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 # --- 1. 配置与常量 --- PUSHOVER_TOKEN = "a5cynwsr19y8bdvn2y447bg3vc51fr" PUSHOVER_USER = "uc86mdifsboas7bczuxmnoyq6qyouw" TOKEN_FILE = 'token_data.json' CINEMA_ID = "44001291" # 页面配置 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): 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): 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("🔄 尝试后台自动登录...") 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': 'chenhy', 'password': '123456', 'type': '1', 'resCode': '44001291', 'deviceid': '1517bfd3f6e7fef1333', '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): """ 获取排片数据。 返回 None 表示网络或API错误 (需要处理)。 返回 [] 表示当天没排片或数据为空。 """ 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 # 网络异常返回 None # --- 4. 监控服务主逻辑 (修复版) --- 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.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() # --- 1. 非营业时间休眠 --- 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() 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) # 如果网络失败，realtime_schedule 为 None 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')}" # 关键修复：绝对不能使用 fallback 回退到缓存数据！ 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: # 确认为0票，加入候选 self.zero_ticket_candidates.add(unique_id) display_monitors.append(f"👁️ {start_str} {movie_name} (0票, 监控中)") else: # 现在的票数 > 0 if unique_id in self.zero_ticket_candidates: # 之前我们确认过它是0票，现在变了 -> 报警 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: # 它从未进入过0票名单，说明一开始就有票 -> 忽略 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) def trigger_simulation(self): self.simulation_mode = True fake_time = (get_beijing_now() + timedelta(minutes=5)).strftime("%H:%M") # 1. 先注入 0 票 self.simulation_data = [{ 'movieName': '测试影片-防误报验证', 'hallName': '模拟厅', 'hallId': 'SIM_SAFE', 'showStartTime': fake_time, 'soldTicketNum': '0' }] self.log(f"已注入 0票 模拟场次: {fake_time} 开场") # 2. 10秒后改为 1 票 def simulate_buy(): time.sleep(10) self.simulation_data[0]['soldTicketNum'] = '1' self.log("⚡ [测试] 模拟动作：购票成功！应触发报警。") threading.Thread(target=simulate_buy).start() # --- 5. Streamlit 前端 --- @st.cache_resource def get_monitor(): return CinemaMonitor() def main(): monitor = get_monitor() st.title("📽️ 影城排程防撤场监控终端 Pro") utc_now = datetime.now(timezone.utc).strftime("%H:%M:%S") bj_now = get_beijing_now().strftime("%H:%M:%S") 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("系统时间 (UTC / BJ)", f"{utc_now} / {bj_now}") st.divider() col_logs, col_list = st.columns([3, 2]) with col_logs: st.subheader("📜 系统运行日志") if st.button("刷新日志"): pass st.text_area("Logs", "\n".join(list(monitor.logs)), height=450, disabled=True) with col_list: st.subheader("🎯 实时监控列表") st.info("说明：👁️=0票监控中 | ✅=突发购票(报警) | 🛡️=原有票(安全)") 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("当前休眠中，无临近场次") st.write("---") if st.button("模拟 '0票 -> 1票' 验证"): monitor.trigger_simulation() st.toast("测试已触发") if __name__ == "__main__": main()