TrendRadar/trendradar/utils/time.py

# coding=utf-8
"""
时间工具模块

本模块提供统一的时间处理函数，所有时区相关操作都应使用 DEFAULT_TIMEZONE 常量。
"""

from datetime import datetime
from typing import Optional, Tuple

import pytz

# 默认时区常量 - 仅作为 fallback，正常运行时使用 config.yaml 中的 app.timezone
DEFAULT_TIMEZONE = "Asia/Shanghai"


def get_configured_time(timezone: str = DEFAULT_TIMEZONE) -> datetime:
    """
    获取配置时区的当前时间

    Args:
        timezone: 时区名称，如 'Asia/Shanghai', 'America/Los_Angeles'

    Returns:
        带时区信息的当前时间
    """
    try:
        tz = pytz.timezone(timezone)
    except pytz.UnknownTimeZoneError:
        print(f"[警告] 未知时区 '{timezone}'，使用默认时区 {DEFAULT_TIMEZONE}")
        tz = pytz.timezone(DEFAULT_TIMEZONE)
    return datetime.now(tz)


def format_date_folder(
    date: Optional[str] = None, timezone: str = DEFAULT_TIMEZONE
) -> str:
    """
    格式化日期文件夹名 (ISO 格式: YYYY-MM-DD)

    Args:
        date: 指定日期字符串，为 None 则使用当前日期
        timezone: 时区名称

    Returns:
        格式化后的日期字符串，如 '2025-12-09'
    """
    if date:
        return date
    return get_configured_time(timezone).strftime("%Y-%m-%d")


def format_time_filename(timezone: str = DEFAULT_TIMEZONE) -> str:
    """
    格式化时间文件名 (格式: HH-MM，用于文件名)

    Windows 系统不支持冒号作为文件名，因此使用连字符

    Args:
        timezone: 时区名称

    Returns:
        格式化后的时间字符串，如 '15-30'
    """
    return get_configured_time(timezone).strftime("%H-%M")


def get_current_time_display(timezone: str = DEFAULT_TIMEZONE) -> str:
    """
    获取当前时间显示 (格式: HH:MM，用于显示)

    Args:
        timezone: 时区名称

    Returns:
        格式化后的时间字符串，如 '15:30'
    """
    return get_configured_time(timezone).strftime("%H:%M")


def convert_time_for_display(time_str: str) -> str:
    """
    将 HH-MM 格式转换为 HH:MM 格式用于显示

    Args:
        time_str: 输入时间字符串，如 '15-30'

    Returns:
        转换后的时间字符串，如 '15:30'
    """
    if time_str and "-" in time_str and len(time_str) == 5:
        return time_str.replace("-", ":")
    return time_str


def format_iso_time_friendly(
    iso_time: str,
    timezone: str = DEFAULT_TIMEZONE,
    include_date: bool = True,
) -> str:
    """
    将 ISO 格式时间转换为用户时区的友好显示格式

    Args:
        iso_time: ISO 格式时间字符串，如 '2025-12-29T00:20:00' 或 '2025-12-29T00:20:00+00:00'
        timezone: 目标时区名称
        include_date: 是否包含日期部分

    Returns:
        友好格式的时间字符串，如 '12-29 08:20' 或 '08:20'
    """
    if not iso_time:
        return ""

    try:
        # 尝试解析各种 ISO 格式
        dt = None

        # 尝试解析带时区的格式
        if "+" in iso_time or iso_time.endswith("Z"):
            iso_time = iso_time.replace("Z", "+00:00")
            try:
                dt = datetime.fromisoformat(iso_time)
            except ValueError:
                pass

        # 尝试解析不带时区的格式（假设为 UTC）
        if dt is None:
            try:
                # 处理 T 分隔符
                if "T" in iso_time:
                    dt = datetime.fromisoformat(iso_time.replace("T", " ").split(".")[0])
                else:
                    dt = datetime.fromisoformat(iso_time.split(".")[0])
                # 假设为 UTC 时间
                dt = pytz.UTC.localize(dt)
            except ValueError:
                pass

        if dt is None:
            # 无法解析，返回原始字符串的简化版本
            if "T" in iso_time:
                parts = iso_time.split("T")
                if len(parts) == 2:
                    date_part = parts[0][5:]  # MM-DD
                    time_part = parts[1][:5]  # HH:MM
                    return f"{date_part} {time_part}" if include_date else time_part
            return iso_time

        # 转换到目标时区
        try:
            target_tz = pytz.timezone(timezone)
        except pytz.UnknownTimeZoneError:
            target_tz = pytz.timezone(DEFAULT_TIMEZONE)

        dt_local = dt.astimezone(target_tz)

        # 格式化输出
        if include_date:
            return dt_local.strftime("%m-%d %H:%M")
        else:
            return dt_local.strftime("%H:%M")

    except Exception:
        # 出错时返回原始字符串的简化版本
        if "T" in iso_time:
            parts = iso_time.split("T")
            if len(parts) == 2:
                date_part = parts[0][5:]  # MM-DD
                time_part = parts[1][:5]  # HH:MM
                return f"{date_part} {time_part}" if include_date else time_part
        return iso_time


