Update pages/monitor.py

pages/monitor.py

@@ -4,7 +4,6 @@ import time
 import json
 import os
 import threading
-import queue
 import re
 from datetime import datetime, timedelta, timezone, time as dt_time
 from collections import deque
@@ -22,12 +21,20 @@ except ImportError:
 # --- 1. 配置与常量 ---
 load_dotenv()
 
+# Pushover 配置
 PUSHOVER_TOKEN = os.getenv("PUSHOVER_TOKEN")
 PUSHOVER_USER = os.getenv("PUSHOVER_USER")
-CINEMA_ID = os.getenv("CINEMA_ID")
 
-# 新增：NTFY 配置
-NTFY_TOPIC = os.getenv("ntfy_code")  # 获取 NTFY 主题名
+# 微信配置
+WX_APP_ID = os.getenv("WX_APP_ID")
+WX_APP_SECRET = os.getenv("WX_APP_SECRET")
+# 支持多个用户，逗号分隔
+WX_USER_OPEN_IDS = os.getenv("WX_USER_OPEN_ID", "").split(',')
+WX_TEMPLATE_ID_TICKET = os.getenv("WX_TEMPLATE_ID_TICKET")
+WX_TEMPLATE_ID_DAILY = os.getenv("WX_TEMPLATE_ID_DAILY")
+
+# 影城配置
+CINEMA_ID = os.getenv("CINEMA_ID")
 
 # 获取 Token 文件路径
 CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -77,34 +84,77 @@ def simplify_hall_name(raw_name):
     return raw_name
 
 
-def send_pushover(title, message, priority=0):
-    """发送 Pushover 通知 (返回是否成功)"""
-    if not PUSHOVER_TOKEN or not PUSHOVER_USER:
-        return False
-
-    url = "https://api.pushover.net/1/messages.json"
-    data = {
-        "token": PUSHOVER_TOKEN,
-        "user": PUSHOVER_USER,
-        "title": title,
-        "message": message,
-        "sound": "pushover",
-        "priority": priority
-    }
-    try:
-        response = requests.post(url, data=data, timeout=10)
-        if response.status_code != 200:
-            print(f"❌ Pushover 拒绝: {response.status_code}")
+# --- 3. 微信推送模块 ---
+class WeChatPusher:
+    def __init__(self, app_id, app_secret):
+        self.app_id = app_id
+        self.app_secret = app_secret
+        self.token = None
+        self.token_expires_time = 0
+
+    def get_access_token(self):
+        """获取并缓存微信 Access Token"""
+        if self.token and datetime.now().timestamp() < self.token_expires_time:
+            return self.token
+
+        if not self.app_id or not self.app_secret:
+            return None
+
+        url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.app_id}&secret={self.app_secret}"
+        try:
+            resp = requests.get(url, timeout=10)
+            data = resp.json()
+            if 'access_token' in data:
+                self.token = data['access_token']
+                # 提前 200 秒过期，防止临界点问题
+                self.token_expires_time = datetime.now().timestamp() + data['expires_in'] - 200
+                return self.token
+            else:
+                print(f"❌ 微信 Token 获取失败: {data}")
+        except Exception as e:
+            print(f"❌ 微信 API 请求异常: {e}")
+        return None
+
+    def send_template_message(self, user_ids, template_id, data_map):
+        """发送模板消息给多个用户"""
+        token = self.get_access_token()
+        if not token:
             return False
-        else:
-            print(f"✅ Pushover 发送成功: {title}")
-            return True
-    except Exception as e:
-        print(f"❌ Pushover 网络异常: {e}")
-        return False
+
+        url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={token}"
+        
+        # 组装微信数据格式
+        formatted_data = {}
+        for key, value in data_map.items():
+            # 默认深蓝色
+            formatted_data[key] = {"value": value, "color": "#173177"}
+
+        # 清洗 user_ids
+        valid_ids = [uid.strip() for uid in user_ids if uid and uid.strip()]
+        success_flag = True
+
+        for user_id in valid_ids:
+            payload = {
+                "touser": user_id,
+                "template_id": template_id,
+                "data": formatted_data
+            }
+            try:
+                resp = requests.post(url, json=payload, timeout=10)
+                result = resp.json()
+                if result.get("errcode") != 0:
+                    print(f"❌ 微信发送失败 [{user_id}]: {result}")
+                    success_flag = False
+                else:
+                    print(f"✅ 微信发送成功 [{user_id}]")
+            except Exception as e:
+                print(f"❌ 微信发送异常 [{user_id}]: {e}")
+                success_flag = False
+        
+        return success_flag
 
 
-# --- 3. API 管理模块 ---
+# --- 4. API 管理模块 ---
 class APIManager:
     def __init__(self, logger):
         self.last_login_fail_time = 0
@@ -192,7 +242,7 @@ class APIManager:
             return None
 
 
-# --- 4. 监控服务主逻辑 ---
+# --- 5. 监控服务主逻辑 ---
 class CinemaMonitor:
     def __init__(self):
         self.logs = deque(maxlen=50)
