fix: 控制台强制 UTF-8，避免 emoji 打印异常拖垮采集事务

Windows 中文控制台默认 cp936(GBK)，而 app.py / services.core / crawler_106 /
crawler_82 大量使用 emoji 打印日志。print 一旦抛 UnicodeEncodeError，异常会
落进采集任务的 try 块，导致整个事务被 rollback 并报“数据写入失败”，表现就是
定时采集长期静默不落库。

实测：在 GBK 管道下 create_app() 直接崩在 app.py 的 emoji print 上。

- app.py 顶部（所有业务 import 之前）强制 stdout/stderr 为 UTF-8。
- 比常见写法多两层守卫：encoding 可能为 None（.lower() 会 AttributeError），
  PyInstaller --noconsole 或重定向时可能没有 buffer —— 这两个恰好是我们要防的
  场景。替换前先 flush，否则旧缓冲区里未写出的内容会随旧对象一起丢。
- 保留原始流引用，防止被 GC 回收时连带关闭底层 buffer。
- auto_monitor_job 瘦身：入库细节全部交给 ingest_device_data。

同时补入 device_monitor.spec（此前未纳入版本管理）与一次性清洗脚本
fix_fake_time.py。

2_1banben/app.py

@@ -1,4 +1,5 @@
 import os
+import io
 import sys
 import json
 import mimetypes
@@ -10,6 +11,41 @@ from flask import Flask, send_from_directory, jsonify
 from flask_cors import CORS
 from flask_apscheduler import APScheduler
 
+# ==============================================================================
+# ✅ 0. 控制台编码兜底（必须最先执行）
+# ==============================================================================
+# 保留原始流引用，防止被 GC 回收时连带关闭底层 buffer
+_CONSOLE_ORIGINALS = []
+
+
+def _force_utf8_console():
+    """
+    强制 stdout/stderr 以 UTF-8 输出，防止 Windows GBK 终端下 emoji 崩溃。
+
+    Windows 中文控制台默认 cp936(GBK)，而本应用（app.py / services.core /
+    crawler_106 / crawler_82）大量使用 emoji 打印日志。一旦 print 抛
+    UnicodeEncodeError，异常会落进采集任务的 try 块，导致整个事务被 rollback
+    并报"数据写入失败" —— 表现就是定时采集长期静默不落库。
+    """
+    for name in ('stdout', 'stderr'):
+        stream = getattr(sys, name, None)
+        # PyInstaller --noconsole 或输出重定向时，stream 或其 buffer 可能不存在
+        if stream is None or not hasattr(stream, 'buffer'):
+            continue
+        enc = (getattr(stream, 'encoding', None) or '').lower().replace('-', '')
+        if enc == 'utf8':
+            continue
+        # 替换前先 flush：否则旧流缓冲区里尚未写出的内容会随旧对象一起丢掉
+        try:
+            stream.flush()
+        except Exception:
+            pass
+        _CONSOLE_ORIGINALS.append(stream)
+        setattr(sys, name, io.TextIOWrapper(stream.buffer, encoding='utf-8', errors='replace'))
+
+
+_force_utf8_console()
+
 # ==============================================================================
 # ✅ 1. 核心模块引用
 # ==============================================================================
@@ -19,6 +55,8 @@ try:
     from models import Device, DeviceHistory
     # 引入核心爬虫调度
     from services.core import execute_monitor_task
+    # 引入统一入库管道
+    from services.db_ingest import ingest_device_data
 
     try:
         from services.iot_api import sync_iot_data_service
@@ -80,14 +118,13 @@ mimetypes.add_type('text/css', '.css')
 # ==============================================================================
 def auto_monitor_job(app):
     """
-    [关键修复]
+    每天的定时采集任务。
-    1. 使用 app.app_context() 确保线程中有 Flask 上下文
+
-    2. 使用 db.session.remove() 强制清理旧连接
+    入库细节全部收敛到 services.db_ingest.ingest_device_data，本函数只负责：
-    3. 使用 db.session.merge() 确保对象状态被正确追踪
+    建立应用上下文 -> 触发爬虫 -> 调用入库管道 -> 提交事务。
-    4. 增加详细日志，对比爬虫返回的数据与入库行为
     """
     with app.app_context():
-        # A. 强制清理会话，确保线程获取的是全新的数据库连接
+        # 强制清理会话，确保线程获取的是全新的数据库连接
         db.session.remove()
 
         tz = pytz.timezone('Asia/Shanghai')
@@ -101,7 +138,6 @@ def auto_monitor_job(app):
             return
 
         try:
-            # B. 执行爬虫
             task_result = execute_monitor_task()
 
             if not task_result:
@@ -111,96 +147,18 @@ def auto_monitor_job(app):
             scraped_list = task_result.get('device_list', [])
             print(f"📦 [数据获取] 爬取到 {len(scraped_list)} 条设备数据")
 
-            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            updated, history = ingest_device_data(scraped_list)
-            stats = {'updated': 0, 'history': 0}
 
