From d66e2826e1e83ea57ebdc29f344633d3ddfc7c92 Mon Sep 17 00:00:00 2001
From: DXC <duxingchen@iris-rs.cn>
Date: Tue, 15 Sep 2026 17:05:38 +0800
Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=8A=BD=E5=87=BA=E7=BB=9F?=
 =?UTF-8?q?=E4=B8=80=E5=85=A5=E5=BA=93=E7=AE=A1=E9=81=93=EF=BC=8C=E5=90=88?=
 =?UTF-8?q?=E5=B9=B6=E5=AE=9A=E6=97=B6=E4=B8=8E=E6=89=8B=E5=8A=A8=E4=B8=A4?=
 =?UTF-8?q?=E6=9D=A1=E9=87=8D=E5=A4=8D=E8=B7=AF=E5=BE=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

问题：app.py:auto_monitor_job（定时）和 routes/api.py:run_monitor（手动）
各自维护了一份几乎相同但细节不一致的写入逻辑，同一条数据经不同入口落库后
latest_time / source / offset / file_count 可能不同。

- 新增 services/time_utils.py：calculate_offset 下沉到无依赖模块，避免
  db_ingest 与 routes.api 互相 import 形成循环。routes/api.py 里 re-export
  一次，保证 app.py 原有的 from routes.api import calculate_offset 不失效。
- 新增 services/db_ingest.py:ingest_device_data()，承载全部入库细节，只
  add/flush 不 commit，事务边界交给调用方。
- routes/api.py:run_monitor 瘦身为「触发爬虫 -> 调管道 -> 提交」。
- models.py:to_dict() 的 offset 改为读取时用 calculate_offset(latest_time)
  实时计算，不再读 offset 列。该列是写入时刻的快照，采集一停摆就整体失真
  （库里停在 2026-02-06，offset 却仍显示“当天”）。列保留但已废弃，
  不执行 ALTER TABLE DROP COLUMN。

入库语义（db_ingest）：
- 只有爬虫拿到真实业务时间才更新主表 latest_time，拿不到就保留上一个有效值
  （冻结），不再回退到 current_time。
- DeviceHistory 单独用 history_time：主表回答“数据到什么时候”，历史回答
  “什么时候采过”。
- 历史表 json_data 改存本次增量切片，不再复制主表那份越滚越大的累积 JSON。
- 主表 json_data 改为覆盖式更新以切断无限膨胀，但保留 APP_OWNED_KEYS
  （bound_iccid / is_whitelist）—— 这两个键由 /bind_device_card 和
  /toggle_whitelist 写入，按原方案直接覆盖会清空所有设备-流量卡绑定。
---
 2_1banben/models.py              |   7 +-
 2_1banben/routes/api.py          |  98 ++++---------------------
 2_1banben/services/db_ingest.py  | 119 +++++++++++++++++++++++++++++++
 2_1banben/services/time_utils.py |  25 +++++++
 4 files changed, 165 insertions(+), 84 deletions(-)
 create mode 100644 2_1banben/services/db_ingest.py
 create mode 100644 2_1banben/services/time_utils.py

diff --git a/2_1banben/models.py b/2_1banben/models.py
index 8b25284..0be7ac7 100644
--- a/2_1banben/models.py
+++ b/2_1banben/models.py
@@ -1,6 +1,8 @@
 # models.py
 from datetime import datetime
 from extensions import db
+# services.time_utils 只依赖 datetime，不会与 extensions/models 形成循环导入
+from services.time_utils import calculate_offset
 
 
 class Device(db.Model):
@@ -46,7 +48,10 @@ class Device(db.Model):
             'is_maintaining': self.is_maintaining,
             'is_hidden': self.is_hidden,
             'is_whitelist': self.is_whitelist,
-            'offset': self.offset,
+            # 滞后天数在读取时实时计算，不读 self.offset。
+            # offset 列是写入时算好的"化石"，采集一旦停摆就会整体腐烂
+            # （例如库停在 2026-02-06，offset 却仍显示"当天"）。
+            'offset': calculate_offset(self.latest_time),
             'file_count': self.file_count  # ✅ 返回给前端
         }
 
diff --git a/2_1banben/routes/api.py b/2_1banben/routes/api.py
index 4e6d256..1ee67ab 100644
--- a/2_1banben/routes/api.py
+++ b/2_1banben/routes/api.py
@@ -15,6 +15,11 @@ try:
 except ImportError:
     execute_monitor_task = None
 
+try:
+    from services.db_ingest import ingest_device_data
+except ImportError:
+    ingest_device_data = None
+
 try:
     from services.iot_api import sync_iot_data_service
 except ImportError:
@@ -27,21 +32,11 @@ api_bp = Blueprint('api', __name__, url_prefix='/api')
 # 0. 核心算法区：数据质量分析与辅助函数
 # =========================================================
 
