消息通知

api/auth.py

@@ -2,6 +2,7 @@ import hmac
 import hashlib
 import time
 from typing import Optional
+from urllib.parse import urlparse
 
 import requests
 from flask import Blueprint, render_template, request, session, redirect, url_for, flash
@@ -33,6 +34,19 @@ def _generate_signature(secret: str, params: dict) -> str:
     ).hexdigest()
 
 
+def _is_safe_redirect_url(target: str) -> bool:
+    """检查 next 是否为安全重定向目标（相对路径或同域）"""
+    if not target or not target.strip():
+        return False
+    parsed = urlparse(target)
+    # 允许相对路径（无 netloc）
+    if not parsed.netloc:
+        return target.startswith('/')
+    # 允许同域
+    ref = urlparse(request.host_url)
+    return parsed.netloc == ref.netloc
+
+
 def _validate_ticket(ticket: str) -> Optional[dict]:
     """调用 UAP 验证票据"""
     payload = {
@@ -89,6 +103,9 @@ def sso_callback():
     if result and result.get('valid') is True:
         logger.info(f"[SSO] 验证通过，登录成功: user_id={result.get('user_id')}, mobile={result.get('mobile')}")
         session['logged_in'] = True
+        next_url = request.args.get('next')
+        if next_url and _is_safe_redirect_url(next_url):
+            return redirect(next_url)
         return redirect(url_for('main.index'))
     else:
         logger.warning(f"[SSO] 验证失败: result={result}")

api/main.py

@@ -2,6 +2,7 @@ from flask import Blueprint, render_template, send_from_directory, current_app,
 from api.utils import login_required, load_led_config
 from application.scheduler_service import scheduler_service
 from application.self_check_service import run_all_checks
+from application.uap_message_service import send_self_check_notification
 from utils.logger_config import logger
 
 main_bp = Blueprint('main', __name__)
@@ -61,9 +62,13 @@ def send_report_api():
         subject = "手动触发设备自检报告"
         scheduler_service.send_email(subject, html_report, receivers=target_email)
         
+        # 向统一登录平台运维人员推送通知
+        summary = scheduler_service._format_summary_text(results)
+        send_self_check_notification(summary)
+        
         return jsonify({
             "success": True,
-            "message": f"自检报告已发送至 {target_email if target_email else '默认邮箱'}"
+            "message": f"自检报告已发送至 {target_email if target_email else '默认邮箱'}，并已通知运维人员"
         })
     except Exception as e:
         logger.error(f"发送报告失败: {e}")

application/scheduler_service.py

@@ -6,6 +6,7 @@ import yaml
 import datetime
 from apscheduler.schedulers.background import BackgroundScheduler
 from application.self_check_service import run_all_checks
+from application.uap_message_service import send_self_check_notification
 from utils.logger_config import logger
 
 class SelfCheckScheduler:
@@ -121,6 +122,19 @@ class SelfCheckScheduler:
         """
         return html
 
+    def _format_summary_text(self, results):
+        """从检测结果生成简短摘要文本"""
+        total_devices = total_online = total_offline = 0
+        modules = {"kodi", "door", "led", "ha", "pc"}
+        for key in modules:
+            for device in results.get(key, []):
+                total_devices += 1
+                if device.get('is_online'):
+                    total_online += 1
+                else:
+                    total_offline += 1
+        return f"总设备数 {total_devices}，在线 {total_online}，离线/异常 {total_offline}"
+
     def send_email(self, subject, content, receivers=None):
         config = self.load_email_config()
         if not config:
@@ -173,6 +187,9 @@ class SelfCheckScheduler:
             results = run_all_checks()
             html_report = self.format_report_html(results)
             self.send_email("每日设备自检报告", html_report)
+            # 向统一登录平台运维人员推送通知
+            summary = self._format_summary_text(results)
+            send_self_check_notification(summary)
         except Exception as e:
             logger.error(f"每日自检任务异常: {e}")
 

application/uap_message_service.py

@@ -0,0 +1,77 @@
+"""
+统一登录平台 (UAP) 消息推送服务
+用于向运维人员发送设备自检等通知消息
+"""
+import hmac
+import hashlib
+import time
+from typing import List
+
+import requests
+from utils.logger_config import logger
+
+# 复用 auth 中的 SSO 配置
+from api.auth import SSO_APP_ID, SSO_APP_SECRET
+
+UAP_MESSAGES_URL = 'https://api.hnyunzhu.com/api/v1/messages/'
+
+# 运维人员账号列表（统一登录平台 app_user_id）
+OPS_USER_IDS: List[str] = ['liuq', 'longb']
+
+# 自检页面链接（生产环境需配置为实际域名）
+SELF_CHECK_BASE_URL = 'http://localhost:5050'
+
+
+def _generate_signature(secret: str, params: dict) -> str:
+    """生成 HMAC-SHA256 签名（与 auth 中规则一致）"""
+    data = {k: v for k, v in params.items() if k != "sign" and v is not None}
+    sorted_keys = sorted(data.keys())
+    query_string = "&".join([f"{k}={data[k]}" for k in sorted_keys])
+    return hmac.new(
+        secret.encode('utf-8'),
+        query_string.encode('utf-8'),
+        hashlib.sha256
+    ).hexdigest()
+
+
+def send_self_check_notification(content: str, target_url: str = None) -> None:
+    """
+    向运维人员发送自检通知消息
+    :param content: 消息内容（如：总设备数 X，在线 Y，离线 Z）
+    :param target_url: 点击跳转地址，默认 /self_check
+    """
+    if target_url is None:
+        target_url = f"{SELF_CHECK_BASE_URL}/self_check"
+
+    timestamp = int(time.time())
+    params = {"app_id": SSO_APP_ID, "timestamp": timestamp}
+    sign = _generate_signature(SSO_APP_SECRET, params)
+
+    headers = {
+        "Content-Type": "application/json",
+        "X-App-Id": SSO_APP_ID,
+        "X-Timestamp": str(timestamp),
+        "X-Sign": sign,
+    }
+
+    payload = {
+        "app_id": SSO_APP_ID,
+        "type": "NOTIFICATION",
+        "content_type": "TEXT",
+        "title": "韫珠展厅设备每日自检信息",
+        "content": content,
+        "auto_sso": True,
+        "target_url": target_url,
+        "action_text": "前往查看",
+    }
+
+    for app_user_id in OPS_USER_IDS:
+        try:
+            body = {**payload, "app_user_id": app_user_id}
+            resp = requests.post(UAP_MESSAGES_URL, json=body, headers=headers, timeout=10)
+            if resp.ok:
+                logger.info(f"[UAP] 自检通知已发送至 {app_user_id}")
+            else:
+                logger.warning(f"[UAP] 发送通知失败 {app_user_id}: status={resp.status_code}, body={resp.text}")
+        except Exception as e:
+            logger.error(f"[UAP] 发送通知异常 {app_user_id}: {e}")