demand_server/app/strategies/base.py

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class GenerateContext:
    """策略产出上下文。"""

    batch_date: str
    params: dict[str, Any]
    strategy_id: str
    strategy_name: str
    window_start: str | None = None
    window_end: str | None = None


@dataclass(frozen=True)
class StrategySkipDecision:
    """策略跳过决策，由各策略在 should_skip() 中自行返回。"""

    skip: bool
    reason: str = ""
    detail: dict[str, Any] = field(default_factory=dict)


@dataclass
class DemandCandidate:
    """
    策略产出的候选需求。

    generate() 返回此结构；写入 strategy_staging 时映射为：
    - content -> demand_name
    - priority_score -> weight
    - extra -> extend
    """

    content: str
    confidence: float | None = None
    priority_score: float | None = None
    demand_id: str | None = None
    demand_type: str | None = None
    video_count: int | None = None
    video_list: list[Any] | None = None
    extra: dict[str, Any] = field(default_factory=dict)


class BaseStrategy(ABC):
    """所有策略实现的统一接口。"""

    strategy_id: str
    name: str
    version: str

    @abstractmethod
    def validate_config(self, config: dict[str, Any]) -> bool:
        """校验策略参数，注册或热更新时调用。"""

    @abstractmethod
    def generate(self, context: GenerateContext) -> list[DemandCandidate]:
        """产出候选需求列表。"""

    def should_skip(self, context: GenerateContext) -> StrategySkipDecision:
        """各策略自行决定是否跳过本次产出，默认不跳过。"""
        return StrategySkipDecision(skip=False)

    def write_staging(
        self,
        *,
        context: GenerateContext,
        candidates: list[DemandCandidate],
    ) -> dict[str, Any]:
        """写入 strategy_staging，默认按 context.batch_date 整批替换。"""
        from app.strategies.staging_store import replace_staging_batch

        written = replace_staging_batch(
            strategy_config_id=self.strategy_id,
            strategy_name=self.name,
            batch_date=context.batch_date,
            candidates=candidates,
        )
        return {"written": written, "skipped_duplicates": 0}