admin/railway_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.优化daily的数据结构

Browse Source 
2.修改定时任务
This commit is contained in:
whm
2025-11-18 12:09:26 +08:00
parent d8b6247094
commit 83925382ed
3 changed files with 76 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
app/services/daily.py
View File

@@ -90,15 +90,17 @@ class DailyDataService(BaseService[DailyData]):
def get_nyid_by_point_id(
self,
db: Session,
point_ids: List[int] = None,
point_ids: Optional[List[int]] = None,
max_num: int = 1
) -> List[List[dict]]:
"""
获取指定point_id的记录，修复子查询中模型对象访问错误，同时过滤useflag=0的数据
获取指定point_id的记录，每个point_id的前max_num条记录放在同一个子列表中
返回格式：[[point1_records...], [point2_records...]]
"""
# 处理参数默认值
point_ids = point_ids or []
max_num = max(max_num, 1)
if max_num <= 0:
return []
# 窗口函数：按point_id分组，每组内按NYID降序编号
row_num = over(
@@ -107,60 +109,54 @@ class DailyDataService(BaseService[DailyData]):
order_by=desc(SettlementData.NYID)
).label("row_num")
# 子查询：查询模型的所有字段 + 行号（不保留模型对象，只展平字段）
# 先获取模型的所有字段列表
# 模型字段列表
model_columns = [getattr(SettlementData, col.name) for col in SettlementData.__table__.columns]
# -------------------------- 新增过滤条件：useflag IS NOT NULL AND useflag != 0 --------------------------
# 基础条件：过滤useflag为0或不存在的记录
# 基础条件
base_conditions = [
SettlementData.useflag.isnot(None), # 确保useflag字段存在（非NULL）
SettlementData.useflag != 0 # 确保useflag值不为0
SettlementData.useflag.isnot(None),
SettlementData.useflag != 0
]
# 若指定了point_ids，添加point_id过滤条件
if point_ids:
base_conditions.append(SettlementData.point_id.in_(point_ids))
# 子查询
subquery = (
select(*model_columns, row_num) # 展开所有字段 + 行号
.where(*base_conditions) # 应用组合条件（包含useflag过滤）
select(*model_columns, row_num)
.where(*base_conditions)
.subquery()
)
# ------------------------------------------------------------------------------------------------------
# 主查询：筛选行号<=max_num的记录
# 主查询：筛选每个point_id的前max_num条
query = (
select(subquery)
.where(subquery.c.row_num <= max_num)
.order_by(subquery.c.point_id, subquery.c.row_num)
)
# 执行查询（结果为包含字段值的行对象）
# 执行查询
results = db.execute(query).all()
grouped: Dict[int, List[dict]] = {}
# 获取模型字段名列表（用于映射行对象到字典）
grouped: Dict[int, List[dict]] = {} # 键：point_id，值：该point_id的所有记录（子列表）
field_names = [col.name for col in SettlementData.__table__.columns]
for row in results:
# 将行对象转换为字典（忽略最后一个字段row_num）
item_dict = {
field: getattr(row, field)
for field in field_names
}
pid = item_dict["point_id"]
item_dict = {field: getattr(row, field) for field in field_names}
try:
pid = int(item_dict["point_id"]) # 确保point_id为整数
except (KeyError, ValueError):
continue # 跳过无效记录
# 同一point_id的记录放入同一个列表
if pid not in grouped:
grouped[pid] = []
grouped[pid].append(item_dict)
grouped[pid] = [] # 初始化子列表
grouped[pid].append(item_dict) # 追加到子列表
# 按输入point_ids顺序整理结果
# 按输入point_ids顺序整理结果（关键：每个point_id对应一个子列表）
if not point_ids:
point_ids = sorted(grouped.keys())
point_ids = sorted(grouped.keys()) # 若无指定，按point_id排序
# 构建[[{}], [{}]]格式
return [
[record] for pid in point_ids for record in grouped.get(pid, [])
]
# 构建最终结果：每个point_id的记录作为一个子列表
return [grouped.get(pid, []) for pid in point_ids]
# 获取所有的今日数据
def get_all_daily_data(

4
app/utils/construction_monitor.py
View File

@@ -2,7 +2,8 @@ from datetime import datetime
from typing import List, Dict
import warnings
import copy
from ..core.logging_config import get_logger
logger = get_logger(__name__)
class ConstructionMonitorUtils:
def __init__(self):
# 原始工况周期映射表（保持不变）
@@ -155,6 +156,7 @@ class ConstructionMonitorUtils:
f"【超期警报】测点{point_idx} 最新工况'{latest_condition}'（{create_date}）"
f"已超期{abs(due_days)}天！基准工况：{base_condition}，周期{period}天"
)
logger.warning(warn_msg)
warnings.warn(warn_msg, UserWarning)
elif start <= due_days <= end:
item_copy["remaining"] = due_days

