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() # Pushover 配置 PUSHOVER_TOKEN = os.getenv("PUSHOVER_TOKEN") PUSHOVER_USER = os.getenv("PUSHOVER_USER") # 微信配置 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__)) 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 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, force_refresh=False): """获取并缓存微信 Access Token :param force_refresh: 是否强制刷新 Token """ if not force_refresh and 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): """发送模板消息给多个用户 (带 40001 错误重试机制)""" # 组装微信数据格式 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()] if not valid_ids: return False success_any = False for user_id in valid_ids: # 每个用户最多尝试 2 次（1次正常，1次刷新 Token 后重试） for attempt in range(2): # 如果是重试（attempt > 0），强制刷新 Token token = self.get_access_token(force_refresh=(attempt > 0)) if not token: break # Token 都拿不到，不用重试了 url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={token}" payload = { "touser": user_id, "template_id": template_id, "data": formatted_data } try: resp = requests.post(url, json=payload, timeout=10) result = resp.json() errcode = result.get("errcode") if errcode == 0: # 发送成功 # print(f"✅ 微信发送成功 [{user_id}]") success_any = True break # 跳出重试循环，处理下一个用户 # 40001: Token 无效 # 40014: Token 不合法 # 42001: Token 过期 elif errcode in [40001, 40014, 42001]: print(f"🔄 微信 Token 失效 ({errcode})，正在刷新重试 [{user_id}]...") continue # 进入下一次循环，强制刷新 Token else: print(f"❌ 微信发送失败 [{user_id}]: {result}") break # 其他错误，不重试 except Exception as e: print(f"❌ 微信发送异常 [{user_id}]: {e}") break return success_any # --- 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.wx_pusher = WeChatPusher(WX_APP_ID, WX_APP_SECRET) 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_pushover(self, 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: return True else: print(f"❌ Pushover 拒绝: {response.status_code}") return False except Exception as e: print(f"❌ Pushover 网络异常: {e}") return False 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 # 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 + WeChat) """ # 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. 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("🚀 监控服务已启动 (WeChat + Pushover)") while True: try: now = get_beijing_now() biz_date = get_business_date() today_str = now.strftime("%Y-%m-%d") # --- 每日健康检查 (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日") 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 } # 休眠逻辑 (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', 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', 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("🎯 当前监控窗口 (开场前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.info(s) else: st.caption("暂无临近开场的监控目标。") if __name__ == "__main__": main()