def is_within_days(
    iso_time: str,
    max_days: int,
    timezone: str = DEFAULT_TIMEZONE,
) -> bool:
    """
    检查 ISO 格式时间是否在指定天数内

    用于 RSS 文章新鲜度过滤，判断文章发布时间是否超过指定天数。

    Args:
        iso_time: ISO 格式时间字符串（如 '2025-12-29T00:20:00' 或带时区）
        max_days: 最大天数（文章发布时间距今不超过此天数则返回 True）
            - max_days > 0: 正常过滤，保留 N 天内的文章
            - max_days <= 0: 禁用过滤，保留所有文章
        timezone: 时区名称（用于获取当前时间）

    Returns:
        True 如果时间在指定天数内（应保留），False 如果超过指定天数（应过滤）
        如果无法解析时间，返回 True（保留文章）
    """
    # 无时间戳或禁用过滤时，保留文章
    if not iso_time:
        return True
    if max_days <= 0:
        return True  # max_days=0 表示禁用过滤

    try:
        dt = None

        # 尝试解析带时区的格式
        if "+" in iso_time or iso_time.endswith("Z"):
            iso_time_normalized = iso_time.replace("Z", "+00:00")
            try:
                dt = datetime.fromisoformat(iso_time_normalized)
            except ValueError:
                pass

        # 尝试解析不带时区的格式（假设为 UTC）
        if dt is None:
            try:
                if "T" in iso_time:
                    dt = datetime.fromisoformat(iso_time.replace("T", " ").split(".")[0])
                else:
                    dt = datetime.fromisoformat(iso_time.split(".")[0])
                dt = pytz.UTC.localize(dt)
            except ValueError:
                pass

        if dt is None:
            # 无法解析时间，保留文章
            return True

        # 获取当前时间（配置的时区，带时区信息）
        now = get_configured_time(timezone)

        # 计算时间差（两个带时区的 datetime 相减会自动处理时区差异）
        diff = now - dt
        days_diff = diff.total_seconds() / (24 * 60 * 60)

        return days_diff <= max_days

    except Exception:
        # 出错时保留文章
        return True


def calculate_days_old(iso_time: str, timezone: str = DEFAULT_TIMEZONE) -> Optional[float]:
    """
    计算 ISO 格式时间距今多少天

    Args:
        iso_time: ISO 格式时间字符串
        timezone: 时区名称

    Returns:
        距今天数（浮点数），如果无法解析返回 None
    """
    if not iso_time:
        return None

    try:
        dt = None

        # 尝试解析带时区的格式
        if "+" in iso_time or iso_time.endswith("Z"):
            iso_time_normalized = iso_time.replace("Z", "+00:00")
            try:
                dt = datetime.fromisoformat(iso_time_normalized)
            except ValueError:
                pass

        # 尝试解析不带时区的格式（假设为 UTC）
        if dt is None:
            try:
                if "T" in iso_time:
                    dt = datetime.fromisoformat(iso_time.replace("T", " ").split(".")[0])
                else:
                    dt = datetime.fromisoformat(iso_time.split(".")[0])
                dt = pytz.UTC.localize(dt)
            except ValueError:
                pass

        if dt is None:
            return None

        now = get_configured_time(timezone)
        diff = now - dt
        return diff.total_seconds() / (24 * 60 * 60)

    except Exception:
        return None


