from typing import Any from app.strategies.batch_date import today_yyyymmdd from app.strategies.base import ( BaseStrategy, DemandCandidate, GenerateContext, StrategySkipDecision, ) from app.strategies.odps.lunar_same_period_demands import query_lunar_same_period_demands from app.strategies.staging_store import count_staging_batch, has_staging_batch _NUMERIC_PARAM_KEYS = ( ("view_pv_count", ("view_pv_count",)), ("min_contribution_score", ("min_contribution_score", "贡献分")), ("rov_avg", ("rov_avg",)), ) _OPTIONAL_PARAM_KEYS = ( ("period_days", ("period_days",), 7), ) class LunarSamePeriodStrategy(BaseStrategy): """去年同期阴历:阴历减一年后转阳历,取 7 天视频窗口聚合实质元素 ROV。""" strategy_id = "lunar_same_period_last_year" name = "去年同期阴历" version = "1.0.0" def validate_config(self, config: dict[str, Any]) -> bool: try: self._resolve_params(config) except (TypeError, ValueError, KeyError): return False return True def should_skip(self, context: GenerateContext) -> StrategySkipDecision: if not has_staging_batch( strategy_config_id=self.strategy_id, batch_date=context.batch_date, ): return StrategySkipDecision(skip=False) existing_count = count_staging_batch( strategy_config_id=self.strategy_id, batch_date=context.batch_date, ) return StrategySkipDecision( skip=True, reason="strategy_staging already has data for batch_date", detail={"existing_count": existing_count}, ) def generate(self, context: GenerateContext) -> list[DemandCandidate]: params = self._resolve_params(context.params) rows = query_lunar_same_period_demands( bizdate=today_yyyymmdd(), period_days=params["period_days"], view_pv_count=params["view_pv_count"], min_contribution_score=params["min_contribution_score"], rov_avg=params["rov_avg"], ) candidates: list[DemandCandidate] = [] for row in rows: demand_name = str(row.get("demand_name") or "").strip() if not demand_name: continue weight = row.get("weight") priority_score = float(weight) if weight is not None else None video_count = row.get("video_count") parsed_video_count = int(video_count) if video_count is not None else None candidates.append( DemandCandidate( content=demand_name, priority_score=priority_score, demand_id=str(row["demand_id"]) if row.get("demand_id") else None, demand_type=str(row.get("type") or "特征点"), video_count=parsed_video_count, video_list=row.get("video_list"), ) ) return candidates @staticmethod def _pick_param(config: dict[str, Any], keys: tuple[str, ...]) -> Any: for key in keys: if key in config: return config[key] raise KeyError(keys[0]) @classmethod def _resolve_params(cls, config: dict[str, Any]) -> dict[str, int | float]: resolved: dict[str, int | float] = {} for canonical, aliases in _NUMERIC_PARAM_KEYS: raw = cls._pick_param(config, aliases) if canonical == "view_pv_count": value = int(raw) if value < 0: raise ValueError(f"{canonical} 不能为负") resolved[canonical] = value else: value = float(raw) if value < 0: raise ValueError(f"{canonical} 不能为负") resolved[canonical] = value for canonical, aliases, default in _OPTIONAL_PARAM_KEYS: raw = default for key in aliases: if key in config: raw = config[key] break value = int(raw) if value < 1: raise ValueError(f"{canonical} 须 >= 1") resolved[canonical] = value return resolved