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

登录系统以及超级管理员权限

Browse Source
This commit is contained in:
YueL1331
2026-01-09 09:47:27 +08:00
parent e67edec876
commit ca895af384
3 changed files with 155 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
2.1版本/app.py
View File

@ -1,41 +1,128 @@
# app.py
import os
import sys
import json
from datetime import datetime
from flask import Flask
from extensions import db, cors # ✅ 从 extensions 导入
from models import Device, DeviceHistory, MaintenanceLog # 导入模型以便 SQLAlchemy 识别
from flask_apscheduler import APScheduler # ✅ 新增：定时任务控制器
from extensions import db, cors
from models import Device, DeviceHistory, MaintenanceLog
from routes.api import api_bp
# 尝试导入爬虫核心逻辑，以便定时任务调用
try:
from services.core import execute_monitor_task
except ImportError:
execute_monitor_task = None
# 解决 Windows 下控制台输出乱码问题
sys.stdout.reconfigure(encoding='utf-8')
# 初始化调度器
scheduler = APScheduler()
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("❌ 错误: 找不到 services.core.execute_monitor_task 模块")
return
try:
# 1. 执行爬取任务
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")
# 2. 这里的逻辑需要和 api.py 保持高度一致（复用更新逻辑）
# 为了防止代码重复，建议以后将此逻辑封装在 services 里的一个独立函数中
from routes.api import calculate_offset
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)}")
def create_app():
app = Flask(__name__)
# 1. 配置路径
# 1. 配置路径与数据库
basedir = os.path.abspath(os.path.dirname(__file__))
instance_path = os.path.join(basedir, 'instance')
if not os.path.exists(instance_path):
os.makedirs(instance_path)
print(f"📁 检测到目录不存在，已自动创建: {instance_path}")
db_path = os.path.join(instance_path, 'devices.db')
# 配置
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JSON_AS_ASCII'] = False
# 2. 初始化插件 (使用 init_app 模式)
cors.init_app(app) # ✅
db.init_app(app) # ✅
# 定时任务配置
app.config['SCHEDULER_API_ENABLED'] = True
app.config['SCHEDULER_TIMEZONE'] = "Asia/Shanghai" # 设置时区
# 3. 注册蓝图
# 注意：api.py 里已经写了 url_prefix='/api'，这里不要再写，否则变 /api/api/...
# 2. 初始化插件
cors.init_app(app)
db.init_app(app)
# 3. 初始化并启动调度器
scheduler.init_app(app)
# 4. 注册定时任务：每天 10:00 运行
# 如果你想测试是否生效，可以暂时把 hour=10 改为每隔一分钟运行一次：trigger='interval', minutes=1
scheduler.add_job(
id='daily_crawl_task',
func=auto_monitor_job,
args=[app],
trigger='cron',
hour=10,
minute=0
)
scheduler.start()
# 5. 注册蓝图
app.register_blueprint(api_bp)
# 4. 初始化数据库表
# 6. 初始化数据库表
with app.app_context():
db.create_all()
@ -44,6 +131,7 @@ def create_app():
app = create_app()
@app.shell_context_processor
def make_shell_context():
return {
@ -53,6 +141,9 @@ def make_shell_context():
'MaintenanceLog': MaintenanceLog
}
if __name__ == '__main__':
print("🚀 服务正在启动: http://127.0.0.1:5000")
app.run(debug=True, host='0.0.0.0', port=5000)
print("⏰ 定时任务已就绪：每天 10:00 自动执行爬取")
# 注意：在生产环境中使用 debug=True 会导致调度器运行两次，建议生产环境设为 False
app.run(debug=False, host='0.0.0.0', port=5000)

26
2.1版本/models.py
View File

