1.自动更新今日需要抓取数据

2.工况接口返回

app/utils/construction_monitor.py

@@ -0,0 +1,159 @@
+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():
+            compatible_map[original_key] = period
+            # 生成中文括号版key并添加到映射表
+            chinese_bracket_key = original_key.replace("(", "（").replace(")", "）")
+            if chinese_bracket_key != original_key:
+                compatible_map[chinese_bracket_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"]:
+            # 用字典临时存储：key=NYID，value=该NYID下剩余天数最少的记录
+            nyid_min_remaining = {}
+            for record in result["data"]:
+                nyid = record.get("NYID")  # 假设NYID字段名是"NYID"，若实际不同需调整
+                if not nyid:
+                    continue  # 无NYID的记录直接保留（或按需求处理）
+                
+                # 若该NYID未存储，或当前记录剩余天数更少，则更新
+                if nyid not in nyid_min_remaining or record["remaining"] < nyid_min_remaining[nyid]["remaining"]:
+                    nyid_min_remaining[nyid] = record
+            
+            # 将字典 values 转换为列表，作为去重后的data
+            result["data"] = list(nyid_min_remaining.values())
+
+        return result