From 74fa1bfbed70d7fbbd447be75d1b6e7ed484ec1e Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Wed, 30 Apr 2025 04:23:05 +0300 Subject: [PATCH] feat(shell): Add support to curl `-b` argument --- scrapling/core/shell.py | 106 ++++++++++++++++++++++++++++------------ 1 file changed, 74 insertions(+), 32 deletions(-) diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 663a168..30f8a4c 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -35,6 +35,7 @@ from scrapling.fetchers import ( Response, ) + _known_logging_levels = { "debug": DEBUG, "info": INFO, @@ -105,6 +106,13 @@ class CurlParser: "-G", "--get", action="store_true" ) # 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 _parser.add_argument("-x", "--proxy", default=None) _parser.add_argument("-U", "--proxy-user", default=None) # Basic proxy auth @@ -154,7 +162,7 @@ class CurlParser: cookie_dict[key] = morsel.value except Exception as e: log.error( - f"Could not parse cookie string '{header_value}': {e}" + f"Could not parse cookie string from -H '{header_value}': {e}" ) else: header_dict[header_key] = header_value @@ -210,6 +218,21 @@ class CurlParser: 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 --- params = dict() data_payload: Union[str, bytes, Dict, None] = None @@ -316,7 +339,7 @@ class CurlParser: 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 if isinstance(curl_command, (Request, str)): request = ( @@ -324,37 +347,53 @@ class CurlParser: if isinstance(curl_command, str) 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() method = request_args.pop("method").strip().lower() if method in self._supported_methods: request_args["json"] = request_args.pop("json_data") - if method not in ("post", "put"): - _ = request_args.pop("data") - _ = request_args.pop("json") - return getattr(Fetcher, method)(**request_args) + # 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) + except Exception as e: + log.error(f"Error calling Fetcher.{method}: {e}") + return None else: log.error( f'Request method "{method}" isn\'t supported by Scrapling yet' ) + return None - if request is None: - log.error( - "This class accepts `Request` objects only generated by the `uncurl` command or a curl command passed as string." - ) + else: + log.error("Input must be a valid curl command string or a Request object.") return None -def show_page_in_browser(page): - if not page: +def show_page_in_browser(page: Adaptor): + if not page or not isinstance(page, Adaptor): log.error("Input must be of type `Adaptor`") return - fd, fname = make_temp_file(".html") - os.write(fd, page.body.encode("utf-8")) - os.close(fd) - open_in_browser(f"file://{fname}") + try: + fd, fname = make_temp_file(".html") + os.write(fd, page.body.encode("utf-8")) + os.close(fd) + 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: @@ -370,7 +409,7 @@ class CustomShell: if _known_logging_levels.get(log_level): self.log_level = _known_logging_levels[log_level] 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.shell = None @@ -385,8 +424,8 @@ class CustomShell: getLogger("scrapling").setLevel(self.log_level) settings = Fetcher.display_config() - _ = settings.pop("storage") - _ = settings.pop("storage_args") + settings.pop("storage", None) + settings.pop("storage_args", None) log.info(f"Scrapling {__version__} shell started") log.info(f"Logging level is set to '{getLevelName(self.log_level)}'") log.info(f"Fetchers' parsing settings: {settings}") @@ -412,8 +451,8 @@ class CustomShell: -> Useful commands - {"page / response":<30} The response object of the last page 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.) - - {"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.) + - {"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 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 - {"help()":<30} Show this help message (Shell help) @@ -423,15 +462,16 @@ Type 'exit' or press Ctrl+D to exit. def update_page(self, result): """Update current page and add to pages history""" self.page = result - self.pages.append(result) - if len(self.pages) > 5: - self.pages.pop(0) # Remove oldest item + if isinstance(result, (Response, Adaptor)): + self.pages.append(result) + if len(self.pages) > 5: + self.pages.pop(0) # Remove oldest item - # Update in IPython namespace too - if self.shell: - self.shell.user_ns["page"] = self.page - self.shell.user_ns["response"] = self.page - self.shell.user_ns["pages"] = self.pages + # Update in IPython namespace too + if self.shell: + self.shell.user_ns["page"] = self.page + self.shell.user_ns["response"] = self.page + self.shell.user_ns["pages"] = self.pages return result @@ -497,9 +537,11 @@ Type 'exit' or press Ctrl+D to exit. ipython_shell.user_ns.update(namespace) # If a command was provided, execute it and exit if self.code: - # Execute the command in the namespace - ipython_shell.run_cell(self.code, store_history=False) + log.info(f"Executing provided code: {self.code}") + try: + ipython_shell.run_cell(self.code, store_history=False) + except Exception as e: + log.error(f"Error executing initial code: {e}") return - # Start the shell with our namespace ipython_shell(local_ns=namespace)