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

"""
Demand pool ODPS → MySQL sync stages used by the durable pipeline.

Stages owned by this module:
1. source row sync (`_sync_pool_rows`)
2. demand word classification (`_classify_words`)
3. real rov/vov enrichment (`enrich_real_rov_vov_7d`)
4. popularity stats (`compute_popularity_stats`)

Related pipeline stages live in sibling modules (belong_rel / tree_weight / videos).
"""
from __future__ import annotations

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

from agents.demand_belong_category_agent.run import main as classify_demand_words
from supply_infra.db.repositories.demand_belong_category_repo import (
    DemandBelongCategoryRepository,
)
from supply_infra.db.repositories.demand_popularity_stats_repo import (
    DemandPopularityStatsRepository,
)
from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
from supply_infra.db.session import get_session
from supply_infra.odps.client import get_odps_client

logger = logging.getLogger(__name__)

_WORD_BATCH_SIZE = 50
_STATS_UPSERT_BATCH = 200
_REAL_METRIC_LIMIT = 1000
_REAL_METRIC_LOOKBACK_DAYS = 7
_GLOBAL_FEATURE_VALUE = "全局SUM"
_VIDEO_LIST_LIMIT = 10

# 策略名 → 统计字段前缀；去年同期阳历/阴历合并为 plat_ly_pop
_STRATEGY_METRIC: dict[str, str] = {
    "新热事件": "ext_pop",
    "逐月": "plat_sust_pop",
    "去年同期阳历": "plat_ly_pop",
    "去年同期阴历": "plat_ly_pop",
    "当下供需gap": "recent_pop",
}
_TARGET_STRATEGIES = list(_STRATEGY_METRIC.keys())
_METRIC_KEYS = ("ext_pop", "plat_sust_pop", "plat_ly_pop", "recent_pop")
_REAL_METRIC_KEYS = ("real_rov_7d", "real_vov_7d")
_ALL_METRIC_KEYS = _METRIC_KEYS + _REAL_METRIC_KEYS
_METRIC_DECIMAL_PLACES: dict[str, int] = {
    "ext_pop": 2,
    "plat_sust_pop": 2,
    "plat_ly_pop": 2,
    "recent_pop": 2,
    "real_rov_7d": 4,
    "real_vov_7d": 4,
}

RowKey = tuple[str, str]


def _row_key(row: dict[str, Any]) -> RowKey:
    return (row["strategy"], row["demand_id"])


def _normalize_video_list(raw: Any) -> tuple[str | None, int]:
    """取前 N 个 video_id，返回 (JSON 文本, count)，二者保持一致。"""
    if raw is None:
        return None, 0

    items: list[Any]
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return None, 0
        try:
            parsed = json.loads(text)
            items = list(parsed) if isinstance(parsed, list) else [text]
        except json.JSONDecodeError:
            items = [part.strip() for part in text.split(",") if part.strip()]
    elif isinstance(raw, (list, tuple)):
        items = list(raw)
    else:
        try:
            items = list(raw)
        except TypeError:
            return None, 0

    truncated = [
        str(v).strip()
        for v in items[:_VIDEO_LIST_LIMIT]
        if v is not None and str(v).strip()
    ]
    if not truncated:
        return None, 0
    return json.dumps(truncated, ensure_ascii=False), len(truncated)


def _normalize_weight(raw: Any) -> float | None:
    """weight 为 0 或无效时视为无分数，存 null。"""
    if raw is None:
        return None
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return None
    return None if value == 0 else value


def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str, Any]]:
    """转换为 MySQL 行，按 (strategy, demand_id) 去重（保留最后一条）。"""
    by_key: dict[RowKey, dict[str, Any]] = {}
    for row in raw_rows:
        strategy = row.get("strategy")
        demand_id = row.get("demand_id")
        demand_name = row.get("demand_name")
        if strategy is None or demand_id is None or demand_name is None:
            logger.warning("Skip row with missing required fields: %s", row)
            continue

        video_list, video_count = _normalize_video_list(row.get("video_list"))
        mapped = {
            "strategy": str(strategy),
            "demand_id": str(demand_id),
            "demand_name": str(demand_name),
            "weight": _normalize_weight(row.get("weight")),
            "type": str(row["type"]) if row.get("type") is not None else None,
            "video_count": video_count,
            "video_list": video_list,
            "extend": str(row["extend"]) if row.get("extend") is not None else None,
            "biz_dt": biz_dt,
        }
        by_key[_row_key(mapped)] = mapped
    return list(by_key.values())


