1.修改推理逻辑,useflag为1的也参与推理,之前是不用的点是1,正常的是0,现在发现正常的也有1,目前未知此参数具体作用。

This commit is contained in:
whm
2026-03-13 16:46:10 +08:00
parent ed0d840aee
commit dd9f31ee52
4 changed files with 74 additions and 85 deletions

View File

@@ -115,11 +115,8 @@ class DailyDataService(BaseService[DailyData]):
# 模型字段列表 # 模型字段列表
model_columns = [getattr(SettlementData, col.name) for col in SettlementData.__table__.columns] model_columns = [getattr(SettlementData, col.name) for col in SettlementData.__table__.columns]
# 基础条件 # 基础条件:不按 useflag 过滤,确保能取到每个 point 的真正最新一期(按 NYID 最大)
base_conditions = [ base_conditions = []
SettlementData.useflag.isnot(None),
SettlementData.useflag != 0
]
if point_ids: if point_ids:
base_conditions.append(SettlementData.point_id.in_(point_ids)) base_conditions.append(SettlementData.point_id.in_(point_ids))

View File

@@ -30,6 +30,21 @@ class SettlementDataService(BaseService[SettlementData]):
def get_by_nyid(self, db: Session, nyid: str) -> List[SettlementData]: def get_by_nyid(self, db: Session, nyid: str) -> List[SettlementData]:
"""根据期数ID获取沉降数据""" """根据期数ID获取沉降数据"""
return self.get_by_field(db, "NYID", nyid) return self.get_by_field(db, "NYID", nyid)
def get_one_dict_by_nyid(self, db: Session, nyid: str) -> Optional[Dict[str, Any]]:
"""根据期数ID取一条沉降记录并转为字典供推理/到期计算用datetime 转为字符串)"""
row = db.query(SettlementData).filter(SettlementData.NYID == nyid).first()
if not row:
return None
field_names = [c.name for c in SettlementData.__table__.columns]
item = {}
for k in field_names:
v = getattr(row, k)
if isinstance(v, datetime):
item[k] = v.strftime("%Y-%m-%d %H:%M:%S")
else:
item[k] = v
return item
def get_by_nyid_and_point_id(self, db: Session, nyid: str, point_id: str) -> List[SettlementData]: def get_by_nyid_and_point_id(self, db: Session, nyid: str, point_id: str) -> List[SettlementData]:
"""根据期数ID和观测点ID获取沉降数据""" """根据期数ID和观测点ID获取沉降数据"""
return self.get_by_field(db, "NYID", nyid, "point_id", point_id) return self.get_by_field(db, "NYID", nyid, "point_id", point_id)

View File

@@ -75,17 +75,13 @@ class ConstructionMonitorUtils:
if not point_data: if not point_data:
continue continue
# 过滤逻辑:仅保留 useflag 存在且值≠0 的记录 # 推理用最新一期:取按 NYID 排序后的第一条(上游已保证倒序),不因 useflag 排除最新期
latest_item = point_data[0]
# 用于冬休回溯等:仅 useflag 有效的历史记录
filtered_point_data = [ filtered_point_data = [
item for item in point_data item for item in point_data
if "useflag" in item and item["useflag"] != 0 # 核心条件:字段存在 + 非0 if "useflag" in item and item["useflag"] != 0
] ]
# 过滤后无数据则跳过当前测点
if not filtered_point_data:
continue
# 使用过滤后的数据处理
latest_item = filtered_point_data[0]
latest_condition = latest_item.get("workinfoname") latest_condition = latest_item.get("workinfoname")
if not latest_condition: if not latest_condition:
result["error_data"].append(latest_item) result["error_data"].append(latest_item)

View File

