|
@@ -21,7 +21,7 @@ import json
|
|
|
import logging
|
|
import logging
|
|
|
from datetime import datetime, timedelta
|
|
from datetime import datetime, timedelta
|
|
|
from decimal import Decimal
|
|
from decimal import Decimal
|
|
|
-from typing import Any
|
|
|
|
|
|
|
+from typing import Any, Callable
|
|
|
|
|
|
|
|
from agents.demand_belong_category_agent.run import main as classify_demand_words
|
|
from agents.demand_belong_category_agent.run import main as classify_demand_words
|
|
|
from supply_infra.db.repositories.demand_belong_category_repo import (
|
|
from supply_infra.db.repositories.demand_belong_category_repo import (
|
|
@@ -111,6 +111,17 @@ def _normalize_video_list(raw: Any) -> tuple[str | None, int]:
|
|
|
return json.dumps(truncated, ensure_ascii=False), len(truncated)
|
|
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]]:
|
|
def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str, Any]]:
|
|
|
"""转换为 MySQL 行,按 (strategy, demand_id) 去重(保留最后一条)。"""
|
|
"""转换为 MySQL 行,按 (strategy, demand_id) 去重(保留最后一条)。"""
|
|
|
by_key: dict[RowKey, dict[str, Any]] = {}
|
|
by_key: dict[RowKey, dict[str, Any]] = {}
|
|
@@ -127,7 +138,7 @@ def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str
|
|
|
"strategy": str(strategy),
|
|
"strategy": str(strategy),
|
|
|
"demand_id": str(demand_id),
|
|
"demand_id": str(demand_id),
|
|
|
"demand_name": str(demand_name),
|
|
"demand_name": str(demand_name),
|
|
|
- "weight": row.get("weight"),
|
|
|
|
|
|
|
+ "weight": _normalize_weight(row.get("weight")),
|
|
|
"type": str(row["type"]) if row.get("type") is not None else None,
|
|
"type": str(row["type"]) if row.get("type") is not None else None,
|
|
|
"video_count": video_count,
|
|
"video_count": video_count,
|
|
|
"video_list": video_list,
|
|
"video_list": video_list,
|
|
@@ -173,21 +184,40 @@ def _classify_words(biz_dt: str) -> dict:
|
|
|
len(batches),
|
|
len(batches),
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
+ failed_batches: list[dict[str, Any]] = []
|
|
|
for idx, batch in enumerate(batches, start=1):
|
|
for idx, batch in enumerate(batches, start=1):
|
|
|
logger.info("Classifying batch %d/%d (%d words)", idx, len(batches), len(batch))
|
|
logger.info("Classifying batch %d/%d (%d words)", idx, len(batches), len(batch))
|
|
|
- classify_demand_words(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 {
|
|
return {
|
|
|
|
|
+ "success": not failed_batches,
|
|
|
"demand_names": len(demand_names),
|
|
"demand_names": len(demand_names),
|
|
|
"words": len(word_set),
|
|
"words": len(word_set),
|
|
|
"existing_filtered": len(existing),
|
|
"existing_filtered": len(existing),
|
|
|
"pending": len(pending),
|
|
"pending": len(pending),
|
|
|
"batches": len(batches),
|
|
"batches": len(batches),
|
|
|
|
|
+ "failed_batches": failed_batches,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
def _sync_diff(partition_date: str) -> dict[str, Any]:
|
|
def _sync_diff(partition_date: str) -> dict[str, Any]:
|
|
|
- """行数不同时拉取 ODPS,只同步 (strategy, demand_id) 差异,并回填已有行的 video 字段。"""
|
|
|
|
|
|
|
+ """行数不同时拉取 ODPS,只同步 (strategy, demand_id) 差异,并回填已有行的 video/weight 字段。"""
|
|
|
odps = get_odps_client()
|
|
odps = get_odps_client()
|
|
|
raw_rows = odps.fetch_multi_demand_pool(partition_date)
|
|
raw_rows = odps.fetch_multi_demand_pool(partition_date)
|
|
|
mysql_rows = _to_mysql_rows(raw_rows, partition_date)
|
|
mysql_rows = _to_mysql_rows(raw_rows, partition_date)
|
|
@@ -227,30 +257,6 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
-def backfill_video_list(partition_date: str) -> dict[str, Any]:
|
|
|
|
|
- """从 ODPS 回填指定分区的 video_list / video_count(每条最多前 10 个 video_id)。"""
|
|
|
|
|
- logger.info("Backfill video_list for partition: %s", partition_date)
|
|
|
|
|
- odps = get_odps_client()
|
|
|
|
|
- raw_rows = odps.fetch_multi_demand_pool(partition_date)
|
|
|
|
|
- mysql_rows = _to_mysql_rows(raw_rows, partition_date)
|
|
|
|
|
-
|
|
|
|
|
- with get_session() as session:
|
|
|
|
|
- updated = MultiDemandPoolDiRepository(session).update_video_fields(
|
|
|
|
|
- partition_date,
|
|
|
|
|
- mysql_rows,
|
|
|
|
|
- )
|
|
|
|
|
-
|
|
|
|
|
- result = {
|
|
|
|
|
- "partition_date": partition_date,
|
|
|
|
|
- "fetched": len(raw_rows),
|
|
|
|
|
- "unique_rows": len(mysql_rows),
|
|
|
|
|
- "updated": updated,
|
|
|
|
|
- "with_video": sum(1 for r in mysql_rows if r.get("video_list")),
|
|
|
|
|
- }
|
|
|
|
|
- logger.info("Backfill video_list completed: %s", result)
|
|
|
|
|
- return result
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
def _to_float(value: Any) -> float | None:
|
|
def _to_float(value: Any) -> float | None:
|
|
|
if value is None:
|
|
if value is None:
|
|
|
return None
|
|
return None
|
|
@@ -349,15 +355,14 @@ def enrich_real_rov_vov_7d(biz_dt: str) -> dict[str, Any]:
|
|
|
def _calc_metric_stats(
|
|
def _calc_metric_stats(
|
|
|
weights: list[float],
|
|
weights: list[float],
|
|
|
places: int = 2,
|
|
places: int = 2,
|
|
|
-) -> tuple[Decimal, int]:
|
|
|
|
|
|
|
+) -> tuple[Decimal | None, int]:
|
|
|
"""
|
|
"""
|
|
|
计算单策略 avg / count。
|
|
计算单策略 avg / count。
|
|
|
- 权重为 0 不参与平均值;全部为 0(或无有效权重)时 avg=0、count=0。
|
|
|
|
|
|
|
+ 权重为 0 不参与平均值;全部为 0(或无有效权重)时 avg=null、count=0。
|
|
|
"""
|
|
"""
|
|
|
nonzero = [w for w in weights if w != 0]
|
|
nonzero = [w for w in weights if w != 0]
|
|
|
if not nonzero:
|
|
if not nonzero:
|
|
|
- zero = Decimal("0").quantize(Decimal("0." + "0" * places))
|
|
|
|
|
- return zero, 0
|
|
|
|
|
|
|
+ return None, 0
|
|
|
|
|
|
|
|
avg_val = sum(nonzero) / len(nonzero)
|
|
avg_val = sum(nonzero) / len(nonzero)
|
|
|
q = Decimal("0." + "0" * places)
|
|
q = Decimal("0." + "0" * places)
|
|
@@ -367,12 +372,10 @@ def _calc_metric_stats(
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
-def _empty_metric_stats() -> dict[str, Decimal | int]:
|
|
|
|
|
- result: dict[str, Decimal | int] = {}
|
|
|
|
|
|
|
+def _empty_metric_stats() -> dict[str, Decimal | int | None]:
|
|
|
|
|
+ result: dict[str, Decimal | int | None] = {}
|
|
|
for key in _ALL_METRIC_KEYS:
|
|
for key in _ALL_METRIC_KEYS:
|
|
|
- places = _METRIC_DECIMAL_PLACES[key]
|
|
|
|
|
- zero = Decimal("0").quantize(Decimal("0." + "0" * places))
|
|
|
|
|
- result[f"{key}_avg"] = zero
|
|
|
|
|
|
|
+ result[f"{key}_avg"] = None
|
|
|
result[f"{key}_count"] = 0
|
|
result[f"{key}_count"] = 0
|
|
|
return result
|
|
return result
|
|
|
|
|
|
|
@@ -388,7 +391,7 @@ def _build_stats_row(
|
|
|
grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
|
|
grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
|
|
|
for strategy, weight in strategy_weights:
|
|
for strategy, weight in strategy_weights:
|
|
|
metric = _STRATEGY_METRIC.get(strategy)
|
|
metric = _STRATEGY_METRIC.get(strategy)
|
|
|
- if metric is None or weight is None:
|
|
|
|
|
|
|
+ if metric is None or weight is None or weight == 0:
|
|
|
continue
|
|
continue
|
|
|
grouped[metric].append(float(weight))
|
|
grouped[metric].append(float(weight))
|
|
|
|
|
|
|
@@ -476,21 +479,8 @@ def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
|
|
|
return result
|
|
return result
|
|
|
|
|
|
|
|
|
|
|
|
|
-def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> dict:
|
|
|
|
|
- """
|
|
|
|
|
- 从 ODPS 增量同步策略需求天级数据到 MySQL,并对新词做归属分类与热度统计。
|
|
|
|
|
-
|
|
|
|
|
- Args:
|
|
|
|
|
- partition_date: 分区日期 (YYYYMMDD),默认当天
|
|
|
|
|
- """
|
|
|
|
|
- if partition_date is None:
|
|
|
|
|
- partition_date = datetime.now().strftime("%Y%m%d")
|
|
|
|
|
-
|
|
|
|
|
- logger.info(
|
|
|
|
|
- "Starting multi demand pool ODPS → MySQL sync for partition: %s",
|
|
|
|
|
- partition_date,
|
|
|
|
|
- )
|
|
|
|
|
-
|
|
|
|
|
|
|
+def _sync_pool_rows(partition_date: str) -> dict[str, Any]:
|
|
|
|
|
+ """同步当天需求池主数据;供完整任务按独立阶段捕获异常。"""
|
|
|
odps = get_odps_client()
|
|
odps = get_odps_client()
|
|
|
odps_count = odps.count_multi_demand_pool(partition_date)
|
|
odps_count = odps.count_multi_demand_pool(partition_date)
|
|
|
with get_session() as session:
|
|
with get_session() as session:
|
|
@@ -499,7 +489,8 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
|
|
|
logger.info("Count check: odps=%d mysql=%d", odps_count, mysql_count)
|
|
logger.info("Count check: odps=%d mysql=%d", odps_count, mysql_count)
|
|
|
|
|
|
|
|
if odps_count == mysql_count:
|
|
if odps_count == mysql_count:
|
|
|
- sync_stats: dict[str, Any] = {
|
|
|
|
|
|
|
+ logger.info("Same count, skip ODPS data sync")
|
|
|
|
|
+ return {
|
|
|
"skipped_same_count": True,
|
|
"skipped_same_count": True,
|
|
|
"odps_count": odps_count,
|
|
"odps_count": odps_count,
|
|
|
"mysql_count": mysql_count,
|
|
"mysql_count": mysql_count,
|
|
@@ -507,31 +498,82 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
|
|
|
"inserted": 0,
|
|
"inserted": 0,
|
|
|
"deleted": 0,
|
|
"deleted": 0,
|
|
|
}
|
|
}
|
|
|
- logger.info("Same count, skip ODPS data sync")
|
|
|
|
|
- else:
|
|
|
|
|
- sync_stats = {
|
|
|
|
|
- "skipped_same_count": False,
|
|
|
|
|
- "odps_count": odps_count,
|
|
|
|
|
- "mysql_count": mysql_count,
|
|
|
|
|
- **_sync_diff(partition_date),
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ return {
|
|
|
|
|
+ "skipped_same_count": False,
|
|
|
|
|
+ "odps_count": odps_count,
|
|
|
|
|
+ "mysql_count": mysql_count,
|
|
|
|
|
+ **_sync_diff(partition_date),
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _run_sync_stage(
|
|
|
|
|
+ stage_name: str,
|
|
|
|
|
+ action: Callable[[], dict[str, Any]],
|
|
|
|
|
+) -> tuple[dict[str, Any], str | None]:
|
|
|
|
|
+ """执行需求池同步子阶段,错误转为结构化结果并允许后续阶段继续。"""
|
|
|
|
|
+ try:
|
|
|
|
|
+ payload = action()
|
|
|
|
|
+ except Exception as exc:
|
|
|
|
|
+ logger.exception("Multi demand pool stage failed; continue: stage=%s", stage_name)
|
|
|
|
|
+ return {"success": False, "error": str(exc)}, str(exc)
|
|
|
|
|
+
|
|
|
|
|
+ if payload.get("success") is False:
|
|
|
|
|
+ error = str(payload.get("error") or f"{stage_name} returned success=False")
|
|
|
|
|
+ logger.error(
|
|
|
|
|
+ "Multi demand pool stage reported failure; continue: stage=%s error=%s",
|
|
|
|
|
+ stage_name,
|
|
|
|
|
+ error,
|
|
|
|
|
+ )
|
|
|
|
|
+ return payload, error
|
|
|
|
|
+ return payload, None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> dict:
|
|
|
|
|
+ """
|
|
|
|
|
+ 从 ODPS 增量同步策略需求天级数据到 MySQL,并对新词做归属分类与热度统计。
|
|
|
|
|
+
|
|
|
|
|
+ Args:
|
|
|
|
|
+ partition_date: 分区日期 (YYYYMMDD),默认当天
|
|
|
|
|
+ """
|
|
|
|
|
+ if partition_date is None:
|
|
|
|
|
+ partition_date = datetime.now().strftime("%Y%m%d")
|
|
|
|
|
+
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "Starting multi demand pool ODPS → MySQL sync for partition: %s",
|
|
|
|
|
+ partition_date,
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ stage_definitions: list[tuple[str, Callable[[], dict[str, Any]]]] = [
|
|
|
|
|
+ ("source_sync", lambda: _sync_pool_rows(partition_date)),
|
|
|
|
|
+ ("classify", lambda: _classify_words(partition_date)),
|
|
|
|
|
+ ("belong_pool_rel", sync_demand_belong_pool_rel),
|
|
|
|
|
+ ("real_metrics", lambda: enrich_real_rov_vov_7d(partition_date)),
|
|
|
|
|
+ ("popularity", lambda: compute_popularity_stats(partition_date)),
|
|
|
|
|
+ ("tree_weight", lambda: compute_category_tree_weight(partition_date)),
|
|
|
|
|
+ ("videos", lambda: sync_multi_demand_videos(limit=None)),
|
|
|
|
|
+ ]
|
|
|
|
|
+ stage_results: dict[str, dict[str, Any]] = {}
|
|
|
|
|
+ errors: list[dict[str, str]] = []
|
|
|
|
|
+ for stage_name, action in stage_definitions:
|
|
|
|
|
+ payload, error = _run_sync_stage(stage_name, action)
|
|
|
|
|
+ stage_results[stage_name] = payload
|
|
|
|
|
+ if error:
|
|
|
|
|
+ errors.append({"stage": stage_name, "error": error})
|
|
|
|
|
|
|
|
- classify_stats = _classify_words(partition_date)
|
|
|
|
|
- belong_pool_rel_stats = sync_demand_belong_pool_rel()
|
|
|
|
|
- real_metric_stats = enrich_real_rov_vov_7d(partition_date)
|
|
|
|
|
- popularity_stats = compute_popularity_stats(partition_date)
|
|
|
|
|
- tree_weight_stats = compute_category_tree_weight(partition_date)
|
|
|
|
|
- video_stats = sync_multi_demand_videos(limit=None)
|
|
|
|
|
|
|
+ sync_stats = stage_results["source_sync"]
|
|
|
|
|
|
|
|
result = {
|
|
result = {
|
|
|
|
|
+ "success": not errors,
|
|
|
"partition_date": partition_date,
|
|
"partition_date": partition_date,
|
|
|
- **sync_stats,
|
|
|
|
|
- "real_metrics": real_metric_stats,
|
|
|
|
|
- "classify": classify_stats,
|
|
|
|
|
- "belong_pool_rel": belong_pool_rel_stats,
|
|
|
|
|
- "popularity": popularity_stats,
|
|
|
|
|
- "tree_weight": tree_weight_stats,
|
|
|
|
|
- "videos": video_stats,
|
|
|
|
|
|
|
+ **({} if errors and sync_stats.get("success") is False else sync_stats),
|
|
|
|
|
+ "source_sync": sync_stats,
|
|
|
|
|
+ "real_metrics": stage_results["real_metrics"],
|
|
|
|
|
+ "classify": stage_results["classify"],
|
|
|
|
|
+ "belong_pool_rel": stage_results["belong_pool_rel"],
|
|
|
|
|
+ "popularity": stage_results["popularity"],
|
|
|
|
|
+ "tree_weight": stage_results["tree_weight"],
|
|
|
|
|
+ "videos": stage_results["videos"],
|
|
|
|
|
+ "errors": errors,
|
|
|
"synced_at": datetime.now().isoformat(),
|
|
"synced_at": datetime.now().isoformat(),
|
|
|
}
|
|
}
|
|
|
logger.info("Multi demand pool sync completed: %s", result)
|
|
logger.info("Multi demand pool sync completed: %s", result)
|