app.py

"""Failed attempt to fix "This event loop is already running" in Widows:
from platform import platform
if platform().lower().startswith("windows"):
    from install import install
    try:
        import nest_asyncio
    except ModuleNotFoundError:
        try:
            install("nest-asyncio")
            import nest_asyncio
        except Exception as exc:
            print(exc)
            raise SystemExit(1)
    nest_asyncio.apply()

from gevent import monkey
monkey.patch_all()

# """
# pylint: disable=broad-except
# from gevent import monkey
# monkey.patch_all()

# import nest_asyncio
# nest_asyncio.apply()

import asyncio
from multiprocessing import Process  # , freeze_support

import gradio as gr
import httpx

# from deepl_fastapi.run_uvicorn import main
# from deepl_fastapi_pw.run_uvicorn_async import main
from deepl_fastapi_pw.deepl_server_async import main
from logzero import logger

# from deepl_scraper_pw.deepl_tr import deepl_tr
from split_text import split_text

arun = asyncio.get_event_loop().run_until_complete


def deepl(text, from_lang, to_lang):
    """Translate."""
    try:
        text = str(text).strip()
    except Exception:
        text = ""
    if not text:
        return "Put something there, man."

    # "http://127.0.0.1:8000/text/?q=test%20me&to_lang=zh"
    url = "http://127.0.0.1:8000/text/"
    url = "http://127.0.0.1:8001/text/"

    res_tot = ""
    text_tot = text[:]
    slash = "_slash_"
    text_tot = text_tot.replace("/", slash)

    _ = split_text(text_tot)
    len_ = len(_)
    logger.info("Split to %s", len_)
    for idx, chunk in enumerate(_):
        logger.info("%s/%s", idx + 1, len_)
        try:
            # resp = httpx.get(f"{url}?q={text}&from_lang={from_lang}&to_lang={to_lang}")
            resp = httpx.post(
                url,
                json={
                    "text": chunk,
                    "from_lang": from_lang,
                    "to_lang": to_lang,
                },
                timeout=90,
            )
            resp.raise_for_status()
        except Exception as exc:
            logger.error(exc)
            return str(exc)

        try:
            jdata = resp.json()
        except Exception as exc:
            logger.error(exc)
            return str(exc)

        try:
            # res = jdata.get("trtext")
            res = jdata.get("result")
        except Exception as exc:
            logger.error(exc)
            res = str(exc)

        if res_tot:
            res_tot = res_tot + "\n" + res
        else:
            res_tot = res

    _ = """
    try:
        # _ = deepl_tr(text, from_lang, to_lang)
        _ = arun(deepl_tr(text, from_lang, to_lang))
    except Exception as exc:
        logger.error(exc)
        return str(exc)
    # """

    # return res

    res_tot = res_tot.replace(slash, "/")

    return res_tot


if __name__ == "__main__":
    # freeze_support()
    Process(target=main).start()

    iface = gr.Interface(
        fn=deepl,
        inputs=[
            gr.Textbox(
                placeholder="Paste text here (not limited to 5000 chars)",
                lines=7,
            ),
            gr.Textbox(label="from_lang", value="en", lines=1),
            gr.Textbox(label="to_lang", value="zh", lines=1),
        ],
        outputs="textarea",
        allow_flagging="never",
    )
    iface.launch()