feat(shell): Add support to curl -b argument
This commit is contained in:
+60
-18
@@ -35,6 +35,7 @@ from scrapling.fetchers import (
|
|||||||
Response,
|
Response,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
_known_logging_levels = {
|
_known_logging_levels = {
|
||||||
"debug": DEBUG,
|
"debug": DEBUG,
|
||||||
"info": INFO,
|
"info": INFO,
|
||||||
@@ -105,6 +106,13 @@ class CurlParser:
|
|||||||
"-G", "--get", action="store_true"
|
"-G", "--get", action="store_true"
|
||||||
) # Use GET and put data in URL
|
) # Use GET and put data in URL
|
||||||
|
|
||||||
|
_parser.add_argument(
|
||||||
|
"-b",
|
||||||
|
"--cookie",
|
||||||
|
default=None,
|
||||||
|
help="Send cookies from string/file (string format used by DevTools)",
|
||||||
|
)
|
||||||
|
|
||||||
# Proxy
|
# Proxy
|
||||||
_parser.add_argument("-x", "--proxy", default=None)
|
_parser.add_argument("-x", "--proxy", default=None)
|
||||||
_parser.add_argument("-U", "--proxy-user", default=None) # Basic proxy auth
|
_parser.add_argument("-U", "--proxy-user", default=None) # Basic proxy auth
|
||||||
@@ -154,7 +162,7 @@ class CurlParser:
|
|||||||
cookie_dict[key] = morsel.value
|
cookie_dict[key] = morsel.value
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
log.error(
|
log.error(
|
||||||
f"Could not parse cookie string '{header_value}': {e}"
|
f"Could not parse cookie string from -H '{header_value}': {e}"
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
header_dict[header_key] = header_value
|
header_dict[header_key] = header_value
|
||||||
@@ -210,6 +218,21 @@ class CurlParser:
|
|||||||
|
|
||||||
headers, cookies = self.parse_headers(parsed_args.header)
|
headers, cookies = self.parse_headers(parsed_args.header)
|
||||||
|
|
||||||
|
if parsed_args.cookie:
|
||||||
|
# We are focusing on the string format from DevTools.
|
||||||
|
try:
|
||||||
|
cookie_parser = Cookie.SimpleCookie()
|
||||||
|
cookie_parser.load(parsed_args.cookie)
|
||||||
|
for key, morsel in cookie_parser.items():
|
||||||
|
# Update the cookies dict, potentially overwriting
|
||||||
|
# cookies with the same name from -H 'Cookie:'
|
||||||
|
cookies[key] = morsel.value
|
||||||
|
log.debug(f"Parsed cookies from -b argument: {list(cookies.keys())}")
|
||||||
|
except Exception as e:
|
||||||
|
log.error(
|
||||||
|
f"Could not parse cookie string from -b '{parsed_args.cookie}': {e}"
|
||||||
|
)
|
||||||
|
|
||||||
# --- Process Data Payload ---
|
# --- Process Data Payload ---
|
||||||
params = dict()
|
params = dict()
|
||||||
data_payload: Union[str, bytes, Dict, None] = None
|
data_payload: Union[str, bytes, Dict, None] = None
|
||||||
@@ -316,7 +339,7 @@ class CurlParser:
|
|||||||
follow_redirects=True, # Scrapling default is True
|
follow_redirects=True, # Scrapling default is True
|
||||||
)
|
)
|
||||||
|
|
||||||
def convert2fetcher(self, curl_command: [Request, str]) -> Optional[Response]:
|
def convert2fetcher(self, curl_command: Union[Request, str]) -> Optional[Response]:
|
||||||
request = None
|
request = None
|
||||||
if isinstance(curl_command, (Request, str)):
|
if isinstance(curl_command, (Request, str)):
|
||||||
request = (
|
request = (
|
||||||
@@ -324,37 +347,53 @@ class CurlParser:
|
|||||||
if isinstance(curl_command, str)
|
if isinstance(curl_command, str)
|
||||||
else curl_command
|
else curl_command
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Ensure request parsing was successful before proceeding
|
||||||
|
if request is None:
|
||||||
|
log.error("Failed to parse curl command, cannot convert to fetcher.")
|
||||||
|
return None
|
||||||
|
|
||||||
request_args = request._asdict()
|
request_args = request._asdict()
|
||||||
method = request_args.pop("method").strip().lower()
|
method = request_args.pop("method").strip().lower()
|
||||||
if method in self._supported_methods:
|
if method in self._supported_methods:
|
||||||
request_args["json"] = request_args.pop("json_data")
|
request_args["json"] = request_args.pop("json_data")
|
||||||
if method not in ("post", "put"):
|
|
||||||
_ = request_args.pop("data")
|
|
||||||
_ = request_args.pop("json")
|
|
||||||
|
|
||||||
|
# Ensure data/json are removed for non-POST/PUT methods
|
||||||
|
if method not in ("post", "put"):
|
||||||
|
_ = request_args.pop("data", None)
|
||||||
|
_ = request_args.pop("json", None)
|
||||||
|
|
||||||
|
try:
|
||||||
return getattr(Fetcher, method)(**request_args)
|
return getattr(Fetcher, method)(**request_args)
|
||||||
|
except Exception as e:
|
||||||
|
log.error(f"Error calling Fetcher.{method}: {e}")
|
||||||
|
return None
|
||||||
else:
|
else:
|
||||||
log.error(
|
log.error(
|
||||||
f'Request method "{method}" isn\'t supported by Scrapling yet'
|
f'Request method "{method}" isn\'t supported by Scrapling yet'
|
||||||
)
|
)
|
||||||
|
return None
|
||||||
|
|
||||||
if request is None:
|
else:
|
||||||
log.error(
|
log.error("Input must be a valid curl command string or a Request object.")
|
||||||
"This class accepts `Request` objects only generated by the `uncurl` command or a curl command passed as string."
|
|
||||||
)
|
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def show_page_in_browser(page):
|
def show_page_in_browser(page: Adaptor):
|
||||||
if not page:
|
if not page or not isinstance(page, Adaptor):
|
||||||
log.error("Input must be of type `Adaptor`")
|
log.error("Input must be of type `Adaptor`")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
fd, fname = make_temp_file(".html")
|
fd, fname = make_temp_file(".html")
|
||||||
os.write(fd, page.body.encode("utf-8"))
|
os.write(fd, page.body.encode("utf-8"))
|
||||||
os.close(fd)
|
os.close(fd)
|
||||||
open_in_browser(f"file://{fname}")
|
open_in_browser(f"file://{fname}")
|
||||||
|
except IOError as e:
|
||||||
|
log.error(f"Failed to write temporary file for viewing: {e}")
|
||||||
|
except Exception as e:
|
||||||
|
log.error(f"An unexpected error occurred while viewing the page: {e}")
|
||||||
|
|
||||||
|
|
||||||
class CustomShell:
|
class CustomShell:
|
||||||
@@ -370,7 +409,7 @@ class CustomShell:
|
|||||||
if _known_logging_levels.get(log_level):
|
if _known_logging_levels.get(log_level):
|
||||||
self.log_level = _known_logging_levels[log_level]
|
self.log_level = _known_logging_levels[log_level]
|
||||||
else:
|
else:
|
||||||
log.error(f'Unknown log level "{log_level}", defaulting to "DEBUG"')
|
log.warning(f'Unknown log level "{log_level}", defaulting to "DEBUG"')
|
||||||
self.log_level = DEBUG
|
self.log_level = DEBUG
|
||||||
|
|
||||||
self.shell = None
|
self.shell = None
|
||||||
@@ -385,8 +424,8 @@ class CustomShell:
|
|||||||
getLogger("scrapling").setLevel(self.log_level)
|
getLogger("scrapling").setLevel(self.log_level)
|
||||||
|
|
||||||
settings = Fetcher.display_config()
|
settings = Fetcher.display_config()
|
||||||
_ = settings.pop("storage")
|
settings.pop("storage", None)
|
||||||
_ = settings.pop("storage_args")
|
settings.pop("storage_args", None)
|
||||||
log.info(f"Scrapling {__version__} shell started")
|
log.info(f"Scrapling {__version__} shell started")
|
||||||
log.info(f"Logging level is set to '{getLevelName(self.log_level)}'")
|
log.info(f"Logging level is set to '{getLevelName(self.log_level)}'")
|
||||||
log.info(f"Fetchers' parsing settings: {settings}")
|
log.info(f"Fetchers' parsing settings: {settings}")
|
||||||
@@ -412,8 +451,8 @@ class CustomShell:
|
|||||||
-> Useful commands
|
-> Useful commands
|
||||||
- {"page / response":<30} The response object of the last page you fetched
|
- {"page / response":<30} The response object of the last page you fetched
|
||||||
- {"pages":<30} Adaptors object of the last 5 response objects you fetched
|
- {"pages":<30} Adaptors object of the last 5 response objects you fetched
|
||||||
- {"uncurl('curl_command')":<30} Convert a curl command to a Fetcher's request and return the Request object for you. (Optimized to handle curl commands copied from DevTools network tab.)
|
- {"uncurl('curl_command')":<30} Convert curl command to a Request object. (Optimized to handle curl commands copied from DevTools network tab.)
|
||||||
- {"curl2fetcher('curl_command')":<30} Convert a curl command to a Fetcher's request and execute it. (Optimized to handle curl commands copied from DevTools network tab.)
|
- {"curl2fetcher('curl_command')":<30} Convert curl command and make the request with Fetcher. (Optimized to handle curl commands copied from DevTools network tab.)
|
||||||
- {"view(page)":<30} View page in a browser
|
- {"view(page)":<30} View page in a browser
|
||||||
- {"help()":<30} Show this help message (Shell help)
|
- {"help()":<30} Show this help message (Shell help)
|
||||||
|
|
||||||
@@ -423,6 +462,7 @@ Type 'exit' or press Ctrl+D to exit.
|
|||||||
def update_page(self, result):
|
def update_page(self, result):
|
||||||
"""Update current page and add to pages history"""
|
"""Update current page and add to pages history"""
|
||||||
self.page = result
|
self.page = result
|
||||||
|
if isinstance(result, (Response, Adaptor)):
|
||||||
self.pages.append(result)
|
self.pages.append(result)
|
||||||
if len(self.pages) > 5:
|
if len(self.pages) > 5:
|
||||||
self.pages.pop(0) # Remove oldest item
|
self.pages.pop(0) # Remove oldest item
|
||||||
@@ -497,9 +537,11 @@ Type 'exit' or press Ctrl+D to exit.
|
|||||||
ipython_shell.user_ns.update(namespace)
|
ipython_shell.user_ns.update(namespace)
|
||||||
# If a command was provided, execute it and exit
|
# If a command was provided, execute it and exit
|
||||||
if self.code:
|
if self.code:
|
||||||
# Execute the command in the namespace
|
log.info(f"Executing provided code: {self.code}")
|
||||||
|
try:
|
||||||
ipython_shell.run_cell(self.code, store_history=False)
|
ipython_shell.run_cell(self.code, store_history=False)
|
||||||
|
except Exception as e:
|
||||||
|
log.error(f"Error executing initial code: {e}")
|
||||||
return
|
return
|
||||||
|
|
||||||
# Start the shell with our namespace
|
|
||||||
ipython_shell(local_ns=namespace)
|
ipython_shell(local_ns=namespace)
|
||||||
|
|||||||
Reference in New Issue
Block a user