feat: 添加中国节假日感知定时功能

- 新增 chinese_holidays 表,通过 timor.tech API 同步节假日数据
- 修正 holiday 字段解读:holiday=true → 休息日,holiday=false → 调休工作日
- 工作日 8:00-22:00 每小时爬取,周末/节假日/夜间自动跳过
- 新增 /api/v1/holidays/sync 和 /api/v1/holidays/today 接口
- 企微菜单新增「同步节假日」按钮,支持手动触发同步
This commit is contained in:
2026-05-09 18:27:27 +08:00
parent c3e06c997f
commit 1f18d2ec87
12 changed files with 381 additions and 15 deletions
+1 -1
View File
@@ -23,7 +23,7 @@ WECHAT_HOST=0.0.0.0
# 定时任务 # 定时任务
SCHEDULER_ENABLED=true SCHEDULER_ENABLED=true
SCHEDULER_CRON=0 8,14,18 * * * SCHEDULER_CRON=0 8-21 * * *
# Markdown # Markdown
MARKDOWN_ENABLED=true MARKDOWN_ENABLED=true
+1
View File
@@ -2,6 +2,7 @@ import asyncio
from alembic import context from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.ext.asyncio import create_async_engine
from app.models.announcement import Base from app.models.announcement import Base
from app.models.holiday import ChineseHoliday # noqa: F401
from app.config import settings from app.config import settings
target_metadata = Base.metadata target_metadata = Base.metadata
@@ -0,0 +1,37 @@
"""add_chinese_holidays_table
Revision ID: 1f59799a5083
Revises: eed20ee8cc26
Create Date: 2026-05-09 18:14:31.144256
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '1f59799a5083'
down_revision: Union[str, Sequence[str], None] = 'eed20ee8cc26'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
op.create_table(
'chinese_holidays',
sa.Column('date', sa.Date(), nullable=False),
sa.Column('is_workday', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('year', sa.Integer(), nullable=False),
sa.Column('description', sa.String(length=100), nullable=False, server_default=''),
sa.PrimaryKeyConstraint('date'),
)
op.create_index('idx_chinese_holidays_year', 'chinese_holidays', ['year'])
def downgrade() -> None:
"""Downgrade schema."""
op.drop_index('idx_chinese_holidays_year', table_name='chinese_holidays')
op.drop_table('chinese_holidays')
+30
View File
@@ -0,0 +1,30 @@
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_db
from app.services.holiday_service import now_in_china, sync_holidays
router = APIRouter()
@router.post("/holidays/sync")
async def sync_holidays_endpoint(
year: int | None = Query(None, description="同步年份,默认当前年份"),
db: AsyncSession = Depends(get_db),
):
if year is None:
year = now_in_china().year
count = await sync_holidays(db, year)
return {"status": "ok", "year": year, "synced": count}
@router.get("/holidays/today")
async def get_today_status(db: AsyncSession = Depends(get_db)):
from app.services.holiday_service import is_workday
today = now_in_china()
workday = await is_workday(db, today)
return {
"date": today.isoformat(),
"is_workday": workday,
"message": "工作日,正常爬取" if workday else "非工作日,跳过爬取",
}
+2 -1
View File
@@ -1,10 +1,11 @@
from fastapi import APIRouter from fastapi import APIRouter
from app.api import announcements, crawl, wechat from app.api import announcements, crawl, holidays, wechat
from app.api import scheduler as scheduler_module from app.api import scheduler as scheduler_module
api_router = APIRouter(prefix="/api/v1") api_router = APIRouter(prefix="/api/v1")
api_router.include_router(announcements.router, tags=["announcements"]) api_router.include_router(announcements.router, tags=["announcements"])
api_router.include_router(crawl.router, tags=["crawl"]) api_router.include_router(crawl.router, tags=["crawl"])
api_router.include_router(holidays.router, tags=["holidays"])
api_router.include_router(wechat.router, tags=["wechat"]) api_router.include_router(wechat.router, tags=["wechat"])
api_router.include_router(scheduler_module.router) api_router.include_router(scheduler_module.router)
+2 -2
View File
@@ -52,7 +52,7 @@ async def wechat_callback(request: Request):
event = xml_tree.find("Event") event = xml_tree.find("Event")
event_key = xml_tree.find("EventKey") event_key = xml_tree.find("EventKey")
from_user = xml_tree.find("FromUserName") from_user = xml_tree.find("FromUserName")
handler.handle_event( await handler.handle_event(
event.text if event is not None else "", event.text if event is not None else "",
event_key.text if event_key is not None else None, event_key.text if event_key is not None else None,
from_user.text if from_user is not None else "", from_user.text if from_user is not None else "",
@@ -60,7 +60,7 @@ async def wechat_callback(request: Request):
elif msg_type == "text": elif msg_type == "text":
content = xml_tree.find("Content") content = xml_tree.find("Content")
from_user = xml_tree.find("FromUserName") from_user = xml_tree.find("FromUserName")
handler.handle_text( await handler.handle_text(
content.text if content is not None else "", content.text if content is not None else "",
from_user.text if from_user is not None else "", from_user.text if from_user is not None else "",
) )
+1 -1
View File
@@ -31,7 +31,7 @@ class Settings(BaseSettings):
# 定时任务 # 定时任务
scheduler_enabled: bool = True scheduler_enabled: bool = True
scheduler_cron: str = "0 8,14,18 * * *" scheduler_cron: str = "0 8-21 * * *"
# Markdown # Markdown
markdown_enabled: bool = True markdown_enabled: bool = True
+15
View File
@@ -0,0 +1,15 @@
from datetime import date
from sqlalchemy import Boolean, Date, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from app.models.announcement import Base
class ChineseHoliday(Base):
__tablename__ = "chinese_holidays"
date: Mapped[date] = mapped_column(Date, primary_key=True)
is_workday: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
year: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
description: Mapped[str] = mapped_column(String(100), nullable=False, default="")
+21 -8
View File
@@ -1,30 +1,43 @@
import logging import logging
from datetime import datetime, time from datetime import datetime
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.cron import CronTrigger
from app.api.deps import get_crawl_service from app.api.deps import get_db, get_crawl_service
from app.config import settings from app.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
scheduler = AsyncIOScheduler() scheduler = AsyncIOScheduler()
NIGHT_START = time(22, 0) NIGHT_START = 22 # 22:00
NIGHT_END = time(6, 0) NIGHT_END = 8 # 08:00
TZ = ZoneInfo("Asia/Shanghai") TZ = ZoneInfo("Asia/Shanghai")
def _is_night_time() -> bool: def _is_night_time() -> bool:
"""22:00 ~ 次日 06:00 夜间时段""" """22:00 ~ 次日 08:00 夜间时段"""
current = datetime.now(TZ).time() current = datetime.now(TZ).hour
return current >= NIGHT_START or current < NIGHT_END return current >= NIGHT_START or current < NIGHT_END
async def scheduled_crawl(): async def _should_skip() -> bool:
"""检查是否应该跳过爬取"""
if _is_night_time(): if _is_night_time():
logger.info("夜间时段 (22:00-06:00),跳过爬取") logger.info("夜间时段 (22:00-08:00),跳过爬取")
return True
from app.services.holiday_service import is_workday
async for db in get_db():
if not await is_workday(db):
logger.info("非工作日,跳过爬取")
return True
return False
async def scheduled_crawl():
if await _should_skip():
return return
logger.info("开始定时爬取任务") logger.info("开始定时爬取任务")
+101
View File
@@ -0,0 +1,101 @@
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)
+83 -2
View File
@@ -1,7 +1,11 @@
import logging
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from app.config import settings from app.config import settings
from app.wechat.crypto import WXBizMsgCrypt from app.wechat.crypto import WXBizMsgCrypt
from app.wechat.client import WeChatClient
logger = logging.getLogger(__name__)
class WeChatMessageHandler: class WeChatMessageHandler:
@@ -11,6 +15,7 @@ class WeChatMessageHandler:
sEncodingAESKey=settings.wechat_encoding_aes_key, sEncodingAESKey=settings.wechat_encoding_aes_key,
sReceiveId=settings.wechat_corp_id, sReceiveId=settings.wechat_corp_id,
) )
self.client = WeChatClient()
def verify_url( def verify_url(
self, msg_signature: str, timestamp: str, nonce: str, echostr: str self, msg_signature: str, timestamp: str, nonce: str, echostr: str
@@ -48,10 +53,86 @@ class WeChatMessageHandler:
return encrypted return encrypted
return None return None
def handle_event( async def handle_event(
self, event: str, event_key: str | None, from_user: str self, event: str, event_key: str | None, from_user: str
) -> str | None: ) -> str | None:
if event != "click" or not event_key:
return None return None
def handle_text(self, content: str, from_user: str) -> str | None: if event_key == "today_stats":
return await self._handle_today_stats(from_user)
elif event_key == "trigger_crawl":
return await self._handle_trigger_crawl(from_user)
elif event_key == "sync_holidays":
return await self._handle_sync_holidays(from_user)
return None return None
async def handle_text(self, content: str, from_user: str) -> str | None:
return None
async def _handle_today_stats(self, from_user: str) -> str | None:
from app.api.deps import get_db
try:
async for db in get_db():
from sqlalchemy import func, select
from app.models.announcement import Announcement
total_result = await db.execute(
select(func.count()).select_from(Announcement)
)
total = total_result.scalar() or 0
today_result = await db.execute(
select(func.count()).where(
func.date(Announcement.publish_date) == func.current_date()
).select_from(Announcement)
)
today = today_result.scalar() or 0
text = f"今日新增: {today}\n累计公告: {total}"
await self.client.send_text(text, from_user)
except Exception as e:
logger.error(f"查询统计失败: {e}")
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_trigger_crawl(self, from_user: str) -> str | None:
from app.api.deps import get_crawl_service
await self.client.send_text("开始爬取,请稍候...", from_user)
try:
service = get_crawl_service()
results = await service.run_all()
total = sum(r.total_count for r in results)
stored = sum(
r.pipeline_result.stored for r in results
if r.pipeline_result
)
notified = sum(
r.pipeline_result.notified for r in results
if r.pipeline_result
)
errors = [r.error_message for r in results if not r.success]
msg = f"爬取完成\n抓取: {total}\n新增: {stored}\n推送: {notified}"
if errors:
msg += f"\n异常: {errors[0][:50]}"
await self.client.send_text(msg, from_user)
except Exception as e:
logger.error(f"手动爬取失败: {e}")
await self.client.send_text(f"爬取失败: {e}", from_user)
async def _handle_sync_holidays(self, from_user: str) -> str | None:
from app.services.holiday_service import now_in_china, sync_holidays
from app.api.deps import get_db
try:
async for db in get_db():
year = now_in_china().year
count = await sync_holidays(db, year)
await self.client.send_text(
f"已同步 {year} 年节假日\n{count} 条记录", from_user,
)
except Exception as e:
logger.error(f"同步节假日失败: {e}")
await self.client.send_text(f"同步失败: {e}", from_user)
+87
View File
@@ -0,0 +1,87 @@
import logging
import httpx
from app.config import settings
logger = logging.getLogger(__name__)
MENU = {
"button": [
{
"name": "今日公告",
"type": "click",
"key": "today_stats",
},
{
"name": "系统管理",
"sub_button": [
{
"name": "立即爬取",
"type": "click",
"key": "trigger_crawl",
},
{
"name": "同步节假日",
"type": "click",
"key": "sync_holidays",
},
],
},
]
}
class MenuManager:
def __init__(self, client=None):
self.client = client
async def _get_token(self) -> str | None:
from app.wechat.client import WeChatClient
c = self.client or WeChatClient()
return await c._get_access_token()
async def create(self) -> bool:
token = await self._get_token()
if not token:
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/create"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
response = await client.post(url, params=params, json=MENU)
data = response.json()
if data.get("errcode") == 0:
logger.info("企微菜单创建成功")
return True
logger.error(f"企微菜单创建失败: {data}")
return False
async def delete(self) -> bool:
token = await self._get_token()
if not token:
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/delete"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
response = await client.get(url, params=params)
data = response.json()
return data.get("errcode") == 0
async def get(self) -> dict | None:
token = await self._get_token()
if not token:
return None
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/get"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
response = await client.get(url, params=params)
data = response.json()
if data.get("errcode") == 0:
return data
return None