SupplyAgent/api/services/demand_grade_videos.py

"""Resolve demand_video_expansion → videos + expansion points for the web UI."""
from __future__ import annotations

import json
from typing import Any

from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
from supply_infra.db.repositories.demand_video_expansion_repo import (
    DemandVideoExpansionRepository,
    DemandVideoExpansionRunRepository,
)
from supply_infra.db.repositories.multi_demand_video_detail_repo import (
    MultiDemandVideoDetailRepository,
)
from supply_infra.db.session import get_session


def _parse_json_list(raw: str | None) -> list[Any]:
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
    except json.JSONDecodeError:
        return []
    return parsed if isinstance(parsed, list) else []


def list_videos_for_demand_grade(demand_grade_id: int) -> dict[str, Any] | None:
    """
    按 demand_grade.id 返回该需求在对应 biz_dt 下的拓展视频与选题结果。

    数据来源：
    - demand_video_expansion_run：判断是否已完成拓展（含零结果）
    - demand_video_expansion：真实视频实例与拓展点位
    """
    with get_session() as session:
        grade = DemandGradeRepository(session).get_by_id(demand_grade_id)
        if grade is None:
            return None

        biz_dt = str(grade.biz_dt)
        run = DemandVideoExpansionRunRepository(session).get_by_demand_grade(
            biz_dt, demand_grade_id
        )
        expansions = DemandVideoExpansionRepository(session).list_by_demand_grade(
            biz_dt, demand_grade_id
        )

        vids: list[str] = []
        seen_vids: set[str] = set()
        points_by_vid: dict[str, list[dict[str, Any]]] = {}
        for row in expansions:
            vid = str(row.video_id).strip()
            if not vid:
                continue
            if vid not in seen_vids:
                seen_vids.add(vid)
                vids.append(vid)
            points_by_vid.setdefault(vid, []).append(
                {
                    "expanded_text": row.expanded_text,
                    "point_type": row.point_type,
                    "point_desc": row.point_desc,
                    "reason": row.reason,
                }
            )

        details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
        videos: list[dict[str, Any]] = []
        for vid in vids:
            row = details.get(vid)
            videos.append(
                {
                    "vid": vid,
                    "title": row.title if row else None,
                    "expansion_points": points_by_vid.get(vid, []),
                }
            )

        expansion_status: str | None = None
        expansion_saved_count = 0
        if run is not None:
            expansion_status = str(run.status)
            expansion_saved_count = int(run.saved_count or 0)

        return {
            "demand_grade_id": grade.id,
            "demand_name": grade.demand_name,
            "biz_dt": biz_dt,
            "category_ids": [int(c) for c in _parse_json_list(grade.category_ids)],
            "grade": grade.grade,
            "strategies": _parse_json_list(grade.strategies),
            "expansion_status": expansion_status,
            "expansion_saved_count": expansion_saved_count,
            "videos": videos,
        }