translate.py

import re
import time
import json
import os
from openai import OpenAI
from googletrans import Translator
from langdetect import detect, DetectorFactory

# 设置种子以确保结果一致
DetectorFactory.seed = 0

def load_translation_cache(cache_file):
    if os.path.exists(cache_file):
        with open(cache_file, 'r', encoding='utf-8') as f:
            translations = json.load(f)
        
        # 将缓存中的所有翻译结果首字母转为小写并去除末尾标点
        for key in translations:
            translations[key] = clean_translation(translations[key])
        
        return translations
    return {}

def save_translation_cache(cache_file, translations):
    # 在保存缓存前确保所有翻译结果首字母转为小写并去除末尾标点
    for key in translations:
        translations[key] = clean_translation(translations[key])
    
    with open(cache_file, 'w', encoding='utf-8') as f:
        json.dump(translations, f, ensure_ascii=False, indent=2)

def get_version(version_file):
    if os.path.exists(version_file):
        with open(version_file, 'r', encoding='utf-8') as f:
            return f.read().strip()
    return None

def update_version(version_file):
    timestamp = str(int(time.time()))
    with open(version_file, 'w', encoding='utf-8') as f:
        f.write(timestamp)
    return timestamp

def contains_chinese(text):
    return any('\u4e00' <= char <= '\u9fff' for char in text)

# def contains_target_language_characters(text, target_lang):
#     try:
#         detected_lang = detect(text)
#         return detected_lang == target_lang
#     except Exception as e:
#         print(f"Language detection failed: {e}")
#         return False

