v6ole
1f18d2ec87
- 新增 chinese_holidays 表，通过 timor.tech API 同步节假日数据 - 修正 holiday 字段解读：holiday=true → 休息日，holiday=false → 调休工作日 - 工作日 8:00-22:00 每小时爬取，周末/节假日/夜间自动跳过 - 新增 /api/v1/holidays/sync 和 /api/v1/holidays/today 接口 - 企微菜单新增「同步节假日」按钮，支持手动触发同步
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
import logging
|
|
from datetime import date
|
|
|
|
import httpx
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.holiday import ChineseHoliday
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
HOLIDAY_API = "http://timor.tech/api/holiday/year"
|
|
|
|
|
|
def now_in_china() -> date:
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
return datetime.now(ZoneInfo("Asia/Shanghai")).date()
|
|
|
|
|
|
async def sync_holidays(db: AsyncSession, year: int) -> int:
|
|
"""同步中国节假日数据，返回更新的记录数"""
|
|
url = f"{HOLIDAY_API}/{year}"
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
response = await client.get(url)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
logger.error(f"节假日 API 返回错误: {data}")
|
|
return 0
|
|
|
|
holidays = data.get("holiday", {})
|
|
if not holidays:
|
|
return 0
|
|
|
|
# 也标记周末（周六日但非调休工作日）
|
|
from datetime import timedelta
|
|
current = date(year, 1, 1)
|
|
end = date(year, 12, 31)
|
|
|
|
records: dict[date, dict] = {}
|
|
while current <= end:
|
|
dow = current.weekday() # 0=Mon, 6=Sun
|
|
# 默认：周一~五为工作日，周六日为非工作日
|
|
default_workday = dow < 5
|
|
records[current] = {
|
|
"date": current,
|
|
"is_workday": default_workday,
|
|
"year": year,
|
|
"description": "",
|
|
}
|
|
current += timedelta(days=1)
|
|
|
|
# 覆盖节假日数据
|
|
# holiday=true → 休息日（无论 wage 值）
|
|
# holiday=false → 调休补班日（周末也要上班）
|
|
for date_str, info_str in holidays.items():
|
|
d = date.fromisoformat(f"{year}-{date_str}")
|
|
if d.year != year:
|
|
continue
|
|
info = info_str if isinstance(info_str, dict) else {}
|
|
holiday = info.get("holiday", False)
|
|
name = info.get("name", "")
|
|
|
|
records[d]["description"] = name
|
|
records[d]["is_workday"] = not holiday # holiday=false → 调休工作日
|
|
|
|
# Upsert
|
|
values = list(records.values())
|
|
stmt = pg_insert(ChineseHoliday).values(values)
|
|
stmt = stmt.on_conflict_do_update(
|
|
index_elements=["date"],
|
|
set_={"is_workday": stmt.excluded.is_workday,
|
|
"description": stmt.excluded.description},
|
|
)
|
|
result = await db.execute(stmt)
|
|
await db.commit()
|
|
logger.info(f"已同步 {year} 年节假日，{len(values)} 天")
|
|
return result.rowcount
|
|
|
|
|
|
async def is_workday(db: AsyncSession, day: date | None = None) -> bool:
|
|
"""判断某天是否为工作日"""
|
|
if day is None:
|
|
day = now_in_china()
|
|
from sqlalchemy import select
|
|
result = await db.execute(
|
|
select(ChineseHoliday.is_workday).where(ChineseHoliday.date == day)
|
|
)
|
|
row = result.fetchone()
|
|
if row is None:
|
|
# 无数据时，按周判断
|
|
return day.weekday() < 5
|
|
return row[0]
|
|
|
|
|
|
async def was_yesterday_workday(db: AsyncSession) -> bool:
|
|
"""昨天是工作日吗"""
|
|
from datetime import timedelta
|
|
yesterday = now_in_china() - timedelta(days=1)
|
|
return await is_workday(db, yesterday)
|