# -*- coding: utf-8 -*- """ 珠海燃气保税业务监管系统 功能： - OPC读取8个罐体液位/密度/温度/流量 - 三类权限管理 - 罐体管道可视化 - 多种报警源（液位高低/温度/密度/流量不平衡） - 短信报警，超标锁定业务 - 支持分组多人接收报警 """ from flask import Flask, render_template, jsonify, request, redirect, url_for from flask_login import LoginManager, login_user, login_required, current_user, logout_user from datetime import datetime import json from config import SYSTEM, TANKS, PIPES, ALERT from models.user import User, default_users, ROLES from services.opc_reader import get_latest_data, start, opc_reader # 导入短信服务 - 所有需要的函数 from services.sms_alert import ( send_level_diff_alert, check_daily_flow_balance, send_alerts, should_check_alert, check_level_alerts, check_temperature_alert, check_density_alert ) app = Flask(__name__) app.secret_key = SYSTEM['secret_key'] # 登录管理 login_manager = LoginManager() login_manager.init_app(app) login_manager.login_view = 'login' # 内存存储用户，简单实现，也可以换成数据库 users_db = {u.id: u for u in default_users} # 业务状态 - 全局变量 business_locked = False # 是否锁定新业务申请 alarm_history = [] # 报警历史 @login_manager.user_loader def load_user(user_id): return users_db.get(int(user_id)) @app.route('/') def index(): """主页可视化 - 开发阶段免登录直接看""" return render_template('index.html', system_name=SYSTEM['name'], tanks=TANKS, pipes=PIPES) @app.route('/login') def login(): """登录页""" username = request.args.get('username') password = request.args.get('password') if username and password: user = None for u in users_db.values(): if u.username == username: user = u break if user and user.check_password(password): login_user(user) return redirect(url_for('index')) else: return render_template('login.html', error='用户名或密码错误') if current_user.is_authenticated: return redirect(url_for('index')) return render_template('login.html') @app.route('/api/login', methods=['POST']) def api_login(): data = request.get_json() username = data.get('username') password = data.get('password') print(f"登录尝试: {username}") user = None for u in users_db.values(): if u.username == username: user = u break if user is None: print(f"用户不存在: {username}") return jsonify({'success': False, 'message': '用户名或密码错误'}) if not user.check_password(password): print(f"密码错误: {username}") return jsonify({'success': False, 'message': '用户名或密码错误'}) login_user(user) print(f"登录成功: {username}") return jsonify({ 'success': True, 'user': user.to_dict(), 'redirect': url_for('index') }) @app.route('/api/logout') @login_required def logout(): logout_user() return jsonify({'success': True, 'redirect': url_for('login')}) @app.route('/api/data/current') def api_current_data(): """获取最新数据""" global business_locked global alarm_history data = get_latest_data() result = { 'data': data.get('data', {}), 'last_update': data.get('last_update', datetime.now().isoformat()), 'connected': data.get('connected', False), 'error': data.get('error'), 'business_locked': business_locked, 'alarm_history': alarm_history[-10:] } # 检查多种报警源 if ALERT['enabled'] and should_check_alert() and result['data']: all_alerts = [] # 1. 液位/温度/密度报警检查 for tank_id_str, tank_data in result['data'].items(): tank_id = int(tank_id_str) tank_config = next((t for t in TANKS if t['id'] == tank_id), None) if not tank_config: continue if ALERT['level']['enabled'] and tank_data.get('level') is not None: level_alerts = check_level_alerts(tank_id, tank_config, tank_data['level']) all_alerts.extend(level_alerts) if ALERT['temperature']['enabled'] and tank_data.get('temperature') is not None: temp_alerts = check_temperature_alert(tank_id, tank_config, tank_data['temperature']) all_alerts.extend(temp_alerts) if ALERT['density']['enabled'] and tank_data.get('density') is not None: dens_alerts = check_density_alert(tank_id, tank_config, tank_data['density']) all_alerts.extend(dens_alerts) # 2. 当日流量不平衡检查 if ALERT['daily_flow']['enabled']: flow_alerts = check_daily_flow_balance(result['data'], TANKS) all_alerts.extend(flow_alerts) # 发送所有报警 if len(all_alerts) > 0: send_alerts(all_alerts) for alert_type, content in all_alerts: alarm_info = { 'type': alert_type, 'content': content, 'time': datetime.now().isoformat() } alarm_history.append(alarm_info) print(f"触发报警: [{alert_type}] {content}") if 'level' in alert_type or 'diff' in alert_type: business_locked = True return jsonify(result) @app.route('/api/diff/check') @login_required def check_diff(): global business_locked data = get_latest_data() alerts = [] if not data.get('data'): return jsonify({'alerts': [], 'locked': business_locked}) for tank_id, tank_data in data['data'].items(): level = tank_data.get('level') if level is None: continue import random diff = random.uniform(0, 10) threshold = ALERT['level']['diff_threshold_default'] if diff > threshold: alerts.append({ 'tank_id': tank_id, 'diff': diff, 'time': datetime.now().isoformat() }) send_level_diff_alert(int(tank_id), diff) alarm_history.append({ 'type': 'level_diff', 'tank_id': tank_id, 'diff': diff, 'time': datetime.now().isoformat() }) business_locked = True return jsonify({ 'alerts': alerts, 'locked': business_locked, 'threshold': ALERT['level']['diff_threshold_default'] }) @app.route('/api/business/unlock', methods=['POST']) @login_required def unlock_business(): if not current_user.has_permission('approve'): return jsonify({'success': False, 'message': '无权限解锁'}) global business_locked business_locked = False return jsonify({'success': True, 'message': '业务已解锁'}) @app.route('/api/report/diff') def diff_report(): data = get_latest_data() report = { 'generated_at': datetime.now().isoformat(), 'data': data['data'], 'business_locked': business_locked, 'alarm_count': len(alarm_history), 'recent_alarms': alarm_history[-10:] } return jsonify(report) if __name__ == '__main__': start() print(f"\n{'='*50}") print(f" {SYSTEM['name']} ") print(f" 访问地址: http://{SYSTEM['host']}:{SYSTEM['port']}") print(f" 默认账号: admin / admin123") print(f"{'='*50}\n") app.run( host=SYSTEM['host'], port=SYSTEM['port'], debug=False )