SupplyAgent/api/services/video_discovery.py

"""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_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.session import get_session

_POINT_TYPES = {"inspiration", "purpose", "key"}


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,
) -> 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,
    }


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
    }


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 = biz_dt or grade_repo.get_latest_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))
        return {
            "biz_dt": str(resolved_biz_dt),
            "items": [
                _serialize_demand(
                    row,
                    category_names,
                    video_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.

    multi_demand_video_detail contributes only the title and no other field.
    """
    with get_session() as session:
        grade = DemandGradeRepository(session).get_by_id(demand_grade_id)
        if grade is None:
            return None

        expansions = DemandVideoExpansionRepository(session).list_by_demand_grade(
            str(grade.biz_dt),
            demand_grade_id,
        )
        category_names = _category_name_map(session, [grade])

        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(
                {
                    "point_type": row.point_type,
                    "expanded_text": row.expanded_text,
                    "point_desc": row.point_desc,
                    "reason": row.reason,
                }
            )

        details = MultiDemandVideoDetailRepository(session).list_by_vids(video_ids)
        videos: list[dict[str, Any]] = []
        for video_id in video_ids:
            points = points_by_video[video_id]
            detail = details.get(video_id)
            videos.append(
                {
                    "vid": video_id,
                    "title": detail.title if detail else None,
                    "point_count": len(points),
                    "points": points,
                }
            )

        demand = _serialize_demand(grade, category_names, len(videos))
        demand["point_count"] = sum(video["point_count"] for video in videos)
        demand["videos"] = videos
        return demand