from typing import Any from app.core.config import settings from app.strategies.batch_date import today_yyyymmdd from app.strategies.base import ( BaseStrategy, DemandCandidate, GenerateContext, ) from app.strategies.odps._utils import round_weight from app.strategies.sources.hot_content_sync_log import fetch_hot_content_sync_log_by_partition from app.strategies.staging_store import insert_staging_rows_skip_duplicates class HotEventStrategy(BaseStrategy): """新热事件:从 hot_content_odps_sync_log 同步至 strategy_staging。""" strategy_id = "hot_event" name = "新热事件" version = "1.0.0" def validate_config(self, config: dict[str, Any]) -> bool: if not settings.hot_content_mysql_configured: return False return isinstance(config, dict) def generate(self, context: GenerateContext) -> list[DemandCandidate]: partition_dt = today_yyyymmdd() source_strategy = str(context.params.get("source_strategy") or self.name).strip() rows = fetch_hot_content_sync_log_by_partition( partition_dt=partition_dt, strategy=source_strategy, ) candidates: list[DemandCandidate] = [] for row in rows: partition = str(row.get("partition_dt") or partition_dt).strip() demand_name = str(row.get("demand_name") or "").strip() if not partition or not demand_name: continue demand_type = row.get("demand_type") parsed_type = str(demand_type).strip() if demand_type else None candidates.append( DemandCandidate( content=demand_name, demand_id=str(row["demand_id"]), demand_type=parsed_type, priority_score=round_weight(row.get("weight")), extra={"batch_date": partition}, ) ) return candidates def write_staging( self, *, context: GenerateContext, candidates: list[DemandCandidate], ) -> dict[str, Any]: return insert_staging_rows_skip_duplicates( strategy_config_id=self.strategy_id, strategy_name=self.name, candidates=candidates, )