"""Resolve demand_belong_category.video_list → multi_demand_video_detail.""" from __future__ import annotations import json from typing import Any from supply_infra.db.repositories.demand_belong_category_repo import ( DemandBelongCategoryRepository, ) from supply_infra.db.repositories.multi_demand_video_detail_repo import ( MultiDemandVideoDetailRepository, ) from supply_infra.db.repositories.multi_demand_video_point_repo import ( MultiDemandVideoPointRepository, ) from supply_infra.db.session import get_session def _parse_video_ids(raw: str | None) -> list[str]: if not raw: return [] try: parsed = json.loads(raw) except json.JSONDecodeError: return [] if not isinstance(parsed, list): return [] return [str(v).strip() for v in parsed if v is not None and str(v).strip()] def list_videos_for_demand_belong(belong_id: int) -> dict[str, Any] | None: """ 按 demand_belong_category.id 返回关联视频详情。 顺序与 video_list 一致;详情表缺失的 vid 仍返回,title/三点 JSON 为空。 """ with get_session() as session: belong = DemandBelongCategoryRepository(session).get_by_id(belong_id) if belong is None or belong.is_delete: return None vids = _parse_video_ids(belong.video_list) details = MultiDemandVideoDetailRepository(session).list_by_vids(vids) points_by_vid = MultiDemandVideoPointRepository(session).json_fields_by_video_ids( vids ) videos: list[dict[str, Any]] = [] for vid in vids: row = details.get(vid) point_fields = points_by_vid.get(vid, {}) videos.append( { "vid": vid, "title": row.title if row else None, "inspiration_points_json": point_fields.get( "inspiration_points_json" ), "purpose_points_json": point_fields.get("purpose_points_json"), "key_points_json": point_fields.get("key_points_json"), } ) return { "demand_belong_id": belong.id, "name": belong.name, "category_id": belong.category_id, "videos": videos, }