-
Notifications
You must be signed in to change notification settings - Fork 12
/
run_webapi.py
168 lines (124 loc) · 5.2 KB
/
run_webapi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
# ----------
from fastapi import FastAPI, HTTPException
from starlette.responses import Response, HTMLResponse
import uvicorn
import multiprocessing
from starlette.staticfiles import StaticFiles
from oneringcore import OneRingCore, version
import asyncio
app = FastAPI()
webapi_version = "2.1"
core:OneRingCore = None
app.mount("/webapi_client", StaticFiles(directory="webapi_client", html = True), name="webapi_client")
@app.get("/", response_class=HTMLResponse)
async def read_items():
html_content = f"""
<html>
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>OneRingTranslator</title>
<link rel="stylesheet" href="/webapi_client/chota.min.css">
</head>
<body>
<div id="top" class="container" role="document">
<h1>OneRingTranslator {version}</h1>
<a href="/webapi_client" class="button">Web interface (simple)</a><br /><br />
<a href="/docs" class="button">API and docs</a><br /><br />
<a href="https://github.com/janvarev/OneRingTranslator" class="button" target="_blank">Github</a><br /><br />
</div>
</body>
</html>
"""
return HTMLResponse(content=html_content, status_code=200)
@app.on_event("startup")
async def startup_event():
global core
core = OneRingCore()
core.init_with_plugins()
pass
@app.get(
"/translate",
# Set what the media type will be in the autogenerated OpenAPI specification.
# fastapi.tiangolo.com/advanced/additional-responses/#additional-media-types-for-the-main-response
# responses = {
# 200: {
# "content": {"text": {}}
# }
# },
# Prevent FastAPI from adding "application/json" as an additional
# response media type in the autogenerated OpenAPI specification.
# https://github.com/tiangolo/fastapi/issues/3258
# response_class=Response
)
async def translate(text:str, from_lang:str = "", to_lang:str = "", translator_plugin:str = "", add_params:str = "", api_key:str = ""):
"""
Return translation
:param str text: text to translate
:param str from_lang: from language (2 symbols, like "en"). May be "user" (will be replaced to "user_lang" from options)
:param str to_lang: to language (2 symbols, like "en"). May be "user" (will be replaced to "user_lang" from options)
:param str translator_plugin: to use. If blank, default will be used. If not inited plugin will call, core try to init plugin
:param str add_params: additional params for translation (depends on plugin)
:param str api_key: api key for access (if service setup in security mode with api keys)
:return: dict (result: text)
"""
#return Response(content=data, media_type="text/wav")
if len(core.api_keys_allowed) > 0: # there are some api keys
if api_key == "":
return {"error": "API key required"}
if not (api_key in core.api_keys_allowed):
return {"error": "No valid API key provided"}
if core.is_multithread:
#print("Multithread")
#res = await asyncio.to_thread(core.translators[translator_plugin][1], core, text, from_lang, to_lang, add_params)
# init before other threads will run
if translator_plugin != "":
core.init_translator_engine(translator_plugin)
if translator_plugin not in core.inited_translator_engines:
return {"error": "Translator plugin not inited"}
res = await asyncio.to_thread(core.translate, text, from_lang, to_lang, translator_plugin,
add_params)
else:
res = core.translate(text,from_lang,to_lang,translator_plugin,add_params)
# import time
# time.sleep(1)
return res #{"result": res}
@app.get(
"/translator_plugin_info",
)
async def translator_plugin_info(api_key:str = ""):
"""
Return list of available translator plugins
:param int api_key: api key for access (if service setup in security mode with api keys)
:return: dict with info
"""
if len(core.api_keys_allowed) > 0: # there are some api keys
if api_key == "":
return {"error": "API key required"}
if not (api_key in core.api_keys_allowed):
return {"error": "No valid API key provided"}
full_list = list(core.translators.keys())
inited_list = core.inited_translator_engines
return {"result": {
"default": core.default_translator,
"all_translator": full_list,
"inited_translator": inited_list,
"user_lang": core.user_lang
}}
@app.get(
"/whois",
)
async def whois():
"""
Return whois service info
:return: dict with info
"""
from oneringcore import version
return {"result": {
"service": "OneRingTranslator",
"version": version,
}}
if __name__ == "__main__":
#multiprocessing.freeze_support()
print("Running OneRingTranslator v{0}, web server v{1}...".format(version, webapi_version))
uvicorn.run("run_webapi:app", host="127.0.0.1", port=4990, log_level="info")