实时音视频翻译

import os
import time
import base64
import asyncio
import json
import websockets
import pyaudio
import queue
import threading
import traceback

class LiveTranslateClient:
  def __init__(self, api_key: str, target_language: str = "en", *, audio_enabled: bool = True):
    if not api_key:
      raise ValueError("API key 不能为空。")

    self.api_key = api_key
    self.target_language = target_language
    self.audio_enabled = audio_enabled
    self.voice = voice if audio_enabled else "Cherry"
    self.ws = None
    self.api_url = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

    # 音频输入配置（麦克风）
    self.input_rate = 16000
    self.input_chunk = 1600
    self.input_format = pyaudio.paInt16
    self.input_channels = 1

    # 音频输出配置（播放）
    self.output_rate = 24000
    self.output_chunk = 2400
    self.output_format = pyaudio.paInt16
    self.output_channels = 1

    # 状态管理
    self.is_connected = False
    self.audio_player_thread = None
    self.audio_playback_queue = queue.Queue()
    self.pyaudio_instance = pyaudio.PyAudio()

  async def connect(self):
    """建立与翻译服务的 WebSocket 连接。"""
    headers = {"Authorization": f"Bearer {self.api_key}"}
    try:
      self.ws = await websockets.connect(self.api_url, additional_headers=headers)
      self.is_connected = True
      print(f"已成功连接到服务器：{self.api_url}")
      await self.configure_session()
    except Exception as e:
      print(f"连接失败：{e}")
      self.is_connected = False
      raise

  async def configure_session(self):
    """配置翻译会话，设置目标语言、音色等参数。"""
    config = {
      "event_id": f"event_{int(time.time() * 1000)}",
      "type": "session.update",
      "session": {
        # 'modalities' 控制输出类型。
        # ["text", "audio"]：同时返回翻译文本和合成语音（推荐）。
        # ["text"]：仅返回翻译文本。
        "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
        **({"voice": self.voice} if self.audio_enabled and self.voice else {}),
        "input_audio_format": "pcm",
        "output_audio_format": "pcm",
        # 'input_audio_transcription' 配置源语言识别。
        # 将 'model' 设置为 'qwen3-asr-flash-realtime' 可同时输出源语言识别结果。
        # "input_audio_transcription": {
        #     "model": "qwen3-asr-flash-realtime",
        #     "language": "zh"  # 源语言，默认为 'en'
        # },
        "translation": {
          "language": self.target_language
        }
      }
    }
    print(f"正在发送会话配置：{json.dumps(config, indent=2, ensure_ascii=False)}")
    await self.ws.send(json.dumps(config))

  async def send_audio_chunk(self, audio_data: bytes):
    """编码并发送音频数据块到服务器。"""
    if not self.is_connected:
      return

    event = {
      "event_id": f"event_{int(time.time() * 1000)}",
      "type": "input_audio_buffer.append",
      "audio": base64.b64encode(audio_data).decode()
    }
    await self.ws.send(json.dumps(event))

  async def send_image_frame(self, image_bytes: bytes, *, event_id: str | None = None):
    # 发送图像数据到服务器
    if not self.is_connected:
      return

    if not image_bytes:
      raise ValueError("image_bytes 不能为空")

    # 编码为 Base64
    image_b64 = base64.b64encode(image_bytes).decode()

    event = {
      "event_id": event_id or f"event_{int(time.time() * 1000)}",
      "type": "input_image_buffer.append",
      "image": image_b64,
    }

    await self.ws.send(json.dumps(event))

  def _audio_player_task(self):
    stream = self.pyaudio_instance.open(
      format=self.output_format,
      channels=self.output_channels,
      rate=self.output_rate,
      output=True,
      frames_per_buffer=self.output_chunk,
    )
    try:
      while self.is_connected or not self.audio_playback_queue.empty():
        try:
          audio_chunk = self.audio_playback_queue.get(timeout=0.1)
          if audio_chunk is None: # 终止信号
            break
          stream.write(audio_chunk)
          self.audio_playback_queue.task_done()
        except queue.Empty:
          continue
    finally:
      stream.stop_stream()
      stream.close()

  def start_audio_player(self):
    """启动音频播放线程（仅在启用音频输出时有效）。"""
    if not self.audio_enabled:
      return
    if self.audio_player_thread is None or not self.audio_player_thread.is_alive():
      self.audio_player_thread = threading.Thread(target=self._audio_player_task, daemon=True)
      self.audio_player_thread.start()

  async def handle_server_messages(self, on_text_received):
    """循环处理服务器返回的消息。"""
    try:
      async for message in self.ws:
        event = json.loads(message)
        event_type = event.get("type")
        if event_type == "response.audio.delta" and self.audio_enabled:
          audio_b64 = event.get("delta", "")
          if audio_b64:
            audio_data = base64.b64decode(audio_b64)
            self.audio_playback_queue.put(audio_data)

        elif event_type == "response.done":
          print("\n[INFO] 一轮响应已完成。")
          usage = event.get("response", {}).get("usage", {})
          if usage:
            print(f"[INFO] Token 用量：{json.dumps(usage, indent=2, ensure_ascii=False)}")
        # 处理源语言识别结果（需启用 input_audio_transcription.model）
        # elif event_type == "conversation.item.input_audio_transcription.text":
        #     stash = event.get("stash", "")  # 未确认的识别文本
        #     print(f"[识别中] {stash}")
        # elif event_type == "conversation.item.input_audio_transcription.completed":
        #     transcript = event.get("transcript", "")  # 最终识别结果
        #     print(f"[源语言] {transcript}")
        elif event_type == "response.audio_transcript.done":
          print("\n[INFO] 文本翻译完成。")
          text = event.get("transcript", "")
          if text:
            print(f"[INFO] 翻译文本：{text}")
        elif event_type == "response.text.done":
          print("\n[INFO] 文本翻译完成。")
          text = event.get("text", "")
          if text:
            print(f"[INFO] 翻译文本：{text}")

    except websockets.exceptions.ConnectionClosed as e:
      print(f"[WARNING] 连接已关闭：{e}")
      self.is_connected = False
    except Exception as e:
      print(f"[ERROR] 处理消息时发生未知错误：{e}")
      traceback.print_exc()
      self.is_connected = False

  async def start_microphone_streaming(self):
    """从麦克风采集音频并流式发送到服务器。"""
    stream = self.pyaudio_instance.open(
      format=self.input_format,
      channels=self.input_channels,
      rate=self.input_rate,
      input=True,
      frames_per_buffer=self.input_chunk
    )
    print("麦克风已开启，请开始说话...")
    try:
      while self.is_connected:
        audio_chunk = await asyncio.get_event_loop().run_in_executor(
          None, stream.read, self.input_chunk
        )
        await self.send_audio_chunk(audio_chunk)
    finally:
      stream.stop_stream()
      stream.close()

  async def close(self):
    """优雅关闭连接并释放资源。"""
    self.is_connected = False
    if self.ws:
      await self.ws.close()
      print("WebSocket 连接已关闭。")

    if self.audio_player_thread:
      self.audio_playback_queue.put(None) # 发送终止信号
      self.audio_player_thread.join(timeout=1)
      print("音频播放线程已停止。")

    self.pyaudio_instance.terminate()
    print("PyAudio 实例已释放。")

import os
import asyncio
from livetranslate_client import LiveTranslateClient

def print_banner():
  print("=" * 60)
  print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime")
  print("=" * 60 + "\n")

def get_user_config():
  """获取用户配置"""
  print("请选择模式：")
  print("1. 语音 + 文本 [默认] | 2. 仅文本")
  mode_choice = input("请输入选项（回车默认为语音 + 文本）：").strip()
  audio_enabled = (mode_choice != "2")

  if audio_enabled:
    lang_map = {
      "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt",
      "7": "es", "8": "it", "9": "ko", "10": "ja", "11": "yue"
    }
    print("请选择目标翻译语言（语音 + 文本模式）：")
    print("1. 英语 | 2. 中文 | 3. 俄语 | 4. 法语 | 5. 德语 | 6. 葡萄牙语 | 7. 西班牙语 | 8. 意大利语 | 9. 韩语 | 10. 日语 | 11. 粤语")
  else:
    lang_map = {
      "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt", "7": "es", "8": "it",
      "9": "id", "10": "ko", "11": "ja", "12": "vi", "13": "th", "14": "ar",
      "15": "yue", "16": "hi", "17": "el", "18": "tr"
    }
    print("请选择目标翻译语言（仅文本模式）：")
    print("1. 英语 | 2. 中文 | 3. 俄语 | 4. 法语 | 5. 德语 | 6. 葡萄牙语 | 7. 西班牙语 | 8. 意大利语 | 9. 印尼语 | 10. 韩语 | 11. 日语 | 12. 越南语 | 13. 泰语 | 14. 阿拉伯语 | 15. 粤语 | 16. 印地语 | 17. 希腊语 | 18. 土耳其语")

  choice = input("请输入选项（默认为第一个选项）：").strip()
  target_language = lang_map.get(choice, next(iter(lang_map.values())))

  voice = None
  if audio_enabled:
    print("\n请选择语音合成音色：")
    voice_map = {"1": "Cherry", "2": "Nofish", "3": "Sunny", "4": "Jada", "5": "Dylan", "6": "Peter", "7": "Eric", "8": "Kiki", "9": "Ethan"}
    print("1. Cherry（女声）[默认] | 2. Nofish（男声） | 3. Sunny（四川女声） | 4. Jada（上海女声） | 5. Dylan（北京男声） | 6. Peter（天津男声） | 7. Eric（四川男声） | 8. Kiki（粤语女声） | 9. Ethan（男声）")
    voice_choice = input("请输入选项（回车默认为 Cherry）：").strip()
    voice = voice_map.get(voice_choice, "Cherry")
  return target_language, voice, audio_enabled

async def main():
  """主程序入口"""
  print_banner()

  api_key = os.environ.get("DASHSCOPE_API_KEY")
  if not api_key:
    print("[ERROR] 请设置 DASHSCOPE_API_KEY 环境变量。")
    print("  示例：export DASHSCOPE_API_KEY='YOUR_API_KEY'")
    return

  target_language, voice, audio_enabled = get_user_config()
  print("\n配置完成：")
  print(f"  - 目标语言：{target_language}")
  if audio_enabled:
    print(f"  - 合成音色：{voice}")
  else:
    print("  - 输出模式：仅文本")

  client = LiveTranslateClient(api_key=api_key, target_language=target_language, voice=voice, audio_enabled=audio_enabled)

  # 定义回调函数
  def on_translation_text(text):
    print(text, end="", flush=True)

  try:
    print("正在连接翻译服务...")
    await client.connect()

    # 根据模式启动音频播放
    client.start_audio_player()

    print("\n" + "-" * 60)
    print("连接成功！请对着麦克风说话。")
    print("程序将实时翻译您的语音并播放结果。按 Ctrl+C 退出。")
    print("-" * 60 + "\n")

    # 并发运行消息处理和麦克风录音
    message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
    tasks = [message_handler]
    # 无论是否启用音频输出，都必须从麦克风采集音频用于翻译
    microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
    tasks.append(microphone_streamer)

    await asyncio.gather(*tasks)

  except KeyboardInterrupt:
    print("\n\n用户中断，正在退出...")
  except Exception as e:
    print(f"\n发生严重错误：{e}")
  finally:
    print("\n正在清理资源...")
    await client.close()
    print("程序已退出。")

if __name__ == "__main__":
  asyncio.run(main())

# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

headers = [
  "Authorization: Bearer " + API_KEY
]

def on_open(ws):
  print(f"已连接到服务器：{API_URL}")
def on_message(ws, message):
  data = json.loads(message)
  print("收到事件：", json.dumps(data, indent=2))
def on_error(ws, error):
  print("错误：", error)

ws = websocket.WebSocketApp(
  API_URL,
  header=headers,
  on_open=on_open,
  on_message=on_message,
  on_error=on_error
)

ws.run_forever()