class TimeWindowChecker:
    """
    时间窗口检查器

    统一管理时间窗口控制逻辑，支持：
    - 推送窗口控制 (push_window)
    - AI 分析窗口控制 (analysis_window)
    - once_per_day 功能
    """

    def __init__(
        self,
        storage_backend,
        get_time_func=None,
        window_name: str = "时间窗口",
    ):
        """
        初始化时间窗口检查器

        Args:
            storage_backend: 存储后端实例
            get_time_func: 获取当前时间的函数
            window_name: 窗口名称（用于日志输出）
        """
        self.storage_backend = storage_backend
        self.get_time_func = get_time_func or (lambda: get_configured_time(DEFAULT_TIMEZONE))
        self.window_name = window_name

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

        支持跨日时间窗口，例如：
        - 正常窗口：09:00-21:00（当天 9 点到 21 点）
        - 跨日窗口：22:00-02:00（当天 22 点到次日 2 点）

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

        Returns:
            是否在时间范围内
        """
        now = self.get_time_func()
        current_time = now.strftime("%H:%M")

        normalized_start = self._normalize_time(start_time)
        normalized_end = self._normalize_time(end_time)
        normalized_current = self._normalize_time(current_time)

        # 判断是否跨日窗口（start > end 表示跨日，如 22:00-02:00）
        if normalized_start <= normalized_end:
            # 正常窗口：09:00-21:00
            result = normalized_start <= normalized_current <= normalized_end
        else:
            # 跨日窗口：22:00-02:00
            # 当前时间 >= 开始时间（如 23:00 >= 22:00）或 当前时间 <= 结束时间（如 01:00 <= 02:00）
            result = normalized_current >= normalized_start or normalized_current <= normalized_end

        if not result:
            print(f"[{self.window_name}] 当前 {normalized_current}，窗口 {normalized_start}-{normalized_end}")

        return result

    def _normalize_time(self, time_str: str) -> str:
        """将时间字符串标准化为 HH:MM 格式"""
        try:
            parts = time_str.strip().split(":")
            if len(parts) != 2:
                raise ValueError(f"时间格式错误: {time_str}")

            hour = int(parts[0])
            minute = int(parts[1])

            if not (0 <= hour <= 23 and 0 <= minute <= 59):
                raise ValueError(f"时间范围错误: {time_str}")

            return f"{hour:02d}:{minute:02d}"
        except Exception as e:
            print(f"[{self.window_name}] 时间格式化错误 '{time_str}': {e}")
            return time_str

    def check_window(
        self,
        window_config: dict,
        check_once_per_day_func=None,
        record_func=None,
    ) -> Tuple[bool, str]:
        """
        统一的时间窗口检查逻辑

        Args:
            window_config: 窗口配置字典，包含：
                - ENABLED: 是否启用窗口控制
                - TIME_RANGE: {"START": "HH:MM", "END": "HH:MM"}
                - ONCE_PER_DAY: 是否每天只执行一次
            check_once_per_day_func: 检查今天是否已执行的函数
            record_func: 记录执行的函数（成功后调用）

        Returns:
            (should_proceed, reason) 元组：
            - should_proceed: 是否应该继续执行
            - reason: 原因说明
        """
        if not window_config.get("ENABLED", False):
            return True, "窗口控制未启用"

        time_range = window_config.get("TIME_RANGE", {})
        start_time = time_range.get("START", "00:00")
        end_time = time_range.get("END", "23:59")

        # 检查时间范围
        if not self.is_in_time_range(start_time, end_time):
            now = self.get_time_func()
            return False, f"当前时间 {now.strftime('%H:%M')} 不在窗口 {start_time}-{end_time} 内"

        # 检查 once_per_day
        if window_config.get("ONCE_PER_DAY", False) and check_once_per_day_func:
            if check_once_per_day_func():
                return False, "今天已执行过"
            else:
                print(f"[{self.window_name}] 今天首次执行")

        return True, "在窗口内"

    def get_status(self, window_config: dict, check_once_per_day_func=None) -> dict:
        """
        获取窗口状态信息

        Args:
            window_config: 窗口配置
            check_once_per_day_func: 检查今天是否已执行的函数

        Returns:
            状态信息字典
        """
        now = self.get_time_func()
        status = {
            "enabled": window_config.get("ENABLED", False),
            "current_time": now.strftime("%H:%M:%S"),
            "current_date": now.strftime("%Y-%m-%d"),
            "timezone": str(now.tzinfo),
        }

        if status["enabled"]:
            time_range = window_config.get("TIME_RANGE", {})
            status["window_start"] = time_range.get("START", "00:00")
            status["window_end"] = time_range.get("END", "23:59")
            status["in_window"] = self.is_in_time_range(
                status["window_start"], status["window_end"]
            )
            status["once_per_day"] = window_config.get("ONCE_PER_DAY", False)

            if status["once_per_day"] and check_once_per_day_func:
                status["executed_today"] = check_once_per_day_func()

        return status