| import os
|
| import sys
|
| import glob
|
| import requests
|
| from urllib.parse import urlencode
|
| from dotenv import dotenv_values
|
| import traceback
|
| import time
|
|
|
| flores_to_iso = {
|
| "asm_Beng": "as",
|
| "ben_Beng": "bn",
|
| "brx_Deva": "brx",
|
| "doi_Deva": "doi",
|
| "eng_Latn": "en",
|
| "gom_Deva": "gom",
|
| "guj_Gujr": "gu",
|
| "hin_Deva": "hi",
|
| "kan_Knda": "kn",
|
| "kas_Arab": "ks",
|
| "kas_Deva": "ks_Deva",
|
| "mai_Deva": "mai",
|
| "mal_Mlym": "ml",
|
| "mar_Deva": "mr",
|
| "mni_Beng": "mni_Beng",
|
| "mni_Mtei": "mni",
|
| "npi_Deva": "ne",
|
| "ory_Orya": "or",
|
| "pan_Guru": "pa",
|
| "san_Deva": "sa",
|
| "sat_Olck": "sat",
|
| "snd_Arab": "sd",
|
| "snd_Deva": "sd_Deva",
|
| "tam_Taml": "ta",
|
| "tel_Telu": "te",
|
| "urd_Arab": "ur",
|
| }
|
|
|
|
|
| class AzureTranslator:
|
| def __init__(
|
| self,
|
| subscription_key: str,
|
| region: str,
|
| endpoint: str = "https://api.cognitive.microsofttranslator.com",
|
| ) -> None:
|
| self.http_headers = {
|
| "Ocp-Apim-Subscription-Key": subscription_key,
|
| "Ocp-Apim-Subscription-Region": region,
|
| }
|
| self.translate_endpoint = endpoint + "/translate?api-version=3.0&"
|
| self.languages_endpoint = endpoint + "/languages?api-version=3.0"
|
|
|
| self.supported_languages = self.get_supported_languages()
|
|
|
| def get_supported_languages(self) -> dict:
|
| return requests.get(self.languages_endpoint).json()["translation"]
|
|
|
| def batch_translate(self, texts: list, src_lang: str, tgt_lang: str) -> list:
|
| if not texts:
|
| return texts
|
|
|
| src_lang = flores_to_iso[src_lang]
|
| tgt_lang = flores_to_iso[tgt_lang]
|
|
|
| if src_lang not in self.supported_languages:
|
| raise NotImplementedError(
|
| f"Source language code: `{src_lang}` not supported!"
|
| )
|
|
|
| if tgt_lang not in self.supported_languages:
|
| raise NotImplementedError(
|
| f"Target language code: `{tgt_lang}` not supported!"
|
| )
|
|
|
| body = [{"text": text} for text in texts]
|
| query_string = urlencode(
|
| {
|
| "from": src_lang,
|
| "to": tgt_lang,
|
| }
|
| )
|
|
|
| try:
|
| response = requests.post(
|
| self.translate_endpoint + query_string,
|
| headers=self.http_headers,
|
| json=body,
|
| )
|
| except:
|
| traceback.print_exc()
|
| return None
|
|
|
| try:
|
| response = response.json()
|
| except:
|
| traceback.print_exc()
|
| print("Response:", response.text)
|
| return None
|
|
|
| return [payload["translations"][0]["text"] for payload in response]
|
|
|
| def text_translate(self, text: str, src_lang: str, tgt_lang: str) -> str:
|
| return self.batch_translate([text], src_lang, tgt_lang)[0]
|
|
|
|
|
| if __name__ == "__main__":
|
| root_dir = sys.argv[1]
|
|
|
|
|
| config = dotenv_values(os.path.join(os.path.dirname(__file__), ".env"))
|
|
|
| t = AzureTranslator(
|
| config["AZURE_TRANSLATOR_TEXT_SUBSCRIPTION_KEY"],
|
| config["AZURE_TRANSLATOR_TEXT_REGION"],
|
| config["AZURE_TRANSLATOR_TEXT_ENDPOINT"],
|
| )
|
|
|
| pairs = sorted(glob.glob(os.path.join(root_dir, "*")))
|
|
|
| for i, pair in enumerate(pairs):
|
| basename = os.path.basename(pair)
|
|
|
| print(pair)
|
|
|
| src_lang, tgt_lang = basename.split("-")
|
|
|
| print(f"{src_lang} - {tgt_lang}")
|
|
|
|
|
| src_infname = os.path.join(pair, f"test.{src_lang}")
|
| tgt_outfname = os.path.join(pair, f"test.{tgt_lang}.pred.azure")
|
| if not os.path.exists(src_infname):
|
| continue
|
|
|
| src_sents = [
|
| sent.replace("\n", "").strip()
|
| for sent in open(src_infname, "r").read().split("\n")
|
| if sent
|
| ]
|
|
|
| if not os.path.exists(tgt_outfname):
|
| try:
|
| translations = []
|
| for i in range(0, len(src_sents), 128):
|
| start, end = i, int(min(i + 128, len(src_sents)))
|
| translations.extend(
|
| t.batch_translate(src_sents[start:end], src_lang, tgt_lang)
|
| )
|
| with open(tgt_outfname, "w") as f:
|
| f.write("\n".join(translations))
|
|
|
| time.sleep(10)
|
| except Exception as e:
|
| print(e)
|
| continue
|
|
|
|
|
| tgt_infname = os.path.join(pair, f"test.{tgt_lang}")
|
| src_outfname = os.path.join(pair, f"test.{src_lang}.pred.azure")
|
| if not os.path.exists(tgt_infname):
|
| continue
|
|
|
| tgt_sents = [
|
| sent.replace("\n", "").strip()
|
| for sent in open(tgt_infname, "r").read().split("\n")
|
| if sent
|
| ]
|
|
|
| if not os.path.exists(src_outfname):
|
| try:
|
| translations = []
|
| for i in range(0, len(tgt_sents), 128):
|
| start, end = i, int(min(i + 128, len(tgt_sents)))
|
| translations.extend(
|
| t.batch_translate(tgt_sents[start:end], tgt_lang, src_lang)
|
| )
|
| with open(src_outfname, "w") as f:
|
| f.write("\n".join(translations))
|
| except Exception as e:
|
| continue
|
|
|
| time.sleep(10)
|
|
|