-            for item in scraped_list:
-                d_name = item.get('name')
-                if not d_name: continue
-
-                # --- 1. 数据解包与默认值处理 ---
-                # 显式提取，防止 None 覆盖数据库现有的值（如果业务需要）
-                # 这里假设爬虫返回 None 就是要写入 None，或者空字符串
-                raw_status = item.get('status', '未知')
-                raw_value = item.get('value', '')
-                f_count = item.get('num_files', 0)
-
-                # 时间处理：必须有时间，否则用当前时间
-                target_date = item.get('target_time')
-                if not target_date:
-                    target_date = current_time
-
-                raw_json = item.get('raw_json', {})
-
-                # [调试日志] 仅打印第一条或特定的设备，防止刷屏，但能帮你确认数据是否为空
-                # if '0025' in d_name:
-                #     print(f"   >>> [写入前检查] {d_name}: Value='{raw_value}' | Files={f_count}")
-
-                # --- 2. 数据库操作 (使用 Merge 机制) ---
-                # 先尝试查询
-                device = Device.query.filter_by(name=d_name).first()
-
-                if not device:
-                    # 如果不存在，新建对象
-                    device = Device(name=d_name, source=item.get('source', '自动爬虫'), install_site="")
-                    db.session.add(device)
-                    db.session.flush()  # 立即获取 ID
-
-                # 更新字段
-                device.status = raw_status
-                device.current_value = raw_value
-                device.latest_time = target_date
-                device.check_time = current_time
-                device.file_count = f_count
-
-                # 计算 Offset
-                try:
-                    device.offset = calculate_offset(target_date)
-                except:
-                    device.offset = 0
-
-                # JSON 数据合并
-                old_json = {}
-                try:
-                    if device.json_data:
-                        old_json = json.loads(device.json_data)
-                except:
-                    old_json = {}
-
-                if isinstance(raw_json, dict):
-                    old_json.update(raw_json)
-
-                device.json_data = json.dumps(old_json, ensure_ascii=False)
-
-                # [核心修复] 使用 merge 告诉 Session "这个对象归你管，请更新它"
-                # 这能解决后台线程中 "DetachedInstanceError" 或更新丢失的问题
-                db.session.merge(device)
-                stats['updated'] += 1
-
-                # --- 3. 写入历史记录 ---
-                history = DeviceHistory(
-                    device_id=device.id,
-                    status=raw_status,
-                    result_data=raw_value,
-                    data_time=target_date,
-                    file_count=f_count,
-                    json_data=device.json_data
-                )
-                db.session.add(history)
-                stats['history'] += 1
-
-            # C. 提交事务
             db.session.commit()
-            print(f"✅ [入库成功] 设备更新: {stats['updated']} | 历史追加: {stats['history']}")
+            print(f"✅ [入库成功] 设备更新: {updated} | 历史追加: {history}")
 
         except Exception as e:
             db.session.rollback()
             print(f"❌ [严重异常] 数据写入失败: {e}")
-            # 打印堆栈以便排查
             import traceback
             traceback.print_exc()
         finally:
-            # D. 再次清理 Session，防止内存泄漏或污染下一次任务
+            # 再次清理 Session，防止内存泄漏或污染下一次任务
             db.session.remove()
             print(f"{'=' * 50}\n")
 

2_1banben/device_monitor.spec

@@ -0,0 +1,44 @@
+# -*- mode: python ; coding: utf-8 -*-
+
+
+a = Analysis(
+    ['app.py'],
+    pathex=[],
+    binaries=[],
+    datas=[('web_dist', 'web_dist')],
+    hiddenimports=[],
+    hookspath=[],
+    hooksconfig={},
+    runtime_hooks=[],
+    excludes=[],
+    noarchive=False,
+    optimize=0,
+)
+pyz = PYZ(a.pure)
+
+exe = EXE(
+    pyz,
+    a.scripts,
+    [],
+    exclude_binaries=True,
+    name='device_monitor',
+    debug=False,
+    bootloader_ignore_signals=False,
+    strip=False,
+    upx=True,
+    console=True,
+    disable_windowed_traceback=False,
+    argv_emulation=False,
+    target_arch=None,
+    codesign_identity=None,
+    entitlements_file=None,
+)
+coll = COLLECT(
+    exe,
+    a.binaries,
+    a.datas,
+    strip=False,
+    upx=True,
+    upx_exclude=[],
+    name='device_monitor',
+)

2_1banben/fix_fake_time.py

@@ -0,0 +1,21 @@
+from app import create_app
+from extensions import db
+from models import Device
+
+app = create_app()
+
+with app.app_context():
+    # 查找状态异常但 offset 被判定为当天的假数据
+    fake_devices = Device.query.filter(
+        Device.status.in_(['离线', '异常', '已离线', 'offline', '未知']),
+        Device.offset == '当天'
+    ).all()
+
+    count = 0
+    for dev in fake_devices:
+        dev.latest_time = None
+        dev.offset = '从未同步'
+        count += 1
+
+    db.session.commit()
+    print(f"✅ 成功清洗了 {count} 台处于离线但伪造了时间的设备！")