feat(shell): Add support to curl `-b` argument

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")
+
+                # Ensure data/json are removed for non-POST/PUT methods
                 if method not in ("post", "put"):
-                    _ = request_args.pop("data")
-                    _ = request_args.pop("json")
+                    _ = request_args.pop("data", None)
+                    _ = request_args.pop("json", None)
 
-                return getattr(Fetcher, method)(**request_args)
+                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)