demand_server/app/strategies/impl/solar_same_period_strategy.py

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.solar_same_period_demands import query_solar_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 SolarSamePeriodStrategy(BaseStrategy):
    """去年同期阳历：取去年同日起到 +6 天（共 7 天）视频窗口聚合实质元素 ROV。"""

    strategy_id = "solar_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_solar_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