-def calculate_offset(latest_time_str):
-    """
-    计算时间滞后天数
-    用于前端展示设备数据是否过时
-    """
-    if not latest_time_str or latest_time_str == "N/A":
-        return "从未同步"
-    try:
-        # 兼容处理 2026_01_13 和 2026-01-13 格式
-        clean = str(latest_time_str).split()[0].replace('_', '-')
-        target = datetime.strptime(clean, "%Y-%m-%d").date()
-        diff = (datetime.now().date() - target).days
-        return "当天" if diff == 0 else f"滞后 {diff} 天"
-    except:
-        return "时间解析失败"
+# calculate_offset 已下沉到 services/time_utils.py。
+# 原因：services.db_ingest 需要计算 offset，而本模块又要 import db_ingest，
+# 留在本模块会形成循环导入。这里 re-export 一次，保证
+# `from routes.api import calculate_offset`（app.py 在用）继续可用。
+from services.time_utils import calculate_offset  # noqa: E402
 
 
 def check_data_quality(content_data, source_type, data_time_str=None):
@@ -383,77 +378,14 @@ def run_monitor():
 
     try:
         # --- A. 执行爬虫并入库 ---
-        if execute_monitor_task:
+        # 入库细节统一走 services.db_ingest，与定时任务 auto_monitor_job 保持一致，
+        # 避免同一条数据经手动/自动两个入口落库后字段不一致。
+        if execute_monitor_task and ingest_device_data:
             task_result = execute_monitor_task()
             if task_result:
                 scraped_list = task_result.get('device_list', [])
-                current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-
-                count_crawler = 0
-                for item in scraped_list:
-                    d_name = item.get('name')
-                    if not d_name: continue
-
-                    d_raw = item.get('raw_json', {})
-                    source = item.get('source', '')
-                    target_time = item.get('target_time')
-
-                    if '106' in str(source):
-                        try:
-                            path_str = d_raw.get('path', '')
-                            match = re.search(r'/Data/(\d{4}_\d{2}_\d{2})/\w+_(\d{2}_\d{2}_\d{2})\.csv', path_str)
-                            if match:
-                                date_part = match.group(1).replace('_', '-')
-                                time_part = match.group(2).replace('_', ':')
-                                target_time = f"{date_part} {time_part}"
-                        except:
-                            pass
-
-                    device = Device.query.filter_by(name=d_name).first()
-                    if not device:
-                        device = Device(name=d_name, source=source, install_site="")
-                        db.session.add(device)
-                        db.session.flush()
-
-                    if device.source == 'iot_card':
-                        device.source = source
-
-                    device.status = item.get('status')
-                    device.current_value = item.get('value')
-                    device.latest_time = target_time
-                    device.check_time = current_time
-
-                    # ✅ [核心修改] 获取爬虫返回的文件数量并保存
-                    f_count = item.get('num_files', 0)
-                    device.file_count = f_count
-
-                    old_json = {}
-                    try:
-                        if device.json_data:
-                            old_json = json.loads(device.json_data)
-                    except:
-                        old_json = {}
-
-                    new_json = d_raw if isinstance(d_raw, dict) else item.get('raw_json', {})
-                    if isinstance(new_json, dict):
-                        old_json.update(new_json)
-
-                    device.json_data = json.dumps(old_json, ensure_ascii=False)
-                    device.offset = calculate_offset(device.latest_time)
-
-                    # ✅ [核心修改] 写入历史记录时包含 file_count
-                    new_history = DeviceHistory(
-                        device_id=device.id,
-                        status=item.get('status'),
-                        result_data=item.get('value'),
-                        data_time=target_time,
-                        json_data=device.json_data,
-                        file_count=f_count  # 确保历史数据也记录文件数
-                    )
-                    db.session.add(new_history)
-                    count_crawler += 1
-
-                msg_list.append(f"爬虫更新: {count_crawler}")
+                updated, _ = ingest_device_data(scraped_list)
+                msg_list.append(f"爬虫更新: {updated}")
             else:
                 msg_list.append("爬虫无数据")
 
