TranslatecomTranslate.py

"""
DeepL
About the translation and the language endpoints:
    This implementation of DeepL follows Marocco2's implementation of DeepL's JSONRPC API\n
    Arrangements and optimizations have been made\n
    Refer to Issue Animenosekai/translate#7 on GitHub for further details
© Anime no Sekai — 2021
"""

from time import time, sleep
from re import compile
from random import randint
from bs4 import BeautifulSoup

from translatepy.language import Language
from translatepy.translators.base import BaseTranslator, BaseTranslateException
from translatepy.utils.annotations import Tuple, List
from translatepy.utils.request import Request

SENTENCES_SPLITTING_REGEX = compile('(?<=[.!:?]) +')


class DeeplTranslateException(BaseTranslateException):
    """
    Default DeepL Translate exception
    """

    error_codes = {
        1042911: "Too many requests."
    }


class GetClientState():
    """
    DeepL Translate state manager
    """
    def __init__(self, request: Request):
        self.id_number = randint(100, 9999) * 1000
        self.session = request

    def dump(self):
        self.id_number += 1
        data = {
            "id": self.id_number,
            "jsonrpc": "2.0",
            "method": "getClientState",
            "params": {
                "v": "20180814"
            }
        }
        return data

    def get(self):
        """
        Returns a new Client State ID
        """
        request = self.session.post("https://www.deepl.com/PHP/backend/clientState.php", params={"request_type": "jsonrpc", "il": "EN", "method": "getClientState"}, json=self.dump())
        response = request.json()
        return response["id"]


class JSONRPCRequest():
    """
    JSON RPC Request Sender for DeepL
    """
    def __init__(self, request: Request) -> None:
        self.client_state = GetClientState(request)
        try:
            self.id_number = self.client_state.get()
        except Exception:
            self.id_number = randint(2000000, 3000000) # ? I didn't verify the range, but it's better having only DeepL not working than having Translator() crash for only one service
        self.session = request
        self.last_access = 0

    def dump(self, method, params):
        self.id_number += 1
        data = {
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
            "id": self.id_number
        }
        return data

    def send_jsonrpc(self, method, params):
        # Take a break 3 sec between requests, so as not to get a block by the IP address
        if time() - self.last_access < 3:
            distance = 3 - (time() - self.last_access)
            sleep((distance if distance >= 0 else 0))

        request = self.session.post("https://www2.deepl.com/jsonrpc", json=self.dump(method, params))
        self.last_access = time()
        response = request.json()
        if request.status_code == 200:
            return response["result"]
        else:
            raise DeeplTranslateException(response["error"]["code"])


