import json import re from datetime import datetime from flask import Blueprint, jsonify, request from sqlalchemy import desc, or_ # 引入 get_jwt_identity 用来获取当前用户ID from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity from extensions import db from models import Device, DeviceHistory, MaintenanceLog, User, UserDevicePermission # 尝试导入爬虫模块 try: from services.core import execute_monitor_task except ImportError: execute_monitor_task = None api_bp = Blueprint('api', __name__, url_prefix='/api') # ======================= # 🔧 辅助函数 # ======================= def is_admin(user_id): """ 判断用户权限 兼容 Token 中存储的 String 类型 ID """ # 1. 既然 Token 里是字符串，这里必须转成字符串比较，或者转成 int 比较 if str(user_id) == '0': return True if not user_id: return False try: # 2. 查库时需要 int uid = int(user_id) u = User.query.get(uid) return u and u.role == 'admin' except: return False 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 "时间解析失败" # ======================= # 0. 认证接口 # ======================= @api_bp.route('/login', methods=['POST']) def login(): data = request.get_json() username = data.get('username') password = data.get('password') # 1. 查库登录 user = User.query.filter_by(username=username).first() if user and user.check_password(password): # 🟢 修复核心：必须用 str() 将 ID 转为字符串 # 否则 flask-jwt-extended 会报错 "Subject must be a string" token = create_access_token( identity=str(user.id), additional_claims={'role': user.role} ) return jsonify({ 'code': 200, 'message': '登录成功', 'token': token, 'role': user.role, 'user_id': user.id }) # 2. 硬编码后门 (修正：生成真实 Token) if username == 'admin' and password == 'licahk': # 🟢 修复核心：'0' 必须是字符串 token = create_access_token( identity='0', additional_claims={'role': 'admin'} ) return jsonify({ 'code': 200, 'message': 'Root登录', 'token': token, 'role': 'admin', 'user_id': 0 }) return jsonify({'code': 401, 'message': '用户名或密码错误'}), 401 # ======================= # 1. 设备接口 # ======================= @api_bp.route('/devices_overview', methods=['GET']) @jwt_required() def devices_overview(): try: # 获取到的 user_id 现在是字符串类型 (例如 "1" 或 "0") user_id = get_jwt_identity() target_devices = [] if is_admin(user_id): target_devices = Device.query.all() else: # 普通用户权限查询 (users_db) # 必须转回 int 才能去数据库查 ID try: uid_int = int(user_id) user = User.query.get(uid_int) if user: perms = UserDevicePermission.query.filter_by(user_id=user.id).all() allowed_ids = [p.device_id for p in perms] if allowed_ids: target_devices = Device.query.filter(Device.id.in_(allowed_ids)).all() except ValueError: return jsonify({'code': 401, 'message': '无效的用户ID格式'}), 401 return jsonify({'code': 200, 'data': [d.to_dict() for d in target_devices]}) except Exception as e: print(f"Error in devices_overview: {e}") return jsonify({'code': 500, 'message': str(e)}) @api_bp.route('/device_data_by_date', methods=['GET']) @jwt_required(optional=True) 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 params'}), 400 device = Device.query.filter_by(name=name).first() if not device: return jsonify({'code': 404, 'message': 'Device not found'}), 404 content = None # 优先查历史 hist = DeviceHistory.query.filter( DeviceHistory.device_id == device.id, DeviceHistory.data_time.like(f"{date_str}%") ).order_by(desc(DeviceHistory.id)).first() if hist: content = hist.json_data elif device.latest_time and str(device.latest_time).startswith(date_str): content = device.json_data if content: # 尝试转JSON对象返回 try: if isinstance(content, str): content = json.loads(content) except: pass return jsonify({'code': 200, 'name': device.name, 'source': device.source, 'content': content}) return jsonify({'code': 404, 'message': '无数据'}), 404 # ======================= # 2. 用户管理 (Admin) # ======================= @api_bp.route('/admin/users', methods=['GET']) @jwt_required() def admin_get_users(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 clients = User.query.filter_by(role='client').all() result = [] for c in clients: perms = UserDevicePermission.query.filter_by(user_id=c.id).all() result.append({ "id": c.id, "username": c.username, "created_at": c.created_at, "allowed_device_ids": [p.device_id for p in perms] }) return jsonify(result) @api_bp.route('/admin/create_client', methods=['POST']) @jwt_required() def admin_create_client(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 data = request.get_json() if User.query.filter_by(username=data['username']).first(): return jsonify({'code': 400, 'msg': '用户已存在'}), 400 u = User(username=data['username'], role='client') u.set_password(data['password']) db.session.add(u) db.session.commit() return jsonify({'code': 200, 'msg': '创建成功'}) @api_bp.route('/admin/assign_devices', methods=['POST']) @jwt_required() def admin_assign_devices(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 data = request.get_json() uid = data.get('user_id') d_ids = data.get('device_ids', []) UserDevicePermission.query.filter_by(user_id=uid).delete() for did in d_ids: db.session.add(UserDevicePermission(user_id=uid, device_id=did)) db.session.commit() return jsonify({'code': 200, 'msg': '权限已保存'}) # ======================= # 3. 日志与工具 # ======================= @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: s = datetime.strptime(start_date, '%Y-%m-%d') e = datetime.strptime(end_date, '%Y-%m-%d').replace(hour=23, minute=59, second=59) query = query.filter(MaintenanceLog.timestamp.between(s, e)) except: pass logs = query.order_by(desc(MaintenanceLog.timestamp)).all() return jsonify({'code': 200, 'data': [l.to_dict() for l in logs]}) @api_bp.route('/logs/add', methods=['POST']) @jwt_required() def add_log(): data = request.get_json() db.session.add(MaintenanceLog( device_name=data.get('device_name'), engineer=data.get('engineer'), location=data.get('location'), content=data.get('content') )) db.session.commit() return jsonify({'code': 200}) @api_bp.route('/logs/delete', methods=['POST']) @jwt_required() def delete_log(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 log = MaintenanceLog.query.get(request.get_json().get('id')) if log: db.session.delete(log) db.session.commit() return jsonify({'code': 200}) return jsonify({'code': 404}) @api_bp.route('/run_monitor', methods=['POST']) @jwt_required() def run_monitor(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 if not execute_monitor_task: return jsonify({'code': 500, 'msg': '爬虫模块未加载'}) try: task_result = execute_monitor_task() if not task_result: return jsonify({'code': 200, 'msg': '跳过'}) scraped_list = task_result.get('device_list', []) now_str = 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', {}) target_time = item.get('target_time') source = item.get('source', '') 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) 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() device.status = item.get('status') device.current_value = item.get('value') device.latest_time = target_time device.check_time = now_str device.json_data = json_str device.offset = calculate_offset(target_time) db.session.add(DeviceHistory( device_id=device.id, status=device.status, result_data=device.current_value, data_time=target_time, json_data=json_str )) 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']) @jwt_required() def update_site(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 d = Device.query.filter_by(name=request.get_json().get('name')).first() if d: d.install_site = request.get_json().get('site') db.session.commit() return jsonify({'code': 200}) return jsonify({'code': 404}) @api_bp.route('/toggle_maintenance', methods=['POST']) @jwt_required() def toggle_maintenance(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 d = Device.query.filter_by(name=request.get_json().get('name')).first() if d: d.is_maintaining = request.get_json().get('is_maintaining') db.session.commit() return jsonify({'code': 200}) return jsonify({'code': 404}) @api_bp.route('/toggle_hidden', methods=['POST']) @jwt_required() def toggle_hidden(): if not is_admin(get_jwt_identity()): return jsonify({'code': 403}), 403 d = Device.query.filter_by(name=request.get_json().get('name')).first() if d: d.is_hidden = request.get_json().get('is_hidden') db.session.commit() return jsonify({'code': 200}) return jsonify({'code': 404})