@@ -201,6 +251,8 @@ class CinemaMonitor:
         self.monitored_sessions = []
 
         self.api = APIManager(self.log)
+        self.wx_pusher = WeChatPusher(WX_APP_ID, WX_APP_SECRET)
+        
         self.current_business_date = None
         self.daily_schedule_cache = None
 
@@ -216,12 +268,6 @@ class CinemaMonitor:
             "notify_fails": 0  # 发送通知失败次数
         }
 
-        # --- NTFY 队列初始化 ---
-        # 队列项结构: (title, message, priority)
-        self.ntfy_queue = queue.Queue()
-        self.ntfy_thread = threading.Thread(target=self._ntfy_worker, daemon=True)
-        self.ntfy_thread.start()
-
         self.thread = threading.Thread(target=self._run_loop, daemon=True)
         self.thread.start()
 
@@ -231,87 +277,97 @@ class CinemaMonitor:
         print(entry)
         self.logs.appendleft(entry)
 
-    def _ntfy_worker(self):
-        """
-        NTFY 后台消费者线程
-        负责从队列取出消息发送，避免阻塞主线程，并提供基本的缓冲
-        """
-        self.log("📡 NTFY 后台通知服务已启动")
-        while True:
-            try:
-                # 阻塞获取任务 (title, msg, priority)
-                item = self.ntfy_queue.get()
-                title, message, priority = item
-
-                if not NTFY_TOPIC:
-                    # 如果未配置，直接跳过
-                    self.ntfy_queue.task_done()
-                    continue
-
-                try:
-                    url = f"https://ntfy.sh/{NTFY_TOPIC}"
-
-                    # 映射优先级:
-                    # Pushover: 0(Normal) -> NTFY: 3(Default)
-                    # Pushover: 1(High) -> NTFY: 4(High) or 5(Urgent)
-                    ntfy_priority = "default"
-                    tags = []
-
-                    if priority == 1:
-                        ntfy_priority = "high"
-                        # tags = ["warning", "ticket"]
-                    else:
-                        ntfy_priority = "default"
-                        # tags = ["white_check_mark"]
-
-                    headers = {
-                        "Title": title.encode('utf-8'),
-                        "Priority": ntfy_priority,
-                        "Tags": ",".join(tags)
-                    }
-
-                    # 发送请求
-                    resp = requests.post(
-                        url,
-                        data=message.encode('utf-8'),
-                        headers=headers,
-                        timeout=10
-                    )
-
-                    if resp.status_code == 200:
-                        print(f"✅ NTFY 发送成功")
-                    else:
-                        print(f"❌ NTFY 发送失败: {resp.status_code} - {resp.text}")
-                        self.stats["notify_fails"] += 1
-
-                except Exception as e:
-                    print(f"❌ NTFY 网络异常: {e}")
-                    self.stats["notify_fails"] += 1
-
-                self.ntfy_queue.task_done()
+    def _send_pushover(self, title, message, priority=0):
+        """发送 Pushover 通知"""
+        if not PUSHOVER_TOKEN or not PUSHOVER_USER:
+            return False
 
-                # 稍微休眠一下，避免极高并发冲击
-                time.sleep(2)
+        url = "https://api.pushover.net/1/messages.json"
+        data = {
+            "token": PUSHOVER_TOKEN,
+            "user": PUSHOVER_USER,
+            "title": title,
+            "message": message,
+            "sound": "pushover",
+            "priority": priority
+        }
+        try:
+            response = requests.post(url, data=data, timeout=10)
+            if response.status_code == 200:
+                return True
+            else:
+                print(f"❌ Pushover 拒绝: {response.status_code}")
+                return False
+        except Exception as e:
+            print(f"❌ Pushover 网络异常: {e}")
+            return False
 
-            except Exception as e:
-                print(f"NTFY 线程错误: {e}")
-                time.sleep(5)
+    def send_ticket_alert(self, hall_name, movie_name, show_time, ticket_count):
+        """
+        发送突发购票告警 (Pushover + WeChat)
+        """
+        # 1. Pushover
+        title = f"🎟 发现 {hall_name} {show_time} 场次有票！"
+        msg = f"{hall_name} {show_time}《{movie_name}》\n⚠️ 突增至 {ticket_count} 张"
+        
+        if not self._send_pushover(title, msg, priority=1):
+            self.stats["notify_fails"] += 1
 
-    def send_notifications(self, title, message, priority=0):
+        # 2. WeChat
+        if WX_TEMPLATE_ID_TICKET:
+            data = {
+                "first": f"发现 {hall_name} {show_time} 场次有票！\n{hall_name} {show_time}《{movie_name}》\n⚠️ 突增至 {ticket_count} 张\n",
+                "keyword1": hall_name,
+                "keyword2": movie_name,
+                "keyword3": show_time,
+                "keyword4": f"{ticket_count} 张",
+                "remark": "\n请尽快处理。"
+            }
+            if not self.wx_pusher.send_template_message(WX_USER_OPEN_IDS, WX_TEMPLATE_ID_TICKET, data):
+                self.stats["notify_fails"] += 1
+
+    def send_daily_report(self, date_str):
         """
-        统一通知入口
-        Pushover 和 NTFY 使用完全相同的 Title 和 Message
+        发送每日统计报告 (Pushover + WeChat)
         """
