SupplyAgent/supply_infra/scheduler/jobs/demand_pool/videos.py

"""
从 multi_demand_pool_di 收集视频，增量写入 multi_demand_video_detail，
并从 ODPS 昨天分区拉取最终选题 JSON 与标题。
"""
from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta
from typing import Any

from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
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
from supply_infra.odps.client import ODPSClient, get_odps_client
from supply_infra.video_points import (
    extract_points_json_from_decode,
    points_from_decode_payload,
)

logger = logging.getLogger(__name__)

_FINAL_TOPIC_KEY = "最终选题"
_TARGET_POST_KEY = "target_post"
_TITLE_KEY = "title"
VIDEO_SYNC_BATCH_SIZE = 100


def _collect_vids_from_video_lists(video_lists: list[str]) -> set[str]:
    """解析 video_list JSON 文本，收集去重 vid。"""
    vids: set[str] = set()
    for text in video_lists:
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError:
            logger.warning("Skip invalid video_list JSON: %s", text[:120])
            continue
        if not isinstance(parsed, list):
            logger.warning("Skip non-list video_list: %s", text[:120])
            continue
        for item in parsed:
            if item is None:
                continue
            vid = str(item).strip()
            if vid:
                vids.add(vid)
    return vids


def _parse_decode_result(decode_result: Any) -> dict[str, Any] | None:
    """将 decode_result 解析为 dict。"""
    if decode_result is None:
        return None

    if isinstance(decode_result, dict):
        return decode_result

    if isinstance(decode_result, str):
        text = decode_result.strip()
        if not text:
            return None
        try:
            payload = json.loads(text)
        except json.JSONDecodeError:
            logger.warning("Skip invalid decode_result JSON: %s", text[:120])
            return None
        return payload if isinstance(payload, dict) else None

    return None


def _extract_final_topic_json(payload: dict[str, Any]) -> str | None:
    """从 decode_result 取出「最终选题」并序列化为 JSON 文本。"""
    final_topic = payload.get(_FINAL_TOPIC_KEY)
    if final_topic is None:
        return None
    return json.dumps(final_topic, ensure_ascii=False)


def _extract_all_points(payload: dict[str, Any]) -> dict[str, str | None]:
    """从 decode_result 提取灵感点/目的点/关键点三个 JSON 列。"""
    return extract_points_json_from_decode(payload)


def _extract_title(payload: dict[str, Any]) -> str | None:
    """从 decode_result.target_post.title 取标题。"""
    target_post = payload.get(_TARGET_POST_KEY)
    if not isinstance(target_post, dict):
        return None
    title = target_post.get(_TITLE_KEY)
    if title is None:
        return None
    text = str(title).strip()
    return text[:512] if text else None


def _extract_url(value: Any, max_len: int = 1024) -> str | None:
    """从 ODPS 行字段提取 URL 文本（与 vid 同级）。"""
    if value is None:
        return None
    text = str(value).strip()
    return text[:max_len] if text else None


def _sync_one_batch(
    odps: ODPSClient,
    decode_dt: str,
    batch_vids: list[str],
    batch_idx: int,
    batch_total: int,
) -> dict[str, int]:
    """查询一批 vid，立刻写入 MySQL。"""
    logger.info(
        "Batch %d/%d: query %d vids, decode_dt=%s",
        batch_idx,
        batch_total,
        len(batch_vids),
        decode_dt,
    )
    odps_rows = odps.fetch_topic_decode_results(
        decode_dt, batch_vids, batch_size=len(batch_vids)
    )

    insert_rows: list[dict[str, Any]] = []
    point_rows_by_vid: dict[str, list[dict[str, Any]]] = {}
    skipped_no_topic = 0
    seen_vids: set[str] = set()
    for row in odps_rows:
        raw_vid = row.get("vid")
        if raw_vid is None:
            continue
        vid = str(raw_vid).strip()
        if not vid or vid in seen_vids:
            continue
        payload = _parse_decode_result(row.get("decode_result"))
        if payload is None:
            skipped_no_topic += 1
            continue
        topic_json = _extract_final_topic_json(payload)
        if topic_json is None:
            skipped_no_topic += 1
            continue
        seen_vids.add(vid)
        insert_rows.append(
            {
                "vid": vid,
                "url1": _extract_url(row.get("url1")),
                "url2": _extract_url(row.get("url2")),
                "title": _extract_title(payload),
                "final_topic_json": topic_json,
                **_extract_all_points(payload),
            }
        )
        point_rows = points_from_decode_payload(vid, payload)
        if point_rows:
            point_rows_by_vid[vid] = point_rows

    with get_session() as session:
        detail_repo = MultiDemandVideoDetailRepository(session)
        inserted = detail_repo.bulk_insert_ignore(insert_rows)
        if point_rows_by_vid:
            MultiDemandVideoPointRepository(session).replace_for_video_ids(
                point_rows_by_vid
            )

    odps_vids = {
        str(r.get("vid")).strip()
        for r in odps_rows
        if r.get("vid") is not None and str(r.get("vid")).strip()
    }
    missing_in_odps = len(set(batch_vids) - odps_vids)
    stats = {
        "odps_rows": len(odps_rows),
        "inserted": inserted,
        "skipped_no_topic": skipped_no_topic,
        "missing_in_odps": missing_in_odps,
    }
    logger.info("Batch %d/%d done: %s", batch_idx, batch_total, stats)
    return stats


