beige/backend/app/services/fission_pipeline.py

"""
app/services/fission_pipeline.py — 第11环裂变 编排层（从 fission_service 拆出）

Celery 内调用：一次 LLM 出 N 套完整笔记包 → 解析/兜底 → 评分@80 → 落 FissionNote
→ 逐套生图存 images_json。拆分原因：fission_service.py 超 200 行红线上限。
🔴 生图引擎零改动：复用单篇 generate_storyboard_images（codeproxy edits 带产品参考图）。
🔴 评分沿用 llm_score_copy，合格线 QUALITY_PASS_SCORE=80（禁改）。
🔴 FISSION_SYSTEM 作 messages[0].role=system 传（chat_complete 是 OpenAI 兼容）。
"""
import logging

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

MAX_FANOUT = 5  # 每用户并发上限5(红线)；裂变 N 套直出受同等约束


def _parse_fission_response(
    raw: str, source_note: dict, product: dict, note_count: int, image_count: int,
) -> list[dict]:
    """解析 LLM 返回的 N 套笔记；不可解析则品类兜底（不卡用户）。

    解析链：parse_json_array(容错markdown) → notes_array_from_parsed(拎数组) →
    空则 build_fallback_notes 品类草稿。每套补 imagePlan/tags 归一化。
    """
    import json as _json
    from app.services.ai_engine._text_prompt import parse_json_array
    from app.services.ai_engine.fission_prompt import (
        normalize_tags, notes_array_from_parsed,
    )
    from app.services.ai_engine.fission_fallback import (
        build_fallback_image_plan, build_fallback_notes,
    )

    notes = notes_array_from_parsed(parse_json_array(raw))
    if not notes:
        # parse_json_array 只认数组；再试整段当对象解析（容错单对象返回）
        try:
            notes = notes_array_from_parsed(_json.loads(raw))
        except (ValueError, TypeError):
            notes = []
    if not notes:
        logger.warning("裂变 LLM 返回不可解析，启用品类兜底。raw[:120]=%s", str(raw)[:120])
        return build_fallback_notes(source_note, product, note_count, image_count)

    out: list[dict] = []
    for n in notes:
        if not isinstance(n, dict):
            continue
        n["tags"] = normalize_tags(n.get("tags", []), n.get("keywords", []))
        plan = n.get("imagePlan")
        if not isinstance(plan, list) or len(plan) != image_count:
            n["imagePlan"] = build_fallback_image_plan(n, image_count)
        out.append(n)
    return out or build_fallback_notes(source_note, product, note_count, image_count)


def _score_notes(clients, notes: list[dict], source_note: dict, banned: list[str]) -> None:
    """对每套笔记 LLM 评分（限2并发），结果就地写回 note['_score']/_passed/_block。

    Celery 同步环境：用 asyncio.run 跑一个内部 gather（信号量限2并发，
    对齐生图限流，避免中转站 429）。
    """
    import asyncio
    from app.services.ai_engine.llm_scorer import llm_score_copy
    from app.services.ai_engine.constants import QUALITY_PASS_SCORE

    async def _run() -> None:
        sem = asyncio.Semaphore(2)

        async def _one(note: dict) -> None:
            copy = {
                "title": note.get("title", ""),
                "content": note.get("content", ""),
                "tags": note.get("tags", []),
            }
            async with sem:
                try:
                    r = await llm_score_copy(
                        clients, copy, source_note, banned, pass_score=QUALITY_PASS_SCORE,
                    )
                except Exception as exc:  # noqa: BLE001
                    logger.warning("裂变评分异常，记0分不达标: %s", exc)
                    r = {"score": 0, "passed": False, "banned_words_found": []}
            note["_score"] = int(r.get("score", 0))
            note["_block"] = bool(r.get("banned_words_found"))
            # 过线 = score≥80 且无硬拦违禁词
            note["_passed"] = bool(r.get("passed")) and not note["_block"]

        await asyncio.gather(*(_one(n) for n in notes))

    asyncio.run(_run())


