-
-
Notifications
You must be signed in to change notification settings - Fork 11
/
parser.py
94 lines (69 loc) · 2.6 KB
/
parser.py
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
from typing import Any, Dict, List, Optional, Tuple, Union
from curl_cffi import CurlOpt
from scrapy.http import Request
def curl_option_method(func):
func._is_curl_option = True
return func
class CurlOptionsParser:
def __init__(self, request: Request) -> None:
self.request = request
self.curl_options = {}
@curl_option_method
def _set_proxy_auth(self):
"""Add support for proxy auth headers"""
if proxy_authorization := self.request.headers.pop(b"Proxy-Authorization", None):
proxy_header = [b"Proxy-Authorization: " + proxy_authorization[0]]
self.curl_options[CurlOpt.PROXYHEADER] = proxy_header
def as_dict(self):
for method_name in dir(self):
method = getattr(self, method_name)
if callable(method) and getattr(method, "_is_curl_option", False):
method()
return self.curl_options
class RequestParser:
# TODO: Implement @request_arg_method instead of @property
def __init__(self, request: Request) -> None:
self._request = request
self._impersonate_args = request.meta.get("impersonate_args", {})
@property
def method(self) -> str:
return self._request.method
@property
def url(self) -> str:
return self._request.url
@property
def params(self) -> Optional[Union[Dict, List, Tuple]]:
return self._impersonate_args.get("params")
@property
def data(self) -> Optional[Any]:
return self._request.body
@property
def headers(self) -> dict:
headers = self._request.headers.to_unicode_dict()
return dict(headers)
@property
def cookies(self) -> dict:
cookies = self._request.cookies
if isinstance(cookies, list):
return {k: v for cookie in cookies for k, v in cookie.items()}
elif isinstance(cookies, dict):
return {k: v for k, v in cookies.items()}
else:
return {}
@property
def allow_redirects(self) -> bool:
return False if self._request.meta.get("dont_redirect") else True
@property
def proxy(self) -> Optional[str]:
return self._request.meta.get("proxy")
@property
def impersonate(self) -> Optional[str]:
return self._request.meta.get("impersonate")
def as_dict(self) -> dict:
request_args = {
property_name: getattr(self, property_name)
for property_name, method in self.__class__.__dict__.items()
if isinstance(method, property)
}
request_args.update(self._impersonate_args)
return request_args