TrendRadar/trendradar/notification/push_manager.py

# coding=utf-8
"""
推送记录管理模块

管理推送记录，支持每日只推送一次和时间窗口控制
通过 storage_backend 统一存储，支持本地 SQLite 和远程云存储
"""

from datetime import datetime
from typing import Callable, Optional, Any, Tuple

import pytz

from trendradar.utils.time import DEFAULT_TIMEZONE, TimeWindowChecker


class PushRecordManager:
    """
    推送记录管理器

    通过 storage_backend 统一管理推送记录：
    - 本地环境：使用 LocalStorageBackend，数据存储在本地 SQLite
    - GitHub Actions：使用 RemoteStorageBackend，数据存储在云端

    这样 once_per_day 功能在 GitHub Actions 上也能正常工作。
    """

    def __init__(
        self,
        storage_backend: Any,
        get_time_func: Optional[Callable[[], datetime]] = None,
    ):
        """
        初始化推送记录管理器

        Args:
            storage_backend: 存储后端实例（LocalStorageBackend 或 RemoteStorageBackend）
            get_time_func: 获取当前时间的函数（应使用配置的时区）
        """
        self.storage_backend = storage_backend
        self.get_time = get_time_func or self._default_get_time

        print(f"[推送记录] 使用 {storage_backend.backend_name} 存储后端")

    def _default_get_time(self) -> datetime:
        """默认时间获取函数（使用 storage_backend 的时区配置）"""
        timezone = getattr(self.storage_backend, 'timezone', DEFAULT_TIMEZONE)
        return datetime.now(pytz.timezone(timezone))

    def has_pushed_today(self) -> bool:
        """
        检查今天是否已经推送过

        Returns:
            是否已推送
        """
        return self.storage_backend.has_pushed_today()

    def record_push(self, report_type: str) -> bool:
        """
        记录推送

        Args:
            report_type: 报告类型

        Returns:
            是否记录成功
        """
        return self.storage_backend.record_push(report_type)

    def is_in_time_range(self, start_time: str, end_time: str) -> bool:
        """
        检查当前时间是否在指定时间范围内

        Args:
            start_time: 开始时间（格式：HH:MM）
            end_time: 结束时间（格式：HH:MM）

        Returns:
            是否在时间范围内
        """
        checker = TimeWindowChecker(
            storage_backend=self.storage_backend,
            get_time_func=self.get_time,
            window_name="推送窗口",
        )
        return checker.is_in_time_range(start_time, end_time)

    def check_push_window(self, window_config: dict) -> Tuple[bool, str]:
        """
        检查推送窗口控制

        Args:
            window_config: 推送窗口配置

        Returns:
            (should_push, reason) 元组
        """
        checker = TimeWindowChecker(
            storage_backend=self.storage_backend,
            get_time_func=self.get_time,
            window_name="推送窗口",
        )
        return checker.check_window(
            window_config=window_config,
            check_once_per_day_func=self.has_pushed_today,
        )

    def check_ai_analysis_window(self, window_config: dict) -> Tuple[bool, str]:
        """
        检查 AI 分析窗口控制

        Args:
            window_config: AI 分析窗口配置

        Returns:
            (should_analyze, reason) 元组
        """
        checker = TimeWindowChecker(
            storage_backend=self.storage_backend,
            get_time_func=self.get_time,
            window_name="AI 分析窗口",
        )
        return checker.check_window(
            window_config=window_config,
            check_once_per_day_func=self.storage_backend.has_ai_analyzed_today,
        )

    def get_push_status(self, window_config: dict) -> dict:
        """
        获取推送状态信息

        Args:
            window_config: 推送窗口配置

        Returns:
            状态信息字典
        """
        checker = TimeWindowChecker(
            storage_backend=self.storage_backend,
            get_time_func=self.get_time,
            window_name="推送窗口",
        )
        status = checker.get_status(
            window_config=window_config,
            check_once_per_day_func=self.has_pushed_today,
        )
        status["window_type"] = "push"
        return status

    def get_ai_analysis_status(self, window_config: dict) -> dict:
        """
        获取 AI 分析状态信息

        Args:
            window_config: AI 分析窗口配置

        Returns:
            状态信息字典
        """
        checker = TimeWindowChecker(
            storage_backend=self.storage_backend,
            get_time_func=self.get_time,
            window_name="AI 分析窗口",
        )
        status = checker.get_status(
            window_config=window_config,
            check_once_per_day_func=self.storage_backend.has_ai_analyzed_today,
        )
        status["window_type"] = "ai_analysis"
        return status

    def reset_push_state(self) -> bool:
        """
        重置今日推送状态

        Returns:
            是否重置成功
        """
        try:
            # 通过存储后端重置推送记录
            if hasattr(self.storage_backend, 'reset_push_state'):
                return self.storage_backend.reset_push_state()
            else:
                print("[推送记录] 存储后端不支持重置推送状态")
                return False
        except Exception as e:
            print(f"[推送记录] 重置推送状态失败: {e}")
            return False

    def reset_ai_analysis_state(self) -> bool:
        """
        重置今日 AI 分析状态

        Returns:
            是否重置成功
        """
        try:
            if hasattr(self.storage_backend, 'reset_ai_analysis_state'):
                return self.storage_backend.reset_ai_analysis_state()
            else:
                print("[推送记录] 存储后端不支持重置 AI 分析状态")
                return False
        except Exception as e:
            print(f"[推送记录] 重置 AI 分析状态失败: {e}")
            return False