import copy import functools import importlib import os import warnings import requests from .config import ( DEFAULT_PROVIDER, DEFAULT_PROVIDER_API_KEY, MIN_WORD_THRESHOLD, IMAGE_DESCRIPTION_MIN_WORD_THRESHOLD, PROVIDER_MODELS, PROVIDER_MODELS_PREFIXES, SCREENSHOT_HEIGHT_TRESHOLD, PAGE_TIMEOUT, IMAGE_SCORE_THRESHOLD, SOCIAL_MEDIA_DOMAINS, ) from .user_agent_generator import UAGen, ValidUAGenerator # , OnlineUAGenerator from .extraction_strategy import ExtractionStrategy, LLMExtractionStrategy from .chunking_strategy import ChunkingStrategy, RegexChunking from .markdown_generation_strategy import MarkdownGenerationStrategy, DefaultMarkdownGenerator from .content_scraping_strategy import ContentScrapingStrategy, LXMLWebScrapingStrategy from .deep_crawling import DeepCrawlStrategy from .table_extraction import TableExtractionStrategy, DefaultTableExtraction from .cache_context import CacheMode from .proxy_strategy import ProxyRotationStrategy import inspect from typing import Any, Awaitable, Callable, Dict, List, Optional, Union from enum import Enum # Type alias for URL matching UrlMatcher = Union[str, Callable[[str], bool], List[Union[str, Callable[[str], bool]]]] def _with_defaults(cls): """Class decorator: adds set_defaults/get_defaults/reset_defaults classmethods. After decorating, every new instance resolves parameters as: explicit arg > class-level user defaults > hardcoded default Usage:: BrowserConfig.set_defaults(headless=False, viewport_width=1920) cfg = BrowserConfig() # headless=False, viewport_width=1920 cfg = BrowserConfig(headless=True) # explicit wins → headless=True """ original_init = cls.__init__ sig = inspect.signature(original_init) param_names = [p for p in sig.parameters if p != "self"] valid_params = frozenset(param_names) @functools.wraps(original_init) def wrapped_init(self, *args, **kwargs): user_defaults = type(self)._user_defaults if user_defaults: # Determine which params the caller passed explicitly explicit = set(kwargs.keys()) for i in range(len(args)): if i < len(param_names): explicit.add(param_names[i]) # Inject user defaults for non-explicit params for key, value in user_defaults.items(): if key not in explicit: kwargs[key] = copy.deepcopy(value) original_init(self, *args, **kwargs) cls.__init__ = wrapped_init cls._user_defaults = {} @classmethod def set_defaults(klass, **kwargs): """Set class-level default overrides for new instances. Args: **kwargs: Parameter names and their default values. Raises: ValueError: If any key is not a valid ``__init__`` parameter. """ invalid = set(kwargs) - valid_params if invalid: raise ValueError( f"Invalid parameter(s) for {klass.__name__}: {invalid}" ) for k, v in kwargs.items(): klass._user_defaults[k] = copy.deepcopy(v) @classmethod def get_defaults(klass): """Return a deep copy of the current class-level defaults.""" return copy.deepcopy(klass._user_defaults) @classmethod def reset_defaults(klass, *names): """Clear class-level defaults. With no arguments, removes all overrides. With arguments, removes only the named overrides. """ if names: for n in names: klass._user_defaults.pop(n, None) else: klass._user_defaults.clear() cls.set_defaults = set_defaults cls.get_defaults = get_defaults cls.reset_defaults = reset_defaults return cls class MatchMode(Enum): OR = "or" AND = "and" # from .proxy_strategy import ProxyConfig # Allowlist of types that can be deserialized via from_serializable_dict(). # This prevents arbitrary class instantiation from untrusted input (e.g. API requests). ALLOWED_DESERIALIZE_TYPES = { # Config classes "BrowserConfig", "CrawlerRunConfig", "HTTPCrawlerConfig", "LLMConfig", "ProxyConfig", "GeolocationConfig", "SeedingConfig", "VirtualScrollConfig", "LinkPreviewConfig", "DomainMapperConfig", # Extraction strategies "JsonCssExtractionStrategy", "JsonXPathExtractionStrategy", "JsonLxmlExtractionStrategy", "LLMExtractionStrategy", "CosineStrategy", "RegexExtractionStrategy", # Markdown / content "DefaultMarkdownGenerator", "PruningContentFilter", "BM25ContentFilter", "LLMContentFilter", # Scraping "LXMLWebScrapingStrategy", "PDFContentScrapingStrategy", # Chunking "RegexChunking", # Deep crawl "BFSDeepCrawlStrategy", "DFSDeepCrawlStrategy", "BestFirstCrawlingStrategy", # Filters & scorers "FilterChain", "URLPatternFilter", "DomainFilter", "ContentTypeFilter", "URLFilter", "SEOFilter", "ContentRelevanceFilter", "KeywordRelevanceScorer", "URLScorer", "CompositeScorer", "DomainAuthorityScorer", "FreshnessScorer", "PathDepthScorer", # Enums "CacheMode", "MatchMode", "DisplayMode", # Dispatchers "MemoryAdaptiveDispatcher", "SemaphoreDispatcher", # Table extraction "DefaultTableExtraction", "NoTableExtraction", "LLMTableExtraction", # Proxy "RoundRobinProxyStrategy", } # ───────────────────────── untrusted-input trust boundary ───────────────────────── # When a config is deserialized from an untrusted source (a network request body # on the Docker server), it may only construct a strict subset of types and may # only set scalar, non-power fields, which are then clamped. SDK / in-process use # is TRUSTED by default, so this changes nothing for library callers. class Provenance(Enum): """Where deserialized config data came from. TRUSTED - SDK / in-process construction (unchanged behavior). UNTRUSTED - a network request body; gated by the allowlists below. """ TRUSTED = "trusted" UNTRUSTED = "untrusted" class UntrustedConfigError(ValueError): """An untrusted request tried to construct a forbidden type or set a forbidden power-field. The Docker layer maps this to HTTP 400.""" # Types an untrusted body may construct - a strict subset of # ALLOWED_DESERIALIZE_TYPES. Deliberately EXCLUDES everything that can run code, # read secrets, route traffic, or recurse the crawl: LLMConfig, # LLMExtractionStrategy, LLMContentFilter, LLMTableExtraction, ProxyConfig, # RoundRobinProxyStrategy, all DeepCrawl*/Filter/Scorer/Dispatcher classes, # SeedingConfig and DomainMapperConfig. UNTRUSTED_ALLOWED_TYPES = { "CrawlerRunConfig", "BrowserConfig", "HTTPCrawlerConfig", "GeolocationConfig", "VirtualScrollConfig", "LinkPreviewConfig", # non-LLM extraction / markdown / scraping / chunking strategies "JsonCssExtractionStrategy", "JsonXPathExtractionStrategy", "JsonLxmlExtractionStrategy", "RegexExtractionStrategy", "CosineStrategy", "DefaultMarkdownGenerator", "PruningContentFilter", "BM25ContentFilter", "LXMLWebScrapingStrategy", "PDFContentScrapingStrategy", "RegexChunking", "DefaultTableExtraction", "NoTableExtraction", # safe scalar enums "CacheMode", "MatchMode", "DisplayMode", } # Fields that must NEVER come from an untrusted body. Presence => 400 (loud), so # a client smuggling these gets an explicit error rather than a silent drop. UNTRUSTED_FORBIDDEN_FIELDS = { "BrowserConfig": { "proxy", "proxy_config", "extra_args", "user_data_dir", "channel", "chrome_channel", "cdp_url", "debugging_port", "host", "storage_state", "cookies", "headers", "init_scripts", "browser_context_id", "target_id", }, "CrawlerRunConfig": { "js_code", "js_code_before_wait", "c4a_script", "deep_crawl_strategy", "proxy_config", "proxy_rotation_strategy", "proxy_session_id", "proxy_session_ttl", "proxy_session_auto_release", "fallback_fetch_function", "experimental", "base_url", "simulate_user", "override_navigator", "magic", "process_in_browser", "shared_data", "session_id", }, } # Scalar knobs an untrusted body MAY set, per class. A field not listed here is # dropped (forward-compatible: new SDK fields don't open a hole). Strategy-object # fields are kept so the nested object is itself type-gated by the recursion. UNTRUSTED_FIELD_ALLOWLIST = { "BrowserConfig": { "browser_type", "headless", "browser_mode", "viewport_width", "viewport_height", "viewport", "device_scale_factor", "accept_downloads", "java_script_enabled", "text_mode", "light_mode", "enable_stealth", "avoid_ads", "avoid_css", "user_agent", "user_agent_mode", "user_agent_generator_config", "verbose", "memory_saving_mode", "max_pages_before_recycle", }, "CrawlerRunConfig": { # content selection / cleaning "word_count_threshold", "only_text", "css_selector", "target_elements", "excluded_tags", "excluded_selector", "keep_data_attributes", "keep_attrs", "remove_forms", "prettiify", "parser_type", # strategy objects (nested type is gated by the recursion) "extraction_strategy", "chunking_strategy", "markdown_generator", "scraping_strategy", "table_extraction", # locale / geo "locale", "timezone_id", "geolocation", # cache "cache_mode", "bypass_cache", "disable_cache", "no_cache_read", "no_cache_write", "check_cache_freshness", "cache_validation_timeout", "fetch_ssl_certificate", # timing / waiting "wait_until", "page_timeout", "wait_for", "wait_for_timeout", "wait_for_images", "delay_before_return_html", "mean_delay", "max_range", # scrolling / rendering "ignore_body_visibility", "scan_full_page", "scroll_delay", "max_scroll_steps", "process_iframes", "flatten_shadow_dom", "remove_overlay_elements", "remove_consent_popups", "adjust_viewport_to_content", "virtual_scroll_config", # media / capture "screenshot", "screenshot_wait_for", "screenshot_height_threshold", "force_viewport_screenshot", "pdf", "capture_mhtml", "image_description_min_word_threshold", "image_score_threshold", "table_score_threshold", # links / images filtering "exclude_external_images", "exclude_all_images", "exclude_social_media_domains", "exclude_external_links", "exclude_social_media_links", "exclude_domains", "exclude_internal_links", "score_links", "preserve_https_for_internal_links", "link_preview_config", # misc safe knobs "verbose", "log_console", "capture_network_requests", "capture_console_messages", "method", "stream", "prefetch", "url", "check_robots_txt", "user_agent", "user_agent_mode", "user_agent_generator_config", "url_matcher", "match_mode", "max_retries", }, } # Upper bounds applied to attacker-influenced quantities after filtering. _MAX_TIMEOUT_MS = 60_000 _MAX_SCROLL_STEPS = 1000 _MAX_VIEWPORT = 4000 def _filter_untrusted_fields(type_name: str, params: dict) -> dict: """Drop non-allowlisted fields and raise on forbidden (power) fields.""" forbidden = UNTRUSTED_FORBIDDEN_FIELDS.get(type_name, set()) allowlist = UNTRUSTED_FIELD_ALLOWLIST.get(type_name) # None => keep all non-forbidden out = {} for key, value in params.items(): if key in forbidden: raise UntrustedConfigError( f"field '{key}' is not permitted on {type_name} from an untrusted request" ) if allowlist is not None and key not in allowlist: continue # silently drop unknown/unsafe fields (forward-compatible) out[key] = value return out def _clamp_untrusted(type_name: str, params: dict) -> dict: """Clamp attacker-influenced quantities to safe upper bounds.""" def _cap_timeout(v): # 0 historically meant "no timeout"; treat as the cap, never unbounded. if not isinstance(v, (int, float)) or v <= 0: return _MAX_TIMEOUT_MS return min(int(v), _MAX_TIMEOUT_MS) if type_name == "CrawlerRunConfig": for f in ("page_timeout", "wait_for_timeout"): if f in params: params[f] = _cap_timeout(params[f]) if isinstance(params.get("max_scroll_steps"), int): params["max_scroll_steps"] = min(params["max_scroll_steps"], _MAX_SCROLL_STEPS) elif type_name == "BrowserConfig": for f in ("viewport_width", "viewport_height"): if isinstance(params.get(f), int): params[f] = max(1, min(params[f], _MAX_VIEWPORT)) return params def _enforce_untrusted(type_name: str, params: dict) -> dict: """Filter then clamp a parameter dict for an untrusted construction.""" return _clamp_untrusted(type_name, _filter_untrusted_fields(type_name, params)) def to_serializable_dict(obj: Any, ignore_default_value : bool = False): """ Recursively convert an object to a serializable dictionary using {type, params} structure for complex objects. """ if obj is None: return None # Handle basic types if isinstance(obj, (str, int, float, bool)): return obj # Handle Enum if isinstance(obj, Enum): return {"type": obj.__class__.__name__, "params": obj.value} # Handle datetime objects if hasattr(obj, "isoformat"): return obj.isoformat() # Handle lists, tuples, and sets, and basically any iterable if isinstance(obj, (list, tuple, set)) or hasattr(obj, '__iter__') and not isinstance(obj, dict): return [to_serializable_dict(item) for item in obj] # Handle frozensets, which are not iterable if isinstance(obj, frozenset): return [to_serializable_dict(item) for item in list(obj)] # Handle dictionaries - preserve them as-is if isinstance(obj, dict): return { "type": "dict", # Mark as plain dictionary "value": {str(k): to_serializable_dict(v) for k, v in obj.items()}, } _type = obj.__class__.__name__ # Handle class instances if hasattr(obj, "__class__"): # Skip types that cannot be deserialized (e.g. logging.Logger, callables). # Only serialize objects whose type is in ALLOWED_DESERIALIZE_TYPES so that # from_serializable_dict can reconstruct them on the other side. if _type not in ALLOWED_DESERIALIZE_TYPES: return None # Get constructor signature sig = inspect.signature(obj.__class__.__init__) params = sig.parameters # Get current values current_values = {} for name, param in params.items(): if name == "self": continue value = getattr(obj, name, param.default) # Only include if different from default, considering empty values if not (is_empty_value(value) and is_empty_value(param.default)): if value != param.default and not ignore_default_value: current_values[name] = to_serializable_dict(value) # Don't serialize private __slots__ - they're internal implementation details # not constructor parameters. This was causing URLPatternFilter to fail # because _simple_suffixes was being serialized as 'simple_suffixes' # if hasattr(obj, '__slots__'): # for slot in obj.__slots__: # if slot.startswith('_'): # Handle private slots # attr_name = slot[1:] # Remove leading '_' # value = getattr(obj, slot, None) # if value is not None: # current_values[attr_name] = to_serializable_dict(value) return { "type": obj.__class__.__name__, "params": current_values } return str(obj) def from_serializable_dict(data: Any, provenance: "Provenance" = None) -> Any: """ Recursively convert a serializable dictionary back to an object instance. provenance defaults to TRUSTED (SDK/in-process). When UNTRUSTED, every typed object must be in UNTRUSTED_ALLOWED_TYPES and its params are filtered (drop unknown, raise on forbidden) and clamped before construction. """ if provenance is None: provenance = Provenance.TRUSTED if data is None: return None # Handle basic types if isinstance(data, (str, int, float, bool)): return data # Handle typed data. # Only enter the typed-object path for dicts that match the shapes produced # by to_serializable_dict(): {"type": "", "params": {...}} or # {"type": "dict", "value": {...}}. Plain business dicts that happen to # carry a "type" key (e.g. JSON-Schema fragments, JsonCss field specs like # {"type": "text", "name": "..."}) have neither "params" nor "value" and # must fall through to the raw-dict path below so they are passed as data. if ( isinstance(data, dict) and "type" in data and ("params" in data or (data["type"] == "dict" and "value" in data)) ): # Handle plain dictionaries if data["type"] == "dict" and "value" in data: return {k: from_serializable_dict(v, provenance) for k, v in data["value"].items()} # Security: only allow known-safe types to be deserialized. # Unknown types (e.g. logging.Logger serialized by older clients) are # silently dropped (returned as None) instead of crashing the request. type_name = data["type"] if type_name not in ALLOWED_DESERIALIZE_TYPES: return None # Untrusted bodies may only construct the strict subset of types and # may not set forbidden power-fields. if provenance == Provenance.UNTRUSTED and type_name not in UNTRUSTED_ALLOWED_TYPES: raise UntrustedConfigError( f"type '{type_name}' may not be constructed from an untrusted request" ) cls = None module_paths = ["crawl4ai"] for module_path in module_paths: try: mod = importlib.import_module(module_path) if hasattr(mod, type_name): cls = getattr(mod, type_name) break except (ImportError, AttributeError): continue if cls is not None: # Handle Enum if issubclass(cls, Enum): return cls(data["params"]) if "params" in data: params = data["params"] if provenance == Provenance.UNTRUSTED: params = _enforce_untrusted(type_name, dict(params)) # Handle class instances constructor_args = { k: from_serializable_dict(v, provenance) for k, v in params.items() } return cls(**constructor_args) # Handle lists if isinstance(data, list): return [from_serializable_dict(item, provenance) for item in data] # Handle raw dictionaries (legacy support) if isinstance(data, dict): return {k: from_serializable_dict(v, provenance) for k, v in data.items()} return data def is_empty_value(value: Any) -> bool: """Check if a value is effectively empty/null.""" if value is None: return True if isinstance(value, (list, tuple, set, dict, str)) and len(value) == 0: return True return False class GeolocationConfig: def __init__( self, latitude: float, longitude: float, accuracy: Optional[float] = 0.0 ): """Configuration class for geolocation settings. Args: latitude: Latitude coordinate (e.g., 37.7749) longitude: Longitude coordinate (e.g., -122.4194) accuracy: Accuracy in meters. Default: 0.0 """ self.latitude = latitude self.longitude = longitude self.accuracy = accuracy @staticmethod def from_dict(geo_dict: Dict) -> "GeolocationConfig": """Create a GeolocationConfig from a dictionary.""" return GeolocationConfig( latitude=geo_dict.get("latitude"), longitude=geo_dict.get("longitude"), accuracy=geo_dict.get("accuracy", 0.0) ) def to_dict(self) -> Dict: """Convert to dictionary representation.""" return { "latitude": self.latitude, "longitude": self.longitude, "accuracy": self.accuracy } def clone(self, **kwargs) -> "GeolocationConfig": """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: GeolocationConfig: A new instance with the specified updates """ config_dict = self.to_dict() config_dict.update(kwargs) return GeolocationConfig.from_dict(config_dict) class ProxyConfig: DIRECT = "direct" # Sentinel: use in proxy_config list to mean "no proxy" def __init__( self, server: str, username: Optional[str] = None, password: Optional[str] = None, ip: Optional[str] = None, ): """Configuration class for a single proxy. Args: server: Proxy server URL (e.g., "http://127.0.0.1:8080") username: Optional username for proxy authentication password: Optional password for proxy authentication ip: Optional IP address for verification purposes """ self.server = server self.username = username self.password = password # Extract IP from server if not explicitly provided self.ip = ip or self._extract_ip_from_server() def _extract_ip_from_server(self) -> Optional[str]: """Extract IP address from server URL.""" try: # Simple extraction assuming http://ip:port format if "://" in self.server: parts = self.server.split("://")[1].split(":") return parts[0] else: parts = self.server.split(":") return parts[0] except Exception: return None @staticmethod def from_string(proxy_str: str) -> "ProxyConfig": """Create a ProxyConfig from a string. Supported formats: - 'http://username:password@ip:port' - 'http://ip:port' - 'socks5://ip:port' - 'ip:port:username:password' - 'ip:port' """ s = (proxy_str or "").strip() # URL with credentials if "@" in s and "://" in s: auth_part, server_part = s.split("@", 1) protocol, credentials = auth_part.split("://", 1) if ":" in credentials: username, password = credentials.split(":", 1) return ProxyConfig( server=f"{protocol}://{server_part}", username=username, password=password, ) # URL without credentials (keep scheme) if "://" in s and "@" not in s: return ProxyConfig(server=s) # Colon separated forms parts = s.split(":") if len(parts) == 4: ip, port, username, password = parts return ProxyConfig(server=f"http://{ip}:{port}", username=username, password=password) if len(parts) == 2: ip, port = parts return ProxyConfig(server=f"http://{ip}:{port}") raise ValueError(f"Invalid proxy string format: {proxy_str}") @staticmethod def from_dict(proxy_dict: Dict) -> "ProxyConfig": """Create a ProxyConfig from a dictionary.""" return ProxyConfig( server=proxy_dict.get("server"), username=proxy_dict.get("username"), password=proxy_dict.get("password"), ip=proxy_dict.get("ip"), ) @staticmethod def from_env(env_var: str = "PROXIES") -> List["ProxyConfig"]: """Load proxies from environment variable. Args: env_var: Name of environment variable containing comma-separated proxy strings Returns: List of ProxyConfig objects """ proxies = [] try: proxy_list = os.getenv(env_var, "").split(",") for proxy in proxy_list: if not proxy: continue proxies.append(ProxyConfig.from_string(proxy)) except Exception as e: print(f"Error loading proxies from environment: {e}") return proxies def to_dict(self) -> Dict: """Convert to dictionary representation.""" return { "server": self.server, "username": self.username, "password": self.password, "ip": self.ip, } def clone(self, **kwargs) -> "ProxyConfig": """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: ProxyConfig: A new instance with the specified updates """ config_dict = self.to_dict() config_dict.update(kwargs) return ProxyConfig.from_dict(config_dict) @_with_defaults class BrowserConfig: """ Configuration class for setting up a browser instance and its context in AsyncPlaywrightCrawlerStrategy. This class centralizes all parameters that affect browser and context creation. Instead of passing scattered keyword arguments, users can instantiate and modify this configuration object. The crawler code will then reference these settings to initialize the browser in a consistent, documented manner. Attributes: browser_type (str): The type of browser to launch. Supported values: "chromium", "firefox", "webkit". Default: "chromium". headless (bool): Whether to run the browser in headless mode (no visible GUI). Default: True. browser_mode (str): Determines how the browser should be initialized: "builtin" - use the builtin CDP browser running in background "dedicated" - create a new dedicated browser instance each time "cdp" - use explicit CDP settings provided in cdp_url "docker" - run browser in Docker container with isolation Default: "dedicated" use_managed_browser (bool): Launch the browser using a managed approach (e.g., via CDP), allowing advanced manipulation. Default: False. cdp_url (str): URL for the Chrome DevTools Protocol (CDP) endpoint. Default: "ws://localhost:9222/devtools/browser/". browser_context_id (str or None): Pre-existing CDP browser context ID to use. When provided along with cdp_url, the crawler will reuse this context instead of creating a new one. Useful for cloud browser services that pre-create isolated contexts. Default: None. target_id (str or None): Pre-existing CDP target ID (page) to use. When provided along with browser_context_id, the crawler will reuse this target instead of creating a new page. Default: None. cdp_cleanup_on_close (bool): When True and using cdp_url, the close() method will still clean up the local Playwright client resources. Useful for cloud/server scenarios where you don't own the remote browser but need to prevent memory leaks from accumulated Playwright instances. Default: False. cdp_close_delay (float): Seconds to wait after disconnecting a CDP WebSocket before stopping the Playwright subprocess. Gives the connection time to fully release. Set to 0 to skip the delay entirely. Only applies when cdp_cleanup_on_close=True. Default: 1.0. cache_cdp_connection (bool): When True and using cdp_url, the Playwright subprocess and CDP WebSocket are cached at the class level and shared across multiple BrowserManager instances connecting to the same cdp_url. Reference-counted; the connection is only closed when the last user releases it. Eliminates the overhead of repeated Playwright/CDP setup and teardown. Default: False. create_isolated_context (bool): When True and using cdp_url, forces creation of a new browser context instead of reusing the default context. Essential for concurrent crawls on the same browser to prevent navigation conflicts. Default: False. debugging_port (int): Port for the browser debugging protocol. Default: 9222. use_persistent_context (bool): Use a persistent browser context (like a persistent profile). Automatically sets use_managed_browser=True. Default: False. user_data_dir (str or None): Path to a user data directory for persistent sessions. If None, a temporary directory may be used. Default: None. chrome_channel (str): The Chrome channel to launch (e.g., "chrome", "msedge"). Only applies if browser_type is "chromium". Default: "chromium". channel (str): The channel to launch (e.g., "chromium", "chrome", "msedge"). Only applies if browser_type is "chromium". Default: "chromium". proxy (Optional[str]): Proxy server URL (e.g., "http://username:password@proxy:port"). If None, no proxy is used. Default: None. proxy_config (ProxyConfig or dict or None): Detailed proxy configuration, e.g. {"server": "...", "username": "..."}. If None, no additional proxy config. Default: None. viewport_width (int): Default viewport width for pages. Default: 1080. viewport_height (int): Default viewport height for pages. Default: 600. viewport (dict): Default viewport dimensions for pages. If set, overrides viewport_width and viewport_height. Default: None. device_scale_factor (float): The device pixel ratio used for rendering pages. Controls how many physical pixels map to one CSS pixel, allowing simulation of HiDPI or Retina displays. For example, a viewport of 1920x1080 with a device_scale_factor of 2.0 produces screenshots at 3840x2160 resolution. Increasing this value improves screenshot quality but may increase memory usage and rendering time. Default: 1.0. verbose (bool): Enable verbose logging. Default: True. accept_downloads (bool): Whether to allow file downloads. If True, requires a downloads_path. Default: False. downloads_path (str or None): Directory to store downloaded files. If None and accept_downloads is True, a default path will be created. Default: None. storage_state (str or dict or None): An in-memory storage state (cookies, localStorage). Default: None. ignore_https_errors (bool): Ignore HTTPS certificate errors. Default: True. java_script_enabled (bool): Enable JavaScript execution in pages. Default: True. cookies (list): List of cookies to add to the browser context. Each cookie is a dict with fields like {"name": "...", "value": "...", "url": "..."}. Default: []. headers (dict): Extra HTTP headers to apply to all requests in this context. Default: {}. user_agent (str): Custom User-Agent string to use. Default: "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36". user_agent_mode (str or None): Mode for generating the user agent (e.g., "random"). If None, use the provided user_agent as-is. Default: None. user_agent_generator_config (dict or None): Configuration for user agent generation if user_agent_mode is set. Default: None. text_mode (bool): If True, disables images and other rich content for potentially faster load times. Default: False. light_mode (bool): Disables certain background features for performance gains. Default: False. extra_args (list): Additional command-line arguments passed to the browser. Default: []. enable_stealth (bool): If True, applies playwright-stealth to bypass basic bot detection. Cannot be used with use_undetected browser mode. Default: False. memory_saving_mode (bool): If True, adds aggressive cache discard and V8 heap cap flags to reduce Chromium memory growth. Recommended for high-volume crawling (1000+ pages). May slightly reduce performance due to cache eviction. Default: False. max_pages_before_recycle (int): Number of pages to crawl before recycling the browser process to reclaim leaked memory. 0 = disabled. Recommended: 500-1000 for long-running crawlers. Default: 0. avoid_ads (bool): If True, blocks ad-related and tracker network requests at the browser context level using a curated blocklist of top ad/tracker domains. Default: False. avoid_css (bool): If True, blocks loading of CSS files (css, less, scss, sass) to reduce resource usage and speed up crawling. Default: False. """ def __init__( self, browser_type: str = "chromium", headless: bool = True, browser_mode: str = "dedicated", use_managed_browser: bool = False, cdp_url: str = None, browser_context_id: str = None, target_id: str = None, cdp_cleanup_on_close: bool = False, cdp_close_delay: float = 1.0, cache_cdp_connection: bool = False, create_isolated_context: bool = False, use_persistent_context: bool = False, user_data_dir: str = None, chrome_channel: str = "chromium", channel: str = "chromium", proxy: str = None, proxy_config: Union[ProxyConfig, dict, None] = None, viewport_width: int = 1080, viewport_height: int = 600, viewport: dict = None, device_scale_factor: float = 1.0, accept_downloads: bool = False, downloads_path: str = None, storage_state: Union[str, dict, None] = None, ignore_https_errors: bool = True, java_script_enabled: bool = True, sleep_on_close: bool = False, verbose: bool = True, cookies: list = None, headers: dict = None, user_agent: str = ( # "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) AppleWebKit/537.36 " # "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " # "(KHTML, like Gecko) Chrome/116.0.5845.187 Safari/604.1 Edg/117.0.2045.47" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/116.0.0.0 Safari/537.36" ), user_agent_mode: str = "", user_agent_generator_config: dict = {}, text_mode: bool = False, light_mode: bool = False, extra_args: list = None, debugging_port: int = 9222, host: str = "localhost", enable_stealth: bool = False, avoid_ads: bool = False, avoid_css: bool = False, init_scripts: List[str] = None, memory_saving_mode: bool = False, max_pages_before_recycle: int = 0, ): self.browser_type = browser_type self.headless = headless self.browser_mode = browser_mode self.use_managed_browser = use_managed_browser self.cdp_url = cdp_url self.browser_context_id = browser_context_id self.target_id = target_id self.cdp_cleanup_on_close = cdp_cleanup_on_close self.cdp_close_delay = cdp_close_delay self.cache_cdp_connection = cache_cdp_connection self.create_isolated_context = create_isolated_context self.use_persistent_context = use_persistent_context self.user_data_dir = user_data_dir self.chrome_channel = chrome_channel or self.browser_type or "chromium" self.channel = channel or self.browser_type or "chromium" if self.browser_type in ["firefox", "webkit"]: self.channel = "" self.chrome_channel = "" if proxy: warnings.warn("The 'proxy' parameter is deprecated and will be removed in a future release. Use 'proxy_config' instead.", UserWarning) self.proxy = proxy self.proxy_config = proxy_config if isinstance(self.proxy_config, dict): self.proxy_config = ProxyConfig.from_dict(self.proxy_config) if isinstance(self.proxy_config, str): self.proxy_config = ProxyConfig.from_string(self.proxy_config) if self.proxy and self.proxy_config: warnings.warn("Both 'proxy' and 'proxy_config' are provided. 'proxy_config' will take precedence.", UserWarning) self.proxy = None elif self.proxy: # Convert proxy string to ProxyConfig if proxy_config is not provided self.proxy_config = ProxyConfig.from_string(self.proxy) self.proxy = None self.viewport_width = viewport_width self.viewport_height = viewport_height self.viewport = viewport if self.viewport is not None: self.viewport_width = self.viewport.get("width", 1080) self.viewport_height = self.viewport.get("height", 600) self.device_scale_factor = device_scale_factor self.accept_downloads = accept_downloads self.downloads_path = downloads_path self.storage_state = storage_state self.ignore_https_errors = ignore_https_errors self.java_script_enabled = java_script_enabled self.cookies = cookies if cookies is not None else [] self.headers = headers if headers is not None else {} self.user_agent = user_agent self.user_agent_mode = user_agent_mode self.user_agent_generator_config = user_agent_generator_config self.text_mode = text_mode self.light_mode = light_mode self.extra_args = extra_args if extra_args is not None else [] self.sleep_on_close = sleep_on_close self.verbose = verbose self.debugging_port = debugging_port self.host = host self.enable_stealth = enable_stealth self.avoid_ads = avoid_ads self.avoid_css = avoid_css self.init_scripts = init_scripts if init_scripts is not None else [] self.memory_saving_mode = memory_saving_mode self.max_pages_before_recycle = max_pages_before_recycle fa_user_agenr_generator = ValidUAGenerator() if self.user_agent_mode == "random": self.user_agent = fa_user_agenr_generator.generate( **(self.user_agent_generator_config or {}) ) else: pass self.browser_hint = UAGen.generate_client_hints(self.user_agent) self.headers.setdefault("sec-ch-ua", self.browser_hint) # Set appropriate browser management flags based on browser_mode if self.browser_mode == "builtin": # Builtin mode uses managed browser connecting to builtin CDP endpoint self.use_managed_browser = True # cdp_url will be set later by browser_manager elif self.browser_mode == "docker": # Docker mode uses managed browser with CDP to connect to browser in container self.use_managed_browser = True # cdp_url will be set later by docker browser strategy elif self.browser_mode == "custom" and self.cdp_url: # Custom mode with explicit CDP URL self.use_managed_browser = True elif self.browser_mode == "dedicated": # Dedicated mode uses a new browser instance each time pass # If persistent context is requested, ensure managed browser is enabled if self.use_persistent_context: self.use_managed_browser = True # Validate stealth configuration if self.enable_stealth and self.use_managed_browser and self.browser_mode == "builtin": raise ValueError( "enable_stealth cannot be used with browser_mode='builtin'. " "Stealth mode requires a dedicated browser instance." ) @staticmethod def from_kwargs(kwargs: dict) -> "BrowserConfig": # Auto-deserialize any dict values that use the {"type": ..., "params": ...} # serialization format (e.g. from JSON API requests or dump()/load() roundtrips). kwargs = { k: from_serializable_dict(v) if isinstance(v, dict) and "type" in v else v for k, v in kwargs.items() } # Only pass keys present in kwargs so that __init__ defaults (and # set_defaults() overrides) are respected for missing keys. valid = inspect.signature(BrowserConfig.__init__).parameters.keys() - {"self"} return BrowserConfig(**{k: v for k, v in kwargs.items() if k in valid}) def to_dict(self): result = { "browser_type": self.browser_type, "headless": self.headless, "browser_mode": self.browser_mode, "use_managed_browser": self.use_managed_browser, "cdp_url": self.cdp_url, "browser_context_id": self.browser_context_id, "target_id": self.target_id, "cdp_cleanup_on_close": self.cdp_cleanup_on_close, "create_isolated_context": self.create_isolated_context, "use_persistent_context": self.use_persistent_context, "user_data_dir": self.user_data_dir, "chrome_channel": self.chrome_channel, "channel": self.channel, "proxy": self.proxy, "proxy_config": self.proxy_config.to_dict() if hasattr(self.proxy_config, 'to_dict') else self.proxy_config, "viewport_width": self.viewport_width, "viewport_height": self.viewport_height, "device_scale_factor": self.device_scale_factor, "accept_downloads": self.accept_downloads, "downloads_path": self.downloads_path, "storage_state": self.storage_state, "ignore_https_errors": self.ignore_https_errors, "java_script_enabled": self.java_script_enabled, "cookies": self.cookies, "headers": self.headers, "user_agent": self.user_agent, "user_agent_mode": self.user_agent_mode, "user_agent_generator_config": self.user_agent_generator_config, "text_mode": self.text_mode, "light_mode": self.light_mode, "extra_args": self.extra_args, "sleep_on_close": self.sleep_on_close, "verbose": self.verbose, "debugging_port": self.debugging_port, "host": self.host, "enable_stealth": self.enable_stealth, "avoid_ads": self.avoid_ads, "avoid_css": self.avoid_css, "init_scripts": self.init_scripts, "memory_saving_mode": self.memory_saving_mode, "max_pages_before_recycle": self.max_pages_before_recycle, } return result def clone(self, **kwargs): """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: BrowserConfig: A new instance with the specified updates """ config_dict = self.to_dict() config_dict.update(kwargs) return BrowserConfig.from_kwargs(config_dict) # Create a funciton returns dict of the object def dump(self) -> dict: # Serialize the object to a dictionary return to_serializable_dict(self) @staticmethod def load(data: dict, provenance: "Provenance" = None) -> "BrowserConfig": # Deserialize the object from a dictionary if provenance is None: provenance = Provenance.TRUSTED config = from_serializable_dict(data, provenance) if isinstance(config, BrowserConfig): return config # Plain-dict path: enforce the untrusted gate on top-level fields too, # so a body that sends bare kwargs (no {type,params}) cannot bypass it. if provenance == Provenance.UNTRUSTED and isinstance(config, dict): config = _enforce_untrusted("BrowserConfig", config) return BrowserConfig.from_kwargs(config) def set_nstproxy( self, token: str, channel_id: str, country: str = "ANY", state: str = "", city: str = "", protocol: str = "http", session_duration: int = 10, ): """ Fetch a proxy from NSTProxy API and automatically assign it to proxy_config. Get your NSTProxy token from: https://app.nstproxy.com/profile Args: token (str): NSTProxy API token. channel_id (str): NSTProxy channel ID. country (str, optional): Country code (default: "ANY"). state (str, optional): State code (default: ""). city (str, optional): City name (default: ""). protocol (str, optional): Proxy protocol ("http" or "socks5"). Defaults to "http". session_duration (int, optional): Session duration in minutes (0 = rotate each request). Defaults to 10. Raises: ValueError: If the API response format is invalid. PermissionError: If the API returns an error message. """ # --- Validate input early --- if not token or not channel_id: raise ValueError("[NSTProxy] token and channel_id are required") if protocol not in ("http", "socks5"): raise ValueError(f"[NSTProxy] Invalid protocol: {protocol}") # --- Build NSTProxy API URL --- params = { "fType": 2, "count": 1, "channelId": channel_id, "country": country, "protocol": protocol, "sessionDuration": session_duration, "token": token, } if state: params["state"] = state if city: params["city"] = city url = "https://api.nstproxy.com/api/v1/generate/apiproxies" try: response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() # --- Handle API error response --- if isinstance(data, dict) and data.get("err"): raise PermissionError(f"[NSTProxy] API Error: {data.get('msg', 'Unknown error')}") if not isinstance(data, list) or not data: raise ValueError("[NSTProxy] Invalid API response — expected a non-empty list") proxy_info = data[0] # --- Apply proxy config --- self.proxy_config = ProxyConfig( server=f"{protocol}://{proxy_info['ip']}:{proxy_info['port']}", username=proxy_info["username"], password=proxy_info["password"], ) except Exception as e: print(f"[NSTProxy] ❌ Failed to set proxy: {e}") raise class VirtualScrollConfig: """Configuration for virtual scroll handling. This config enables capturing content from pages with virtualized scrolling (like Twitter, Instagram feeds) where DOM elements are recycled as user scrolls. """ def __init__( self, container_selector: str, scroll_count: int = 10, scroll_by: Union[str, int] = "container_height", wait_after_scroll: float = 0.5, ): """ Initialize virtual scroll configuration. Args: container_selector: CSS selector for the scrollable container scroll_count: Maximum number of scrolls to perform scroll_by: Amount to scroll - can be: - "container_height": scroll by container's height - "page_height": scroll by viewport height - int: fixed pixel amount wait_after_scroll: Seconds to wait after each scroll for content to load """ self.container_selector = container_selector self.scroll_count = scroll_count self.scroll_by = scroll_by self.wait_after_scroll = wait_after_scroll def to_dict(self) -> dict: """Convert to dictionary for serialization.""" return { "container_selector": self.container_selector, "scroll_count": self.scroll_count, "scroll_by": self.scroll_by, "wait_after_scroll": self.wait_after_scroll, } @classmethod def from_dict(cls, data: dict) -> "VirtualScrollConfig": """Create instance from dictionary.""" return cls(**data) class LinkPreviewConfig: """Configuration for link head extraction and scoring.""" def __init__( self, include_internal: bool = True, include_external: bool = False, include_patterns: Optional[List[str]] = None, exclude_patterns: Optional[List[str]] = None, concurrency: int = 10, timeout: int = 5, max_links: int = 100, query: Optional[str] = None, score_threshold: Optional[float] = None, verbose: bool = False ): """ Initialize link extraction configuration. Args: include_internal: Whether to include same-domain links include_external: Whether to include different-domain links include_patterns: List of glob patterns to include (e.g., ["*/docs/*", "*/api/*"]) exclude_patterns: List of glob patterns to exclude (e.g., ["*/login*", "*/admin*"]) concurrency: Number of links to process simultaneously timeout: Timeout in seconds for each link's head extraction max_links: Maximum number of links to process (prevents overload) query: Query string for BM25 contextual scoring (optional) score_threshold: Minimum relevance score to include links (0.0-1.0, optional) verbose: Show detailed progress during extraction """ self.include_internal = include_internal self.include_external = include_external self.include_patterns = include_patterns self.exclude_patterns = exclude_patterns self.concurrency = concurrency self.timeout = timeout self.max_links = max_links self.query = query self.score_threshold = score_threshold self.verbose = verbose # Validation if concurrency <= 0: raise ValueError("concurrency must be positive") if timeout <= 0: raise ValueError("timeout must be positive") if max_links <= 0: raise ValueError("max_links must be positive") if score_threshold is not None and not (0.0 <= score_threshold <= 1.0): raise ValueError("score_threshold must be between 0.0 and 1.0") if not include_internal and not include_external: raise ValueError("At least one of include_internal or include_external must be True") @staticmethod def from_dict(config_dict: Dict[str, Any]) -> "LinkPreviewConfig": """Create LinkPreviewConfig from dictionary (for backward compatibility).""" if not config_dict: return None return LinkPreviewConfig( include_internal=config_dict.get("include_internal", True), include_external=config_dict.get("include_external", False), include_patterns=config_dict.get("include_patterns"), exclude_patterns=config_dict.get("exclude_patterns"), concurrency=config_dict.get("concurrency", 10), timeout=config_dict.get("timeout", 5), max_links=config_dict.get("max_links", 100), query=config_dict.get("query"), score_threshold=config_dict.get("score_threshold"), verbose=config_dict.get("verbose", False) ) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary format.""" return { "include_internal": self.include_internal, "include_external": self.include_external, "include_patterns": self.include_patterns, "exclude_patterns": self.exclude_patterns, "concurrency": self.concurrency, "timeout": self.timeout, "max_links": self.max_links, "query": self.query, "score_threshold": self.score_threshold, "verbose": self.verbose } def clone(self, **kwargs) -> "LinkPreviewConfig": """Create a copy with updated values.""" config_dict = self.to_dict() config_dict.update(kwargs) return LinkPreviewConfig.from_dict(config_dict) class HTTPCrawlerConfig: """HTTP-specific crawler configuration""" method: str = "GET" headers: Optional[Dict[str, str]] = None data: Optional[Dict[str, Any]] = None json: Optional[Dict[str, Any]] = None follow_redirects: bool = True verify_ssl: bool = True downloads_path: Optional[str] = None def __init__( self, method: str = "GET", headers: Optional[Dict[str, str]] = None, data: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, follow_redirects: bool = True, verify_ssl: bool = True, downloads_path: Optional[str] = None, ): self.method = method self.headers = headers self.data = data self.json = json self.follow_redirects = follow_redirects self.verify_ssl = verify_ssl self.downloads_path = downloads_path @staticmethod def from_kwargs(kwargs: dict) -> "HTTPCrawlerConfig": return HTTPCrawlerConfig( method=kwargs.get("method", "GET"), headers=kwargs.get("headers"), data=kwargs.get("data"), json=kwargs.get("json"), follow_redirects=kwargs.get("follow_redirects", True), verify_ssl=kwargs.get("verify_ssl", True), downloads_path=kwargs.get("downloads_path"), ) def to_dict(self): return { "method": self.method, "headers": self.headers, "data": self.data, "json": self.json, "follow_redirects": self.follow_redirects, "verify_ssl": self.verify_ssl, "downloads_path": self.downloads_path, } def clone(self, **kwargs): """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: HTTPCrawlerConfig: A new instance with the specified updates """ config_dict = self.to_dict() config_dict.update(kwargs) return HTTPCrawlerConfig.from_kwargs(config_dict) def dump(self) -> dict: return to_serializable_dict(self) @staticmethod def load(data: dict, provenance: "Provenance" = None) -> "HTTPCrawlerConfig": if provenance is None: provenance = Provenance.TRUSTED config = from_serializable_dict(data, provenance) if isinstance(config, HTTPCrawlerConfig): return config if provenance == Provenance.UNTRUSTED and isinstance(config, dict): config = _enforce_untrusted("HTTPCrawlerConfig", config) return HTTPCrawlerConfig.from_kwargs(config) @_with_defaults class CrawlerRunConfig(): """ Configuration class for controlling how the crawler runs each crawl operation. This includes parameters for content extraction, page manipulation, waiting conditions, caching, and other runtime behaviors. This centralizes parameters that were previously scattered as kwargs to `arun()` and related methods. By using this class, you have a single place to understand and adjust the crawling options. Attributes: # Deep Crawl Parameters deep_crawl_strategy (DeepCrawlStrategy or None): Strategy to use for deep crawling. # Content Processing Parameters word_count_threshold (int): Minimum word count threshold before processing content. Default: MIN_WORD_THRESHOLD (typically 200). extraction_strategy (ExtractionStrategy or None): Strategy to extract structured data from crawled pages. Default: None (NoExtractionStrategy is used if None). chunking_strategy (ChunkingStrategy): Strategy to chunk content before extraction. Default: RegexChunking(). markdown_generator (MarkdownGenerationStrategy): Strategy for generating markdown. Default: None. only_text (bool): If True, attempt to extract text-only content where applicable. Default: False. css_selector (str or None): CSS selector to extract a specific portion of the page. Default: None. target_elements (list of str or None): List of CSS selectors for specific elements for Markdown generation and structured data extraction. When you set this, only the contents of these elements are processed for extraction and Markdown generation. If you do not set any value, the entire page is processed. The difference between this and css_selector is that this will shrink the initial raw HTML to the selected element, while this will only affect the extraction and Markdown generation. Default: None excluded_tags (list of str or None): List of HTML tags to exclude from processing. Default: None. excluded_selector (str or None): CSS selector to exclude from processing. Default: None. keep_data_attributes (bool): If True, retain `data-*` attributes while removing unwanted attributes. Default: False. keep_attrs (list of str): List of HTML attributes to keep during processing. Default: []. remove_forms (bool): If True, remove all `
` elements from the HTML. Default: False. prettiify (bool): If True, apply `fast_format_html` to produce prettified HTML output. Default: False. parser_type (str): Type of parser to use for HTML parsing. Default: "lxml". scraping_strategy (ContentScrapingStrategy): Scraping strategy to use. Default: LXMLWebScrapingStrategy. proxy_config (ProxyConfig or dict or None): Detailed proxy configuration, e.g. {"server": "...", "username": "..."}. If None, no additional proxy config. Default: None. # Sticky Proxy Session Parameters proxy_session_id (str or None): When set, maintains the same proxy for all requests sharing this session ID. The proxy is acquired on first request and reused for subsequent requests. Session expires when explicitly released or crawler context is closed. Default: None. proxy_session_ttl (int or None): Time-to-live for sticky session in seconds. After TTL expires, a new proxy is acquired on next request. Default: None (session lasts until explicitly released or crawler closes). proxy_session_auto_release (bool): If True, automatically release the proxy session after a batch operation. Useful for arun_many() to clean up sessions automatically. Default: False. # Browser Location and Identity Parameters locale (str or None): Locale to use for the browser context (e.g., "en-US"). Default: None. timezone_id (str or None): Timezone identifier to use for the browser context (e.g., "America/New_York"). Default: None. geolocation (GeolocationConfig or None): Geolocation configuration for the browser. Default: None. # SSL Parameters fetch_ssl_certificate: bool = False, # Caching Parameters cache_mode (CacheMode or None): Defines how caching is handled. If None, defaults to CacheMode.ENABLED internally. Default: CacheMode.BYPASS. session_id (str or None): Optional session ID to persist the browser context and the created page instance. If the ID already exists, the crawler does not create a new page and uses the current page to preserve the state. bypass_cache (bool): Legacy parameter, if True acts like CacheMode.BYPASS. Default: False. disable_cache (bool): Legacy parameter, if True acts like CacheMode.DISABLED. Default: False. no_cache_read (bool): Legacy parameter, if True acts like CacheMode.WRITE_ONLY. Default: False. no_cache_write (bool): Legacy parameter, if True acts like CacheMode.READ_ONLY. Default: False. shared_data (dict or None): Shared data to be passed between hooks. Default: None. # Cache Validation Parameters (Smart Cache) check_cache_freshness (bool): If True, validates cached content freshness using HTTP conditional requests (ETag/Last-Modified) and head fingerprinting before returning cached results. Avoids full browser crawls when content hasn't changed. Only applies when cache_mode allows reads. Default: False. cache_validation_timeout (float): Timeout in seconds for cache validation HTTP requests. Default: 10.0. # Page Navigation and Timing Parameters wait_until (str): The condition to wait for when navigating, e.g. "domcontentloaded". Default: "domcontentloaded". page_timeout (int): Timeout in ms for page operations like navigation. Default: 60000 (60 seconds). wait_for (str or None): A CSS selector or JS condition to wait for before extracting content. Default: None. wait_for_timeout (int or None): Specific timeout in ms for the wait_for condition. If None, uses page_timeout instead. Default: None. wait_for_images (bool): If True, wait for images to load before extracting content. Default: False. delay_before_return_html (float): Delay in seconds before retrieving final HTML. Default: 0.1. mean_delay (float): Mean base delay between requests when calling arun_many. Default: 0.1. max_range (float): Max random additional delay range for requests in arun_many. Default: 0.3. semaphore_count (int): Number of concurrent operations allowed. Default: 5. # Page Interaction Parameters js_code (str or list of str or None): JavaScript code/snippets to run on the page after wait_for and delay_before_return_html. Default: None. js_code_before_wait (str or list of str or None): JavaScript to run BEFORE wait_for. Use for triggering loading that wait_for then checks. Default: None. js_only (bool): If True, indicates subsequent calls are JS-driven updates, not full page loads. Default: False. ignore_body_visibility (bool): If True, ignore whether the body is visible before proceeding. Default: True. scan_full_page (bool): If True, scroll through the entire page to load all content. Default: False. scroll_delay (float): Delay in seconds between scroll steps if scan_full_page is True. Default: 0.2. max_scroll_steps (Optional[int]): Maximum number of scroll steps to perform during full page scan. If None, scrolls until the entire page is loaded. Default: None. process_iframes (bool): If True, attempts to process and inline iframe content. Default: False. flatten_shadow_dom (bool): If True, flatten shadow DOM content into the light DOM before HTML capture so page.content() includes it. Also injects an init script to force-open closed shadow roots. Default: False. remove_overlay_elements (bool): If True, remove overlays/popups before extracting HTML. Default: False. remove_consent_popups (bool): If True, remove GDPR/cookie consent popups (IAB TCF/CMP) before extracting HTML. Targets known CMP providers like OneTrust, Cookiebot, TrustArc, Quantcast, Didomi, etc. Default: False. simulate_user (bool): If True, simulate user interactions (mouse moves, clicks) for anti-bot measures. Default: False. override_navigator (bool): If True, overrides navigator properties for more human-like behavior. Default: False. magic (bool): If True, attempts automatic handling of overlays/popups. Default: False. adjust_viewport_to_content (bool): If True, adjust viewport according to the page content dimensions. Default: False. # Media Handling Parameters screenshot (bool): Whether to take a screenshot after crawling. Default: False. screenshot_wait_for (float or None): Additional wait time before taking a screenshot. Default: None. screenshot_height_threshold (int): Threshold for page height to decide screenshot strategy. Default: SCREENSHOT_HEIGHT_TRESHOLD (from config, e.g. 20000). force_viewport_screenshot (bool): If True, always take viewport-only screenshots regardless of page height. When False, uses automatic decision (viewport for short pages, full-page for long pages). Default: False. pdf (bool): Whether to generate a PDF of the page. Default: False. image_description_min_word_threshold (int): Minimum words for image description extraction. Default: IMAGE_DESCRIPTION_MIN_WORD_THRESHOLD (e.g., 50). image_score_threshold (int): Minimum score threshold for processing an image. Default: IMAGE_SCORE_THRESHOLD (e.g., 3). exclude_external_images (bool): If True, exclude all external images from processing. Default: False. table_score_threshold (int): Minimum score threshold for processing a table. Default: 7. table_extraction (TableExtractionStrategy): Strategy to use for table extraction. Default: DefaultTableExtraction with table_score_threshold. # Virtual Scroll Parameters virtual_scroll_config (VirtualScrollConfig or dict or None): Configuration for handling virtual scroll containers. Used for capturing content from pages with virtualized scrolling (e.g., Twitter, Instagram feeds). Default: None. # Link and Domain Handling Parameters exclude_social_media_domains (list of str): List of domains to exclude for social media links. Default: SOCIAL_MEDIA_DOMAINS (from config). exclude_external_links (bool): If True, exclude all external links from the results. Default: False. exclude_internal_links (bool): If True, exclude internal links from the results. Default: False. exclude_social_media_links (bool): If True, exclude links pointing to social media domains. Default: False. exclude_domains (list of str): List of specific domains to exclude from results. Default: []. exclude_internal_links (bool): If True, exclude internal links from the results. Default: False. score_links (bool): If True, calculate intrinsic quality scores for all links using URL structure, text quality, and contextual relevance metrics. Separate from link_preview_config. Default: False. # Debugging and Logging Parameters verbose (bool): Enable verbose logging. Default: True. log_console (bool): If True, log console messages from the page. Default: False. # HTTP Crwler Strategy Parameters method (str): HTTP method to use for the request, when using AsyncHTTPCrwalerStrategy. Default: "GET". data (dict): Data to send in the request body, when using AsyncHTTPCrwalerStrategy. Default: None. json (dict): JSON data to send in the request body, when using AsyncHTTPCrwalerStrategy. # Connection Parameters stream (bool): If True, enables streaming of crawled URLs as they are processed when used with arun_many. Default: False. process_in_browser (bool): If True, forces raw:/file:// URLs to be processed through the browser pipeline (enabling js_code, wait_for, scrolling, etc.). When False (default), raw:/file:// URLs use a fast path that returns HTML directly without browser interaction. This is automatically enabled when browser-requiring parameters are detected (js_code, wait_for, screenshot, pdf, etc.). Default: False. check_robots_txt (bool): Whether to check robots.txt rules before crawling. Default: False Default: False. user_agent (str): Custom User-Agent string to use. Default: None. user_agent_mode (str or None): Mode for generating the user agent (e.g., "random"). If None, use the provided user_agent as-is. Default: None. user_agent_generator_config (dict or None): Configuration for user agent generation if user_agent_mode is set. Default: None. # Experimental Parameters experimental (dict): Dictionary containing experimental parameters that are in beta phase. This allows passing temporary features that are not yet fully integrated into the main parameter set. Default: None. url: str = None # This is not a compulsory parameter """ _UNWANTED_PROPS = { 'disable_cache' : 'Instead, use cache_mode=CacheMode.DISABLED', 'bypass_cache' : 'Instead, use cache_mode=CacheMode.BYPASS', 'no_cache_read' : 'Instead, use cache_mode=CacheMode.WRITE_ONLY', 'no_cache_write' : 'Instead, use cache_mode=CacheMode.READ_ONLY', } def __init__( self, # Content Processing Parameters word_count_threshold: int = MIN_WORD_THRESHOLD, extraction_strategy: ExtractionStrategy = None, chunking_strategy: ChunkingStrategy = RegexChunking(), markdown_generator: MarkdownGenerationStrategy = DefaultMarkdownGenerator(), only_text: bool = False, css_selector: str = None, target_elements: List[str] = None, excluded_tags: list = None, excluded_selector: str = None, keep_data_attributes: bool = False, keep_attrs: list = None, remove_forms: bool = False, prettiify: bool = False, parser_type: str = "lxml", scraping_strategy: ContentScrapingStrategy = None, proxy_config: Union["ProxyConfig", List["ProxyConfig"], dict, str, None] = None, proxy_rotation_strategy: Optional[ProxyRotationStrategy] = None, # Sticky Proxy Session Parameters proxy_session_id: Optional[str] = None, proxy_session_ttl: Optional[int] = None, proxy_session_auto_release: bool = False, # Browser Location and Identity Parameters locale: Optional[str] = None, timezone_id: Optional[str] = None, geolocation: Optional[GeolocationConfig] = None, # SSL Parameters fetch_ssl_certificate: bool = False, # Caching Parameters cache_mode: CacheMode = CacheMode.BYPASS, session_id: str = None, bypass_cache: bool = False, disable_cache: bool = False, no_cache_read: bool = False, no_cache_write: bool = False, shared_data: dict = None, # Cache Validation Parameters (Smart Cache) check_cache_freshness: bool = False, cache_validation_timeout: float = 10.0, # Page Navigation and Timing Parameters wait_until: str = "domcontentloaded", page_timeout: int = PAGE_TIMEOUT, wait_for: str = None, wait_for_timeout: int = None, wait_for_images: bool = False, delay_before_return_html: float = 0.1, mean_delay: float = 0.1, max_range: float = 0.3, semaphore_count: int = 5, # Page Interaction Parameters js_code: Union[str, List[str]] = None, js_code_before_wait: Union[str, List[str]] = None, c4a_script: Union[str, List[str]] = None, js_only: bool = False, ignore_body_visibility: bool = True, scan_full_page: bool = False, scroll_delay: float = 0.2, max_scroll_steps: Optional[int] = None, process_iframes: bool = False, flatten_shadow_dom: bool = False, remove_overlay_elements: bool = False, remove_consent_popups: bool = False, simulate_user: bool = False, override_navigator: bool = False, magic: bool = False, adjust_viewport_to_content: bool = False, # Media Handling Parameters screenshot: bool = False, screenshot_wait_for: float = None, screenshot_height_threshold: int = SCREENSHOT_HEIGHT_TRESHOLD, force_viewport_screenshot: bool = False, pdf: bool = False, capture_mhtml: bool = False, image_description_min_word_threshold: int = IMAGE_DESCRIPTION_MIN_WORD_THRESHOLD, image_score_threshold: int = IMAGE_SCORE_THRESHOLD, table_score_threshold: int = 7, table_extraction: TableExtractionStrategy = None, exclude_external_images: bool = False, exclude_all_images: bool = False, # Link and Domain Handling Parameters exclude_social_media_domains: list = None, exclude_external_links: bool = False, exclude_social_media_links: bool = False, exclude_domains: list = None, exclude_internal_links: bool = False, score_links: bool = False, preserve_https_for_internal_links: bool = False, # Debugging and Logging Parameters verbose: bool = True, log_console: bool = False, # Network and Console Capturing Parameters capture_network_requests: bool = False, capture_console_messages: bool = False, # Connection Parameters method: str = "GET", stream: bool = False, prefetch: bool = False, # When True, return only HTML + links (skip heavy processing) process_in_browser: bool = False, # Force browser processing for raw:/file:// URLs url: str = None, base_url: str = None, # Base URL for markdown link resolution (used with raw: HTML) check_robots_txt: bool = False, user_agent: str = None, user_agent_mode: str = None, user_agent_generator_config: dict = {}, # Deep Crawl Parameters deep_crawl_strategy: Optional[DeepCrawlStrategy] = None, # Link Extraction Parameters link_preview_config: Union[LinkPreviewConfig, Dict[str, Any]] = None, # Virtual Scroll Parameters virtual_scroll_config: Union[VirtualScrollConfig, Dict[str, Any]] = None, # URL Matching Parameters url_matcher: Optional[UrlMatcher] = None, match_mode: MatchMode = MatchMode.OR, # Experimental Parameters experimental: Dict[str, Any] = None, # Anti-Bot Retry Parameters max_retries: int = 0, fallback_fetch_function: Optional[Callable[[str], Awaitable[str]]] = None, ): # TODO: Planning to set properties dynamically based on the __init__ signature self.url = url self.base_url = base_url # Base URL for markdown link resolution # Content Processing Parameters self.word_count_threshold = word_count_threshold self.extraction_strategy = extraction_strategy self.chunking_strategy = chunking_strategy self.markdown_generator = markdown_generator self.only_text = only_text self.css_selector = css_selector self.target_elements = target_elements or [] self.excluded_tags = excluded_tags or [] self.excluded_selector = excluded_selector or "" self.keep_data_attributes = keep_data_attributes self.keep_attrs = keep_attrs or [] self.remove_forms = remove_forms self.prettiify = prettiify self.parser_type = parser_type self.scraping_strategy = scraping_strategy or LXMLWebScrapingStrategy() self.proxy_config = proxy_config # runs through property setter self.proxy_rotation_strategy = proxy_rotation_strategy # Sticky Proxy Session Parameters self.proxy_session_id = proxy_session_id self.proxy_session_ttl = proxy_session_ttl self.proxy_session_auto_release = proxy_session_auto_release # Browser Location and Identity Parameters self.locale = locale self.timezone_id = timezone_id self.geolocation = geolocation # SSL Parameters self.fetch_ssl_certificate = fetch_ssl_certificate # Caching Parameters self.cache_mode = cache_mode self.session_id = session_id self.bypass_cache = bypass_cache self.disable_cache = disable_cache self.no_cache_read = no_cache_read self.no_cache_write = no_cache_write self.shared_data = shared_data # Cache Validation (Smart Cache) self.check_cache_freshness = check_cache_freshness self.cache_validation_timeout = cache_validation_timeout # Page Navigation and Timing Parameters self.wait_until = wait_until self.page_timeout = page_timeout self.wait_for = wait_for self.wait_for_timeout = wait_for_timeout self.wait_for_images = wait_for_images self.delay_before_return_html = delay_before_return_html self.mean_delay = mean_delay self.max_range = max_range self.semaphore_count = semaphore_count # Page Interaction Parameters self.js_code = js_code self.js_code_before_wait = js_code_before_wait self.c4a_script = c4a_script self.js_only = js_only self.ignore_body_visibility = ignore_body_visibility self.scan_full_page = scan_full_page self.scroll_delay = scroll_delay self.max_scroll_steps = max_scroll_steps self.process_iframes = process_iframes self.flatten_shadow_dom = flatten_shadow_dom self.remove_overlay_elements = remove_overlay_elements self.remove_consent_popups = remove_consent_popups self.simulate_user = simulate_user self.override_navigator = override_navigator self.magic = magic self.adjust_viewport_to_content = adjust_viewport_to_content # Media Handling Parameters self.screenshot = screenshot self.screenshot_wait_for = screenshot_wait_for self.screenshot_height_threshold = screenshot_height_threshold self.force_viewport_screenshot = force_viewport_screenshot self.pdf = pdf self.capture_mhtml = capture_mhtml self.image_description_min_word_threshold = image_description_min_word_threshold self.image_score_threshold = image_score_threshold self.exclude_external_images = exclude_external_images self.exclude_all_images = exclude_all_images self.table_score_threshold = table_score_threshold # Table extraction strategy (default to DefaultTableExtraction if not specified) if table_extraction is None: self.table_extraction = DefaultTableExtraction(table_score_threshold=table_score_threshold) else: self.table_extraction = table_extraction # Link and Domain Handling Parameters self.exclude_social_media_domains = ( exclude_social_media_domains or SOCIAL_MEDIA_DOMAINS ) self.exclude_external_links = exclude_external_links self.exclude_social_media_links = exclude_social_media_links self.exclude_domains = exclude_domains or [] self.exclude_internal_links = exclude_internal_links self.score_links = score_links self.preserve_https_for_internal_links = preserve_https_for_internal_links # Debugging and Logging Parameters self.verbose = verbose self.log_console = log_console # Network and Console Capturing Parameters self.capture_network_requests = capture_network_requests self.capture_console_messages = capture_console_messages # Connection Parameters self.stream = stream self.prefetch = prefetch # Prefetch mode: return only HTML + links self.process_in_browser = process_in_browser # Force browser processing for raw:/file:// URLs self.method = method # Robots.txt Handling Parameters self.check_robots_txt = check_robots_txt # User Agent Parameters self.user_agent = user_agent self.user_agent_mode = user_agent_mode self.user_agent_generator_config = user_agent_generator_config # Validate type of extraction strategy and chunking strategy if they are provided if self.extraction_strategy is not None and not isinstance( self.extraction_strategy, ExtractionStrategy ): raise ValueError( "extraction_strategy must be an instance of ExtractionStrategy" ) if self.chunking_strategy is not None and not isinstance( self.chunking_strategy, ChunkingStrategy ): raise ValueError( "chunking_strategy must be an instance of ChunkingStrategy" ) if self.markdown_generator is not None and not isinstance( self.markdown_generator, MarkdownGenerationStrategy ): hint = "" if isinstance(self.markdown_generator, dict): hint = ( ' The JSON format must be {"type": "", "params": {...}}.' ' Note: "params" is required — "options" or other keys are not recognized.' ) raise ValueError( "markdown_generator must be an instance of MarkdownGenerationStrategy, " f"got {type(self.markdown_generator).__name__}.{hint}" ) # Set default chunking strategy if None if self.chunking_strategy is None: self.chunking_strategy = RegexChunking() # Deep Crawl Parameters self.deep_crawl_strategy = deep_crawl_strategy # Link Extraction Parameters if link_preview_config is None: self.link_preview_config = None elif isinstance(link_preview_config, LinkPreviewConfig): self.link_preview_config = link_preview_config elif isinstance(link_preview_config, dict): # Convert dict to config object for backward compatibility self.link_preview_config = LinkPreviewConfig.from_dict(link_preview_config) else: raise ValueError("link_preview_config must be LinkPreviewConfig object or dict") # Virtual Scroll Parameters if virtual_scroll_config is None: self.virtual_scroll_config = None elif isinstance(virtual_scroll_config, VirtualScrollConfig): self.virtual_scroll_config = virtual_scroll_config elif isinstance(virtual_scroll_config, dict): # Convert dict to config object for backward compatibility self.virtual_scroll_config = VirtualScrollConfig.from_dict(virtual_scroll_config) else: raise ValueError("virtual_scroll_config must be VirtualScrollConfig object or dict") # URL Matching Parameters self.url_matcher = url_matcher self.match_mode = match_mode # Experimental Parameters self.experimental = experimental or {} # Anti-Bot Retry Parameters self.max_retries = max_retries self.fallback_fetch_function = fallback_fetch_function # Compile C4A scripts if provided if self.c4a_script and not self.js_code: self._compile_c4a_script() @staticmethod def _normalize_proxy_config(value): """Normalize proxy_config to ProxyConfig, list of ProxyConfig/None, or None.""" if isinstance(value, list): normalized = [] for p in value: if p is None or p == "direct": normalized.append(None) elif isinstance(p, dict): normalized.append(ProxyConfig.from_dict(p)) elif isinstance(p, str): normalized.append(ProxyConfig.from_string(p)) else: normalized.append(p) return normalized elif isinstance(value, dict): return ProxyConfig.from_dict(value) elif isinstance(value, str): if value == "direct": return None return ProxyConfig.from_string(value) return value # ProxyConfig or None @property def proxy_config(self): return self._proxy_config @proxy_config.setter def proxy_config(self, value): self._proxy_config = CrawlerRunConfig._normalize_proxy_config(value) def _get_proxy_list(self) -> list: """Normalize proxy_config to a list for the retry loop.""" if self.proxy_config is None: return [None] if isinstance(self.proxy_config, list): return self.proxy_config if self.proxy_config else [None] return [self.proxy_config] def _compile_c4a_script(self): """Compile C4A script to JavaScript""" try: # Try importing the compiler try: from .script import compile except ImportError: from crawl4ai.script import compile # Handle both string and list inputs if isinstance(self.c4a_script, str): scripts = [self.c4a_script] else: scripts = self.c4a_script # Compile each script compiled_js = [] for i, script in enumerate(scripts): result = compile(script) if result.success: compiled_js.extend(result.js_code) else: # Format error message following existing patterns error = result.first_error error_msg = ( f"C4A Script compilation error (script {i+1}):\n" f" Line {error.line}, Column {error.column}: {error.message}\n" f" Code: {error.source_line}" ) if error.suggestions: error_msg += f"\n Suggestion: {error.suggestions[0].message}" raise ValueError(error_msg) self.js_code = compiled_js except ImportError: raise ValueError( "C4A script compiler not available. " "Please ensure crawl4ai.script module is properly installed." ) except Exception as e: # Re-raise with context if "compilation error" not in str(e).lower(): raise ValueError(f"Failed to compile C4A script: {str(e)}") raise def is_match(self, url: str) -> bool: """Check if this config matches the given URL. Args: url: The URL to check against this config's matcher Returns: bool: True if this config should be used for the URL or if no matcher is set. """ if self.url_matcher is None: return True if callable(self.url_matcher): # Single function matcher return self.url_matcher(url) elif isinstance(self.url_matcher, str): # Single pattern string from fnmatch import fnmatch return fnmatch(url, self.url_matcher) elif isinstance(self.url_matcher, list): # List of mixed matchers if not self.url_matcher: # Empty list return False results = [] for matcher in self.url_matcher: if callable(matcher): results.append(matcher(url)) elif isinstance(matcher, str): from fnmatch import fnmatch results.append(fnmatch(url, matcher)) else: # Skip invalid matchers continue # Apply match mode logic if self.match_mode == MatchMode.OR: return any(results) if results else False else: # AND mode return all(results) if results else False return False def __getattr__(self, name): """Handle attribute access.""" if name in self._UNWANTED_PROPS: raise AttributeError(f"Getting '{name}' is deprecated. {self._UNWANTED_PROPS[name]}") raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{name}'") def __setattr__(self, name, value): """Handle attribute setting.""" # TODO: Planning to set properties dynamically based on the __init__ signature sig = inspect.signature(self.__init__) all_params = sig.parameters # Dictionary of parameter names and their details if name in self._UNWANTED_PROPS and value is not all_params[name].default: raise AttributeError(f"Setting '{name}' is deprecated. {self._UNWANTED_PROPS[name]}") super().__setattr__(name, value) @staticmethod def from_kwargs(kwargs: dict) -> "CrawlerRunConfig": # Auto-deserialize any dict values that use the {"type": ..., "params": ...} # serialization format (e.g. from JSON API requests or dump()/load() roundtrips). # This covers markdown_generator, extraction_strategy, content_filter, etc. kwargs = { k: from_serializable_dict(v) if isinstance(v, dict) and "type" in v else v for k, v in kwargs.items() } # Only pass keys present in kwargs so that __init__ defaults (and # set_defaults() overrides) are respected for missing keys. valid = inspect.signature(CrawlerRunConfig.__init__).parameters.keys() - {"self"} return CrawlerRunConfig(**{k: v for k, v in kwargs.items() if k in valid}) # Create a funciton returns dict of the object def dump(self) -> dict: # Serialize the object to a dictionary return to_serializable_dict(self) @staticmethod def load(data: dict, provenance: "Provenance" = None) -> "CrawlerRunConfig": # Deserialize the object from a dictionary if provenance is None: provenance = Provenance.TRUSTED config = from_serializable_dict(data, provenance) if isinstance(config, CrawlerRunConfig): return config if provenance == Provenance.UNTRUSTED and isinstance(config, dict): config = _enforce_untrusted("CrawlerRunConfig", config) return CrawlerRunConfig.from_kwargs(config) def to_dict(self): return { "word_count_threshold": self.word_count_threshold, "extraction_strategy": self.extraction_strategy, "chunking_strategy": self.chunking_strategy, "markdown_generator": self.markdown_generator, "only_text": self.only_text, "css_selector": self.css_selector, "target_elements": self.target_elements, "excluded_tags": self.excluded_tags, "excluded_selector": self.excluded_selector, "keep_data_attributes": self.keep_data_attributes, "keep_attrs": self.keep_attrs, "remove_forms": self.remove_forms, "prettiify": self.prettiify, "parser_type": self.parser_type, "scraping_strategy": self.scraping_strategy, "proxy_config": ( [p.to_dict() if hasattr(p, 'to_dict') else p for p in self.proxy_config] if isinstance(self.proxy_config, list) else (self.proxy_config.to_dict() if hasattr(self.proxy_config, 'to_dict') else self.proxy_config) ), "proxy_rotation_strategy": self.proxy_rotation_strategy, "proxy_session_id": self.proxy_session_id, "proxy_session_ttl": self.proxy_session_ttl, "proxy_session_auto_release": self.proxy_session_auto_release, "locale": self.locale, "timezone_id": self.timezone_id, "geolocation": self.geolocation, "fetch_ssl_certificate": self.fetch_ssl_certificate, "cache_mode": self.cache_mode, "session_id": self.session_id, "bypass_cache": self.bypass_cache, "disable_cache": self.disable_cache, "no_cache_read": self.no_cache_read, "no_cache_write": self.no_cache_write, "shared_data": self.shared_data, "wait_until": self.wait_until, "page_timeout": self.page_timeout, "wait_for": self.wait_for, "wait_for_timeout": self.wait_for_timeout, "wait_for_images": self.wait_for_images, "delay_before_return_html": self.delay_before_return_html, "mean_delay": self.mean_delay, "max_range": self.max_range, "semaphore_count": self.semaphore_count, "js_code": self.js_code, "js_code_before_wait": self.js_code_before_wait, "js_only": self.js_only, "ignore_body_visibility": self.ignore_body_visibility, "scan_full_page": self.scan_full_page, "scroll_delay": self.scroll_delay, "max_scroll_steps": self.max_scroll_steps, "process_iframes": self.process_iframes, "flatten_shadow_dom": self.flatten_shadow_dom, "remove_overlay_elements": self.remove_overlay_elements, "remove_consent_popups": self.remove_consent_popups, "simulate_user": self.simulate_user, "override_navigator": self.override_navigator, "magic": self.magic, "adjust_viewport_to_content": self.adjust_viewport_to_content, "screenshot": self.screenshot, "screenshot_wait_for": self.screenshot_wait_for, "screenshot_height_threshold": self.screenshot_height_threshold, "pdf": self.pdf, "capture_mhtml": self.capture_mhtml, "image_description_min_word_threshold": self.image_description_min_word_threshold, "image_score_threshold": self.image_score_threshold, "table_score_threshold": self.table_score_threshold, "table_extraction": self.table_extraction, "exclude_all_images": self.exclude_all_images, "exclude_external_images": self.exclude_external_images, "exclude_social_media_domains": self.exclude_social_media_domains, "exclude_external_links": self.exclude_external_links, "exclude_social_media_links": self.exclude_social_media_links, "exclude_domains": self.exclude_domains, "exclude_internal_links": self.exclude_internal_links, "score_links": self.score_links, "preserve_https_for_internal_links": self.preserve_https_for_internal_links, "verbose": self.verbose, "log_console": self.log_console, "capture_network_requests": self.capture_network_requests, "capture_console_messages": self.capture_console_messages, "method": self.method, "stream": self.stream, "prefetch": self.prefetch, "process_in_browser": self.process_in_browser, "check_robots_txt": self.check_robots_txt, "user_agent": self.user_agent, "user_agent_mode": self.user_agent_mode, "user_agent_generator_config": self.user_agent_generator_config, "deep_crawl_strategy": self.deep_crawl_strategy, "link_preview_config": self.link_preview_config.to_dict() if self.link_preview_config else None, "url": self.url, "url_matcher": self.url_matcher, "match_mode": self.match_mode, "experimental": self.experimental, "max_retries": self.max_retries, } def clone(self, **kwargs): """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: CrawlerRunConfig: A new instance with the specified updates Example: ```python # Create a new config with streaming enabled stream_config = config.clone(stream=True) # Create a new config with multiple updates new_config = config.clone( stream=True, cache_mode=CacheMode.BYPASS, verbose=True ) ``` """ config_dict = self.to_dict() config_dict.update(kwargs) return CrawlerRunConfig.from_kwargs(config_dict) class LLMConfig: def __init__( self, provider: str = DEFAULT_PROVIDER, api_token: Optional[str] = None, base_url: Optional[str] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, top_p: Optional[float] = None, frequency_penalty: Optional[float] = None, presence_penalty: Optional[float] = None, stop: Optional[List[str]] = None, n: Optional[int] = None, backoff_base_delay: Optional[int] = None, backoff_max_attempts: Optional[int] = None, backoff_exponential_factor: Optional[int] = None, provenance: "Provenance" = None, ): """Configuaration class for LLM provider and API token.""" if provenance is None: provenance = Provenance.TRUSTED # Defense in depth: untrusted callers can already not reach here (the # type gate forbids constructing LLMConfig from a request body), but if # they ever do, never resolve env vars or read provider keys from the # environment - that is the credential-exfil gadget. if provenance == Provenance.UNTRUSTED: if api_token and api_token.startswith("env:"): raise UntrustedConfigError( "LLMConfig.api_token may not reference an environment variable " "from an untrusted request" ) self.provider = provider self.api_token = api_token # never os.getenv self.base_url = base_url self.temperature = temperature self.max_tokens = max_tokens self.top_p = top_p self.frequency_penalty = frequency_penalty self.presence_penalty = presence_penalty self.stop = stop self.n = n self.backoff_base_delay = backoff_base_delay if backoff_base_delay is not None else 2 self.backoff_max_attempts = backoff_max_attempts if backoff_max_attempts is not None else 3 self.backoff_exponential_factor = backoff_exponential_factor if backoff_exponential_factor is not None else 2 return self.provider = provider if api_token and not api_token.startswith("env:"): self.api_token = api_token elif api_token and api_token.startswith("env:"): self.api_token = os.getenv(api_token[4:]) else: # Check if given provider starts with any of key in PROVIDER_MODELS_PREFIXES # If not, check if it is in PROVIDER_MODELS prefixes = PROVIDER_MODELS_PREFIXES.keys() if any(provider.startswith(prefix) for prefix in prefixes): selected_prefix = next( (prefix for prefix in prefixes if provider.startswith(prefix)), None, ) self.api_token = PROVIDER_MODELS_PREFIXES.get(selected_prefix) else: self.provider = DEFAULT_PROVIDER self.api_token = os.getenv(DEFAULT_PROVIDER_API_KEY) self.base_url = base_url self.temperature = temperature self.max_tokens = max_tokens self.top_p = top_p self.frequency_penalty = frequency_penalty self.presence_penalty = presence_penalty self.stop = stop self.n = n self.backoff_base_delay = backoff_base_delay if backoff_base_delay is not None else 2 self.backoff_max_attempts = backoff_max_attempts if backoff_max_attempts is not None else 3 self.backoff_exponential_factor = backoff_exponential_factor if backoff_exponential_factor is not None else 2 @staticmethod def from_kwargs(kwargs: dict) -> "LLMConfig": return LLMConfig( provider=kwargs.get("provider", DEFAULT_PROVIDER), api_token=kwargs.get("api_token"), base_url=kwargs.get("base_url"), temperature=kwargs.get("temperature"), max_tokens=kwargs.get("max_tokens"), top_p=kwargs.get("top_p"), frequency_penalty=kwargs.get("frequency_penalty"), presence_penalty=kwargs.get("presence_penalty"), stop=kwargs.get("stop"), n=kwargs.get("n"), backoff_base_delay=kwargs.get("backoff_base_delay"), backoff_max_attempts=kwargs.get("backoff_max_attempts"), backoff_exponential_factor=kwargs.get("backoff_exponential_factor") ) def to_dict(self): return { "provider": self.provider, "api_token": self.api_token, "base_url": self.base_url, "temperature": self.temperature, "max_tokens": self.max_tokens, "top_p": self.top_p, "frequency_penalty": self.frequency_penalty, "presence_penalty": self.presence_penalty, "stop": self.stop, "n": self.n, "backoff_base_delay": self.backoff_base_delay, "backoff_max_attempts": self.backoff_max_attempts, "backoff_exponential_factor": self.backoff_exponential_factor } def clone(self, **kwargs): """Create a copy of this configuration with updated values. Args: **kwargs: Key-value pairs of configuration options to update Returns: llm_config: A new instance with the specified updates """ config_dict = self.to_dict() config_dict.update(kwargs) return LLMConfig.from_kwargs(config_dict) class SeedingConfig: """ Configuration class for URL discovery and pre-validation via AsyncUrlSeeder. """ def __init__( self, source: str = "sitemap+cc", pattern: Optional[str] = "*", live_check: bool = False, extract_head: bool = False, max_urls: int = -1, concurrency: int = 1000, hits_per_sec: int = 5, force: bool = False, base_directory: Optional[str] = None, llm_config: Optional[LLMConfig] = None, verbose: Optional[bool] = None, query: Optional[str] = None, score_threshold: Optional[float] = None, scoring_method: str = "bm25", filter_nonsense_urls: bool = True, cache_ttl_hours: int = 24, validate_sitemap_lastmod: bool = True, ): """ Initialize URL seeding configuration. Args: source: Discovery source(s) to use. Options: "sitemap", "cc" (Common Crawl), or "sitemap+cc" (both). Default: "sitemap+cc" pattern: URL pattern to filter discovered URLs (e.g., "*example.com/blog/*"). Supports glob-style wildcards. Default: "*" (all URLs) live_check: Whether to perform HEAD requests to verify URL liveness. Default: False extract_head: Whether to fetch and parse section for metadata extraction. Required for BM25 relevance scoring. Default: False max_urls: Maximum number of URLs to discover. Use -1 for no limit. Default: -1 concurrency: Maximum concurrent requests for live checks/head extraction. Default: 1000 hits_per_sec: Rate limit in requests per second to avoid overwhelming servers. Default: 5 force: If True, bypasses the AsyncUrlSeeder's internal .jsonl cache and re-fetches URLs. Default: False base_directory: Base directory for UrlSeeder's cache files (.jsonl). If None, uses default ~/.crawl4ai/. Default: None llm_config: LLM configuration for future features (e.g., semantic scoring). Currently unused. Default: None verbose: Override crawler's general verbose setting for seeding operations. Default: None (inherits from crawler) query: Search query for BM25 relevance scoring (e.g., "python tutorials"). Requires extract_head=True. Default: None score_threshold: Minimum relevance score (0.0-1.0) to include URL. Only applies when query is provided. Default: None scoring_method: Scoring algorithm to use. Currently only "bm25" is supported. Future: "semantic". Default: "bm25" filter_nonsense_urls: Filter out utility URLs like robots.txt, sitemap.xml, ads.txt, favicon.ico, etc. Default: True cache_ttl_hours: Hours before sitemap cache expires. Set to 0 to disable TTL (only lastmod validation). Default: 24 validate_sitemap_lastmod: If True, compares sitemap's with cache timestamp and refetches if sitemap is newer. Default: True """ self.source = source self.pattern = pattern self.live_check = live_check self.extract_head = extract_head self.max_urls = max_urls self.concurrency = concurrency self.hits_per_sec = hits_per_sec self.force = force self.base_directory = base_directory self.llm_config = llm_config self.verbose = verbose self.query = query self.score_threshold = score_threshold self.scoring_method = scoring_method self.filter_nonsense_urls = filter_nonsense_urls self.cache_ttl_hours = cache_ttl_hours self.validate_sitemap_lastmod = validate_sitemap_lastmod # Add to_dict, from_kwargs, and clone methods for consistency def to_dict(self) -> Dict[str, Any]: return {k: v for k, v in self.__dict__.items() if k != 'llm_config' or v is not None} @staticmethod def from_kwargs(kwargs: Dict[str, Any]) -> 'SeedingConfig': return SeedingConfig(**kwargs) def clone(self, **kwargs: Any) -> 'SeedingConfig': config_dict = self.to_dict() config_dict.update(kwargs) return SeedingConfig.from_kwargs(config_dict) class DomainMapperConfig: """ Configuration for comprehensive domain URL discovery via DomainMapper. Discovers all URLs under a domain using multiple sources (sitemap, Common Crawl, Wayback Machine, Certificate Transparency, path probing, robots.txt mining, RSS/Atom feeds, homepage link extraction) without deep crawling. """ def __init__( self, source: str = "sitemap+cc+crt+probe", max_urls: int = -1, concurrency: int = 50, hits_per_sec: int = 10, force: bool = False, extract_head: bool = True, filter_nonsense_urls: bool = True, soft_404_detection: bool = True, query: Optional[str] = None, score_threshold: Optional[float] = None, scoring_method: str = "bm25", probe_paths: Optional[List[str]] = None, common_subdomains: Optional[List[str]] = None, include_subdomains: bool = True, source_timeout: float = 30.0, use_browser_for_homepage: bool = False, verbose: Optional[bool] = None, cache_ttl_hours: int = 24, base_directory: Optional[str] = None, dns_timeout: float = 3.0, http_timeout: float = 10.0, ): """ Args: source: Discovery sources joined by '+'. Valid: sitemap, cc, wayback, crt, probe, robots, feed, homepage. Default: "sitemap+cc+crt+probe" max_urls: Maximum URLs to return. -1 for unlimited. concurrency: Max concurrent requests across all hosts. hits_per_sec: Rate limit in requests/second. force: Bypass all caches. extract_head: Fetch and parse metadata for each URL. filter_nonsense_urls: Filter utility URLs (robots.txt, favicon, etc.). soft_404_detection: Fingerprint and filter soft-404 pages (SPAs). query: BM25 relevance query. Requires extract_head=True. score_threshold: Minimum relevance score (0.0-1.0) to include URL. scoring_method: Scoring algorithm. Currently only "bm25". probe_paths: Extra paths to probe on each host (added to defaults). common_subdomains: Extra subdomain prefixes to guess (added to defaults). include_subdomains: Discover and scan subdomains. When False, only scans the exact domain provided (skips crt.sh, DNS guessing, Wayback/CC host discovery). Default True. source_timeout: Max seconds per discovery source. Sources that exceed this are killed and skipped. Prevents slow sources from blocking fast results. Default 30.0. use_browser_for_homepage: Use Playwright for JS-rendered homepages. verbose: Override logger verbose setting. cache_ttl_hours: Hours before cached results expire. base_directory: Base directory for cache files. dns_timeout: Timeout in seconds for DNS resolution. http_timeout: Timeout in seconds for HTTP requests. """ self.source = source self.max_urls = max_urls self.concurrency = concurrency self.hits_per_sec = hits_per_sec self.force = force self.extract_head = extract_head self.filter_nonsense_urls = filter_nonsense_urls self.soft_404_detection = soft_404_detection self.query = query self.score_threshold = score_threshold self.scoring_method = scoring_method self.probe_paths = probe_paths self.common_subdomains = common_subdomains self.include_subdomains = include_subdomains self.source_timeout = source_timeout self.use_browser_for_homepage = use_browser_for_homepage self.verbose = verbose self.cache_ttl_hours = cache_ttl_hours self.base_directory = base_directory self.dns_timeout = dns_timeout self.http_timeout = http_timeout def to_dict(self) -> Dict[str, Any]: return {k: v for k, v in self.__dict__.items()} @staticmethod def from_kwargs(kwargs: Dict[str, Any]) -> 'DomainMapperConfig': return DomainMapperConfig(**kwargs) def clone(self, **kwargs: Any) -> 'DomainMapperConfig': config_dict = self.to_dict() config_dict.update(kwargs) return DomainMapperConfig.from_kwargs(config_dict)