# 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