demand_server/app/strategies/impl/hot_event_strategy.py

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,
        )