@ -1,4 +1,3 @@
# models.py
from datetime import datetime
import json
from extensions import db
@ -10,24 +9,23 @@ class Device(db.Model):
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)) # 系统检查时间
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)) # 滞后描述
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 返回
# 统一状态映射逻辑
api_status = 'offline' if self.status in ['离线', '异常', '已离线'] else 'online'
# 注意：列表接口不返回 json_data 以提高性能
return {
'id': self.id,
'name': self.name,
@ -37,7 +35,7 @@ class Device(db.Model):
'status_text': self.status,
'value': self.current_value,
'reason': self.reason,
'install_site': self.install_site,
'install_site': self.install_site or '',
'is_maintaining': self.is_maintaining,
'is_hidden': self.is_hidden,
'offset': self.offset
@ -61,10 +59,10 @@ 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) # 自动记录时间
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 {

78
2.1版本/routes/api.py
View File

@ -1,7 +1,7 @@
import os
import shutil
import json
import re # [新增] 引入正则模块用于解析路径
import re
from datetime import datetime
from flask import Blueprint, jsonify, request
from sqlalchemy import desc, or_
@ -17,6 +17,27 @@ except ImportError:
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. 设备概览与详情接口
# =======================
@ -34,7 +55,7 @@ def devices_overview():
@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') # 前端传来的格式通常是 YYYY-MM-DD
date_str = request.args.get('date')
if not name or not date_str:
return jsonify({'code': 400, 'message': 'Missing name or date'}), 400
@ -44,9 +65,6 @@ def device_data_by_date():
return jsonify({'code': 404, 'message': 'Device not found'}), 404
content = None
# 1. 查历史表
# 注意：如果数据是通过 run_monitor 经过处理保存的，data_time 应该是标准的 YYYY-MM-DD 格式
history_record = DeviceHistory.query.filter(
DeviceHistory.device_id == device.id,
DeviceHistory.data_time.like(f"{date_str}%")
@ -54,8 +72,6 @@ def device_data_by_date():
if history_record:
content = history_record.json_data
# 2. 查当前状态 (如果历史表没查到，且当前状态的时间也匹配)
elif device.latest_time and device.latest_time.startswith(date_str):
content = device.json_data
@ -66,7 +82,6 @@ def device_data_by_date():
'source': device.source,
'content': content
})
else:
return jsonify({'code': 404, 'message': 'No data for this date'}), 404
@ -81,7 +96,6 @@ def get_logs():
end_date = request.args.get('end_date')
query = MaintenanceLog.query
if keyword:
kw = f"%{keyword}%"
query = query.filter(or_(
@ -125,20 +139,14 @@ def add_log():
def update_log():
data = request.get_json()
log_id = data.get('id')
if not log_id:
return jsonify({'code': 400, 'message': 'Missing Log ID'}), 400
log = MaintenanceLog.query.get(log_id)
if not log:
return jsonify({'code': 404, 'message': 'Log not found'}), 404
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:
@ -158,15 +166,13 @@ def delete_log():
# =======================
# 3. 辅助与控制接口 (核心修改逻辑在 run_monitor)
# 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('_', '-')
if len(clean) < 8: return latest_time_str
target = datetime.strptime(clean, "%Y-%m-%d").date()
diff = (datetime.now().date() - target).days
return "当天已同步" if diff == 0 else f"滞后 {diff} 天"
@ -195,40 +201,30 @@ def run_monitor():
source = item.get('source', '')
target_time = item.get('target_time')
# ================= [核心修改部分 START] =================
# 针对 106 网站的数据，从 path 中解析标准时间
# 格式示例: "path": "/Data/2026_01_08/xiwuzhu_16_29_28.csv"
# 目标格式: "2026-01-08 16:29:28"
# 处理 106 路径时间
if '106' in str(source):
try:
path_str = d_raw.get('path', '')
# 正则匹配日期 (YYYY_MM_DD) 和 时间 (HH_MM_SS)
# 解释：/Data/(年_月_日)/任意字符_(时_分_秒).csv
match = re.search(r'/Data/(\d{4}_\d{2}_\d{2})/\w+_(\d{2}_\d{2}_\d{2})\.csv', path_str)
if match:
date_part = match.group(1).replace('_', '-') # 2026_01_08 -> 2026-01-08
time_part = match.group(2).replace('_', ':') # 16_29_28 -> 16:29:28
extracted_time = f"{date_part} {time_part}"
# 覆盖爬虫原本获取的可能不准确的时间
target_time = extracted_time
item['target_time'] = extracted_time
except Exception as parse_err:
print(f"Error parsing 106 path: {parse_err}")
# ================= [核心修改部分 END] =================
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)
# 只有新设备才初始化静态字段
device = Device(name=d_name, source=source, install_site="")
db.session.add(device)
db.session.flush() # 获取ID
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.latest_time = target_time
device.check_time = current_check_time
device.json_data = json_str
device.offset = calculate_offset(target_time)
@ -237,14 +233,14 @@ def run_monitor():
device_id=device.id,
status=item.get('status'),
result_data=item.get('value'),
data_time=target_time, # 存入标准时间，方便查询
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} 台设备'})
return jsonify({'code': 200, 'message': f'成功更新 {count} 台设备，资料已保留'})
except Exception as e:
db.session.rollback()
return jsonify({'code': 500, 'message': str(e)})