def _build_word_set(demand_names: list[str]) -> set[str]:
    """demand_name 按空格分词，全部落入同一个 set。"""
    words: set[str] = set()
    for name in demand_names:
        for token in str(name).split():
            word = token.strip()
            if word:
                words.add(word)
    return words


def _chunked(items: list[str], size: int) -> list[list[str]]:
    return [items[i : i + size] for i in range(0, len(items), size)]


def _classify_words(biz_dt: str) -> dict:
    """查询 demand_name → 空格分词入 set → 过滤已有词 → 100 词一批调用 agent。"""
    with get_session() as session:
        demand_names = MultiDemandPoolDiRepository(session).list_demand_names_by_biz_dt(biz_dt)
        word_set = _build_word_set(demand_names)
        existing = DemandBelongCategoryRepository(session).get_existing_names(word_set)
        pending = word_set - existing

    word_list = list(pending)
    batches = _chunked(word_list, _WORD_BATCH_SIZE)

    logger.info(
        "Classify prepare: demand_names=%d words=%d existing=%d pending=%d batches=%d",
        len(demand_names),
        len(word_set),
        len(existing),
        len(pending),
        len(batches),
    )

    failed_batches: list[dict[str, Any]] = []
    for idx, batch in enumerate(batches, start=1):
        logger.info("Classifying batch %d/%d (%d words)", idx, len(batches), len(batch))
        try:
            classify_demand_words(batch)
        except Exception as exc:
            logger.exception(
                "Demand belong classification batch failed; continue remaining batches: "
                "biz_dt=%s batch=%s/%s",
                biz_dt,
                idx,
                len(batches),
            )
            failed_batches.append(
                {
                    "batch": idx,
                    "size": len(batch),
                    "error": str(exc),
                }
            )

    return {
        "success": not failed_batches,
        "demand_names": len(demand_names),
        "words": len(word_set),
        "existing_filtered": len(existing),
        "pending": len(pending),
        "batches": len(batches),
        "failed_batches": failed_batches,
    }


def _sync_diff(partition_date: str) -> dict[str, Any]:
    """行数不同时拉取 ODPS，只同步 (strategy, demand_id) 差异，并回填已有行的 video/weight 字段。"""
    odps = get_odps_client()
    raw_rows = odps.fetch_multi_demand_pool(partition_date)
    mysql_rows = _to_mysql_rows(raw_rows, partition_date)
    odps_keys = {_row_key(r) for r in mysql_rows}
    odps_by_key = {_row_key(r): r for r in mysql_rows}

    with get_session() as session:
        repo = MultiDemandPoolDiRepository(session)
        mysql_keys = repo.list_keys_by_biz_dt(partition_date)

        to_insert_keys = odps_keys - mysql_keys
        to_delete_keys = mysql_keys - odps_keys
        to_update_keys = odps_keys & mysql_keys

        insert_rows = [odps_by_key[k] for k in to_insert_keys]
        update_rows = [odps_by_key[k] for k in to_update_keys]
        deleted = repo.delete_by_keys(partition_date, list(to_delete_keys))
        inserted = repo.bulk_insert(insert_rows)
        updated = repo.update_video_fields(partition_date, update_rows)

    logger.info(
        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d video_update=%d",
        len(odps_keys),
        len(mysql_keys),
        inserted,
        deleted,
        updated,
    )
    return {
        "fetched": len(raw_rows),
        "odps_unique": len(odps_keys),
        "mysql_before": len(mysql_keys),
        "inserted": inserted,
        "deleted": deleted,
        "video_updated": updated,
        "skipped_invalid": len(raw_rows) - len(mysql_rows),
    }


