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

打包上传的2.0版本

Browse Source
This commit is contained in:
YueL1331
2026-01-09 12:48:50 +08:00
parent ca895af384
commit ffbd494b7b
14 changed files with 240 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
2_1banben/app.py Normal file
View File

@ -0,0 +1,202 @@
import os
import sys
import json
import logging
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:
# 数据库实例 (在根目录 extensions.py 中)
from extensions import db
# 数据模型 (在根目录 models.py 中)
from models import Device, DeviceHistory, MaintenanceLog
# 核心业务逻辑 (在 services/core.py 中)
from services.core import execute_monitor_task
# 路由蓝图 (在 routes/api.py 中)
try:
from routes.api import api_bp as device_bp
except ImportError:
from routes.api import device_bp
# 工具函数 (在 routes/api.py 中)
from routes.api import calculate_offset
except ImportError as e:
print(f"❌ 严重错误: 模块导入失败。请检查文件名和变量名。详细信息: {e}")
print(f"系统路径: {sys.path}")
sys.exit(1)
# ==============================================================================
# 2. 路径计算模块 (兼容 PyInstaller 打包)
# ==============================================================================
def get_base_path():
"""获取运行时基准路径，兼容开发环境和打包环境"""
if getattr(sys, 'frozen', False):
if hasattr(sys, '_MEIPASS'):
return sys._MEIPASS # --onefile 模式
else:
return os.path.dirname(os.path.abspath(sys.executable)) # --onedir 模式
else:
return os.path.abspath(os.path.dirname(__file__))
BASE_DIR = get_base_path()
STATIC_FOLDER = os.path.join(BASE_DIR, 'web_dist')
INSTANCE_FOLDER = os.path.join(BASE_DIR, 'instance')
DB_PATH = os.path.join(INSTANCE_FOLDER, 'devices.db')
# 修复 Windows 下注册表 MIME 类型缺失导致网页白屏的问题
mimetypes.add_type('application/javascript', '.js')
mimetypes.add_type('text/css', '.css')
print(f"🚀 运行环境: {'Packaged' if getattr(sys, 'frozen', False) else 'Dev'}")
print(f"📂 基准路径: {BASE_DIR}")
print(f"💾 数据库路径: {DB_PATH}")
# ==============================================================================
# 3. 定时任务逻辑
# ==============================================================================
def auto_monitor_job(app):
"""定时任务具体执行逻辑"""
with app.app_context():
print(f"⏰ [定时任务] 启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if not execute_monitor_task:
print("❌ 错误: 无法加载爬虫核心模块 (execute_monitor_task is Missing)")
return
try:
# 执行爬取
task_result = execute_monitor_task()
if not task_result:
print("⚠️ [定时任务] 爬虫未获取到数据")
return
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
db.session.commit()
print(f"✅ [定时任务] 成功更新 {count} 台设备状态")
except Exception as e:
db.session.rollback()
print(f"❌ [定时任务] 异常: {str(e)}")
# ==============================================================================
# 4. Flask 应用工厂
# ==============================================================================
def create_app():
# 🔴 关键修复：移除了 static_url_path=''
# 这样 Flask 就不会强制拦截所有根路径请求，让下面的 serve_static 有机会处理 /dashboard
app = Flask(__name__, static_folder=STATIC_FOLDER)
CORS(app)
# 确保 instance 目录存在
if not os.path.exists(INSTANCE_FOLDER):
os.makedirs(INSTANCE_FOLDER, exist_ok=True)
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DB_PATH}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SCHEDULER_API_ENABLED'] = True
# 初始化数据库
db.init_app(app)
# 初始化定时任务
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
# 添加定时任务 (每天 10:00)
scheduler.add_job(
id='daily_monitor_task',
func=auto_monitor_job,
args=[app],
trigger='cron',
hour=10,
minute=0
)
# 注册蓝图
app.register_blueprint(device_bp)
# -------------------------------------------------
# 前端路由支持 (Vue History Mode)
# -------------------------------------------------
@app.route('/')
def serve_index():
if not os.path.exists(os.path.join(app.static_folder, 'index.html')):
return "❌ 错误: 前端文件丢失 (web_dist/index.html)", 404
return send_from_directory(app.static_folder, 'index.html')
@app.route('/<path:path>')
def serve_static(path):
# 1. 优先尝试直接返回实际存在的文件 (js, css, img等)
file_path = os.path.join(app.static_folder, path)
if os.path.exists(file_path):
return send_from_directory(app.static_folder, path)
# 2. 如果是 API 请求但没找到对应接口，返回 404 JSON (不返回 HTML)
if path.startswith('api') or path.startswith('static'):
return jsonify({'code': 404, 'message': 'Not Found'}), 404
# 3. 关键逻辑：
# 访问 /dashboard 等前端路由时，文件系统中并没有 dashboard 这个文件
# 所以会走到这里，返回 index.html，让 Vue 及其 Router 接管页面渲染
return send_from_directory(app.static_folder, 'index.html')
with app.app_context():
db.create_all()
return app
if __name__ == '__main__':
app = create_app()
# 生产环境/打包环境通常设为 False
debug_mode = not getattr(sys, 'frozen', False)
print("🚀 服务启动中...")
app.run(host='0.0.0.0', port=5000, debug=debug_mode, use_reloader=False)