98
app/utils/scheduler.py
View File

@@ -1,6 +1,7 @@
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
@@ -17,6 +18,8 @@ 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__)
@@ -216,7 +219,7 @@ def database_cleanup_task():
return "数据库清理完成"
# 每日自动写入获取最新工况信息
def scheduled_get_max_nyid_by_point_id(start: int = 0, end: int = 0):
def scheduled_get_max_nyid_by_point_id(start: int = 0,end: int = 0):
"""定时任务：获取max NYID关联数据并批量创建DailyData记录"""
db: Session = None
try:
@@ -225,9 +228,11 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0, end: int = 0):
db = SessionLocal()
logger.info("定时任务开始执行：获取max NYID关联数据并处理")
# 核心新增：清空DailyData表所有数据
delete_count = db.query(DailyData).delete()
db.commit()
logger.info(f"DailyData表清空完成，共删除{delete_count}条历史记录")
# delete_count = db.query(DailyData).delete()
# db.commit()
db.execute(text(f"TRUNCATE TABLE {DailyData.__tablename__}"))
db.commit() # 必须提交事务
# logger.info(f"DailyData表清空完成，共删除{delete_count}条历史记录")
# 1. 获取沉降数据（返回 List[List[dict]]）
daily_service = DailyDataService()
@@ -235,7 +240,7 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0, end: int = 0):
# 2. 计算到期数据
monitor = ConstructionMonitorUtils()
daily_data = monitor.get_due_data(result, start, end)
daily_data = monitor.get_due_data(result,start=start,end=end)
data = daily_data['data']
error_data = daily_data['error_data']
@@ -250,86 +255,55 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0, end: int = 0):
new_list = [w['point_id'] for w in winters]
# 获取更多历史记录
nyid_list = daily_service.get_nyid_by_point_id(db, new_list, max_num)
w_list = monitor.get_due_data(nyid_list, start, end)
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'])
print(w_list)
data.extend(error_data)
# 4. 初始化服务实例
level_service = LevelDataService()
checkpoint_db = CheckpointService()
section_db = SectionDataService()
account_service = AccountService()
# 5. 批量查询优化
logger.info("批量获取关联数据")
# 提取所有需要的ID列表
nyid_list = list(set(d['NYID'] for d in data if d.get('NYID')))
point_id_list = list(set(d['point_id'] for d in data if d.get('point_id')))
# 批量查询LevelData
logger.info(f"批量查询LevelData，nyid数量: {len(nyid_list)}")
level_results = level_service.get_by_nyids(db, nyid_list)
level_dict = {level.NYID: level for level in level_results}
# 批量查询CheckpointData
logger.info(f"批量查询CheckpointData，point_id数量: {len(point_id_list)}")
checkpoint_results = checkpoint_db.get_by_point_ids_batch(db, point_id_list)
checkpoint_dict = {cp.point_id: cp for cp in checkpoint_results}
# 提取所有section_id
section_id_list = list(set(
cp.section_id for cp in checkpoint_results
if cp.section_id and isinstance(cp, object)
))
# 批量查询SectionData
logger.info(f"批量查询SectionData，section_id数量: {len(section_id_list)}")
section_results = section_db.get_by_section_ids_batch(db, section_id_list)
section_dict = {s.section_id: s for s in section_results}
# 提取所有account_id
account_id_list = list(set(
s.account_id for s in section_results
if s.account_id and isinstance(s, object)
))
# 批量查询AccountData
logger.info(f"批量查询AccountData，account_id数量: {len(account_id_list)}")
account_results = account_service.get_accounts_batch(db, account_id_list)
account_dict = {acc.id: acc for acc in account_results}
logger.info("批量查询完成，开始关联数据")
# 6. 关联数据到原记录
print(len(data))
# 5. 关联其他表数据（核心逻辑保留）
for d in data:
# 关联LevelData
level_instance = level_dict.get(d['NYID'])
# 处理 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_dict.get(d['point_id'])
# 处理 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
section_id = d['checkpoint_data']['section_id'] if d.get('checkpoint_data') else None
section_instance = section_dict.get(section_id) if section_id 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
account_id = d.get('section_data', {}).get('account_id') if d.get('section_data') else None
account_instance = account_dict.get(account_id) if account_id else None
d['account_data'] = account_instance.__dict__ if account_instance else 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
# 7. 构造DailyData数据并批量创建
# 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']) 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'],
@@ -338,8 +312,8 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0, end: int = 0):
'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)
# 批量创建记录
if daily_create_data:
created_records = daily_service.batch_create_by_account_nyid(db, daily_create_data)