Create pages/monitor.py

pages/monitor.py

@@ -0,0 +1,398 @@
+import streamlit as st
+import requests
+import time
+import json
+import os
+import threading
+from datetime import datetime, timedelta, timezone, time as dt_time
+from collections import deque
+from dotenv import load_dotenv  # 引入 dotenv
+
+# --- 1. 配置与常量 ---
+load_dotenv()  # 加载根目录下的 .env 文件
+
+# 从环境变量获取配置 (请确保你的 .env 文件里配置了这些)
+PUSHOVER_TOKEN = os.getenv("PUSHOVER_TOKEN", "") 
+PUSHOVER_USER = os.getenv("PUSHOVER_USER", "")
+CINEMA_ID = os.getenv("CINEMA_ID", "44001291")
+# 登录信息也建议从环境变量取，这里为了逻辑独立暂时保留 APIManager 里的读取逻辑
+
+TOKEN_FILE = 'token_data.json' # 与主程序共用同一个 Token 文件
+
+# 页面配置
+st.set_page_config(page_title="影城防撤场监控系统 Pro", page_icon="🛡️", layout="wide")
+
+# --- 2. 核心时间处理 ---
+# ... (以下代码保持你提供的内容不变) ...
+
+def get_beijing_now():
+    """获取当前的北京时间 (Naive)"""
+    utc_now = datetime.now(timezone.utc)
+    beijing_now = utc_now.astimezone(timezone(timedelta(hours=8)))
+    return beijing_now.replace(tzinfo=None)
+
+def get_business_date():
+    """获取当前的营业日日期"""
+    now = get_beijing_now()
+    if now.time() < dt_time(6, 0):
+        return (now - timedelta(days=1)).strftime("%Y-%m-%d")
+    return now.strftime("%Y-%m-%d")
+
+def parse_show_datetime(date_str, time_str):
+    """解析排片时间"""
+    try:
+        show_t = datetime.strptime(time_str, "%H:%M").time()
+        base_date = datetime.strptime(date_str, "%Y-%m-%d")
+        if show_t < dt_time(6, 0):
+            base_date += timedelta(days=1)
+        return datetime.combine(base_date.date(), show_t)
+    except:
+        return None
+
+def send_pushover(title, message, priority=0):
+    if not PUSHOVER_TOKEN or not PUSHOVER_USER:
+        return # 如果没配置 Pushover，直接跳过
+    
+    url = "https://api.pushover.net/1/messages.json"
+    data = {
+        "token": PUSHOVER_TOKEN,
+        "user": PUSHOVER_USER,
+        "title": title,
+        "message": message,
+        "sound": "pushover",
+        "priority": priority
+    }
+    try:
+        requests.post(url, data=data, timeout=5)
+    except:
+        pass
+
+# --- 3. API 管理模块 ---
+
+class APIManager:
+    def __init__(self, logger):
+        self.last_login_fail_time = 0
+        self.logger = logger
+
+    def save_token(self, token_data):
+        try:
+            with open(TOKEN_FILE, 'w', encoding='utf-8') as f:
+                json.dump(token_data, f, ensure_ascii=False, indent=4)
+        except:
+            pass
+
+    def load_token(self):
+        # 优先读取本地文件，这样能复用 app2.py 登录生成的 Token
+        if os.path.exists(TOKEN_FILE):
+            try:
+                with open(TOKEN_FILE, 'r', encoding='utf-8') as f:
+                    return json.load(f).get('token')
+            except:
+                pass
+        return None
+
+    def login(self):
+        if time.time() - self.last_login_fail_time < 600:
+            self.logger("⏳ 登录冷却中，跳过重试。")
+            return None
+
+        self.logger("🔄 尝试后台自动登录...")
+        
+        # 使用环境变量获取账号密码
+        username = os.getenv("CINEMA_USERNAME")
+        password = os.getenv("CINEMA_PASSWORD")
+        res_code = os.getenv("CINEMA_RES_CODE")
+        device_id = os.getenv("CINEMA_DEVICE_ID")
+        
+        if not all([username, password, res_code]):
+            self.logger("❌ 环境变量缺失，无法自动登录")
+            return None
+
+        session = requests.Session()
+        session.headers.update({'User-Agent': 'Mozilla/5.0'})
+        login_url = 'https://app.bi.piao51.cn/cinema-app/credential/login.action'
+        login_data = {
+            'username': username, 'password': password, 'type': '1',
+            'resCode': res_code, 'deviceid': device_id, 'dtype': 'ios',
+        }
+
+        try:
+            session.post(login_url, data=login_data, timeout=15)
+            resp = session.get('https://app.bi.piao51.cn/cinema-app/security/logined.action', timeout=10)
+            info = resp.json()
+            if info.get("success") and info.get("data", {}).get("token"):
+                self.save_token(info['data'])
+                self.logger("✅ 登录成功。")
+                return info['data']['token']
+            else:
+                raise Exception("未获取到Token")
+        except Exception as e:
+            self.last_login_fail_time = time.time()
+            msg = f"登录失败: {e}"
+            self.logger(f"❌ {msg}")
+            send_pushover("监控异常", msg, priority=0)
+            return None
+
+    def fetch_schedule(self, date_str):
+        token = self.load_token()
+        if not token:
+            token = self.login()
+            if not token: return None
+
+        url = 'https://cawapi.yinghezhong.com/showInfo/getHallShowInfo'
+        params = {'showDate': date_str, 'token': token, '_': int(time.time() * 1000)}
+        headers = {'Origin': 'https://caw.yinghezhong.com', 'User-Agent': 'Mozilla/5.0'}
+
+        try:
+            response = requests.get(url, params=params, headers=headers, timeout=15)
+            data = response.json()
+            if data.get('code') == 1:
+                return data.get('data', [])
+            elif data.get('code') == 500:
+                self.logger("⚠️ Token失效，重试中...")
+                token = self.login()
+                if token:
+                    params['token'] = token
+                    response = requests.get(url, params=params, headers=headers, timeout=15)
+                    return response.json().get('data', [])
+            return None
+        except Exception as e:
+            self.logger(f"API请求异常: {e}")
+            return None
+
+# --- 4. 监控服务主逻辑 ---
+# ... (CinemaMonitor 类及后续代码完全保持不变，直接粘贴即可) ...
+
+class CinemaMonitor:
+    def __init__(self):
+        self.logs = deque(maxlen=50)
+        self.status_text = "初始化中..."
+        self.next_wakeup = None
+        self.monitored_sessions = []
+
+        self.api = APIManager(self.log)
+        self.current_business_date = None
+        self.daily_schedule_cache = None
+
+        self.zero_ticket_candidates = set()
+        self.alerted_sessions = set()
+
+        self.last_daily_report_date = None
+        self.simulation_mode = False
+        self.simulation_data = []
+
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.thread.start()
+
+    def log(self, msg):
+        timestamp = get_beijing_now().strftime("%H:%M:%S")
+        entry = f"[{timestamp}] {msg}"
+        print(entry)
+        self.logs.appendleft(entry)
+
+    def _run_loop(self):
+        self.log("🚀 监控服务已启动")
+
+        while True:
+            try:
+                now = get_beijing_now()
+                biz_date = get_business_date()
+                today_str = now.strftime("%Y-%m-%d")
+
+                # --- 0. 每日健康检查 (09:00) ---
+                if now.time() >= dt_time(9, 0) and self.last_daily_report_date != today_str:
+                    msg = f"日期: {today_str}\n状态: 监控服务正常运行中\nToken: 有效"
+                    self.log(f"🔔 发送每日健康检查通知: {today_str}")
+                    send_pushover("✅ 影城脚本今日运行正常", msg, priority=0)
+                    self.last_daily_report_date = today_str
+
+                # --- 1. 非营业时间休眠 (06:00 - 09:30) ---
+                start_check_time = now.replace(hour=9, minute=30, second=0, microsecond=0)
+                is_early_morning = (dt_time(6, 0) <= now.time() < dt_time(9, 30))
+
+                if is_early_morning and not self.simulation_mode:
+                    self.status_text = "非监控时段 (等待 09:30)"
+                    self.next_wakeup = start_check_time
+                    self.zero_ticket_candidates.clear()
+                    self.alerted_sessions.clear()
+
+                    sleep_sec = (start_check_time - now).total_seconds()
+                    # 最多睡5分钟醒来一次
+                    self.log(f"💤 系统休眠中... 预计唤醒: {start_check_time.strftime('%H:%M')}")
+                    time.sleep(min(sleep_sec, 300))
+                    continue
+
+                # --- 2. 缓存刷新 ---
+                if self.daily_schedule_cache is None or self.current_business_date != biz_date:
+                    self.log(f"📅 获取 {biz_date} 全天排片...")
+                    schedule = self.api.fetch_schedule(biz_date)
+                    if schedule is None:
+                        self.log("❌ 初始化获取失败，1分钟后重试")
+                        time.sleep(60)
+                        continue
+                    self.daily_schedule_cache = schedule
+                    self.current_business_date = biz_date
+                    self.zero_ticket_candidates.clear()
+                    self.alerted_sessions.clear()
+                    self.log(f"✅ 数据已更新，共 {len(schedule)} 场")
+
+                # --- 3. 筛选窗口期场次 ---
+                active_check_needed = False
+                min_sleep_seconds = 3600
+                sessions_in_window = []
+                current_schedule = self.daily_schedule_cache + self.simulation_data
+
+                for item in current_schedule:
+                    start_time_str = item.get('showStartTime')
+                    if not start_time_str: continue
+                    show_dt = parse_show_datetime(biz_date, start_time_str)
+                    if not show_dt: continue
+
+                    if now >= show_dt: continue
+
+                    monitor_start_dt = show_dt - timedelta(minutes=11)
+
+                    if now >= monitor_start_dt:
+                        sessions_in_window.append({'data': item, 'dt': show_dt})
+                        active_check_needed = True
+                    else:
+                        seconds_until_window = (monitor_start_dt - now).total_seconds()
+                        if seconds_until_window < min_sleep_seconds:
+                            min_sleep_seconds = seconds_until_window
+
+                # --- 4. 严格执行逻辑 ---
+                if active_check_needed:
+                    self.status_text = "🔥 监控进行中 (高频轮询)"
+                    self.next_wakeup = now + timedelta(seconds=60)
+
+                    realtime_schedule = self.api.fetch_schedule(biz_date)
+
+                    if realtime_schedule is None and not self.simulation_mode:
+                        self.log("⚠️ 网络/API请求失败，跳过本次判定，防止误报！")
+                        time.sleep(60)
+                        continue
+
+                    if realtime_schedule is None: realtime_schedule = []
+                    if self.simulation_mode: realtime_schedule += self.simulation_data
+
+                    realtime_map = {f"{x.get('hallId')}_{x.get('showStartTime')}": x for x in realtime_schedule}
+
+                    display_monitors = []
+
+                    for session in sessions_in_window:
+                        unique_id = f"{biz_date}_{session['data'].get('hallId')}_{session['data'].get('showStartTime')}"
+                        key_for_map = f"{session['data'].get('hallId')}_{session['data'].get('showStartTime')}"
+
+                        latest = realtime_map.get(key_for_map)
+                        if not latest: continue
+
+                        movie_name = latest.get('movieName')
+                        hall_name = latest.get('hallName')
+                        sold = int(latest.get('soldTicketNum', 0))
+                        start_str = latest.get('showStartTime')
+                        mins_left = (session['dt'] - now).total_seconds() / 60
+
+                        if sold == 0:
+                            self.zero_ticket_candidates.add(unique_id)
+                            display_monitors.append(f"👁️ {start_str} {movie_name} (0票, 监控中)")
+                        else:
+                            if unique_id in self.zero_ticket_candidates:
+                                if unique_id not in self.alerted_sessions:
+                                    self.log(f"🚨 突发购票检测！{hall_name}《{movie_name}》")
+                                    send_pushover(
+                                        f"禁止撤场：{hall_name}有票！",
+                                        f"《{movie_name}》{start_str}\n倒计时 {mins_left:.1f}分\n⚠️ 从0票突增至{sold}张",
+                                        priority=0
+                                    )
+                                    self.alerted_sessions.add(unique_id)
+                                    self.zero_ticket_candidates.remove(unique_id)
+
+                                display_monitors.append(f"✅ {start_str} {movie_name} (新售出{sold}张, 已报警)")
+                            else:
+                                display_monitors.append(f"🛡️ {start_str} {movie_name} (原有{sold}张, 忽略)")
+
+                    self.monitored_sessions = display_monitors
+                    time.sleep(60)
+
+                else:
+                    if 0 < min_sleep_seconds < 86400:
+                        wakeup_dt = now + timedelta(seconds=min_sleep_seconds)
+                        self.status_text = "💤 智能休眠中"
+                        self.next_wakeup = wakeup_dt
+                        self.monitored_sessions = []
+                        self.log(f"休眠 {min_sleep_seconds / 60:.1f} 分钟，直到 {wakeup_dt.strftime('%H:%M')}")
+                        time.sleep(min_sleep_seconds)
+                    else:
+                        self.log("今日无更多监控场次，休眠 5 分钟")
+                        time.sleep(300)
+
+            except Exception as e:
+                self.log(f"主循环异常: {e}")
+                time.sleep(60)
+
+
+# --- 5. Streamlit 前端 ---
+
+@st.cache_resource
+def get_monitor():
+    return CinemaMonitor()
+
+
+def main():
+    monitor = get_monitor()
+    st.title("📢 零票影前十分钟内突发购票通知")
+
+    # 计算统计数据：剩余0票场次 / 总场次
+    total_sessions = 0
+    zero_remaining = 0
+
+    if monitor.daily_schedule_cache:
+        total_sessions = len(monitor.daily_schedule_cache)
+        now = get_beijing_now()
+        biz_date = monitor.current_business_date or get_business_date()
+
+        for item in monitor.daily_schedule_cache:
+            start_str = item.get('showStartTime')
+            if not start_str: continue
+
+            show_dt = parse_show_datetime(biz_date, start_str)
+            if not show_dt: continue
+
+            # 只统计还没开场的
+            if show_dt > now:
+                # 统计票数为0的 (基于缓存数据)
+                if int(item.get('soldTicketNum', 0)) == 0:
+                    zero_remaining += 1
+
+    c1, c2, c3 = st.columns(3)
+    with c1:
+        st.metric("当前状态", monitor.status_text)
+    with c2:
+        st.metric("下次唤醒", monitor.next_wakeup.strftime("%H:%M:%S") if monitor.next_wakeup else "--")
+    with c3:
+        st.metric("剩余零票场次 / 当日全部场次", f"{zero_remaining} / {total_sessions}")
+
+    st.divider()
+
+    col_logs, col_list = st.columns([3, 2])
+    with col_logs:
+        st.subheader("📜 运行日志")
+        # 移除按钮
+        st.text_area("Logs", "\n".join(list(monitor.logs)), height=450, disabled=True)
+    with col_list:
+        st.subheader("🎯 实时监控列表")
+        st.markdown("> 说明：当日 0 票场次会被监控中，在 0 票电影开场前 10 分钟内突发购票会触发通知。")
+        if monitor.monitored_sessions:
+            for s in monitor.monitored_sessions:
+                if "✅" in s:
+                    st.success(s)
+                elif "👁️" in s:
+                    st.error(s)
+                else:
+                    st.caption(s)
+        else:
+            st.info("当前休眠中，无临近场次")
+
+
+if __name__ == "__main__":
+    main()