From bd01218f67066016a3db29dc58ea99e8d65be09f Mon Sep 17 00:00:00 2001 From: DedInc Date: Fri, 13 Dec 2024 09:15:12 +0600 Subject: [PATCH] v1.1.0 --- LICENSE | 2 +- mintrans/__init__.py | 26 +++++- mintrans/async_mintrans.py | 169 +++++++++++++++++++++++++++++++++++++ mintrans/mintrans.py | 146 +++++++++++++++++++++----------- setup.py | 4 +- 5 files changed, 294 insertions(+), 53 deletions(-) create mode 100644 mintrans/async_mintrans.py diff --git a/LICENSE b/LICENSE index d0ce461..2bbeabd 100644 --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2023 Vladislav Zenkevich +Copyright (c) 2024 Vladislav Zenkevich Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/mintrans/__init__.py b/mintrans/__init__.py index 2b2727b..e049278 100644 --- a/mintrans/__init__.py +++ b/mintrans/__init__.py @@ -1 +1,25 @@ -from .mintrans import * \ No newline at end of file +from .mintrans import * +try: + from .async_mintrans import ( + AsyncBingTranslator, + AsyncDeepLTranslator, + AsyncGoogleTranslator, + get_random_user_agent_async + ) + __all__ = [ + 'BingTranslator', + 'DeepLTranslator', + 'GoogleTranslator', + 'RateLimitException', + 'AsyncBingTranslator', + 'AsyncDeepLTranslator', + 'AsyncGoogleTranslator', + 'get_random_user_agent_async' + ] +except ImportError: + __all__ = [ + 'BingTranslator', + 'DeepLTranslator', + 'GoogleTranslator', + 'RateLimitException' + ] \ No newline at end of file diff --git a/mintrans/async_mintrans.py b/mintrans/async_mintrans.py new file mode 100644 index 0000000..c3f9083 --- /dev/null +++ b/mintrans/async_mintrans.py @@ -0,0 +1,169 @@ +import aiohttp +import re +import time +import random +import asyncio +from functools import wraps +from .mintrans import RateLimitException, _CHROME_VERSION + +async def get_random_user_agent_async(): + try: + async with aiohttp.ClientSession() as session: + async with session.get('https://fingerprints.bablosoft.com/preview?tags=Desktop,' + random.choice(['Firefox', 'Chrome', 'Safari'])) as response: + if response.status == 200: + data = await response.json() + return data['ua'] + raise Exception("Failed to fetch user agent") + except: + return f'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{_CHROME_VERSION} Safari/537.36' + +def async_retry_on_rate_limit(max_attempts=3, delay=1): + def decorator(func): + @wraps(func) + async def wrapper(*args, **kwargs): + attempts = 0 + while attempts < max_attempts: + try: + return await func(*args, **kwargs) + except RateLimitException: + attempts += 1 + if attempts == max_attempts: + raise + await asyncio.sleep(delay) + return None + return wrapper + return decorator + +class AsyncBingTranslator: + def __init__(self): + self.session = None + self.headers = None + + async def _get_session(self): + async with aiohttp.ClientSession() as session: + headers = { + 'User-Agent': await get_random_user_agent_async(), + 'Referer': 'https://www.bing.com/translator' + } + async with session.get('https://www.bing.com/translator', headers=headers) as response: + content = await response.text() + params_pattern = re.compile(r'params_AbusePreventionHelper\s*=\s*(\[.*?\]);', re.DOTALL) + match = params_pattern.search(content) + if match: + params = match.group(1) + key, token, _ = [p.strip('"').replace('[', '').replace(']', '') for p in params.split(',')] + headers.update({'key': key, 'token': token}) + match = re.search(r'IG:"(\w+)"', content) + if match: + ig_value = match.group(1) + headers.update({'IG': ig_value}) + return headers + + @async_retry_on_rate_limit(max_attempts=3) + async def translate(self, text, from_lang, to_lang): + if not self.headers: + self.headers = await self._get_session() + + url = f'https://www.bing.com/ttranslatev3?isVertical=1&&IG={self.headers.get("IG")}&IID=translator.{random.randint(5019, 5026)}.{random.randint(1, 3)}' + data = { + '': '', + 'fromLang': from_lang, + 'text': text, + 'to': to_lang, + 'token': self.headers.get('token'), + 'key': self.headers.get('key') + } + + async with aiohttp.ClientSession() as session: + async with session.post(url, data=data, headers=self.headers) as response: + response_json = await response.json() + if isinstance(response_json, dict): + if 'ShowCaptcha' in response_json: + self.headers = await self._get_session() + raise RateLimitException('Bing Translate rate limit reached (Captcha)!') + elif 'statusCode' in response_json: + if response_json['statusCode'] == 400: + response_json['errorMessage'] = f'1000 characters limit! You send {len(text)} characters.' + elif response_json['statusCode'] == 429: + raise RateLimitException('Bing Translate rate limit reached!') + else: + return response_json[0] + return response_json + +class AsyncDeepLTranslator: + def __init__(self): + pass + + @async_retry_on_rate_limit(max_attempts=3) + async def translate(self, text, from_lang, to_lang): + json_data = { + "jsonrpc": "2.0", + "method": "LMT_handle_jobs", + "params": { + "jobs": [{ + "kind": "default", + "sentences": [{ + "text": text, + "id": 1, + "prefix": "" + }], + "raw_en_context_before": [], + "raw_en_context_after": [], + "preferred_num_beams": 4 + }], + "lang": { + "target_lang": to_lang.upper(), + "preference": { + "weight": {}, + "default": "default" + }, + "source_lang_computed": from_lang.upper() + }, + "priority": 1, + "commonJobParams": { + "quality": "normal", + "mode": "translate", + "browserType": 1, + "textType": "plaintext", + }, + "timestamp": round(time.time()) + }, + "id": random.randint(88720004, 100000000) + } + + headers = {'User-Agent': await get_random_user_agent_async()} + async with aiohttp.ClientSession() as session: + async with session.post('https://www2.deepl.com/jsonrpc?method=LMT_handle_jobs', json=json_data, headers=headers) as response: + response_json = await response.json() + try: + translated_text = response_json['result'] + return translated_text + except KeyError: + if 'many requests' in response_json['error']['message']: + raise RateLimitException('Rate limit error!') + raise KeyError + +class AsyncGoogleTranslator: + def __init__(self): + pass + + @async_retry_on_rate_limit(max_attempts=3) + async def translate(self, text, from_lang, to_lang): + url = 'https://translate.googleapis.com/translate_a/single' + params = { + 'client': 'gtx', + 'sl': 'auto', + 'tl': to_lang, + 'hl': from_lang, + 'dt': ['t', 'bd'], + 'dj': '1', + 'source': 'popup5', + 'q': text + } + + headers = {'User-Agent': await get_random_user_agent_async()} + async with aiohttp.ClientSession() as session: + async with session.get(url, params=params, headers=headers) as response: + if response.status == 429: + raise RateLimitException('Google Translate rate limit reached!') + return await response.json() diff --git a/mintrans/mintrans.py b/mintrans/mintrans.py index 1224fe4..6905239 100644 --- a/mintrans/mintrans.py +++ b/mintrans/mintrans.py @@ -2,6 +2,44 @@ import re import time import random +from functools import wraps + +class RateLimitException(Exception): + pass + +_CHROME_VERSION = None +def get_google_latest_version(): + global _CHROME_VERSION + if _CHROME_VERSION is None: + try: + data = requests.get('https://googlechromelabs.github.io/chrome-for-testing/last-known-good-versions.json').json() + _CHROME_VERSION = data['channels']['Stable']['version'] + except: + _CHROME_VERSION = '131.0.6778.139' + return _CHROME_VERSION + +def get_random_user_agent(): + try: + return requests.get('https://fingerprints.bablosoft.com/preview?tags=Desktop,' + random.choice(['Firefox', 'Chrome', 'Safari'])).json()['ua'] + except: + return f'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{get_google_latest_version()} Safari/537.36' + +def retry_on_rate_limit(max_attempts=3, delay=1): + def decorator(func): + @wraps(func) + def wrapper(*args, **kwargs): + attempts = 0 + while attempts < max_attempts: + try: + return func(*args, **kwargs) + except RateLimitException: + attempts += 1 + if attempts == max_attempts: + raise + time.sleep(delay) + return None + return wrapper + return decorator class BingTranslator: def __init__(self): @@ -10,7 +48,7 @@ def __init__(self): def _get_session(self): session = requests.Session() headers = { - 'User-Agent': '', + 'User-Agent': get_random_user_agent(), 'Referer': 'https://www.bing.com/translator' } session.headers.update(headers) @@ -28,6 +66,7 @@ def _get_session(self): session.headers.update({'IG': ig_value}) return session + @retry_on_rate_limit(max_attempts=3) def translate(self, text, from_lang, to_lang): url = f'https://www.bing.com/ttranslatev3?isVertical=1&&IG={self.session.headers.get("IG")}&IID=translator.{random.randint(5019, 5026)}.{random.randint(1, 3)}' data = {'': '', 'fromLang': from_lang, 'text': text, 'to': to_lang, 'token': self.session.headers.get('token'), 'key': self.session.headers.get('key')} @@ -35,79 +74,88 @@ def translate(self, text, from_lang, to_lang): if type(response) is dict: if 'ShowCaptcha' in response.keys(): self.session = self._get_session() - return self.translate(text, from_lang, to_lang) + raise RateLimitException('Bing Translate rate limit reached (Captcha)!') elif 'statusCode' in response.keys(): if response['statusCode'] == 400: response['errorMessage'] = f'1000 characters limit! You send {len(text)} characters.' + elif response['statusCode'] == 429: + raise RateLimitException('Bing Translate rate limit reached!') else: return response[0] return response -class RateLimitException(Exception): - pass - class DeepLTranslator: def __init__(self): pass + @retry_on_rate_limit(max_attempts=3) def translate(self, text, from_lang, to_lang): json = { - "jsonrpc": "2.0", - "method": "LMT_handle_jobs", - "params": { - "jobs": [{ - "kind": "default", - "sentences": [{ - "text": text, - "id": 1, - "prefix": "" - }], - "raw_en_context_before": [], - "raw_en_context_after": [], - "preferred_num_beams": 4 - }], - "lang": { - "target_lang": to_lang.upper(), - "preference": { - "weight": {}, - "default": "default" + "jsonrpc": "2.0", + "method": "LMT_handle_jobs", + "params": { + "jobs": [{ + "kind": "default", + "sentences": [{ + "text": text, + "id": 1, + "prefix": "" + }], + "raw_en_context_before": [], + "raw_en_context_after": [], + "preferred_num_beams": 4 + }], + "lang": { + "target_lang": to_lang.upper(), + "preference": { + "weight": {}, + "default": "default" + }, + "source_lang_computed": from_lang.upper() + }, + "priority": 1, + "commonJobParams": { + "quality": "normal", + "mode": "translate", + "browserType": 1, + "textType": "plaintext", + }, + "timestamp": round(time.time()) }, - "source_lang_computed": from_lang.upper() - }, - "priority": 1, - "commonJobParams": { - "regionalVariant": "en-US", - "mode": "translate", - "textType": "plaintext", - "browserType": 1 - }, - "timestamp": round(time.time() * 1.5) - }, - "id": random.randint(100000000, 9999999999) -} - r = requests.post('https://www2.deepl.com/jsonrpc?method=LMT_handle_jobs', json=json) + "id": random.randint(88720004, 100000000) + } + headers = {'User-Agent': get_random_user_agent()} + r = requests.post('https://www2.deepl.com/jsonrpc?method=LMT_handle_jobs', json=json, headers=headers) + print(r.json()) try: - translated_text = r.json()['result']['translations'][-1]['beams'][-1]['sentences'][-1]['text'] + translated_text = r.json()['result'] return translated_text except KeyError: - raise RateLimitException('Rate limit error!') + if 'many requests' in r.json()['error']['message']: + raise RateLimitException('Rate limit error!') + raise KeyError class GoogleTranslator: def __init__(self): pass + @retry_on_rate_limit(max_attempts=3) def translate(self, text, from_lang, to_lang): url = 'https://translate.googleapis.com/translate_a/single' params = { - 'client': 'gtx', - 'sl': 'auto', - 'tl': to_lang, - 'hl': from_lang, - 'dt': ['t', 'bd'], - 'dj': '1', - 'source': 'popup5', - 'q': text + 'client': 'gtx', + 'sl': 'auto', + 'tl': to_lang, + 'hl': from_lang, + 'dt': ['t', 'bd'], + 'dj': '1', + 'source': 'popup5', + 'q': text } - return requests.get(url, params=params).json() \ No newline at end of file + headers = {'User-Agent': get_random_user_agent()} + response = requests.get(url, params=params, headers=headers) + if response.status_code == 429: + raise RateLimitException('Google Translate rate limit reached!') + return response.json() diff --git a/setup.py b/setup.py index e226cd4..c2a22c3 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="mintrans", - version="1.0.1", + version="1.1.0", author="Maehdakvan", author_email="visitanimation@google.com", description="Mintrans is a free API wrapper that utilizes Bing, DeepL, and Google Translate for translation purposes.", @@ -21,6 +21,6 @@ "Operating System :: OS Independent", ], packages=find_packages(), - install_requires=['requests'], + install_requires=['requests', 'aiohttp', 'asyncio'], python_requires='>=3.6' )