DXC
15f79e69ad
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。
241 lines
8.1 KiB
Python
241 lines
8.1 KiB
Python
import os
|
||
import io
|
||
import sys
|
||
import json
|
||
import mimetypes
|
||
import logging
|
||
from datetime import datetime
|
||
import pytz
|
||
|
||
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. 核心模块引用
|
||
# ==============================================================================
|
||
try:
|
||
from config import Config
|
||
from extensions import db
|
||
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
|
||
except ImportError:
|
||
sync_iot_data_service = None
|
||
|
||
try:
|
||
from routes.api import api_bp as device_bp
|
||
from routes.api import calculate_offset
|
||
except ImportError:
|
||
# 兜底逻辑，防止缺失 calculate_offset 导致崩溃
|
||
def calculate_offset(target_time):
|
||
return 0
|
||
|
||
|
||
from routes.api import device_bp
|
||
|
||
except ImportError as e:
|
||
print(f"❌ [启动错误] 模块导入失败: {e}")
|
||
sys.exit(1)
|
||
|
||
# ==============================================================================
|
||
# 2. 智能路径配置
|
||
# ==============================================================================
|
||
RESOURCE_BASE = Config.BASE_DIR
|
||
INSTANCE_PATH = Config.INSTANCE_DIR
|
||
|
||
|
||
def find_static_folder(base_path):
|
||
"""
|
||
全能路径搜寻逻辑，适配 PyInstaller 打包环境
|
||
"""
|
||
if getattr(sys, 'frozen', False):
|
||
if hasattr(sys, '_MEIPASS'):
|
||
mei_path = os.path.join(sys._MEIPASS, 'web_dist')
|
||
if os.path.exists(os.path.join(mei_path, 'index.html')):
|
||
return mei_path
|
||
internal_path = os.path.join(base_path, '_internal', 'web_dist')
|
||
if os.path.exists(os.path.join(internal_path, 'index.html')):
|
||
return internal_path
|
||
|
||
path = os.path.join(base_path, 'web_dist')
|
||
if os.path.exists(os.path.join(path, 'index.html')):
|
||
return path
|
||
|
||
parent_path = os.path.join(os.path.dirname(base_path), 'web_dist')
|
||
if os.path.exists(os.path.join(parent_path, 'index.html')):
|
||
return parent_path
|
||
return path
|
||
|
||
|
||
STATIC_FOLDER = find_static_folder(RESOURCE_BASE)
|
||
mimetypes.add_type('application/javascript', '.js')
|
||
mimetypes.add_type('text/css', '.css')
|
||
|
||
|
||
# ==============================================================================
|
||
# 3. 核心定时任务逻辑 (深度优化版)
|
||
# ==============================================================================
|
||
def auto_monitor_job(app):
|
||
"""
|
||
每天的定时采集任务。
|
||
|
||
入库细节全部收敛到 services.db_ingest.ingest_device_data，本函数只负责：
|
||
建立应用上下文 -> 触发爬虫 -> 调用入库管道 -> 提交事务。
|
||
"""
|
||
with app.app_context():
|
||
# 强制清理会话，确保线程获取的是全新的数据库连接
|
||
db.session.remove()
|
||
|
||
tz = pytz.timezone('Asia/Shanghai')
|
||
now_str = datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
print(f"\n{'=' * 50}")
|
||
print(f"⏰ [定时任务启动] {now_str}")
|
||
|
||
if not execute_monitor_task:
|
||
print("❌ 错误: execute_monitor_task 未定义")
|
||
return
|
||
|
||
try:
|
||
task_result = execute_monitor_task()
|
||
|
||
if not task_result:
|
||
print("⚠️ [警告] 爬虫执行完毕，但返回空数据")
|
||
return
|
||
|
||
scraped_list = task_result.get('device_list', [])
|
||
print(f"📦 [数据获取] 爬取到 {len(scraped_list)} 条设备数据")
|
||
|
||
updated, history = ingest_device_data(scraped_list)
|
||
|
||
db.session.commit()
|
||
print(f"✅ [入库成功] 设备更新: {updated} | 历史追加: {history}")
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
print(f"❌ [严重异常] 数据写入失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
finally:
|
||
# 再次清理 Session，防止内存泄漏或污染下一次任务
|
||
db.session.remove()
|
||
print(f"{'=' * 50}\n")
|
||
|
||
|
||
# ==============================================================================
|
||
# 4. Flask 应用工厂
|
||
# ==============================================================================
|
||
def create_app():
|
||
print(f"🔍 [前端路径锁定] {STATIC_FOLDER}")
|
||
|
||
app = Flask(__name__, static_folder=STATIC_FOLDER, instance_path=INSTANCE_PATH)
|
||
CORS(app)
|
||
|
||
if not os.path.exists(app.instance_path):
|
||
os.makedirs(app.instance_path, exist_ok=True)
|
||
|
||
app.config.from_object(Config)
|
||
|
||
# 初始化 DB
|
||
db.init_app(app)
|
||
|
||
# 初始化调度器
|
||
scheduler = APScheduler()
|
||
scheduler.init_app(app)
|
||
scheduler.start()
|
||
|
||
# --- 添加定时任务 ---
|
||
# 注意：这里我们传递 [app] 作为参数，确保 job 函数内能获取到 app 上下文
|
||
scheduler.add_job(
|
||
id='daily_monitor_task',
|
||
func=auto_monitor_job,
|
||
args=[app],
|
||
trigger='cron',
|
||
hour=17,
|
||
minute=00,
|
||
second=00,
|
||
misfire_grace_time=3600,
|
||
timezone=pytz.timezone('Asia/Shanghai')
|
||
)
|
||
print(f"📅 定时任务已锁定: 每天北京时间 17:00 执行")
|
||
|
||
app.register_blueprint(device_bp)
|
||
|
||
@app.route('/api/force_run')
|
||
def force_run_task():
|
||
"""手动触发接口：复用同一个 auto_monitor_job 函数，确保逻辑一致"""
|
||
auto_monitor_job(app)
|
||
return jsonify({'code': 200, 'msg': '手动触发成功，请查看服务器日志'})
|
||
|
||
@app.route('/')
|
||
def serve_index():
|
||
try:
|
||
return send_from_directory(app.static_folder, 'index.html')
|
||
except Exception:
|
||
return "Frontend Error", 404
|
||
|
||
@app.route('/<path:path>')
|
||
def serve_static(path):
|
||
if path.startswith('api'):
|
||
return jsonify({'code': 404, 'message': 'API endpoint not found'}), 404
|
||
|
||
file_path = os.path.join(app.static_folder, path)
|
||
if os.path.exists(file_path):
|
||
return send_from_directory(app.static_folder, path)
|
||
|
||
return send_from_directory(app.static_folder, 'index.html')
|
||
|
||
with app.app_context():
|
||
db.create_all()
|
||
|
||
return app
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app = create_app()
|
||
debug_mode = not getattr(sys, 'frozen', False)
|
||
|
||
print(f"🚀 服务启动中... 数据库: {app.config['SQLALCHEMY_DATABASE_URI']}")
|
||
# 注意：use_reloader=False 防止调度器在 Debug 模式下运行两次
|
||
app.run(host='0.0.0.0', port=5000, debug=debug_mode, use_reloader=False) |