""" app/services/task_service.py — 任务创建 service 校验有无 key → 建 GenerationTask → 只推 task_id 入队,绝不传 key(基石B)。 """ import logging from sqlalchemy.orm import Session from app.core.response import raise_business from app.middleware.workspace_guard import CurrentUser from app.models.task import GenerationTask from app.models.workspace import UserApiKey logger = logging.getLogger(__name__) def _check_user_has_key(db: Session, user_id: int, workspace_id: int) -> None: """校验用户在此 workspace 是否有可用 API Key(openai/apiports均可),没有则引导去配置。""" key = ( db.query(UserApiKey) .filter( UserApiKey.user_id == user_id, UserApiKey.workspace_id == workspace_id, UserApiKey.provider.in_(["openai", "apiports"]), # G6坑修复:接受主备通道名 ) .first() ) if not key: raise_business("尚未配置 API Key,请先在设置中录入") def create_generation_task( db: Session, current_user: CurrentUser, body, # CreateTaskRequest ) -> GenerationTask: """ 建 GenerationTask 并推 Celery 队列。 只传 task_id,绝不传 key(基石B)。 """ if body.track == "ai": # 轨A:先检查有没有 key _check_user_has_key(db, current_user.user_id, current_user.workspace_id) # 禁降级铁律:本次产品入镜(need_product_image=True)时,产品必须已上传参考图, # 否则拒绝建任务(不允许降级纯文生图,防产品包装跑偏/过抽检失败)。 need_img = getattr(body, "need_product_image", True) if need_img: from app.models.product import Product product = db.query(Product).filter( Product.id == body.product_id, Product.workspace_id == current_user.workspace_id, ).first() if not product: raise_business("产品不存在") if not (product.image_path or "").strip(): raise_business("该产品未上传参考图,无法生成产品入镜内容;请先到产品库上传产品图,或关闭「产品入镜」开关") task = GenerationTask( workspace_id=current_user.workspace_id, product_id=body.product_id, operator_id=current_user.user_id, theme=body.theme, text_count=body.text_count, image_count=body.image_count, track=body.track, need_product_image=need_img, status="pending", ) db.add(task) db.commit() db.refresh(task) logger.info("GenerationTask created: id=%s ws=%s", task.id, current_user.workspace_id) if body.track == "ai": enqueue_generation(task.id) return task def enqueue_generation(task_id: int) -> None: """只推 task_id 入队,绝不推 key(基石B)。""" from app.workers.tasks import run_generation_pipeline run_generation_pipeline.delay(task_id) logger.info("Enqueued task_id=%s", task_id)