|
|
@@ -6,7 +6,7 @@
|
|
|
"""
|
|
|
|
|
|
from datetime import datetime
|
|
|
-from typing import Optional, Tuple
|
|
|
+from typing import Optional
|
|
|
|
|
|
import pytz
|
|
|
|
|
|
@@ -285,161 +285,3 @@ def calculate_days_old(iso_time: str, timezone: str = DEFAULT_TIMEZONE) -> Optio
|
|
|
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
|
sansan