class DeeplTranslate(BaseTranslator):

    _supported_languages = {'AUTO', 'BG', 'ZH', 'CS', 'DA', 'NL', 'NL', 'EN', 'ET', 'FI', 'FR', 'DE', 'EL', 'HU', 'IT', 'JA', 'LV', 'LT', 'PL', 'PT', 'RO', 'RO', 'RO', 'RU', 'SK', 'SL', 'ES', 'ES', 'SV'}

    def __init__(self, request: Request = Request(), preferred_langs: List = ["EN", "RU"]) -> None:
        self.session = request
        self.jsonrpc = JSONRPCRequest(request)
        self.user_preferred_langs = preferred_langs

    def _split_into_sentences(self, text: str, destination_language: str, source_language: str) -> Tuple[List[str], str]:
        """
        Split a string into sentences using the DeepL API.\n
        Fallbacks to a simple Regex splitting if an error occurs or no result is found
        Returned tuple: (Result, Computed Language (None if same as source_language))
        """
        REGEX_SPLIT = True

        if REGEX_SPLIT is True:
            SENTENCES_SPLITTING_REGEX.split(text), None

        params = {
            "texts": [text.strip()],  # What for need strip there?
            "lang": {
                "lang_user_selected": source_language,
                "user_preferred_langs": list(set(self.user_preferred_langs + [destination_language]))
            }
        }
        resp = self.jsonrpc.send_jsonrpc("LMT_split_into_sentences", params)

        return resp["splitted_texts"][0], resp["lang"]

    def _translate(self, text: str, destination_language: str, source_language: str) -> str:
        priority = 1
        quality = ""

        # splitting the text into sentences
        sentences, computed_lang = self._split_into_sentences(text, destination_language, source_language)

        # building the a job per sentence
        jobs = self._build_jobs(sentences, quality)

        # timestamp generation
        i_count = 1
        for sentence in sentences:
            i_count += sentence.count("i")
        ts = int(time() * 10) * 100 + 1000

        # params building
        params = {
            "jobs": jobs,
            "lang": {
                "target_lang": destination_language,
                "user_preferred_langs": [destination_language]
            },
            "priority": priority,
            "timestamp": ts + (i_count - ts % i_count)
        }

        if source_language == "auto":
            params["lang"]["source_lang_computed"] = computed_lang
            params["lang"]["user_preferred_langs"].append(computed_lang)
        else:
            params["lang"]["source_lang_user_selected"] = source_language

        results = self.jsonrpc.send_jsonrpc("LMT_handle_jobs", params)

        try:
            _detected_language = results["source_lang"]
        except:
            _detected_language = source_language

        if results is not None:
            translations = results["translations"]
            return _detected_language, " ".join(obj["beams"][0]["postprocessed_sentence"] for obj in translations if obj["beams"])

    def _language(self, text: str) -> str:
        priority = 1
        quality = ""

        # splitting the text into sentences
        sentences, computed_lang = self._split_into_sentences(text, "EN", "AUTO")

        # building the a job per sentence
        jobs = self._build_jobs(sentences, quality)

        # timestamp generation
        i_count = 1
        for sentence in sentences:
            i_count += sentence.count("i")
        ts = int(time() * 10) * 100 + 1000

        # params building
        params = {
            "jobs": jobs,
            "lang": {
                "target_lang": "EN",
                "user_preferred_langs": ["EN"]
            },
            "priority": priority,
            "timestamp": ts + (i_count - ts % i_count)
        }

        if computed_lang is not None:
            params["lang"]["source_lang_computed"] = computed_lang
            params["lang"]["user_preferred_langs"].append(computed_lang)
        else:
            params["lang"]["source_lang_user_selected"] = "AUTO"

        results = self.jsonrpc.send_jsonrpc("LMT_handle_jobs", params)

        if results is not None:
            return results["source_lang"]

    def _dictionary(self, text: str, destination_language: str, source_language: str) -> str:
        if source_language == "AUTO":
            source_language = self._language(text)

        destination_language = Language(destination_language).name.lower()
        source_language = Language(source_language).name.lower()

        request = self.session.post("https://dict.deepl.com/" + source_language + "-" + destination_language + "/search?ajax=1&source=" + source_language + "&onlyDictEntries=1&translator=dnsof7h3k2lgh3gda&delay=800&jsStatus=0&kind=full&eventkind=keyup&forleftside=true", data={"query": text})
        if request.status_code < 400:
            response = BeautifulSoup(request.text, "html.parser")
            _result = []
            for element in response.find_all("a"):
                if element.has_attr('class'):
                    if "dictLink" in element["class"]:
                        _result.append(element.text.replace("\n", ""))
                        # if "featured" in element["class"]:
                        #     results["featured"].append(element.text.replace("\n", ""))
                        # else:
                        #     results["less_common"].append(element.text.replace("\n", ""))
            return source_language, _result

    def _build_jobs(self, sentences, quality=""):
        """
        Builds a job for each sentence for DeepL
        """
        jobs = []
        for index, sentence in enumerate(sentences):
            if index == 0:
                try:
                    before = []
                    after = [sentences[index + 1]]
                except IndexError:  # index == len(sentences) - 1
                    before = []
                    after = []
            else:
                if len(before) > 4:
                    before.pop(0)  # the "before" array cannot be more than 5 elements long i guess?
                before.extend([sentences[index - 1]])
                if index > len(sentences) - 2:
                    after = []
                else:
                    after = [sentences[index + 1]]

            job = {
                "kind": "default",
                "raw_en_context_after": after.copy(),
                "raw_en_context_before": before.copy(),
                "raw_en_sentence": sentence,
            }
            if quality != "":
                job["quality"] = quality
            jobs.append(job)

        return jobs

    def _language_normalize(self, language):
        if language.id == "zho":
            return "ZH"
        return language.alpha2.upper()

    def _language_denormalize(self, language_code):
        if str(language_code).lower() in {"zh", "zh-cn"}:
            return Language("zho")
        return Language(language_code)

    def __str__(self) -> str:
        return "DeepL"