def translate_text_qwen_mt(text, target_lang):
    api_key = os.getenv("AI_API_KEY")
    if not api_key:
        raise ValueError("AI_API_KEY environment variable is not set.")
    
    client = OpenAI(
        api_key=api_key,
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    messages = [
        {'role': 'user', 'content': text}
    ]
    translation_options = {
        "source_lang": "zh",  # 指定源语言为中文
        "target_lang": target_lang
    }
    try:
        completion = client.chat.completions.create(
            model="qwen-mt-plus",
            messages=messages,
            extra_body={
                "translation_options": translation_options
            }
        )
        translated_text = completion.choices[0].message.content
        return translated_text.capitalize().lower().rstrip('.,!?;:')
    except Exception as e:
        print(f"Qwen-MT-Plus translation failed: {e}")
        return ""

def translate_text_google(text, target_lang):
    try:
        translator = Translator(service_urls=['translate.google.com'])
        translation = translator.translate(text, src='zh-cn', dest=target_lang)  # 指定源语言为中文
        translated_text = translation.text
        return translated_text.capitalize().lower().rstrip('.,!?;:')
    except Exception as e:
        print(f"Google Translate failed: {e}")
        return ""

def needs_fallback_translation(translated_text):
    return '\n' in translated_text or '"' in translated_text

def clean_translation(text):
    return text.replace('\n', '').replace('"', '')

def translate_po_file(input_file, output_file, target_lang_code, target_lang_name):
    # 获取目标语言目录
    lang_dir = os.path.dirname(output_file)
    
    # 构建 LC_MESSAGES 目录路径
    lc_messages_dir = os.path.join('languages', target_lang_code, 'LC_MESSAGES')
    
    # 确保 LC_MESSAGES 目录存在
    if not os.path.exists(lc_messages_dir):
        os.makedirs(lc_messages_dir)
    
    # 构建缓存文件和版本文件的路径
    cache_file = os.path.join(lang_dir, f'cache_{target_lang_name}.json')
    version_file = os.path.join(lc_messages_dir, 'version')
    
    translations = load_translation_cache(cache_file)
    current_version = get_version(version_file)
    
    with open(input_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 匹配msgid和空msgstr
    pattern = r'msgid "(.+?)"\s*\nmsgstr "(.*?)"'
    matches = re.finditer(pattern, content)

    updated = False
    used_translations = set()  # 用于跟踪已使用的翻译

    for match in matches:
        msgid_text = match.group(1)

        # 检查缓存
        if msgid_text in translations:
            translated_text = translations[msgid_text]
            # 清理缓存中的翻译结果
            translated_text = clean_translation(translated_text)
            # 直接使用缓存的翻译，不再检查目标语言
            if translated_text == "":
                updated = True
                print(f"Cached translation is empty for: {msgid_text}. Re-translating...")
            else:
                print(f"Using cached translation: {msgid_text} -> {translated_text}")
                # 更新content以反映翻译结果
                content = re.sub(
                    rf'msgid "{re.escape(msgid_text)}"\s*\nmsgstr ".*?"',
                    rf'msgid "{msgid_text}"\nmsgstr "{translated_text}"',
                    content
                )
                used_translations.add(msgid_text)  # 标记为已使用
                continue  # 跳过翻译步骤

        # 进行翻译
        try:
            # 增加重试机制
            max_retries = 3
            for attempt in range(max_retries):
                try:
                    time.sleep(0.1)  # 增加延迟以避免请求过快
                    translated_text = translate_text_qwen_mt(msgid_text, target_lang_code)
                    
                    # 检查翻译结果是否仍包含中文或需要回退翻译
                    if (contains_chinese(translated_text) or 
                        needs_fallback_translation(translated_text)):
                        print(f"Translation does not meet criteria using Qwen-MT-Plus. Using Google Translate...")
                        translated_text = translate_text_google(msgid_text, target_lang_code)
                    
                    # 清理Google翻译结果
                    translated_text = clean_translation(translated_text)
                    
                    # 检查翻译是否有变更
                    if msgid_text in translations and translations[msgid_text] != translated_text:
                        print(f"Translation changed for: {msgid_text} -> {translated_text}")
                        updated = True
                    
                    # 更新缓存
                    translations[msgid_text] = translated_text  # 存储翻译到缓存
                    print(f"New translation [{target_lang_code}]: {msgid_text} -> {translated_text}")
                    used_translations.add(msgid_text)  # 标记为已使用
                    break  # 成功翻译后跳出重试循环
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"Retry {attempt + 1}/{max_retries} for: {msgid_text}")
                    time.sleep(0.1)  # 重试前等待更长时间
        except Exception as e:
            print(f"Translation failed for: {msgid_text}")
            print(f"Error: {e}")
            # 处理翻译失败的情况，删除该条目
            if msgid_text in translations:
                del translations[msgid_text]  # 从缓存中删除该条目
                content = re.sub(rf'msgid "{re.escape(msgid_text)}"\nmsgstr ".*?"\n?', '', content)
                updated = True  # 标记为已更新
                continue  # 继续处理下一个条目

        # 更新content以反映翻译结果
        if translated_text:  # 确保翻译成功
            content = re.sub(
                rf'msgid "{re.escape(msgid_text)}"\s*\nmsgstr ".*?"',
                rf'msgid "{msgid_text}"\nmsgstr "{translated_text}"',
                content
            )
            updated = True
            used_translations.add(msgid_text)  # 标记为已使用

    # 删除未使用的缓存项
    for key in list(translations.keys()):
        if key not in used_translations:
            print(f"Removing unused cache entry: {key}")
            del translations[key]
            updated = True

    if updated:
        save_translation_cache(cache_file, translations)
        new_version = update_version(version_file)
        print(f"Updated version from {current_version} to {new_version}")
    else:
        print("No updates.")
        # 创建一个额外的文件来指示没有更新
        no_update_file = os.path.join(os.path.dirname(output_file), f'{os.path.basename(output_file)}.no-update')
        with open(no_update_file, 'w', encoding='utf-8') as f:
            f.write("# No updates.\n")

    # 确保每个 msgid 和 msgstr 之间没有多余的空格或换行符
    content = re.sub(r'\n\s*msgstr', '\nmsgstr', content)
    
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(content)

if __name__ == '__main__':
    for lang_code, lang_name in [('en', 'English'), ('fa', 'Persian'), ('ru', 'Russian'), ('ko', 'Korean'), ('fr', 'French')]:
        print(f"\nTranslating to {lang_name} ({lang_code})...")
        input_file = f'po/{lang_code}.po'
        output_file = f'po/{lang_code}.po'
        translate_po_file(input_file, output_file, lang_code, lang_name)