-        # 1. 发送 Pushover
-        success = send_pushover(title, message, priority)
-        if not success:
+        # 1. Pushover
+        stats_text_full = (
+            f"0票总场次：{self.stats['zero_total']}\n"
+            f"触发通知次数：{self.stats['alert_count']}\n"
+            f"发送通知失败次数：{self.stats['notify_fails']}\n"
+            f"API获取失败次数：{self.stats['api_fails']}"
+        )
+        
+        title = "🟢 0票场次开场前10分钟内突发购票检查就绪"
+        msg = (
+            f"{date_str}，今日排片数据已加载，开始智能检查。\n"
+            f"昨日情况：\n{stats_text_full}"
+        )
+        if not self._send_pushover(title, msg, priority=0):
             self.stats["notify_fails"] += 1
 
-        # 2. 发送 NTFY (放入队列)
-        self.ntfy_queue.put((title, message, priority))
+        # 2. WeChat
+        if WX_TEMPLATE_ID_DAILY:
+            # 格式化简短的昨日数据，避免微信字段超长
+            stats_short = (
+                f"0票场次({self.stats['zero_total']}) "
+                f"触发({self.stats['alert_count']}) "
+                f"s失败({self.stats['notify_fails']}) "
+                f"g失败({self.stats['api_fails']})"
+            )
+            
+            data = {
+                "first": f"0票场次开场前10分钟内突发购票检查就绪\n{date_str}，今日排片数据已加载，开始智能检查。\n",
+                "keyword1": "0票场次开场前10分钟内突发购票",
+                "keyword2": date_str,
+                "keyword3": stats_short,
+                "remark": "\n服务正常运行中。"
+            }
+            if not self.wx_pusher.send_template_message(WX_USER_OPEN_IDS, WX_TEMPLATE_ID_DAILY, data):
+                self.stats["notify_fails"] += 1
 
     def _run_loop(self):
-        self.log("🚀 监控服务已启动")
+        self.log("🚀 监控服务已启动 (WeChat + Pushover)")
         while True:
             try:
                 now = get_beijing_now()
@@ -320,25 +376,11 @@ class CinemaMonitor:
 
                 # --- 每日健康检查 (09:00) ---
                 if now.time() >= dt_time(9, 0) and self.last_daily_report_date != today_str:
-                    # 生成报告文案
                     report_date_str = now.strftime("%Y年%m月%d日")
-
-                    # 报告正文
-                    report_body = (
-                        f"{report_date_str}，今日排片数据已加载，开始智能检查。\n"
-                        f"昨日情况：\n"
-                        f"0票总场次：{self.stats['zero_total']}\n"
-                        f"触发通知次数：{self.stats['alert_count']}\n"
-                        f"发送通知失败次数：{self.stats['notify_fails']}\n"
-                        f"API获取失败次数：{self.stats['api_fails']}"
-                    )
-
-                    report_title = "🟢 0票场次开场前10分钟内突发购票检查就绪"
-
                     self.log(f"🔔 发送每日统计报告: {today_str}")
 
-                    # 发送报告 (Pushover 和 NTFY 内容一致)
-                    self.send_notifications(report_title, report_body, priority=0)
+                    # 发送报告
+                    self.send_daily_report(report_date_str)
 
                     # 重置统计数据 & 更新日期
                     self.last_daily_report_date = today_str
@@ -443,12 +485,8 @@ class CinemaMonitor:
                                     # 统计触发次数
                                     self.stats["alert_count"] += 1
 
-                                    # 统一文案
-                                    n_title = f"🎟 发现 {hall_short} {start} 场次有票！"
-                                    n_msg = f"{hall_short} {start}《{movie}》\n⚠️ 突增至 {sold} 张"
-
-                                    # 发送通知 (Pushover + NTFY)
-                                    self.send_notifications(n_title, n_msg, priority=1)
+                                    # 发送通知 (Pushover + WeChat)
+                                    self.send_ticket_alert(hall_short, movie, start, sold)
 
                                     self.alerted_sessions.add(unique_id)
                                     self.zero_ticket_candidates.remove(unique_id)
@@ -476,7 +514,7 @@ class CinemaMonitor:
                 time.sleep(60)
 
 
-# --- 5. Streamlit 前端 ---
+# --- 6. Streamlit 前端 ---
 
 @st.cache_resource
 def get_monitor():
@@ -492,10 +530,6 @@ def main():
 
     st.title("📢 零票影前十分钟内突发购票检查")
 
-    # 检查 NTFY 配置
-    if not NTFY_TOPIC:
-        st.warning("⚠️ 检测到 ntfy_code 环境变量未配置，NTFY 通知将不可用。")
-
     c1, c2, c3 = st.columns(3)
     with c1:
         st.metric("运行状态", monitor.status_text)