yawan-workbench/server/app/campaign_jobs.py

"""Content-production jobs shared by inline threads and the DSH worker.

The FastAPI workbench validates and queues work. This module contains the
actual campaign parsing/composition orchestration that must behave identically
whether it runs in an API background thread or in an external DSH process.
"""
from __future__ import annotations

from sqlmodel import Session, select

from . import progress
from .agents.composer import compose as compose_artifact
from .agents.coordinator import parse_mindmap
from .models import Agent, Campaign, Notification, Segment

# Near-synonym content tags used when assigning parsed sections to canvas nodes.
TAG_ALIAS = {
    "产品回指": ["产品销售", "产品体验"],
    "转场": ["开场", "互动"],
    "预告": ["钩子"],
    "见证": ["案例"],
    "金句": ["心法"],
    "共鸣": ["心法"],
    "安全提示": ["技法"],
    "跟练": ["技法", "功法"],
    "氛围": ["场景"],
    "情绪": ["场景", "心法"],
}


def _match_agent(session: Session, tag: str,
                 candidates: list[Agent] | None = None,
                 *, allow_unassigned: bool = False) -> Agent | None:
    aliases = TAG_ALIAS.get(tag, [])
    if candidates is None:
        candidates = list(session.exec(select(Agent).where(
            Agent.role == "production", Agent.status == "active"
        )).all())
    for alias in [tag, *aliases]:
        for agent in candidates:
            if alias in (agent.content_tags or []):
                return agent
    if candidates:
        return candidates[0]
    if allow_unassigned:
        return None
    raise RuntimeError(f"没有能处理标签[{tag}]的启用生产 Agent——请先在 Agent 管理中配置")


def _match_node(session: Session, tag: str, canvas_nodes: list, used: dict,
                *, allow_unassigned: bool = False) -> tuple[Agent | None, str]:
    aliases = [tag, *TAG_ALIAS.get(tag, [])]
    node = None
    for alias in aliases:
        hits = [n for n in canvas_nodes
                if str(n.get("content_tag") or n.get("tag") or "") == alias]
        if hits:
            node = hits[used.get(alias, 0) % len(hits)]
            used[alias] = used.get(alias, 0) + 1
            break
    if node is None:
        return _match_agent(session, tag, allow_unassigned=allow_unassigned), f"seg-{len(used) + 1}"
    agent = session.get(Agent, int(node.get("agent_id")))
    if not agent:
        if allow_unassigned:
            return None, str(node.get("id") or node.get("node_id") or f"seg-{len(used) + 1}")
        raise RuntimeError("画布节点引用的 Agent 不存在")
    if allow_unassigned and (agent.status != "active" or agent.role != "production"):
        return None, str(node.get("id") or node.get("node_id") or f"seg-{len(used) + 1}")
    return agent, str(node.get("id") or node.get("node_id") or f"seg-{len(used) + 1}")


