File size: 1,663 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """Finance app定时任务.
提供LPR数据自动同步等定时任务功能.
"""
from __future__ import annotations
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
def sync_lpr_rates() -> dict:
"""同步LPR利率数据的定时任务.
每月20日（遇节假日顺延）从央行官网获取最新LPR数据。
Returns:
同步结果
"""
logger.info("[LPRSchedule] Starting scheduled LPR sync task")
try:
from apps.finance.services.lpr import LPRSyncService
service = LPRSyncService()
result = service.sync_latest()
logger.info(f"[LPRSchedule] LPR sync completed: {result}")
return result
except Exception as e:
logger.error(f"[LPRSchedule] LPR sync failed: {e}")
raise
def setup_lpr_sync_schedule() -> None:
"""设置LPR同步定时任务.
在系统启动时调用，创建每月20日的定时同步任务。
"""
from django_q.models import Schedule
# 检查是否已存在
existing = Schedule.objects.filter(name="lpr_monthly_sync").first()
if existing:
logger.info("[LPRSchedule] LPR sync schedule already exists")
return
# 创建定时任务：每月20日 9:30 执行
# 央行通常在每月20日9:15公布LPR
Schedule.objects.create(
name="lpr_monthly_sync",
func="apps.finance.tasks.sync_lpr_rates",
schedule_type=Schedule.MONTHLY,
repeats=-1, # 无限重复
next_run=datetime.now().replace(day=20, hour=9, minute=30, second=0, microsecond=0),
)
logger.info("[LPRSchedule] LPR monthly sync schedule created")
|