diff --git a/2_1banben/services/db_ingest.py b/2_1banben/services/db_ingest.py
new file mode 100644
index 0000000..6558f52
--- /dev/null
+++ b/2_1banben/services/db_ingest.py
@@ -0,0 +1,119 @@
+# services/db_ingest.py
+"""
+统一的设备数据入库管道。
+
+此前 app.py:auto_monitor_job（定时任务）和 routes/api.py:run_monitor（手动触发）
+各自维护了一份几乎相同但细节不一致的写入逻辑，导致同一条数据经不同入口落库后
+latest_time / offset / source / file_count 可能不同。这里合并为一份。
+
+本函数只负责 add/flush，不 commit —— 事务边界由调用方掌握。
+"""
+import json
+from datetime import datetime
+
+from extensions import db
+from models import Device, DeviceHistory
+
+# 由本应用接口写入 Device.json_data、而非爬虫返回的键，覆盖时必须保留：
+#   bound_iccid —— routes/api.py:/bind_device_card 写入的设备-流量卡手工绑定
+#   is_whitelist —— routes/api.py:/toggle_whitelist 写入的白名单标记
+APP_OWNED_KEYS = ('bound_iccid', 'is_whitelist')
+
+
+def ingest_device_data(scraped_list):
+    """
+    把爬虫返回的设备列表写入 Device（快照）和 DeviceHistory（历史）。
+
+    返回 (设备更新数, 历史追加数)。
+    """
+    if not scraped_list:
+        return 0, 0
+
+    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    stats = {'updated': 0, 'history': 0}
+
+    for item in scraped_list:
+        d_name = item.get('name')
+        if not d_name:
+            continue
+
+        # --- 1. 数据解包 ---
+        raw_status = item.get('status', '未知')
+        raw_value = item.get('value', '')
+        f_count = item.get('num_files', 0)
+        source = item.get('source', '自动爬虫')
+
+        # target_time 由爬虫层解析，为真实记录时间；离线/异常/没抓到时为 None。
+        # 注意：这里绝不回退到 current_time —— 那会把"采集时刻"冒充成"数据时刻"。
+        target_date = item.get('target_time')
+
+        raw_json = item.get('raw_json', {})
+
+        # --- 2. 设备主表更新 ---
+        device = Device.query.filter_by(name=d_name).first()
+
+        if not device:
+            device = Device(name=d_name, source=source, install_site="")
+            db.session.add(device)
+            db.session.flush()
+        elif device.source == 'iot_card':
+            # IoT 卡与爬虫设备共用 devices 表且靠 name 关联，撞名时以爬虫来源为准
+            device.source = source
+
+        device.status = raw_status
+        device.current_value = raw_value
+        device.check_time = current_time
+        device.file_count = f_count
+
+        # ✅ [核心逻辑] 只有爬虫拿到了真实业务时间才更新设备主表的时间。
+        # 拿不到（离线/异常）就保留上一次的有效值 —— 即"冻结"。
+        # 这样一台断线多天的设备，offset 会停在"滞后 N 天"而不是被刷成"当天"。
+        if target_date:
+            device.latest_time = target_date
+
+        # 注意：不再写入 device.offset。
+        # 该列是"写入时刻"的快照，采集一停摆就整体失真，现改为 Device.to_dict()
+        # 读取时用 calculate_offset(latest_time) 实时计算。列保留但已废弃。
+
+        # --- 3. JSON 数据：覆盖为本次最新，切断无限膨胀 ---
+        # 旧写法 old_json.update(raw_json) 会让主表 JSON 只增不减、永久累积。
+        # 但 APP_OWNED_KEYS 里的键由本应用自己的接口写入（不属于爬虫数据），
+        # 必须原样保留 —— 否则每次采集都会把用户的设备-流量卡绑定清掉。
+        old_json = {}
+        if device.json_data:
+            try:
+                old_json = json.loads(device.json_data)
+            except Exception:
+                pass
+
+        if isinstance(raw_json, dict) and raw_json:
+            new_json = dict(raw_json)
+            for key in APP_OWNED_KEYS:
+                if old_json.get(key) is not None:
+                    new_json[key] = old_json[key]
+            device.json_data = json.dumps(new_json, ensure_ascii=False)
+        # raw_json 为空（离线/异常）时保持原有 json_data 不变，避免单次爬虫失误清空状态
+
+        db.session.merge(device)
+        stats['updated'] += 1
+
+        # --- 4. 历史表写入 ---
+        # 历史记录的是"事件"，所以没有数据时间时用本次观测时间，
+        # 与主表的"冻结"策略不同：主表回答"数据到什么时候"，历史回答"什么时候采过"。
+        history_time = target_date if target_date else current_time
+        # [核心修复] 只存本次抓取的切片，不再复制主表那份越滚越大的累积 JSON。
+        # 旧写法 json_data=device.json_data 会让每一条历史都完整复制一遍全量数据，
+        # 历史表体积随采集次数线性膨胀。
+        incremental_json = json.dumps(raw_json, ensure_ascii=False) if raw_json else "{}"
+        history = DeviceHistory(
+            device_id=device.id,
+            status=raw_status,
+            result_data=raw_value,
+            data_time=history_time,
+            file_count=f_count,
+            json_data=incremental_json
+        )
+        db.session.add(history)
+        stats['history'] += 1
+
+    return stats['updated'], stats['history']
diff --git a/2_1banben/services/time_utils.py b/2_1banben/services/time_utils.py
new file mode 100644
index 0000000..b2bdf3b
--- /dev/null
+++ b/2_1banben/services/time_utils.py
@@ -0,0 +1,25 @@
+# services/time_utils.py
+"""
+时间辅助函数。
+
+刻意不依赖 flask / db / models，避免被 services.db_ingest 和 routes.api
+互相引用时产生循环导入。
+"""
+from datetime import datetime
+
+
+def calculate_offset(latest_time_str):
+    """
+    计算时间滞后天数
+    用于前端展示设备数据是否过时
+    """
+    if not latest_time_str or latest_time_str == "N/A":
+        return "从未同步"
+    try:
+        # 兼容处理 2026_01_13 和 2026-01-13 格式
+        clean = str(latest_time_str).split()[0].replace('_', '-')
+        target = datetime.strptime(clean, "%Y-%m-%d").date()
+        diff = (datetime.now().date() - target).days
+        return "当天" if diff == 0 else f"滞后 {diff} 天"
+    except:
+        return "时间解析失败"