duxingchen/ZDXX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test

Browse Source
This commit is contained in:
YueL1331
2026-01-07 15:57:34 +08:00
parent 15d66d6694
commit ef440177b3
16 changed files with 354 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
1.1/test1.py
View File

@ -1,10 +1,11 @@
import os
import sys
import json
import threading
import requests
import logging
from datetime import datetime
from flask import Flask, jsonify
from flask import Flask, jsonify, send_from_directory
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from flask_apscheduler import APScheduler
@ -13,35 +14,62 @@ from lxml import etree
# --- 配置日志 ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
app = Flask(__name__)
# --- 关键路径处理函数 (适配 PyInstaller) ---
def get_base_path():
"""获取运行时及其所在目录，适配开发环境和打包后的EXE环境"""
if getattr(sys, 'frozen', False):
# 如果是打包后的 exe，sys.executable 是 exe 的路径
return os.path.dirname(sys.executable)
# 开发环境下，是当前脚本的路径
return os.path.dirname(os.path.abspath(__file__))
def get_static_path():
"""获取 Vue 静态资源 dist 的路径"""
if getattr(sys, 'frozen', False):
# PyInstaller 打包时，资源文件会被解压到 sys._MEIPASS 临时目录
# 我们需要在打包命令中指定 --add-data "dist;dist"
return os.path.join(sys._MEIPASS, 'dist')
# 开发环境
return os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dist')
# --- Flask 初始化 ---
# static_folder 指向 Vue 打包后的 dist 目录
# static_url_path='' 表示静态文件不需要 /static 前缀
dist_folder = get_static_path()
app = Flask(__name__, static_folder=dist_folder, static_url_path='')
CORS(app)
# --- 数据库配置 ---
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///monitor_data.db'
# 确保数据库生成在 exe 同级目录下，而不是临时文件夹中
db_path = os.path.join(get_base_path(), 'monitor_data.db')
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SCHEDULER_API_ENABLED'] = True # 允许通过API查看调度任务
app.config['SCHEDULER_API_ENABLED'] = True
db = SQLAlchemy(app)
scheduler = APScheduler()
# --- 模型定义 ---
# --- 模型定义 (保持不变) ---
class MonitorRecord(db.Model):
id = db.Column(db.Integer, primary_key=True)
source = db.Column(db.String(50))
name = db.Column(db.String(100))
status = db.Column(db.String(50)) # 正常 / 离线 / 异常
status = db.Column(db.String(50))
reason = db.Column(db.String(255))
offset = db.Column(db.String(50))
latest_time = db.Column(db.String(50)) # 数据时间
check_time = db.Column(db.String(50)) # 爬取时间
latest_time = db.Column(db.String(50))
check_time = db.Column(db.String(50))
content = db.Column(db.Text, nullable=True)
with app.app_context():
db.create_all()
# --- 爬虫配置 ---
# --- 爬虫配置 (保持不变) ---
CONFIG = {
"106": {
"base_url": "http://106.75.72.40:7500/api/proxy/tcp",
@ -57,7 +85,7 @@ CONFIG = {
is_running = False
# --- 核心辅助函数 ---
# --- 核心辅助函数 (保持不变) ---
def calculate_offset(latest_time_str):
if not latest_time_str or latest_time_str == "N/A":
return "从未同步"
@ -72,9 +100,6 @@ def calculate_offset(latest_time_str):
def save_record(source, name, status, reason, latest_time="N/A", content=None):
"""
Upsert 逻辑: 有则更新，无则插入
"""
record = MonitorRecord.query.filter_by(source=source, name=name).first()
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
current_offset = calculate_offset(latest_time)
@ -82,11 +107,9 @@ def save_record(source, name, status, reason, latest_time="N/A", content=None):
if record:
if content is not None: record.content = content
if latest_time != "N/A": record.latest_time = latest_time
record.status = status
record.reason = reason
record.check_time = now_str
# 使用当前库里的时间重新计算 offset
time_base = latest_time if latest_time != "N/A" else record.latest_time
record.offset = calculate_offset(time_base)
else:
@ -96,17 +119,15 @@ def save_record(source, name, status, reason, latest_time="N/A", content=None):
check_time=now_str, content=content
)
db.session.add(new_record)
try:
db.session.commit()
except Exception as e:
db.session.rollback()
logging.error(f"DB Error: {e}")
return f"{source}_{name}"
# --- 106 逻辑 ---
# --- 业务逻辑函数 (保持不变) ---
def get_106_dynamic_token(port):
try:
login_url = f"http://106.75.72.40:{port}/api/login"
@ -140,6 +161,7 @@ def find_closest_item(items, is_date_level=True):
def run_106_logic(active_set):
# (保持原样，省略以节省空间，直接用你原本的逻辑即可)
c = CONFIG["106"]
today_str = datetime.now().strftime("%Y_%m_%d")
main_headers = {"Authorization": c["primary_auth"], "User-Agent": "Mozilla/5.0"}
@ -150,12 +172,10 @@ def run_106_logic(active_set):
name = item.get('name', '')
if not name.lower().endswith('_data'): continue
if "TOWER" not in name.upper(): continue
if str(item.get('status')).lower() != 'online':
key = save_record("106网站", name, "离线", f"设备状态: {item.get('status')}")
active_set.add(key)
continue
try:
port = item.get('conf', {}).get('remote_port')
token = get_106_dynamic_token(port)
@ -163,35 +183,24 @@ def run_106_logic(active_set):
key = save_record("106网站", name, "异常", "Token获取失败")
active_set.add(key)
continue
headers = {"Authorization": c["primary_auth"], "x-auth": token}
api_root = "/api/resources/Data/" if "TOWER_" in name.upper() else "/api/resources/data/"
# 寻找日期
res1 = requests.get(f"http://106.75.72.40:{port}{api_root}", headers=headers, timeout=10)
best_date = find_closest_item(res1.json().get('items', []), True)
if not best_date or best_date[2] != today_str:
key = save_record("106网站", name, "正常", "未找到今日文件夹",
latest_time=best_date[2] if best_date else "N/A")
active_set.add(key)
continue
# 寻找文件
date_path = f"{api_root}{best_date[2]}/"
res2 = requests.get(f"http://106.75.72.40:{port}{date_path}", headers=headers, timeout=10)
best_file = find_closest_item(res2.json().get('items', []), False)
if not best_file:
key = save_record("106网站", name, "正常", "今日文件夹为空", latest_time=today_str)
active_set.add(key)
continue
# 获取内容
file_item = best_file[1]
full_path = file_item.get('path') or f"{date_path}{file_item.get('name')}"
# 判断下载方式
is_tower_i = "TOWER" in name.upper() and "TOWER_" not in name.upper()
if is_tower_i:
download_url = f"http://106.75.72.40:{port}/api/raw{full_path}"
@ -201,10 +210,8 @@ def run_106_logic(active_set):
file_api_url = f"http://106.75.72.40:{port}/api/resources{full_path}"
res3 = requests.get(file_api_url, headers=headers, timeout=20)
final_content = res3.json().get('content', '')
key = save_record("106网站", name, "正常", "同步成功", latest_time=today_str, content=final_content)
active_set.add(key)
except Exception as e:
key = save_record("106网站", name, "异常", f"采集错误: {str(e)[:50]}")
active_set.add(key)
@ -212,15 +219,14 @@ def run_106_logic(active_set):
logging.error(f"106 Global Error: {e}")
# --- 82 逻辑 ---
def run_82_logic(active_set):
# (保持原样，直接用你原本的逻辑即可)
c = CONFIG["82"]
session = requests.Session()
try:
session.post(f"{c['base_url']}/login.php", data=c["login"], timeout=10)
resp = session.post(f"{c['base_url']}/GetStationList.php", timeout=10)
stations = etree.HTML(resp.content).xpath('//option/@value')
for sid in [s for s in stations if s]:
try:
r = session.post(f"{c['base_url']}/getLastWeatherData.php", data=str(sid),
@ -242,22 +248,15 @@ def run_82_logic(active_set):
logging.error(f"82 Global Error: {e}")
# --- 核心任务逻辑 ---
def execute_monitor_task():
global is_running
if is_running:
logging.warning("Task already running, skipping...")
return
if is_running: return
is_running = True
logging.info("Starting monitor task...")
with app.app_context():
active_set = set()
run_106_logic(active_set)
run_82_logic(active_set)
# 掉线处理
all_records = MonitorRecord.query.all()
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
for record in all_records:
@ -266,17 +265,15 @@ def execute_monitor_task():
record.reason = "设备本次未出现"
record.check_time = now_str
record.offset = calculate_offset(record.latest_time)
try:
db.session.commit()
except:
db.session.rollback()
is_running = False
logging.info("Monitor task finished.")
# --- 路由 ---
# --- API 路由 (保持不变) ---
@app.route('/api/run', methods=['POST'])
def manual_start():
if is_running: return jsonify({"status": "busy"}), 400
@ -298,18 +295,33 @@ def logs():
} for l in data])
# --- 调度器配置 ---
# id: 任务ID, func: 任务函数(字符串路径或引用), trigger: cron(定时), hour/minute: 时间
# --- 新增: 前端页面托管路由 ---
@app.route('/')
def serve_index():
return send_from_directory(app.static_folder, 'index.html')
@app.route('/<path:path>')
def serve_static_files(path):
# 尝试在 dist 目录寻找文件 (css, js, icons)
file_path = os.path.join(app.static_folder, path)
if os.path.exists(file_path):
return send_from_directory(app.static_folder, path)
# 如果找不到文件（例如刷新页面时的路由），返回index.html让Vue Router处理
return send_from_directory(app.static_folder, 'index.html')
# --- 调度器与启动 ---
@scheduler.task('cron', id='daily_job', hour=10, minute=0)
def auto_run_task():
with app.app_context():
logging.info("Auto scheduler triggered.")
# 在新线程中运行，避免阻塞调度器主线程
threading.Thread(target=execute_monitor_task).start()
if __name__ == "__main__":
scheduler.init_app(app)
scheduler.start()
# 注意：debug=True 可能会导致调度器在开发模式下运行两次，生产环境建议关闭 debug
app.run(debug=True, port=5000, use_reloader=False)
# Host='0.0.0.0' 允许外部IP访问
# Port=5000 (确保 Windows 防火墙开放了此端口)
print("应用正在启动... 请确保 dist 文件夹与脚本/exe 同级或已被打包")
app.run(host='0.0.0.0', port=5000, debug=False, use_reloader=False)