42
2_1banben/config.py Normal file
View File

@ -0,0 +1,42 @@
import os
import sys
def get_base_path():
"""获取运行时路径 (兼容打包后的 exe 和开发环境)"""
if getattr(sys, 'frozen', False):
return os.path.dirname(sys.executable)
return os.path.dirname(os.path.abspath(__file__))
def get_static_path():
"""获取 dist 静态资源路径"""
if getattr(sys, 'frozen', False):
return os.path.join(sys._MEIPASS, 'dist')
return os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dist')
class Config:
BASE_DIR = get_base_path()
# 数据库路径：保存在运行目录下，文件名为 monitor_data.db
# Windows 下路径需要注意转义，这里使用 os.path.join 最安全
SQLALCHEMY_DATABASE_URI = f'sqlite:///{os.path.join(BASE_DIR, "monitor_data.db")}'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# --- 定时任务配置 ---
SCHEDULER_API_ENABLED = True
SCHEDULER_TIMEZONE = "Asia/Shanghai" # 👈 必须加这个，否则 APScheduler 可能报错
# --- 爬虫配置 (Service层会读取这里) ---
CRAWLER_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'}
}
}

9
2_1banben/extensions.py Normal file
View File

@ -0,0 +1,9 @@
#extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from flask_apscheduler import APScheduler
# 这里只创建对象，不绑定 app
db = SQLAlchemy()
cors = CORS()
scheduler = APScheduler()

75
2_1banben/models.py Normal file
View File

@ -0,0 +1,75 @@
from datetime import datetime
import json
from extensions import db
class Device(db.Model):
__tablename__ = 'devices'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), unique=True, index=True)
source = db.Column(db.String(50))
# 快照字段（爬虫更新）
status = db.Column(db.String(50))
current_value = db.Column(db.String(200))
latest_time = db.Column(db.String(50))
json_data = db.Column(db.Text)
check_time = db.Column(db.String(50))
reason = db.Column(db.String(255))
offset = db.Column(db.String(50))
# 手动录入字段（受保护，run_monitor 不主动覆盖）
install_site = db.Column(db.String(100), default="")
is_maintaining = db.Column(db.Boolean, default=False)
is_hidden = db.Column(db.Boolean, default=False)
def to_dict(self):
# 统一状态映射逻辑
api_status = 'offline' if self.status in ['离线', '异常', '已离线'] else 'online'
return {
'id': self.id,
'name': self.name,
'source': self.source,
'latest_time': self.latest_time,
'status': api_status,
'status_text': self.status,
'value': self.current_value,
'reason': self.reason,
'install_site': self.install_site or '',
'is_maintaining': self.is_maintaining,
'is_hidden': self.is_hidden,
'offset': self.offset
}
class DeviceHistory(db.Model):
__tablename__ = 'device_history'
id = db.Column(db.Integer, primary_key=True)
device_id = db.Column(db.Integer, db.ForeignKey('devices.id'))
data_time = db.Column(db.String(50))
status = db.Column(db.String(50))
result_data = db.Column(db.String(200), default="")
json_data = db.Column(db.Text)
file_path = db.Column(db.String(255))
recorded_at = db.Column(db.DateTime, default=datetime.now)
class MaintenanceLog(db.Model):
__tablename__ = 'maintenance_logs'
id = db.Column(db.Integer, primary_key=True)
device_name = db.Column(db.String(100), nullable=False)
engineer = db.Column(db.String(50))
location = db.Column(db.String(100))
content = db.Column(db.Text)
timestamp = db.Column(db.DateTime, default=datetime.now)
def to_dict(self):
return {
'id': self.id,
'device_name': self.device_name,
'engineer': self.engineer or '',
'location': self.location or '',
'content': self.content,
'timestamp': self.timestamp.strftime('%Y-%m-%d %H:%M:%S')
}

