initial commit
This commit is contained in:
f1den committed
1 parent
eeab594b68
commit
e1ed811084
57 files changed
+18704
No files matched your search
@@ -0,0 +1,96 @@
|
||||
import time
|
||||
from DrissionPage import ChromiumPage
|
||||
|
||||
class CloudflareBypasser:
|
||||
def __init__(self, driver: ChromiumPage, max_retries=-1, log=True):
|
||||
self.driver = driver
|
||||
self.max_retries = max_retries
|
||||
self.log = log
|
||||
|
||||
def search_recursively_shadow_root_with_iframe(self,ele):
|
||||
if ele.shadow_root:
|
||||
if ele.shadow_root.child().tag == "iframe":
|
||||
return ele.shadow_root.child()
|
||||
else:
|
||||
for child in ele.children():
|
||||
result = self.search_recursively_shadow_root_with_iframe(child)
|
||||
if result:
|
||||
return result
|
||||
return None
|
||||
|
||||
def search_recursively_shadow_root_with_cf_input(self,ele):
|
||||
if ele.shadow_root:
|
||||
if ele.shadow_root.ele("tag:input"):
|
||||
return ele.shadow_root.ele("tag:input")
|
||||
else:
|
||||
for child in ele.children():
|
||||
result = self.search_recursively_shadow_root_with_cf_input(child)
|
||||
if result:
|
||||
return result
|
||||
return None
|
||||
|
||||
def locate_cf_button(self):
|
||||
button = None
|
||||
eles = self.driver.eles("tag:input")
|
||||
for ele in eles:
|
||||
if "name" in ele.attrs.keys() and "type" in ele.attrs.keys():
|
||||
if "turnstile" in ele.attrs["name"] and ele.attrs["type"] == "hidden":
|
||||
button = ele.parent().shadow_root.child()("tag:body").shadow_root("tag:input")
|
||||
break
|
||||
|
||||
if button:
|
||||
return button
|
||||
else:
|
||||
# If the button is not found, search it recursively
|
||||
self.log_message("Basic search failed. Searching for button recursively.")
|
||||
ele = self.driver.ele("tag:body")
|
||||
iframe = self.search_recursively_shadow_root_with_iframe(ele)
|
||||
if iframe:
|
||||
button = self.search_recursively_shadow_root_with_cf_input(iframe("tag:body"))
|
||||
else:
|
||||
self.log_message("Iframe not found. Button search failed.")
|
||||
return button
|
||||
|
||||
def log_message(self, message):
|
||||
if self.log:
|
||||
print(message)
|
||||
|
||||
def click_verification_button(self):
|
||||
try:
|
||||
button = self.locate_cf_button()
|
||||
if button:
|
||||
self.log_message("Verification button found. Attempting to click.")
|
||||
button.click()
|
||||
else:
|
||||
self.log_message("Verification button not found.")
|
||||
|
||||
except Exception as e:
|
||||
self.log_message(f"Error clicking verification button: {e}")
|
||||
|
||||
def is_bypassed(self):
|
||||
try:
|
||||
title = self.driver.title.lower()
|
||||
return "just a moment" not in title
|
||||
except Exception as e:
|
||||
self.log_message(f"Error checking page title: {e}")
|
||||
return False
|
||||
|
||||
def bypass(self):
|
||||
|
||||
try_count = 0
|
||||
|
||||
while not self.is_bypassed():
|
||||
if 0 < self.max_retries + 1 <= try_count:
|
||||
self.log_message("Exceeded maximum retries. Bypass failed.")
|
||||
break
|
||||
|
||||
self.log_message(f"Attempt {try_count + 1}: Verification page detected. Trying to bypass...")
|
||||
self.click_verification_button()
|
||||
|
||||
try_count += 1
|
||||
time.sleep(2)
|
||||
|
||||
if self.is_bypassed():
|
||||
self.log_message("Bypass successful.")
|
||||
else:
|
||||
self.log_message("Bypass failed.")
|
||||
@@ -0,0 +1,294 @@
|
||||
from curl_cffi import requests
|
||||
from typing import Optional, Dict, Any, Generator, Literal
|
||||
import json
|
||||
from .pow import DeepSeekPOW
|
||||
import pkg_resources
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
ThinkingMode = Literal['detailed', 'simple', 'disabled']
|
||||
SearchMode = Literal['enabled', 'disabled']
|
||||
|
||||
class DeepSeekError(Exception):
|
||||
"""Base exception for all DeepSeek API errors"""
|
||||
pass
|
||||
|
||||
class AuthenticationError(DeepSeekError):
|
||||
"""Raised when authentication fails"""
|
||||
pass
|
||||
|
||||
class RateLimitError(DeepSeekError):
|
||||
"""Raised when API rate limit is exceeded"""
|
||||
pass
|
||||
|
||||
class NetworkError(DeepSeekError):
|
||||
"""Raised when network communication fails"""
|
||||
pass
|
||||
|
||||
class CloudflareError(DeepSeekError):
|
||||
"""Raised when Cloudflare blocks the request"""
|
||||
pass
|
||||
|
||||
class APIError(DeepSeekError):
|
||||
"""Raised when API returns an error response"""
|
||||
def __init__(self, message: str, status_code: Optional[int] = None):
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
|
||||
class DeepSeekAPI:
|
||||
BASE_URL = "https://chat.deepseek.com/api/v0"
|
||||
|
||||
def __init__(self, auth_token: str):
|
||||
if not auth_token or not isinstance(auth_token, str):
|
||||
raise AuthenticationError("Invalid auth token provided")
|
||||
|
||||
try:
|
||||
curl_cffi_version = pkg_resources.get_distribution('curl-cffi').version
|
||||
if curl_cffi_version != '0.8.1b9':
|
||||
print("\033[93mWarning: DeepSeek API requires curl-cffi version 0.8.1b9", file=sys.stderr)
|
||||
print("Please install the correct version using: pip install curl-cffi==0.8.1b9\033[0m", file=sys.stderr)
|
||||
except pkg_resources.DistributionNotFound:
|
||||
print("\033[93mWarning: curl-cffi not found. Please install version 0.8.1b9:", file=sys.stderr)
|
||||
print("pip install curl-cffi==0.8.1b9\033[0m", file=sys.stderr)
|
||||
|
||||
self.auth_token = auth_token
|
||||
self.pow_solver = DeepSeekPOW()
|
||||
|
||||
# Load cookies from JSON file
|
||||
cookies_path = Path(__file__).parent / 'cookies.json'
|
||||
try:
|
||||
with open(cookies_path, 'r') as f:
|
||||
cookie_data = json.load(f)
|
||||
self.cookies = cookie_data.get('cookies', {})
|
||||
except (FileNotFoundError, json.JSONDecodeError) as e:
|
||||
print(f"\033[93mWarning: Could not load cookies from {cookies_path}: {e}\033[0m", file=sys.stderr)
|
||||
self.cookies = {}
|
||||
|
||||
def _get_headers(self, pow_response: Optional[str] = None) -> Dict[str, str]:
|
||||
headers = {
|
||||
'accept': '*/*',
|
||||
'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
|
||||
'authorization': f'Bearer {self.auth_token}',
|
||||
'content-type': 'application/json',
|
||||
'origin': 'https://chat.deepseek.com',
|
||||
'referer': 'https://chat.deepseek.com/',
|
||||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36',
|
||||
'x-app-version': '20241129.1',
|
||||
'x-client-locale': 'en_US',
|
||||
'x-client-platform': 'web',
|
||||
'x-client-version': '1.0.0-always',
|
||||
}
|
||||
|
||||
if pow_response:
|
||||
headers['x-ds-pow-response'] = pow_response
|
||||
|
||||
return headers
|
||||
|
||||
def _refresh_cookies(self) -> None:
|
||||
"""Run the cookie refresh script and reload cookies"""
|
||||
try:
|
||||
# Get path to bypass.py
|
||||
script_path = Path(__file__).parent / 'bypass.py'
|
||||
|
||||
# Run the script
|
||||
subprocess.run([sys.executable, script_path], check=True)
|
||||
|
||||
# Wait briefly for cookies file to be written
|
||||
time.sleep(2)
|
||||
|
||||
# Reload cookies
|
||||
cookies_path = Path(__file__).parent / 'cookies.json'
|
||||
with open(cookies_path, 'r') as f:
|
||||
cookie_data = json.load(f)
|
||||
self.cookies = cookie_data.get('cookies', {})
|
||||
|
||||
except Exception as e:
|
||||
print(f"\033[93mWarning: Failed to refresh cookies: {e}\033[0m", file=sys.stderr)
|
||||
|
||||
def _make_request(self, method: str, endpoint: str, json_data: Dict[str, Any], pow_required: bool = False) -> Any:
|
||||
url = f"{self.BASE_URL}{endpoint}"
|
||||
|
||||
retry_count = 0
|
||||
max_retries = 2
|
||||
|
||||
while retry_count < max_retries:
|
||||
try:
|
||||
headers = self._get_headers()
|
||||
if pow_required:
|
||||
challenge = self._get_pow_challenge()
|
||||
pow_response = self.pow_solver.solve_challenge(challenge)
|
||||
headers = self._get_headers(pow_response)
|
||||
|
||||
response = requests.request(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=headers,
|
||||
json=json_data,
|
||||
cookies=self.cookies,
|
||||
impersonate='chrome120',
|
||||
timeout=None
|
||||
)
|
||||
|
||||
# Check if we hit Cloudflare protection
|
||||
if "<!DOCTYPE html>" in response.text and "Just a moment" in response.text:
|
||||
print("\033[93mWarning: Cloudflare protection detected. Bypassing...\033[0m", file=sys.stderr)
|
||||
if retry_count < max_retries - 1:
|
||||
self._refresh_cookies() # Refresh cookies
|
||||
retry_count += 1
|
||||
continue
|
||||
|
||||
# Handle other response codes
|
||||
if response.status_code == 401:
|
||||
raise AuthenticationError("Invalid or expired authentication token")
|
||||
elif response.status_code == 429:
|
||||
raise RateLimitError("API rate limit exceeded")
|
||||
elif response.status_code >= 500:
|
||||
raise APIError(f"Server error occurred: {response.text}", response.status_code)
|
||||
elif response.status_code != 200:
|
||||
raise APIError(f"API request failed: {response.text}", response.status_code)
|
||||
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error occurred: {str(e)}")
|
||||
except json.JSONDecodeError:
|
||||
raise APIError("Invalid JSON response from server")
|
||||
|
||||
raise APIError("Failed to bypass Cloudflare protection after multiple attempts")
|
||||
|
||||
def _get_pow_challenge(self) -> Dict[str, Any]:
|
||||
try:
|
||||
response = self._make_request(
|
||||
'POST',
|
||||
'/chat/create_pow_challenge',
|
||||
{'target_path': '/api/v0/chat/completion'}
|
||||
)
|
||||
return response['data']['biz_data']['challenge']
|
||||
except KeyError:
|
||||
raise APIError("Invalid challenge response format from server")
|
||||
|
||||
def create_chat_session(self) -> str:
|
||||
"""Creates a new chat session and returns the session ID"""
|
||||
try:
|
||||
response = self._make_request(
|
||||
'POST',
|
||||
'/chat_session/create',
|
||||
{'character_id': None}
|
||||
)
|
||||
return response['data']['biz_data']['id']
|
||||
except KeyError:
|
||||
raise APIError("Invalid session creation response format from server")
|
||||
|
||||
def chat_completion(self,
|
||||
chat_session_id: str,
|
||||
prompt: str,
|
||||
parent_message_id: Optional[str] = None,
|
||||
thinking_enabled: bool = True,
|
||||
search_enabled: bool = False) -> Generator[Dict[str, Any], None, None]:
|
||||
"""
|
||||
Send a message and get streaming response
|
||||
|
||||
Args:
|
||||
chat_session_id (str): The ID of the chat session
|
||||
prompt (str): The message to send
|
||||
parent_message_id (Optional[str]): ID of the parent message for threading
|
||||
thinking_enabled (bool): Whether to show the thinking process
|
||||
search_enabled (bool): Whether to enable web search for up-to-date information
|
||||
|
||||
Returns:
|
||||
Generator[Dict[str, Any], None, None]: Yields message chunks with content and type
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If the authentication token is invalid
|
||||
RateLimitError: If the API rate limit is exceeded
|
||||
NetworkError: If a network error occurs
|
||||
APIError: If any other API error occurs
|
||||
"""
|
||||
if not prompt or not isinstance(prompt, str):
|
||||
raise ValueError("Prompt must be a non-empty string")
|
||||
if not chat_session_id or not isinstance(chat_session_id, str):
|
||||
raise ValueError("Chat session ID must be a non-empty string")
|
||||
|
||||
json_data = {
|
||||
'chat_session_id': chat_session_id,
|
||||
'parent_message_id': parent_message_id,
|
||||
'prompt': prompt,
|
||||
'ref_file_ids': [],
|
||||
'thinking_enabled': thinking_enabled,
|
||||
'search_enabled': search_enabled,
|
||||
}
|
||||
|
||||
try:
|
||||
headers = self._get_headers(
|
||||
pow_response=self.pow_solver.solve_challenge(
|
||||
self._get_pow_challenge()
|
||||
)
|
||||
)
|
||||
|
||||
response = requests.post(
|
||||
f"{self.BASE_URL}/chat/completion",
|
||||
headers=headers,
|
||||
json=json_data,
|
||||
cookies=self.cookies, # Add cookies
|
||||
impersonate='chrome120',
|
||||
stream=True,
|
||||
timeout=None
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
error_text = next(response.iter_lines(), b'').decode('utf-8', 'ignore')
|
||||
if response.status_code == 401:
|
||||
raise AuthenticationError("Invalid or expired authentication token")
|
||||
elif response.status_code == 429:
|
||||
raise RateLimitError("API rate limit exceeded")
|
||||
else:
|
||||
raise APIError(f"API request failed: {error_text}", response.status_code)
|
||||
|
||||
for chunk in response.iter_lines():
|
||||
try:
|
||||
parsed = self._parse_chunk(chunk)
|
||||
if parsed:
|
||||
yield parsed
|
||||
# print(parsed)
|
||||
if parsed.get('finish_reason') == 'stop':
|
||||
break
|
||||
except Exception as e:
|
||||
raise APIError(f"Error parsing response chunk: {str(e)}")
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error occurred during streaming: {str(e)}")
|
||||
|
||||
def _parse_chunk(self, chunk: bytes) -> Optional[Dict[str, Any]]:
|
||||
"""Parse a SSE chunk from the API response"""
|
||||
if not chunk:
|
||||
return None
|
||||
|
||||
try:
|
||||
if chunk.startswith(b'data: '):
|
||||
data = json.loads(chunk[6:])
|
||||
|
||||
if 'v' in data and isinstance(data['v'], str):
|
||||
return {
|
||||
'content': data['v'],
|
||||
'type': 'message',
|
||||
'finish_reason': 'finish_reason'
|
||||
}
|
||||
|
||||
if 'choices' in data and data['choices']:
|
||||
choice = data['choices'][0]
|
||||
if 'delta' in choice:
|
||||
delta = choice['delta']
|
||||
|
||||
return {
|
||||
'content': delta.get('content', ''),
|
||||
'type': delta.get('type', ''),
|
||||
'finish_reason': choice.get('finish_reason')
|
||||
}
|
||||
except json.JSONDecodeError:
|
||||
raise APIError("Invalid JSON in response chunk")
|
||||
except Exception as e:
|
||||
raise APIError(f"Error parsing chunk: {str(e)}")
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,87 @@
|
||||
# run_and_get_cookies.py
|
||||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import requests
|
||||
import json
|
||||
|
||||
def validate_cookies(cookies_data):
|
||||
"""Validate that cf_clearance cookie is present and not empty"""
|
||||
cookies = cookies_data.get('cookies', {})
|
||||
return 'cf_clearance' in cookies and cookies['cf_clearance'].strip() != ''
|
||||
|
||||
def get_and_save_cookies(server_url, cookie_file_path, max_retries=3):
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
response = requests.get(server_url)
|
||||
response.raise_for_status()
|
||||
cookies_data = response.json()
|
||||
|
||||
if not validate_cookies(cookies_data):
|
||||
print(f"Attempt {attempt + 1}: cf_clearance cookie not found, retrying...")
|
||||
time.sleep(5)
|
||||
continue
|
||||
|
||||
cookies_to_save = {
|
||||
'cookies': cookies_data.get('cookies', {}),
|
||||
'user_agent': cookies_data.get('user_agent', '')
|
||||
}
|
||||
|
||||
os.makedirs(os.path.dirname(cookie_file_path), exist_ok=True)
|
||||
with open(cookie_file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(cookies_to_save, f, indent=4, ensure_ascii=False)
|
||||
print("Successfully obtained and saved cookies with cf_clearance!")
|
||||
return True
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
print(f"Connection error on attempt {attempt + 1}: {str(e)}")
|
||||
if attempt < max_retries - 1:
|
||||
time.sleep(5)
|
||||
else:
|
||||
print("Max retries reached. Failed to get valid cookies.")
|
||||
return False
|
||||
|
||||
print("Failed to obtain valid cf_clearance cookie after all attempts")
|
||||
return False
|
||||
|
||||
def run_server_background():
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
server_script = os.path.abspath(os.path.join(script_dir, "server.py"))
|
||||
server_dir = os.path.dirname(server_script)
|
||||
|
||||
os.makedirs(server_dir, exist_ok=True)
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
[sys.executable, server_script],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
cwd=server_dir,
|
||||
start_new_session=True
|
||||
)
|
||||
return process
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Getting the cookies...")
|
||||
server_process = run_server_background()
|
||||
|
||||
if server_process:
|
||||
# Increase initial wait time to ensure server is fully started
|
||||
time.sleep(10)
|
||||
server_url = "http://localhost:8000/cookies?url=https://chat.deepseek.com"
|
||||
cookie_file = "dsk/cookies.json"
|
||||
|
||||
# Increase max retries for more reliability
|
||||
success = get_and_save_cookies(server_url, cookie_file, max_retries=5)
|
||||
|
||||
if not success:
|
||||
print("Failed to obtain valid cookies.")
|
||||
server_process.terminate()
|
||||
sys.exit(1)
|
||||
server_process.terminate()
|
||||
else:
|
||||
print("Failed to start server.")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,111 @@
|
||||
"""
|
||||
DeepSeek Proof of Work Challenge Implementation
|
||||
Author: @xtekky
|
||||
Date: 2024
|
||||
|
||||
This module implements a proof-of-work challenge solver using WebAssembly (WASM)
|
||||
for Custom sha3 hashing. It provides functionality to solve computational challenges
|
||||
required for authentication or rate limiting purposes.
|
||||
"""
|
||||
|
||||
import json
|
||||
import base64
|
||||
import wasmtime
|
||||
import numpy as np
|
||||
from typing import Dict, Any
|
||||
import os
|
||||
|
||||
WASM_PATH = f'{os.path.dirname(__file__)}/wasm/sha3_wasm_bg.7b9ca65ddd.wasm'
|
||||
|
||||
class DeepSeekHash:
|
||||
def __init__(self):
|
||||
self.instance = None
|
||||
self.memory = None
|
||||
self.store = None
|
||||
|
||||
def init(self, wasm_path: str):
|
||||
engine = wasmtime.Engine()
|
||||
|
||||
with open(wasm_path, 'rb') as f:
|
||||
wasm_bytes = f.read()
|
||||
|
||||
module = wasmtime.Module(engine, wasm_bytes)
|
||||
|
||||
self.store = wasmtime.Store(engine)
|
||||
linker = wasmtime.Linker(engine)
|
||||
linker.define_wasi()
|
||||
|
||||
self.instance = linker.instantiate(self.store, module)
|
||||
self.memory = self.instance.exports(self.store)["memory"]
|
||||
|
||||
return self
|
||||
|
||||
def _write_to_memory(self, text: str) -> tuple[int, int]:
|
||||
encoded = text.encode('utf-8')
|
||||
length = len(encoded)
|
||||
ptr = self.instance.exports(self.store)["__wbindgen_export_0"](self.store, length, 1)
|
||||
|
||||
memory_view = self.memory.data_ptr(self.store)
|
||||
for i, byte in enumerate(encoded):
|
||||
memory_view[ptr + i] = byte
|
||||
|
||||
return ptr, length
|
||||
|
||||
def calculate_hash(self, algorithm: str, challenge: str, salt: str,
|
||||
difficulty: int, expire_at: int) -> float:
|
||||
|
||||
prefix = f"{salt}_{expire_at}_"
|
||||
retptr = self.instance.exports(self.store)["__wbindgen_add_to_stack_pointer"](self.store, -16)
|
||||
|
||||
try:
|
||||
challenge_ptr, challenge_len = self._write_to_memory(challenge)
|
||||
prefix_ptr, prefix_len = self._write_to_memory(prefix)
|
||||
|
||||
self.instance.exports(self.store)["wasm_solve"](
|
||||
self.store,
|
||||
retptr,
|
||||
challenge_ptr,
|
||||
challenge_len,
|
||||
prefix_ptr,
|
||||
prefix_len,
|
||||
float(difficulty)
|
||||
)
|
||||
|
||||
memory_view = self.memory.data_ptr(self.store)
|
||||
status = int.from_bytes(bytes(memory_view[retptr:retptr + 4]), byteorder='little', signed=True)
|
||||
|
||||
if status == 0:
|
||||
return None
|
||||
|
||||
value_bytes = bytes(memory_view[retptr + 8:retptr + 16])
|
||||
value = np.frombuffer(value_bytes, dtype=np.float64)[0]
|
||||
|
||||
return int(value)
|
||||
|
||||
finally:
|
||||
self.instance.exports(self.store)["__wbindgen_add_to_stack_pointer"](self.store, 16)
|
||||
|
||||
class DeepSeekPOW:
|
||||
def __init__(self):
|
||||
self.hasher = DeepSeekHash().init(WASM_PATH)
|
||||
|
||||
def solve_challenge(self, config: Dict[str, Any]) -> str:
|
||||
"""Solves a proof-of-work challenge and returns the encoded response"""
|
||||
answer = self.hasher.calculate_hash(
|
||||
config['algorithm'],
|
||||
config['challenge'],
|
||||
config['salt'],
|
||||
config['difficulty'],
|
||||
config['expire_at']
|
||||
)
|
||||
|
||||
result = {
|
||||
'algorithm': config['algorithm'],
|
||||
'challenge': config['challenge'],
|
||||
'salt': config['salt'],
|
||||
'answer': answer,
|
||||
'signature': config['signature'],
|
||||
'target_path': config['target_path']
|
||||
}
|
||||
|
||||
return base64.b64encode(json.dumps(result).encode()).decode()
|
||||
@@ -0,0 +1,61 @@
|
||||
# run_and_get_cookies.py
|
||||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import requests
|
||||
import json
|
||||
|
||||
def get_and_save_cookies(server_url, cookie_file_path):
|
||||
for attempt in range(5):
|
||||
try:
|
||||
response = requests.get(server_url)
|
||||
response.raise_for_status()
|
||||
cookies_data = response.json()
|
||||
|
||||
cookies_to_save = {
|
||||
'cookies': cookies_data.get('cookies', {}),
|
||||
'user_agent': cookies_data.get('user_agent', '')
|
||||
}
|
||||
|
||||
os.makedirs(os.path.dirname(cookie_file_path), exist_ok=True)
|
||||
with open(cookie_file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(cookies_to_save, f, indent=4, ensure_ascii=False)
|
||||
return
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
if attempt < 4:
|
||||
time.sleep(5)
|
||||
else:
|
||||
raise
|
||||
|
||||
def run_server_background():
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
server_script = os.path.abspath(os.path.join(script_dir, "server.py"))
|
||||
server_dir = os.path.dirname(server_script)
|
||||
|
||||
os.makedirs(server_dir, exist_ok=True)
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
[sys.executable, server_script],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
cwd=server_dir,
|
||||
start_new_session=True
|
||||
)
|
||||
return process
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Getting the cookies...")
|
||||
server_process = run_server_background()
|
||||
|
||||
if server_process:
|
||||
time.sleep(5)
|
||||
server_url = "http://localhost:8000/cookies?url=https://chat.deepseek.com"
|
||||
cookie_file = "dsk/cookies.json"
|
||||
get_and_save_cookies(server_url, cookie_file)
|
||||
else:
|
||||
print("Failed to start server.")
|
||||
@@ -0,0 +1,176 @@
|
||||
import json
|
||||
import re
|
||||
import os
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from CloudflareBypasser import CloudflareBypasser
|
||||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||||
from fastapi import FastAPI, HTTPException, Response
|
||||
from pydantic import BaseModel
|
||||
from typing import Dict
|
||||
import argparse
|
||||
|
||||
from pyvirtualdisplay import Display
|
||||
import uvicorn
|
||||
import atexit
|
||||
import time
|
||||
|
||||
# Check if running in Docker mode
|
||||
DOCKER_MODE = os.getenv("DOCKERMODE", "false").lower() == "true"
|
||||
|
||||
SERVER_PORT = int(os.getenv("SERVER_PORT", 8000))
|
||||
|
||||
# Chromium options arguments
|
||||
arguments = [
|
||||
# "--remote-debugging-port=9222", # Add this line for remote debugging
|
||||
"-no-first-run",
|
||||
"-force-color-profile=srgb",
|
||||
"-metrics-recording-only",
|
||||
"-password-store=basic",
|
||||
"-use-mock-keychain",
|
||||
"-export-tagged-pdf",
|
||||
"-no-default-browser-check",
|
||||
"-disable-background-mode",
|
||||
"-enable-features=NetworkService,NetworkServiceInProcess,LoadCryptoTokenExtension,PermuteTLSExtensions",
|
||||
"-disable-features=FlashDeprecationWarning,EnablePasswordsAccountStorage",
|
||||
"-deny-permission-prompts",
|
||||
"-disable-gpu",
|
||||
"-accept-lang=en-US",
|
||||
#"-incognito" # You can add this line to open the browser in incognito mode by default
|
||||
]
|
||||
|
||||
browser_path = "/usr/bin/google-chrome"
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
# Pydantic model for the response
|
||||
class CookieResponse(BaseModel):
|
||||
cookies: Dict[str, str]
|
||||
user_agent: str
|
||||
|
||||
|
||||
# Function to check if the URL is safe
|
||||
def is_safe_url(url: str) -> bool:
|
||||
parsed_url = urlparse(url)
|
||||
ip_pattern = re.compile(
|
||||
r"^(127\.0\.0\.1|localhost|0\.0\.0\.0|::1|10\.\d+\.\d+\.\d+|172\.1[6-9]\.\d+\.\d+|172\.2[0-9]\.\d+\.\d+|172\.3[0-1]\.\d+\.\d+|192\.168\.\d+\.\d+)$"
|
||||
)
|
||||
hostname = parsed_url.hostname
|
||||
if (hostname and ip_pattern.match(hostname)) or parsed_url.scheme == "file":
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
# Function to verify if the page has loaded properly
|
||||
def verify_page_loaded(driver: ChromiumPage) -> bool:
|
||||
"""Verify if the page has loaded properly"""
|
||||
try:
|
||||
# Wait for body element to be present
|
||||
body = driver.ele('tag:body', timeout=10)
|
||||
# Check if page has actual content
|
||||
return len(body.html) > 100
|
||||
except:
|
||||
return False
|
||||
|
||||
|
||||
# Function to bypass Cloudflare protection
|
||||
def bypass_cloudflare(url: str, retries: int, log: bool, proxy: str = None) -> ChromiumPage:
|
||||
max_load_retries = 3
|
||||
|
||||
for load_attempt in range(max_load_retries):
|
||||
options = ChromiumOptions().auto_port()
|
||||
if DOCKER_MODE:
|
||||
options.set_argument("--auto-open-devtools-for-tabs", "true")
|
||||
options.set_argument("--remote-debugging-port=9222")
|
||||
options.set_argument("--no-sandbox") # Necessary for Docker
|
||||
options.set_argument("--disable-gpu") # Optional, helps in some cases
|
||||
options.set_paths(browser_path=browser_path).headless(False)
|
||||
else:
|
||||
options.set_paths(browser_path=browser_path).headless(False)
|
||||
|
||||
if proxy:
|
||||
options.set_proxy(proxy)
|
||||
|
||||
driver = ChromiumPage(addr_or_opts=options)
|
||||
try:
|
||||
driver.get(url)
|
||||
# Wait for initial page load
|
||||
time.sleep(5)
|
||||
|
||||
if not verify_page_loaded(driver):
|
||||
driver.quit()
|
||||
if load_attempt < max_load_retries - 1:
|
||||
time.sleep(3)
|
||||
continue
|
||||
else:
|
||||
raise Exception("Failed to load page properly after multiple attempts")
|
||||
|
||||
cf_bypasser = CloudflareBypasser(driver, retries, log)
|
||||
cf_bypasser.bypass()
|
||||
return driver
|
||||
except Exception as e:
|
||||
driver.quit()
|
||||
if load_attempt < max_load_retries - 1:
|
||||
time.sleep(3)
|
||||
continue
|
||||
raise e
|
||||
|
||||
|
||||
# Endpoint to get cookies
|
||||
@app.get("/cookies", response_model=CookieResponse)
|
||||
async def get_cookies(url: str, retries: int = 5, proxy: str = None):
|
||||
if not is_safe_url(url):
|
||||
raise HTTPException(status_code=400, detail="Invalid URL")
|
||||
try:
|
||||
driver = bypass_cloudflare(url, retries, log, proxy)
|
||||
cookies = {cookie.get("name", ""): cookie.get("value", " ") for cookie in driver.cookies()}
|
||||
user_agent = driver.user_agent
|
||||
driver.quit()
|
||||
return CookieResponse(cookies=cookies, user_agent=user_agent)
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# Endpoint to get HTML content and cookies
|
||||
@app.get("/html")
|
||||
async def get_html(url: str, retries: int = 5, proxy: str = None):
|
||||
if not is_safe_url(url):
|
||||
raise HTTPException(status_code=400, detail="Invalid URL")
|
||||
try:
|
||||
driver = bypass_cloudflare(url, retries, log, proxy)
|
||||
html = driver.html
|
||||
cookies_json = {cookie.get("name", ""): cookie.get("value", " ") for cookie in driver.cookies()}
|
||||
response = Response(content=html, media_type="text/html")
|
||||
response.headers["cookies"] = json.dumps(cookies_json)
|
||||
response.headers["user_agent"] = driver.user_agent
|
||||
driver.quit()
|
||||
return response
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# Main entry point
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Cloudflare bypass api")
|
||||
|
||||
parser.add_argument("--nolog", action="store_true", help="Disable logging")
|
||||
parser.add_argument("--headless", action="store_true", help="Run in headless mode")
|
||||
|
||||
args = parser.parse_args()
|
||||
display = None
|
||||
|
||||
if args.headless or DOCKER_MODE:
|
||||
display = Display(visible=0, size=(1920, 1080))
|
||||
display.start()
|
||||
|
||||
def cleanup_display():
|
||||
if display:
|
||||
display.stop()
|
||||
atexit.register(cleanup_display)
|
||||
|
||||
if args.nolog:
|
||||
log = False
|
||||
else:
|
||||
log = True
|
||||
|
||||
uvicorn.run(app, host="0.0.0.0", port=SERVER_PORT)
|
||||
Binary file not shown.
@@ -0,0 +1,178 @@
|
||||
|
||||
import sys, os
|
||||
from typing import Generator, Dict, Any
|
||||
from googletrans import Translator, constants
|
||||
|
||||
try: from dsk.api import DeepSeekAPI, AuthenticationError, RateLimitError, NetworkError, APIError
|
||||
except ImportError: from .dsk.api import DeepSeekAPI, AuthenticationError, RateLimitError, NetworkError, APIError
|
||||
|
||||
try: from yat.yandextranslatefree import YandexFreeTranslate
|
||||
except ImportError: from .yat.yandextranslatefree import YandexFreeTranslate
|
||||
|
||||
def unimplemented(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
print(f"⚠️ Warning: Function '{func.__name__}' is not implemented yet and is just a stub!")
|
||||
# raise NotImplementedError(f"Function '{func.__name__}' is not implemented yet!")
|
||||
return wrapper
|
||||
|
||||
class TranslationHelper():
|
||||
|
||||
def __init__(self, deepseekApiToken: str, yandexApiType: str = "ios"):
|
||||
if not deepseekApiToken or not isinstance(deepseekApiToken, str):
|
||||
raise AuthenticationError("[DEEPSEEK] Invalid auth token provided")
|
||||
if not yandexApiType or not isinstance(yandexApiType, str):
|
||||
raise AuthenticationError("[YANDEX] Invalid auth token provided")
|
||||
|
||||
self.deepseek = DeepSeekAPI(auth_token=deepseekApiToken)
|
||||
self.yandex = YandexFreeTranslate(api=yandexApiType)
|
||||
self.google = Translator()
|
||||
|
||||
def _get_deepseek_response(self, chunks: Generator[Dict[str, Any], None, None]) -> None:
|
||||
"""Helper function to print response chunks in a clean format"""
|
||||
thinking_lines = []
|
||||
text_content = []
|
||||
|
||||
try:
|
||||
for chunk in chunks:
|
||||
print(chunk)
|
||||
if chunk['type'] == 'thinking':
|
||||
if chunk['content'] and chunk['content'] not in thinking_lines:
|
||||
thinking_lines.append(chunk['content'])
|
||||
elif chunk['type'] == 'message':
|
||||
text_content.append(chunk['content'])
|
||||
except KeyError as e:
|
||||
print(f"❌ Error: Malformed response chunk - missing key {str(e)}")
|
||||
return
|
||||
|
||||
if thinking_lines:
|
||||
print("\n🤔 Thinking:")
|
||||
for line in thinking_lines:
|
||||
print(f" • {line}")
|
||||
print()
|
||||
|
||||
response = ''.join(text_content)
|
||||
print("[DEBUG] 💬 AI Response output:")
|
||||
print(response)
|
||||
return response
|
||||
|
||||
def _run_deepseek_chat(self, title: str, prompt: str, thinking_enabled: bool = True, search_enabled: bool = False) -> None:
|
||||
"""Run a chat example with error handling"""
|
||||
# print(f"[DEBUG] \n{title}")
|
||||
# print("-" * 80)
|
||||
|
||||
try:
|
||||
chunks = self.deepseek.chat_completion(
|
||||
self.deepseek.create_chat_session(),
|
||||
prompt,
|
||||
thinking_enabled=thinking_enabled,
|
||||
search_enabled=search_enabled
|
||||
)
|
||||
return self._get_deepseek_response(chunks)
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"[DEEPSEEK] ❌ Authentication Error: {str(e)}")
|
||||
print("[DEEPSEEK] Please check your authentication token and try again.")
|
||||
sys.exit(1)
|
||||
except RateLimitError as e:
|
||||
print(f"[DEEPSEEK] ❌ Rate Limit Error: {str(e)}")
|
||||
print("[DEEPSEEK] Please wait a moment before making more requests.")
|
||||
except NetworkError as e:
|
||||
print(f"[DEEPSEEK] ❌ Network Error: {str(e)}")
|
||||
print("[DEEPSEEK] Please check your internet connection and try again.")
|
||||
except APIError as e:
|
||||
print(f"[DEEPSEEK] ❌ API Error: {str(e)}")
|
||||
if e.status_code:
|
||||
print(f"[DEEPSEEK] Status code: {e.status_code}")
|
||||
except Exception as e:
|
||||
print(f"[DEEPSEEK] ❌ Unexpected Error: {str(e)}")
|
||||
print("[DEEPSEEK] Please report this issue if it persists.")
|
||||
|
||||
|
||||
|
||||
def getDeepseekTranslation(self, text: str, prompt: str = None) -> list[str]:
|
||||
"""Get multiple creative translations from DeepSeek AI.
|
||||
|
||||
Args:
|
||||
text: The English text to translate
|
||||
prompt: Preprompt that will make AI work as expected (10 enumerated lines of russian text)
|
||||
|
||||
Returns:
|
||||
List of 10 Russian translations following the specified format
|
||||
"""
|
||||
if not text or not isinstance(text, str):
|
||||
raise ValueError("[DEEPSEEK] Invalid text provided")
|
||||
if not prompt or not isinstance(prompt, str):
|
||||
raise ValueError("[DEEPSEEK] Invalid pre-prompt provided")
|
||||
if not text.strip():
|
||||
return []
|
||||
|
||||
try:
|
||||
full_prompt = f"{prompt}\n\nEnglish phrase: {text}"
|
||||
|
||||
response = self._run_deepseek_chat(
|
||||
"DeepSeek AI translation",
|
||||
full_prompt,
|
||||
thinking_enabled=False
|
||||
)
|
||||
numerated_translations = response.split("\n")
|
||||
translations = [
|
||||
parts[1].strip()
|
||||
for t in numerated_translations
|
||||
if (parts := t.split(". ", 1)) and # walrus operator requires Python 3.8+
|
||||
len(parts) == 2 and
|
||||
parts[0].strip().isdigit()
|
||||
]
|
||||
if len(translations) >= 10:
|
||||
return translations[:10]
|
||||
|
||||
return translations + ["[Translation not available]"] * (10 - len(translations))
|
||||
|
||||
except Exception as e:
|
||||
print(f"[DEEPSEEK] error: {str(e)}")
|
||||
return [f"DeepSeek Error: {str(e)}"]
|
||||
|
||||
def getYandexTranslation(self, text, source_language: str = "en", target_language: str = "ru",) -> str:
|
||||
if not text or not isinstance(text, str):
|
||||
raise ValueError("[YANDEX] Invalid text provided")
|
||||
if not target_language or not isinstance(target_language, str):
|
||||
raise ValueError("[YANDEX] Invalid target language provided")
|
||||
if not source_language or not isinstance(source_language, str):
|
||||
raise ValueError("[YANDEX] Invalid source language provided")
|
||||
|
||||
try:
|
||||
translation: str = self.yandex.translate(source_language, target_language, text)
|
||||
return translation.strip() if translation else "Yandex translation not available"
|
||||
except Exception as e:
|
||||
print(f"[YANDEX] translation error: {str(e)}")
|
||||
return "Yandex translation error"
|
||||
|
||||
def getGoogleTranslation(self, text, source_language: str = "en", target_language: str = "ru",) -> str:
|
||||
if not text or not isinstance(text, str):
|
||||
raise ValueError("[GOOGLE] Invalid text provided")
|
||||
if not target_language or not isinstance(target_language, str):
|
||||
raise ValueError("[GOOGLE] Invalid target language provided")
|
||||
if not source_language or not isinstance(source_language, str):
|
||||
raise ValueError("[GOOGLE] Invalid source language provided")
|
||||
|
||||
try:
|
||||
translation = self.google.translate(text=text, dest=target_language, src=source_language)
|
||||
translation_text: str = translation.text
|
||||
return translation_text.strip() if translation else "Google translation not available"
|
||||
except Exception as e:
|
||||
print(f"[GOOGLE] translation error: {str(e)}")
|
||||
return "Google translation error"
|
||||
|
||||
@unimplemented
|
||||
def getDeeplTranslation(self, text, source_language: str = "en", target_language: str = "ru",) -> str:
|
||||
if not text or not isinstance(text, str):
|
||||
raise ValueError("[DEEPL] Invalid text provided")
|
||||
if not target_language or not isinstance(target_language, str):
|
||||
raise ValueError("[DEEPL] Invalid target language provided")
|
||||
if not source_language or not isinstance(source_language, str):
|
||||
raise ValueError("[DEEPL] Invalid source language provided")
|
||||
|
||||
try:
|
||||
return "DeepL translation not available"
|
||||
except Exception as e:
|
||||
print(f"[DEEPL] translation error: {str(e)}")
|
||||
return "Deepl translation error"
|
||||
@@ -0,0 +1,257 @@
|
||||
#!/usr/bin/env python3
|
||||
import ssl
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import os.path
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
try:
|
||||
import urllib.parse as urllibparse
|
||||
import urllib.request as urllibrequest
|
||||
import urllib.error as urlliberror
|
||||
except ImportError:
|
||||
import urllib as urllibrequest
|
||||
import urllib as urllibparse
|
||||
import urllib as urlliberror
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
import socks
|
||||
try: from sockshandler import SocksiPyHandler
|
||||
except ImportError: from .sockshandler import SocksiPyHandler
|
||||
del sys.path[0]
|
||||
|
||||
def smartsplit(t, s, e):
|
||||
"""
|
||||
smartsplit(text, start_position, end_position)
|
||||
Splits a string into parts according to the number of characters.
|
||||
If there is a space between the start and end positions, split before a space, the word will not end in the middle.
|
||||
If there are no spaces in the specified range, divide as is, by the finite number of characters.
|
||||
"""
|
||||
t = t.replace("\r\n", "\n").replace("\r", "\n")
|
||||
if(e >= len(t)): return [t]
|
||||
l = []
|
||||
tmp=""
|
||||
i=0
|
||||
for sim in t:
|
||||
i = i + 1
|
||||
tmp = tmp + sim
|
||||
if(i<s): continue
|
||||
if(i==e):
|
||||
l.append(tmp)
|
||||
tmp=""
|
||||
i=0
|
||||
continue
|
||||
if((i > s and i < e) and (sim == chr(160) or sim == chr(9) or sim == chr(10) or sim == chr(32))):
|
||||
l.append(tmp)
|
||||
tmp=""
|
||||
i=0
|
||||
continue
|
||||
if(len(tmp)>0): l.append(tmp)
|
||||
return l
|
||||
|
||||
class YandexFreeTranslateError(Exception): pass
|
||||
|
||||
class YandexFreeTranslate():
|
||||
error_count = 0
|
||||
siteurl = "https://translate.yandex.ru/"
|
||||
apibaseurl = "https://translate.yandex.net/api/v1/tr.json/"
|
||||
api=""
|
||||
ua = r"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:79.0) Gecko/20100101 Firefox/79.0"
|
||||
key = ""
|
||||
keysuffix = "-0-0"
|
||||
keyfilename = os.path.join(os.path.expanduser("~"), ".YandexFreeTranslate.key")
|
||||
expiretime = 60*60*24*4
|
||||
backfilename = keyfilename+".back"
|
||||
useProxy = False
|
||||
proxy_protocol = ""
|
||||
proxy_host = ""
|
||||
proxy_port = 0
|
||||
proxy_username = ""
|
||||
proxy_password = ""
|
||||
def _getparams(self, **p):
|
||||
params = {}
|
||||
params["ios"] = {
|
||||
"srv":"ios", "ucid":"9676696D-0B56-4F13-B4D5-4A3DA2A3344D",
|
||||
"sid":"1A5A10A952AB4A3B82533F44B87EE696", "id":"1A5A10A952AB4A3B82533F44B87EE696-0-0"
|
||||
}
|
||||
params["web"] = {
|
||||
"id":self.key, "srv":"tr-text", "reason":"paste", "options": 4
|
||||
}
|
||||
params[self.api].update(p)
|
||||
return params[self.api]
|
||||
def decode_response(self, response):
|
||||
try:
|
||||
res = response.decode("UTF8")
|
||||
except UnicodeDecodeError:
|
||||
res = gzip.decompress(response).decode("UTF8")
|
||||
return res
|
||||
|
||||
def set_proxy(self, protocol, host, port, username="", password=""):
|
||||
self.useProxy = True
|
||||
self.proxy_protocol = protocol
|
||||
self.proxy_host = host
|
||||
self.proxy_port = int(port)
|
||||
self.proxy_username = username
|
||||
self.proxy_password = password
|
||||
|
||||
def _create_opener(self):
|
||||
opener = urllibrequest.build_opener()
|
||||
if self.useProxy:
|
||||
if self.proxy_protocol == "socks4":
|
||||
opener = urllibrequest.build_opener(SocksiPyHandler(socks.SOCKS4,
|
||||
self.proxy_host, self.proxy_port, username=self.proxy_username, password=self.proxy_password))
|
||||
if self.proxy_protocol == "socks5":
|
||||
opener = urllibrequest.build_opener(SocksiPyHandler(socks.SOCKS5,
|
||||
self.proxy_host, self.proxy_port, username=self.proxy_username, password=self.proxy_password))
|
||||
return opener
|
||||
|
||||
def _create_request(self, *ar, **kw):
|
||||
if len(ar) > 0 and "http" in ar[0]: url = ar[0]
|
||||
if "url" in kw: url = kw["url"]
|
||||
rq = urllibrequest.Request(*ar, **kw)
|
||||
if self.useProxy:
|
||||
if self.proxy_protocol == "http" or self.proxy_protocol == "https":
|
||||
fullHost = ":".join([self.proxy_host, str(self.proxy_port)])
|
||||
if len(self.proxy_username) + len(self.proxy_password) > 0:
|
||||
fullHost = ":".join([self.proxy_username, self.proxy_password]) + "@" + fullHost
|
||||
rq.set_proxy(fullHost, self.proxy_protocol)
|
||||
rq.add_header("Host", urllibparse.urlparse(url)[1])
|
||||
return rq
|
||||
|
||||
def _sid_to_key(self, sid):
|
||||
splitter = "."
|
||||
l = []
|
||||
for item in sid.split(splitter): l.append(item[::-1])
|
||||
return splitter.join(l)+self.keysuffix
|
||||
def _parse_sid(self):
|
||||
if self.api != "web": ValueError("available only for web API. Now using "+self.api)
|
||||
try:
|
||||
if self.useProxy:
|
||||
old_context = ssl._create_default_https_context
|
||||
ssl._create_default_https_context = ssl.create_default_context
|
||||
req = self._create_request(self.siteurl)
|
||||
req.add_header("User-Agent", self.ua)
|
||||
req.add_header("Accept", r"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8 ")
|
||||
req.add_header("Accept-Language", r"ru-RU,ru;q=0.8,en-US;q=0.5,en;q=0.3")
|
||||
req.add_header("DNT", "1")
|
||||
req.add_header("Accept-Encoding", "gzip")
|
||||
response = self._create_opener().open(req).read()
|
||||
page = self.decode_response(response)
|
||||
#open("page.html", "w", encoding="utf8").write(page)
|
||||
try:
|
||||
return re.search(r'''SID[\s]?[:][\s]?['"]([^'"]+)['"]''', page).group(1)
|
||||
except AttributeError:
|
||||
raise YandexFreeTranslateError("blocked or not found")
|
||||
except: raise
|
||||
finally:
|
||||
if self.useProxy:
|
||||
ssl._create_default_https_context = old_context
|
||||
|
||||
def _save_key(self, key):
|
||||
with open(self.keyfilename, "w", encoding="utf8") as f:
|
||||
f.write(key)
|
||||
def _get_key(self):
|
||||
if os.path.isfile(self.keyfilename) and (time.time() - os.path.getmtime(self.keyfilename)) < self.expiretime:
|
||||
# print("from file")
|
||||
with open(self.keyfilename, "r", encoding="utf8") as f:
|
||||
return f.read()
|
||||
else:
|
||||
# print("from internet")
|
||||
sid = self._parse_sid()
|
||||
key = self._sid_to_key(sid)
|
||||
self._save_key(key)
|
||||
return key
|
||||
def get_key(self): return self._get_key()
|
||||
def regenerate_key(self):
|
||||
if os.path.isfile(self.backfilename): os.remove(self.backfilename)
|
||||
if os.path.isfile(self.keyfilename):
|
||||
os.rename(self.keyfilename, self.backfilename)
|
||||
self.key = self._get_key()
|
||||
return self.key
|
||||
def __init__(self, api="web"):
|
||||
api = api.lower()
|
||||
self.api = api
|
||||
if not os.path.isfile(self.keyfilename) and os.path.isfile(self.backfilename):
|
||||
os.rename(self.backfilename, self.keyfilename)
|
||||
def translate(self, source = "auto", target="", text=""):
|
||||
resp = {}
|
||||
content = None
|
||||
try:
|
||||
if self.useProxy:
|
||||
old_context = ssl._create_default_https_context
|
||||
ssl._create_default_https_context = ssl.create_default_context
|
||||
if self.key == "": self.key = self._get_key()
|
||||
if source == "auto": source = ""
|
||||
if len(source) != 0 and len(source) != 2: raise ValueError("source")
|
||||
if len(target) == 0 or len(target) > 2: raise ValueError("target")
|
||||
if text == "": raise ValueError("text")
|
||||
if source==target: return text
|
||||
if source == "": lang = target
|
||||
else: lang = source+"-"+target
|
||||
p=[]
|
||||
for part in smartsplit(text, 500, 550):
|
||||
req = self._create_request(self.apibaseurl+"translate?"+urllibparse.urlencode(self._getparams(lang=lang)))
|
||||
req.add_header("User-Agent", self.ua)
|
||||
req.add_header("Accept", r"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8 ")
|
||||
req.add_header("Accept-Language", r"ru-RU,ru;q=0.8,en-US;q=0.5,en;q=0.3")
|
||||
req.add_header("DNT", "1")
|
||||
req.add_header("Accept-Encoding", "gzip, deflate, br")
|
||||
try:
|
||||
response = self._create_opener().open(req, data = urllibparse.urlencode({
|
||||
"text":part
|
||||
}).encode("UTF8")).read()
|
||||
content = self.decode_response(response)
|
||||
resp = json.loads(content)
|
||||
except (urlliberror.HTTPError, json.JSONDecodeError):
|
||||
if self.error_count >= 2:
|
||||
self.error_count = 0
|
||||
if sys.exc_info()[0] == json.JSONDecodeError:
|
||||
raise YandexFreeTranslateError(content)
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
self.error_count = self.error_count + 1
|
||||
self.regenerate_key()
|
||||
return self.translate(source, target, text)
|
||||
if "text" not in resp:
|
||||
raise YandexFreeTranslateError(content)
|
||||
p.append(resp["text"][0])
|
||||
return "\n".join(p)
|
||||
except: raise
|
||||
finally:
|
||||
if self.useProxy:
|
||||
ssl._create_default_https_context = old_context
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 4:
|
||||
print("usage: "+sys.argv[0]+" \t sourcelang \t targetlang \t text or file.txt \t [output_file_name.txt]")
|
||||
print('example: {0} "{1}" "{2}" "{3}"'.format(sys.argv[0], "en", "ru", "hello, world!"))
|
||||
print('example: {0} "{1}" "{2}" "{3}"'.format(sys.argv[0], "en", "ru", "in.txt"))
|
||||
print("\n")
|
||||
print("Install a proxy server: create an environment variable before calling:")
|
||||
print("\t export https_proxy=127.0.0.1:8080")
|
||||
print("\t export socks5_proxy=127.0.0.1:9050")
|
||||
print("\t export socks5_proxy=username:password@127.0.0.1:9050")
|
||||
sys.exit(0)
|
||||
yt = YandexFreeTranslate()
|
||||
for protocol in ["http", "https", "socks4", "socks5"]:
|
||||
if protocol in os.environ:
|
||||
if "@" in os.environ[protocol]:
|
||||
auth, hostport = os.environ[protocol].split("@")
|
||||
host, port = hostport.split(":")
|
||||
login, password = auth.split(":")
|
||||
else:
|
||||
host, port = hostport.split(":")
|
||||
login, password = ("", "")
|
||||
yt.setProxy(protocol, host, port, login, password)
|
||||
text = sys.argv[3]
|
||||
if len(text) < 150 and os.path.isfile(text): text = open(text, "r", encoding="utf8").read()
|
||||
tr_text = yt.translate(sys.argv[1], sys.argv[2], text)
|
||||
if len(sys.argv) == 5:
|
||||
with open(sys.argv[4], "w") as f: f.write(tr_text)
|
||||
print("text saved in file: "+sys.argv[4])
|
||||
else:
|
||||
print(tr_text)
|
||||
Reference in new issue
Block a user