import os import sys import json import mimetypes from datetime import datetime from flask import Flask, send_from_directory, jsonify from flask_cors import CORS from flask_apscheduler import APScheduler # ============================================================================== # ✅ 1. 核心模块引用 # ============================================================================== try: # 导入配置类 from config import Config # 数据库实例 from extensions import db # 数据模型 from models import Device, DeviceHistory # 核心业务逻辑 (爬虫) from services.core import execute_monitor_task # 核心业务逻辑 (IoT) - 用于定时任务 from services.iot_api import sync_iot_data_service # 路由蓝图 try: from routes.api import api_bp as device_bp # 导入保存逻辑，供定时任务复用 from routes.api import save_iot_cards_to_db, calculate_offset except ImportError: from routes.api import device_bp, save_iot_cards_to_db, calculate_offset except ImportError as e: print(f"❌ 严重错误: 模块导入失败。请检查文件名和变量名。详细信息: {e}") sys.exit(1) # ============================================================================== # 2. 路径计算 (核心修复：区分资源路径和数据路径) # ============================================================================== def get_paths(): """ 计算关键路径： 1. resource_base: 用于存放 web_dist (打包后在临时目录) 2. data_base: 用于存放数据库 (打包后在 exe 旁边) """ if getattr(sys, 'frozen', False): # --- 打包环境 (PyInstaller) --- # 资源文件在临时解压目录 (sys._MEIPASS) resource_base = sys._MEIPASS # 数据文件(数据库)在 exe 所在目录 (sys.executable 的父目录) data_base = os.path.dirname(sys.executable) else: # --- 开发环境 --- base = os.path.abspath(os.path.dirname(__file__)) resource_base = base data_base = base return resource_base, data_base # 获取路径 RESOURCE_BASE, DATA_BASE = get_paths() # 定义具体文件夹路径 STATIC_FOLDER = os.path.join(RESOURCE_BASE, 'web_dist') # ⚠️ 关键：强制将 instance 文件夹定位到数据目录 (exe旁边)，而不是临时目录 INSTANCE_PATH = os.path.join(DATA_BASE, 'instance') # 修复 Windows MIME 类型 mimetypes.add_type('application/javascript', '.js') mimetypes.add_type('text/css', '.css') # ============================================================================== # 3. 定时任务逻辑 (保持不变) # ============================================================================== def auto_monitor_job(app): """定时任务具体执行逻辑""" with app.app_context(): print(f"⏰ [定时任务] 启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # --- 任务 A: 爬虫更新 --- if execute_monitor_task: try: 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 = 0 for item in scraped_list: d_name = item.get('name') if not d_name: continue 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() device.status = item.get('status') device.current_value = item.get('value') device.latest_time = item.get('target_time') device.check_time = current_time device.json_data = json.dumps(item.get('raw_json', {}), ensure_ascii=False) device.offset = calculate_offset(item.get('target_time')) db.session.add(DeviceHistory( device_id=device.id, status=device.status, result_data=device.current_value, data_time=item.get('target_time'), json_data=device.json_data )) count += 1 print(f"✅ [定时任务-爬虫] 更新 {count} 台") else: print("⚠️ [定时任务-爬虫] 未获取到数据") except Exception as e: print(f"❌ [定时任务-爬虫] 异常: {e}") # --- 任务 B: IoT 同步 --- if sync_iot_data_service: try: # 1. 获取数据 iot_list = sync_iot_data_service() # 2. 保存入库 (复用 api.py 中的逻辑) count_iot, err = save_iot_cards_to_db(iot_list) if err: print(f"❌ [定时任务-IoT] 错误: {err}") else: print(f"✅ [定时任务-IoT] 更新 {count_iot} 张") except Exception as e: print(f"❌ [定时任务-IoT] 异常: {e}") # 统一提交事务 try: db.session.commit() print("💾 [定时任务] 数据库事务提交完成") except Exception as e: db.session.rollback() print(f"❌ [定时任务] 提交失败: {e}") # ============================================================================== # 4. Flask 应用工厂 # ============================================================================== def create_app(): # ⚠️ 关键修改：显式传入 instance_path，告诉 Flask 去哪里找/存 数据库文件 app = Flask(__name__, static_folder=STATIC_FOLDER, instance_path=INSTANCE_PATH) CORS(app) # 1. 确保 instance 目录存在 (在 exe 旁边创建文件夹) if not os.path.exists(app.instance_path): try: os.makedirs(app.instance_path, exist_ok=True) print(f"📁 已创建数据目录: {app.instance_path}") except OSError as e: print(f"❌ 无法创建数据目录 {app.instance_path}: {e}") # 2. 加载配置 app.config.from_object(Config) # ⚠️ 关键修改：强制重写数据库 URI，确保使用绝对路径 # 即使 Config 里写了，这里也要确保它指向我们刚才计算出的 INSTANCE_PATH db_name = 'monitor_data.db' # 你的数据库文件名 db_path = os.path.join(app.instance_path, db_name) # Windows下路径分隔符处理，防止报错 if sys.platform.startswith('win'): app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}' else: app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:////{db_path}' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False print(f"💾 数据库路径锁定为: {db_path}") # 3. 初始化扩展 db.init_app(app) scheduler = APScheduler() scheduler.init_app(app) scheduler.start() # 4. 添加定时任务 (每天 12:00) scheduler.add_job( id='daily_monitor_task', func=auto_monitor_job, args=[app], trigger='cron', hour=12, minute=0 ) # 5. 注册路由蓝图 app.register_blueprint(device_bp) # ------------------------------------------------- # 前端路由支持 # ------------------------------------------------- @app.route('/') def serve_index(): index_path = os.path.join(app.static_folder, 'index.html') if not os.path.exists(index_path): return f"❌ 错误: 前端文件丢失 ({index_path})", 404 return send_from_directory(app.static_folder, 'index.html') @app.route('/') def serve_static(path): file_path = os.path.join(app.static_folder, path) if os.path.exists(file_path): return send_from_directory(app.static_folder, path) if path.startswith('api') or path.startswith('static'): return jsonify({'code': 404, 'message': 'Not Found'}), 404 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("🚀 服务启动中...") # use_reloader=False 防止定时任务执行两次 app.run(host='0.0.0.0', port=5000, debug=debug_mode, use_reloader=False)