v6ole
7dcb22f573
- 新增 holidays.py 服务：按年获取节假日数据，缓存到 system_config - 区分 holiday（工作日假日）和 workday（调休上班日）两种类型 - is_working_day() 支持调休判断：周末在 workdays 集合中视为工作日 - 启动时自动从 API 刷新当年+明年数据，失败则使用缓存 - 系统设置新增「从 API 刷新」按钮，支局长可手动触发 - 支持 2010-2026 年数据，API 免费免鉴权 Co-Authored-By: Claude <noreply@anthropic.com>
137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
"""Chinese holiday data service — fetches from apisbo.com and caches to system_config."""
|
|
|
|
import json
|
|
import httpx
|
|
from datetime import date
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.system_config import SystemConfig
|
|
|
|
|
|
HOLIDAY_API = "https://api.apisbo.com/holidays/year"
|
|
|
|
|
|
async def _get_config(db: AsyncSession, key: str) -> str | None:
|
|
row = await db.get(SystemConfig, key)
|
|
return row.value if row else None
|
|
|
|
|
|
async def _set_config(db: AsyncSession, key: str, value: str):
|
|
row = await db.get(SystemConfig, key)
|
|
if row:
|
|
row.value = value
|
|
else:
|
|
db.add(SystemConfig(key=key, value=value))
|
|
await db.commit()
|
|
|
|
|
|
async def refresh_holidays(db: AsyncSession, year: int | None = None) -> dict:
|
|
"""Fetch Chinese holidays for a year from apisbo.com and cache to DB.
|
|
|
|
Caches two config keys:
|
|
- holidays: comma-separated rest-day dates (holidays + weekends already handled by weekday check)
|
|
We only store holiday dates here, since weekends are auto-detected.
|
|
- workdays: comma-separated makeup workday dates (调休, when Sat/Sun becomes a workday)
|
|
|
|
Returns dict with counts.
|
|
"""
|
|
year = year or date.today().year
|
|
|
|
# Fetch current year + next year on first load
|
|
years_to_fetch = {year}
|
|
existing = await _get_config(db, f"holidays_{year}")
|
|
if not existing:
|
|
years_to_fetch.add(year + 1)
|
|
|
|
all_holidays: list[str] = []
|
|
all_workdays: list[str] = []
|
|
|
|
for y in years_to_fetch:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
resp = await client.get(f"{HOLIDAY_API}/{y}")
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
|
|
if result.get("code") != 0:
|
|
continue
|
|
|
|
holidays = []
|
|
workdays = []
|
|
for item in result.get("data", []):
|
|
d = item["date"]
|
|
if item["type"] == "holiday" and date.fromisoformat(d).weekday() < 5:
|
|
# Only cache weekday-holidays (weekend holidays are already skipped)
|
|
holidays.append(d)
|
|
elif item["type"] == "workday":
|
|
# Makeup workday — Saturday/Sunday that becomes a workday
|
|
workdays.append(d)
|
|
|
|
# Store per-year for reference, and accumulate for runtime use
|
|
await _set_config(db, f"holidays_{y}", ",".join(holidays))
|
|
await _set_config(db, f"workdays_{y}", ",".join(workdays))
|
|
all_holidays.extend(holidays)
|
|
all_workdays.extend(workdays)
|
|
|
|
except Exception:
|
|
# API unavailable — fall back to existing cached data
|
|
cached_h = await _get_config(db, f"holidays_{y}")
|
|
if cached_h:
|
|
all_holidays.extend([d for d in cached_h.split(",") if d.strip()])
|
|
cached_w = await _get_config(db, f"workdays_{y}")
|
|
if cached_w:
|
|
all_workdays.extend([d for d in cached_w.split(",") if d.strip()])
|
|
|
|
# Write runtime config keys (used by is_working_day)
|
|
if all_holidays or all_workdays:
|
|
await _set_config(db, "holidays", ",".join(sorted(all_holidays)))
|
|
await _set_config(db, "workdays", ",".join(sorted(all_workdays)))
|
|
|
|
return {
|
|
"holidays": len(all_holidays),
|
|
"workdays": len(all_workdays),
|
|
"years": sorted(years_to_fetch),
|
|
}
|
|
|
|
|
|
async def load_holiday_sets(db: AsyncSession) -> tuple[set[date], set[date]]:
|
|
"""Load holiday and workday date sets from cached config.
|
|
|
|
Returns (holidays_set, workdays_set).
|
|
- holidays_set: dates that are rest days (weekday holidays)
|
|
- workdays_set: dates that are workdays despite being weekends (调休)
|
|
"""
|
|
holidays: set[date] = set()
|
|
workdays: set[date] = set()
|
|
|
|
# Try runtime config first
|
|
h_val = await _get_config(db, "holidays")
|
|
if h_val:
|
|
for s in h_val.split(","):
|
|
s = s.strip()
|
|
if s:
|
|
try:
|
|
holidays.add(date.fromisoformat(s))
|
|
except ValueError:
|
|
pass
|
|
|
|
w_val = await _get_config(db, "workdays")
|
|
if w_val:
|
|
for s in w_val.split(","):
|
|
s = s.strip()
|
|
if s:
|
|
try:
|
|
workdays.add(date.fromisoformat(s))
|
|
except ValueError:
|
|
pass
|
|
|
|
# If no data cached yet, try to refresh
|
|
if not holidays and not workdays:
|
|
try:
|
|
await refresh_holidays(db)
|
|
# Reload after refresh
|
|
return await load_holiday_sets(db)
|
|
except Exception:
|
|
pass
|
|
|
|
return holidays, workdays
|