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
|
import log
import time
import json
import tls_client
import binascii
import threading
import config
from enum import Enum
from blake3 import blake3
from collections import OrderedDict
from typing import Optional, Self, Any
from post import Post, User
type ParsedJson = Any
type HttpResponse = tls_client.Response
USER_AGENT = config.USER_AGENT
class Method(Enum):
HEAD = 'HEAD'
GET = 'GET'
POST = 'POST'
class Search():
def __init__(self) -> None:
self.pages: dict[int, list[str]] = {}
self.completed: bool = False
self.errored: bool = False
self.iter_page: int = 0
self.iter_index: int = 0
def __iter__(self) -> Self:
self.iter_page = 0
self.iter_index = 0
return self
def __next__(self) -> str:
page = self.get_page(self.iter_page)
if not page:
raise StopIteration
unique_id = page[self.iter_index]
self.iter_index += 1
if self.iter_index >= len(page):
self.iter_page += 1
self.iter_index = 0
return unique_id
def request_page(self, num: int) -> bool:
return False
def get_page(self, num: int) -> Optional[list[str]]:
if self.errored:
return None
if num not in self.pages:
if self.completed:
return None
if not self.request_page(num):
if not self.completed:
self.errored = True
return None
return self.pages[num]
class Module():
def __init__(self) -> None:
self.headers: dict[str, str] = {}
self.cookies: dict[str, str] = {}
self.mutex: threading.Lock = threading.Lock()
self.raw_responses: dict[str, str] = {}
self.unique_id_map: OrderedDict = OrderedDict()
self.session = tls_client.Session(client_identifier=config.TLS_CLIENT_IDENTIFIER, random_tls_extension_order=True)
def init(self) -> bool:
return True
def do_request(self, method: Method, url: str, params: Optional[dict[str, str]] = None, retries: int = 2) -> Optional[HttpResponse]:
log.debug(f'HTTP {method.value} request {url} {params}.')
response: Optional[HttpResponse] = None
for _ in range(0, retries + 1):
try:
response = self.session.execute_request(method.value, url, headers=self.headers, cookies=self.cookies, params=params, timeout=600)
except Exception as e:
log.info(f'Connection exception ({repr(e)}), retrying in 20 seconds...')
time.sleep(20)
continue
if response is None:
log.error('Invalid None response from requests library.')
break
if response.status_code == 429:
log.info('Got too many requests, waiting for 4 minutes...')
time.sleep(60 * 4)
response = None
retries += 1
continue
if response.status_code == 404 or response.status_code == 403: # Shortcut 404, 403
log.warn(f'Returning None, {response.status_code}')
return None
if int(response.status_code / 100) != 2: # Non-200 response
log.error(f'Non-200 status code ({response.status_code}), retrying in 5 seconds...')
time.sleep(5)
response = None
continue
'''
try:
response.read()
except Exception as e:
log.info(f'Read exception ({e}), retrying in 20 seconds...')
time.sleep(20)
response = None
continue
else:
break
'''
break
return response
def add_raw_response(self, data: ParsedJson) -> str:
with self.mutex:
dump = json.dumps(data, separators=(',', ':'), indent=None)
hash_str = binascii.hexlify(blake3(dump.encode('utf-8')).digest()).decode('ascii')
self.raw_responses[hash_str] = dump
return hash_str
def add_to_map(self, unique_id: str, item: Post | User) -> None:
with self.mutex:
self.unique_id_map[unique_id] = item
if len(self.unique_id_map) > 10240:
self.unique_id_map.popitem(last=False)
def search(self, query: str, *extra_args: Any) -> Optional[Search]:
return None
def get_item(self, unique_id: str) -> Optional[Post | User]:
with self.mutex:
return self.unique_id_map.get(unique_id, None)
def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
return None
|