Karim shoair
549 lines
20 KiB
Python
549 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
import os
|
|
import json
|
|
from sys import stderr
|
|
from functools import wraps
|
|
from http import cookies as Cookie
|
|
from collections import namedtuple
|
|
from shlex import split as shlex_split
|
|
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 logging import (
|
|
DEBUG,
|
|
INFO,
|
|
WARNING,
|
|
ERROR,
|
|
CRITICAL,
|
|
FATAL,
|
|
getLogger,
|
|
getLevelName,
|
|
)
|
|
|
|
from IPython.terminal.embed import InteractiveShellEmbed
|
|
|
|
from scrapling import __version__
|
|
from scrapling.core.utils import log
|
|
from scrapling.parser import Adaptor, Adaptors
|
|
from scrapling.core._types import List, Optional, Dict, Tuple, Any, Union
|
|
from scrapling.fetchers import (
|
|
Fetcher,
|
|
AsyncFetcher,
|
|
DynamicFetcher,
|
|
StealthyFetcher,
|
|
Response,
|
|
)
|
|
|
|
|
|
_known_logging_levels = {
|
|
"debug": DEBUG,
|
|
"info": INFO,
|
|
"warning": WARNING,
|
|
"error": ERROR,
|
|
"critical": CRITICAL,
|
|
"fatal": FATAL,
|
|
}
|
|
|
|
|
|
# Define the structure for parsed context - Simplified for Fetcher args
|
|
Request = namedtuple(
|
|
"Request",
|
|
[
|
|
"method",
|
|
"url",
|
|
"params",
|
|
"data", # Can be str, bytes, or dict (for urlencoded)
|
|
"json_data", # Python object (dict/list) for JSON payload
|
|
"headers",
|
|
"cookies",
|
|
"proxy",
|
|
"follow_redirects", # Added for -L flag
|
|
],
|
|
)
|
|
|
|
|
|
# Suppress exit on error to handle parsing errors gracefully
|
|
class NoExitArgumentParser(ArgumentParser):
|
|
def error(self, message):
|
|
log.error(f"Curl arguments parsing error: {message}")
|
|
raise ValueError(f"Curl arguments parsing error: {message}")
|
|
|
|
def exit(self, status=0, message=None):
|
|
if message:
|
|
log.error(f"Scrapling shell exited with status {status}: {message}")
|
|
self._print_message(message, stderr)
|
|
raise ValueError(
|
|
f"Scrapling shell exited with status {status}: {message or 'Unknown reason'}"
|
|
)
|
|
|
|
|
|
class CurlParser:
|
|
"""Builds the argument parser for relevant curl flags from DevTools."""
|
|
|
|
def __init__(self):
|
|
# We will use argparse parser to parse the curl command directly instead of regex
|
|
# We will focus more on flags that will show up on curl commands copied from DevTools's network tab
|
|
_parser = NoExitArgumentParser(add_help=False) # Disable default help
|
|
# Basic curl arguments
|
|
_parser.add_argument("curl_command_placeholder", nargs="?", help=SUPPRESS)
|
|
_parser.add_argument("url")
|
|
_parser.add_argument("-X", "--request", dest="method", default=None)
|
|
_parser.add_argument("-H", "--header", action="append", default=[])
|
|
_parser.add_argument(
|
|
"-A", "--user-agent", help="Will be parsed from -H if present"
|
|
) # Note: DevTools usually includes this in -H
|
|
|
|
# Data arguments (prioritizing types common from DevTools)
|
|
_parser.add_argument("-d", "--data", default=None)
|
|
_parser.add_argument(
|
|
"--data-raw", default=None
|
|
) # Often used by browsers for JSON body
|
|
_parser.add_argument("--data-binary", default=None)
|
|
# Keep urlencode for completeness, though less common from browser copy/paste
|
|
_parser.add_argument("--data-urlencode", action="append", default=[])
|
|
_parser.add_argument(
|
|
"-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
|
|
|
|
# Connection/Security
|
|
_parser.add_argument("-k", "--insecure", action="store_true")
|
|
_parser.add_argument(
|
|
"--compressed", action="store_true"
|
|
) # Very common from browsers
|
|
|
|
# Other flags often included but may not map directly to request args
|
|
_parser.add_argument("-i", "--include", action="store_true")
|
|
_parser.add_argument("-s", "--silent", action="store_true")
|
|
_parser.add_argument("-v", "--verbose", action="store_true")
|
|
|
|
self.parser: NoExitArgumentParser = _parser
|
|
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_parser = Cookie.SimpleCookie()
|
|
cookie_parser.load(header_value)
|
|
for key, morsel in cookie_parser.items():
|
|
cookie_dict[key] = morsel.value
|
|
except Exception as e:
|
|
log.error(
|
|
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]:
|
|
"""Parses the curl command string into a structured context for Fetcher."""
|
|
|
|
clean_command = curl_command.strip().lstrip("curl").strip().replace("\\\n", " ")
|
|
|
|
try:
|
|
tokens = shlex_split(
|
|
clean_command
|
|
) # Split the string using shell-like syntax
|
|
except ValueError as e:
|
|
log.error(f"Could not split command line: {e}")
|
|
return None
|
|
|
|
try:
|
|
parsed_args, unknown = self.parser.parse_known_args(tokens)
|
|
if unknown:
|
|
log.warning(f"Ignored unknown curl arguments: {unknown}")
|
|
|
|
except ValueError:
|
|
return None
|
|
|
|
except Exception as e:
|
|
log.error(
|
|
f"An unexpected error occurred during curl arguments parsing: {e}"
|
|
)
|
|
return None
|
|
|
|
# --- Determine Method ---
|
|
method = "get" # Default
|
|
if parsed_args.get: # -G forces GET
|
|
method = "get"
|
|
|
|
elif parsed_args.method:
|
|
method = parsed_args.method.strip().lower()
|
|
|
|
# Infer POST if data is present (unless overridden by -X or -G)
|
|
elif any(
|
|
[
|
|
parsed_args.data,
|
|
parsed_args.data_raw,
|
|
parsed_args.data_binary,
|
|
parsed_args.data_urlencode,
|
|
]
|
|
):
|
|
method = "post"
|
|
|
|
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
|
|
json_payload: Optional[Any] = None
|
|
|
|
# DevTools often uses --data-raw for JSON bodies
|
|
# Precedence: --data-binary > --data-raw / -d > --data-urlencode
|
|
if parsed_args.data_binary is not None:
|
|
try:
|
|
data_payload = parsed_args.data_binary.encode("utf-8")
|
|
log.debug("Using data from --data-binary as bytes.")
|
|
except Exception as e:
|
|
log.warning(
|
|
f"Could not encode binary data '{parsed_args.data_binary}' as bytes: {e}. Using raw string."
|
|
)
|
|
data_payload = parsed_args.data_binary # Fallback to string
|
|
|
|
elif parsed_args.data_raw is not None:
|
|
data_payload = parsed_args.data_raw
|
|
|
|
elif parsed_args.data is not None:
|
|
data_payload = parsed_args.data
|
|
|
|
elif parsed_args.data_urlencode:
|
|
# Combine and parse urlencoded data
|
|
combined_data = "&".join(parsed_args.data_urlencode)
|
|
try:
|
|
data_payload = dict(parse_qsl(combined_data, keep_blank_values=True))
|
|
except Exception as e:
|
|
log.warning(
|
|
f"Could not parse urlencoded data '{combined_data}': {e}. Treating as raw string."
|
|
)
|
|
data_payload = combined_data
|
|
|
|
# Check if raw data looks like JSON, prefer 'json' param if so
|
|
if isinstance(data_payload, str):
|
|
try:
|
|
maybe_json = json.loads(data_payload)
|
|
if isinstance(maybe_json, (dict, list)):
|
|
json_payload = maybe_json
|
|
data_payload = None
|
|
except json.JSONDecodeError:
|
|
pass # Not JSON, keep it in data_payload
|
|
|
|
# Handle -G: Move data to params if method is GET
|
|
if method == "get" and data_payload:
|
|
if isinstance(data_payload, dict): # From --data-urlencode likely
|
|
params.update(data_payload)
|
|
elif isinstance(data_payload, str):
|
|
try:
|
|
params.update(dict(parse_qsl(data_payload, keep_blank_values=True)))
|
|
except ValueError:
|
|
log.warning(
|
|
f"Could not parse data '{data_payload}' into GET parameters for -G."
|
|
)
|
|
|
|
if params:
|
|
data_payload = None # Clear data as it's moved to params
|
|
json_payload = None # Should not have JSON body with -G
|
|
|
|
# --- Process Proxy ---
|
|
proxies: Optional[Dict[str, str]] = None
|
|
if parsed_args.proxy:
|
|
proxy_url = (
|
|
f"http://{parsed_args.proxy}"
|
|
if "://" not in parsed_args.proxy
|
|
else parsed_args.proxy
|
|
)
|
|
|
|
if parsed_args.proxy_user:
|
|
user_pass = parsed_args.proxy_user
|
|
parts = urlparse(proxy_url)
|
|
netloc_parts = parts.netloc.split("@")
|
|
netloc = (
|
|
f"{user_pass}@{netloc_parts[-1]}"
|
|
if len(netloc_parts) > 1
|
|
else f"{user_pass}@{parts.netloc}"
|
|
)
|
|
proxy_url = urlunparse(
|
|
(
|
|
parts.scheme,
|
|
netloc,
|
|
parts.path,
|
|
parts.params,
|
|
parts.query,
|
|
parts.fragment,
|
|
)
|
|
)
|
|
|
|
# Standard proxy dict format
|
|
proxies = {"http": proxy_url, "https": proxy_url}
|
|
log.debug(f"Using proxy configuration: {proxies}")
|
|
|
|
# --- Final Context ---
|
|
return Request(
|
|
method=method,
|
|
url=parsed_args.url,
|
|
params=params,
|
|
data=data_payload,
|
|
json_data=json_payload,
|
|
headers=headers,
|
|
cookies=cookies,
|
|
proxy=proxies,
|
|
follow_redirects=True, # Scrapling default is True
|
|
)
|
|
|
|
def convert2fetcher(self, curl_command: Union[Request, str]) -> Optional[Response]:
|
|
request = None
|
|
if isinstance(curl_command, (Request, str)):
|
|
request = (
|
|
self.parse(curl_command)
|
|
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", 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
|
|
|
|
else:
|
|
log.error("Input must be a valid curl command string or a Request object.")
|
|
|
|
return None
|
|
|
|
|
|
def show_page_in_browser(page: Adaptor):
|
|
if not page or not isinstance(page, Adaptor):
|
|
log.error("Input must be of type `Adaptor`")
|
|
return
|
|
|
|
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:
|
|
"""A custom IPython shell with minimal dependencies"""
|
|
|
|
def __init__(self, code, log_level="debug"):
|
|
self.code = code
|
|
self.page = None
|
|
self.pages = Adaptors([])
|
|
self._curl_parser = CurlParser()
|
|
log_level = log_level.strip().lower()
|
|
|
|
if _known_logging_levels.get(log_level):
|
|
self.log_level = _known_logging_levels[log_level]
|
|
else:
|
|
log.warning(f'Unknown log level "{log_level}", defaulting to "DEBUG"')
|
|
self.log_level = DEBUG
|
|
|
|
self.shell = None
|
|
|
|
# Initialize your application components
|
|
self.init_components()
|
|
|
|
def init_components(self):
|
|
"""Initialize application components"""
|
|
# This is where you'd set up your application-specific objects
|
|
if self.log_level:
|
|
getLogger("scrapling").setLevel(self.log_level)
|
|
|
|
settings = Fetcher.display_config()
|
|
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}")
|
|
|
|
@staticmethod
|
|
def banner():
|
|
"""Create a custom banner for the shell"""
|
|
return f"""
|
|
-> Available Scrapling objects:
|
|
- Fetcher/AsyncFetcher
|
|
- DynamicFetcher
|
|
- StealthyFetcher
|
|
- Adaptor
|
|
|
|
-> Useful shortcuts:
|
|
- {"get":<30} Shortcut for `Fetcher.get`
|
|
- {"post":<30} Shortcut for `Fetcher.post`
|
|
- {"put":<30} Shortcut for `Fetcher.put`
|
|
- {"delete":<30} Shortcut for `Fetcher.delete`
|
|
- {"fetch":<30} Shortcut for `DynamicFetcher.fetch`
|
|
- {"stealthy_fetch":<30} Shortcut for `StealthyFetcher.fetch`
|
|
|
|
-> 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 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)
|
|
|
|
Type 'exit' or press Ctrl+D to exit.
|
|
"""
|
|
|
|
def update_page(self, result):
|
|
"""Update current page and add to pages history"""
|
|
self.page = result
|
|
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
|
|
|
|
return result
|
|
|
|
def create_wrapper(self, func):
|
|
"""Create a wrapper that preserves function signature but updates page"""
|
|
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
result = func(*args, **kwargs)
|
|
return self.update_page(result)
|
|
|
|
return wrapper
|
|
|
|
def get_namespace(self):
|
|
"""Create a namespace with application-specific objects"""
|
|
|
|
# Create wrapped versions of fetch functions
|
|
get = self.create_wrapper(Fetcher.get)
|
|
post = self.create_wrapper(Fetcher.post)
|
|
put = self.create_wrapper(Fetcher.put)
|
|
delete = self.create_wrapper(Fetcher.delete)
|
|
dynamic_fetch = self.create_wrapper(DynamicFetcher.fetch)
|
|
stealthy_fetch = self.create_wrapper(StealthyFetcher.fetch)
|
|
curl2fetcher = self.create_wrapper(self._curl_parser.convert2fetcher)
|
|
|
|
# Create the namespace dictionary
|
|
return {
|
|
"get": get,
|
|
"post": post,
|
|
"put": put,
|
|
"delete": delete,
|
|
"Fetcher": Fetcher,
|
|
"AsyncFetcher": AsyncFetcher,
|
|
"fetch": dynamic_fetch,
|
|
"DynamicFetcher": DynamicFetcher,
|
|
"stealthy_fetch": stealthy_fetch,
|
|
"StealthyFetcher": StealthyFetcher,
|
|
"Adaptor": Adaptor,
|
|
"page": self.page,
|
|
"response": self.page,
|
|
"pages": self.pages,
|
|
"view": show_page_in_browser,
|
|
"uncurl": self._curl_parser.parse,
|
|
"curl2fetcher": curl2fetcher,
|
|
"help": self.show_help,
|
|
}
|
|
|
|
def show_help(self):
|
|
"""Show help information"""
|
|
print(self.banner())
|
|
|
|
def start(self):
|
|
"""Start the interactive shell"""
|
|
# Get our namespace with application objects
|
|
namespace = self.get_namespace()
|
|
ipython_shell = InteractiveShellEmbed(
|
|
banner1=self.banner(),
|
|
banner2="",
|
|
enable_tip=False,
|
|
exit_msg="Bye Bye",
|
|
user_ns=namespace,
|
|
)
|
|
self.shell = ipython_shell
|
|
|
|
# If a command was provided, execute it and exit
|
|
if self.code:
|
|
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
|
|
|
|
ipython_shell()
|