def sync_multi_demand_videos(
    limit: int | None = None,
    offset: int = 0,
    batch_size: int = VIDEO_SYNC_BATCH_SIZE,
    *,
    decode_dt: str | None = None,
) -> dict[str, Any]:
    """
    增量同步需求池视频与最终选题。

    - 视频来源：multi_demand_pool_di 全表 video_list
    - 已有 vid 跳过
    - ODPS decode 分区由批次冻结参数传入；兼容调用未传时使用「今天的昨天」
    - 按 batch_size 分批：每批查询后立刻写入
    - limit: 本轮最多处理的 pending 数；None 表示从 offset 起全部
    - offset: pending 列表起始偏移（手动分批用）
    """
    resolved_decode_dt = decode_dt or (
        datetime.now() - timedelta(days=1)
    ).strftime("%Y%m%d")
    start = max(0, int(offset))
    chunk = max(1, int(batch_size))
    logger.info(
        "Starting multi demand video sync: decode_dt=%s limit=%s offset=%d batch_size=%d",
        resolved_decode_dt,
        limit,
        start,
        chunk,
    )

    with get_session() as session:
        video_lists = MultiDemandPoolDiRepository(session).list_all_video_lists()
        all_vids = _collect_vids_from_video_lists(video_lists)
        existing = (
            MultiDemandVideoDetailRepository(session).get_existing_vids(all_vids)
            if all_vids
            else set()
        )

    pending_all = sorted(all_vids - existing)
    pending = pending_all[start:]
    if limit is not None:
        pending = pending[: max(0, int(limit))]

    logger.info(
        "Vids: pool_lists=%d unique=%d existing=%d pending_total=%d to_process=%d",
        len(video_lists),
        len(all_vids),
        len(existing),
        len(pending_all),
        len(pending),
    )

    empty = {
        "decode_dt": resolved_decode_dt,
        "pool_video_lists": len(video_lists),
        "unique_vids": len(all_vids),
        "pending_total": len(pending_all),
        "offset": start,
        "batch_size": chunk,
        "batches": 0,
        "processed": 0,
        "odps_rows": 0,
        "inserted": 0,
        "skipped_no_topic": 0,
        "missing_in_odps": 0,
        "next_offset": start,
        "remaining": max(0, len(pending_all) - start),
    }
    if not pending:
        logger.info("No pending vids: %s", empty)
        return empty

    odps = get_odps_client()
    batches = [pending[i : i + chunk] for i in range(0, len(pending), chunk)]
    total_odps_rows = 0
    total_inserted = 0
    total_skipped = 0
    total_missing = 0

    for idx, batch_vids in enumerate(batches, start=1):
        stats = _sync_one_batch(odps, resolved_decode_dt, batch_vids, idx, len(batches))
        total_odps_rows += stats["odps_rows"]
        total_inserted += stats["inserted"]
        total_skipped += stats["skipped_no_topic"]
        total_missing += stats["missing_in_odps"]

    next_offset = start + len(pending)
    result = {
        "decode_dt": resolved_decode_dt,
        "pool_video_lists": len(video_lists),
        "unique_vids": len(all_vids),
        "pending_total": len(pending_all),
        "offset": start,
        "batch_size": chunk,
        "batches": len(batches),
        "processed": len(pending),
        "odps_rows": total_odps_rows,
        "inserted": total_inserted,
        "skipped_no_topic": total_skipped,
        "missing_in_odps": total_missing,
        "next_offset": next_offset,
        "remaining": max(0, len(pending_all) - next_offset),
    }
    logger.info("Multi demand video sync completed: %s", result)
    return result


