diff --git a/app/services/daily.py b/app/services/daily.py index 06d09b1..4e386f0 100644 --- a/app/services/daily.py +++ b/app/services/daily.py @@ -115,11 +115,8 @@ class DailyDataService(BaseService[DailyData]): # 模型字段列表 model_columns = [getattr(SettlementData, col.name) for col in SettlementData.__table__.columns] - # 基础条件 - base_conditions = [ - SettlementData.useflag.isnot(None), - SettlementData.useflag != 0 - ] + # 基础条件：不按 useflag 过滤，确保能取到每个 point 的真正最新一期（按 NYID 最大） + base_conditions = [] if point_ids: base_conditions.append(SettlementData.point_id.in_(point_ids)) diff --git a/app/services/settlement_data.py b/app/services/settlement_data.py index 65a9338..35dc30e 100644 --- a/app/services/settlement_data.py +++ b/app/services/settlement_data.py @@ -30,6 +30,21 @@ class SettlementDataService(BaseService[SettlementData]): def get_by_nyid(self, db: Session, nyid: str) -> List[SettlementData]: """根据期数ID获取沉降数据""" 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]: """根据期数ID和观测点ID获取沉降数据""" return self.get_by_field(db, "NYID", nyid, "point_id", point_id) diff --git a/app/utils/construction_monitor.py b/app/utils/construction_monitor.py index 16c066d..b5790b0 100644 --- a/app/utils/construction_monitor.py +++ b/app/utils/construction_monitor.py @@ -75,17 +75,13 @@ class ConstructionMonitorUtils: if not point_data: continue - # 过滤逻辑：仅保留 useflag 存在且值≠0 的记录 + # 推理用最新一期：取按 NYID 排序后的第一条（上游已保证倒序），不因 useflag 排除最新期 + latest_item = point_data[0] + # 用于冬休回溯等：仅 useflag 有效的历史记录 filtered_point_data = [ - item for item in point_data - if "useflag" in item and item["useflag"] != 0 # 核心条件：字段存在 + 非0 + item for item in point_data + 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") if not latest_condition: result["error_data"].append(latest_item) diff --git a/app/utils/scheduler.py b/app/utils/scheduler.py index dc550fa..820067d 100644 --- a/app/utils/scheduler.py +++ b/app/utils/scheduler.py @@ -16,7 +16,9 @@ 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 ..models.level_data import LevelData +from ..services.settlement_data import SettlementDataService +from typing import List, Tuple, Any from ..utils.construction_monitor import ConstructionMonitorUtils import time import json @@ -95,10 +97,8 @@ class TaskScheduler: # 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关联数据任务 + # if existing_job is None: # self.scheduler.add_job( # scheduled_get_max_nyid_by_point_id, # 'cron', @@ -234,92 +234,73 @@ def scheduled_get_max_nyid_by_point_id(start: int = 0,end: int = 0): # 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) - 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. 初始化服务实例 + # 1. 以 level_data 为来源：每个 linecode 取最新一期（NYID 最大），再按该 NYID 从 settlement 取一条 level_service = LevelDataService() + settlement_service = SettlementDataService() + daily_service = DailyDataService() checkpoint_db = CheckpointService() section_db = SectionDataService() account_service = AccountService() - # 5. 关联其他表数据（核心逻辑保留） + monitor = ConstructionMonitorUtils() + + 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 / account（level 已带在 __level_data） 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（返回单实例，直接使用） + d['level_data'] = d.pop('__level_data', None) + d.pop('__linecode', None) 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() + + # 4. 构造 DailyData（每条已是「每 linecode 最新一期」） 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) + 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'): + 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'], + } + daily_create_data.append(tem) # 批量创建记录 print(daily_create_data) if daily_create_data: