'commit'

WeiXin/T7_TTS_VoiceReply.py

@@ -0,0 +1,145 @@
+# coding=utf-8
+import os
+import sys
+import time
+import threading
+import logging
+
+# 添加项目根目录到 sys.path 以便导入 Util
+project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if project_root not in sys.path:
+    sys.path.append(project_root)
+
+from Util.AlyTtsKit import QwenTTSManager
+from WeiXin import WxUtil
+
+# ======= 配置 =======
+# 这里的 Voice ID 是从 T2_PlayVoice.py 中复制过来的
+MY_VOICE_ID = "qwen-tts-vc-guanyu-voice-20260131160431051-8e51"
+
+# 目标文本
+REPLY_TEXT = "我是少惠林的大张老师，您的孩子几年级了？我们周六周日上班，您可以带孩子过来试听一下。"
+
+# 设置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.StreamHandler(sys.stdout),
+        logging.FileHandler(os.path.join(WxUtil.LOG_DIR, "T7_TTS_VoiceReply.log"), mode='w', encoding='utf-8')
+    ]
+)
+logger = logging.getLogger("T7_TTS")
+
+def tts_worker(tts_manager, voice_id, text, finished_event, start_event):
+    """TTS 播音工作线程"""
+    try:
+        logger.info(f"[TTS] 启动语音合成，音色ID: {voice_id}")
+        # 增加缓冲区到 1.0 秒以解决连贯性问题
+        tts_manager.start_synthesis(voice_id, [text], wait_finished=False, buffer_seconds=1.0)
+        
+        # 等待播放实际开始并通知主线程
+        tts_manager.wait_for_playback_start()
+        start_event.set()
+        
+        # 此时等待全部播放完成
+        if tts_manager.callback:
+            tts_manager.callback.wait_for_finished()
+            
+        logger.info("[TTS] 语音播放完成")
+    except Exception as e:
+        logger.error(f"[TTS] 播放过程中发生异常: {e}")
+    finally:
+        finished_event.set()
+
+def run_t7_task():
+    logger.info("开始执行 T7: 集成 TTS 克隆引擎与微信语音回复 (优化连贯性版本)")
+    
+    # 1. 连接设备
+    d = WxUtil.connect_device()
+    if not d:
+        logger.error("无法连接设备，任务终止")
+        return
+
+    # 2. 初始化 TTS 管理器
+    try:
+        # 增加缓冲区到 1.0 秒以解决断续问题
+        tts_manager = QwenTTSManager()
+        # 这里虽然 init 没传参，但我们可以在 start_synthesis 逻辑里或者修改 AlyTtsKit 让它支持传参
+    except Exception as e:
+        logger.error(f"初始化 TTS 失败: {e}")
+        return
+
+    # 3. 准备微信环境：确保处于语音输入模式
+    logger.info("步骤1: 准备微信环境...")
+    # 先尝试寻找 "按住说话" 按钮，如果找不到则尝试切换
+    tmp_screen = os.path.join(WxUtil.OUTPUT_DIR, "t7_check_mode.jpg")
+    d.screenshot(tmp_screen)
+    press_say_template = os.path.join(WxUtil.TEMPLATE_DIR, "press_say.jpg")
+    pos = WxUtil.match_template_center(tmp_screen, press_say_template, threshold=0.8)
+    
+    if not pos:
+        logger.info(">>> [状态] 未发现 '按住说话' 按钮，尝试切换模式...")
+        # 尝试点击键盘图标切换回语音模式
+        audio_reply_template = os.path.join(WxUtil.TEMPLATE_DIR, "audio_reply.jpg")
+        switch_pos = WxUtil.match_template_center(tmp_screen, audio_reply_template, threshold=0.8)
+        if switch_pos:
+            logger.info(f">>> [切换] 找到切换按钮 {switch_pos}，点击切换...")
+            d.click(switch_pos[0], switch_pos[1])
+            time.sleep(1.5)
+            d.screenshot(tmp_screen)
+            pos = WxUtil.match_template_center(tmp_screen, press_say_template, threshold=0.8)
+        
+    if not pos:
+        logger.error("无法定位到 '按住说话' 按钮，请检查手机界面")
+        return
+
+    logger.info(f">>> [定位] 成功定位到语音按钮中心: {pos}")
+
+    # 4. 同步执行：开始长按 + 启动 TTS
+    finished_event = threading.Event()
+    playback_start_event = threading.Event()
+    tts_thread = threading.Thread(target=tts_worker, args=(tts_manager, MY_VOICE_ID, REPLY_TEXT, finished_event, playback_start_event))
+    
+    logger.info("步骤2: 开始同步录音与播放 (使用 Jitter Buffer 优化)...")
+    
+    # 启动 TTS 线程
+    tts_thread.start()
+    
+    # 等待 TTS 真正开始发出声音
+    logger.info(">>> [等待] 等待音频缓冲区填充并开始播放...")
+    if not playback_start_event.wait(timeout=10):
+        logger.error("等待音频播放超时")
+        return
+    
+    # 按住按钮
+    logger.info(">>> [发送] 检测到音频播放开始，立即按住发送语音按钮...")
+    d.touch.down(pos[0], pos[1])
+    
+    # 模拟计时
+    start_time = time.time()
+    last_second = 0
+    while not finished_event.is_set():
+        elapsed = int(time.time() - start_time)
+        if elapsed > last_second:
+            logger.info(f">>> [计时] {elapsed}")
+            last_second = elapsed
+        time.sleep(0.1)
+        if elapsed > 30:
+            logger.warning("录音时间超过 30 秒，强制结束")
+            break
+            
+    # 释放按钮
+    d.touch.up(pos[0], pos[1])
+    total_duration = time.time() - start_time
+    logger.info(f">>> [完成] 录音完成，总时长约 {total_duration:.2f}s")
+    
+    tts_thread.join()
+    
+    # 5. 截图保存结果
+    final_screen = os.path.join(WxUtil.OUTPUT_DIR, "T7_Final_Result_Optimized.jpg")
+    d.screenshot(final_screen)
+    logger.info(f"任务结束，结果截图已保存至: {final_screen}")
+
+if __name__ == "__main__":
+    run_t7_task()