345 lines
14 KiB
Python
345 lines
14 KiB
Python
from apscheduler.schedulers.background import BackgroundScheduler
|
||
from apscheduler.executors.pool import ThreadPoolExecutor
|
||
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
|
||
from sqlalchemy import text
|
||
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
|
||
from ..core.config import settings
|
||
from sqlalchemy.orm import Session
|
||
from ..core.database import SessionLocal
|
||
from ..core.logging_config import get_logger
|
||
from ..models.account import Account
|
||
# from ..services.daily import get_nyid_by_point_id
|
||
from ..services.daily import DailyDataService
|
||
from ..services.level_data import LevelDataService
|
||
from ..services.checkpoint import CheckpointService
|
||
from ..services.section_data import SectionDataService
|
||
from ..services.account import AccountService
|
||
from ..models.daily import DailyData
|
||
from ..models.settlement_data import SettlementData
|
||
from typing import List
|
||
from ..utils.construction_monitor import ConstructionMonitorUtils
|
||
import time
|
||
import json
|
||
|
||
# 获取日志记录器
|
||
logger = get_logger(__name__)
|
||
|
||
class TaskScheduler:
|
||
def __init__(self):
|
||
# 配置作业存储
|
||
jobstores = {
|
||
'default': SQLAlchemyJobStore(url=settings.DATABASE_URL)
|
||
}
|
||
|
||
# 配置执行器
|
||
executors = {
|
||
'default': ThreadPoolExecutor(20)
|
||
}
|
||
|
||
# 作业默认配置
|
||
job_defaults = {
|
||
'coalesce': False,
|
||
'max_instances': 3
|
||
}
|
||
|
||
# 创建调度器
|
||
self.scheduler = BackgroundScheduler(
|
||
jobstores=jobstores,
|
||
executors=executors,
|
||
job_defaults=job_defaults,
|
||
timezone='Asia/Shanghai'
|
||
)
|
||
|
||
# 添加事件监听器
|
||
self.scheduler.add_listener(self._job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
|
||
|
||
def _job_listener(self, event):
|
||
"""作业执行监听器"""
|
||
if event.exception:
|
||
logger.error(f"Job {event.job_id} crashed: {event.exception}")
|
||
else:
|
||
logger.info(f"Job {event.job_id} executed successfully")
|
||
|
||
def start(self):
|
||
"""启动调度器"""
|
||
if not self.scheduler.running:
|
||
self.scheduler.start()
|
||
logger.info("定时任务调度器已启动")
|
||
|
||
# 启动时自动添加系统定时任务
|
||
self._setup_system_tasks()
|
||
|
||
def shutdown(self):
|
||
"""关闭调度器"""
|
||
if self.scheduler.running:
|
||
self.scheduler.shutdown()
|
||
logger.info("定时任务调度器已关闭")
|
||
|
||
def _setup_system_tasks(self):
|
||
"""设置系统定时任务"""
|
||
try:
|
||
logger.info("无定时任务")
|
||
# 检查是否已存在每日重置任务
|
||
|
||
# existing_job = self.scheduler.get_job("daily_reset_today_updated")
|
||
# # existing_job = self.scheduler.get_job("get_max_nyid")
|
||
# if not existing_job:
|
||
# # 添加每天午夜12点重置today_updated字段的任务
|
||
# self.scheduler.add_job(
|
||
# reset_today_updated_task,
|
||
# 'cron',
|
||
# id='daily_reset_today_updated',
|
||
# hour=0,
|
||
# minute=0,
|
||
# second=0,
|
||
# name='每日重置账号更新状态'
|
||
# )
|
||
# logger.info("系统定时任务：每日重置账号更新状态已添加")
|
||
# existing_job = None
|
||
# existing_job = self.scheduler.get_job("scheduled_get_max_nyid_by_point_id")
|
||
# if not existing_job:
|
||
# # 添加每天凌晨1点执行获取max NYID关联数据任务
|
||
# self.scheduler.add_job(
|
||
# scheduled_get_max_nyid_by_point_id,
|
||
# 'cron',
|
||
# id='scheduled_get_max_nyid_by_point_id',
|
||
# hour=1,
|
||
# minute=0,
|
||
# second=0,
|
||
# name='每日获取max NYID关联数据并创建DailyData记录'
|
||
# )
|
||
# logger.info("系统定时任务：每日获取max NYID关联数据任务已添加")
|
||
except Exception as e:
|
||
logger.error(f"设置系统定时任务失败: {e}")
|
||
|
||
# 设置测试任务
|
||
# try:
|
||
# existing_job = self.scheduler.get_job("test_task")
|
||
# if not existing_job:
|
||
# # 添加每天每小时重置today_updated字段的任务
|
||
# self.scheduler.add_job(
|
||
# reset_today_updated_task,
|
||
# 'cron',
|
||
# id='test_task',
|
||
# hour='*',
|
||
# minute=0,
|
||
# second=0,
|
||
# name='测试任务'
|
||
# )
|
||
# logger.info("系统定时任务：测试任务已添加")
|
||
# except Exception as e:
|
||
# logger.error(f"设置测试任务失败: {e}")
|
||
|
||
def add_cron_job(self, func, job_id: str, **kwargs):
|
||
"""添加cron定时任务"""
|
||
return self.scheduler.add_job(func, 'cron', id=job_id, **kwargs)
|
||
|
||
def add_interval_job(self, func, job_id: str, **kwargs):
|
||
"""添加间隔执行任务"""
|
||
return self.scheduler.add_job(func, 'interval', id=job_id, **kwargs)
|
||
|
||
def add_date_job(self, func, job_id: str, **kwargs):
|
||
"""添加指定时间执行任务"""
|
||
return self.scheduler.add_job(func, 'date', id=job_id, **kwargs)
|
||
|
||
def remove_job(self, job_id: str):
|
||
"""移除任务"""
|
||
try:
|
||
self.scheduler.remove_job(job_id)
|
||
logger.info(f"任务 {job_id} 已移除")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"移除任务失败: {e}")
|
||
return False
|
||
|
||
def get_job(self, job_id: str):
|
||
"""获取任务"""
|
||
return self.scheduler.get_job(job_id)
|
||
|
||
def get_jobs(self):
|
||
"""获取所有任务"""
|
||
return self.scheduler.get_jobs()
|
||
|
||
def pause_job(self, job_id: str):
|
||
"""暂停任务"""
|
||
try:
|
||
self.scheduler.pause_job(job_id)
|
||
logger.info(f"任务 {job_id} 已暂停")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"暂停任务失败: {e}")
|
||
return False
|
||
|
||
def resume_job(self, job_id: str):
|
||
"""恢复任务"""
|
||
try:
|
||
self.scheduler.resume_job(job_id)
|
||
logger.info(f"任务 {job_id} 已恢复")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"恢复任务失败: {e}")
|
||
return False
|
||
|
||
# 全局调度器实例
|
||
task_scheduler = TaskScheduler()
|
||
|
||
# 系统定时任务函数
|
||
def reset_today_updated_task():
|
||
"""每日重置账号today_updated字段为0的任务"""
|
||
db = SessionLocal()
|
||
try:
|
||
logger.info("开始执行每日重置账号更新状态任务")
|
||
|
||
# 更新所有账号的today_updated字段为0
|
||
updated_need_count = db.query(Account).filter(Account.today_updated == 1).count()
|
||
updated_count = db.query(Account).update({Account.today_updated: 0})
|
||
db.commit()
|
||
|
||
logger.info(f"每日重置任务完成，已重置 {updated_count} 个账号的today_updated字段")
|
||
return f"成功重置 {updated_count} 个账号的更新状态"
|
||
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"每日重置任务执行失败: {e}")
|
||
raise e
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# 示例定时任务函数
|
||
def example_task():
|
||
"""示例定时任务"""
|
||
logger.info("执行示例定时任务")
|
||
# 这里可以添加具体的业务逻辑
|
||
return "任务执行完成"
|
||
|
||
def database_cleanup_task():
|
||
"""数据库清理任务示例"""
|
||
logger.info("执行数据库清理任务")
|
||
# 这里可以添加数据库清理逻辑
|
||
return "数据库清理完成"
|
||
|
||
# 每日自动写入获取最新工况信息
|
||
def scheduled_get_max_nyid_by_point_id(start: int = 0,end: int = 0):
|
||
"""定时任务：获取max NYID关联数据并批量创建DailyData记录"""
|
||
db: Session = None
|
||
try:
|
||
logger.info("定时任务触发：开始获取max NYID关联数据并处理")
|
||
# 初始化数据库会话（替代接口的Depends依赖）
|
||
db = SessionLocal()
|
||
logger.info("定时任务开始执行：获取max NYID关联数据并处理")
|
||
# 核心新增：清空DailyData表所有数据
|
||
# delete_count = db.query(DailyData).delete()
|
||
# db.commit()
|
||
|
||
# logger.info(f"DailyData表清空完成，共删除{delete_count}条历史记录")
|
||
|
||
# 1. 获取沉降数据（返回 List[List[dict]]）
|
||
daily_service = DailyDataService()
|
||
result = daily_service.get_nyid_by_point_id(db, [], 1)
|
||
|
||
# 2. 计算到期数据
|
||
monitor = ConstructionMonitorUtils()
|
||
daily_data = monitor.get_due_data(result,start=start,end=end)
|
||
print(daily_data)
|
||
data = daily_data['data']
|
||
error_data = daily_data['error_data']
|
||
|
||
winters = daily_data['winter']
|
||
logger.info(f"首次获取数据完成，共{len(result)}条记录")
|
||
|
||
# 3. 循环处理冬休数据，追溯历史非冬休记录
|
||
max_num = 1
|
||
print(f"首次获取冬休数据完成，共{len(winters)}条记录")
|
||
while 1:
|
||
max_num += 1
|
||
print(max_num)
|
||
# 提取冬休数据的point_id列表
|
||
new_list = [int(w['point_id']) for w in winters]
|
||
if new_list == []:
|
||
break
|
||
# 获取更多历史记录
|
||
nyid_list = daily_service.get_nyid_by_point_id(db, new_list, max_num)
|
||
w_list = monitor.get_due_data(nyid_list,start=start,end=end)
|
||
# 更新冬休、待处理、错误数据
|
||
winters = w_list['winter']
|
||
data.extend(w_list['data'])
|
||
# 过期数据一并处理
|
||
# data.extend(w_list['error_data'])
|
||
error_data.extend(w_list['error_data'])
|
||
if winters == []:
|
||
break
|
||
data.extend(error_data)
|
||
# 4. 初始化服务实例
|
||
level_service = LevelDataService()
|
||
checkpoint_db = CheckpointService()
|
||
section_db = SectionDataService()
|
||
account_service = AccountService()
|
||
# 5. 关联其他表数据（核心逻辑保留）
|
||
for d in data:
|
||
# 处理 LevelData（假设返回列表，取第一条）
|
||
level_results = level_service.get_by_nyid(db, d['NYID'])
|
||
level_instance = level_results[0] if isinstance(level_results, list) and level_results else level_results
|
||
d['level_data'] = level_instance.to_dict() if level_instance else None
|
||
|
||
# 处理 CheckpointData（返回单实例，直接使用）
|
||
checkpoint_instance = checkpoint_db.get_by_point_id(db, d['point_id'])
|
||
d['checkpoint_data'] = checkpoint_instance.to_dict() if checkpoint_instance else None
|
||
|
||
# 处理 SectionData（根据checkpoint_data关联）
|
||
if d['checkpoint_data']:
|
||
section_instance = section_db.get_by_section_id(db, d['checkpoint_data']['section_id'])
|
||
d['section_data'] = section_instance.to_dict() if section_instance else None
|
||
else:
|
||
d['section_data'] = None
|
||
|
||
# 处理 AccountData
|
||
if d.get('section_data') and d['section_data'].get('account_id'):
|
||
account_response = account_service.get_account(db, account_id=d['section_data']['account_id'])
|
||
d['account_data'] = account_response.__dict__ if account_response else None
|
||
else:
|
||
d['account_data'] = None
|
||
print(f"一共有{len(data)}条数据")
|
||
# 6. 构造DailyData数据并批量创建
|
||
# daily_create_data1 = set()
|
||
daily_create_data = []
|
||
nyids = []
|
||
for d in data:
|
||
# 过滤无效数据（避免缺失关键字段报错）
|
||
if all(key in d for key in ['NYID', 'point_id','remaining']) and d.get('level_data') and d.get('account_data') and d.get('section_data'):
|
||
if d['NYID'] in nyids:
|
||
continue
|
||
tem = {
|
||
'NYID': d['NYID'],
|
||
'point_id': d['point_id'],
|
||
'linecode': d['level_data']['linecode'],
|
||
'account_id': d['account_data']['account_id'],
|
||
'section_id': d['section_data']['section_id'],
|
||
'remaining': (0-int(d['overdue'])) if 'overdue' in d else d['remaining'],
|
||
}
|
||
nyids.append(d['NYID'])
|
||
daily_create_data.append(tem)
|
||
# 批量创建记录
|
||
print(daily_create_data)
|
||
if daily_create_data:
|
||
db.execute(text(f"TRUNCATE TABLE {DailyData.__tablename__}"))
|
||
db.commit() # 必须提交事务
|
||
# with open('data.json', 'w', encoding='utf-8') as f:
|
||
# json.dump(daily_create_data, f, ensure_ascii=False, indent=4)
|
||
created_records = daily_service.batch_create_by_account_nyid(db, daily_create_data)
|
||
logger.info(f"定时任务完成：成功创建{len(created_records)}条DailyData记录，共处理{len(data)}个point_id数据")
|
||
else:
|
||
db.execute(text(f"TRUNCATE TABLE {DailyData.__tablename__}"))
|
||
db.commit() # 必须提交事务
|
||
logger.warning("定时任务完成：无有效数据可创建DailyData记录")
|
||
|
||
except Exception as e:
|
||
# 异常时回滚事务
|
||
if db:
|
||
db.rollback()
|
||
logger.error(f"定时任务执行失败：{str(e)}", exc_info=True)
|
||
raise e # 抛出异常便于定时框架捕获告警
|
||
finally:
|
||
# 确保数据库会话关闭
|
||
if db:
|
||
db.close() |