asyncio 速度比 ThreadPool 慢？DeepL Pro 实测告诉你原因。
```python"""Translate using deeplpro's dl_session cookie."""
# pylint: disable=broad-exception-caught
import asyncio
import sys
from time import monotonic
from random import randrange
from typing import Union
from concurrent.futures import ThreadPoolExecutor
import httpx
from loadtext import loadtext
COOKIE = "dl_session=81f94b98-497f-4f18-ab79-1ef25076f4c9"
HEADERS = {"Content-Type": "application/json", "Cookie": COOKIE}
URL = "https://api.deepl.com/jsonrpc"
def deepl_tr(text: str, source_lang: str = "auto", target_lang: str = "zh") -> Union:
"""Translate using deeplpro's dl_session cookie."""
data = {
"jsonrpc": "2.0",
"method": "LMT_handle_texts",
"id": randrange(sys.maxsize),
"params": {
"splitting": "newlines",
"lang": {"source_lang_user_selected": source_lang, "target_lang": target_lang},
"texts": [{"text": text, "requestAlternatives": 3}],
},
}
try:
_ = httpx.post(URL, json=data, headers=HEADERS)
jdata = _.json()
return jdata
except Exception as exc:
return exc
async def deepl_tr_async(text: str, source_lang: str = "auto", target_lang: str = "zh") -> Union:
"""Translate using deeplpro's dl_session cookie."""
data = {
"jsonrpc": "2.0",
"method": "LMT_handle_texts",
"id": randrange(sys.maxsize),
"params": {
"splitting": "newlines",
"lang": {"source_lang_user_selected": source_lang, "target_lang": target_lang},
"texts": [{"text": text, "requestAlternatives": 3}],
},
}
async with httpx.AsyncClient() as client:
try:
_ = await client.post(URL, json=data, headers=HEADERS)
jdata = _.json()
return jdata
except Exception as exc:
return exc
def main():
"""Run."""
texts = loadtext(r"2024-08-20.txt")
then = monotonic()
with ThreadPoolExecutor() as executor:
_ = executor.map(deepl_tr, texts)
print(*_)
time_el = monotonic() - then
print(f"{len(texts)}, {time_el:.2f} {time_el / len(texts):.2f}")
async def main_a():
"""Run async."""
texts = loadtext(r"2024-08-20.txt")
then = monotonic()
coros =
_ = await asyncio.gather(*coros)
print(_)
time_el = monotonic() - then
print(f"{len(texts)}, {time_el:.2f} {time_el / len(texts):.2f}")
if __name__ == "__main__":
asyncio.run(main_a())
main()
```
页:
[1]