def _backfill_urls_one_batch(
    odps: ODPSClient,
    decode_dt: str,
    batch_vids: list[str],
    batch_idx: int,
    batch_total: int,
) -> dict[str, int]:
    """查询一批 vid 的 url1/url2，立刻更新 MySQL。"""
    logger.info(
        "Backfill batch %d/%d: query %d vids, decode_dt=%s",
        batch_idx,
        batch_total,
        len(batch_vids),
        decode_dt,
    )
    odps_rows = odps.fetch_topic_decode_results(
        decode_dt, batch_vids, batch_size=len(batch_vids)
    )

    urls_by_vid: dict[str, dict[str, str | None]] = {}
    seen_vids: set[str] = set()
    for row in odps_rows:
        raw_vid = row.get("vid")
        if raw_vid is None:
            continue
        vid = str(raw_vid).strip()
        if not vid or vid in seen_vids:
            continue
        url1 = _extract_url(row.get("url1"))
        url2 = _extract_url(row.get("url2"))
        if url1 is None and url2 is None:
            continue
        seen_vids.add(vid)
        urls_by_vid[vid] = {"url1": url1, "url2": url2}

    with get_session() as session:
        updated = MultiDemandVideoDetailRepository(session).update_urls(urls_by_vid)

    odps_vids = {
        str(r.get("vid")).strip()
        for r in odps_rows
        if r.get("vid") is not None and str(r.get("vid")).strip()
    }
    stats = {
        "odps_rows": len(odps_rows),
        "matched": len(urls_by_vid),
        "updated": updated,
        "missing_in_odps": len(set(batch_vids) - odps_vids),
        "no_urls_in_odps": len(odps_rows) - len(urls_by_vid),
    }
    logger.info("Backfill batch %d/%d done: %s", batch_idx, batch_total, stats)
    return stats


def backfill_multi_demand_video_urls(
    limit: int | None = None,
    offset: int = 0,
    batch_size: int = VIDEO_SYNC_BATCH_SIZE,
    *,
    decode_dt: str | None = None,
) -> dict[str, Any]:
    """
    回填已有 multi_demand_video_detail 行的 url1/url2。

    - 目标：url1 或 url2 为空的 vid
    - 数据来源：ODPS dwd_topic_decode_result_di 表字段（与 vid 同级）
    - decode_dt 默认「今天的昨天」
    """
    resolved_decode_dt = decode_dt or (
        datetime.now() - timedelta(days=1)
    ).strftime("%Y%m%d")
    start = max(0, int(offset))
    chunk = max(1, int(batch_size))
    logger.info(
        "Starting multi demand video url backfill: decode_dt=%s limit=%s offset=%d batch_size=%d",
        resolved_decode_dt,
        limit,
        start,
        chunk,
    )

    with get_session() as session:
        pending_all = sorted(
            MultiDemandVideoDetailRepository(session).list_vids_missing_urls()
        )

    pending = pending_all[start:]
    if limit is not None:
        pending = pending[: max(0, int(limit))]

    logger.info(
        "Url backfill vids: pending_total=%d to_process=%d",
        len(pending_all),
        len(pending),
    )

    empty = {
        "decode_dt": resolved_decode_dt,
        "pending_total": len(pending_all),
        "offset": start,
        "batch_size": chunk,
        "batches": 0,
        "processed": 0,
        "odps_rows": 0,
        "matched": 0,
        "updated": 0,
        "missing_in_odps": 0,
        "no_urls_in_odps": 0,
        "next_offset": start,
        "remaining": max(0, len(pending_all) - start),
    }
    if not pending:
        logger.info("No pending vids for url backfill: %s", empty)
        return empty

    odps = get_odps_client()
    batches = [pending[i : i + chunk] for i in range(0, len(pending), chunk)]
    total_odps_rows = 0
    total_matched = 0
    total_updated = 0
    total_missing = 0
    total_no_urls = 0

    for idx, batch_vids in enumerate(batches, start=1):
        stats = _backfill_urls_one_batch(
            odps, resolved_decode_dt, batch_vids, idx, len(batches)
        )
        total_odps_rows += stats["odps_rows"]
        total_matched += stats["matched"]
        total_updated += stats["updated"]
        total_missing += stats["missing_in_odps"]
        total_no_urls += stats["no_urls_in_odps"]

    next_offset = start + len(pending)
    result = {
        "decode_dt": resolved_decode_dt,
        "pending_total": len(pending_all),
        "offset": start,
        "batch_size": chunk,
        "batches": len(batches),
        "processed": len(pending),
        "odps_rows": total_odps_rows,
        "matched": total_matched,
        "updated": total_updated,
        "missing_in_odps": total_missing,
        "no_urls_in_odps": total_no_urls,
        "next_offset": next_offset,
        "remaining": max(0, len(pending_all) - next_offset),
    }
    logger.info("Multi demand video url backfill completed: %s", result)
    return result