xiaoxia-saas/apps/api/app/services/generation_common.py

"""智能剪辑公共服务辅助函数（从 route 层下沉）。

集中管理：
- query_voice_durations：批量查询配音素材时长
- writeback_edit_plan_config：任务入队后回写 EditPlan.config
- collect_plan_segments：分页读取 plan clips 构建素材区间表（变体避让用）
- resolve_latest_plan_by_template：按 template_id + user_id 查最新 EditPlan

设计原则：
- 无副作用的纯查询 / 幂等写回；失败一律不阻断主流程（记日志 + 返回安全默认值）
- 不依赖 FastAPI / HTTPException，便于 service 层和 worker 复用
"""

from __future__ import annotations

import logging
from typing import Any, Optional

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)


def query_voice_durations(db: Session, voice_ids: list[str]) -> list[float]:
    """批量查询配音素材时长（秒），#1749 配音时长分配用。

    逐项 try/float 硬化：MagicMock/异常/缺失 → 0.0（无配音不分配，不阻断）。

    #1855 P0修复：不再对 voice_ids 去重，保持与调用方传入顺序/长度一致，
    允许同配音id多次出现时返回相同时长（支持"同配音N变体"的时长对齐）。
    """
    raw_ids = list(voice_ids or [])
    if not raw_ids:
        return []
    unique_ids: list[str] = []
    _seen: set[str] = set()
    for v in raw_ids:
        if v and v not in _seen:
            _seen.add(v)
            unique_ids.append(v)
    if not unique_ids:
        return [0.0 for _ in raw_ids]
    try:
        from packages.adapters.sqlalchemy_impl.models import AssetModel

        rows = db.query(AssetModel.id, AssetModel.duration).filter(AssetModel.id.in_(unique_ids)).all()
        dur_map: dict[str, float] = {}
        for row in rows:
            try:
                dur_map[row[0]] = float(row[1] or 0.0)
            except (TypeError, ValueError):
                dur_map[row[0]] = 0.0
        return [dur_map.get(v, 0.0) if v else 0.0 for v in raw_ids]
    except Exception:
        logger.warning("[generation_common] 配音时长查询失败（按无配音处理，不阻断）", exc_info=True)
        return [0.0 for _ in raw_ids]


def writeback_edit_plan_config(
    plan_id: str,
    task_id: str,
    title_config: dict | None,
    db: Session,
    dedup_enabled: bool | None = None,
    video_index: int | None = None,
    assembly_mode: str | None = None,
    script_id: str | None = None,
    video_ratio: str | None = None,
) -> None:
    """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。

    用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
    #1970：dedup_enabled 非 None 时一并写入，worker 据此决定 edge_crop/微变换；
    PR3 叙事模式再写 assembly_mode/script_id/video_ratio（可追溯，不影响渲染）。
    失败只记日志，不影响任务创建。
    """
    if not plan_id:
        return
    try:
        from packages.adapters.sqlalchemy_impl.models import EditPlanModel

        plan_model = db.query(EditPlanModel).filter(EditPlanModel.id == plan_id).first()
        if plan_model is None:
            logger.warning("[generation_common] 回写plan.config失败: plan不存在 plan_id=%s", plan_id)
            return

        current_config = plan_model.config if isinstance(plan_model.config, dict) else {}
        merged = dict(current_config)
        merged["generation_task_id"] = task_id
        if dedup_enabled is not None:
            merged["dedup_enabled"] = bool(dedup_enabled)
        if video_index is not None:
            merged["video_index"] = int(video_index)
        if assembly_mode:
            merged["assembly_mode"] = assembly_mode
        if script_id:
            merged["script_id"] = script_id
        if video_ratio:
            merged["video_ratio"] = video_ratio

        if title_config:
            # #1901 统一字段名为 "title"（worker sync_configs_to_plan 写的是 "title"）
            # 先读取新旧两个 key，判断标题文字是否变化
            old_title_cfg = merged.get("title", {}) or {}
            if not isinstance(old_title_cfg, dict) or not (old_title_cfg.get("text") or "").strip():
                old_title_cfg = merged.get("title_config", {}) or {}
            old_title_text = (old_title_cfg.get("text") or "").strip() if isinstance(old_title_cfg, dict) else ""
            new_title_text = (title_config.get("text") or "").strip()
            if old_title_text != new_title_text:
                if "cover" in merged:
                    del merged["cover"]
                    logger.info(
                        "[generation_common] 标题变化，清除旧封面: plan_id=%s old_title=%s new_title=%s",
                        plan_id,
                        old_title_text,
                        new_title_text,
                    )
            # 字段名归一化（font_size→size, font_preset→font, font_color→color），与 worker sync_configs_to_plan 保持一致
            normalized = dict(title_config)
            if "font_size" in normalized and "size" not in normalized:
                normalized["size"] = normalized["font_size"]
            if "font_preset" in normalized and "font" not in normalized:
                normalized["font"] = normalized["font_preset"]
            if "font_color" in normalized and "color" not in normalized:
                normalized["color"] = normalized["font_color"]
            merged["title"] = normalized
            # 清掉旧 key，避免双字段并存
            merged.pop("title_config", None)

        plan_model.config = merged
        db.commit()
        logger.info(
            "[generation_common] 回写plan.config成功: plan_id=%s task_id=%s keys=%s",
            plan_id,
            task_id,
            list(merged.keys()),
        )
    except Exception as e:
        logger.warning(
            "[generation_common] 回写plan.config异常（不影响任务创建）: plan_id=%s error=%s",
            plan_id,
            e,
            exc_info=True,
        )
        try:
            db.rollback()
        except Exception:
            pass


