""" LLM-based translation helper using Qwen chat model. This module provides a thin wrapper around DashScope's `qwen-flash` model for high-quality, prompt-controlled translation, independent of the main `Translator` (machine translation) pipeline. Usage example: from query.llm_translate import llm_translate result = llm_translate( text="我看到这个视频后没有笑", target_lang="en", source_lang="zh", source_lang_label="中文", target_lang_label="英文", ) """ from __future__ import annotations import logging import os import time from typing import Dict, Optional from openai import OpenAI from config.env_config import DASHSCOPE_API_KEY from config.services_config import get_translation_config logger = logging.getLogger(__name__) # 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1 # 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1 # 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1 # # 默认保持与现有翻译/索引脚本相同的美国地域，可通过环境变量覆盖： # DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 DEFAULT_QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1" QWEN_MODEL_NAME = "qwen-flash" # 由调用方提供的语言标签/代码填充，占位符说明： # - source_lang: 源语言的人类可读名称（按目标语言本地化，例如 "中文", "English"） # - target_lang: 目标语言的人类可读名称 # - src_lang_code: 源语言代码，例如 "zh" # - tgt_lang_code: 目标语言代码，例如 "en" TRANSLATION_PROMPTS: Dict[str, str] = { "zh": """你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）翻译员。你的目标是在遵循 {target_lang} 的语法、词汇和文化习惯的前提下，准确传达原始 {source_lang} 文本的含义和细微差别。请只输出 {target_lang} 的翻译内容，不要包含任何额外的解释或评论。请将以下 {source_lang} 文本翻译成 {target_lang}： {text}""", "en": """You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) translator. Your goal is to accurately convey the meaning and nuances of the original {source_lang} text while adhering to {target_lang} grammar, vocabulary, and cultural sensitivities. Produce only the {target_lang} translation, without any additional explanations or commentary. Please translate the following {source_lang} text into {target_lang}: {text}""", "ru": """Вы профессиональный переводчик с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Ваша задача — точно передать смысл и нюансы исходного текста на {source_lang}, соблюдая грамматику, лексику и культурные особенности {target_lang}. Выводите только перевод на {target_lang}, без каких-либо дополнительных объяснений или комментариев. Пожалуйста, переведите следующий текст с {source_lang} на {target_lang}: {text}""", "ar": """أنت مترجم محترف من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). هدفك هو نقل المعنى والدلالات الدقيقة للنص الأصلي بلغة {source_lang} بدقة، مع الالتزام بقواعد اللغة والمفردات والحساسيات الثقافية الخاصة بلغة {target_lang}. قم بإنتاج الترجمة إلى {target_lang} فقط دون أي شروحات أو تعليقات إضافية. يرجى ترجمة النص التالي من {source_lang} إلى {target_lang}: {text}""", "ja": """あなたは {source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）へのプロの翻訳者です。{target_lang} の文法、語彙、文化的配慮に従いながら、元の {source_lang} テキストの意味やニュアンスを正確に伝えることが目的です。追加の説明やコメントは一切含めず、{target_lang} の翻訳のみを出力してください。次の {source_lang} テキストを {target_lang} に翻訳してください： {text}""", "es": """Eres un traductor profesional de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Tu objetivo es transmitir con precisión el significado y los matices del texto original en {source_lang}, respetando la gramática, el vocabulario y las sensibilidades culturales de {target_lang}. Produce únicamente la traducción en {target_lang}, sin explicaciones ni comentarios adicionales. Por favor, traduce el siguiente texto de {source_lang} a {target_lang}: {text}""", "de": """Du bist ein professioneller Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Dein Ziel ist es, die Bedeutung und Nuancen des ursprünglichen {source_lang}-Textes genau zu vermitteln und dabei die Grammatik, den Wortschatz und die kulturellen Besonderheiten von {target_lang} zu berücksichtigen. Gib ausschließlich die Übersetzung in {target_lang} aus, ohne zusätzliche Erklärungen oder Kommentare. Bitte übersetze den folgenden {source_lang}-Text in {target_lang}: {text}""", "fr": """Vous êtes un traducteur professionnel de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Votre objectif est de transmettre fidèlement le sens et les nuances du texte original en {source_lang}, tout en respectant la grammaire, le vocabulaire et les sensibilités culturelles de {target_lang}. Produisez uniquement la traduction en {target_lang}, sans explications ni commentaires supplémentaires. Veuillez traduire le texte suivant de {source_lang} vers {target_lang} : {text}""", "it": """Sei un traduttore professionista da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Il tuo obiettivo è trasmettere con precisione il significato e le sfumature del testo originale in {source_lang}, rispettando la grammatica, il vocabolario e le sensibilità culturali di {target_lang}. Produci solo la traduzione in {target_lang}, senza spiegazioni o commenti aggiuntivi. Per favore traduci il seguente testo da {source_lang} a {target_lang}: {text}""", "pt": """Você é um tradutor profissional de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Seu objetivo é transmitir com precisão o significado e as nuances do texto original em {source_lang}, respeitando a gramática, o vocabulário e as sensibilidades culturais de {target_lang}. Produza apenas a tradução em {target_lang}, sem quaisquer explicações ou comentários adicionais. Por favor, traduza o seguinte texto de {source_lang} para {target_lang}: {text}""", } def _get_qwen_client(base_url: Optional[str] = None) -> Optional[OpenAI]: """ Lazily construct an OpenAI-compatible client for DashScope. Uses DASHSCOPE_API_KEY and base_url (provider config / env) to configure endpoint. """ api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY") if not api_key: logger.warning("DASHSCOPE_API_KEY not set; llm-based translation will be disabled") return None # 优先使用显式传入的 base_url，其次环境变量，最后默认地域。 base_url = ( (base_url or "").strip() or os.getenv("DASHSCOPE_BASE_URL") or DEFAULT_QWEN_BASE_URL ) try: client = OpenAI(api_key=api_key, base_url=base_url) return client except Exception as exc: logger.error("Failed to initialize DashScope OpenAI client: %s", exc, exc_info=True) return None def _build_prompt( text: str, target_lang: str, source_lang_label: str, target_lang_label: str, src_lang_code: str, tgt_lang_code: str, ) -> str: """ Build translation prompt for given target language, defaulting to English template. """ key = (target_lang or "").lower() template = TRANSLATION_PROMPTS.get(key) or TRANSLATION_PROMPTS["en"] return template.format( source_lang=source_lang_label, target_lang=target_lang_label, src_lang_code=src_lang_code, tgt_lang_code=tgt_lang_code, text=text, ) def llm_translate( text: str, target_lang: str, *, source_lang: Optional[str] = None, source_lang_label: Optional[str] = None, target_lang_label: Optional[str] = None, timeout_sec: Optional[float] = None, ) -> Optional[str]: """ Translate text with Qwen chat model using rich prompts. - 根据目标语言选择提示词，如果没匹配到则退回英文模板。 - 不对 text 做语言检测或缓存，调用方自行控制。 Args: text: 原始文本 target_lang: 目标语言代码（如 "zh", "en"） source_lang: 源语言代码（可选，不影响提示词选择，仅用于日志） source_lang_label: 源语言展示名称，用于 prompt（默认使用 source_lang） target_lang_label: 目标语言展示名称，用于 prompt（默认使用 target_lang） timeout_sec: 请求超时时间（秒，可选；若未配置则从 config 读取或采用默认） Returns: 翻译后的文本；如失败则返回 None。 """ if not text or not str(text).strip(): return text cfg = get_translation_config() provider_cfg = cfg.providers.get("llm", {}) if isinstance(cfg.providers, dict) else {} model_name = provider_cfg.get("model") or QWEN_MODEL_NAME req_timeout = float(provider_cfg.get("timeout_sec") or timeout_sec or 30.0) base_url = (provider_cfg.get("base_url") or "").strip() or None client = _get_qwen_client(base_url=base_url) if not client: # 无法调用云端，直接回退 logger.warning( "[llm_translate] Client init failed; returning original text. " "text=%r target_lang=%s source_lang=%s", text[:80], target_lang, source_lang or "auto", ) return text tgt = (target_lang or "").lower() or "en" src = (source_lang or "auto").lower() src_label = source_lang_label or src tgt_label = target_lang_label or tgt prompt = _build_prompt( text=text, target_lang=tgt, source_lang_label=src_label, target_lang_label=tgt_label, src_lang_code=src, tgt_lang_code=tgt, ) start = time.time() try: completion = client.chat.completions.create( model=model_name, messages=[ { "role": "user", "content": prompt, } ], timeout=req_timeout, ) content = (completion.choices[0].message.content or "").strip() duration_ms = (time.time() - start) * 1000 logger.info( "[llm_translate] Success | model=%s src=%s tgt=%s latency=%.1fms text=%r -> %r", model_name, src, tgt, duration_ms, text[:80], content[:80], ) return content or text except Exception as exc: duration_ms = (time.time() - start) * 1000 logger.warning( "[llm_translate] Failed | model=%s src=%s tgt=%s latency=%.1fms error=%s", model_name, src, tgt, duration_ms, exc, exc_info=True, ) # 安全回退：出错时返回原文，避免中断上游流程 return text __all__ = [ "TRANSLATION_PROMPTS", "llm_translate", ]