import configparser import ctypes import json import os import platform import hashlib import random import shutil import subprocess import sys import threading import time import traceback import uuid import logging from pathlib import Path import cv2 import numpy as np import paddle import pyautogui import pydirectinput import pyperclip import pyttsx3 import requests from PIL import Image, ImageDraw, ImageFont from paddleocr import PaddleOCR # 当前版本号 VERSION = "2.1.9" # 客户端升级API UPDATE_API = "https://www.liminqi.top/api/client/update" def check_for_updates(): """ 检查是否有新版本可用 """ try: print(f"正在检查更新... 当前版本: {VERSION}") # 发送POST请求，包含客户端版本号 data = {"client_version": VERSION} response = requests.post(UPDATE_API, json=data, timeout=10) if response.status_code == 200: data = response.json() if data.get("success"): download_url = data.get("download_url") if download_url: print(f"发现新版本，下载地址: {download_url}") return True, download_url else: print(f"API返回错误: {data.get('message', '未知错误')}") else: print(f"检查更新失败，HTTP状态码: {response.status_code}") except requests.exceptions.RequestException as e: print(f"网络错误，检查更新失败: {e}") except Exception as e: print(f"检查更新时发生错误: {e}") return False, None def download_and_replace(file_url, local_path): """ 下载新版本文件并替换当前文件 """ try: print(f"正在下载新版本文件: {file_url}") # 下载文件 response = requests.get(file_url, timeout=30) if response.status_code == 200: # 保存到临时文件 temp_file = local_path + ".tmp" with open(temp_file, 'wb') as f: f.write(response.content) print("下载完成，正在替换文件...") # 替换当前文件 if os.path.exists(local_path): os.remove(local_path) os.rename(temp_file, local_path) print("文件替换成功") return True else: print(f"下载失败，HTTP状态码: {response.status_code}") except Exception as e: print(f"下载或替换文件时发生错误: {e}") return False def restart_program(): """ 重启当前程序 """ try: print("正在重启程序...") # 获取当前Python解释器和脚本路径 python_exe = sys.executable script_path = os.path.abspath(__file__) # 启动新进程 subprocess.Popen([python_exe, script_path]) # 退出当前进程 sys.exit(0) except Exception as e: print(f"重启程序时发生错误: {e}") def auto_update(): """ 自动升级检查流程 返回: True 如果进行了升级并重启，False 如果没有升级或升级失败 """ print("=== 自动升级检查 ===") print(f"当前版本: {VERSION}") # 检查更新 has_update, download_url = check_for_updates() if has_update and download_url: print("开始升级...") # 下载并替换文件 local_path = os.path.abspath(__file__) if download_and_replace(download_url, local_path): print("升级成功，准备重启...") # 重启程序 restart_program() return True else: print("升级失败，继续使用当前版本") return False # 尝试导入socketio-client try: import socketio SOCKETIO_AVAILABLE = True except ImportError as e: print("⚠️ 未安装socketio-client，WebSocket功能将不可用") print("请运行: pip install python-socketio[client]") SOCKETIO_AVAILABLE = False raise e # 设置GPU设备（如果可用） try: paddle.set_device('gpu') print("使用GPU加速") except Exception as e: print("使用CPU运行") paddle.set_device('cpu') def check_admin_privileges(): """ 检查管理员权限，如果没有则提示用户 """ try: is_admin = ctypes.windll.shell32.IsUserAnAdmin() except Exception as e: print(f"检查管理员权限时出错: {str(e)}") is_admin = False if not is_admin: print("当前未以管理员权限运行，部分功能可能无法正常工作") class ScreenMonitor: """ 屏幕监控类，负责监控游戏屏幕、OCR识别文本、语音播报等功能 """ def __init__(self, x, y, width, height): """ 初始化屏幕监控器 Args: x (int): 监控区域左上角x坐标 y (int): 监控区域左上角y坐标 width (int): 监控区域宽度 height (int): 监控区域高度 """ self.system_id = None self.blacklist_songs = None self.config = None self.vip_players = None self.playmusic = False self.base_dir = Path(__file__).parent self.region = (x, y, width, height) self.running = False self.last_spoken_text = "" self.cooldown_time = 5 # 语音播报冷却 # WebSocket相关配置 self.sio = None self.websocket_connected = False self.heartbeat_thread = None self.heartbeat_running = False self.heartbeat_interval = 30 # 心跳包发送间隔（秒） self.server_url = "wss://www.liminqi.top" # 服务器地址 self.current_task = "空闲" # 当前任务状态 # 新增：连接管理锁和状态控制 self.connection_lock = threading.Lock() # 连接操作锁 self.is_connecting = False # 是否正在连接中 self.last_connection_check = 0 # 上次连接检查时间 self.connection_check_interval = 10 # 连接检查间隔（秒） self.reconnect_thread = None # 重连线程 self.reconnect_running = False # 重连线程运行状态 self.last_connection_attempt = 0 # 记录最近连接尝试的时间戳 # 新增：心跳包响应时间记录 self.last_heartbeat_response = 0 # 上次收到心跳包响应的时间 self.heartbeat_timeout = 60 # 心跳包超时时间（秒） self.last_recognized_text = "" # 关键词播报记录（防止重复播报） self.last_announced_for_11 = "" # "11"关键词上次播报内容 self.last_announced_for_00 = "" # "00"关键词上次播报内容 self.last_announced_for_naimei = "" # "奶妈"关键词上次播报内容 self.last_announced_for_qiufan = "" # "囚犯"关键词上次播报内容 self.last_announced_for_help = "" # "求助"关键词上次播报内容 self.last_announced_for_vip = "" # VIP玩家上次播报内容 self.last_query_content = "" # 记录上次查询的内容 self._last_speak_attempt = time.time() self._last_special_announcement = time.time() # 添加用于记录小文API调用的变量 self.tykey = "" # 通译灵码key self.last_xiaowen_content = "" # 记录上次发送给小文API的内容 self.xiaowen_cooldown_time = 12 # 小文API调用冷却时间（秒） self._last_xiaowen_call = 0 # 上次调用小文API的时间 # 音乐控制 self.musickey = "" # 音乐APIkey self.current_music_duration = 0 # 当前播放音乐的时长（秒） self.music_start_time = 0 # 音乐开始播放时间 self.playmusic = False # 点歌功能开关状态 self.is_playing_music = False # 当前是否正在播放歌曲 self.current_task = "空闲" self._last_music_request = "" # 记录上次的点歌请求 self.audio_operation_lock = threading.Lock() # 清理线程锁 self.is_downloading = False # 标记是否正在下载文件 # 点歌队列 self.music_queue = [] # 点歌队列 self.queue_processing = False # 队列处理状态 # F键自动按键线程控制 self.f_key_thread = None self.f_key_running = False self.f_key_interval = 5 # 每5秒按一次F键 self.f_key_lock = threading.Lock() # F键操作锁 self.is_sending_message = False # 标记是否正在发送消息 # 添加wav_filepath属性的初始化 self.temp_audio_dir = "temp_audio" # 临时音频文件夹 self.mp3_dir = self.base_dir / "MP3" self._setup_temp_dir() # 初始化临时文件夹 self.user_name = "" # 玩家名 self.config_path = self.base_dir / "config.ini" # 初始化 PaddleOCR self.ocr = PaddleOCR(use_doc_orientation_classify=True, use_doc_unwarping=False, use_textline_orientation=False, text_detection_model_name="PP-OCRv5_mobile_det", text_recognition_model_name="PP-OCRv5_server_rec", text_detection_model_dir="model/det", # 替换为您的检测模型路径 # text_recognition_model_dir="model/rec", # 替换为您的识别模型路径 ) # 通义实验室角色扮演API配置 self.api_key = self.tykey # 请替换为您的实际API Key self.service_name = "aca-chat-send" self.url = "https://nlp.aliyuncs.com/v2/api/chat/send" self.headers = { "Content-Type": "application/json", "X-AcA-DataInspection": "enable", "x-fag-servicename": self.service_name, "x-fag-appcode": "aca", "Authorization": f"Bearer {self.api_key}" } # 添加语音播报时间记录 self._last_speak_attempt = time.time() - self.cooldown_time self._last_special_announcement = time.time() # 记录特殊提示的播报时间 self.special_announcement_interval = 300 # 特殊提示间隔时间（秒） # 定时空格 self.space_pressed_today = False # 标记今天是否已经按过空格键 # 加载中文字体 self.font_path = "simhei.ttf" # 黑体字体文件 # 如果没有字体文件，尝试使用系统默认字体 try: self.font = ImageFont.truetype("simhei.ttf", 18) except: try: # 尝试使用系统默认中文字体 self.font = ImageFont.truetype("C:/Windows/Fonts/msyh.ttc", 18) except: # 如果都失败，使用默认字体（可能仍显示为?） self.font = ImageFont.load_default() def ensure_mp3_dir(self): """确保MP3目录存在""" if not self.mp3_dir.exists(): self.mp3_dir.mkdir() print("✅ 创建MP3目录") else: print("✅ MP3目录已存在") def get_system_uuid(self): """获取系统UUID""" try: # 获取系统UUID（如果可用） system_uuid = uuid.getnode() self.system_id = str(system_uuid) except: # 如果无法获取系统UUID，生成一个基于硬件的UUID self.system_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, platform.node())) print(f"📋 系统UUID: {self.system_id}") def submit_robot_mapping(self, fid): """向服务器提交robot_id和fid的映射关系""" try: # 检查是否已获取系统UUID if not hasattr(self, 'system_id') or not self.system_id: self.get_system_uuid() robot_id = self.system_id print(f"🔗 提交robot_id和fid映射: robot_id={robot_id}, fid={fid}") # 构建请求数据 data = { 'robot_id': robot_id, 'fid': fid } # 发送POST请求到服务器 url = "https://www.liminqi.top/api/robot/mapping" response = requests.post(url, json=data, timeout=30) if response.status_code == 200: result = response.json() if result.get('success'): action = result.get('action', 'unknown') print(f"✅ robot_id和fid映射{action}成功") return True else: error_msg = result.get('message', '未知错误') logger.error(f"❌ robot_id和fid映射失败: {error_msg}") return False else: logger.error(f"❌ 服务器响应错误: HTTP {response.status_code}") try: error_data = response.json() error_msg = error_data.get('message', '未知错误') logger.error(f"❌ 服务器错误详情: {error_msg}") except: print(f"❌ 响应内容: {response.text[:100]}...") return False except requests.exceptions.ConnectionError: logger.error("❌ 连接服务器失败，请检查网络连接") return False except requests.exceptions.Timeout: logger.error("❌ 请求超时，请检查服务器状态") return False except Exception as e: logger.error(f"❌ 提交robot mapping失败: {e}") return False def load_or_create_config(self): """加载或创建配置文件""" self.config = configparser.ConfigParser() if self.config_path.exists(): self.config.read(self.config_path, encoding='utf-8') print("✅ 配置文件加载成功") # 从配置文件读取VIP玩家列表 try: vip_players_str = self.config.get('VIP', 'players') self.vip_players = [name.strip() for name in vip_players_str.split(',')] except (configparser.NoSectionError, configparser.NoOptionError) as e: # 如果配置文件不存在或格式错误，使用默认列表 self.vip_players = [] # 从配置文件读取黑名单歌曲列表 try: blacklist_songs_str = self.config.get('MUSIC', 'blacklist_songs') self.blacklist_songs = [song.strip().lower() for song in blacklist_songs_str.split(',')] print(f"已加载黑名单歌曲: {len(self.blacklist_songs)}首") except (configparser.NoSectionError, configparser.NoOptionError) as e: # 默认黑名单歌曲 self.blacklist_songs = ["几度梦回大唐", "眼保健操", "舞动青春"] # 从配置文件读取通义灵码和音乐APIkey try: tykey = self.config.get('CONFIG', 'tykey') musickey = self.config.get('CONFIG', 'musickey') self.musickey = musickey self.tykey = tykey # 更新API配置 self.api_key = self.tykey self.headers = { "Content-Type": "application/json", "X-AcA-DataInspection": "enable", "x-fag-servicename": self.service_name, "x-fag-appcode": "aca", "Authorization": f"Bearer {self.api_key}" } except (configparser.NoSectionError, configparser.NoOptionError) as e: self.musickey = "" # 音乐APIkey self.tykey = "" # 通义灵码key logger.error(f"❌ 配置文件{e}，请检查tykey和musickey") # 点歌开关检测 try: self.playmusic = self.config.getboolean('CONFIG', 'playmusic') except (configparser.NoSectionError, configparser.NoOptionError, ValueError) as e: # 默认关闭音乐播放 self.playmusic = False if self.playmusic: print("🎶 点歌功能已开启") else: print("🎶 点歌功能已关闭") else: print("📄 创建默认配置文件") self.config['CONFIG'] = { 'fid': '00000000000' } with open(self.config_path, 'w', encoding='utf-8') as f: self.config.write(f) return self.config def check_fid(self, config): """检查FID配置""" try: fid = self.config.get('CONFIG', 'fid') print(f"📋 当前FID: {fid}") if fid == '00000000000': logger.warning("⚠️ 请手动修改config.ini中的fid为你的实际FID") return False else: print("✅ FID配置正确") return True except: logger.error("❌ 配置文件格式错误") return False def download_server_config(self, fid): """从服务器下载配置文件并替换本地""" try: url = f"https://www.liminqi.top/rebot?fid={fid}" print(f"🌐 下载云端配置: {url}") response = requests.get(url, timeout=30) # 检查响应内容类型 content_type = response.headers.get('content-type', '') if 'application/json' in content_type: # 服务器返回JSON错误消息 try: error_data = response.json() error_msg = error_data.get('message', '未知错误') print(f"❌ 服务器返回: {error_msg}") except: print(f"❌ 服务器返回JSON解析失败: {response.text}") return False elif response.status_code == 200: # 保存服务器配置到本地 with open(self.config_path, 'w', encoding='utf-8') as f: f.write(response.text) print("✅ 云端配置已替换本地配置") return True else: # 处理其他HTTP错误状态 print(f"❌ 下载配置失败: HTTP {response.status_code}") try: # 尝试解析可能的JSON错误消息 error_data = response.json() error_msg = error_data.get('message', '未知错误') print(f"❌ 服务器错误详情: {error_msg}") except: print(f"❌ 响应内容: {response.text[:100]}...") return False except Exception as e: print(f"❌ 连接服务器失败: {e}") return False def download_missing_mp3(self, config): """对比VIP名字下载缺失的MP3，并检查已存在文件的哈希一致性""" try: # 使用传入的config参数，而不是self.config if config.has_section('CONFIG') and config.has_option('CONFIG', 'fid'): fid = config.get('CONFIG', 'fid') else: print("❌ 配置文件中缺少fid配置") return # 读取VIP玩家名字 if config.has_section('VIP') and config.has_option('VIP', 'players'): vip_players = config.get('VIP', 'players').split(',') print(f"👥 VIP玩家: {len(vip_players)} 个") # 检查本地MP3文件 local_mp3 = {f.stem for f in self.mp3_dir.glob('*.mp3')} print(f"🎵 本地MP3: {len(local_mp3)} 个") # 处理所有VIP玩家 for player in vip_players: player = player.strip() if not player: continue # 检查文件是否存在 if player not in local_mp3: # 文件不存在，直接下载 print(f"📥 下载缺失文件: {player}.mp3") self.download_single_mp3(fid, player) else: # 文件存在，检查哈希一致性 if not self.check_mp3_hash_consistency(fid, player): print(f"🔄 文件不一致，重新下载: {player}.mp3") self.download_single_mp3(fid, player) else: print(f"✅ 文件一致: {player}.mp3") print("✅ MP3文件检查完成") else: print("❌ 配置文件中缺少VIP玩家列表") except Exception as e: print(f"❌ 下载MP3时出错: {e}") def check_mp3_hash_consistency(self, fid, player_name): """检查本地MP3文件与服务器端是否一致""" try: # 查询服务器端文件哈希值 server_hash_info = self.get_mp3_file_hash(fid, player_name) if not server_hash_info: print(f"❌ 无法获取服务器端 {player_name}.mp3 的哈希信息") return False # 计算本地文件哈希值 local_file_path = self.mp3_dir / f"{player_name}.mp3" try: with open(local_file_path, 'rb') as f: local_content = f.read() local_hash = hashlib.md5(local_content).hexdigest() local_size = len(local_content) except Exception as e: print(f"❌ 读取本地文件 {player_name}.mp3 失败: {e}") return False # 比较哈希值 if local_hash == server_hash_info['file_hash'] and local_size == server_hash_info['file_size']: return True # 文件一致 else: print(f"🔄 文件不一致: {player_name}.mp3") print(f" 本地: 大小={local_size}, 哈希={local_hash}") print(f" 服务器: 大小={server_hash_info['file_size']}, 哈希={server_hash_info['file_hash']}") return False except Exception as e: print(f"❌ 检查文件 {player_name}.mp3 一致性时出错: {e}") return False def get_mp3_file_hash(self, fid, player_name): """查询服务器端MP3文件的哈希值""" try: # URL编码文件名 encoded_name = requests.utils.quote(player_name) url = f"https://www.liminqi.top/rebot?fid={fid}&name={encoded_name}.mp3&hash=true" response = requests.get(url, timeout=30) # 检查响应内容类型 content_type = response.headers.get('content-type', '') if 'application/json' in content_type and response.status_code == 200: hash_data = response.json() if hash_data.get('success'): return { 'filename': hash_data.get('filename'), 'file_size': hash_data.get('file_size'), 'file_hash': hash_data.get('file_hash') } else: error_msg = hash_data.get('message', '未知错误') print(f"❌ 服务器返回错误: {error_msg}") return None else: print(f"❌ 查询哈希失败: HTTP {response.status_code}") return None except Exception as e: print(f"❌ 查询文件哈希时出错: {player_name}.mp3 - {e}") return None def download_single_mp3(self, fid, player_name): """下载单个MP3文件""" try: # URL编码文件名 encoded_name = requests.utils.quote(player_name) url = f"https://www.liminqi.top/rebot?fid={fid}&name={encoded_name}.mp3" print(f"📥 下载: {player_name}.mp3") response = requests.get(url, timeout=30) # 检查响应内容类型 content_type = response.headers.get('content-type', '') if 'application/json' in content_type: # 服务器返回JSON错误消息 try: error_data = response.json() error_msg = error_data.get('message', '未知错误') print(f"❌ 服务器返回: {error_msg}") except: print(f"❌ 服务器返回JSON解析失败: {response.text}") elif 'audio/mpeg' in content_type and response.status_code == 200: mp3_path = self.mp3_dir / f"{player_name}.mp3" with open(mp3_path, 'wb') as f: f.write(response.content) print(f"✅ 下载完成: {player_name}.mp3") else: print(f"❌ 下载失败: HTTP {response.status_code}") try: # 尝试解析可能的JSON错误消息 error_data = response.json() error_msg = error_data.get('message', '未知错误') print(f"❌ 服务器错误详情: {error_msg}") except: print(f"❌ 响应内容: {response.text[:100]}...") except Exception as e: print(f"❌ 下载错误: {player_name}.mp3 - {e}") def _setup_temp_dir(self): """初始化临时音频文件夹""" if not os.path.exists(self.temp_audio_dir): os.makedirs(self.temp_audio_dir) else: # 清空临时文件夹 self._cleanup_temp_dir() def _cleanup_temp_dir(self): """清空临时音频文件夹""" if os.path.exists(self.temp_audio_dir): for filename in os.listdir(self.temp_audio_dir): file_path = os.path.join(self.temp_audio_dir, filename) try: # 检查是否有锁属性，如果没有则直接删除 if hasattr(self, 'audio_operation_lock'): # 尝试获取锁，如果获取不到说明正在播放音频 if self.audio_operation_lock.acquire(timeout=0): try: if os.path.isfile(file_path): os.remove(file_path) finally: self.audio_operation_lock.release() else: # 获取不到锁，跳过此文件的清理 print(f"跳过清理文件 {file_path}，正在播放音频") else: # 如果没有锁属性，直接尝试删除 if os.path.isfile(file_path): os.remove(file_path) except Exception as e: print(f"清理文件失败: {e}") return True def cleanup_temp_dir_periodically(self): """定期清理临时文件夹""" def periodic_cleanup(): while self.running: time.sleep(21600) # 每6小时钟清理一次 if self.running: # 确保程序仍在运行 self._cleanup_temp_dir() logger.info("已执行定期清理临时音频文件夹") cleanup_thread = threading.Thread(target=periodic_cleanup, daemon=True) cleanup_thread.start() def press_f_key(self): """按F键""" try: pydirectinput.moveTo(828, 431, duration=0.5) time.sleep(0.1) pydirectinput.click() time.sleep(0.1) pydirectinput.press('f') except Exception as e: logger.error(f"按F键失败: {e}") def init_websocket(self): """初始化WebSocket连接""" with self.connection_lock: # 检查是否已经在连接中 if self.is_connecting: logger.info("⚠️ WebSocket连接正在进行中，跳过重复初始化") return False self.is_connecting = True self.last_connection_attempt = time.time() # 记录连接尝试时间 try: # 先检查服务器是否可达 print(f"🔍 检查服务器连接性: {self.server_url}") # 如果已有连接，先正确断开 if self.sio and hasattr(self.sio, 'connected') and self.sio.connected: print("🔌 断开现有WebSocket连接...") try: self.sio.disconnect() time.sleep(1) # 等待断开完成 except Exception as e: print(f"⚠️ 断开连接时出错: {e}") # 配置自动重连参数 - 优化重连设置 self.sio = socketio.Client( reconnection=True, reconnection_attempts=3, # 增加重试次数 reconnection_delay=2, reconnection_delay_max=5, randomization_factor=0.5, ) # 设置事件处理器 @self.sio.event def skip_song_command(data): """处理跳过歌曲命令""" logger.info(f"🎵 收到跳过歌曲命令: {data}") # 修复：调整判断顺序，先检查播放状态，再检查下载状态 # 检查是否正在播放音乐 if not self.is_playing_music: logger.info("⚠️ 当前没有正在播放的歌曲，无需切歌") return # 修复：只有在真正下载时才检查下载状态 # 检查是否正在下载歌曲（只有在当前没有播放音乐时才需要检查） if self.is_downloading and not self.is_playing_music: logger.info("⚠️ 当前正在下载歌曲，无法切歌，请等待下载完成") self.send_message_to_game("正在下载歌曲，无法切歌，请等待下载完成") return # 检查队列中是否有下一首歌曲 if not self.music_queue: logger.info("⚠️ 队列中没有下一首歌曲，无法切歌") self.send_message_to_game("队列已空，无法切歌，请及时点歌~") return # 获取下一首歌曲名称 next_song = self.music_queue[0] logger.info(f"🎵 准备切歌到下一首: {next_song}") # 停止当前播放的歌曲 try: # 停止当前音乐播放 if hasattr(self, 'current_music_duration') and self.current_music_duration > 0: # 发送歌曲结束事件 if hasattr(self, '_last_music_request') and self._last_music_request: self.send_song_end(self._last_music_request) # 修复：不要重置播放状态，保持点歌功能开启 # 只重置时间相关的状态，保持is_playing_music为True self.current_music_duration = 0 self.music_start_time = 0 self._last_music_request = "" logger.info(f"✅ 已停止当前歌曲播放") except Exception as e: logger.error(f"❌ 停止当前歌曲失败: {e}") # 立即开始播放下一首歌曲 try: # 从队列中取出下一首歌曲 song_name = self.music_queue.pop(0) logger.info(f"🎵 开始播放下一首歌曲: {song_name}") # 修复：确保点歌功能状态正确 if not self.playmusic: self.playmusic = True self.current_task = "点歌功能已开启" # 使用线程来处理下一首歌曲，避免阻塞 threading.Thread(target=self._play_song_from_queue, args=(song_name,), daemon=True).start() # 修复：移除重复的消息发送，只在_play_song_from_queue中发送一次 logger.info(f"✅ 切歌完成，正在播放: {song_name}") except Exception as e: logger.error(f"❌ 切歌失败: {e}") self.send_message_to_game("切歌失败，请重试") @self.sio.event def reset_state_command(data): """处理重置状态命令""" logger.info(f"🔄 收到重置状态命令: {data}") # 重新加载云端配置 logger.info("🔄 重置时重新加载云端配置...") try: # 加载当前配置获取FID config = self.load_or_create_config() if config.has_section('CONFIG') and config.has_option('CONFIG', 'fid'): fid = config.get('CONFIG', 'fid') print(f"📋 当前FID: {fid}") # 下载云端配置替换本地 if self.download_server_config(fid): # 重新加载更新后的配置 config = self.load_or_create_config() # 对比VIP名字下载缺失MP3 self.download_missing_mp3(config) print("✅ 云端配置重新加载完成") else: print("⚠️ 云端配置下载失败，使用本地配置") else: logger.warning("⚠️ 配置文件中缺少FID，跳过云端配置加载") except Exception as e: logger.warning(f"❌ 重新加载云端配置时出错: {e}") # 调用reset_state方法重置所有状态 self.reset_state(speak_confirmation=True) logger.info("✅ 状态重置完成") @self.sio.event def connect(): print("✅ WebSocket连接成功") self.websocket_connected = True self.is_connecting = False # 启动心跳包线程（确保只启动一次） if not self.heartbeat_running: self.start_heartbeat() # 等待2秒再发送初始心跳包，给服务器准备时间 threading.Timer(2, self.send_heartbeat).start() @self.sio.event def connect_error(data): print(f"❌ WebSocket连接错误: {data}") self.websocket_connected = False self.is_connecting = False # 延迟重连，避免立即重连 threading.Timer(5, self._safe_reconnect).start() @self.sio.event def disconnect(): print("❌ WebSocket连接断开") self.websocket_connected = False # 延迟重连，避免立即重连 threading.Timer(5, self._safe_reconnect).start() @self.sio.event def robot_status_update(data): """处理机器人状态更新事件""" # 获取robot_id进行验证 robot_id = data.get('robot_id', '') if robot_id and robot_id != self.system_id: return # 不是当前机器人的状态更新，忽略 # print(f"🤖 收到机器人状态更新: {data}") # 更新当前任务状态 current_task = data.get('current_task', '') if current_task: self.current_task = current_task # 修复：优先根据current_task同步playmusic状态 if current_task == '点歌功能已开启': # 只有在状态真正发生变化时才播报 if not self.playmusic: self.playmusic = True self.speak_text("点歌功能已开启") logger.info("🎵 点歌功能已开启，语音播报已触发") elif current_task == '点歌功能已关闭': # 只有在状态真正发生变化时才播报 if self.playmusic: self.playmusic = False self.speak_text("点歌功能已关闭") logger.info("🎵 点歌功能已关闭，语音播报已触发") @self.sio.event def heartbeat_response(data): """处理心跳包响应事件""" robot_id = data.get('robot_id', '') if robot_id and robot_id != self.system_id: return # 不是当前机器人的响应，忽略 # 更新心跳包响应时间 self.last_heartbeat_response = time.time() logger.info(f"💓 收到心跳包响应: {robot_id}") # 连接服务器 - 添加超时和错误处理 print(f"🔗 正在连接WebSocket服务器: {self.server_url}") try: # 设置连接超时 self.sio.connect(self.server_url, wait_timeout=10) print("✅ WebSocket连接请求已发送") return True except Exception as connect_error: print(f"❌ WebSocket连接失败: {str(connect_error)}") self.websocket_connected = False self.is_connecting = False # 延迟重连 threading.Timer(5, self._safe_reconnect).start() return False except Exception as e: logger.error(f"❌ WebSocket初始化失败: {str(e)}") self.is_connecting = False # 延迟重连 threading.Timer(5, self._safe_reconnect).start() return False def _safe_reconnect(self): """安全的重连方法，避免重复重连""" with self.connection_lock: if not self.websocket_connected and not self.is_connecting and hasattr(self, 'running') and self.running: logger.info("🔄 安全重连：启动重连尝试") self._attempt_reconnect() def _attempt_reconnect(self): """尝试重新连接WebSocket - 优化重连逻辑""" # 检查是否已经有重连线程在运行 if self.reconnect_thread and self.reconnect_thread.is_alive(): logger.info("⚠️ 重连线程已在运行，跳过重复重连") return def reconnect_worker(): retry_count = 0 max_retry_delay = 120 # 减少最大重试间隔 self.reconnect_running = True while not self.websocket_connected and hasattr(self, 'running') and self.running: try: retry_count += 1 # 使用指数退避算法，但限制最大重试次数 if retry_count > 10: # 增加最大重试次数 logger.error("⚠️ 重连尝试次数过多，暂停重连") break delay = min(10 * (2 ** (retry_count - 1)), max_retry_delay) # 减少初始延迟 logger.error(f"⏳ 尝试重连WebSocket (第{retry_count}次)，{delay}秒后重试...") time.sleep(delay) if hasattr(self, 'running') and self.running and not self.websocket_connected: logger.info("🔄 正在尝试重新连接WebSocket...") try: # 先彻底清理现有连接 self._cleanup_websocket() # 重新初始化socketio客户端 self.init_websocket() except Exception as e: logger.error(f"❌ 重连尝试失败: {e}") continue except Exception as e: logger.error(f"❌ 重连工作线程异常: {e}") break self.reconnect_running = False # 启动重连线程 self.reconnect_thread = threading.Thread(target=reconnect_worker, daemon=True) self.reconnect_thread.start() def _cleanup_websocket(self): """彻底清理WebSocket连接""" try: # 停止心跳包线程 if self.heartbeat_running: self.heartbeat_running = False if self.heartbeat_thread and self.heartbeat_thread.is_alive(): self.heartbeat_thread.join(timeout=5) # 断开现有连接 if self.sio and hasattr(self.sio, 'connected') and self.sio.connected: try: self.sio.disconnect() except: pass # 重置连接状态 self.websocket_connected = False self.is_connecting = False self.last_heartbeat_response = 0 logger.info("🧹 WebSocket连接已彻底清理") except Exception as e: logger.error(f"❌ 清理WebSocket连接失败: {e}") def start_heartbeat(self): """启动心跳包发送线程""" if not self.websocket_connected: logger.error("❌ WebSocket未连接，无法启动心跳包") return False # 检查是否已经有心跳包线程在运行 if self.heartbeat_running and self.heartbeat_thread and self.heartbeat_thread.is_alive(): logger.info("⚠️ 心跳包线程已在运行，跳过重复启动") return False self.heartbeat_running = True self.heartbeat_thread = threading.Thread(target=self._heartbeat_worker, daemon=True) self.heartbeat_thread.start() logger.info("💓 心跳包线程已启动") return True def _heartbeat_worker(self): """心跳包工作线程""" while self.heartbeat_running and self.websocket_connected: try: self.send_heartbeat() time.sleep(self.heartbeat_interval) except Exception as e: logger.error(f"❌ 心跳包发送失败: {str(e)}") time.sleep(5) # 出错后等待5秒再重试 def send_heartbeat(self): """发送心跳包""" if not self.sio or not self.websocket_connected: return False try: # 简化连接状态检查，主要依赖websocket_connected标志 if not self.websocket_connected: logger.warning("⚠️ WebSocket连接已断开，跳过心跳包发送") return False # 检查心跳包响应超时 current_time = time.time() if (self.last_heartbeat_response > 0 and current_time - self.last_heartbeat_response > self.heartbeat_timeout): logger.warning(f"⚠️ 心跳包响应超时 ({current_time - self.last_heartbeat_response:.1f}秒)，连接可能已失效") self.websocket_connected = False self._attempt_reconnect() return False # 获取系统UUID作为robot_id if not hasattr(self, 'system_id') or not self.system_id: self.get_system_uuid() # 修复：正确区分点歌功能开关状态和当前播放状态 # playmusic_enabled: 点歌功能是否开启（开关状态） # playmusic: 当前是否正在播放歌曲（播放状态） heartbeat_data = { 'robot_id': self.system_id, 'current_task': self.current_task, 'playmusic': self.is_playing_music if hasattr(self, 'is_playing_music') else False, # 当前播放状态 'playmusic_enabled': self.playmusic if hasattr(self, 'playmusic') else False, # 点歌功能开关状态 'version': '1.0.0', 'timestamp': int(time.time()) } self.sio.emit('robot_heartbeat', heartbeat_data) return True except Exception as e: logger.error(f"❌ 发送心跳包失败: {str(e)}") self.websocket_connected = False self._attempt_reconnect() return False def send_song_start(self, song_name): """发送歌曲开始播放事件""" if not self.websocket_connected or not self.sio: return False try: if not hasattr(self, 'system_id') or not self.system_id: self.get_system_uuid() # 修复：正确区分点歌功能开关状态和当前播放状态 song_data = { 'robot_id': self.system_id, 'song_name': song_name, 'timestamp': int(time.time()), 'playmusic_enabled': self.playmusic if hasattr(self, 'playmusic') else False, # 点歌功能开关状态 'playmusic': True # 歌曲开始，当前正在播放歌曲 } self.sio.emit('robot_song_start', song_data) logger.info(f"🎵 歌曲开始播放: {song_name}") return True except Exception as e: logger.error(f"❌ 发送歌曲开始事件失败: {str(e)}") return False def send_song_end(self, song_name): """发送歌曲播放结束事件""" if not self.websocket_connected or not self.sio: return False try: if not hasattr(self, 'system_id') or not self.system_id: self.get_system_uuid() # 修复：根据点歌模式状态正确设置本地任务状态 if hasattr(self, 'playmusic') and self.playmusic: # 点歌模式开启时，保持"点歌中"状态 self.current_task = "点歌功能已开启" logger.info(f"🎵 歌曲播放结束，点歌模式保持开启: {song_name}") else: # 点歌模式关闭时，设为"空闲"状态 self.current_task = "空闲" logger.info(f"🎵 歌曲播放结束，设为空闲状态: {song_name}") # 修复：正确区分点歌功能开关状态和当前播放状态 song_data = { 'robot_id': self.system_id, 'song_name': song_name, 'timestamp': int(time.time()), 'playmusic_enabled': self.playmusic if hasattr(self, 'playmusic') else False, # 点歌功能开关状态 'playmusic': False # 歌曲结束，当前没有播放歌曲 } self.sio.emit('robot_song_end', song_data) logger.info(f"🎵 歌曲播放结束事件已发送: {song_name}, 点歌模式: {'开启' if self.playmusic else '关闭'}") return True except Exception as e: logger.error(f"❌ 发送歌曲结束事件失败: {str(e)}") return False def stop_websocket(self): """停止WebSocket连接""" logger.info("🛑 正在停止WebSocket连接...") # 停止重连线程 if self.reconnect_running: self.reconnect_running = False if self.reconnect_thread and self.reconnect_thread.is_alive(): self.reconnect_thread.join(timeout=5) # 彻底清理WebSocket连接 self._cleanup_websocket() logger.info("✅ WebSocket连接已停止") def f_key_worker(self): """F键自动按键线程工作函数""" while self.f_key_running: # 检查是否正在发送消息 with self.f_key_lock: if not self.is_sending_message: self.press_f_key() # 等待5秒 for i in range(self.f_key_interval): if not self.f_key_running: break time.sleep(1) def start_f_key_thread(self): """启动F键自动按键线程""" if self.f_key_thread is None or not self.f_key_thread.is_alive(): self.f_key_running = True self.f_key_thread = threading.Thread(target=self.f_key_worker, daemon=True) self.f_key_thread.start() logger.info("F键自动按键线程已启动") def stop_f_key_thread(self): """停止F键自动按键线程""" self.f_key_running = False if self.f_key_thread and self.f_key_thread.is_alive(): self.f_key_thread.join(timeout=2) logger.info("F键自动按键线程已停止") def copy_mp3_to_temp_dir(self, mp3_file_path): """将MP3文件复制到临时目录并返回新路径""" if not os.path.exists(mp3_file_path): return None try: # 获取文件名 filename = os.path.basename(mp3_file_path) # 创建临时文件路径 temp_mp3_path = os.path.join(self.temp_audio_dir, filename) # 复制文件到临时目录 shutil.copy2(mp3_file_path, temp_mp3_path) return temp_mp3_path except Exception as e: logger.error(f"复制MP3文件失败: {e}") return None def capture_region(self): """捕获指定屏幕区域并拉伸高度以提高OCR识别率""" x, y, w, h = self.region screenshot = pyautogui.screenshot(region=(x, y, w, h)) # 将图像转换为OpenCV格式 image = cv2.cvtColor(np.array(screenshot), cv2.COLOR_RGB2BGR) # 拉伸图像宽高以提高OCR识别率 # stretched_width = int(w * 1.1) # 宽度拉伸到原来的1.1倍 # stretched_height = int(h * 1.1) # 高度拉伸到原来的1.1倍 # stretched_image = cv2.resize(image, (stretched_width, stretched_height), interpolation=cv2.INTER_LINEAR) return image def extract_text(self, image): """使用 PaddleOCR 提取图像中的文本""" try: # 将 OpenCV BGR 图像转换为 RGB 格式 rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # 使用 PaddleOCR 进行文本检测和识别 result = self.ocr.predict(rgb_image) # 提取识别到的文本 texts = [] scores = [] if result is not None and len(result) > 0: # print(result) # 根据调试信息，文本存储在 rec_texts 字段中 if 'rec_texts' in result[0]: texts = result[0]['rec_texts'] scores = result[0]['rec_scores'] if 'rec_scores' in result[0] else [] # 检查是否满足特定条件：3个值，第2个是时间格式，所有分数>0.7 if (len(texts) == 3 and len(scores) == 3 and all(score > 0.7 for score in scores) and self._is_time_format(texts[1])): # 设置用户名 self.user_name = texts[0] # 返回第三个值（内容） return texts[2] return " ".join(texts) if texts else "" except Exception as e: print(f"文本识别错误: {e}") traceback.print_exc() return "" def _is_time_format(self, text): """检查文本是否为时间格式（如 11:39）""" import re # 匹配 HH:MM 或 H:MM 格式 time_pattern = r'^\d{1,2}:\d{2}$' return bool(re.match(time_pattern, text)) def call_role_play_api(self, user_input): """调用通义实验室角色扮演API""" try: logger.info(f"准备调用API，发送内容: {user_input}") # 添加调试信息 payload = { "input": { "messages": [ {"role": "user", "content": user_input} # 用户输入的内容 ], "aca": { "botProfile": { "characterId": "2b7abc2a69bf4aeb8979fb30fb2ee412" }, "userProfile": { "userId": "123456", "basicInfo": "" }, "scenario": { "description": "" }, "context": { "useChatHistory": False # 不使用历史对话 } } } } response = requests.post(self.url, headers=self.headers, json=payload, timeout=30) if response.status_code == 200: result = json.loads(response.text) # 修复响应解析逻辑 - 根据实际返回的结构进行解析 if "data" in result and "choices" in result["data"]: api_response = result["data"]["choices"][0]["messages"][0]["content"] return api_response except Exception as e: logger.error(f"调用角色扮演API时出错: {e}") traceback.print_exc() return None def search_and_play_music(self, song_name, max_retries=3): """搜索并播放音乐""" def sanitize_filename(filename): illegal_chars = '<>:"/\\|?*' for char in illegal_chars: filename = filename.replace(char, '_') return filename # 检查下载状态 if self.is_downloading: logger.info(f"正在下载其他歌曲，请稍后") return False # 修复：在开始下载前更新状态为"正在下载歌曲" self.is_downloading = True self.current_task = "正在下载歌曲" # 修复：设置下载状态 logger.info(f"🎵 开始下载歌曲: {song_name}") try: # 主接口下载逻辑 - 使用GET请求，适配新的JSON格式 for attempt in range(max_retries): try: # 使用GET请求，参数放在URL中 api_url = "https://api.yaohud.cn/api/music/wyvip" params = { "key": "Fb6gqCmNQ1mu4FWGDOS", "msg": song_name, "level": "standard", "n": 1 } response = requests.get(api_url, params=params, timeout=30) if response.status_code == 200: data = response.json() if data.get("code") == 200 and "data" in data: song_info = data["data"] # 固定使用vipmusic里的url music_url = None if song_info.get("vipmusic") and song_info["vipmusic"].get("url"): music_url = song_info["vipmusic"]["url"] if not music_url: # 主接口没有下载链接，立即尝试备用接口 logger.info("主接口未返回下载链接，尝试备用接口") backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result # 检查歌曲时长是否超过5分钟（300秒） duration_str = "180" # 默认时长 if song_info.get("vipmusic") and song_info["vipmusic"].get("duration"): duration_str = song_info["vipmusic"]["duration"] duration_seconds = self.parse_duration(duration_str) if duration_seconds > 330: # 超过5:30分钟 logger.info(f"歌曲时长 {duration_seconds} 秒超过5分钟，抛弃该歌曲") self.speak_text(f"歌曲 {song_name} 时长超过5分钟，无法播放") # 直接返回False，不尝试备用接口 return False # 文件下载逻辑 - 适配新的字段名称 song_title = song_info.get("name", "未知歌曲") singer = song_info.get("songname", "未知歌手") music_filename = f"{song_title}_{singer}.mp3" music_filepath = os.path.join(self.temp_audio_dir, sanitize_filename(music_filename)) # 下载重试机制 for download_attempt in range(2): try: music_response = requests.get(music_url, timeout=30, stream=True) if music_response.status_code == 200: with open(music_filepath, "wb") as f: for chunk in music_response.iter_content(chunk_size=8192): if chunk: f.write(chunk) # 下载成功后立即设置播放状态 with self.audio_operation_lock: self.play_audio_file(music_filepath) time.sleep(2) self.current_music_duration = duration_seconds - 5 # 修复：只有在当前状态不是服务器端关闭时才恢复为"点歌功能已开启" if self.current_task != "点歌功能已关闭": self.current_task = "点歌功能已开启" return True except Exception as e: if download_attempt == 1: logger.info(f"主接口下载失败，尝试备用接口: {str(e)}") backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result else: logger.info(f"主API请求失败，状态码: {response.status_code}") # 主API请求失败，立即尝试备用接口 backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result except requests.exceptions.ReadTimeout: if attempt == max_retries - 1: backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result except Exception as e: logger.info(f"搜索和播放音乐时出错: {e}") if attempt == max_retries - 1: backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result # 如果所有重试都失败，尝试备用接口 logger.error("主接口所有重试均失败，尝试备用接口") backup_result = self._search_and_play_with_backup_api(song_name) if not backup_result: self.speak_text("点歌失败，请重试") return backup_result finally: # 修复：确保下载状态正确重置 self.is_downloading = False # 修复：只有在当前状态不是服务器端关闭时才恢复为"点歌功能已开启" if self.current_task == "正在下载歌曲" and self.current_task != "点歌功能已关闭": self.current_task = "点歌功能已开启" logger.info(f"🎵 下载状态重置完成") def _search_and_play_with_backup_api(self, song_name): """使用备用API搜索并播放音乐""" # 修复：在备用接口中也设置下载状态 self.current_task = "正在下载歌曲" logger.info(f"🎵 使用备用接口下载歌曲: {song_name}") try: def sanitize_filename(filename): """移除文件名中的非法字符""" illegal_chars = '<>:"/\\|?*' for char in illegal_chars: filename = filename.replace(char, '_') return filename try: # 构建备用API URL backup_api_url = "https://jkapi.com/api/music" params = { "plat": "qq", # qq/wy QQ或者网易 "type": "json", "apiKey": self.musickey, "name": song_name } for attempt in range(2): try: response = requests.get(backup_api_url, params=params, timeout=30) break except requests.exceptions.Timeout: if attempt == 0: logger.warning(f"备用API第{attempt + 1}次请求超时，重试...") continue else: logger.error("备用API两次请求均超时") return False except Exception as e: logger.error(f"备用API请求异常: {e}") return False if response.status_code == 200: data = response.json() # 检查返回结果是否成功 if data.get("code") == 1 and "music_url" in data: song_title = data.get("name", "未知歌曲") singer = data.get("artist", "未知歌手") music_url = data.get("music_url") # 备用接口没有返回时长信息，所以跳过时长检查 if music_url: # 生成文件名 music_filename = f"{song_title}_{singer}.mp3" music_filename = sanitize_filename(music_filename) music_filepath = os.path.join(self.temp_audio_dir, music_filename) # 下载音乐文件 logger.info(f"正在下载歌曲: {song_title} - {singer}") for download_attempt in range(2): try: music_response = requests.get(music_url, timeout=60, stream=True) break except requests.exceptions.Timeout: if download_attempt == 0: logger.warning(f"歌曲下载第{download_attempt + 1}次超时，重试...") continue else: logger.error("歌曲下载两次均超时") return False if music_response.status_code == 200: with open(music_filepath, "wb") as f: for chunk in music_response.iter_content(chunk_size=8192): if chunk: f.write(chunk) logger.info(f"歌曲下载完成: {music_filepath}") # 播放音乐（跳过时长检查） with self.audio_operation_lock: self.play_audio_file(music_filepath) time.sleep(2) # 设置一个默认播放时长（因为接口没有返回时长） self.current_music_duration = 180 # 默认约3分钟 # 修复：只有在当前状态不是服务器端关闭时才恢复为"点歌功能已开启" if self.current_task != "点歌功能已关闭": self.current_task = "点歌功能已开启" return True else: logger.error(f"下载失败，状态码: {music_response.status_code}") return False else: self.speak_text(f"未找到{song_title}下载链接") return False else: error_msg = data.get("msg", "未知错误") logger.error(f"备用API返回错误: {error_msg}") return False else: logger.error(f"备用API请求失败，状态码: {response.status_code}") return False except Exception as e: logger.error(f"使用备用API时出错: {e}") traceback.print_exc() return False finally: # 修复：确保状态正确重置 if self.current_task == "正在下载歌曲" and self.current_task != "点歌功能已关闭": self.current_task = "点歌功能已开启" def check_blacklist_and_announce(self, player_id): """检查玩家是否在黑名单中并语音播报结果""" try: # 调用黑名单API api_url = f"https://www.liminqi.top/api/blacklist/check?player_id={player_id}" response = requests.get(api_url, timeout=30) if response.status_code == 200: data = response.json() if data.get('success'): if data.get('is_blacklisted'): reason = data.get('reason', '未知原因') message = f"玩家 {player_id} 在黑名单中，原因：{reason}" logger.info(f"黑名单查询结果: {message}") self.playmusic == False self.speak_text(f"玩家 {player_id} 在黑名单中") # 新增：发送查询结果到游戏聊天框 self.send_message_to_game(f"#R查询结果：{message}") else: message = f"玩家 {player_id} 不在黑名单中" logger.info(f"黑名单查询结果: {message}") self.speak_text(f"该ID未查询到黑名单信息") # 新增：发送查询结果到游戏聊天框 self.send_message_to_game(f"#G查询结果：{message}") else: error_msg = data.get('message', '查询失败') logger.error(f"黑名单查询失败: {error_msg}") self.speak_text("黑名单查询失败") # 新增：发送错误信息到游戏聊天框 self.send_message_to_game(f"查询失败：{error_msg}") else: logger.error(f"API请求失败，状态码: {response.status_code}") self.speak_text("黑名单服务不可用") # 新增：发送错误信息到游戏聊天框 self.send_message_to_game(f"API请求失败，状态码: {response.status_code}") except requests.exceptions.ConnectionError: logger.warning("无法连接到黑名单API服务") self.speak_text("无法连接到黑名单服务") # 新增：发送错误信息到游戏聊天框 self.send_message_to_game("无法连接到黑名单服务") except requests.exceptions.Timeout: logger.warning("黑名单查询超时") self.speak_text("查询超时") # 新增：发送错误信息到游戏聊天框 self.send_message_to_game("查询超时") except Exception as e: logger.warning(f"黑名单查询异常: {e}") self.speak_text("查询异常") # 新增：发送错误信息到游戏聊天框 self.send_message_to_game(f"查询异常：{str(e)}") def send_message_to_game(self, message): """发送消息到游戏聊天框""" try: # 标记开始发送消息，暂停F键操作 with self.f_key_lock: self.is_sending_message = True # 移动鼠标到聊天框位置 (548, 1106) pydirectinput.moveTo(543, 1016, duration=0.5) time.sleep(0.2) # 点击聊天框激活输入 pydirectinput.click() time.sleep(0.2) # 清空当前输入内容（如果有） pydirectinput.keyDown('ctrl') pydirectinput.press('a') pydirectinput.keyUp('ctrl') time.sleep(0.2) pydirectinput.press('backspace') time.sleep(0.1) # 使用pyperclip复制消息到剪贴板，然后粘贴 pyperclip.copy(message) time.sleep(0.1) # 使用Ctrl+V粘贴 pydirectinput.keyDown('ctrl') pydirectinput.press('v') pydirectinput.keyUp('ctrl') time.sleep(0.2) # 按回车发送 pydirectinput.press('enter') time.sleep(0.1) logger.info(f"已发送消息到游戏聊天框: {message}") except Exception as e: logger.error(f"发送消息到游戏失败: {e}") finally: # 标记发送消息完成，恢复F键操作 with self.f_key_lock: self.is_sending_message = False # 添加检查音乐播放状态的方法 def is_music_playing(self): """检查是否有音乐正在播放""" if self.is_playing_music and self.current_music_duration > 0: elapsed_time = time.time() - self.music_start_time if elapsed_time < self.current_music_duration + 5: # 增加5秒缓冲时间 return True else: # 时间到了，重置状态并检查队列 self.is_playing_music = False self.current_music_duration = 0 self.music_start_time = 0 # 修复：防止重复发送歌曲结束事件 if hasattr(self, '_last_music_request') and self._last_music_request: # 记录当前歌曲名称，然后立即清空，避免重复发送 last_song = self._last_music_request self._last_music_request = "" # 立即清空，防止重复发送 self.send_song_end(last_song) # 检查队列中是否有下一首歌曲 if self.music_queue: song_name = self.music_queue.pop(0) logger.info(f"🎵 当前歌曲播放结束，开始播放队列中的下一首: {song_name}") # 修复：在播放下一首之前保持播放状态 self.is_playing_music = True # 使用线程来处理下一首歌曲，避免阻塞主循环 threading.Thread(target=self._play_song_from_queue, args=(song_name,), daemon=True).start() # 返回True表示正在处理下一首歌曲 return True else: # 队列为空时，只重置播放状态，不关闭点歌功能 logger.info("🎵 队列播放完毕，等待新的点歌请求") # 修复：队列空了也不关闭点歌功能，保持开启状态 # 确保点歌功能状态正确保持 self.playmusic = True # 修复：保持点歌功能开启 self.current_task = "点歌功能已开启" # 修复：保持当前任务状态 self._last_music_request = "" # 修复：队列播放完毕后，立即发送状态更新到前端 if self.websocket_connected and self.sio: try: status_data = { 'robot_id': self.system_id, 'current_task': self.current_task, 'playmusic_enabled': True, # 点歌功能开启 'playmusic': False, # 当前没有播放歌曲 'timestamp': int(time.time()) } self.sio.emit('robot_status_update', status_data) logger.info("🎵 队列播放完毕，已发送状态更新到前端") except Exception as e: logger.error(f"❌ 发送状态更新失败: {str(e)}") return False def _play_song_from_queue(self, song_name): """从队列中播放歌曲""" # 添加播放锁，防止并发播放 if not hasattr(self, '_play_lock'): self._play_lock = threading.Lock() with self._play_lock: # 修复：检查队列中是否有下一首歌曲，但不要提前取出 next_song = "" if self.music_queue: next_song = self.music_queue[0] # 查看队列中的第一首歌，但不取出 if self.search_and_play_music(song_name): # 设置音乐播放状态 self._last_music_request = song_name self.is_playing_music = True # 确保播放状态正确 self.music_start_time = time.time() # 修复：确保点歌功能状态正确 if not self.playmusic: self.playmusic = True self.current_task = "点歌功能已开启" # 修复：更新当前任务状态 # 修复：只在播放成功时发送一次消息，避免重复 if next_song: self.send_message_to_game(f"开始播放 {song_name} 下一首 {next_song}") else: self.send_message_to_game(f"开始播放 {song_name},已是最后一首请及时点歌~") # 修复：发送歌曲开始事件，包含正确的状态字段 self.send_song_start(song_name) else: logger.error(f"🎵 队列歌曲播放失败: {song_name}") self.send_message_to_game(f"队列歌曲 {song_name} 播放失败") # 修复：如果播放失败，正确处理队列中的下一首歌曲 if self.music_queue: next_song = self.music_queue.pop(0) logger.info(f"🎵 继续处理队列中的下一首歌曲: {next_song}") # 使用线程来避免递归调用导致的栈溢出 threading.Thread(target=self._play_song_from_queue, args=(next_song,), daemon=True).start() else: # 队列为空时，只重置播放状态，不关闭点歌功能 logger.info("🎵 队列已空，等待新的点歌请求") self.is_playing_music = False self.current_music_duration = 0 self.music_start_time = 0 # 修复：队列空了也不关闭点歌功能，保持开启状态 self.playmusic = True # 修复：保持点歌功能开启 self.current_task = "点歌功能已开启" # 修复：保持当前任务状态 # 修复：发送歌曲结束事件，包含正确的状态字段 self.send_song_end(song_name) def parse_duration(self, interval_str): """解析歌曲时长字符串，返回秒数""" try: # 处理"04:28"格式（新API格式：分:秒） if ":" in interval_str: parts = interval_str.split(":") if len(parts) == 2: minutes = int(parts[0]) seconds = int(parts[1]) return minutes * 60 + seconds elif len(parts) == 3: # 处理"时:分:秒"格式 hours = int(parts[0]) minutes = int(parts[1]) seconds = int(parts[2]) return hours * 3600 + minutes * 60 + seconds # 处理类似"4分29秒"的格式（旧API格式兼容） if "分" in interval_str and "秒" in interval_str: parts = interval_str.replace("秒", "").split("分") minutes = int(parts[0]) seconds = int(parts[1]) if parts[1] else 0 return minutes * 60 + seconds elif "秒" in interval_str: # 只有秒的情况 seconds = int(interval_str.replace("秒", "")) return seconds else: # 默认返回180秒（3分钟） return 180 except: return 180 def speak_text(self, text): """使用 pyttsx3 生成WAV并通过系统播放器播放""" current_time = time.time() # 修改：允许在下载失败时播报错误信息，但阻止其他类型的播报 # 同时允许时长超过限制的提示语音播报 if hasattr(self, 'is_downloading') and self.is_downloading and not text.startswith( "点歌失败") and not text.startswith("歌曲") and "时长超过" not in text: logger.info("文件下载中，跳过语音播报") return False # 简化冷却检查 if hasattr(self, '_last_speak_attempt'): if current_time - self._last_speak_attempt < self.cooldown_time: remaining = self.cooldown_time - (current_time - self._last_speak_attempt) logger.info(f"播报冷却中，还需等待 {remaining:.1f} 秒") return False # 如果正在播放音乐，也阻止语音播报 if self.is_music_playing(): logger.info("音乐播放中，跳过语音播报") return False try: # 生成带时间戳的唯一文件名（保存在临时文件夹中） timestamp = int(time.time() * 1000) # 毫秒级时间戳 unique_wav_file = os.path.join(self.temp_audio_dir, f"speech_{timestamp}.wav") # 每次都创建新的 TTS 引擎实例 tts_engine = pyttsx3.init() tts_engine.setProperty('rate', 160) tts_engine.setProperty('volume', 1.0) # 保存为唯一文件名的WAV文件 tts_engine.save_to_file(text, unique_wav_file) tts_engine.runAndWait() # 清理引擎 del tts_engine # 确保文件完全写入 time.sleep(0.2) # 调用播放器播放WAV文件 self.play_audio_file(unique_wav_file) self._last_speak_attempt = current_time return True except Exception as e: logger.error(f"语音生成错误: {e}") traceback.print_exc() self._last_speak_attempt = current_time # 出错也更新时间 return False def play_audio_file(self, file_path): """播放音频文件""" try: if os.path.exists(file_path): # 确保文件存在后再播放 time.sleep(0.1) # 短暂延迟确保文件完全写入 # 直接使用系统默认播放器播放 os.startfile(file_path) else: logger.error(f"音频文件不存在: {file_path}") except Exception as e: logger.error(f"播放音频文件失败: {e}") def speak_special_announcement(self): """播报特殊提示（从API接口获取），点歌模式下改为发送到游戏聊天框""" current_time = time.time() if current_time - self._last_special_announcement > self.special_announcement_interval: try: # 调用API获取随机文本 api_url = "https://api.yaohud.cn/api/randtext/twqh" params = { "key": "Fb6gqCmNQ1mu4FWGDOS", } response = requests.get(api_url, params=params, timeout=10) if response.status_code == 200: data = response.json() if data.get("code") == 200: special_text = data.get("data", "") if special_text: # 检查是否在点歌模式下 if self.playmusic: # 点歌模式下：发送文本到游戏聊天框，不朗读 self.send_message_to_game(special_text) self._last_special_announcement = current_time logger.info(f"点歌模式下发送特殊提示到游戏聊天框: {special_text}") return True else: # 非点歌模式下：正常朗读 if self.speak_text(special_text): self._last_special_announcement = current_time logger.info(f"播报特殊提示: {special_text}") return True else: logger.warning("API返回的data字段为空") else: logger.error(f"API返回错误: {data.get('msg', '未知错误')}") else: logger.error(f"API请求失败: HTTP {response.status_code}") except requests.exceptions.Timeout: logger.error("API请求超时") except requests.exceptions.ConnectionError: logger.error("API连接失败") except Exception as e: logger.error(f"获取特殊提示失败: {e}") # API调用失败时，使用默认提示 default_text = "我是文津馆助手小文，你可以通过发送小文和我交流哦" # 检查是否在点歌模式下 if self.playmusic: # 点歌模式下：发送文本到游戏聊天框，不朗读 self.send_message_to_game(default_text) self._last_special_announcement = current_time logger.info(f"点歌模式下发送默认特殊提示到游戏聊天框: {default_text}") return True else: # 非点歌模式下：正常朗读 if self.speak_text(default_text): self._last_special_announcement = current_time logger.info(f"播报默认特殊提示: {default_text}") return True return False def draw_region_box(self, image, text): """在图像上绘制区域框和识别文本（支持中文显示）""" # 绘制绿色边框 cv2.rectangle(image, (0, 0), (image.shape[1] - 1, image.shape[0] - 1), (0, 255, 0), 2) # 添加文本背景 text_bg = np.zeros((80, image.shape[1], 3), dtype=np.uint8) text_bg[:] = (0, 0, 0) # 将OpenCV图像转换为PIL图像以支持中文 pil_image = Image.fromarray(cv2.cvtColor(np.vstack([image, text_bg]), cv2.COLOR_BGR2RGB)) draw = ImageDraw.Draw(pil_image) # 添加识别到的文本 display_text = text[:50] + "..." if len(text) > 50 else text # 使用PIL绘制中文文本 draw.text((10, image.shape[0] + 10), f"识别: {display_text}", font=self.font, fill=(0, 255, 0)) draw.text((10, image.shape[0] + 30), f"区域: X={self.region[0]}, Y={self.region[1]}, 宽={self.region[2]}, 高={self.region[3]}", font=self.font, fill=(255, 255, 255)) draw.text((10, image.shape[0] + 50), "按 'q' 退出", font=self.font, fill=(255, 255, 255)) # 转换回OpenCV格式 result_image = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR) return result_image def reset_state(self, speak_confirmation=True): """重置所有状态变量到初始状态""" # 如果存在播放锁，确保重置时不会造成死锁 if hasattr(self, '_play_lock') and self._play_lock.locked(): try: # 尝试获取锁然后立即释放，确保锁状态正常 acquired = self._play_lock.acquire(blocking=False) if acquired: self._play_lock.release() logger.info("🔓 播放锁已安全释放") except Exception as e: logger.error(f"⚠️ 释放播放锁时出错: {e}") self.last_spoken_text = "" self.last_recognized_text = "" # 重置所有关键词播报记录 self.last_announced_for_11 = "" self.last_announced_for_00 = "" self.last_announced_for_naimei = "" self.last_announced_for_qiufan = "" self.last_announced_for_help = "" self.last_announced_for_vip = "" self.last_query_content = "" # 重置小文API相关变量 self.last_xiaowen_content = "" self._last_xiaowen_call = 0 # 重置F键线程状态 self.is_sending_message = False # 重置音乐相关变量 self._last_music_request = "" self.is_playing_music = False self.current_music_duration = 0 self.music_start_time = 0 self.playmusic = False # 点歌功能开关状态 self.current_task = "空闲" # 当前任务状态 # 清空音乐队列 if hasattr(self, 'music_queue'): self.music_queue.clear() logger.info("🎵 音乐队列已清空") # 重置语音播报时间（设置为过去时间，避免冷却） self._last_speak_attempt = time.time() - self.cooldown_time self._last_special_announcement = time.time() if speak_confirmation: self.speak_text("程序状态已重置") # 发送状态更新事件到前端 if SOCKETIO_AVAILABLE and self.websocket_connected: try: self.socketio.emit('robot_status_update', { 'current_task': self.current_task, 'playmusic_enabled': self.playmusic, # 点歌功能开关状态 'playmusic': False # 重置后没有播放歌曲 }) logger.info("🔄 重置后状态更新事件已发送") except Exception as e: logger.error(f"❌ 发送重置后状态更新事件失败: {e}") def run(self): print("🚀 烟雨阁客户端启动") print("=" * 50) # 确保MP3目录存在 self.ensure_mp3_dir() # 加载配置 config = self.load_or_create_config() # 检查FID if not self.check_fid(config): print("\n📝 请修改配置文件后重新启动") return # 获取FID并提交robot_id映射关系 fid = config.get('CONFIG', 'fid') print(f"🔗 开始提交robot_id和fid映射关系") self.submit_robot_mapping(fid) # 下载云端配置替换本地 self.get_system_uuid() if self.download_server_config(fid): # 重新加载更新后的配置 config = self.load_or_create_config() # 对比VIP名字下载缺失MP3 self.download_missing_mp3(config) print("=" * 50) # 初始化WebSocket连接 if SOCKETIO_AVAILABLE: # 检查是否已经连接，避免重复初始化 if not self.websocket_connected and not self.is_connecting: print("🔄 正在初始化WebSocket连接...") if self.init_websocket(): print("✅ WebSocket连接初始化成功") # 注意：心跳包线程已在connect()事件中启动，这里不需要重复启动 else: print("❌ WebSocket连接初始化失败") else: print("✅ WebSocket连接已存在，跳过重复初始化") else: print("⚠️ WebSocket功能不可用，跳过连接") """主监控循环""" self.running = True last_processed_text = "" print("开始监控屏幕区域... 按 'q' 键退出") print(f"监控区域: X={self.region[0]}, Y={self.region[1]}, 宽度={self.region[2]}, 高度={self.region[3]}") # 启动定期清理线程 self.cleanup_temp_dir_periodically() # 启动F键自动按键线程 # self.start_f_key_thread() # 创建窗口并设置属性 cv2.namedWindow('Screen Monitor - PaddleOCR', cv2.WINDOW_NORMAL) cv2.resizeWindow('Screen Monitor - PaddleOCR', self.region[2], self.region[3] + 80) # 高度增加80像素用于显示文本信息 while self.running: loop_start_time = time.time() try: # 捕获屏幕区域 frame = self.capture_region() # 检查配置文件是否发生变化 # if self._check_config_changed(): # self._reload_config() # 提取文本 text = self.extract_text(frame) self.last_recognized_text = text # 在主循环中添加时间检查 current_time = time.localtime() # 每天5点01分按空格键 if current_time.tm_hour == 5 and current_time.tm_min == 2 and not self.space_pressed_today: print("🕒 检测到5:02，自动按空格键") pyautogui.press('space') pyautogui.press('space') self.space_pressed_today = True else: # 重置标记，确保第二天可以再次触发 self.space_pressed_today = False # 定期检查WebSocket连接状态 if loop_start_time - self.last_connection_check >= self.connection_check_interval: self.last_connection_check = loop_start_time # 使用时间戳赋值 # 检查连接状态 if not self.websocket_connected and not self.is_connecting: # 添加额外的检查：如果最近刚初始化过，跳过重连 if loop_start_time - self.last_connection_attempt < 10: # 10秒内刚尝试过连接 print("🔍 最近刚尝试过连接，跳过重连检查") continue print("🔍 检测到WebSocket连接断开，尝试重连...") self._safe_reconnect() # 输出识别结果 if text.strip() and text != last_processed_text: last_processed_text = text # if "闭嘴" in text or "关机" in text: # self.reset_state(speak_confirmation=True) # 第一次调用时播报 # # 添加延时避免立即重复识别 # time.sleep(1) # continue # 音量控制功能 if "增加音量" == text: try: # 按下Ctrl+Alt+Up组合键 pyautogui.keyDown('ctrl') pyautogui.keyDown('alt') pyautogui.press('up') time.sleep(0.1) pyautogui.press('up') pyautogui.keyUp('alt') pyautogui.keyUp('ctrl') self.send_message_to_game("#R音量+10%") # 添加短暂延迟避免重复触发 time.sleep(0.5) continue except Exception as e: logger.error(f"❌ 音量控制失败: {e}") if "降低音量" == text: try: # 按下Ctrl+Alt+Down组合键 pyautogui.keyDown('ctrl') pyautogui.keyDown('alt') pyautogui.press('down') time.sleep(0.1) pyautogui.press('down') pyautogui.keyUp('alt') pyautogui.keyUp('ctrl') self.send_message_to_game("#G音量-10%") # 添加短暂延迟避免重复触发 time.sleep(0.5) continue except Exception as e: logger.error(f"❌ 音量控制失败: {e}") # 新增：如果文本包含"+"则跳过本次循环 if "+" in text or (text and text[-1] == "国"): time.sleep(0.2) continue # 手动朗读 if "say/" in text and self.playmusic == False: prefix_index = text.find("say") if prefix_index != -1: content_after_prefix = text[prefix_index + 4:].strip() # 跳过 "say/" if content_after_prefix != self.last_xiaowen_content: if self.speak_text(content_after_prefix): self.last_xiaowen_content = content_after_prefix continue # 处理点歌请求 if text.startswith("点歌") and self.playmusic == True: song_name = text[2:].strip() if song_name: # 检查是否为黑名单歌曲 if self._is_blacklisted_song(song_name): # 即使播报冷却中，也要记录黑名单歌曲检测 logger.info(f"检测到黑名单歌曲: {song_name}") self.send_message_to_game(f"#R歌曲 {song_name} 在黑名单中，无法点播") continue # 检查队列中是否有重复歌曲 if hasattr(self, 'music_queue') and song_name in self.music_queue: logger.info(f"检测到重复歌曲: {song_name}") continue # 添加去重检查，避免重复处理相同的点歌请求 if not hasattr(self, '_last_music_request') or self._last_music_request != song_name: # 即使播报冷却中，也要记录点歌请求 logger.info(f"收到点歌请求: {song_name}") # 修复：检查当前是否正在播放音乐，使用is_playing_music而不是is_music_playing()的返回值 # 因为is_music_playing()在队列播放完毕后返回False，但点歌功能应该保持开启 if self.is_playing_music: # 如果正在播放，将歌曲加入队列 self.music_queue.append(song_name) queue_position = len(self.music_queue) # 检查队列中下一首播放的歌曲 next_song_info = "" if queue_position > 1: # 如果队列中已经有至少一首歌在等待 next_song = self.music_queue[0] next_song_info = f"，下一首播放: {next_song}" self.send_message_to_game( f"#R点歌成功！#Y{song_name} #R已加入队列，当前排在#B第 {queue_position} 位{next_song_info}") else: # 如果没有播放，直接播放歌曲 # 检查队列中是否有下一首 self.speak_text(f"收到点歌请求: {song_name}") next_song_info = "" if self.music_queue: # 如果队列中有歌曲 next_song = self.music_queue[0] next_song_info = f"，下一首播放: {next_song}" if self.search_and_play_music(song_name): # 设置音乐播放状态 self._last_music_request = song_name self.is_playing_music = True self.music_start_time = time.time() self.speak_text(f"开始播放 {song_name}{next_song_info}") self.send_message_to_game(f"开始播放 {song_name}{next_song_info}") else: # 即使播报冷却中，也要记录点歌失败 logger.error(f"点歌失败: {song_name}") self.speak_text(f"点歌失败，请重试") continue # 角色扮演API if text.startswith("小文"): content_after_prefix = text[2:].strip() if content_after_prefix and content_after_prefix != self.last_xiaowen_content: if loop_start_time - self._last_xiaowen_call >= self.xiaowen_cooldown_time: response_text = self.call_role_play_api(content_after_prefix) if response_text : if not self.playmusic: # 播放音乐时，跳过语音转文字发送对话 self.speak_text(response_text) self.send_message_to_game(f"#B{response_text}") self.last_xiaowen_content = content_after_prefix self._last_xiaowen_call = loop_start_time else: logger.error("API未返回有效内容") else: remaining = self.xiaowen_cooldown_time - (loop_start_time - self._last_xiaowen_call) logger.info(f"小文API调用冷却中，还需等待 {remaining:.1f} 秒") # 查询黑名单功能 if text.startswith("查询"): prefix_index = text.find("查询") if prefix_index != -1: content_after_prefix = text[prefix_index + 2:].strip() # 跳过 "查询" if content_after_prefix: # 提取数字（玩家ID） import re player_id_match = re.search(r'\d+', content_after_prefix) if player_id_match: player_id = player_id_match.group() # 检查是否与上次查询内容相同 current_query = f"查询{player_id}" if current_query == self.last_query_content: continue logger.info(f"检测到查询请求，玩家ID: {player_id}") self.check_blacklist_and_announce(player_id) # 更新查询记录 self.last_query_content = current_query else: logger.info("查询请求中未找到有效的玩家ID") # 检查VIP玩家 for player_name in self.vip_players: if player_name in text and "房间" in text and not self.playmusic: if player_name != self.last_announced_for_vip: announced = False mp3_file = os.path.join(self.mp3_dir, f"{player_name}.MP3") temp_mp3_file = self.copy_mp3_to_temp_dir(mp3_file) if temp_mp3_file: self.play_audio_file(temp_mp3_file) announced = True else: if self.speak_text(f"最爱天亮的妈妈 {player_name} 进入了烟雨阁"): announced = True if announced: self.last_announced_for_vip = player_name logger.info(f"检测到VIP玩家 '{player_name}'，已触发语音播报") break # 通用播报处理 - 仅在音乐未播放时处理 if not self.playmusic: self._handle_keyword_announcement(text, "111", "捕快.MP3", "last_announced_for_11") self._handle_keyword_announcement(text, "000", "劫狱.MP3", "last_announced_for_00") self._handle_keyword_announcement(text, "奶妈", "奶妈.MP3", "last_announced_for_naimei") self._handle_keyword_announcement(text, "救我", "求助.MP3", "last_announced_for_help") self._handle_keyword_announcement(text, "救命", "求助.MP3", "last_announced_for_help") self._handle_keyword_announcement(text, "有没有犯人", "犯人.MP3", "last_announced_for_qiufan") else: # 当没有识别到文本时，重置所有播报记录 self.last_announced_for_11 = "" self.last_announced_for_00 = "" self.last_announced_for_naimei = "" self.last_announced_for_qiufan = "" self.last_announced_for_help = "" # 检查是否需要播报特殊提示（长时间未播报时） if loop_start_time - self._last_speak_attempt > self.special_announcement_interval: self.speak_special_announcement() self.is_music_playing() # 显示结果 frame_with_box = self.draw_region_box(frame, text) cv2.imshow('Screen Monitor - PaddleOCR', frame_with_box) # 检查按键 key = cv2.waitKey(1) & 0xFF if key == ord('q'): break except Exception as e: logger.error(f"运行时错误: {e}") traceback.print_exc() # 添加短暂延迟，减轻CPU负担 time.sleep(0.2) cv2.destroyAllWindows() self.running = False # 停止WebSocket连接 if SOCKETIO_AVAILABLE: self.stop_websocket() print("监控已停止") def _reload_config(self): """重新加载配置文件""" print("检测到配置文件变化，正在重新加载配置...") try: # 重新读取配置文件 new_config = configparser.ConfigParser() new_config.read(self.config_path, encoding='utf-8') # 更新VIP玩家列表 try: vip_players_str = new_config.get('VIP', 'players') new_vip_players = [name.strip() for name in vip_players_str.split(',')] if new_vip_players != self.vip_players: self.vip_players = new_vip_players print(f"已更新VIP玩家列表: {self.vip_players}") except (configparser.NoSectionError, configparser.NoOptionError) as e: print(f"VIP配置读取错误: {e}") # 更新黑名单歌曲列表 try: blacklist_songs_str = new_config.get('MUSIC', 'blacklist_songs') new_blacklist_songs = [song.strip().lower() for song in blacklist_songs_str.split(',')] if new_blacklist_songs != self.blacklist_songs: self.blacklist_songs = new_blacklist_songs print(f"已更新黑名单歌曲: {self.blacklist_songs}") except (configparser.NoSectionError, configparser.NoOptionError) as e: print(f"黑名单歌曲配置读取错误: {e}") # 从配置文件读取通义灵码和音乐APIkey try: tykey = self.config.get('CONFIG', 'tykey') musickey = self.config.get('CONFIG', 'musickey') self.musickey = musickey self.tykey = tykey except (configparser.NoSectionError, configparser.NoOptionError) as e: # 默认关闭音乐播放 self.tykey = "" self.musickey = "" print("配置文件重载完成") return True except Exception as e: print(f"配置文件重载失败: {e}") return False def _handle_keyword_announcement(self, text, keyword, mp3_filename, last_attr_name): """通用关键词播报处理函数""" if keyword in text: last_value = getattr(self, last_attr_name, "") if text != last_value: # 检查冷却时间 current_time = time.time() last_attempt_attr = f"_last_{last_attr_name}_attempt" last_attempt_time = getattr(self, last_attempt_attr, 0) if current_time - last_attempt_time < self.cooldown_time: remaining = self.cooldown_time - (current_time - last_attempt_time) logger.info(f"'{keyword}'播报冷却中，还需等待 {remaining:.1f} 秒") return False # 新增：检查音乐播放状态，如果正在播放音乐则跳过播报 if self.is_music_playing(): logger.info(f"音乐播放中，跳过'{keyword}'关键词播报") return False mp3_file = os.path.join(self.mp3_dir, mp3_filename) temp_mp3_file = self.copy_mp3_to_temp_dir(mp3_file) if temp_mp3_file: self.play_audio_file(temp_mp3_file) setattr(self, last_attr_name, text) # 更新该关键词的最后播报时间 setattr(self, last_attempt_attr, current_time) logger.info(f"检测到目标文本 '{keyword}'，已触发语音播报") return True return False def _is_blacklisted_song(self, song_name): """检查歌曲是否在黑名单中（智能模糊匹配）""" song_name_lower = song_name.lower().strip() # 首先检查完全匹配（优先检查，不受长度限制） if song_name_lower in [song.lower().strip() for song in self.blacklist_songs]: print(f"完全匹配: '{song_name_lower}' 在黑名单中，返回True") return True # 如果输入歌曲名很短（少于2个字符），直接返回False，避免误判 if len(song_name_lower) < 2: print(f"歌曲名过短，直接返回False") return False for blacklisted_song in self.blacklist_songs: blacklisted_lower = blacklisted_song.lower().strip() # 如果黑名单歌曲名很短（少于1个字符），跳过（基本不会出现） if len(blacklisted_lower) < 1: continue # 1. 直接包含匹配：检查黑名单歌曲名是否包含在输入歌曲名中 if blacklisted_lower in song_name_lower: print(f"直接包含匹配: '{blacklisted_lower}' 在 '{song_name_lower}' 中，返回True") return True # 2. 反向包含匹配：检查输入歌曲名是否包含在黑名单歌曲名中 if song_name_lower in blacklisted_lower: print(f"反向包含匹配: '{song_name_lower}' 在 '{blacklisted_lower}' 中，返回True") return True # 3. 分词匹配：将歌曲名按常见分隔符分词，检查是否有重叠 separators = [' ', '-', '_', '·', '，', '、', '—', '–'] # 对输入歌曲名进行分词 input_words = [song_name_lower] # 默认包含完整歌曲名 for sep in separators: if sep in song_name_lower: input_words = song_name_lower.split(sep) break # 只按第一个找到的分隔符分词 # 对黑名单歌曲名进行分词 blacklisted_words = [blacklisted_lower] # 默认包含完整黑名单歌曲名 for sep in separators: if sep in blacklisted_lower: blacklisted_words = blacklisted_lower.split(sep) break # 只按第一个找到的分隔符分词 # 检查分词后的关键词匹配 for input_word in input_words: if len(input_word) >= 1: # 降低长度限制到1个字符 for blacklisted_word in blacklisted_words: if len(blacklisted_word) >= 1: # 降低长度限制到1个字符 # 检查双向包含关系 if input_word in blacklisted_word or blacklisted_word in input_word: print(f"分词匹配: '{input_word}' 与 '{blacklisted_word}' 匹配，返回True") return True # 检查完全相等 elif input_word == blacklisted_word: print(f"分词完全相等匹配: '{input_word}' == '{blacklisted_word}'，返回True") return True # 4. 相似度匹配：计算编辑距离，如果相似度很高则认为是同一首歌 if len(song_name_lower) > 2 and len(blacklisted_lower) > 2: # 降低长度限制 # 计算最长公共子序列长度 def lcs_length(s1, s2): m, n = len(s1), len(s2) dp = [[0] * (n + 1) for _ in range(m + 1)] for i in range(1, m + 1): for j in range(1, n + 1): if s1[i - 1] == s2[j - 1]: dp[i][j] = dp[i - 1][j - 1] + 1 else: dp[i][j] = max(dp[i - 1][j], dp[i][j - 1]) return dp[m][n] lcs_len = lcs_length(song_name_lower, blacklisted_lower) min_len = min(len(song_name_lower), len(blacklisted_lower)) # 如果最长公共子序列长度占较短字符串长度的70%以上，认为是同一首歌 if min_len > 0 and lcs_len / min_len >= 0.7: # 降低相似度阈值 print( f"相似度匹配: '{song_name}' 与黑名单歌曲 '{blacklisted_song}' 相似度 {lcs_len / min_len:.2f}，返回True") return True return False # 使用示例 if __name__ == "__main__": # 首先进行自动升级检查 if auto_update(): # 如果进行了升级，程序会重启，这里不会执行 pass else: # 如果没有升级或升级失败，继续正常启动 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 检查管理员权限 check_admin_privileges() x = 490 y = 870 width = 340 height = 55 # x = 84 # y = 1015 # width = 360 # height = 30 try: monitor = ScreenMonitor(x, y, width, height) monitor.run() except Exception as e: print(f"发生错误: {e}") traceback.print_exc()