From 15f79e69ad69d04ec0bf9ba1958ae671913cde82 Mon Sep 17 00:00:00 2001 From: DXC Date: Tue, 15 Sep 2026 17:05:43 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8E=A7=E5=88=B6=E5=8F=B0=E5=BC=BA?= =?UTF-8?q?=E5=88=B6=20UTF-8=EF=BC=8C=E9=81=BF=E5=85=8D=20emoji=20?= =?UTF-8?q?=E6=89=93=E5=8D=B0=E5=BC=82=E5=B8=B8=E6=8B=96=E5=9E=AE=E9=87=87?= =?UTF-8?q?=E9=9B=86=E4=BA=8B=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 134 ++++++++++++---------------------- 2_1banben/device_monitor.spec | 44 +++++++++++ 2_1banben/fix_fake_time.py | 21 ++++++ 3 files changed, 111 insertions(+), 88 deletions(-) create mode 100644 2_1banben/device_monitor.spec create mode 100644 2_1banben/fix_fake_time.py diff --git a/2_1banben/app.py b/2_1banben/app.py index 4692552..5295e9e 100644 --- a/2_1banben/app.py +++ b/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() 强制清理旧连接 - 3. 使用 db.session.merge() 确保对象状态被正确追踪 - 4. 增加详细日志，对比爬虫返回的数据与入库行为 + 每天的定时采集任务。 + + 入库细节全部收敛到 services.db_ingest.ingest_device_data，本函数只负责： + 建立应用上下文 -> 触发爬虫 -> 调用入库管道 -> 提交事务。 """ 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") - stats = {'updated': 0, 'history': 0} + updated, history = ingest_device_data(scraped_list) - 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") diff --git a/2_1banben/device_monitor.spec b/2_1banben/device_monitor.spec new file mode 100644 index 0000000..4aad3f7 --- /dev/null +++ b/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', +) diff --git a/2_1banben/fix_fake_time.py b/2_1banben/fix_fake_time.py new file mode 100644 index 0000000..5fe7ca1 --- /dev/null +++ b/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} 台处于离线但伪造了时间的设备！")