# -*- coding: utf-8 -*- """上海时区与每日任务时间计算。 B 站数据口径使用 ``Asia/Shanghai``。中国自 1991 年起不再使用夏令时,因此这里 用固定 UTC+8,避免 Windows 上依赖未安装的 ``tzdata`` 包。 """ from __future__ import annotations import re import time from datetime import datetime, timedelta, timezone SHANGHAI_TZ = timezone(timedelta(hours=8)) _HHMM_RE = re.compile(r"^([01]\d|2[0-3]):([0-5]\d)$") _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") def shanghai_datetime(ts: float | None = None) -> datetime: return datetime.fromtimestamp(time.time() if ts is None else ts, SHANGHAI_TZ) def shanghai_date(ts: float | None = None) -> str: return shanghai_datetime(ts).strftime("%Y-%m-%d") def parse_hhmm(value: str) -> tuple[int, int]: match = _HHMM_RE.match(value or "") if not match: raise ValueError(f"时间必须是 HH:MM(24 小时制): {value!r}") return int(match.group(1)), int(match.group(2)) def next_daily_run(now: float, hhmm: str) -> float: """返回 ``now`` 之后(严格大于)最近的 ``hh:mm`` 上海时间时间戳。""" hour, minute = parse_hhmm(hhmm) current = shanghai_datetime(now) target = current.replace(hour=hour, minute=minute, second=0, microsecond=0) if target.timestamp() <= now: target += timedelta(days=1) return target.timestamp() def is_date(value: str) -> bool: if not _DATE_RE.match(value or ""): return False try: datetime.strptime(value, "%Y-%m-%d") except ValueError: return False return True def day_bounds(date: str) -> tuple[float, float]: """返回上海时区某一天的 [start, end) 时间戳。""" if not is_date(date): raise ValueError(f"日期格式必须是 YYYY-MM-DD: {date!r}") start = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ) return start.timestamp(), (start + timedelta(days=1)).timestamp()