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 hashlib
import json
import logging
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any

from supply_infra.category_match import (
    CategoryMatchClient,
    get_category_match_client,
)
from supply_infra.config import get_infra_settings
from supply_infra.db.repositories.demand_belong_category_repo import (
    DemandBelongCategoryRepository,
)
from supply_infra.db.repositories.demand_belong_pool_rel_repo import (
    DemandBelongPoolRelRepository,
)
from supply_infra.db.repositories.demand_popularity_stats_repo import (
    DemandPopularityStatsRepository,
)
from supply_infra.db.repositories.global_tree_category_repo import (
    GlobalTreeCategoryRepository,
)
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__)

_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"])


_SOURCE_COMPARE_FIELDS = (
    "strategy",
    "demand_id",
    "demand_name",
    "weight",
    "type",
    "video_count",
    "video_list",
    "extend",
    "biz_dt",
)


def _source_signature(row: dict[str, Any]) -> tuple[Any, ...]:
    """生成稳定的源字段比较值；排除数据库 id 和下游回填字段。"""
    return tuple(row.get(field) for field in _SOURCE_COMPARE_FIELDS)


def _source_snapshot_hash(rows: list[dict[str, Any]]) -> str:
    """生成与行顺序无关的内容哈希，供运行审计和水位比较。"""
    canonical = [
        {field: row.get(field) for field in _SOURCE_COMPARE_FIELDS}
        for row in sorted(rows, key=_row_key)
    ]
    payload = json.dumps(
        canonical,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


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 _classify_words(
    biz_dt: str,
    *,
    client: CategoryMatchClient | None = None,
    workers: int | None = None,
) -> dict:
    """逐词调用分类路径接口，将 stable_id 映射到本地分类后写入。"""
    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
        source_id_to_mysql_id = GlobalTreeCategoryRepository(
            session
        ).get_active_source_id_map()

    word_list = sorted(pending)
    matcher = client or get_category_match_client()
    worker_count = min(
        max(1, workers or get_infra_settings().category_match_workers),
        max(1, len(word_list)),
    )

    logger.info(
        "Category match prepare: demand_names=%d words=%d existing=%d "
        "pending=%d workers=%d",
        len(demand_names),
        len(word_set),
        len(existing),
        len(pending),
        worker_count,
    )

    rows: list[dict[str, Any]] = []
    unmatched_terms: list[str] = []
    failed_items: list[dict[str, Any]] = []

    def match_term(item: tuple[int, str]) -> tuple[str, Any, Exception | None]:
        idx, term = item
        logger.info("Matching category %d/%d: %s", idx, len(word_list), term)
        try:
            return term, matcher.match_one(term, description=""), None
        except Exception as exc:
            logger.exception(
                "Category match failed; continue remaining items: "
                "biz_dt=%s item=%s/%s term=%s",
                biz_dt,
                idx,
                len(word_list),
                term,
            )
            return term, None, exc

    indexed_terms = list(enumerate(word_list, start=1))
    with ThreadPoolExecutor(
        max_workers=worker_count,
        thread_name_prefix="category-match",
    ) as executor:
        outcomes = list(executor.map(match_term, indexed_terms))

    for term, match, error_value in outcomes:
        if error_value is not None:
            failed_items.append(
                {
                    "term": term,
                    "error": str(error_value),
                }
            )
            continue
        if match is None:
            unmatched_terms.append(term)
            continue

        category_id = source_id_to_mysql_id.get(match.stable_id)
        if category_id is None:
            error = (
                f"matched stable_id={match.stable_id} is absent from "
                "global_tree_category"
            )
            logger.error("Category match cannot be persisted: term=%s %s", term, error)
            failed_items.append({"term": term, "error": error})
            continue
        rows.append(
            {
                "name": term,
                "category_id": category_id,
                "reason": match.reason,
                "is_delete": 0,
            }
        )

    with get_session() as session:
        persisted = DemandBelongCategoryRepository(
            session
        ).upsert_category_matches(rows)

    result = {
        "success": not failed_items,
        "demand_names": len(demand_names),
        "words": len(word_set),
        "existing_filtered": len(existing),
        "pending": len(pending),
        "api_calls": len(word_list),
        "matched": len(rows),
        "persisted": persisted,
        "unmatched": len(unmatched_terms),
        "unmatched_terms": unmatched_terms,
        "failed_items": failed_items,
    }
    if failed_items:
        result["error_code"] = "category_match_failed_items"
        result["error"] = f"{len(failed_items)} category match item(s) failed"
    return result


def _sync_diff(partition_date: str) -> dict[str, Any]:
    """按主键和完整源内容同步；行数相同也会识别修订。"""
    odps = get_odps_client()
    raw_rows = odps.fetch_multi_demand_pool(partition_date)
    mysql_rows = _to_mysql_rows(raw_rows, partition_date)
    odps_by_key = {_row_key(r): r for r in mysql_rows}
    odps_keys = set(odps_by_key)

    with get_session() as session:
        repo = MultiDemandPoolDiRepository(session)
        existing_rows = repo.list_source_rows_by_biz_dt(partition_date)
        mysql_by_key = {_row_key(row): row for row in existing_rows}
        mysql_keys = set(mysql_by_key)

        to_insert_keys = odps_keys - mysql_keys
        to_delete_keys = mysql_keys - odps_keys
        shared_keys = odps_keys & mysql_keys
        to_update_keys = {
            key
            for key in shared_keys
            if _source_signature(odps_by_key[key])
            != _source_signature(mysql_by_key[key])
        }

        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_pool_ids = [mysql_by_key[key]["id"] for key in to_delete_keys]
        deleted_relations = DemandBelongPoolRelRepository(session).delete_by_pool_ids(
            deleted_pool_ids
        )
        deleted = repo.delete_by_keys(partition_date, list(to_delete_keys))
        inserted = repo.bulk_insert(insert_rows)
        updated = (
            repo.update_source_fields(partition_date, update_rows)
            if update_rows
            else 0
        )

    logger.info(
        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d update=%d unchanged=%d",
        len(odps_keys),
        len(mysql_keys),
        inserted,
        deleted,
        updated,
        len(shared_keys) - len(to_update_keys),
    )
    return {
        "fetched": len(raw_rows),
        "odps_unique": len(odps_keys),
        "mysql_before": len(mysql_keys),
        "inserted": inserted,
        "deleted": deleted,
        "relations_deleted": deleted_relations,
        "updated": updated,
        "unchanged": len(shared_keys) - len(to_update_keys),
        "source_snapshot_hash": _source_snapshot_hash(mysql_rows),
        "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]:
    """同步当天需求池主数据；内容哈希差分替代不可靠的行数门禁。"""
    return _sync_diff(partition_date)