import tkinter as tk from tkinter import ttk, scrolledtext import requests import json import threading import time from datetime import datetime import winsound import os import random import string class TicketMonitor: def __init__(self, root): self.root = root self.root.title("余票监控系统") self.root.geometry("1000x850") # 监控状态 self.is_monitoring = False self.monitor_thread = None # 默认API配置 self.api_url = "https://newmanage.mgk.org.cn/campaign/noTokenapi/listCinemaNumByDay?corpCode=0001" self.headers_template = { 'Connection': 'keep-alive', 'content-type': 'application/json', 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 26_5_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.75(0x18004b3c) NetType/WIFI Language/zh_CN', 'Referer': 'https://servicewechat.com/wx66fdc5f2e062a173/163/page-frame.html' } self.payload = { "corpCode": "0001", "orderFlag": "C" } # 生成固定的Token前缀 + 随机6位 self.token_prefix = "eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiIxODU5MTc3NzI4MSIsImNvcnBOYW1lIjoi5rKz5Y2X5bmz5Y-w6YKu5Lu35Yy6IiwiY29ycENvZGUiOiIwMDAxIiwiZXhwIjoxNzU1NDE1MTY3LCJpYXQiOjE3MjM4NzkxNjcsImp0aSI6IjY2Y2NmODVkLTUzMDEtNDdmMS1hZmI4LWM5NDU5YjI3YzdmOSJ9." # 生成随机6位字符（字母+数字） self.random_suffix = ''.join(random.choices(string.ascii_letters + string.digits, k=6)) self.token = self.token_prefix + self.random_suffix # 请求计数器 self.request_count = 0 self.auto_refresh_interval = 30 # 每30次请求自动刷新Token # 用于记录已通知的日期 self.notified_dates = set() # 日期选择相关变量 self.selected_dates = set() # 选中的日期 # 初始化UI self.setup_ui() # 显示生成的Token信息（仅日志） self.log(f"Token已自动生成，后缀: {self.random_suffix}", "info") def setup_ui(self): # 创建主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.pack(fill=tk.BOTH, expand=True) # ========== 监控设置区域 ========== settings_frame = ttk.LabelFrame(main_frame, text="监控设置", padding="10") settings_frame.pack(fill=tk.X, pady=(0, 10)) # 第一行：基本设置 basic_frame = ttk.Frame(settings_frame) basic_frame.pack(fill=tk.X, pady=5) ttk.Label(basic_frame, text="监控间隔(秒):").pack(side=tk.LEFT, padx=5) self.interval_var = tk.StringVar(value="10") interval_spinbox = ttk.Spinbox(basic_frame, from_=3, to=60, textvariable=self.interval_var, width=5) interval_spinbox.pack(side=tk.LEFT, padx=5) # 应急票监控开关 ttk.Label(basic_frame, text="应急票监控:").pack(side=tk.LEFT, padx=(20, 5)) self.monitor_emergency_var = tk.BooleanVar(value=True) emergency_switch = ttk.Checkbutton(basic_frame, text="开启", variable=self.monitor_emergency_var) emergency_switch.pack(side=tk.LEFT, padx=5) # 只提醒一次 self.notify_once_var = tk.BooleanVar(value=True) ttk.Checkbutton(basic_frame, text="📌 只提醒一次", variable=self.notify_once_var).pack(side=tk.LEFT, padx=20) # 声音提醒 self.sound_var = tk.BooleanVar(value=True) ttk.Checkbutton(basic_frame, text="🔊 声音提醒", variable=self.sound_var).pack(side=tk.LEFT, padx=5) # 第二行：日期选择模式 date_mode_frame = ttk.Frame(settings_frame) date_mode_frame.pack(fill=tk.X, pady=5) ttk.Label(date_mode_frame, text="监控模式:", width=15).pack(side=tk.LEFT, padx=5) self.monitor_mode = tk.StringVar(value="all") ttk.Radiobutton(date_mode_frame, text="所有日期", variable=self.monitor_mode, value="all", command=self.on_mode_change).pack(side=tk.LEFT, padx=5) ttk.Radiobutton(date_mode_frame, text="选择日期", variable=self.monitor_mode, value="selected", command=self.on_mode_change).pack(side=tk.LEFT, padx=5) # 快捷选择按钮 self.select_all_btn = ttk.Button(date_mode_frame, text="全选", command=self.select_all_dates, state=tk.DISABLED) self.select_all_btn.pack(side=tk.LEFT, padx=5) self.deselect_all_btn = ttk.Button(date_mode_frame, text="取消全选", command=self.deselect_all_dates, state=tk.DISABLED) self.deselect_all_btn.pack(side=tk.LEFT, padx=5) # Token状态显示（替代原来的刷新按钮） token_status_frame = ttk.Frame(date_mode_frame) token_status_frame.pack(side=tk.RIGHT, padx=5) self.token_status_label = ttk.Label(token_status_frame, text=f"Token: {self.random_suffix} (0次)") self.token_status_label.pack() # 控制按钮 control_frame = ttk.Frame(settings_frame) control_frame.pack(fill=tk.X, pady=10) self.start_btn = ttk.Button(control_frame, text="▶ 开始监控", command=self.start_monitoring, width=12) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button(control_frame, text="■ 停止监控", command=self.stop_monitoring, state=tk.DISABLED, width=12) self.stop_btn.pack(side=tk.LEFT, padx=5) ttk.Button(control_frame, text="🔍 立即查询", command=self.manual_check, width=12).pack(side=tk.LEFT, padx=5) # ========== 状态显示 ========== status_frame = ttk.LabelFrame(main_frame, text="状态信息", padding="5") status_frame.pack(fill=tk.X, pady=(0, 10)) status_left = ttk.Frame(status_frame) status_left.pack(side=tk.LEFT, fill=tk.X, expand=True) self.status_label = ttk.Label(status_left, text="⏹ 未开始监控", foreground="gray") self.status_label.pack(side=tk.LEFT, padx=5) status_right = ttk.Frame(status_frame) status_right.pack(side=tk.RIGHT) self.last_update_label = ttk.Label(status_right, text="最后更新: --") self.last_update_label.pack(side=tk.RIGHT, padx=5) self.count_label = ttk.Label(status_right, text="发现票数: 0") self.count_label.pack(side=tk.RIGHT, padx=20) # ========== 日期选择区域 ========== date_select_frame = ttk.LabelFrame(main_frame, text="日期选择 (点击选择/取消选择日期)", padding="10") date_select_frame.pack(fill=tk.X, pady=(0, 10)) # 创建日期选择框架 date_canvas_frame = ttk.Frame(date_select_frame) date_canvas_frame.pack(fill=tk.X, pady=5) self.date_checkboxes = {} # 存储日期和对应的Checkbutton变量 self.date_check_vars = {} # 存储BooleanVar # 创建日期选择容器 self.date_container = ttk.Frame(date_canvas_frame) self.date_container.pack(fill=tk.X) # 先创建默认的日期占位 self.init_date_checkboxes() # ========== 结果显示区域 ========== result_frame = ttk.LabelFrame(main_frame, text="余票信息", padding="5") result_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) # 创建Treeview显示数据 columns = ("日期", "常规票(A)", "应急票(H)", "状态") self.tree = ttk.Treeview(result_frame, columns=columns, show="headings", height=12) # 设置列标题和宽度 column_widths = {"日期": 120, "常规票(A)": 100, "应急票(H)": 100, "状态": 100} for col in columns: self.tree.heading(col, text=col) self.tree.column(col, width=column_widths.get(col, 100), anchor="center") # 添加滚动条 scrollbar = ttk.Scrollbar(result_frame, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=scrollbar.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # ========== 日志区域 ========== log_frame = ttk.LabelFrame(main_frame, text="日志", padding="5") log_frame.pack(fill=tk.BOTH, expand=True) self.log_text = scrolledtext.ScrolledText(log_frame, height=6, state=tk.DISABLED) self.log_text.pack(fill=tk.BOTH, expand=True) # 配置日志颜色标签 self.log_text.tag_configure("error", foreground="red") self.log_text.tag_configure("success", foreground="green") self.log_text.tag_configure("info", foreground="blue") self.log_text.tag_configure("warning", foreground="orange") def init_date_checkboxes(self): """初始化日期复选框""" # 清空现有复选框 for widget in self.date_container.winfo_children(): widget.destroy() self.date_checkboxes.clear() self.date_check_vars.clear() # 创建示例日期（实际使用时会从API获取真实日期） example_dates = [] row_frame = None for i, date in enumerate(example_dates): if i % 8 == 0: row_frame = ttk.Frame(self.date_container) row_frame.pack(fill=tk.X, pady=2) var = tk.BooleanVar(value=True) self.date_check_vars[date] = var cb = ttk.Checkbutton(row_frame, text=date, variable=var, command=lambda d=date: self.on_date_toggle(d)) cb.pack(side=tk.LEFT, padx=10) self.date_checkboxes[date] = cb # 默认全选 self.selected_dates.add(date) def update_date_checkboxes(self, dates): """更新日期复选框列表""" # 清空现有复选框 for widget in self.date_container.winfo_children(): widget.destroy() self.date_checkboxes.clear() self.date_check_vars.clear() self.selected_dates.clear() # 按日期排序 sorted_dates = sorted(dates) row_frame = None for i, date in enumerate(sorted_dates): if i % 8 == 0: row_frame = ttk.Frame(self.date_container) row_frame.pack(fill=tk.X, pady=2) var = tk.BooleanVar(value=True) # 默认选中 self.date_check_vars[date] = var cb = ttk.Checkbutton(row_frame, text=date, variable=var, command=lambda d=date: self.on_date_toggle(d)) cb.pack(side=tk.LEFT, padx=10) self.date_checkboxes[date] = cb # 添加到选中集合 self.selected_dates.add(date) def on_date_toggle(self, date): """日期选择切换""" if date in self.date_check_vars: if self.date_check_vars[date].get(): self.selected_dates.add(date) self.log(f"已选择日期: {date}", "info") else: self.selected_dates.discard(date) self.log(f"已取消选择日期: {date}", "warning") def select_all_dates(self): """全选日期""" for date, var in self.date_check_vars.items(): var.set(True) self.selected_dates.add(date) self.log("已全选所有日期", "info") def deselect_all_dates(self): """取消全选""" for date, var in self.date_check_vars.items(): var.set(False) self.selected_dates.discard(date) self.log("已取消全选所有日期", "warning") def on_mode_change(self): """监控模式切换""" if self.monitor_mode.get() == "selected": self.select_all_btn.config(state=tk.NORMAL) self.deselect_all_btn.config(state=tk.NORMAL) self.log("切换到日期选择模式，请选择需要监控的日期", "info") else: self.select_all_btn.config(state=tk.DISABLED) self.deselect_all_btn.config(state=tk.DISABLED) self.log("切换到所有日期模式", "info") def regenerate_token(self): """重新生成Token（随机6位后缀）""" self.random_suffix = ''.join(random.choices(string.ascii_letters + string.digits, k=6)) self.token = self.token_prefix + self.random_suffix self.log(f"Token已重新生成，新后缀: {self.random_suffix}", "info") # 更新Token状态显示 self.token_status_label.config(text=f"Token: {self.random_suffix} ({self.request_count}次)") def get_headers(self): """获取请求头""" headers = self.headers_template.copy() headers['access-token'] = self.token return headers def log(self, message, tag=None): """添加日志消息""" self.log_text.config(state=tk.NORMAL) timestamp = datetime.now().strftime("%H:%M:%S") # 添加时间戳 self.log_text.insert(tk.END, f"[{timestamp}] ", "info") # 添加消息（可带颜色标签） if tag: self.log_text.insert(tk.END, f"{message}\n", tag) else: self.log_text.insert(tk.END, f"{message}\n") self.log_text.see(tk.END) self.log_text.config(state=tk.DISABLED) def update_status(self, message, color="black"): """更新状态显示""" self.status_label.config(text=message, foreground=color) def fetch_ticket_data(self): """获取余票数据""" # 增加请求计数并自动刷新Token self.request_count += 1 # 每30次请求自动刷新Token if self.request_count % self.auto_refresh_interval == 0: self.regenerate_token() self.log(f"🔄 已自动刷新Token (第{self.request_count}次请求)", "info") headers = self.get_headers() try: response = requests.post( self.api_url, headers=headers, json=self.payload, timeout=10 ) if response.status_code == 200: data = response.json() if data.get('status') == 200: return data.get('data', []) else: self.log(f"API返回错误: {data}", "error") return None else: self.log(f"HTTP请求失败: {response.status_code}", "error") return None except requests.exceptions.Timeout: self.log("请求超时，请检查网络连接", "error") return None except requests.exceptions.ConnectionError: self.log("网络连接失败，请检查网络", "error") return None except Exception as e: self.log(f"请求异常: {str(e)}", "error") return None def should_monitor_date(self, date): """检查是否应该监控该日期""" if self.monitor_mode.get() == "all": return True else: return date in self.selected_dates def check_and_notify(self, ticket_data): """检查余票并发送通知""" new_notified = set() total_tickets = 0 # 更新日期复选框（如果有新日期） dates_in_data = [item.get('DAY', '') for item in ticket_data if item.get('DAY')] if dates_in_data: # 获取当前显示的日期列表 current_dates = set(self.date_check_vars.keys()) new_dates = set(dates_in_data) - current_dates if new_dates: # 如果有新日期，更新复选框 self.update_date_checkboxes(dates_in_data) self.log(f"更新日期列表，新增 {len(new_dates)} 个日期", "info") for item in ticket_data: day = item.get('DAY', '') A_num = int(item.get('A_num', 0)) H_num = int(item.get('H_num', 0)) # 检查是否应该监控该日期 if not self.should_monitor_date(day): continue # 检查是否监控应急票 monitor_emergency = self.monitor_emergency_var.get() # 统计有票的日期（根据设置决定是否检查应急票） has_ticket = False if monitor_emergency: has_ticket = A_num > 0 or H_num > 0 else: has_ticket = A_num > 0 # 只监控常规票 if has_ticket: total_tickets += 1 # 判断是否应该提醒 should_notify = False if self.notify_once_var.get(): if day not in self.notified_dates: should_notify = True new_notified.add(day) else: should_notify = True new_notified.add(day) # 发送提醒 if should_notify: ticket_parts = [] if A_num > 0: ticket_parts.append(f"常规票{A_num}张") if monitor_emergency and H_num > 0: ticket_parts.append(f"应急票{H_num}张") msg = f"🎫 {day} 有票了！{' + '.join(ticket_parts)}" self.log(msg, "success") self.update_status(f"🎉 发现余票！{day}", "green") # 声音提醒 if self.sound_var.get(): self.play_sound() # 更新已通知日期集合 if self.notify_once_var.get(): self.notified_dates.update(new_notified) else: self.notified_dates = set() # 更新计数 self.count_label.config(text=f"发现票数: {total_tickets}") return total_tickets def play_sound(self): """播放声音提醒""" try: winsound.MessageBeep(winsound.MB_ICONASTERISK) except Exception as e: pass def update_display(self, ticket_data): """更新表格显示""" # 清空现有数据 for item in self.tree.get_children(): self.tree.delete(item) if not ticket_data: self.log("没有获取到数据", "warning") return # 添加新数据 for item in ticket_data: day = item.get('DAY', '') A_num = item.get('A_num', '0') H_num = item.get('H_num', '0') sumNum = item.get('sumNum', '无票') # 检查是否应该监控该日期 if not self.should_monitor_date(day): continue values = (day, A_num, H_num, sumNum) item_id = self.tree.insert("", tk.END, values=values) # 修改颜色逻辑：常规票有票时显示绿色 monitor_emergency = self.monitor_emergency_var.get() # 检查常规票(A)是否有票 has_regular_ticket = int(A_num) > 0 # 检查应急票(H)是否有票 has_emergency_ticket = monitor_emergency and int(H_num) > 0 # 只要有常规票，就显示绿色 if has_regular_ticket: self.tree.tag_configure('has_regular_ticket', background='#90EE90') # 亮绿色 self.tree.item(item_id, tags=('has_regular_ticket',)) # 如果只有应急票，显示浅蓝色 elif has_emergency_ticket: self.tree.tag_configure('has_emergency_ticket', background='#B0E0E6') # 浅蓝色 self.tree.item(item_id, tags=('has_emergency_ticket',)) # 无票不显示特殊颜色 def manual_check(self): """手动查询""" self.log("手动查询中...", "info") self.update_status("查询中...", "orange") data = self.fetch_ticket_data() if data: self.update_display(data) total = self.check_and_notify(data) self.last_update_label.config(text=f"最后更新: {datetime.now().strftime('%H:%M:%S')}") self.update_status(f"查询完成，发现 {total} 天有票", "blue") self.log(f"查询完成，共 {len(data)} 天数据", "info") else: self.update_status("查询失败", "red") def monitoring_loop(self): """监控循环""" self.log("开始监控...", "info") self.update_status("🔄 监控运行中", "green") check_count = 0 while self.is_monitoring: try: check_count += 1 self.log(f"第 {check_count} 次检测...", "info") # 获取数据 data = self.fetch_ticket_data() if data: self.update_display(data) self.check_and_notify(data) self.last_update_label.config(text=f"最后更新: {datetime.now().strftime('%H:%M:%S')}") self.update_status(f"🔄 监控运行中 (第{check_count}次)", "green") # 更新Token状态显示 self.token_status_label.config(text=f"Token: {self.random_suffix} ({self.request_count}次)") else: self.update_status("⚠️ 监控运行中 (获取数据失败)", "orange") # 等待指定的间隔 interval = int(self.interval_var.get()) for _ in range(interval): if not self.is_monitoring: break time.sleep(1) except Exception as e: self.log(f"监控循环错误: {str(e)}", "error") time.sleep(5) self.log("监控已停止", "info") self.update_status("⏹ 已停止", "gray") def start_monitoring(self): """开始监控""" if self.is_monitoring: return self.is_monitoring = True self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) # 重置计数器 self.request_count = 0 self.token_status_label.config(text=f"Token: {self.random_suffix} (0次)") # 清空已通知列表 self.notified_dates.clear() # 在新线程中运行监控 self.monitor_thread = threading.Thread(target=self.monitoring_loop, daemon=True) self.monitor_thread.start() def stop_monitoring(self): """停止监控""" self.is_monitoring = False self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) if self.monitor_thread: self.monitor_thread.join(timeout=2) self.update_status("⏹ 已停止", "gray") def main(): root = tk.Tk() app = TicketMonitor(root) # 窗口关闭处理 def on_closing(): app.stop_monitoring() root.destroy() root.protocol("WM_DELETE_WINDOW", on_closing) # 居中显示 root.update_idletasks() width = root.winfo_width() height = root.winfo_height() x = (root.winfo_screenwidth() // 2) - (width // 2) y = (root.winfo_screenheight() // 2) - (height // 2) root.geometry(f'+{x}+{y}') root.mainloop() if __name__ == "__main__": main()