def collect_plan_segments(
    plan_id: str,
    clip_repo: Any,
    *,
    page_size: int = 500,
) -> dict[str, list[tuple[float, float]]]:
    """分页读取 plan 所有 clips，构建 {asset_id: [(start, end), ...]} 素材区间表。

    用于 #1855 P0 批次内素材区间避让（变体间素材片段重叠控制）。
    """
    segs: dict[str, list[tuple[float, float]]] = {}
    sk, pg = 0, page_size
    while True:
        batch = clip_repo.list_by_plan(plan_id, skip=sk, limit=pg)
        if not batch:
            break
        for c in batch:
            if c.asset_id and float(c.duration or 0) > 0:
                st = float(c.start_time or 0.0)
                segs.setdefault(c.asset_id, []).append((st, st + float(c.duration)))
        if len(batch) < pg:
            break
        sk += pg
    return segs


def collect_plan_atom_clip_ids(
    plan_id: str,
    clip_repo: Any,
    *,
    page_size: int = 500,
) -> list[str]:
    """分页读取 plan 所有 clips，收集已选用的原子片段 ID（#1970）。

    用于批量变体间原子片段级硬避让：同一原子片段在同批次内只用一次。
    旧路径 clips 的 atom_clip_id 为空串，自动忽略。
    """
    ids: list[str] = []
    sk, pg = 0, page_size
    while True:
        batch = clip_repo.list_by_plan(plan_id, skip=sk, limit=pg)
        if not batch:
            break
        for c in batch:
            acid = getattr(c, "atom_clip_id", "") or ""
            if acid:
                ids.append(acid)
        if len(batch) < pg:
            break
        sk += pg
    return ids


def resolve_latest_plan_by_template(
    db: Session,
    *,
    template_id: str,
    user_id: str,
) -> Optional[str]:
    """按 template_id + user_id 查找最新的 EditPlan.id（模板兜底用）。找不到返回 None。"""
    if not (template_id or "").strip():
        return None
    try:
        from packages.adapters.sqlalchemy_impl.models import EditPlanModel

        latest = (
            db.query(EditPlanModel)
            .filter(
                EditPlanModel.template_id == template_id.strip(),
                EditPlanModel.created_by_user_id == user_id,
            )
            .order_by(EditPlanModel.created_at.desc())
            .first()
        )
        return latest.id if latest else None
    except Exception:
        logger.warning(
            "[generation_common] 按template查找最新plan失败: template=%s user=%s",
            template_id,
            user_id,
            exc_info=True,
        )
        return None