feat: 新增节假日缓存,仅工作日生成与推送

- daily/holiday.py: 抓取 xiaoai.me 全年节假日并缓存到 .cache/holidays-<year>.json,
  is_workday() 判定(法定节假日/周末休息,调休补班日上班),联网失败回退本地缓存
- scheduler: tick_once 前置工作日闸门,非工作日标记完成并跳过,判定失败降级为正常工作日避免漏跑
- config/.env.example: 新增 DAILY_WORKDAY_ONLY 开关(默认开)
- tests: test_holiday 8 例 + scheduler 非工作日闸门 2 例

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-23 17:51:09 +08:00
parent d66f2c716c
commit 36085f107d
6 changed files with 232 additions and 0 deletions

106
daily/holiday.py Normal file
View File

@@ -0,0 +1,106 @@
"""法定节假日获取、缓存与工作日判定。
数据来自 xiaoai.me 公共接口,一次性抓取全年并缓存到 .cache/
后续判定只读本地缓存;缓存缺失或过期时才重新请求。
"""
from __future__ import annotations
import json
import logging
import ssl
import urllib.request
from datetime import date
from pathlib import Path
from daily.config import CACHE_DIR
logger = logging.getLogger(__name__)
_API_URL = "https://publicapi.xiaoai.me/holiday/year?date={year}"
_TIMEOUT = 15
def _cache_file(year: int) -> Path:
return CACHE_DIR / f"holidays-{year}.json"
def _ssl_context() -> ssl.SSLContext:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
return ctx
def _fetch_year(year: int) -> dict[str, dict]:
"""请求全年节假日,返回 {date_str: {"rest": bool, "name": str}}。
rest=1 表示休息(法定节假日/调休放假rest=0 表示调休补班(需上班)。
"""
url = _API_URL.format(year=year)
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=_TIMEOUT, context=_ssl_context()) as resp:
payload = json.loads(resp.read().decode("utf-8"))
items = payload.get("data") or []
result: dict[str, dict] = {}
for item in items:
day = item.get("date")
if not day:
continue
result[day] = {
"rest": bool(item.get("rest", 0)),
"name": item.get("holiday", ""),
}
if not result:
raise RuntimeError(f"节假日接口返回为空 year={year}")
return result
def load_holidays(year: int, *, refresh: bool = False) -> dict[str, dict]:
"""加载全年节假日缓存;缺失或 refresh 时联网抓取并写入缓存。
联网失败时若已有缓存则回退用缓存,保证离线可用。
"""
cache = _cache_file(year)
if cache.exists() and not refresh:
try:
raw = json.loads(cache.read_text(encoding="utf-8"))
if isinstance(raw, dict) and raw:
return raw
except (OSError, ValueError):
logger.warning("节假日缓存损坏,重新获取 %s", cache)
try:
data = _fetch_year(year)
except Exception as exc:
if cache.exists():
logger.warning("节假日获取失败,回退到本地缓存:%s", exc)
return json.loads(cache.read_text(encoding="utf-8"))
raise
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
logger.info("已缓存 %d 年节假日 %d 条 -> %s", year, len(data), cache)
return data
def is_workday(day: date, holidays: dict[str, dict] | None = None) -> bool:
"""判断是否为工作日。
规则调休补班日rest=0 且在表中)→ 上班;
法定节假日/调休放假rest=1→ 休息;
周六日 → 休息;其余 → 上班。
"""
holidays = holidays if holidays is not None else load_holidays(day.year)
key = day.isoformat()
entry = holidays.get(key)
if entry is not None:
return not entry["rest"]
return day.weekday() < 5 # 周一~周五
def workday_name(day: date, holidays: dict[str, dict] | None = None) -> str | None:
"""返回该天的节假日名(休息或调休补班),无则 None。用于日志。"""
holidays = holidays if holidays is not None else load_holidays(day.year)
entry = holidays.get(day.isoformat())
return entry["name"] if entry else None