# -*- coding: utf-8 -*- import requests import time import tkinter as tk from tkinter import scrolledtext, ttk, messagebox, colorchooser from threading import Thread from datetime import datetime, timedelta import os import sys import asyncio import edge_tts import pygame import queue import hashlib import json import uuid def resource_path(rel_path): try: base = sys._MEIPASS except: base = os.path.dirname(os.path.abspath(__file__)) return os.path.join(base, rel_path) if getattr(sys, 'frozen', False): BASE_DIR = os.path.dirname(sys.executable) else: BASE_DIR = os.path.dirname(os.path.abspath(__file__)) APP_DATA_DIR = os.path.join(BASE_DIR, "MegaMonitor_Data") RECORDS_DIR = os.path.join(APP_DATA_DIR, "Records") CACHE_DIR = os.path.join(APP_DATA_DIR, "Cache") for d in [APP_DATA_DIR, RECORDS_DIR, CACHE_DIR]: os.makedirs(d, exist_ok=True) def clean_old_logs(): try: now = datetime.now() for f in os.listdir(RECORDS_DIR): path = os.path.join(RECORDS_DIR, f) if os.path.isfile(path): if now - datetime.fromtimestamp(os.path.getctime(path)) > timedelta(hours=48): os.remove(path) except: pass clean_old_logs() AUDIO_AVAILABLE = True try: pygame.mixer.pre_init(44100, -16, 2, 512) pygame.mixer.init() except Exception: AUDIO_AVAILABLE = False def format_amt_display(num): f_num = float(num) return f"{int(f_num)}" if f_num == int(f_num) else f"{f_num:,.2f}" def cn_amount(num): try: units = ['', '十', '百', '千', '万'] digits = '零一二三四五六七八九' def _int_to_cn(n): if n == 0: return digits[0] res = "" s = str(n)[::-1] for i, d in enumerate(s): if d != '0': res = digits[int(d)] + units[i] + res else: if res and res[0] != digits[0]: res = digits[0] + res return res.rstrip('零').lstrip('一') f_num = float(num) if f_num == int(f_num): return _int_to_cn(int(f_num)) i, d = f"{f_num:.2f}".split('.') return _int_to_cn(int(i)) + "点" + "".join(digits[int(x)] for x in d) except: return str(num) LANG_DICT = { "zh": { "title": "MEGA收款播报", "start_btn": "开启播报系统", "lang_sel": "语言:", "status_ok": "系统监控中", "reboot": "优雅重启", "settings": "⚙ 设置", "freq": "检查频率(秒):", "float_title": "悬浮窗设置", "float_a": "开启账号A悬浮窗", "float_b": "开启账号B悬浮窗", "float_w": "宽度", "float_h": "高度", "speech_title": "播报设置", "speed": "语速", "voice_a": "账号A音色", "voice_b": "账号B音色", "payer": "付款人", "wait": "等待第一笔入账...", "total_label": "今日总营业额: $", "float_alpha": "透明度", "remember_acc": "记住并保存账号信息", "check_token": "校验Token有效性", "clear_a": "清空账号A", "clear_b": "清空账号B", "font_size_main": "内容字体大小", "font_size_total": "总额字体大小", "color_main": "内容文字颜色", "color_total": "总额文字颜色", "speech_mode": "播报句式模式", "mode_with_name": "带账号名播报", "mode_no_name": "不带账号名播报", "income": "收款成功", "expense": "支出", "received_success": "收款成功" }, "es": { "title": "MEGA Notificador", "start_btn": "INICIAR MONITOR", "lang_sel": "Idioma:", "status_ok": "En línea", "reboot": "Reiniciar", "settings": "⚙ Ajustes", "freq": "Frecuencia (seg):", "float_title": "Ventana flotante", "float_a": "Ventana Cuenta A", "float_b": "Ventana Cuenta B", "float_w": "Ancho", "float_h": "Alto", "speech_title": "Voz", "speed": "Velocidad", "voice_a": "Voz A", "voice_b": "Voz B", "payer": "Cliente", "wait": "Esperando...", "total_label": "Total Hoy: $", "float_alpha": "Transparencia", "remember_acc": "Recordar cuenta", "check_token": "Validar Token", "clear_a": "Limpiar A", "clear_b": "Limpiar B", "font_size_main": "Tamaño fuente", "font_size_total": "Tamaño total", "color_main": "Color texto", "color_total": "Color total", "speech_mode": "Modo de voz", "mode_with_name": "Con nombre de cuenta", "mode_no_name": "Sin nombre de cuenta", "income": "Ingreso", "expense": "Egreso", "received_success": "Recibido" } } VOICE_MAP = { "zh": {"女声": "zh-CN-XiaoxiaoNeural", "男声": "zh-CN-YunxiNeural"}, "es": {"女声": "es-AR-ElenaNeural", "男声": "es-AR-TomasNeural"} } class BigNotification: def __init__(self, master, amount, payer, lang, is_income=True, is_test=False): self.top = tk.Toplevel(master) self.top.overrideredirect(True) self.top.attributes("-topmost", True, "-alpha", 0.95) color = "#00FF00" if is_income else "#FF0000" if is_income: if lang == "zh": title = "收款成功" else: title = "Recibido" else: if lang == "zh": title = "订单取消" else: title = "Cancelación" if is_test: if lang == "zh": title = f"【测试】{title}" else: title = f"【Prueba】{title}" self.top.configure(bg='#1A1A1A', highlightbackground=color, highlightthickness=3) w, h = 420, 160 sw = self.top.winfo_screenwidth() self.top.geometry(f"{w}x{h}+{sw - w - 20}+50") tk.Label(self.top, text=title, font=("Arial", 12, "bold"), fg=color, bg='#1A1A1A').pack(pady=(12, 0)) tk.Label(self.top, text=f"${format_amt_display(amount)}", font=("Verdana", 38, "bold"), fg=color, bg='#1A1A1A').pack() tk.Label(self.top, text=f"{LANG_DICT[lang]['payer']}: {payer}", font=("Arial", 11), fg="white", bg='#1A1A1A').pack(pady=5) self.top.after(5000, self.top.destroy) class MegaMonitorApp: def __init__(self): self.root = tk.Tk() try: win_icon = resource_path("app_icon.ico") self.root.iconbitmap(win_icon) except: pass self.is_running = True self.processed_ids = set() self.speech_queue = queue.Queue() self.lang = tk.StringVar(value="zh") self.today_total = 0.0 self.current_date = datetime.now().strftime("%Y.%m.%d") self.check_interval = tk.IntVar(value=3) self.float_width = tk.IntVar(value=300) self.float_height = tk.IntVar(value=120) self.float_alpha = tk.IntVar(value=75) self.font_size_main = tk.IntVar(value=10) self.font_size_total = tk.IntVar(value=16) self.color_main = tk.StringVar(value="#FFFFFF") self.color_total = tk.StringVar(value="#FFD700") self.speech_mode = tk.StringVar(value="with_name") self.speech_rate = tk.IntVar(value=0) self.show_float_a = tk.BooleanVar(value=True) self.show_float_b = tk.BooleanVar(value=True) self.voice_a = tk.StringVar(value="女声") self.voice_b = tk.StringVar(value="男声") self.remember_account = tk.BooleanVar(value=True) self.fraud_window = tk.IntVar(value=120) self.last_income_record = {} self.float_wins = {} self.config_file = os.path.join(APP_DATA_DIR, "config_final.json") self.speech_timer = None # ================= ✅ 新增：记录启动时间 ================= self.startup_time = time.time() # ====================================================== self.root.geometry("420x360") self.setup_login_ui() Thread(target=self._speech_worker, daemon=True).start() def get_lang(self): return self.lang.get() def get_mp_account_id(self, token): """用 Token 获取 MercadoPago 账号 ID""" try: headers = {"Authorization": f"Bearer {token}"} r = requests.get("https://api.mercadopago.com/users/me", headers=headers, timeout=10) if r.status_code == 200: return str(r.json().get("id")) except: pass return None def pick_color_main(self): c = colorchooser.askcolor(title="选择内容文字颜色")[1] if c: self.color_main.set(c) self.refresh_float_font() def pick_color_total(self): c = colorchooser.askcolor(title="选择总额文字颜色")[1] if c: self.color_total.set(c) self.refresh_float_font() def refresh_float_font(self): for d in self.float_wins.values(): d['lbl'].config(font=("Arial", self.font_size_main.get()), fg=self.color_main.get()) d['total_lbl'].config(font=("Arial", self.font_size_total.get(), "bold"), fg=self.color_total.get()) def check_token_valid(self, token): if not token: return False try: r = requests.get("https://api.mercadopago.com/users/me", headers={"Authorization": f"Bearer {token}"}, timeout=8) return r.status_code == 200 except: return False def validate_all_tokens(self): t1, t2 = self.e_a_t.get().strip(), self.e_b_t.get().strip() ok1 = self.check_token_valid(t1) if t1 else True ok2 = self.check_token_valid(t2) if t2 else True messagebox.showinfo("校验结果", "✅ 所有Token有效") if ok1 and ok2 else messagebox.showerror("校验结果", "❌ 存在无效Token") def clear_account_a(self): self.e_a_n.delete(0, tk.END) self.e_a_t.delete(0, tk.END) def clear_account_b(self): self.e_b_n.delete(0, tk.END) self.e_b_t.delete(0, tk.END) def setup_login_ui(self): self.root.title("MEGA - 系统初始化") self.login_frame = tk.Frame(self.root) self.login_frame.place(relx=0.5, rely=0.5, anchor=tk.CENTER) saved = self.load_configs() for i, (tag, color) in enumerate([("A", "#009EE3"), ("B", "#9B59B6")]): tk.Label(self.login_frame, text=f"--- 账号 {tag} ---", font=("Arial", 9, "bold"), fg=color).pack(pady=(10,0)) f = tk.Frame(self.login_frame); f.pack() tk.Label(f, text="名称:").grid(row=0, column=0) n_ent = tk.Entry(f, width=15); n_ent.grid(row=0, column=1) n_ent.insert(0, saved[i]['name']) tk.Label(self.login_frame, text="Token:").pack() t_ent = tk.Entry(self.login_frame, width=40, show="*"); t_ent.pack() t_ent.insert(0, saved[i]['token']) if i==0: self.e_a_n, self.e_a_t = n_ent, t_ent else: self.e_b_n, self.e_b_t = n_ent, t_ent tk.Checkbutton(self.login_frame, text=LANG_DICT[self.get_lang()]["remember_acc"], variable=self.remember_account).pack(pady=5) btn_frame = tk.Frame(self.login_frame); btn_frame.pack(pady=5) tk.Button(btn_frame, text=LANG_DICT[self.get_lang()]["clear_a"], command=self.clear_account_a, bg="#c0392b", fg="white", width=8).grid(row=0, column=0, padx=5) tk.Button(btn_frame, text=LANG_DICT[self.get_lang()]["clear_b"], command=self.clear_account_b, bg="#c0392b", fg="white", width=8).grid(row=0, column=1, padx=5) tk.Button(btn_frame, text=LANG_DICT[self.get_lang()]["check_token"], command=self.validate_all_tokens, bg="#27ae60", fg="white", width=10).grid(row=0, column=2, padx=5) tk.Button(self.login_frame, text=" 开启播报系统 ", bg="#FF0000", fg="white", font=("微软雅黑", 14, "bold"), width=25, height=2, command=self.start_app).pack(pady=10) def start_app(self): if self.remember_account.get(): self.save_configs(self.e_a_t.get(), self.e_a_n.get(), self.e_b_t.get(), self.e_b_n.get()) else: self.save_configs("", "", "", "") configs = [] if self.e_a_t.get().strip(): token = self.e_a_t.get().strip() name = self.e_a_n.get().strip() or "A" account_id = self.get_mp_account_id(token) if not account_id: messagebox.showerror("错误", "账号 A Token 无效") return configs.append({"token": token, "name": name, "id": 1, "account_id": account_id}) if self.e_b_t.get().strip(): token = self.e_b_t.get().strip() name = self.e_b_n.get().strip() or "B" account_id = self.get_mp_account_id(token) if not account_id: messagebox.showerror("错误", "账号 B Token 无效") return configs.append({"token": token, "name": name, "id": 2, "account_id": account_id}) if not configs: return self.account_configs = configs self.login_frame.destroy() self.root.geometry("900x750") self.build_monitor_ui() for c in self.account_configs: self.toggle_float(c['id']) Thread(target=self.api_loop, args=(c,), daemon=True).start() def build_monitor_ui(self): L = LANG_DICT[self.get_lang()] self.root.title(L["title"]) f_top = tk.Frame(self.root, pady=10, bg="#E5E7E9"); f_top.pack(fill=tk.X) tk.Label(f_top, text=L["freq"], bg="#E5E7E9", font=("Arial", 10, "bold")).pack(side=tk.LEFT, padx=10) tk.Scale(f_top, from_=2, to=15, orient=tk.HORIZONTAL, variable=self.check_interval, length=250, bg="#E5E7E9", bd=0).pack(side=tk.LEFT) tk.Button(f_top, text=L["settings"], bg="#E67E22", fg="white", font=("Arial", 11, "bold"), width=15, command=self.show_settings).pack(side=tk.RIGHT, padx=20) paned = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL) paned.pack(fill=tk.BOTH, expand=True) self.log_box = scrolledtext.ScrolledText(paned, width=35, font=("Consolas", 8)) paned.add(self.log_box) self.rec_area = tk.Frame(paned) paned.add(self.rec_area) self.views = {} for c in self.account_configs: f = tk.LabelFrame(self.rec_area, text=f" {c['name']} ", fg=("#009EE3" if c['id']==1 else "#9B59B6"), font=("Arial", 10, "bold")) f.pack(fill=tk.BOTH, expand=True, padx=5, pady=2) t = scrolledtext.ScrolledText(f, height=10, font=("Arial", 9)); t.pack(fill=tk.BOTH, expand=True) self.views[c['id']] = t self.total_frame = tk.Frame(self.root, bg="#2C3E50", pady=8) self.total_frame.pack(side=tk.BOTTOM, fill=tk.X) self.total_val_lab = tk.Label( self.total_frame, text=f"{L['total_label']}0", fg="#F1C40F", bg="#2C3E50", font=("Verdana", 20, "bold") ) self.total_val_lab.pack(side=tk.LEFT, padx=20) tk.Button( self.total_frame, text="📂 打开今日账单", command=self.open_today_bill, bg="#FF9800", fg="white", font=("微软雅黑", 10, "bold"), relief=tk.FLAT, padx=10 ).pack(side=tk.RIGHT, padx=20, pady=5) bottom = tk.Frame(self.root, bg="#F0F0F0", height=35) bottom.pack(side=tk.BOTTOM, fill=tk.X) self.led = tk.Canvas(bottom, width=15, height=15, bg="#F0F0F0", highlightthickness=0) self.led_circle = self.led.create_oval(2, 2, 13, 13, fill="green") self.led.pack(side=tk.LEFT, padx=10) tk.Button(bottom, text=L["reboot"], command=self.hot_reboot, font=("Arial", 8), bg="#D5DBDB").pack(side=tk.LEFT, padx=5) self.status_lab = tk.Label(bottom, text=L["status_ok"], font=("Arial", 9), bg="#F0F0F0") self.status_lab.pack(side=tk.LEFT, padx=10) def open_today_bill(self): today = datetime.now().strftime("%Y.%m.%d") file_path = os.path.join(RECORDS_DIR, f"{today}.txt") if os.path.exists(file_path): os.startfile(file_path) else: messagebox.showinfo("提示", "今日尚未产生账单记录") def api_loop(self, config): token, name, aid, my_account_id = config["token"], config["name"], config["id"], config["account_id"] headers = {'Authorization': f'Bearer {token}'} first = True startup_time = self.startup_time # 获取程序启动时间 def _log(msg): """同时输出到控制台和 GUI 日志框""" print(msg) try: self.log_box.insert(tk.END, msg + "\n") self.log_box.see(tk.END) except Exception: pass while self.is_running: try: r = requests.get( "https://api.mercadopago.com/v1/payments/search", headers=headers, params={'sort':'date_created','criteria':'desc','limit':20}, timeout=10 ) if r.status_code != 200: self.set_led(False, f"HTTP {r.status_code}") time.sleep(self.check_interval.get()) continue self.set_led(True) payments = r.json().get('results', []) if first: # 初始化：记录已有的ID，不播报历史交易 for p in payments: self.processed_ids.add(p['id']) first = False _log(f"[INFO] 初始化完成({name})，忽略 {len(payments)} 条历史记录") time.sleep(self.check_interval.get()) continue for p in reversed(payments): pid = p['id'] if p.get('status') != 'approved' or pid in self.processed_ids: continue self.processed_ids.add(pid) # ===== 时间过滤：只跳过明显的旧交易 ===== # 给 5 分钟缓冲，避免因为以下原因误杀刚发生的交易： # 1) MP 服务器与本机有秒级时差 # 2) 交易创建到 API 可见有 10~30 秒延迟 # 3) 用户刚启动程序就让朋友转账 # 已经有 processed_ids 防重复播报，这里只是再加一道"真旧交易"的保险 TIME_FILTER_BUFFER_SEC = 300 # 5 分钟 date_created_str = p.get('date_created', '') if date_created_str: try: # MP 时间格式: "2026-05-11T07:53:29.000-04:00" # 用 fromisoformat 直接解析带时区的时间，避免硬编码 -04:00 iso_str = date_created_str # Python 3.10 之前 fromisoformat 不支持 Z 后缀，做个兼容 if iso_str.endswith('Z'): iso_str = iso_str[:-1] + '+00:00' trans_time = datetime.fromisoformat(iso_str).timestamp() if trans_time < startup_time - TIME_FILTER_BUFFER_SEC: _log(f"[跳过-历史交易] pid={pid} time={date_created_str} " f"(早于启动 {int(startup_time - trans_time)}秒)") continue except Exception as e: _log(f"[WARN] 时间解析失败: {e}, 原始值={date_created_str}") # ===== 取金额 ===== amt = float(p.get('transaction_amount', 0) or 0) if amt <= 0: _log(f"[跳过-金额非正] pid={pid} amt={amt}") continue # ===== 保存完整原始 JSON 到 txt 文件（debug 用）===== # 文件位置: MegaMonitor_Data/debug_payments.txt try: debug_file = os.path.join(APP_DATA_DIR, "debug_payments.txt") now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(debug_file, "a", encoding="utf-8") as f: f.write("=" * 60 + "\n") f.write(f"[{now_str}] account={name} pid={pid} my_id={my_account_id}\n") f.write("=" * 60 + "\n") f.write(json.dumps(p, ensure_ascii=False, indent=2, default=str)) f.write("\n\n") except Exception as e: _log(f"[WARN] 写 debug 文件失败: {e}") # ===== 收/支判定 ===== # MercadoPago Payment 字段说明： # payer.id → 嵌套字段（付款人ID） # collector_id → 顶层字段（收款人ID） ★ 不是 collector.id # operation_type → regular_payment / money_transfer / transfer / payout / ... operation_type = p.get('operation_type', '') payer_obj = p.get('payer') or {} payer_id = str(payer_obj.get('id', '') or '') collector_id = str(p.get('collector_id', '') or '') receiver_id_str = str(p.get('receiver_id', '') or '') real_collector_id = collector_id or receiver_id_str EXPENSE_OPS = ('payout', 'money_out', 'withdrawal') # ★ 关键：先看 collector_id（钱进谁口袋），再看 payer_id（钱从哪来） # MP 在"自己充值(account_fund)"的场景里，payer.id 和 collector_id # 都是用户自己的 MP ID（payer 是绑定的 CVU 银行账户，归属同一个 MP 用户）， # 如果先判 payer 会把"自己充值"误杀成"我是付款人"。 if real_collector_id and real_collector_id == my_account_id: # 钱进了我钱包 → 不管 payer 是谁都播报 # 涵盖：别人转我、自己充值(account_fund)、银行转账入账 is_my_income = True debug_tag = "[播报-我是收款人]" elif payer_id and payer_id == my_account_id: # 我是付款人但 collector 不是我 → 我转出去给别人，不播报 is_my_income = False debug_tag = "[跳过-我转给别人]" elif operation_type in EXPENSE_OPS: is_my_income = False debug_tag = "[跳过-支出类op]" else: # search 接口语义就是查"我相关"的支付，兜底按收入处理 is_my_income = True debug_tag = "[播报-默认入账]" _log(f"{debug_tag} acct={name} pid={pid} op={operation_type} " f"payer={payer_id} collector={real_collector_id} " f"my_id={my_account_id} amt={amt}") if not is_my_income: continue # ===== 触发播报 ===== payer_email = payer_obj.get('email', '') or "User" self.root.after(0, self.on_income, aid, name, amt, payer_email, False) except Exception as e: self.set_led(False, str(e)) _log(f"[ERROR] api_loop 异常({name}): {e}") time.sleep(self.check_interval.get()) def on_income(self, aid, name, raw_amt, payer, is_test=False): L = LANG_DICT[self.get_lang()] now = datetime.now() t_str, d_str = now.strftime("%H:%M:%S"), now.strftime("%Y.%m.%d") abs_amt = abs(raw_amt) cn_money = cn_amount(abs_amt) if not is_test: if d_str != self.current_date: self.today_total = 0.0 self.current_date = d_str self.today_total += abs_amt self.total_val_lab.config(text=f"{L['total_label']}{format_amt_display(self.today_total)}") self.last_income_record[name] = { "amount": abs_amt, "time": time.time(), "payer": payer } display_amt = format_amt_display(abs_amt) voice_txt = "" current_lang = self.get_lang() if current_lang == "zh": prefix = "这是测试，" if is_test else "" voice_txt = f"{prefix}{name}收款成功{cn_money}比索" if self.speech_mode.get() == "with_name" else f"{prefix}收款成功{cn_money}比索" else: prefix = "Prueba: " if is_test else "" voice_txt = f"{prefix}Cuenta {name}, recibido {abs_amt} pesos" if self.speech_mode.get() == "with_name" else f"{prefix}Recibido {abs_amt} pesos" if AUDIO_AVAILABLE: v_key = self.voice_a.get() if aid == 1 else self.voice_b.get() real_voice = VOICE_MAP[current_lang][v_key] self.speech_queue.put((voice_txt, real_voice, self.speech_rate.get())) type_prefix = L["income"] test_tag = "【测试】" if is_test else "" self.views[aid].insert(tk.END, f"[{t_str}] {test_tag}{type_prefix} ${display_amt} | {payer}\n") self.views[aid].see(tk.END) self.log_box.insert(tk.END, f"[{t_str}] {test_tag}{name} {type_prefix} ${display_amt}\n") self.log_box.see(tk.END) if aid in self.float_wins: d = self.float_wins[aid] d['data'].insert(0, f"{t_str} {test_tag}{L['received_success']} ${display_amt}") d['lbl'].config(text="\n".join(d['data'][:5])) d['total_lbl'].config(text=f"今日: ${format_amt_display(self.today_total)}") BigNotification(self.root, abs_amt, payer, current_lang, is_income=True, is_test=is_test) try: with open(os.path.join(RECORDS_DIR, f"{d_str}.txt"), "a", encoding="utf-8") as f: f.write(f"[{t_str}] {test_tag}账号:{name} | 类型:{type_prefix} | 金额:${display_amt} | 付款人:{payer}\n") except: pass def get_available_balance(self, token): try: r = requests.get("https://api.mercadopago.com/users/me", headers={"Authorization": f"Bearer {token}"}, timeout=8) if r.status_code == 200: return float(r.json().get("available_balance", 0)) except: return None def on_balance_drop(self, amount, account_name): L = LANG_DICT[self.get_lang()] current_lang = self.get_lang() if current_lang == "zh": voice_txt = f"这是测试，{account_name}账号有笔付款被取消，请注意查看" else: voice_txt = f"Prueba: Cuenta {account_name}, se canceló un pago, revisá por favor" if AUDIO_AVAILABLE: v_key = self.voice_a.get() if account_name == self.account_configs[0]["name"] else self.voice_b.get() real_voice = VOICE_MAP[current_lang][v_key] self.speech_queue.put((voice_txt, real_voice, self.speech_rate.get())) self.log_box.insert(tk.END, f"[⚠ 疑似恶意取消] {account_name} 支出 ${format_amt_display(amount)}\n") self.log_box.see(tk.END) BigNotification( self.root, amount, account_name, current_lang, is_income=False, is_test=True ) def speak_test(self): if self.speech_timer: self.root.after_cancel(self.speech_timer) self.speech_timer = self.root.after( 200, self._play_test_direct ) def _play_test_direct(self): """直接播放当前语速的测试语音（用于滑块拖拽时）""" try: rate = int(self.speech_rate.get()) except: rate = 0 current_lang = self.get_lang() if current_lang == "zh": text = "这是语速测试，收款成功一千五百比索" else: text = "Esto es una prueba de velocidad, recibido 1500 pesos" if pygame.mixer.music.get_busy(): pygame.mixer.music.stop() pygame.mixer.music.unload() Thread( target=self._speak_thread_direct, args=(text, rate), daemon=True ).start() def _speak_thread_direct(self, text, rate): """直接播放语音，不经过队列""" try: current_lang = self.get_lang() voice = VOICE_MAP[current_lang][self.voice_a.get()] r_str = f"{'+' if rate>=0 else ''}{rate}%" audio_file = os.path.join(CACHE_DIR, f"test_{uuid.uuid4().hex}.mp3") async def generate_audio(): await edge_tts.Communicate(text, voice, rate=r_str).save(audio_file) asyncio.run(generate_audio()) pygame.mixer.music.load(audio_file) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): time.sleep(0.1) except Exception as e: print(f"语音播放错误: {e}") finally: try: if os.path.exists(audio_file): os.remove(audio_file) except: pass def simulate_income(self): if not self.account_configs: messagebox.showwarning("提示", "请先配置至少一个账号") return cfg = self.account_configs[0] self.on_income( aid=cfg["id"], name=cfg["name"], raw_amt=1500, payer="test_user@megamonitor.com", is_test=True ) def simulate_cancel(self): if not self.account_configs: messagebox.showwarning("提示", "请先配置至少一个账号") return cfg = self.account_configs[0] self.on_balance_drop( amount=1500, account_name=cfg["name"] ) def _speak_thread(self, text, rate): try: current_lang = self.get_lang() voice = VOICE_MAP[current_lang][self.voice_a.get()] r_str = f"{'+' if rate>=0 else ''}{rate}%" async def g(): f = os.path.join(CACHE_DIR, "test_speed.mp3") await edge_tts.Communicate(text, voice, rate=r_str).save(f) pygame.mixer.music.load(f) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): await asyncio.sleep(0.1) asyncio.run(g()) except: pass def _speech_worker(self): while True: try: msg, voice, rate = self.speech_queue.get() if not AUDIO_AVAILABLE: self.speech_queue.task_done() continue r_str = f"{'+' if rate>=0 else ''}{rate}%" h = hashlib.md5(f"{msg}{voice}{rate}".encode()).hexdigest() f = os.path.join(CACHE_DIR, f"{h}.mp3") if not os.path.exists(f): async def g(): await edge_tts.Communicate(msg, voice, rate=r_str).save(f) asyncio.run(g()) while pygame.mixer.music.get_busy(): time.sleep(0.1) pygame.mixer.music.load(f) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): time.sleep(0.1) pygame.mixer.music.unload() self.speech_queue.task_done() except: time.sleep(1) def show_settings(self): L = LANG_DICT[self.get_lang()] win = tk.Toplevel(self.root) win.title(L["settings"]) win.geometry("620x520") win.attributes("-topmost", True) main_frame = tk.Frame(win) main_frame.pack(fill=tk.BOTH, expand=True, padx=15, pady=15) col1 = tk.Frame(main_frame) col1.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) col2 = tk.Frame(main_frame) col2.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True) tk.Label(col1, text=f"【{L['float_title']}】", font=("Arial", 11, "bold")).pack(anchor=tk.W, pady=(0,8)) tk.Checkbutton(col1, text=L["float_a"], variable=self.show_float_a, command=lambda: self.toggle_float(1)).pack(anchor=tk.W) tk.Checkbutton(col1, text=L["float_b"], variable=self.show_float_b, command=lambda: self.toggle_float(2)).pack(anchor=tk.W) tk.Label(col1, text=L["float_w"]).pack(anchor=tk.W) tk.Scale(col1, from_=200, to=800, orient=tk.HORIZONTAL, variable=self.float_width, command=self._rescale).pack(fill=tk.X) tk.Label(col1, text=L["float_h"]).pack(anchor=tk.W) tk.Scale(col1, from_=80, to=400, orient=tk.HORIZONTAL, variable=self.float_height, command=self._rescale).pack(fill=tk.X) tk.Label(col1, text=L["float_alpha"]).pack(anchor=tk.W) tk.Scale(col1, from_=20, to=100, orient=tk.HORIZONTAL, variable=self.float_alpha, command=self.update_float_alpha).pack(fill=tk.X) tk.Label(col1, text=L["font_size_main"]).pack(anchor=tk.W) tk.Scale(col1, from_=8, to=20, orient=tk.HORIZONTAL, variable=self.font_size_main, command=lambda x:self.refresh_float_font()).pack(fill=tk.X) tk.Label(col1, text=L["font_size_total"]).pack(anchor=tk.W) tk.Scale(col1, from_=10, to=24, orient=tk.HORIZONTAL, variable=self.font_size_total, command=lambda x:self.refresh_float_font()).pack(fill=tk.X) c_frame = tk.Frame(col1); c_frame.pack(anchor=tk.W, pady=5) tk.Button(c_frame, text=L["color_main"], command=self.pick_color_main, bg="#444", fg="white", width=10).grid(row=0,column=0,padx=3) tk.Button(c_frame, text=L["color_total"], command=self.pick_color_total, bg="#444", fg="white", width=10).grid(row=0,column=1,padx=3) tk.Label(col2, text=f"【{L['speech_title']}】", font=("Arial", 11, "bold")).pack(anchor=tk.W, pady=(0,8)) tk.Label(col2, text=L["lang_sel"]).pack(anchor=tk.W) lang_cb = ttk.Combobox(col2, textvariable=self.lang, state="readonly", width=22) lang_cb['values'] = ["zh", "es"] lang_cb.pack(anchor=tk.W) tk.Label(col2, text=L["speech_mode"]).pack(anchor=tk.W) mode_cb = ttk.Combobox(col2, textvariable=self.speech_mode, state="readonly", width=22) mode_cb['values'] = [L["mode_with_name"], L["mode_no_name"]] mode_cb.pack(anchor=tk.W) tk.Label(col2, text=L["speed"]).pack(anchor=tk.W) def on_speech_rate_change(event=None): self._play_test_direct() speech_scale = tk.Scale( col2, from_=-50, to=50, orient=tk.HORIZONTAL, variable=self.speech_rate ) speech_scale.bind("", on_speech_rate_change) speech_scale.pack(fill=tk.X) tk.Label(col2, text="恶意取消监测窗口(分钟):").pack(anchor=tk.W, pady=(15, 0)) tk.Scale( col2, from_=1, to=120, orient=tk.HORIZONTAL, variable=self.fraud_window ).pack(fill=tk.X, pady=(0, 10)) btn_frame = tk.Frame(col2) btn_frame.pack(pady=10) tk.Button( btn_frame, text="测试收款", command=self.simulate_income, bg="#009EE3", fg="white", font=("微软雅黑", 10, "bold"), width=12 ).pack(side=tk.LEFT, padx=5) tk.Button( btn_frame, text="测试恶意取消", command=self.simulate_cancel, bg="#E74C3C", fg="white", font=("微软雅黑", 10, "bold"), width=12 ).pack(side=tk.LEFT, padx=5) def update_float_alpha(self, *args): val = self.float_alpha.get() / 100 for d in self.float_wins.values(): d['win'].attributes("-alpha", val) def _rescale(self, _=None): for aid, d in self.float_wins.items(): d['win'].geometry(f"{self.float_width.get()}x{self.float_height.get()}") def toggle_float(self, aid): cfg = next((c for c in self.account_configs if c['id'] == aid), None) if not cfg: return show = self.show_float_a.get() if aid == 1 else self.show_float_b.get() if show and aid not in self.float_wins: fw = tk.Toplevel() fw.overrideredirect(True) fw.attributes("-alpha", self.float_alpha.get() / 100, "-topmost", True) fw.configure(bg='#1A1A1A') fw.geometry(f"{self.float_width.get()}x{self.float_height.get()}+{50 + (aid-1)*450}+100") top_frame = tk.Frame(fw, bg="#1A1A1A") top_frame.pack(fill=tk.X) tk.Label(top_frame, text=f"● {cfg['name']}", fg="#00FF00", bg="#1A1A1A", font=("Arial", 9, "bold")).pack(side=tk.LEFT, padx=5) total_lbl = tk.Label(top_frame, text=f"今日: $0", fg=self.color_total.get(), bg="#1A1A1A", font=("Arial", self.font_size_total.get(), "bold")) total_lbl.pack(side=tk.RIGHT, padx=8) lbl = tk.Label(fw, text=LANG_DICT[self.get_lang()]["wait"], fg=self.color_main.get(), bg="#1A1A1A", font=("Arial", self.font_size_main.get()), justify=tk.LEFT) lbl.pack(fill=tk.BOTH, expand=True, padx=10) self.float_wins[aid] = {"win": fw, "lbl": lbl, "data": [], "total_lbl": total_lbl} def b1(e): fw.x, fw.y = e.x, e.y def b2(e): fw.geometry(f"+{fw.winfo_x()+(e.x-fw.x)}+{fw.winfo_y()+(e.y-fw.y)}") fw.bind("", b1) fw.bind("", b2) elif not show and aid in self.float_wins: self.float_wins[aid]['win'].destroy() del self.float_wins[aid] def set_led(self, ok, msg=""): self.led.itemconfig(self.led_circle, fill="green" if ok else "red") self.status_lab.config(text=LANG_DICT[self.get_lang()]["status_ok"] if ok else f"Err: {msg[:15]}") def hot_reboot(self): self.is_running = False self.root.destroy() os.execl(sys.executable, sys.executable, *sys.argv) def load_configs(self): try: with open(self.config_file, "r", encoding="utf-8") as f: d = json.load(f) return [d[0], d[1], {"lang": d[2].get("lang", "zh")}] except: return [{"token":"", "name":""}, {"token":"", "name":""}, {"lang": "zh"}] def save_configs(self, t1=None, n1=None, t2=None, n2=None): data = [ {"token": t1 or self.e_a_t.get(), "name": n1 or self.e_a_n.get()}, {"token": t2 or self.e_b_t.get(), "name": n2 or self.e_b_n.get()}, {"lang": self.lang.get()} ] with open(self.config_file, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) if __name__ == "__main__": app = MegaMonitorApp() app.root.mainloop()