66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""中国节假日日历只读服务"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.cn_holiday import SysCnHolidayDay
|
|
|
|
|
|
def is_rest_day(db: Session, d: date) -> bool:
|
|
"""查表判定是否休息日；无行则按周六日回退。"""
|
|
row: Optional[SysCnHolidayDay] = (
|
|
db.query(SysCnHolidayDay).filter(SysCnHolidayDay.day_date == d).first()
|
|
)
|
|
if row is None:
|
|
return d.weekday() >= 5
|
|
if row.is_rest_day is not None:
|
|
return int(row.is_rest_day) == 1
|
|
return int(row.day_type or 0) in (1, 2, 3)
|
|
|
|
|
|
def list_day_flags(db: Session, start: date, end: date) -> List[Dict[str, Any]]:
|
|
"""返回 [start, end] 闭区间每日标志；缺行用周末回退补齐。"""
|
|
if end < start:
|
|
return []
|
|
|
|
rows = (
|
|
db.query(SysCnHolidayDay)
|
|
.filter(SysCnHolidayDay.day_date >= start, SysCnHolidayDay.day_date <= end)
|
|
.all()
|
|
)
|
|
by_date = {r.day_date: r for r in rows}
|
|
|
|
out: List[Dict[str, Any]] = []
|
|
cur = start
|
|
while cur <= end:
|
|
row = by_date.get(cur)
|
|
if row is not None:
|
|
rest = (
|
|
int(row.is_rest_day) == 1
|
|
if row.is_rest_day is not None
|
|
else int(row.day_type or 0) in (1, 2, 3)
|
|
)
|
|
out.append(
|
|
{
|
|
"dayDate": cur.isoformat(),
|
|
"dayType": int(row.day_type),
|
|
"isRestDay": rest,
|
|
"dayName": row.day_name,
|
|
}
|
|
)
|
|
else:
|
|
is_weekend = cur.weekday() >= 5
|
|
out.append(
|
|
{
|
|
"dayDate": cur.isoformat(),
|
|
"dayType": 1 if is_weekend else 0,
|
|
"isRestDay": is_weekend,
|
|
"dayName": None,
|
|
}
|
|
)
|
|
cur += timedelta(days=1)
|
|
return out
|