From 36085f107d5d30904533474e1c6d42052c6b3b93 Mon Sep 17 00:00:00 2001 From: yumao Date: Thu, 23 Jul 2026 17:51:09 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E8=8A=82=E5=81=87?= =?UTF-8?q?=E6=97=A5=E7=BC=93=E5=AD=98=EF=BC=8C=E4=BB=85=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=97=A5=E7=94=9F=E6=88=90=E4=B8=8E=E6=8E=A8=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - daily/holiday.py: 抓取 xiaoai.me 全年节假日并缓存到 .cache/holidays-.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) --- .env.example | 3 ++ daily/config.py | 5 ++ daily/holiday.py | 106 ++++++++++++++++++++++++++++++++++++++++ daily/scheduler.py | 23 +++++++++ tests/test_holiday.py | 57 +++++++++++++++++++++ tests/test_scheduler.py | 38 ++++++++++++++ 6 files changed, 232 insertions(+) create mode 100644 daily/holiday.py create mode 100644 tests/test_holiday.py diff --git a/.env.example b/.env.example index 6e9fb88..3f47934 100644 --- a/.env.example +++ b/.env.example @@ -69,6 +69,9 @@ DAILY_NEWS_DEDUP_DAYS=7 DAILY_SCHEDULE_TZ=Asia/Shanghai DAILY_SCHEDULE_GENERATE_AT=08:50 DAILY_SCHEDULE_PUSH_AT=09:00 +# 仅工作日生成/推送（法定节假日、周末跳过；调休补班日照常） +# 节假日数据取自 xiaoai.me，缓存于 .cache/holidays-.json，每年首次自动获取一次 +DAILY_WORKDAY_ONLY=1 # 编辑指定今日首推（可选）：关键词，或 关键词|URL # Python Step 0 检索 → featured.json；Agent / classic 企微「今日首推」优先使用 diff --git a/daily/config.py b/daily/config.py index 8aed4f6..b5b0d33 100644 --- a/daily/config.py +++ b/daily/config.py @@ -167,3 +167,8 @@ def narrative_axis_days() -> int: def news_backfill_enabled() -> bool: return env_bool("DAILY_NEWS_BACKFILL", False) + + +def workday_only() -> bool: + """仅工作日生成/推送；法定节假日与周末跳过（调休补班日照常）。""" + return env_bool("DAILY_WORKDAY_ONLY", True) diff --git a/daily/holiday.py b/daily/holiday.py new file mode 100644 index 0000000..60388a4 --- /dev/null +++ b/daily/holiday.py @@ -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 diff --git a/daily/scheduler.py b/daily/scheduler.py index dd27866..ceb4384 100644 --- a/daily/scheduler.py +++ b/daily/scheduler.py @@ -20,7 +20,9 @@ from daily.config import ( schedule_generate_at, schedule_push_at, schedule_timezone_name, + workday_only, ) +from daily.holiday import is_workday, load_holidays, workday_name logger = logging.getLogger(__name__) @@ -200,6 +202,27 @@ def tick_once( push_at = push_at or parse_hhmm(schedule_push_at()) today_str = now.astimezone(tz).date().isoformat() + + # 仅工作日运行：非工作日（法定节假日/周末，调休补班日除外）跳过当日 generate/push。 + # 将两者标记为已完成，避免 plan_next_action 在当天反复补跑。 + if workday_only() and state.last_generate_date != today_str: + today = now.astimezone(tz).date() + try: + holidays = load_holidays(today.year) + if not is_workday(today, holidays): + name = workday_name(today, holidays) or "周末" + logger.info("今日 %s 非工作日（%s），跳过生成与推送", today_str, name) + state.last_generate_date = today_str + state.last_push_date = today_str + if not dry_run: + state.save() + if dry_run: + logger.info("[dry-run] 非工作日，将跳过") + return state + except Exception as exc: + # 节假日数据不可用时按常规工作日处理，避免因接口故障漏跑 + logger.warning("工作日判定失败（%s），按正常工作日处理", exc) + planned = plan_next_action( now=now, tz=tz, diff --git a/tests/test_holiday.py b/tests/test_holiday.py new file mode 100644 index 0000000..97336dd --- /dev/null +++ b/tests/test_holiday.py @@ -0,0 +1,57 @@ +"""Tests for daily.holiday.""" + +from __future__ import annotations + +import unittest +from datetime import date + +from daily.holiday import is_workday, workday_name + +# 模拟 2026 节假日表：1/1 元旦(休)，1/4 元旦调休(补班) +HOLIDAYS = { + "2026-01-01": {"rest": True, "name": "元旦节"}, + "2026-01-02": {"rest": True, "name": "元旦节"}, + "2026-01-03": {"rest": True, "name": "元旦节"}, + "2026-01-04": {"rest": False, "name": "元旦节调休"}, + "2026-02-16": {"rest": True, "name": "春节"}, + "2026-02-14": {"rest": False, "name": "春节调休"}, +} + + +class IsWorkdayTests(unittest.TestCase): + def test_legal_holiday_is_rest(self): + # 2026-01-01 周四，元旦 → 休息 + self.assertFalse(is_workday(date(2026, 1, 1), HOLIDAYS)) + + def test_tiaoxiu_makeup_day_is_workday(self): + # 2026-01-04 周日，调休补班 → 上班 + self.assertEqual(date(2026, 1, 4).weekday(), 6) + self.assertTrue(is_workday(date(2026, 1, 4), HOLIDAYS)) + + def test_spring_festival_holiday(self): + self.assertFalse(is_workday(date(2026, 2, 16), HOLIDAYS)) + + def test_spring_festival_makeup_saturday(self): + # 2026-02-14 周六，调休 → 上班 + self.assertEqual(date(2026, 2, 14).weekday(), 5) + self.assertTrue(is_workday(date(2026, 2, 14), HOLIDAYS)) + + def test_normal_weekday(self): + # 2026-07-23 周四，非节假日 → 上班 + self.assertTrue(is_workday(date(2026, 7, 23), HOLIDAYS)) + + def test_normal_weekend(self): + # 2026-07-25 周六，非节假日 → 休息 + self.assertFalse(is_workday(date(2026, 7, 25), HOLIDAYS)) + + +class WorkdayNameTests(unittest.TestCase): + def test_holiday_name(self): + self.assertEqual(workday_name(date(2026, 1, 1), HOLIDAYS), "元旦节") + + def test_normal_day_has_no_name(self): + self.assertIsNone(workday_name(date(2026, 7, 23), HOLIDAYS)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 34ddbd4..39adfcd 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -4,6 +4,7 @@ from __future__ import annotations import unittest from datetime import datetime +from unittest import mock from zoneinfo import ZoneInfo from daily.scheduler import ( @@ -12,6 +13,7 @@ from daily.scheduler import ( next_occurrence_after, parse_hhmm, plan_next_action, + tick_once, ) @@ -116,3 +118,39 @@ class NextOccurrenceTests(unittest.TestCase): nxt = next_occurrence_after(ClockTime(8, 50), tz, now) self.assertEqual(nxt.date().isoformat(), "2026-07-10") self.assertEqual((nxt.hour, nxt.minute), (8, 50)) + + +class WorkdayGateTests(unittest.TestCase): + """非工作日 tick_once 直接标记完成并跳过，不触发 generate。""" + + def setUp(self) -> None: + self.tz = ZoneInfo("Asia/Shanghai") + self.gen = ClockTime(8, 50) + self.push = ClockTime(9, 0) + + def _tick(self, day, workday): + now = datetime(day.year, day.month, day.day, 8, 30, tzinfo=self.tz) + holidays = {} if workday else {day.isoformat(): {"rest": True, "name": "测试假"}} + with mock.patch("daily.scheduler.load_holidays", return_value=holidays), \ + mock.patch("daily.scheduler.run_scheduled_action") as run: + state = tick_once( + now=now, tz=self.tz, state=SchedulerState(), + generate_at=self.gen, push_at=self.push, dry_run=True, + ) + return state, run + + def test_holiday_skips_and_marks_done(self): + # 2026-07-25 是周六；强制为非工作日 + from datetime import date + day = date(2026, 7, 25) + state, run = self._tick(day, workday=False) + self.assertFalse(run.called) + self.assertEqual(state.last_generate_date, "2026-07-25") + self.assertEqual(state.last_push_date, "2026-07-25") + + def test_workday_proceeds(self): + from datetime import date + day = date(2026, 7, 23) # 周四，工作日 + state, run = self._tick(day, workday=True) + # 工作日不提前标记完成（dry-run 不执行动作，故 last_generate_date 仍为 None） + self.assertIsNone(state.last_generate_date)