def _to_float(value: Any) -> float | None:
    if value is None:
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _pick_better_real_metric(
    current: dict[str, Any] | None,
    candidate: dict[str, Any],
) -> dict[str, Any]:
    """相同特征值时保留 rov_diff、vov_diff 更高的一条。"""
    if current is None:
        return candidate

    current_key = (
        _to_float(current.get("rov_diff")) or 0.0,
        _to_float(current.get("vov_diff")) or 0.0,
    )
    candidate_key = (
        _to_float(candidate.get("rov_diff")) or 0.0,
        _to_float(candidate.get("vov_diff")) or 0.0,
    )
    return candidate if candidate_key > current_key else current


def _build_real_metrics_by_feature(
    raw_rows: list[dict[str, Any]],
) -> dict[str, tuple[float | None, float | None]]:
    """按特征值去重，映射为 demand_name → (real_rov_7d, real_vov_7d)。"""
    best_by_feature: dict[str, dict[str, Any]] = {}
    for row in raw_rows:
        feature = row.get("特征值")
        if feature is None:
            continue
        feature_name = str(feature).strip()
        if not feature_name or feature_name == _GLOBAL_FEATURE_VALUE:
            continue
        best_by_feature[feature_name] = _pick_better_real_metric(
            best_by_feature.get(feature_name),
            row,
        )

    return {
        feature: (_to_float(row.get("rov_diff")), _to_float(row.get("vov_diff")))
        for feature, row in best_by_feature.items()
    }


def enrich_real_rov_vov_7d(biz_dt: str) -> dict[str, Any]:
    """
    从 ODPS 拉取执行日及往前 7 日的 rov_diff/vov_diff，按特征值匹配回填 MySQL。

    相同特征值保留 rov_diff、vov_diff 更高的记录。
    """
    dt_right = biz_dt
    dt_left = (
        datetime.strptime(biz_dt, "%Y%m%d") - timedelta(days=_REAL_METRIC_LOOKBACK_DAYS)
    ).strftime("%Y%m%d")

    logger.info(
        "Enrich real rov/vov: biz_dt=%s range=%s~%s limit=%d",
        biz_dt,
        dt_left,
        dt_right,
        _REAL_METRIC_LIMIT,
    )

    odps = get_odps_client()
    raw_rows = odps.fetch_real_rov_vov_7d(
        dt_left=dt_left,
        dt_right=dt_right,
        limit=_REAL_METRIC_LIMIT,
    )
    metrics_by_name = _build_real_metrics_by_feature(raw_rows)

    with get_session() as session:
        updated = MultiDemandPoolDiRepository(session).update_real_metrics_by_demand_name(
            biz_dt,
            metrics_by_name,
        )

    result = {
        "dt_left": dt_left,
        "dt_right": dt_right,
        "odps_rows": len(raw_rows),
        "unique_features": len(metrics_by_name),
        "updated_rows": updated,
    }
    logger.info("Enrich real rov/vov completed: %s", result)
    return result


def _calc_metric_stats(
    weights: list[float],
    places: int = 2,
) -> tuple[Decimal | None, int]:
    """
    计算单策略 avg / count。
    权重为 0 不参与平均值；全部为 0（或无有效权重）时 avg=null、count=0。
    """
    nonzero = [w for w in weights if w != 0]
    if not nonzero:
        return None, 0

    avg_val = sum(nonzero) / len(nonzero)
    q = Decimal("0." + "0" * places)
    return (
        Decimal(str(round(avg_val, places))).quantize(q),
        len(nonzero),
    )


def _empty_metric_stats() -> dict[str, Decimal | int | None]:
    result: dict[str, Decimal | int | None] = {}
    for key in _ALL_METRIC_KEYS:
        result[f"{key}_avg"] = None
        result[f"{key}_count"] = 0
    return result