2
2_1banben/routes/__init__.py Normal file
View File

@ -0,0 +1,2 @@
# routes/__init__.py
# 这是一个空文件，用于将 routes 文件夹标识为 Python 包。

279
2_1banben/routes/api.py Normal file
View File

@ -0,0 +1,279 @@
import os
import shutil
import json
import re
from datetime import datetime
from flask import Blueprint, jsonify, request
from sqlalchemy import desc, or_
from extensions import db
from models import Device, DeviceHistory, MaintenanceLog
# 尝试导入爬虫模块
try:
from services.core import execute_monitor_task
except ImportError:
execute_monitor_task = None
api_bp = Blueprint('api', __name__, url_prefix='/api')
# =======================
# 0. 认证接口
# =======================
@api_bp.route('/login', methods=['POST'])
def login():
data = request.get_json()
username = data.get('username')
password = data.get('password')
if username == 'admin' and password == 'licahk':
return jsonify({
'code': 200,
'message': '登录成功',
'token': 'super-admin-token-2026',
'user': {'username': 'admin', 'role': 'administrator'}
})
return jsonify({'code': 401, 'message': '用户名或密码错误'}), 401
# =======================
# 1. 设备概览与详情接口
# =======================
@api_bp.route('/devices_overview', methods=['GET'])
def devices_overview():
try:
devices = Device.query.all()
data_list = [d.to_dict() for d in devices]
return jsonify({'code': 200, 'data': data_list})
except Exception as e:
return jsonify({'code': 500, 'message': str(e)})
@api_bp.route('/device_data_by_date', methods=['GET'])
def device_data_by_date():
name = request.args.get('name')
date_str = request.args.get('date')
if not name or not date_str:
return jsonify({'code': 400, 'message': 'Missing name or date'}), 400
device = Device.query.filter_by(name=name).first()
if not device:
return jsonify({'code': 404, 'message': 'Device not found'}), 404
content = None
history_record = DeviceHistory.query.filter(
DeviceHistory.device_id == device.id,
DeviceHistory.data_time.like(f"{date_str}%")
).order_by(desc(DeviceHistory.id)).first()
if history_record:
content = history_record.json_data
elif device.latest_time and device.latest_time.startswith(date_str):
content = device.json_data
if content:
return jsonify({
'code': 200,
'name': device.name,
'source': device.source,
'content': content
})
return jsonify({'code': 404, 'message': 'No data for this date'}), 404
# =======================
# 2. 维修日志接口
# =======================
@api_bp.route('/logs/list', methods=['GET'])
def get_logs():
keyword = request.args.get('keyword', '')
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
query = MaintenanceLog.query
if keyword:
kw = f"%{keyword}%"
query = query.filter(or_(
MaintenanceLog.device_name.like(kw),
MaintenanceLog.engineer.like(kw),
MaintenanceLog.location.like(kw),
MaintenanceLog.content.like(kw)
))
if start_date and end_date:
try:
start_dt = datetime.strptime(start_date, '%Y-%m-%d')
end_dt = datetime.strptime(end_date, '%Y-%m-%d').replace(hour=23, minute=59, second=59)
query = query.filter(MaintenanceLog.timestamp.between(start_dt, end_dt))
except ValueError:
pass
logs = query.order_by(MaintenanceLog.timestamp.desc()).all()
return jsonify({'code': 200, 'data': [l.to_dict() for l in logs]})
@api_bp.route('/logs/add', methods=['POST'])
def add_log():
data = request.get_json()
try:
new_log = MaintenanceLog(
device_name=data.get('device_name', '未知设备'),
engineer=data.get('engineer', ''),
location=data.get('location', ''),
content=data.get('content', '')
)
db.session.add(new_log)
db.session.commit()
return jsonify({'code': 200, 'message': 'Log saved'})
except Exception as e:
db.session.rollback()
return jsonify({'code': 500, 'message': str(e)})
@api_bp.route('/logs/update', methods=['POST'])
def update_log():
data = request.get_json()
log_id = data.get('id')
log = MaintenanceLog.query.get(log_id)
if not log: return jsonify({'code': 404, 'message': 'Not found'}), 404
try:
log.device_name = data.get('device_name', log.device_name)
log.engineer = data.get('engineer', log.engineer)
log.location = data.get('location', log.location)
log.content = data.get('content', log.content)
db.session.commit()
return jsonify({'code': 200, 'message': 'Log updated'})
except Exception as e:
db.session.rollback()
return jsonify({'code': 500, 'message': str(e)})
@api_bp.route('/logs/delete', methods=['POST'])
def delete_log():
data = request.get_json()
log = MaintenanceLog.query.get(data.get('id'))
if log:
db.session.delete(log)
db.session.commit()
return jsonify({'code': 200, 'message': 'Deleted'})
return jsonify({'code': 404, 'message': 'Not found'}), 404
# =======================
# 3. 辅助与控制接口 (核心修复逻辑)
# =======================
def calculate_offset(latest_time_str):
if not latest_time_str or latest_time_str == "N/A": return "从未同步"
try:
clean = str(latest_time_str).split()[0].replace('_', '-')
target = datetime.strptime(clean, "%Y-%m-%d").date()
diff = (datetime.now().date() - target).days
return "当天已同步" if diff == 0 else f"滞后 {diff} 天"
except:
return "时间解析失败"
@api_bp.route('/run_monitor', methods=['POST'])
def run_monitor():
try:
if not execute_monitor_task:
return jsonify({'code': 500, 'message': 'Core module missing'})
task_result = execute_monitor_task()
if not task_result: return jsonify({'code': 200, 'message': '任务跳过'})
scraped_list = task_result.get('device_list', [])
current_check_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
d_raw = item.get('raw_json', {})
source = item.get('source', '')
target_time = item.get('target_time')
# 处理 106 路径时间
if '106' in str(source):
try:
path_str = d_raw.get('path', '')
match = re.search(r'/Data/(\d{4}_\d{2}_\d{2})/\w+_(\d{2}_\d{2}_\d{2})\.csv', path_str)
if match:
target_time = f"{match.group(1).replace('_', '-')} {match.group(2).replace('_', ':')}"
except:
pass
json_str = json.dumps(d_raw, ensure_ascii=False) if isinstance(d_raw, (dict, list)) else str(d_raw)
# --- 关键修改：先查询，后更新 ---
device = Device.query.filter_by(name=d_name).first()
if not device:
# 只有新设备才初始化静态字段
device = Device(name=d_name, source=source, install_site="")
db.session.add(device)
db.session.flush() # 获取 ID 供 History 使用
# 仅更新动态抓取的字段，保留手动填写的 install_site, is_maintaining, is_hidden
device.status = item.get('status')
device.current_value = item.get('value')
device.latest_time = target_time
device.check_time = current_check_time
device.json_data = json_str
device.offset = calculate_offset(target_time)
new_history = DeviceHistory(
device_id=device.id,
status=item.get('status'),
result_data=item.get('value'),
data_time=target_time,
json_data=json_str
)
db.session.add(new_history)
count += 1
db.session.commit()
return jsonify({'code': 200, 'message': f'成功更新 {count} 台设备，资料已保留'})
except Exception as e:
db.session.rollback()
return jsonify({'code': 500, 'message': str(e)})
@api_bp.route('/update_site', methods=['POST'])
def update_site():
data = request.get_json()
device = Device.query.filter_by(name=data.get('name')).first()
if device:
device.install_site = data.get('site')
db.session.commit()
return jsonify({'code': 200})
return jsonify({'code': 404}), 404
@api_bp.route('/toggle_maintenance', methods=['POST'])
def toggle_maintenance():
data = request.get_json()
device = Device.query.filter_by(name=data.get('name')).first()
if device:
device.is_maintaining = data.get('is_maintaining')
db.session.commit()
return jsonify({'code': 200})
return jsonify({'code': 404}), 404
@api_bp.route('/toggle_hidden', methods=['POST'])
def toggle_hidden():
data = request.get_json()
device = Device.query.filter_by(name=data.get('name')).first()
if device:
device.is_hidden = data.get('is_hidden')
db.session.commit()
return jsonify({'code': 200})
return jsonify({'code': 404}), 404

