25
25
26
26
import json
27
27
import random
28
- import urllib .request
29
28
from functools import partial
30
29
from typing import TYPE_CHECKING , Any , Final , TypeVar , cast
31
- from urllib .parse import urlencode
32
30
33
31
import httpx
32
+ import orjson
34
33
import trio
35
34
36
35
from subtitle_translate import agents
@@ -70,22 +69,22 @@ async def collect(
70
69
## return 'http://clients5.google.com/translate_a/t?'+urlencode(query)
71
70
72
71
73
- def get_translation_url (
74
- sentence : str ,
75
- to_language : str ,
76
- source_language : str = "auto" ,
77
- ) -> str :
78
- """Return the URL you should visit to get query translated to language to_language."""
79
- query = {
80
- "client" : "gtx" ,
81
- "dt" : "t" ,
82
- "sl" : source_language ,
83
- "tl" : to_language ,
84
- "q" : sentence ,
85
- }
86
- return "https://translate.googleapis.com/translate_a/single?" + urlencode (
87
- query ,
88
- )
72
+ ## def get_translation_url(
73
+ ## sentence: str,
74
+ ## to_language: str,
75
+ ## source_language: str = "auto",
76
+ ## ) -> str:
77
+ ## """Return the URL you should visit to get query translated to language to_language."""
78
+ ## query = {
79
+ ## "client": "gtx",
80
+ ## "dt": "t",
81
+ ## "sl": source_language,
82
+ ## "tl": to_language,
83
+ ## "q": sentence,
84
+ ## }
85
+ ## return "https://translate.googleapis.com/translate_a/single?" + urlencode(
86
+ ## query,
87
+ ## )
89
88
90
89
91
90
def process_response (result : list [str ] | list [list [Any ]]) -> str :
@@ -99,7 +98,7 @@ def process_response(result: list[str] | list[list[Any]]) -> str:
99
98
part = next_
100
99
else :
101
100
raise ValueError (
102
- f"Unexpected type { type (part )!r} , expected list or str" ,
101
+ f"Unexpected type { type (part )!r} , expected list or str\n { part = } " ,
103
102
)
104
103
raise RuntimeError ("Unreachable" )
105
104
@@ -114,23 +113,65 @@ def is_url(text: str) -> bool:
114
113
)
115
114
116
115
117
- def translate_sync (
118
- sentence : str ,
119
- to_lang : str ,
120
- source_lang : str = "auto" ,
121
- ) -> str :
122
- """Return sentence translated from source_lang to to_lang."""
123
- if is_url (sentence ):
124
- # skip URLs
125
- return sentence
126
-
127
- # Get URL from function, which uses urllib to generate proper query
128
- url = get_translation_url (sentence , to_lang , source_lang )
129
- if not url .startswith ("http" ):
130
- raise ValueError ("URL not http(s), is this intended?" )
131
- with urllib .request .urlopen (url , timeout = 0.5 ) as file : # noqa: S310
132
- request_result = json .loads (file .read ())
133
- return process_response (request_result )
116
+ ##def translate_sync(
117
+ ## sentence: str,
118
+ ## to_lang: str,
119
+ ## source_lang: str = "auto",
120
+ ##) -> str:
121
+ ## """Return sentence translated from source_lang to to_lang."""
122
+ ## if is_url(sentence):
123
+ ## # skip URLs
124
+ ## return sentence
125
+ ##
126
+ ## # Get URL from function, which uses urllib to generate proper query
127
+ ## url = get_translation_url(sentence, to_lang, source_lang)
128
+ ## if not url.startswith("http"):
129
+ ## raise ValueError("URL not http(s), is this intended?")
130
+ ## with urllib.request.urlopen(url, timeout=0.5) as file:
131
+ ## request_result = json.loads(file.read())
132
+ ## return process_response(request_result)
133
+
134
+
135
+ ##async def get_translated_coroutine(
136
+ ## client: httpx.AsyncClient,
137
+ ## sentence: str,
138
+ ## to_lang: str,
139
+ ## source_lang: str = "auto",
140
+ ##) -> str:
141
+ ## """Return the sentence translated, asynchronously."""
142
+ ## global AGENT # pylint: disable=global-statement
143
+ ##
144
+ ## if is_url(sentence):
145
+ ## # skip URLs
146
+ ## return sentence
147
+ ## # Make sure we have a timeout, so that in the event of network failures
148
+ ## # or something code doesn't get stuck
149
+ ## # Get URL from function, which uses urllib to generate proper query
150
+ ## url = get_translation_url(sentence, to_lang, source_lang)
151
+ ##
152
+ ## headers = {
153
+ ## "User-Agent": "",
154
+ ## "Accept": "*/*",
155
+ ## "Accept-Language": "en-US,en-GB; q=0.5",
156
+ ## "Accept-Encoding": "gzip, deflate",
157
+ ## "Connection": "keep-alive",
158
+ ## }
159
+ ##
160
+ ## while True:
161
+ ## AGENT = (AGENT + 1) % len(agents.USER_AGENTS)
162
+ ## headers["User-Agent"] = agents.USER_AGENTS[AGENT]
163
+ ##
164
+ ## try:
165
+ ## # Go to that URL and get our translated response
166
+ ## response = await client.get(url, headers=headers)
167
+ ## # Wait for our response and make it json so we can look at
168
+ ## # it like a dictionary
169
+ ## return process_response(response.json())
170
+ ## except httpx.ConnectTimeout:
171
+ ## pass
172
+ ## except json.decoder.JSONDecodeError:
173
+ ## print(f"{type(response) = }\n{response = }")
174
+ ## raise
134
175
135
176
136
177
async def get_translated_coroutine (
@@ -148,31 +189,33 @@ async def get_translated_coroutine(
148
189
# Make sure we have a timeout, so that in the event of network failures
149
190
# or something code doesn't get stuck
150
191
# Get URL from function, which uses urllib to generate proper query
151
- url = get_translation_url ( sentence , to_lang , source_lang )
192
+ url = "https://translate-pa.googleapis.com/v1/translateHtml"
152
193
153
194
headers = {
154
- "User-Agent" : "" ,
155
195
"Accept" : "*/*" ,
156
- "Accept-Language" : "en-US,en-GB; q=0.5" ,
157
- "Accept-Encoding" : "gzip, deflate" ,
158
- "Connection" : "keep-alive" ,
196
+ "X-Goog-API-Key" : "AIzaSyATBXajvzQLTDHEQbcpq0Ihe0vWDHmO520" ,
197
+ "Content-Type" : "application/json+protobuf" ,
159
198
}
160
199
161
- while True :
162
- AGENT = (AGENT + 1 ) % len (agents .USER_AGENTS )
163
- headers ["User-Agent" ] = agents .USER_AGENTS [AGENT ]
200
+ sentence = sentence .replace ("\n " , "<br>" )
201
+
202
+ raw_content = [[[sentence ], source_lang , to_lang ], "wt_lib" ]
203
+ content = orjson .dumps (raw_content )
164
204
205
+ while True :
165
206
try :
166
207
# Go to that URL and get our translated response
167
- response = await client .get (url , headers = headers )
208
+ response = await client .post (url , content = content , headers = headers )
168
209
# Wait for our response and make it json so we can look at
169
210
# it like a dictionary
170
- return process_response (response .json ())
211
+ return process_response (response .json ()). replace ( "<br>" , " \n " )
171
212
except httpx .ConnectTimeout :
172
213
pass
173
214
except json .decoder .JSONDecodeError :
174
215
print (f"{ type (response ) = } \n { response = } " )
175
216
raise
217
+ AGENT = (AGENT + 1 ) % len (agents .USER_AGENTS )
218
+ headers ["User-Agent" ] = agents .USER_AGENTS [AGENT ]
176
219
177
220
178
221
async def translate_async (
0 commit comments