Compare two translation bundles and translate missing keys via deep_translator (Google) by key = value || value logic ...

translate_missing.py

import json
import requests
import concurrent.futures
import os
import time
from tqdm import tqdm
from deep_translator import GoogleTranslator


def load_json_file(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            return json.load(file)
    except FileNotFoundError:
        return {}

def save_json_file(data, file_path):
    with open(file_path, 'w', encoding='utf-8') as file:
        json.dump(data, file, ensure_ascii=False, indent=2)

def load_json_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        return json.load(file)

def translate_text_with_googletrans(key, value, target_lang="de", max_retries=3):
    translator = GoogleTranslator(target=target_lang)
    text_to_translate = f"'{key}' = {value}"
    for attempt in range(1, max_retries + 1):
        try:
            translated = translator.translate(text_to_translate)
            _k, _v = [x.strip() for x in translated.split('=', maxsplit=1)]
            # Überprüfen, ob das Ergebnis sinnvoll ist
            if _k.startswith("'") and _k.endswith("'") and _v != value:
                return _v
            # Übersetzen nur des Wertes, wenn der erste Versuch fehlschlägt
            return translator.translate(value)
        except Exception as e:
            if attempt == max_retries:
                print(f"Übersetzung fehlgeschlagen nach {max_retries} Versuchen: {text_to_translate}")
                return None  # Rückgabe von None, um anzuzeigen, dass die Übersetzung fehlgeschlagen ist
            print(f"Übersetzungsversuch {attempt} fehlgeschlagen, warte {10 * attempt} Sekunden: {e}")
            time.sleep(10 * attempt)
   

def collect_tasks(en_obj, de_obj, parent_key=''):
    tasks = []
    for key, value in en_obj.items():
        full_key = f"{parent_key}.{key}" if parent_key else key
        if isinstance(value, dict):
            tasks.extend(collect_tasks(value, de_obj.get(key, {}), full_key))
        elif isinstance(value, list):
            # Handle lists if necessary
            pass
        else:
            if key not in de_obj:
                tasks.append((full_key, value))
    return tasks

def translate_task(task):
    key, value = task
    translated_text = translate_text_with_googletrans(key, value)
    return key, translated_text

def update_translations(de_translations, key, value):
    keys = key.split('.')
    for k in keys[:-1]:
        de_translations = de_translations.setdefault(k, {})
    de_translations[keys[-1]] = value

def main():
    global de_translations
    global deepl_api_key
    global translator

    en_file_path = 'locale.constant-en_US.json'
    de_file_path = 'locale.constant-de_DE.json'
    updated_de_file_path = de_file_path.replace('.json', '_UPDATED.json')

    # Laden der aktualisierten Datei, falls vorhanden
    if os.path.exists(updated_de_file_path):
        de_translations = load_json_file(updated_de_file_path)
    else:
        de_translations = load_json_file(de_file_path)

    en_translations = load_json_file(en_file_path)
    de_translations = load_json_file(de_file_path)
    updated_de_file_path = de_file_path.replace('.json', '_UPDATED.json')
    
    tasks = collect_tasks(en_translations, de_translations)

    if tasks:
        worker_count = os.cpu_count()*2
        with concurrent.futures.ThreadPoolExecutor(max_workers=worker_count) as executor:
            future_to_key = {executor.submit(translate_task, task): task for task in tasks}
            for future in tqdm(concurrent.futures.as_completed(future_to_key), total=len(tasks)):
                key, _ = future_to_key[future]
                translated_text = future.result()[1]
                if translated_text and key not in de_translations:  # Prüft, ob der Schlüssel bereits existiert
                    update_translations(de_translations, key, translated_text)
                    save_json_file(de_translations, updated_de_file_path)

        print("Übersetzungen abgeschlossen und gespeichert.")
    else:
        print("Keine neuen Übersetzungen erforderlich.")

if __name__ == "__main__":
    main()