def execute_fission_pipeline(db: Session, clients, fission_id: int, source_task_id: int) -> dict:
    """裂变主编排（Celery 内调用，clients 已构建好）。

    流程：查 FissionTask+源产品 → 一次 chat_complete 出 N 套 → 解析/兜底
    → 每套评分@80 → 按分排序取 N 套（不够用兜底补，不达标标 needs_optimization）
    → 落 FissionNote → 逐套生图存 images_json → 回写 FissionTask.status。
    """
    import json as _json

    from app.models.fission import FissionTask, FissionNote
    from app.models.task import GenerationTask
    from app.models.product import Product
    from app.workers.pipeline_steps import build_product_dict
    from app.services.ai_engine.fission_prompt import build_fission_prompt

    ft = db.query(FissionTask).filter(FissionTask.id == fission_id).first()
    if not ft:
        return {"fission_id": fission_id, "status": "not_found"}

    n = max(1, min(ft.fanout_count or 3, MAX_FANOUT))
    try:
        source_note = _json.loads(ft.source_note) if ft.source_note else {}
    except (ValueError, TypeError):
        source_note = {}

    src = db.query(GenerationTask).filter(GenerationTask.id == source_task_id).first()
    product_row = db.query(Product).filter(Product.id == src.product_id).first() if src else None
    if not src or not product_row:
        ft.status = "failed"; db.commit()
        return {"fission_id": fission_id, "status": "failed", "reason": "源任务或产品缺失"}

    product = build_product_dict(product_row)
    image_count = max(1, src.image_count or 3)
    banned = []  # 违禁词分级表后续接入；评分器内置 BANNED_WORDS_DEFAULT 已兜底

    # 一次 LLM 出 N 套（FISSION_SYSTEM 作 messages[0].role=system）
    from app.services.ai_engine.fission_prompt import FISSION_SYSTEM
    user_prompt = build_fission_prompt(
        source_note, product, ft.reference_level, n, image_count,
    )
    # max_tokens 按套数缩放：每套完整笔记包约 1200 token，留足余量
    max_tokens = min(8192, 1800 + n * 1400)
    raw = ""
    try:
        import asyncio
        raw = asyncio.run(clients.chat_complete(
            messages=[{"role": "system", "content": FISSION_SYSTEM},
                      {"role": "user", "content": user_prompt}],
            model=clients._model, max_tokens=max_tokens, temperature=0.8,
        ))
    except Exception as exc:  # noqa: BLE001
        logger.warning("裂变 LLM 调用失败，启用品类兜底: %s", exc)

    notes = _parse_fission_response(raw, source_note, product, n, image_count)
    _score_notes(clients, notes, source_note, banned)

    # 排序：过线优先，再按分降序；取前 N 套（不足用兜底草稿补齐）
    notes.sort(key=lambda x: (x.get("_passed", False), x.get("_score", 0)), reverse=True)
    chosen = notes[:n]

    saved_ids: list[int] = []
    for seq, note in enumerate(chosen):
        passed = bool(note.get("_passed"))
        fn = FissionNote(
            fission_id=fission_id, workspace_id=ft.workspace_id, seq=seq,
            note_json=_json.dumps(note, ensure_ascii=False),
            score=int(note.get("_score", 0)),
            passed=1 if passed else 0,
            needs_optimization=0 if passed else 1,  # 不达标不丢弃，标降级草稿
            dimension=str(note.get("dimension", ""))[:64],
            status="scored",
        )
        db.add(fn); db.flush()
        saved_ids.append(fn.id)
    db.commit()

    logger.info("裂变出 %s 套已落库 fission=%s ids=%s", len(saved_ids), fission_id, saved_ids)

    # 逐套生图（复用单篇引擎），存 images_json
    from app.services.fission_images import generate_fission_images
    generate_fission_images(db, clients, ft, product, image_count, saved_ids)

    ft.status = "completed"; db.commit()
    return {"fission_id": fission_id, "status": "completed", "note_ids": saved_ids}