241 lines
8.5 KiB
Python
241 lines
8.5 KiB
Python
import os
|
|
import json
|
|
import threading
|
|
import requests
|
|
from datetime import datetime
|
|
from flask import Flask, jsonify
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_cors import CORS
|
|
from lxml import etree
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
|
|
# --- 数据库配置 ---
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///monitor_data.db'
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
db = SQLAlchemy(app)
|
|
|
|
|
|
class ErrorLog(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
source = db.Column(db.String(50))
|
|
name = db.Column(db.String(100))
|
|
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 add_error_to_db(source, name, reason, latest_time="N/A", content=None):
|
|
days_diff = "N/A"
|
|
if latest_time and latest_time != "N/A":
|
|
try:
|
|
# 兼容 2024_05_20 和 2024-05-20 两种格式
|
|
clean_date_str = str(latest_time).split()[0].replace('_', '-')
|
|
target_date = datetime.strptime(clean_date_str, "%Y-%m-%d").date()
|
|
diff = (datetime.now().date() - target_date).days
|
|
days_diff = f"滞后 {diff} 天" if diff > 0 else "当天已同步"
|
|
except:
|
|
days_diff = "解析失败"
|
|
|
|
log = ErrorLog(
|
|
source=source, name=name, reason=reason, offset=days_diff,
|
|
latest_time=latest_time, content=content,
|
|
check_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
)
|
|
db.session.add(log)
|
|
|
|
|
|
# --- 106 专用辅助函数 ---
|
|
def get_106_dynamic_token(port):
|
|
login_url = f"http://106.75.72.40:{port}/api/login"
|
|
try:
|
|
resp = requests.post(login_url, json=CONFIG["106"]["login_payload"], timeout=10)
|
|
if resp.status_code == 200:
|
|
return resp.text.strip().replace('"', '')
|
|
return 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]
|
|
|
|
|
|
# --- 106 核心采集逻辑 ---
|
|
def run_106_logic():
|
|
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=15)
|
|
proxies = resp.json().get('proxies', [])
|
|
|
|
for item in proxies:
|
|
name = item.get('name', '')
|
|
# 严格过滤
|
|
if not name.lower().endswith('_data'): continue
|
|
name_upper = name.upper()
|
|
is_tower_underscore = "TOWER_" in name_upper
|
|
is_tower_i = "TOWER" in name_upper and not is_tower_underscore
|
|
if not (is_tower_underscore or is_tower_i): continue
|
|
|
|
# 状态检查
|
|
if str(item.get('status')).lower() != 'online':
|
|
add_error_to_db("106网站", name, f"离线({item.get('status')})")
|
|
continue
|
|
|
|
try:
|
|
port = item.get('conf', {}).get('remote_port')
|
|
token = get_106_dynamic_token(port)
|
|
if not token:
|
|
add_error_to_db("106网站", name, "Token获取失败")
|
|
continue
|
|
|
|
headers = {"Authorization": c["primary_auth"], "x-auth": token, "User-Agent": "Mozilla/5.0"}
|
|
api_root = "/api/resources/Data/" if is_tower_underscore else "/api/resources/data/"
|
|
|
|
# Step 1: 寻找日期目录
|
|
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', []), is_date_level=True)
|
|
|
|
if not best_date or best_date[2] != today_str:
|
|
add_error_to_db("106网站", name, "未找到今日文件夹", best_date[2] if best_date else "无")
|
|
continue
|
|
|
|
# Step 2: 寻找最新文件
|
|
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', []), is_date_level=False)
|
|
|
|
if not best_file:
|
|
add_error_to_db("106网站", name, "文件夹内无文件", today_str)
|
|
continue
|
|
|
|
file_item = best_file[1]
|
|
full_path = file_item.get('path') or f"{date_path}{file_item.get('name')}"
|
|
|
|
# Step 3: 获取内容
|
|
final_content = ""
|
|
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', '')
|
|
|
|
add_error_to_db("106网站", name, "同步成功", today_str, content=final_content)
|
|
|
|
except Exception as e:
|
|
add_error_to_db("106网站", name, f"采集异常: {str(e)}")
|
|
|
|
except Exception as e:
|
|
add_error_to_db("106网站", "全局错误", str(e))
|
|
|
|
|
|
# --- 82 逻辑 (保持不变) ---
|
|
def run_82_logic():
|
|
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"
|
|
add_error_to_db("82网站", sid, "同步成功", latest, content=json.dumps(data, ensure_ascii=False))
|
|
except:
|
|
add_error_to_db("82网站", sid, "采集异常")
|
|
except Exception as e:
|
|
add_error_to_db("82网站", "初始化错误", str(e))
|
|
|
|
|
|
# --- Flask 路由 ---
|
|
def background_task():
|
|
global is_running
|
|
with app.app_context():
|
|
ErrorLog.query.delete() # 每次开始前清理旧日志
|
|
run_106_logic()
|
|
run_82_logic()
|
|
db.session.commit()
|
|
is_running = False
|
|
|
|
|
|
@app.route('/api/run', methods=['POST'])
|
|
def start():
|
|
global is_running
|
|
if is_running: return jsonify({"status": "busy"}), 400
|
|
is_running = True
|
|
threading.Thread(target=background_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 = ErrorLog.query.all()
|
|
return jsonify([{
|
|
"source": l.source,
|
|
"name": l.name,
|
|
"reason": l.reason,
|
|
"offset": l.offset,
|
|
"latest_time": l.latest_time,
|
|
"check_time": l.check_time,
|
|
"content": l.content
|
|
} for l in data])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=True, port=5000) |