165 lines
7.9 KiB
Python
165 lines
7.9 KiB
Python
from datetime import datetime
|
||
from typing import List, Dict
|
||
import warnings
|
||
import copy
|
||
|
||
class ConstructionMonitorUtils:
|
||
def __init__(self):
|
||
# 原始工况周期映射表（保持不变）
|
||
self.base_periods = {
|
||
"仰拱(底板)施工完成后，第1个月": 7,
|
||
"仰拱(底板)施工完成后，第2至3个月": 14,
|
||
"仰拱(底板)施工完成后，3个月以后": 30,
|
||
"无砟轨道铺设后，第1至3个月": 30,
|
||
"无砟轨道铺设后，4至12个月": 90,
|
||
"无砟轨道铺设后，12个月以后": 180,
|
||
"墩台施工到一定高度": 30,
|
||
"墩台混凝土施工": 30,
|
||
"预制梁桥，架梁前": 30,
|
||
"预制梁桥，预制梁架设前": 1,
|
||
"预制梁桥，预制梁架设后": 7,
|
||
"桥位施工桥梁，制梁前": 30,
|
||
"桥位施工桥梁，上部结构施工中": 1,
|
||
"架桥机(运梁车)通过": 7,
|
||
"桥梁主体工程完工后,第1至3个月": 7, # 模拟包含英文逗号的原始数据
|
||
"桥梁主体工程完工后，第4至6个月": 14,
|
||
"桥梁主体工程完工后，6个月以后": 30,
|
||
"轨道铺设期间，前": 30,
|
||
"轨道铺设期间，后": 14,
|
||
"轨道铺设完成后，第1个月": 14,
|
||
"轨道铺设完成后，2至3个月": 30,
|
||
"轨道铺设完成后，4至12个月": 90,
|
||
"轨道铺设完成后，12个月以后": 180,
|
||
"填筑或堆载，一般情况": 1,
|
||
"填筑或堆载，沉降量突变情况": 1,
|
||
"填筑或堆载，两次填筑间隔时间较长情况": 3,
|
||
"堆载预压或路基填筑完成，第1至3个月": 7,
|
||
"堆载预压或路基填筑完成，第4至6个月": 14,
|
||
"堆载预压或路基填筑完成，6个月以后": 30,
|
||
"架桥机(运梁车)首次通过前": 1,
|
||
"架桥机(运梁车)首次通过后，前3天": 1,
|
||
"架桥机(运梁车)首次通过后": 7,
|
||
"轨道板(道床)铺设后，第1个月": 14,
|
||
"轨道板(道床)铺设后，第2至3个月": 30,
|
||
"轨道板(道床)铺设后，3个月以后": 90
|
||
}
|
||
# 构建中英文括号+逗号兼容映射表
|
||
self.compatible_periods = self._build_compatible_brackets_map()
|
||
|
||
def _build_compatible_brackets_map(self) -> Dict[str, int]:
|
||
"""构建支持中英文括号、中英文逗号的兼容映射表"""
|
||
compatible_map = {}
|
||
for original_key, period in self.base_periods.items():
|
||
# 1. 保留原始key
|
||
compatible_map[original_key] = period
|
||
|
||
# 2. 生成中文括号版key（原逻辑）
|
||
chinese_bracket_key = original_key.replace("(", "（").replace(")", "）")
|
||
if chinese_bracket_key != original_key:
|
||
compatible_map[chinese_bracket_key] = period
|
||
|
||
# 3. 生成英文逗号转中文逗号版key（新增逻辑）
|
||
chinese_comma_key = original_key.replace(",", "，")
|
||
if chinese_comma_key != original_key:
|
||
compatible_map[chinese_comma_key] = period
|
||
|
||
# 4. 生成中文括号+中文逗号混合版key（双重兼容）
|
||
mixed_key = chinese_bracket_key.replace(",", "，")
|
||
if mixed_key != original_key and mixed_key not in compatible_map:
|
||
compatible_map[mixed_key] = period
|
||
|
||
return compatible_map
|
||
|
||
def get_due_data(self, input_data: List[List[Dict]], start: int = 0, end: int = 1, current_date: datetime = None) -> Dict[str, List[Dict]]:
|
||
result = {"winter": [], "data": [], "error_data": []}
|
||
if not input_data:
|
||
return result
|
||
|
||
calc_date = current_date.date() if current_date else datetime.now().date()
|
||
|
||
# 原有核心逻辑（完全不变）
|
||
for point_idx, point_data in enumerate(input_data):
|
||
if not point_data:
|
||
continue
|
||
|
||
latest_item = point_data[0]
|
||
latest_condition = latest_item.get("workinfoname")
|
||
if not latest_condition:
|
||
result["error_data"].append(latest_item)
|
||
warnings.warn(f"【数据错误】测点{point_idx}的最新数据缺少'workinfoname'字段", UserWarning)
|
||
continue
|
||
|
||
base_condition = None
|
||
if latest_condition != "冬休":
|
||
if latest_condition not in self.compatible_periods:
|
||
result["error_data"].append(latest_item)
|
||
warnings.warn(f"【数据错误】测点{point_idx}最新数据存在未定义工况: {latest_condition}", UserWarning)
|
||
continue
|
||
base_condition = latest_condition
|
||
else:
|
||
for history_item in point_data[1:]:
|
||
history_condition = history_item.get("workinfoname")
|
||
if not history_condition:
|
||
result["error_data"].append(history_item)
|
||
warnings.warn(f"【数据错误】测点{point_idx}的历史数据缺少'workinfoname'字段", UserWarning)
|
||
continue
|
||
if history_condition != "冬休":
|
||
if history_condition not in self.compatible_periods:
|
||
result["error_data"].append(history_item)
|
||
warnings.warn(f"【数据错误】测点{point_idx}历史数据存在未定义工况: {history_condition}", UserWarning)
|
||
base_condition = None
|
||
break
|
||
base_condition = history_condition
|
||
break
|
||
|
||
item_copy = copy.deepcopy(latest_item)
|
||
create_date_val = latest_item.get("createdate")
|
||
if not create_date_val:
|
||
result["error_data"].append(item_copy)
|
||
warnings.warn(f"【数据错误】测点{point_idx}的最新数据缺少'createdate'字段", UserWarning)
|
||
continue
|
||
|
||
try:
|
||
if isinstance(create_date_val, datetime):
|
||
create_date = create_date_val.date()
|
||
else:
|
||
create_date = datetime.strptime(create_date_val, "%Y-%m-%d %H:%M:%S").date()
|
||
except ValueError as e:
|
||
result["error_data"].append(item_copy)
|
||
warnings.warn(f"【数据错误】测点{point_idx}最新数据的日期格式错误：{create_date_val}，错误：{str(e)}", UserWarning)
|
||
continue
|
||
|
||
if not base_condition:
|
||
result["winter"].append(item_copy)
|
||
continue
|
||
|
||
period = self.compatible_periods[base_condition]
|
||
days_passed = (calc_date - create_date).days
|
||
due_days = period - days_passed
|
||
|
||
if due_days < 0:
|
||
item_copy["overdue"] = abs(due_days)
|
||
result["error_data"].append(item_copy)
|
||
warn_msg = (
|
||
f"【超期警报】测点{point_idx} 最新工况'{latest_condition}'（{create_date}）"
|
||
f"已超期{abs(due_days)}天！基准工况：{base_condition}，周期{period}天"
|
||
)
|
||
warnings.warn(warn_msg, UserWarning)
|
||
elif start <= due_days <= end:
|
||
item_copy["remaining"] = due_days
|
||
result["data"].append(item_copy)
|
||
|
||
# 新增步骤：data中相同NYID保留剩余天数最少的记录
|
||
if result["data"]:
|
||
nyid_min_remaining = {}
|
||
for record in result["data"]:
|
||
nyid = record.get("NYID")
|
||
if not nyid:
|
||
continue
|
||
|
||
if nyid not in nyid_min_remaining or record["remaining"] < nyid_min_remaining[nyid]["remaining"]:
|
||
nyid_min_remaining[nyid] = record
|
||
|
||
result["data"] = list(nyid_min_remaining.values())
|
||
|
||
return result |