27
2_1banben/routes/web.py Normal file
View File

@ -0,0 +1,27 @@
import os
from flask import Blueprint, send_from_directory
# 👇 确保 config.py 在根目录，且能被引用
from config import get_static_path
web_bp = Blueprint('web', __name__)
@web_bp.route('/')
def index():
"""访问根路径时，返回 dist/index.html"""
try:
return send_from_directory(get_static_path(), 'index.html')
except Exception as e:
return f"前端资源未找到，请确认 dist 文件夹是否存在。错误信息: {e}", 404
@web_bp.route('/<path:path>')
def static_files(path):
"""访问 /css, /js 等静态资源"""
static_folder = get_static_path()
file_path = os.path.join(static_folder, path)
if os.path.exists(file_path):
return send_from_directory(static_folder, path)
# 路由回退：解决 Vue History 模式刷新 404 问题
# 如果找不到文件，就返回 index.html，让 Vue 路由去处理
return send_from_directory(static_folder, 'index.html')

2
2_1banben/services/__init__.py Normal file
View File

@ -0,0 +1,2 @@
# services/__init__.py
# 这是一个空文件，用于将 services 文件夹标识为 Python 包。

37
2_1banben/services/core.py Normal file
View File

@ -0,0 +1,37 @@
# services/core.py
import logging
import threading
from .crawler_106 import run_106_logic
from .crawler_82 import run_82_logic
task_lock = threading.Lock()
def execute_monitor_task():
"""
执行所有爬虫，返回一个大列表：
{'device_list': [item1, item2...], 'target_time': '...'}
"""
if task_lock.locked():
logging.warning(">>> 任务正在运行中，跳过")
return None
with task_lock:
logging.info(">>> 开始执行监控任务...")
# 1. 获取 106 数据列表
list_106 = run_106_logic()
# 2. 获取 82 数据列表
list_82 = run_82_logic()
# 3. 合并
combined_list = list_106 + list_82
logging.info(f">>> 任务完成，共获取 {len(combined_list)} 条数据")
return {
'device_list': combined_list,
'target_time': None, # 具体时间已在 item 里
'temp_file_path': None # 废弃旧逻辑，文件路径已在 item 里
}

