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
|
import log
import time
import json
import httpx
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
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: httpx.Client = httpx.Client(follow_redirects=True, timeout=20.0, http2=False)
def init(self) -> bool:
return True
def do_request(self, method: Method, url: str, params: Optional[dict[str, str]] = None, retries: int = 2) -> Optional[httpx.Response]:
log.debug(f'HTTP {method.value} request {url} {params}.')
response: Optional[httpx.Response] = None
for _ in range(0, retries + 1):
try:
response = self.session.request(method.value, url, headers=self.headers, cookies=self.cookies, params=params)
except Exception as e:
log.info(f'Connection exception ({e}), retrying in 20 seconds...')
time.sleep(20)
continue
if not response:
continue
if response.status_code == 429:
log.info('Got too many requests, waiting for 3 minutes...')
time.sleep(60 * 3)
response = None
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
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
|