""" app/workers/packaging_task.py — 交付打包 Celery 任务 build_delivery_package:查已选文案+图片 → package_exporter → 存路径 """ import json import logging import os from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) def _get_db(): from app.core.database import SessionLocal return SessionLocal() @celery_app.task( bind=True, name="app.workers.tasks.build_delivery_package", max_retries=2, default_retry_delay=10, queue="packaging", ) def build_delivery_package(self, package_id: int) -> dict: """打包交付任务。查 delivery_packages → 收集笔记 → package_exporter""" logger.info("build_delivery_package start: package_id=%s", package_id) db = _get_db() try: from app.models.task import DeliveryPackage, TextCandidate, ImageCandidate from app.constants.enums import PackageStatus pkg = db.query(DeliveryPackage).filter(DeliveryPackage.id == package_id).first() if not pkg: raise ValueError(f"package_id={package_id} not found") workspace_id = pkg.workspace_id task_id = pkg.task_id from app.core.config import get_settings settings = get_settings() upload_root = settings.UPLOAD_ABS_ROOT.rstrip("/") # A8 多套打包:按 strategy(A/B/C 三套正交叙事)分组,每套成一篇独立 note。 # 北哥拿到的交付包含完整 3 套(note_01/02/03),不再只打第 1 条文案。 selected_texts = db.query(TextCandidate).filter( TextCandidate.task_id == task_id, TextCandidate.is_selected == True, ).order_by(TextCandidate.id).all() if not selected_texts: raise ValueError("无已选文案,请先选择文案") # 整套全打:一套内全部图按 seq 排序进包,不只打封面。重生场景同 (strategy,seq) # 可能多条(新增不删旧),去重取最新(id最大),避免包内重复图。 all_images = db.query(ImageCandidate).filter( ImageCandidate.task_id == task_id, ).order_by(ImageCandidate.strategy, ImageCandidate.seq).all() def _read_image(ic) -> dict: img_bytes = b"" if ic.url: path = ic.url if os.path.isabs(path): pass elif path.startswith("/uploads/"): path = f"{upload_root}/{path.removeprefix('/uploads/')}" elif path.startswith("uploads/"): path = f"{upload_root}/{path.removeprefix('uploads/')}" try: with open(path, "rb") as f: img_bytes = f.read() except OSError as e: logger.warning("图片读取失败,跳过:%s %s", ic.url, e) return { "seq": ic.seq, "role": ic.role.value if hasattr(ic.role, "value") else str(ic.role), "data": img_bytes, } # 按 strategy 分组(A/B/C;老数据 strategy=None 归一套,向后兼容) from collections import OrderedDict groups: "OrderedDict[str, dict]" = OrderedDict() for ic in all_images: slot = groups.setdefault(ic.strategy or "_", {}) prev = slot.get(ic.seq) if prev is None or ic.id > prev.id: slot[ic.seq] = ic # 同 seq 留最新 # 文案按 strategy 建映射,供图片组按套精准配对(A8:文图同套对齐,不靠脆弱 idx)。 # 同套多条选中取第 1 条;老数据 strategy=None 的归入 fallback 列表。 texts_by_strategy: dict = {} texts_no_strategy: list = [] for tc in selected_texts: if tc.strategy: texts_by_strategy.setdefault(tc.strategy, tc) else: texts_no_strategy.append(tc) _text_fallback = iter(texts_no_strategy or selected_texts) notes = [] for idx, (_strategy, slot) in enumerate(groups.items()): images_data = [_read_image(slot[k]) for k in sorted(slot)] # 配对优先级:①同 strategy 文案精准对齐 ②无同套则按顺序取无套文案 # ③再兜底用第 idx 条/第 1 条,确保每组图都有文案不漏。 tc = texts_by_strategy.get(_strategy) if tc is None: tc = next(_text_fallback, None) if tc is None: tc = selected_texts[idx] if idx < len(selected_texts) else selected_texts[0] text_data = json.loads(tc.content or "{}") notes.append({ "title": text_data.get("title", ""), "content": text_data.get("content", ""), "tags": text_data.get("tags", []), "images": images_data, "banned_word_status": (tc.banned_word_status.value if hasattr(tc.banned_word_status, "value") else str(tc.banned_word_status)), }) if not notes: raise ValueError("无图片候选,无法打包") from app.services.ai_engine.package_exporter import build_delivery_package as do_build # 打包产物放持久化卷 /app/uploads/packages/,与图片目录 /app/uploads/{ws}/{task}/ 分开。 packages_base = os.path.join(upload_root, "packages") zip_path = do_build(workspace_id, task_id, notes, base_path=packages_base) pkg.package_path = zip_path pkg.download_url = f"/api/v1/delivery-packages/{package_id}/download-file" pkg.status = PackageStatus.READY db.commit() logger.info("delivery package ready: package_id=%s path=%s", package_id, zip_path) return {"package_id": package_id, "status": "ready", "path": zip_path} except Exception as exc: logger.error("build_delivery_package failed: package_id=%s err=%s", package_id, exc) raise self.retry(exc=exc) finally: db.close()