From 8940bdeb5575357e9df8ee825e34b8a38c0a40f3 Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Tue, 25 Nov 2025 21:23:06 +0200 Subject: [PATCH] fix(shell): dynamically build the signature of shortcuts after last changes --- scrapling/core/_shell_signatures.py | 95 +++++++++++++++++++++++++++++ scrapling/core/shell.py | 54 ++++++++++++++-- 2 files changed, 143 insertions(+), 6 deletions(-) create mode 100644 scrapling/core/_shell_signatures.py diff --git a/scrapling/core/_shell_signatures.py b/scrapling/core/_shell_signatures.py new file mode 100644 index 0000000..907a9f8 --- /dev/null +++ b/scrapling/core/_shell_signatures.py @@ -0,0 +1,95 @@ +from scrapling.core._types import ( + Dict, + Any, + List, + Tuple, + Optional, +) + +# Parameter definitions for shell function signatures (defined once at module level) +# Mirrors TypedDict definitions from _types.py but runtime-accessible for IPython introspection +_REQUESTS_PARAMS = { + "params": Optional[Dict | List | Tuple], + "cookies": Any, + "auth": Optional[Tuple[str, str]], + "impersonate": Any, + "http3": Optional[bool], + "stealthy_headers": Optional[bool], + "proxies": Any, + "proxy": Optional[str], + "proxy_auth": Optional[Tuple[str, str]], + "timeout": Optional[int | float], + "headers": Any, + "retries": Optional[int], + "retry_delay": Optional[int], + "follow_redirects": Optional[bool], + "max_redirects": Optional[int], + "verify": Optional[bool], + "cert": Optional[str | Tuple[str, str]], + "selector_config": Optional[Dict], +} + +_FETCH_PARAMS = { + "headless": bool, + "google_search": bool, + "hide_canvas": bool, + "disable_webgl": bool, + "real_chrome": bool, + "stealth": bool, + "wait": int | float, + "page_action": Optional[Any], + "proxy": Optional[str | Dict], + "locale": str, + "extra_headers": Optional[Dict[str, str]], + "useragent": Optional[str], + "cdp_url": Optional[str], + "timeout": int | float, + "disable_resources": bool, + "wait_selector": Optional[str], + "init_script": Optional[str], + "cookies": Optional[List[Dict]], + "network_idle": bool, + "load_dom": bool, + "wait_selector_state": Any, + "extra_flags": Optional[List[str]], + "additional_args": Optional[Dict], + "custom_config": Optional[Dict], +} + +_STEALTHY_FETCH_PARAMS = { + "headless": bool, + "block_images": bool, + "disable_resources": bool, + "block_webrtc": bool, + "allow_webgl": bool, + "network_idle": bool, + "load_dom": bool, + "humanize": bool | float, + "solve_cloudflare": bool, + "wait": int | float, + "timeout": int | float, + "page_action": Optional[Any], + "wait_selector": Optional[str], + "init_script": Optional[str], + "addons": Optional[List[str]], + "wait_selector_state": Any, + "cookies": Optional[List[Dict]], + "google_search": bool, + "extra_headers": Optional[Dict[str, str]], + "proxy": Optional[str | Dict], + "os_randomize": bool, + "disable_ads": bool, + "geoip": bool, + "custom_config": Optional[Dict], + "additional_args": Optional[Dict], +} + +# Mapping of function names to their parameter definitions +Signatures_map = { + "get": _REQUESTS_PARAMS, + "post": {**_REQUESTS_PARAMS, "data": Optional[Dict | str], "json": Optional[Dict | List]}, + "put": {**_REQUESTS_PARAMS, "data": Optional[Dict | str], "json": Optional[Dict | List]}, + "delete": _REQUESTS_PARAMS, + "fetch": _FETCH_PARAMS, + "stealthy_fetch": _STEALTHY_FETCH_PARAMS, +} diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 3e0f1e2..143d8be 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -1,13 +1,14 @@ # -*- coding: utf-8 -*- -from re import sub as re_sub from sys import stderr from functools import wraps +from re import sub as re_sub from collections import namedtuple from shlex import split as shlex_split +from inspect import signature, Parameter from tempfile import mkstemp as make_temp_file -from urllib.parse import urlparse, urlunparse, parse_qsl from argparse import ArgumentParser, SUPPRESS from webbrowser import open as open_in_browser +from urllib.parse import urlparse, urlunparse, parse_qsl from logging import ( DEBUG, INFO, @@ -21,6 +22,7 @@ from logging import ( from orjson import loads as json_loads, JSONDecodeError +from ._shell_signatures import Signatures_map from scrapling import __version__ from scrapling.core.utils import log from scrapling.parser import Selector, Selectors @@ -28,12 +30,12 @@ from scrapling.core.custom_types import TextHandler from scrapling.engines.toolbelt.custom import Response from scrapling.core.utils._shell import _ParseHeaders, _CookieParser from scrapling.core._types import ( - Optional, Dict, Any, cast, - extraction_types, + Optional, Generator, + extraction_types, ) @@ -312,6 +314,40 @@ class CurlParser: return None +def _unpack_signature(func): + """ + Unpack TypedDict from Unpack[TypedDict] annotations in **kwargs and reconstruct the signature. + + This allows the interactive shell to show individual parameters instead of just **kwargs, similar to how IDEs display them. + """ + try: + sig = signature(func) + func_name = getattr(func, "__name__", None) + + # Check if this function has known parameters + if func_name not in Signatures_map: + return sig + + new_params = [] + for param in sig.parameters.values(): + if param.kind == Parameter.VAR_KEYWORD: + # Replace **kwargs with individual keyword-only parameters + for field_name, field_type in Signatures_map[func_name].items(): + new_params.append( + Parameter(field_name, Parameter.KEYWORD_ONLY, default=Parameter.empty, annotation=field_type) + ) + else: + new_params.append(param) + + # Reconstruct signature with unpacked parameters + if len(new_params) != len(sig.parameters): + return sig.replace(parameters=new_params) + return sig + + except Exception: # pragma: no cover + return signature(func) + + def show_page_in_browser(page: Selector): # pragma: no cover if not page or not isinstance(page, Selector): log.error("Input must be of type `Selector`") @@ -431,7 +467,7 @@ Type 'exit' or press Ctrl+D to exit. return result - def create_wrapper(self, func): + def create_wrapper(self, func, get_signature=True): """Create a wrapper that preserves function signature but updates page""" @wraps(func) @@ -439,6 +475,12 @@ Type 'exit' or press Ctrl+D to exit. result = func(*args, **kwargs) return self.update_page(result) + if get_signature: + # Explicitly preserve and unpack signature for IPython introspection and autocompletion + wrapper.__signature__ = _unpack_signature(func) # pyright: ignore + else: + wrapper.__signature__ = signature(func) # pyright: ignore + return wrapper def get_namespace(self): @@ -451,7 +493,7 @@ Type 'exit' or press Ctrl+D to exit. delete = self.create_wrapper(self.__Fetcher.delete) dynamic_fetch = self.create_wrapper(self.__DynamicFetcher.fetch) stealthy_fetch = self.create_wrapper(self.__StealthyFetcher.fetch) - curl2fetcher = self.create_wrapper(self._curl_parser.convert2fetcher) + curl2fetcher = self.create_wrapper(self._curl_parser.convert2fetcher, get_signature=False) # Create the namespace dictionary return {