Delete pages/monitor.py

pages/monitor.py

@@ -1,496 +0,0 @@
-import streamlit as st
-import requests
-import time
-import json
-import os
-import threading
-import re
-import urllib3
-from datetime import datetime, timedelta, timezone, time as dt_time
-from collections import deque
-from dotenv import load_dotenv
-
-# --- 0. 基础配置 ---
-st.set_page_config(page_title="突发购票监控", page_icon="🛡️", layout="wide")
-
-# 屏蔽 HTTPS 证书警告
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-# 尝试导入自动刷新组件
-try:
-    from streamlit_autorefresh import st_autorefresh
-except ImportError:
-    st_autorefresh = None
-
-# --- 1. 配置与常量 ---
-load_dotenv()
-
-# 企业微信机器人配置
-WEWORK_BOT_WEBHOOK = os.getenv("WEWORK_BOT_WEBHOOK")
-
-# 影城配置
-CINEMA_ID = os.getenv("CINEMA_ID")
-
-# 获取 Token 文件路径
-CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
-ROOT_DIR = os.path.dirname(CURRENT_DIR)
-TOKEN_FILE = os.path.join(ROOT_DIR, 'token_data.json')
-
-
-# --- 2. 工具函数 ---
-def get_beijing_now():
-    """获取当前的北京时间"""
-    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 simplify_hall_name(raw_name):
-    """
-    影厅名称简化
-    例如：'【和成天下1号厅】...' -> '1号厅'
-          '6号激光厅' -> '6号厅'
-    """
-    if not raw_name:
-        return "未知影厅"
-    match = re.search(r'(\d+)号', raw_name)
-    if match:
-        return f"{match.group(1)}号厅"
-    return raw_name
-
-
-# --- 3. 企业微信机器人推送模块 ---
-class WeWorkBotPusher:
-    def __init__(self, webhook_url):
-        self.webhook_url = webhook_url
-
-    def send_text(self, content):
-        """发送纯文本消息"""
-        if not self.webhook_url:
-            return False
-
-        headers = {"Content-Type": "application/json"}
-        data = {
-            "msgtype": "text",
-            "text": {
-                "content": content
-            }
-        }
-        
-        try:
-            resp = requests.post(self.webhook_url, json=data, headers=headers, timeout=10)
-            result = resp.json()
-            if result.get("errcode") == 0:
-                return True
-            else:
-                print(f"❌ 企业微信机器人发送失败: {result}")
-                return False
-        except Exception as e:
-            print(f"❌ 企业微信机器人发送异常: {e}")
-            return False
-
-
-# --- 4. 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):
-        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()
-            self.logger(f"❌ 登录失败: {e}")
-            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
-
-
-# --- 5. 监控服务主逻辑 ---
-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.bot = WeWorkBotPusher(WEWORK_BOT_WEBHOOK)
-        
-        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.stats = {
-            "zero_total": 0,  # 纳入监控的0票总场次
-            "alert_count": 0,  # 触发通知次数
-            "api_fails": 0,  # API获取失败次数
-            "notify_fails": 0  # 发送通知失败次数
-        }
-
-        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 send_ticket_alert(self, hall_name, movie_name, show_time, ticket_count):
-        """
-        发送突发购票告警 (企业微信机器人)
-        """
-        msg = (
-            f"发现 {hall_name} {show_time} 场次有票！\n\n"
-            f"{hall_name} {show_time}《{movie_name}》\n"
-            f"突增至 {ticket_count} 张\n"
-            f"请尽快处理。"
-        )
-        
-        if not self.bot.send_text(msg):
-            self.stats["notify_fails"] += 1
-
-    def send_daily_report(self, date_str):
-        """
-        发送每日统计报告 (企业微信机器人)
-        """
-        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']}"
-        )
-        
-        msg = (
-            f"0票场次开场前10分钟内突发购票检查就绪\n\n"
-            f"{date_str}，今日排片数据已加载，开始智能检查。\n\n"
-            f"昨日情况：\n{stats_text_full}\n"
-            f"服务正常运行中。"
-        )
-        if not self.bot.send_text(msg):
-            self.stats["notify_fails"] += 1
-
-    def _run_loop(self):
-        self.log("🚀 突发购票监控服务已启动")
-        is_first_run = True  # 新增：首次运行标志
-
-        while True:
-            try:
-                now = get_beijing_now()
-                biz_date = get_business_date()
-                today_str = now.strftime("%Y-%m-%d")
-
-                # --- 每日健康检查 (09:00 或 首次运行) ---
-                if is_first_run or (now.time() >= dt_time(9, 0) and self.last_daily_report_date != today_str):
-                    report_date_str = now.strftime("%Y年%m月%d日")
-                    
-                    if is_first_run:
-                        self.log(f"🔔 首次运行，发送状态报告: {today_str}")
-                    else:
-                        self.log(f"🔔 发送每日统计报告: {today_str}")
-
-                    # 发送报告
-                    self.send_daily_report(report_date_str)
-
-                    # 重置统计数据 & 更新日期
-                    self.last_daily_report_date = today_str
-                    self.stats = {
-                        "zero_total": 0, "alert_count": 0,
-                        "api_fails": 0, "notify_fails": 0
-                    }
-                    is_first_run = False
-
-                # 休眠逻辑 (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:
-                    self.status_text = "非监控时段 (等待 09:30)"
-                    self.next_wakeup = start_check_time
-                    self.zero_ticket_candidates.clear()
-                    self.alerted_sessions.clear()
-                    time.sleep(60)
-                    continue
-
-                # 缓存刷新
-                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分钟后重试")
-                        self.stats["api_fails"] += 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)} 场")
-
-                # 筛选窗口期
-                active_check_needed = False
-                min_sleep_seconds = 3600
-                sessions_in_window = []
-                current_schedule = self.daily_schedule_cache
-
-                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
-
-                # 执行监控
-                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:
-                        self.log("⚠️ API请求失败，跳过本次判定")
-                        self.stats["api_fails"] += 1
-                        time.sleep(60)
-                        continue
-
-                    realtime_map = {f"{x.get('hallId')}_{x.get('showStartTime')}": x for x in realtime_schedule}
-                    display_monitors = []
-
-                    for session in sessions_in_window:
-                        key = f"{session['data'].get('hallId')}_{session['data'].get('showStartTime')}"
-                        unique_id = f"{biz_date}_{key}"
-
-                        latest = realtime_map.get(key)
-                        if not latest: continue
-
-                        movie = latest.get('movieName')
-                        hall_raw = latest.get('hallName')
-                        # 影厅名简化
-                        hall_short = simplify_hall_name(hall_raw)
-
-                        sold = int(latest.get('soldTicketNum') or 0)
-                        start = latest.get('showStartTime')
-
-                        if sold == 0:
-                            # 统计新加入的0票场次
-                            if unique_id not in self.zero_ticket_candidates:
-                                self.stats["zero_total"] += 1
-
-                            self.zero_ticket_candidates.add(unique_id)
-                            display_monitors.append(f"👁️ {start} {movie} (0票)")
-                        else:
-                            if unique_id in self.zero_ticket_candidates:
-                                if unique_id not in self.alerted_sessions:
-                                    self.log(f"🚨 突发购票！{hall_short}《{movie}》")
-
-                                    # 统计触发次数
-                                    self.stats["alert_count"] += 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)
-                                display_monitors.append(f"✅ {start} {movie} (新售出)")
-                            else:
-                                display_monitors.append(f"🛡️ {start} {movie} (原有票)")
-
-                    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("今日监控结束，长休眠")
-                        time.sleep(300)
-
-            except Exception as e:
-                self.log(f"主循环异常: {e}")
-                time.sleep(60)
-
-
-# --- 6. Streamlit 前端 ---
-
-@st.cache_resource
-def get_monitor():
-    return CinemaMonitor()
-
-
-def main():
-    monitor = get_monitor()
-
-    # 自动刷新：30秒
-    if st_autorefresh:
-        st_autorefresh(interval=30 * 1000, key="monitor_refresh")
-
-    st.title("📢 突发购票监控")
-
-    c1, c2, c3 = st.columns(3)
-    with c1:
-        st.metric("运行状态", monitor.status_text)
-    with c2:
-        wakeup_str = monitor.next_wakeup.strftime("%H:%M:%S") if monitor.next_wakeup else "--"
-        st.metric("下次唤醒", wakeup_str)
-
-    # --- 统计逻辑 ---
-    remaining_zero_count = 0
-    remaining_total_count = 0
-
-    if 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 show_dt and show_dt > now:
-                remaining_total_count += 1
-                if int(item.get('soldTicketNum') or 0) == 0:
-                    remaining_zero_count += 1
-
-    with c3:
-        st.metric("剩余 0 票场次 / 剩余总场次", f"{remaining_zero_count} / {remaining_total_count}")
-
-    st.divider()
-
-    col_logs, col_list = st.columns([3, 2])
-
-    with col_logs:
-        st.subheader("📜 运行日志")
-        log_text = "\n".join(list(monitor.logs))
-        st.text_area("Logs", log_text, height=500, disabled=True)
-
-    with col_list:
-        st.subheader("🎯 实时监控列表")
-        if monitor.monitored_sessions:
-            for s in monitor.monitored_sessions:
-                if "✅" in s:
-                    st.success(s)
-                elif "👁️" in s:
-                    st.error(s)
-                else:
-                    st.info(s)
-        else:
-            st.caption("暂无监控目标")
-
-
-if __name__ == "__main__":
-    main()