315 lines
11 KiB
Python
315 lines
11 KiB
Python
import os
|
||
import json
|
||
import threading
|
||
import requests
|
||
import logging
|
||
from datetime import datetime
|
||
from flask import Flask, jsonify
|
||
from flask_sqlalchemy import SQLAlchemy
|
||
from flask_cors import CORS
|
||
from flask_apscheduler import APScheduler
|
||
from lxml import etree
|
||
|
||
# --- 配置日志 ---
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
||
app = Flask(__name__)
|
||
CORS(app)
|
||
|
||
# --- 数据库配置 ---
|
||
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///monitor_data.db'
|
||
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||
app.config['SCHEDULER_API_ENABLED'] = True # 允许通过API查看调度任务
|
||
|
||
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)) # 正常 / 离线 / 异常
|
||
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)) # 爬取时间
|
||
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",
|
||
"primary_auth": "Basic YWRtaW46bGljYWhr",
|
||
"login_payload": {"username": "admin", "password": "licahk", "recaptcha": ""}
|
||
},
|
||
"82": {
|
||
"base_url": "http://82.156.1.111/weather/php",
|
||
"login": {'username': 'renlixin', 'password': 'licahk', 'login': '123'}
|
||
}
|
||
}
|
||
|
||
is_running = False
|
||
|
||
|
||
# --- 核心辅助函数 ---
|
||
def calculate_offset(latest_time_str):
|
||
if not latest_time_str or latest_time_str == "N/A":
|
||
return "从未同步"
|
||
try:
|
||
clean_date_str = str(latest_time_str).split()[0].replace('_', '-')
|
||
target_date = datetime.strptime(clean_date_str, "%Y-%m-%d").date()
|
||
diff = (datetime.now().date() - target_date).days
|
||
if diff == 0: return "当天已同步"
|
||
return f"滞后 {diff} 天"
|
||
except:
|
||
return "时间解析失败"
|
||
|
||
|
||
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)
|
||
|
||
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:
|
||
new_record = MonitorRecord(
|
||
source=source, name=name, status=status, reason=reason,
|
||
offset=current_offset, latest_time=latest_time,
|
||
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"
|
||
resp = requests.post(login_url, json=CONFIG["106"]["login_payload"], timeout=10)
|
||
return resp.text.strip().replace('"', '') if resp.status_code == 200 else None
|
||
except:
|
||
return None
|
||
|
||
|
||
def find_closest_item(items, is_date_level=True):
|
||
if not items or not isinstance(items, list): return None
|
||
today = datetime.now()
|
||
scored_items = []
|
||
for item in items:
|
||
name_val = item.get('name', '')
|
||
path_val = item.get('path', '')
|
||
target_str = name_val if name_val else path_val.split('/')[-1]
|
||
try:
|
||
if is_date_level:
|
||
current_date = datetime.strptime(target_str, "%Y_%m_%d")
|
||
else:
|
||
mod_str = item.get('modified', '')
|
||
current_date = datetime.fromisoformat(mod_str.replace('Z', '+00:00'))
|
||
diff = abs((today - current_date.replace(tzinfo=None)).total_seconds())
|
||
scored_items.append((diff, item, target_str))
|
||
except:
|
||
continue
|
||
if not scored_items: return None
|
||
scored_items.sort(key=lambda x: x[0])
|
||
return scored_items[0]
|
||
|
||
|
||
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"}
|
||
try:
|
||
resp = requests.get(c["base_url"], headers=main_headers, timeout=20)
|
||
proxies = resp.json().get('proxies', [])
|
||
for item in proxies:
|
||
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)
|
||
if not token:
|
||
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}"
|
||
res3 = requests.get(download_url, headers=headers, timeout=20)
|
||
final_content = f"Binary Data Size: {len(res3.content)}"
|
||
else:
|
||
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)
|
||
except Exception as e:
|
||
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),
|
||
headers={'Content-Type': 'text/plain'}, timeout=10)
|
||
data = r.json()
|
||
if data:
|
||
d_list = data.get('date', [])
|
||
latest = str(d_list[-1]) if d_list else "N/A"
|
||
key = save_record("82网站", sid, "正常", "同步成功", latest_time=latest,
|
||
content=json.dumps(data, ensure_ascii=False))
|
||
active_set.add(key)
|
||
else:
|
||
key = save_record("82网站", sid, "异常", "返回空数据")
|
||
active_set.add(key)
|
||
except:
|
||
key = save_record("82网站", sid, "异常", "单个采集失败")
|
||
active_set.add(key)
|
||
except Exception as e:
|
||
logging.error(f"82 Global Error: {e}")
|
||
|
||
|
||
# --- 核心任务逻辑 ---
|
||
def execute_monitor_task():
|
||
global is_running
|
||
if is_running:
|
||
logging.warning("Task already running, skipping...")
|
||
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:
|
||
if f"{record.source}_{record.name}" not in active_set:
|
||
record.status = "已离线"
|
||
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.")
|
||
|
||
|
||
# --- 路由 ---
|
||
@app.route('/api/run', methods=['POST'])
|
||
def manual_start():
|
||
if is_running: return jsonify({"status": "busy"}), 400
|
||
threading.Thread(target=execute_monitor_task).start()
|
||
return jsonify({"status": "started"})
|
||
|
||
|
||
@app.route('/api/status')
|
||
def status(): return jsonify({"is_running": is_running})
|
||
|
||
|
||
@app.route('/api/logs')
|
||
def logs():
|
||
data = MonitorRecord.query.all()
|
||
return jsonify([{
|
||
"source": l.source, "name": l.name, "status": l.status,
|
||
"reason": l.reason, "offset": l.offset, "latest_time": l.latest_time,
|
||
"check_time": l.check_time, "content": l.content
|
||
} for l in data])
|
||
|
||
|
||
# --- 调度器配置 ---
|
||
# id: 任务ID, func: 任务函数(字符串路径或引用), trigger: cron(定时), hour/minute: 时间
|
||
@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) |