refactor: Optimizations to CLI
This commit is contained in:
Karim shoair
+10
-30
@@ -3,7 +3,7 @@ from subprocess import check_output
|
||||
from sys import executable as python_executable
|
||||
|
||||
from scrapling.core.utils import log
|
||||
from scrapling.core.shell import Convertor, _CookieParser
|
||||
from scrapling.core.shell import Convertor, _CookieParser, _ParseHeaders
|
||||
from scrapling.fetchers import Fetcher, DynamicFetcher, StealthyFetcher
|
||||
|
||||
from orjson import loads as json_loads, JSONDecodeError
|
||||
@@ -22,31 +22,6 @@ def run_command(cmd, line):
|
||||
# I meant to not use try except here
|
||||
|
||||
|
||||
def parse_headers(header_strings):
|
||||
"""Parse header strings into a dictionary"""
|
||||
headers = {}
|
||||
for header in header_strings:
|
||||
if ":" in header:
|
||||
key, value = header.split(":", 1)
|
||||
headers[key.strip()] = value.strip()
|
||||
else:
|
||||
log.warning(f"Invalid header format '{header}', should be 'Key: Value'")
|
||||
return headers
|
||||
|
||||
|
||||
def parse_cookies(cookie_string):
|
||||
"""Parse cookie string into a dictionary"""
|
||||
if not cookie_string:
|
||||
return {}
|
||||
|
||||
try:
|
||||
cookies = {key: value for key, value in _CookieParser(cookie_string)}
|
||||
except Exception as e:
|
||||
raise ValueError(f"Could not parse cookies '{cookie_string}': {e}")
|
||||
|
||||
return cookies
|
||||
|
||||
|
||||
def parse_json_data(json_string):
|
||||
"""Parse JSON string into a Python object"""
|
||||
if not json_string:
|
||||
@@ -140,8 +115,13 @@ def shell(code, level):
|
||||
|
||||
def parse_extract_arguments(headers, cookies, params, json=None):
|
||||
"""Parse arguments for extract command"""
|
||||
parsed_headers = parse_headers(headers)
|
||||
parsed_cookies = parse_cookies(cookies)
|
||||
parsed_headers, parsed_cookies = _ParseHeaders(headers)
|
||||
for key, value in _CookieParser(cookies):
|
||||
try:
|
||||
parsed_cookies[key] = value
|
||||
except Exception as e:
|
||||
raise ValueError(f"Could not parse cookies '{cookies}': {e}")
|
||||
|
||||
parsed_json = parse_json_data(json)
|
||||
parsed_params = {}
|
||||
for param in params:
|
||||
@@ -673,7 +653,7 @@ def fetch(
|
||||
"""
|
||||
|
||||
# Parse parameters
|
||||
parsed_headers = parse_headers(extra_headers)
|
||||
parsed_headers, _ = _ParseHeaders(extra_headers, False)
|
||||
|
||||
# Build request arguments
|
||||
kwargs = {
|
||||
@@ -821,7 +801,7 @@ def stealthy_fetch(
|
||||
"""
|
||||
|
||||
# Parse parameters
|
||||
parsed_headers = parse_headers(extra_headers)
|
||||
parsed_headers, _ = _ParseHeaders(extra_headers, False)
|
||||
|
||||
# Build request arguments
|
||||
kwargs = {
|
||||
|
||||
+41
-36
@@ -73,6 +73,46 @@ def _CookieParser(cookie_string):
|
||||
yield key, morsel.value
|
||||
|
||||
|
||||
def _ParseHeaders(
|
||||
header_lines: List[str], parse_cookies: bool = True
|
||||
) -> Tuple[Dict[str, str], Dict[str, str]]:
|
||||
"""Parses headers into separate header and cookie dictionaries."""
|
||||
header_dict = dict()
|
||||
cookie_dict = dict()
|
||||
|
||||
for header_line in header_lines:
|
||||
if ":" not in header_line:
|
||||
if header_line.endswith(";"):
|
||||
header_key = header_line[:-1].strip()
|
||||
header_value = ""
|
||||
header_dict[header_key] = header_value
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Could not parse header without colon: '{header_line}'."
|
||||
)
|
||||
else:
|
||||
header_key, header_value = header_line.split(":", 1)
|
||||
header_key = header_key.strip()
|
||||
header_value = header_value.strip()
|
||||
|
||||
if parse_cookies:
|
||||
if header_key.lower() == "cookie":
|
||||
try:
|
||||
cookie_dict = {
|
||||
key: value for key, value in _CookieParser(header_value)
|
||||
}
|
||||
except Exception as e:
|
||||
raise ValueError(
|
||||
f"Could not parse cookie string from header '{header_value}': {e}"
|
||||
)
|
||||
else:
|
||||
header_dict[header_key] = header_value
|
||||
else:
|
||||
header_dict[header_key] = header_value
|
||||
|
||||
return header_dict, cookie_dict
|
||||
|
||||
|
||||
# Suppress exit on error to handle parsing errors gracefully
|
||||
class NoExitArgumentParser(ArgumentParser):
|
||||
def error(self, message):
|
||||
@@ -142,41 +182,6 @@ class CurlParser:
|
||||
self._supported_methods = ("get", "post", "put", "delete")
|
||||
|
||||
# --- Helper Functions ---
|
||||
@staticmethod
|
||||
def parse_headers(header_lines: List[str]) -> Tuple[Dict[str, str], Dict[str, str]]:
|
||||
"""Parses -H headers into separate header and cookie dictionaries."""
|
||||
header_dict = dict()
|
||||
cookie_dict = dict()
|
||||
|
||||
for header_line in header_lines:
|
||||
if ":" not in header_line:
|
||||
if header_line.endswith(";"):
|
||||
header_key = header_line[:-1].strip()
|
||||
header_value = ""
|
||||
header_dict[header_key] = header_value
|
||||
else:
|
||||
log.warning(
|
||||
f"Could not parse header without colon: '{header_line}', skipping."
|
||||
)
|
||||
continue
|
||||
else:
|
||||
header_key, header_value = header_line.split(":", 1)
|
||||
header_key = header_key.strip()
|
||||
header_value = header_value.strip()
|
||||
|
||||
if header_key.lower() == "cookie":
|
||||
try:
|
||||
cookie_dict = {
|
||||
key: value for key, value in _CookieParser(header_value)
|
||||
}
|
||||
except Exception as e:
|
||||
raise ValueError(
|
||||
f"Could not parse cookie string from -H '{header_value}': {e}"
|
||||
)
|
||||
else:
|
||||
header_dict[header_key] = header_value
|
||||
|
||||
return header_dict, cookie_dict
|
||||
|
||||
# --- Main Parsing Logic ---
|
||||
def parse(self, curl_command: str) -> Optional[Request]:
|
||||
@@ -225,7 +230,7 @@ class CurlParser:
|
||||
):
|
||||
method = "post"
|
||||
|
||||
headers, cookies = self.parse_headers(parsed_args.header)
|
||||
headers, cookies = _ParseHeaders(parsed_args.header)
|
||||
|
||||
if parsed_args.cookie:
|
||||
# We are focusing on the string format from DevTools.
|
||||
|
||||
Reference in New Issue
Block a user