def _build_stats_row(
    demand_category_id: int,
    demand_word_name: str,
    biz_dt: str,
    strategy_weights: list[tuple[str, float | None]],
    real_metrics: list[tuple[float | None, float | None]],
) -> dict[str, Any]:
    """按策略分组后汇总为 demand_popularity_stats 一行（含 rov_diff/vov_diff）。"""
    grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
    for strategy, weight in strategy_weights:
        metric = _STRATEGY_METRIC.get(strategy)
        if metric is None or weight is None or weight == 0:
            continue
        grouped[metric].append(float(weight))

    rov_values: list[float] = []
    vov_values: list[float] = []
    for rov, vov in real_metrics:
        if rov is not None:
            rov_values.append(float(rov))
        if vov is not None:
            vov_values.append(float(vov))

    row: dict[str, Any] = {
        "demand_category_id": demand_category_id,
        "demand_word_name": demand_word_name[:128],
        "biz_dt": biz_dt,
        **_empty_metric_stats(),
    }
    for key in _METRIC_KEYS:
        avg_val, count = _calc_metric_stats(
            grouped[key], places=_METRIC_DECIMAL_PLACES[key]
        )
        row[f"{key}_avg"] = avg_val
        row[f"{key}_count"] = count

    rov_avg, rov_count = _calc_metric_stats(
        rov_values, places=_METRIC_DECIMAL_PLACES["real_rov_7d"]
    )
    vov_avg, vov_count = _calc_metric_stats(
        vov_values, places=_METRIC_DECIMAL_PLACES["real_vov_7d"]
    )
    row["real_rov_7d_avg"] = rov_avg
    row["real_rov_7d_count"] = rov_count
    row["real_vov_7d_avg"] = vov_avg
    row["real_vov_7d_count"] = vov_count
    return row


def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
    """
    遍历 demand_belong_category 全部词，各自 LIKE 查询当天指定策略权重，
    并汇总 rov_diff/vov_diff，写入 demand_popularity_stats（avg/count）。

    Args:
        biz_dt: 业务日期 (YYYYMMDD)
    """
    with get_session() as session:
        categories = DemandBelongCategoryRepository(session).list_active_id_name()

    if not categories:
        logger.info("Popularity stats: no demand_belong_category rows, skip")
        return {"words": 0, "upserted": 0}

    logger.info("Popularity stats: processing %d words for biz_dt=%s", len(categories), biz_dt)

    rows: list[dict[str, Any]] = []
    upserted = 0

    with get_session() as session:
        pool_repo = MultiDemandPoolDiRepository(session)
        stats_repo = DemandPopularityStatsRepository(session)

        for idx, (category_id, name) in enumerate(categories, start=1):
            matches = pool_repo.list_weights_by_name_like(
                biz_dt, name, _TARGET_STRATEGIES
            )
            real_matches = pool_repo.list_real_metrics_by_name_like(biz_dt, name)
            rows.append(
                _build_stats_row(category_id, name, biz_dt, matches, real_matches)
            )

            if len(rows) >= _STATS_UPSERT_BATCH:
                upserted += stats_repo.upsert_rows(rows)
                logger.info(
                    "Popularity stats progress: %d/%d words, batch upserted",
                    idx,
                    len(categories),
                )
                rows = []

        if rows:
            upserted += stats_repo.upsert_rows(rows)

    result = {"words": len(categories), "upserted": upserted}
    logger.info("Popularity stats completed: %s", result)
    return result


def _sync_pool_rows(partition_date: str) -> dict[str, Any]:
    """同步当天需求池主数据；供完整任务按独立阶段捕获异常。"""
    odps = get_odps_client()
    odps_count = odps.count_multi_demand_pool(partition_date)
    with get_session() as session:
        mysql_count = MultiDemandPoolDiRepository(session).count_by_biz_dt(partition_date)

    logger.info("Count check: odps=%d mysql=%d", odps_count, mysql_count)

    if odps_count == mysql_count:
        logger.info("Same count, skip ODPS data sync")
        return {
            "skipped_same_count": True,
            "odps_count": odps_count,
            "mysql_count": mysql_count,
            "fetched": 0,
            "inserted": 0,
            "deleted": 0,
        }
    return {
        "skipped_same_count": False,
        "odps_count": odps_count,
        "mysql_count": mysql_count,
        **_sync_diff(partition_date),
    }