"""Demand-first video discovery data for the web workspace.""" from __future__ import annotations import json from typing import Any from supply_infra.db.repositories.demand_feedback_repo import DemandFeedbackRepository from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository from supply_infra.db.repositories.demand_video_expansion_repo import ( DemandVideoExpansionRepository, ) from supply_infra.db.repositories.global_tree_category_repo import ( GlobalTreeCategoryRepository, ) from supply_infra.db.repositories.multi_demand_video_detail_repo import ( MultiDemandVideoDetailRepository, ) from supply_infra.db.repositories.pipeline_step_run_repo import PipelineStepRunRepository from supply_infra.db.session import get_session _POINT_TYPES = {"inspiration", "purpose", "key"} _SOURCE_VIDEO_GRADES = frozenset({"B", "C", "D"}) _VIDEO_DETAIL_URL_TEMPLATE = "https://admin.piaoquantv.com/cms/post-detail/{vid}/detail" def _parse_video_ids(raw: Any) -> list[str]: """Parse JSON array or comma-separated text into ordered unique vid list.""" if raw is None: return [] items: list[Any] if isinstance(raw, str): text = raw.strip() if not text: return [] try: parsed = json.loads(text) except (TypeError, ValueError): parsed = None if isinstance(parsed, list): items = parsed elif parsed is not None: items = [parsed] else: items = [part.strip() for part in text.split(",")] elif isinstance(raw, (list, tuple)): items = list(raw) else: return [] result: list[str] = [] seen: set[str] = set() for item in items: vid = str(item).strip() if item is not None else "" if vid and vid not in seen: seen.add(vid) result.append(vid) return result def _normalize_url(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None def _serialize_video( video_id: str, detail: Any | None, points: list[dict[str, Any]], feedback_count: int = 0, ) -> dict[str, Any]: url1 = _normalize_url(detail.url1) if detail else None url2 = _normalize_url(detail.url2) if detail else None return { "vid": video_id, "title": detail.title if detail else None, "url1": url1, "url2": url2, "video_detail_url": _VIDEO_DETAIL_URL_TEMPLATE.format(vid=video_id), "point_count": len(points), "points": points, "feedback_summary": {"count": int(feedback_count)}, } def _source_video_count(row: Any, expansion_counts: dict[int, int]) -> int: grade = str(row.grade or "").upper() if grade in _SOURCE_VIDEO_GRADES: return len(_parse_video_ids(row.video_list)) return expansion_counts.get(int(row.id), 0) def _parse_string_list(raw: Any) -> list[str]: """Parse JSON arrays and legacy comma-separated text into a stable string list.""" if raw is None: return [] values: list[Any] if isinstance(raw, str): text = raw.strip() if not text: return [] try: parsed = json.loads(text) except (TypeError, ValueError): parsed = None if isinstance(parsed, list): values = parsed elif parsed is not None: values = [parsed] else: values = [part.strip() for part in text.split(",")] elif isinstance(raw, (list, tuple)): values = list(raw) else: values = [raw] result: list[str] = [] seen: set[str] = set() for value in values: text = str(value).strip() if value is not None else "" if text and text not in seen: seen.add(text) result.append(text) return result def _parse_int_list(raw: Any) -> list[int]: result: list[int] = [] for value in _parse_string_list(raw): try: result.append(int(value)) except ValueError: continue return result def _number(value: Any) -> float | None: return float(value) if value is not None else None def _serialize_demand( row: Any, category_names: dict[int, str], video_count: int, feedback_count: int = 0, ) -> dict[str, Any]: category_ids = _parse_int_list(row.category_ids) return { "id": int(row.id), "demand_name": row.demand_name, "biz_dt": str(row.biz_dt), "grade": row.grade, "score": _number(row.score), "prior_total_score": _number(row.prior_total_score), "posterior_rov_avg": _number(row.posterior_rov_avg), "posterior_rov_count": int(row.posterior_rov_count or 0), "has_posterior": bool(row.has_posterior), "category_ids": category_ids, "category_names": [ category_names[category_id] for category_id in category_ids if category_id in category_names ], "strategies": _parse_string_list(row.strategies), "reason": row.reason, "video_count": video_count, "feedback_summary": {"count": int(feedback_count)}, } def _category_name_map(session: Any, rows: list[Any]) -> dict[int, str]: category_ids = sorted( { category_id for row in rows for category_id in _parse_int_list(row.category_ids) } ) categories = GlobalTreeCategoryRepository(session).list_active_by_ids(category_ids) return { int(category.id): str(category.name) for category in categories if category.name } _DEMAND_EXPAND_STEP_KEY = "demand_expand" def _resolve_video_discovery_biz_dt(session: Any, biz_dt: str | None) -> str | None: """默认取最近一次成功完成的 demand_expand 步骤所属业务日。""" if biz_dt: return str(biz_dt) latest_expand_dt = PipelineStepRunRepository(session).get_latest_succeeded_biz_dt( _DEMAND_EXPAND_STEP_KEY ) if latest_expand_dt: return latest_expand_dt return DemandGradeRepository(session).get_latest_biz_dt() def list_video_discovery_demands(biz_dt: str | None = None) -> dict[str, Any]: """Return one demand card per demand_grade row for the selected/latest day.""" with get_session() as session: grade_repo = DemandGradeRepository(session) resolved_biz_dt = _resolve_video_discovery_biz_dt(session, biz_dt) if not resolved_biz_dt: return {"biz_dt": None, "items": []} rows = grade_repo.list_by_biz_dt(str(resolved_biz_dt)) category_names = _category_name_map(session, rows) video_counts = DemandVideoExpansionRepository( session ).count_distinct_videos_by_demand_grade(str(resolved_biz_dt)) feedback_counts = DemandFeedbackRepository(session).count_demands( [int(row.id) for row in rows] ) return { "biz_dt": str(resolved_biz_dt), "items": [ _serialize_demand( row, category_names, _source_video_count(row, video_counts), feedback_counts.get(int(row.id), 0), ) for row in rows ], } def get_video_discovery_demand(demand_grade_id: int) -> dict[str, Any] | None: """ Resolve the judged video list and hit evidence from demand_video_expansion. S/A 使用拓展命中视频;B/C/D 使用 demand_grade.video_list 中的源视频。 multi_demand_video_detail 补充 title 与解构 url1/url2。 """ with get_session() as session: grade = DemandGradeRepository(session).get_by_id(demand_grade_id) if grade is None: return None category_names = _category_name_map(session, [grade]) grade_code = str(grade.grade or "").upper() demand_feedback_count, video_feedback_counts, point_feedback_counts = ( DemandFeedbackRepository(session).count_for_demand(demand_grade_id) ) if grade_code in _SOURCE_VIDEO_GRADES: video_ids = _parse_video_ids(grade.video_list) details = MultiDemandVideoDetailRepository(session).list_by_vids(video_ids) videos = [ _serialize_video( video_id, details.get(video_id), [], video_feedback_counts.get(video_id, 0), ) for video_id in video_ids ] else: expansions = DemandVideoExpansionRepository(session).list_by_demand_grade( str(grade.biz_dt), demand_grade_id, ) video_ids: list[str] = [] points_by_video: dict[str, list[dict[str, Any]]] = {} for row in expansions: video_id = str(row.video_id).strip() if not video_id or row.point_type not in _POINT_TYPES: continue if video_id not in points_by_video: video_ids.append(video_id) points_by_video[video_id] = [] points_by_video[video_id].append( { "id": int(row.id), "point_type": row.point_type, "expanded_text": row.expanded_text, "point_desc": row.point_desc, "reason": row.reason, "feedback_summary": { "count": point_feedback_counts.get(int(row.id), 0) }, } ) details = MultiDemandVideoDetailRepository(session).list_by_vids(video_ids) videos = [ _serialize_video( video_id, details.get(video_id), points_by_video[video_id], video_feedback_counts.get(video_id, 0), ) for video_id in video_ids ] demand = _serialize_demand( grade, category_names, len(videos), demand_feedback_count, ) demand["point_count"] = sum(video["point_count"] for video in videos) demand["videos"] = videos demand["video_source"] = ( "pool" if grade_code in _SOURCE_VIDEO_GRADES else "expansion" ) return demand