@@ -16,7 +16,9 @@ from ..services.section_data import SectionDataService
from ..services.account import AccountService from ..services.account import AccountService
from ..models.daily import DailyData from ..models.daily import DailyData
from ..models.settlement_data import SettlementData from ..models.settlement_data import SettlementData
from typing import List from ..models.level_data import LevelData
from ..services.settlement_data import SettlementDataService
from typing import List, Tuple, Any
from ..utils.construction_monitor import ConstructionMonitorUtils from ..utils.construction_monitor import ConstructionMonitorUtils
import time import time
import json import json
@@ -95,10 +97,8 @@ class TaskScheduler:
# name='每日重置账号更新状态' # name='每日重置账号更新状态'
# ) # )
# logger.info("系统定时任务:每日重置账号更新状态已添加") # logger.info("系统定时任务:每日重置账号更新状态已添加")
# existing_job = None
# existing_job = self.scheduler.get_job("scheduled_get_max_nyid_by_point_id") # existing_job = self.scheduler.get_job("scheduled_get_max_nyid_by_point_id")
# if not existing_job: # if existing_job is None:
# # 添加每天凌晨1点执行获取max NYID关联数据任务
# self.scheduler.add_job( # self.scheduler.add_job(
# scheduled_get_max_nyid_by_point_id, # scheduled_get_max_nyid_by_point_id,
# 'cron', # 'cron',
@@ -234,81 +234,63 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0,end: int = 0):
# logger.info(f"DailyData表清空完成共删除{delete_count}条历史记录") # logger.info(f"DailyData表清空完成共删除{delete_count}条历史记录")
# 1. 获取沉降数据(返回 List[List[dict]] # 1. 以 level_data 为来源:每个 linecode 取最新一期NYID 最大),再按该 NYID 从 settlement 取一条
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)
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]
# print(new_list)
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'])
# print(f"第{max_num}次获取冬休数据完成,共{len(winters)}条记录")
if winters == []:
break
data.extend(error_data)
# 4. 初始化服务实例
level_service = LevelDataService() level_service = LevelDataService()
settlement_service = SettlementDataService()
daily_service = DailyDataService()
checkpoint_db = CheckpointService() checkpoint_db = CheckpointService()
section_db = SectionDataService() section_db = SectionDataService()
account_service = AccountService() account_service = AccountService()
# 5. 关联其他表数据(核心逻辑保留) monitor = ConstructionMonitorUtils()
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返回单实例直接使用 linecodes = [r[0] for r in db.query(LevelData.linecode).distinct().all()]
linecode_level_settlement: List[Tuple[str, Any, dict]] = []
for linecode in linecodes:
level_instance = level_service.get_last_by_linecode(db, linecode)
if not level_instance:
continue
nyid = level_instance.NYID
settlement_dict = settlement_service.get_one_dict_by_nyid(db, nyid)
if not settlement_dict:
continue
settlement_dict['__linecode'] = linecode
settlement_dict['__level_data'] = level_instance.to_dict()
linecode_level_settlement.append((linecode, level_instance, settlement_dict))
input_data = [[s] for (_, _, s) in linecode_level_settlement]
if not input_data:
logger.warning("未找到任何 linecode 对应的最新期沉降数据,跳过写 daily")
db.execute(text(f"TRUNCATE TABLE {DailyData.__tablename__}"))
db.commit()
db.close()
return
# 2. 计算到期数据remaining / 冬休等)
daily_data = monitor.get_due_data(input_data, start=start, end=end)
data = daily_data['data']
logger.info(f"按 level_data 最新期获取数据完成,共{len(data)}条有效记录")
# 3. 关联 level / checkpoint / section / accountlevel 已带在 __level_data
for d in data:
d['level_data'] = d.pop('__level_data', None)
d.pop('__linecode', None)
checkpoint_instance = checkpoint_db.get_by_point_id(db, d['point_id']) checkpoint_instance = checkpoint_db.get_by_point_id(db, d['point_id'])
d['checkpoint_data'] = checkpoint_instance.to_dict() if checkpoint_instance else None d['checkpoint_data'] = checkpoint_instance.to_dict() if checkpoint_instance else None
# 处理 SectionData根据checkpoint_data关联
if d['checkpoint_data']: if d['checkpoint_data']:
section_instance = section_db.get_by_section_id(db, d['checkpoint_data']['section_id']) 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 d['section_data'] = section_instance.to_dict() if section_instance else None
else: else:
d['section_data'] = None d['section_data'] = None
# 处理 AccountData
if d.get('section_data') and d['section_data'].get('account_id'): 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']) 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 d['account_data'] = account_response.__dict__ if account_response else None
else: else:
d['account_data'] = None d['account_data'] = None
print(f"一共有{len(data)}条数据")
# 6. 构造DailyData数据并批量创建 # 4. 构造 DailyData(每条已是「每 linecode 最新一期」)
# daily_create_data1 = set()
daily_create_data = [] daily_create_data = []
nyids = []
for d in data: for d in data:
# 过滤无效数据(避免缺失关键字段报错) if not all(key in d for key in ['NYID', 'point_id', 'remaining']) or not d.get('level_data') or not d.get('account_data') or not d.get('section_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 continue
tem = { tem = {
'NYID': d['NYID'], 'NYID': d['NYID'],
@@ -318,7 +300,6 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0,end: int = 0):
'section_id': d['section_data']['section_id'], 'section_id': d['section_data']['section_id'],
'remaining': (0 - int(d['overdue'])) if 'overdue' in d else d['remaining'], 'remaining': (0 - int(d['overdue'])) if 'overdue' in d else d['remaining'],
} }
nyids.append(d['NYID'])
daily_create_data.append(tem) daily_create_data.append(tem)
# 批量创建记录 # 批量创建记录
print(daily_create_data) print(daily_create_data)