TrendRadar/trendradar/core/scheduler.py

# coding=utf-8
"""
时间线调度器

统一的时间线调度系统，替代分散的 push_window / analysis_window 逻辑。
基于 periods + day_plans + week_map 模型实现灵活的时间段调度。
"""

import copy
import re
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

from datetime import datetime


@dataclass
class ResolvedSchedule:
    """当前时间解析后的调度结果"""
    period_key: Optional[str]       # 命中的 period key，None=默认配置
    period_name: Optional[str]      # 命中的展示名称
    day_plan: str                   # 当前日计划
    collect: bool
    analyze: bool
    push: bool
    report_mode: str
    ai_mode: str
    once_analyze: bool
    once_push: bool


class Scheduler:
    """
    时间线调度器

    根据 timeline 配置（periods + day_plans + week_map）解析当前时间应执行的行为。
    支持：
    - 预设模板 + 自定义模式
    - 跨日时间段（如 22:00-07:00）
    - 每天 / 每周差异化配置
    - once 执行去重（analyze / push 独立维度）
    - 冲突策略（error_on_overlap / last_wins）
    """

    def __init__(
        self,
        schedule_config: Dict[str, Any],
        timeline_data: Dict[str, Any],
        storage_backend: Any,
        get_time_func: Callable[[], datetime],
    ):
        """
        初始化调度器

        Args:
            schedule_config: config.yaml 中的 schedule 段（含 preset 等）
            timeline_data: timeline.yaml 的完整数据
            storage_backend: 存储后端（用于 once 去重记录）
            get_time_func: 获取当前时间的函数（应使用配置的时区）
        """
        self.schedule_config = schedule_config
        self.storage = storage_backend
        self.get_time = get_time_func
        self.enabled = schedule_config.get("enabled", True)

        # 加载并构建最终 timeline
        self.timeline = self._build_timeline(schedule_config, timeline_data)
        if self.enabled:
            self._validate_timeline(self.timeline)

    def _build_timeline(
        self,
        schedule_config: Dict[str, Any],
        timeline_data: Dict[str, Any],
    ) -> Dict[str, Any]:
        """从 preset 或 custom 构建 timeline"""
        preset = schedule_config.get("preset", "always_on")

        if preset == "custom":
            timeline = copy.deepcopy(timeline_data.get("custom", {}))
        else:
            presets = timeline_data.get("presets", {})
            if preset not in presets:
                raise ValueError(
                    f"未知的预设模板: '{preset}'，可选值: "
                    f"{', '.join(presets.keys())}, custom"
                )
            timeline = copy.deepcopy(presets[preset])

        # 确保 periods 是 dict（可能为空 {}）
        if timeline.get("periods") is None:
            timeline["periods"] = {}

        return timeline

    def resolve(self) -> ResolvedSchedule:
        """
        解析当前时间对应的调度配置

        Returns:
            ResolvedSchedule 包含当前应执行的行为
        """
        if not self.enabled:
            # 调度未启用时返回默认的全功能配置
            return ResolvedSchedule(
                period_key=None,
                period_name=None,
                day_plan="disabled",
                collect=True,
                analyze=True,
                push=True,
                report_mode="current",
                ai_mode="follow_report",
                once_analyze=False,
                once_push=False,
            )

        now = self.get_time()
        weekday = now.isoweekday()  # 1=周一 ... 7=周日
        now_hhmm = now.strftime("%H:%M")

        # 查找当天的日计划
        day_plan_key = self.timeline["week_map"].get(weekday)
        if day_plan_key is None:
            raise ValueError(f"week_map 缺少星期映射: {weekday}")

        day_plan = self.timeline["day_plans"].get(day_plan_key)
        if day_plan is None:
            raise ValueError(f"week_map[{weekday}] 引用了不存在的 day_plan: {day_plan_key}")

        # 查找当前活跃的时间段
        period_key = self._find_active_period(now_hhmm, day_plan)

        # 合并默认配置和时间段配置
        merged = self._merge_with_default(period_key)

        # 打印调度日志
        weekday_names = {1: "一", 2: "二", 3: "三", 4: "四", 5: "五", 6: "六", 7: "日"}
        period_display = "默认配置（未命中任何时间段）"
        if period_key:
            period_cfg = self.timeline["periods"][period_key]
            period_name = period_cfg.get("name", period_key)
            start = period_cfg.get("start", "?")
            end = period_cfg.get("end", "?")
            period_display = f"{period_name} ({start}-{end})"

        print(f"[调度] 星期{weekday_names.get(weekday, '?')}，日计划: {day_plan_key}")
        print(f"[调度] 当前时间段: {period_display}")

        resolved = ResolvedSchedule(
            period_key=period_key,
            period_name=(
                self.timeline["periods"][period_key].get("name")
                if period_key
                else None
            ),
            day_plan=day_plan_key,
            collect=merged.get("collect", True),
            analyze=merged.get("analyze", False),
            push=merged.get("push", False),
            report_mode=merged.get("report_mode", "current"),
            ai_mode=self._resolve_ai_mode(merged),
            once_analyze=merged.get("once", {}).get("analyze", False),
            once_push=merged.get("once", {}).get("push", False),
        )

        # 打印行为摘要
        actions = []
        if resolved.collect:
            actions.append("采集")
        if resolved.analyze:
            actions.append(f"分析(AI:{resolved.ai_mode})")
        if resolved.push:
            actions.append(f"推送(模式:{resolved.report_mode})")
        print(f"[调度] 行为: {', '.join(actions) if actions else '无'}")

        return resolved

    def _find_active_period(
        self, now_hhmm: str, day_plan: Dict[str, Any]
    ) -> Optional[str]:
        """
        查找当前时间命中的活跃时间段

        Args:
            now_hhmm: 当前时间 HH:MM
            day_plan: 日计划配置

        Returns:
            命中的 period key，或 None
        """
        candidates = []
        for idx, key in enumerate(day_plan.get("periods", [])):
            period = self.timeline["periods"].get(key)
            if period is None:
                continue
            if self._in_range(now_hhmm, period["start"], period["end"]):
                candidates.append((idx, key))

        if not candidates:
            return None

        # 检查冲突
        if len(candidates) > 1:
            policy = self.timeline.get("overlap", {}).get("policy", "error_on_overlap")
            conflicting = [c[1] for c in candidates]

            if policy == "error_on_overlap":
                raise ValueError(
                    f"检测到时间段重叠冲突: {', '.join(conflicting)} 在 {now_hhmm} 重叠。"
                    f"请调整时间段配置，或将 overlap.policy 设为 'last_wins'"
                )

            # last_wins：输出重叠警告，列表中后面的优先
            print(
                f"[调度] 检测到时间段重叠: {', '.join(conflicting)} 在 {now_hhmm} 重叠"
            )
            winner = candidates[-1]
            print(f"[调度] 冲突策略: last_wins，生效时间段: {winner[1]}")
            return winner[1]

        return candidates[0][1]

    @staticmethod
    def _in_range(now_hhmm: str, start: str, end: str) -> bool:
        """
        检查时间是否在范围内（支持跨日）

        Args:
            now_hhmm: 当前时间 HH:MM
            start: 开始时间 HH:MM
            end: 结束时间 HH:MM

        Returns:
            是否在范围内
        """
        if start <= end:
            # 正常范围，如 08:00-09:00
            return start <= now_hhmm <= end
        else:
            # 跨日范围，如 22:00-07:00
            return now_hhmm >= start or now_hhmm <= end

    def _merge_with_default(self, period_key: Optional[str]) -> Dict[str, Any]:
        """合并默认配置和时间段配置"""
        base = copy.deepcopy(self.timeline.get("default", {}))
        if not period_key:
            return base

        period = copy.deepcopy(self.timeline["periods"][period_key])

        # 先合并 once 子对象
        merged_once = dict(base.get("once", {}))
        merged_once.update(period.get("once", {}))

        # 标量字段覆盖
        base.update(period)

        # 恢复合并后的 once
        if merged_once:
            base["once"] = merged_once

        return base

    @staticmethod
    def _resolve_ai_mode(cfg: Dict[str, Any]) -> str:
        """解析最终的 AI 模式"""
        ai_mode = cfg.get("ai_mode", "follow_report")
        if ai_mode == "follow_report":
            return cfg.get("report_mode", "current")
        return ai_mode

    def already_executed(self, period_key: str, action: str, date_str: str) -> bool:
        """
        检查指定时间段的某个 action 今天是否已执行

        Args:
            period_key: 时间段 key
            action: 动作类型 (analyze / push)
            date_str: 日期 YYYY-MM-DD

        Returns:
            是否已执行
        """
        return self.storage.has_period_executed(date_str, period_key, action)

    def record_execution(self, period_key: str, action: str, date_str: str) -> None:
        """
        记录时间段的 action 执行

        Args:
            period_key: 时间段 key
            action: 动作类型 (analyze / push)
            date_str: 日期 YYYY-MM-DD
        """
        self.storage.record_period_execution(date_str, period_key, action)

    # ========================================
    # 校验
    # ========================================

    def _validate_timeline(self, timeline: Dict[str, Any]) -> None:
        """
        启动时校验 timeline 配置

        Raises:
            ValueError: 配置不合法时抛出
        """
        required_top_keys = ["default", "periods", "day_plans", "week_map"]
        for key in required_top_keys:
            if key not in timeline:
                raise ValueError(f"timeline 缺少必须字段: {key}")

        # week_map 必须覆盖 1..7
        for day in range(1, 8):
            if day not in timeline["week_map"]:
                raise ValueError(f"week_map 缺少星期映射: {day}")

        # day_plan 引用完整性
        for day, plan_key in timeline["week_map"].items():
            if plan_key not in timeline["day_plans"]:
                raise ValueError(
                    f"week_map[{day}] 引用了不存在的 day_plan: {plan_key}"
                )

        # period 引用完整性
        for plan_key, plan in timeline["day_plans"].items():
            for period_key in plan.get("periods", []):
                if period_key not in timeline["periods"]:
                    raise ValueError(
                        f"day_plan[{plan_key}] 引用了不存在的 period: {period_key}"
                    )

        # 时间格式校验
        for period_key, period in timeline["periods"].items():
            if "start" not in period or "end" not in period:
                raise ValueError(
                    f"period '{period_key}' 缺少 start 或 end 字段"
                )
            self._validate_hhmm(period["start"], f"{period_key}.start")
            self._validate_hhmm(period["end"], f"{period_key}.end")
            if period["start"] == period["end"]:
                raise ValueError(
                    f"period '{period_key}' 的 start 与 end 不能相同: {period['start']}"
                )

        # 检查冲突策略下的重叠
        policy = timeline.get("overlap", {}).get("policy", "error_on_overlap")
        if policy == "error_on_overlap":
            self._check_period_overlaps(timeline)

    def _check_period_overlaps(self, timeline: Dict[str, Any]) -> None:
        """
        检查每个日计划中的时间段是否存在重叠

        仅在 overlap.policy == "error_on_overlap" 时调用
        """
        periods = timeline.get("periods", {})

        for plan_key, plan in timeline["day_plans"].items():
            period_keys = plan.get("periods", [])
            if len(period_keys) <= 1:
                continue

            # 收集每个时间段的范围
            ranges = []
            for pk in period_keys:
                p = periods.get(pk, {})
                if "start" in p and "end" in p:
                    ranges.append((pk, p["start"], p["end"]))

            # 两两检查重叠
            for i in range(len(ranges)):
                for j in range(i + 1, len(ranges)):
                    if self._ranges_overlap(
                        ranges[i][1], ranges[i][2],
                        ranges[j][1], ranges[j][2],
                    ):
                        raise ValueError(
                            f"day_plan '{plan_key}' 中时间段 '{ranges[i][0]}' "
                            f"({ranges[i][1]}-{ranges[i][2]}) 与 '{ranges[j][0]}' "
                            f"({ranges[j][1]}-{ranges[j][2]}) 存在重叠。"
                            f"请调整时间段，或将 overlap.policy 设为 'last_wins'"
                        )

    @staticmethod
    def _ranges_overlap(s1: str, e1: str, s2: str, e2: str) -> bool:
        """检查两个时间范围是否重叠（支持跨日）"""
        def to_minutes(t: str) -> int:
            h, m = t.split(":")
            return int(h) * 60 + int(m)

        def expand_range(start: str, end: str) -> List[tuple]:
            """将时间范围展开为分钟段列表，跨日时拆分为两段"""
            s = to_minutes(start)
            e = to_minutes(end)
            if s <= e:
                return [(s, e)]
            else:
                # 跨日：拆分为 [start, 23:59] 和 [00:00, end]
                return [(s, 24 * 60 - 1), (0, e)]

        segs1 = expand_range(s1, e1)
        segs2 = expand_range(s2, e2)

        for a_start, a_end in segs1:
            for b_start, b_end in segs2:
                # 两个区间有重叠的条件
                if a_start <= b_end and b_start <= a_end:
                    return True
        return False

    @staticmethod
    def _validate_hhmm(value: str, field_name: str) -> None:
        """校验 HH:MM 格式"""
        if not re.match(r"^\d{2}:\d{2}$", value):
            raise ValueError(f"{field_name} 格式错误: '{value}'，期望 HH:MM")
        h, m = value.split(":")
        if not (0 <= int(h) <= 23 and 0 <= int(m) <= 59):
            raise ValueError(f"{field_name} 时间值超出范围: '{value}'")