Перевод слов на Python: интеграция с внешними сервисами

Расширенные примеры и нестандартные сценарии

Пакетная обработка с ограничением частоты запросов

При большом количестве слов необходимо соблюдать лимиты API. Следующий пример использует time.sleep и повторные попытки для Google Cloud Translation.

import time
from google.cloud import translate_v2 as translate
from tenacity import retry, stop_after_attempt, wait_exponential

client = translate.Client()

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def translate_with_retry(text, target):
    result = client.translate(text, target_language=target)
    return result['translatedText']

words = ['apple', 'banana', 'cherry']
translated = []
for w in words:
    translated.append(translate_with_retry(w, 'ru'))
    time.sleep(0.5)  # не более 2 запросов в секунду
print(translated)

['яблоко', 'банан', 'вишня']

Определение языка исходного текста

Многие API возвращают код обнаруженного языка. Используем googletrans для автоматического определения.

from googletrans import Translator

t = Translator()
detection = t.detect('Bonjour le monde')
print(detection.lang, detection.confidence)  # fr 0.98

fr 0.98

Асинхронные запросы с aiohttp

Для ускорения параллельного перевода используется модуль asyncio и библиотека aiohttp (на примере DeepL).

import asyncio
import aiohttp
import deepl

auth_key = 'ваш_ключ'
translator = deepl.Translator(auth_key)

async def translate_many(texts, target_lang):
    tasks = []
    for text in texts:
        tasks.append(translator.translate_text_async(text, target_lang=target_lang))
    results = await asyncio.gather(*tasks)
    return [r.text for r in results]

words = ['cat', 'dog', 'fish']
loop = asyncio.new_event_loop()
result = loop.run_until_complete(translate_many(words, 'RU'))
print(result)

['кот', 'собака', 'рыба']

Примечание: метод translate_text_async доступен в deepl начиная с версии 1.5.

Кэширование переводов для уменьшения нагрузки

Сохраняем уже переведённые фразы в словарь или Redis, чтобы не повторять запросы.

import hashlib
import json
from google.cloud import translate_v2 as translate

cache = {}
def cached_translate(text, target, client):
    key = hashlib.md5(f'{text}:{target}'.encode()).hexdigest()
    if key in cache:
        return cache[key]
    result = client.translate(text, target_language=target)
    cache[key] = result['translatedText']
    return cache[key]

client = translate.Client()
print(cached_translate('hello', 'ru', client))
print(cached_translate('hello', 'ru', client))  # из кэша

привет
привет

Использование прокси при работе с заблокированными сервисами

Для googletrans в странах, где Google заблокирован, можно указать HTTP-прокси.

import googletrans

translator = googletrans.Translator(proxies={'http': 'http://proxy:8080', 'https': 'http://proxy:8080'})
result = translator.translate('world', dest='ru')
print(result.text)

мир

Обработка ошибок сети и таймаутов

При нестабильном соединении необходимо перехватывать исключения и повторять запрос.

import requests
from requests.exceptions import RequestException
import time

def safe_translate(text, target, api_key, folder_id, max_retries=3):
    url = 'https://translate.api.cloud.yandex.net/translate/v2/translate'
    headers = {'Authorization': f'Api-Key {api_key}', 'Content-Type': 'application/json'}
    data = {'folderId': folder_id, 'texts': [text], 'targetLanguageCode': target}
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data, timeout=5)
            response.raise_for_status()
            return response.json()['translations'][0]['text']
        except RequestException as e:
            print(f'Ошибка {attempt+1}: {e}')
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

print(safe_translate('hello', 'ru', 'ключ', 'folder_id'))

здравствуйте