def run_parse_job(session: Session, campaign_id: int, user_advice: str = "") -> dict:
    """Parse a mind map and replace the campaign's draft segment list."""
    campaign = session.get(Campaign, campaign_id)
    if not campaign:
        raise ValueError(f"任务不存在：{campaign_id}")
    campaign.status = "parsing"
    session.add(campaign)
    session.commit()
    progress.publish(int(campaign.owner_user_id or 0), "parse_start",
                     campaign_id=campaign_id, name=campaign.name, backend="dsh")
    try:
        result = parse_mindmap(session, campaign, user_advice=user_advice)
        if result.get("status") != "ok" or not result.get("segments"):
            if result.get("status") == "ok":
                detail = "；".join(list(result.get("pending_confirms") or []) +
                                   list(result.get("notes") or []))
                campaign.notes = [f"拆解没有产出内容段：{str(detail)[:200]}"]
            else:
                campaign.notes = [f"拆解失败：{str(result.get('error'))[:200]}"]
            campaign.status = "parse_failed"
            session.add(campaign)
            session.commit()
            progress.publish(int(campaign.owner_user_id or 0), "parse_failed",
                             campaign_id=campaign_id,
                             error=str(campaign.notes[0] if campaign.notes else "拆解未产出内容段")[:120])
            return {"status": "parse_failed", "error": campaign.notes[0] if campaign.notes else ""}

        for old_seg in session.exec(select(Segment).where(Segment.campaign_id == campaign_id)).all():
            session.delete(old_seg)
        session.commit()

        snap = campaign.canvas_snapshot or {}
        canvas_nodes = snap.get("nodes", []) if isinstance(snap, dict) else []
        used: dict = {}
        unassigned = 0
        for index, item in enumerate(result["segments"]):
            tag = item.get("content_tag", "心法")
            if canvas_nodes:
                agent, node_id = _match_node(session, tag, canvas_nodes, used, allow_unassigned=True)
            else:
                agent = _match_agent(session, tag, allow_unassigned=True)
                node_id = f"seg-{index + 1}"
            if agent is None:
                unassigned += 1
            session.add(Segment(
                campaign_id=campaign_id,
                node_id=node_id,
                agent_id=agent.id if agent else 0,
                title=item.get("title", f"内容段{index + 1}"),
                content_tag=tag,
                time_range=item.get("time_range", ""),
                mindmap_node=item.get("mindmap_node", item.get("title", "")),
                product_ref=item.get("product_ref", ""),
            ))

        campaign.task_spec = result["segments"]
        campaign.pending_confirms = result.get("pending_confirms", [])
        campaign.notes = list(result.get("notes", []))
        if unassigned:
            campaign.notes.append(
                f"有 {unassigned} 个内容段暂未匹配启用的生产 Agent，请在内容段中选择 Agent 后再生成"
            )
        campaign.status = "parsed"
        session.add(campaign)
        session.commit()
        progress.publish(int(campaign.owner_user_id or 0), "parse_done",
                         campaign_id=campaign_id, segments=len(result["segments"]))
        return {"status": "parsed", "segments": len(result["segments"])}
    except Exception as exc:
        session.rollback()
        campaign = session.get(Campaign, campaign_id)
        if campaign:
            campaign.status = "parse_failed"
            campaign.notes = [f"拆解异常：{str(exc)[:200]}"]
            session.add(campaign)
            session.commit()
        progress.publish(int(campaign.owner_user_id if campaign else 0), "parse_failed",
                         campaign_id=campaign_id, error=str(exc)[:120])
        progress.publish(int(campaign.owner_user_id if campaign else 0), "parse_done",
                         campaign_id=campaign_id, segments=0)
        return {"status": "parse_failed", "error": str(exc)}


def run_compose_job(session: Session, campaign_id: int) -> dict:
    """Run final composition and report progress through the shared bus."""
    campaign = session.get(Campaign, campaign_id)
    if not campaign:
        raise ValueError(f"任务不存在：{campaign_id}")
    progress.publish(int(campaign.owner_user_id or 0), "compose_start",
                     campaign_id=campaign_id, name=campaign.name, backend="dsh")
    try:
        result = compose_artifact(session, campaign_id)
    except Exception:
        # 基础设施异常（文件、模型、数据库等）交给任务队列退避重试/死信，
        # 保持 composing，由 DSH 重试或死信补偿逻辑恢复到可操作状态。
        session.rollback()
        raise
    if result.get("status") == "error":
        # compose 主动返回的业务校验失败（如还有段未审查、没有已通过段）
        # 不应触发队列重试；交还工作台，用户处理后可再次点击组合。
        campaign = session.get(Campaign, campaign_id)
        if campaign and campaign.status == "composing":
            campaign.status = "composing_ready"
            session.add(campaign)
            session.commit()
        owner_id = int(campaign.owner_user_id if campaign else 0)
        error_text = str(result.get("error") or "成稿组合失败")[:200]
        progress.publish(owner_id, "compose_failed",
                         campaign_id=campaign_id, error=error_text)
        if campaign:
            session.add(Notification(
                type="failure", ref_id=campaign_id,
                message=f"DSH 成稿组合失败：{error_text[:160]}",
            ))
            session.commit()
        return result
    progress.publish(int(campaign.owner_user_id or 0), "compose_done",
                     campaign_id=campaign_id, artifact_id=result.get("artifact_id"),
                     status=result.get("status"))
    return result