159
2_1banben/services/crawler_106.py Normal file
View File

@ -0,0 +1,159 @@
# services/crawler_106.py
import os
import requests
import logging
from datetime import datetime
from config import Config
CONFIG = Config.CRAWLER_CONFIG["106"]
def get_temp_dir():
base_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
temp_dir = os.path.join(base_dir, 'instance', 'temp')
if not os.path.exists(temp_dir):
os.makedirs(temp_dir)
return temp_dir
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["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():
"""返回 result_list, 每个元素是一个字典"""
results = []
print(">>> [106爬虫] 启动...")
today_str = datetime.now().strftime("%Y_%m_%d")
main_headers = {"Authorization": CONFIG["primary_auth"], "User-Agent": "Mozilla/5.0"}
try:
resp = requests.get(CONFIG["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
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
# 构建基础数据包
data_packet = {
'source': '106网站',
'name': name,
'status': '正常',
'value': '',
'target_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'raw_json': {},
'temp_file': None
}
if str(item.get('status')).lower() != 'online':
data_packet['status'] = '离线'
data_packet['value'] = f"状态: {item.get('status')}"
results.append(data_packet)
continue
try:
port = item.get('conf', {}).get('remote_port')
token = get_106_dynamic_token(port)
if not token:
data_packet['status'] = '异常'
data_packet['value'] = "Token获取失败"
results.append(data_packet)
continue
headers = {"Authorization": CONFIG["primary_auth"], "x-auth": token}
api_root = "/api/resources/Data/" if is_tower_underscore 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:
data_packet['value'] = "未找到今日文件夹"
data_packet['target_time'] = best_date[2] if best_date else "N/A"
results.append(data_packet)
continue
data_packet['target_time'] = best_date[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', []), False)
if not best_file:
data_packet['value'] = "今日文件夹为空"
results.append(data_packet)
continue
file_item = best_file[1]
full_path = file_item.get('path') or f"{date_path}{file_item.get('name')}"
# 核心逻辑：获取内容
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, stream=True)
if res3.status_code == 200:
safe_name = f"{name}_{datetime.now().strftime('%H%M%S')}.db"
temp_path = os.path.join(get_temp_dir(), safe_name)
with open(temp_path, 'wb') as f:
f.write(res3.content)
data_packet['temp_file'] = temp_path # 🔥 传递给API
data_packet['value'] = f"Binary Downloaded: {len(res3.content)} bytes"
data_packet['raw_json'] = file_item # 用文件属性充当RawData
else:
data_packet['status'] = '异常'
data_packet['value'] = f"下载失败: {res3.status_code}"
else:
# JSON 内容
file_api_url = f"http://106.75.72.40:{port}/api/resources{full_path}"
res3 = requests.get(file_api_url, headers=headers, timeout=20)
try:
json_content = res3.json()
data_packet['raw_json'] = json_content # 🔥 完整保存
data_packet['value'] = json_content.get('content', '')
except:
data_packet['value'] = "JSON解析失败"
results.append(data_packet)
except Exception as e:
data_packet['status'] = '异常'
data_packet['value'] = str(e)[:50]
results.append(data_packet)
except Exception as e:
logging.error(f"106 Crawler Error: {e}")
return results

62
2_1banben/services/crawler_82.py Normal file
View File

@ -0,0 +1,62 @@
# services/crawler_82.py
import requests
import json
import logging
from lxml import etree
from config import Config
from datetime import datetime
CONFIG = Config.CRAWLER_CONFIG["82"]
def run_82_logic():
"""返回 result_list"""
results = []
print(">>> [82爬虫] 启动...")
session = requests.Session()
try:
session.post(f"{CONFIG['base_url']}/login.php", data=CONFIG["login"], timeout=10)
resp = session.post(f"{CONFIG['base_url']}/GetStationList.php", timeout=10)
html = etree.HTML(resp.content)
if html is None: return []
stations = html.xpath('//option/@value')
for sid in [s for s in stations if s]:
data_packet = {
'source': '82网站',
'name': str(sid),
'status': '正常',
'value': '',
'target_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'raw_json': {},
'temp_file': None
}
try:
r = session.post(f"{CONFIG['base_url']}/getLastWeatherData.php", data=str(sid),
headers={'Content-Type': 'text/plain'}, timeout=10)
try:
data = r.json()
except:
data = None
if data:
d_list = data.get('date', [])
latest = str(d_list[-1]) if d_list else "N/A"
data_packet['target_time'] = latest
data_packet['value'] = f"Data Points: {len(d_list)}"
data_packet['raw_json'] = data # 🔥 存完整JSON
else:
data_packet['status'] = '异常'
data_packet['value'] = "返回空数据"
except Exception as e:
data_packet['status'] = '异常'
data_packet['value'] = "单个采集失败"
results.append(data_packet)
except Exception as e:
logging.error(f"82 Crawler Error: {e}")
return results