import os import random import re from dataclasses import dataclass, field from typing import Optional, List, Dict, Any from urllib.parse import urlparse from dotenv import load_dotenv from .url_utils import normalize_url # Modern realistic user agents for rotation to evade bot-detection and WAF blocks DEFAULT_USER_AGENTS: List[str] = [ # Chrome on Windows (v133) "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36", # Chrome on macOS (v133) "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36", # Firefox on Windows (v134) "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:134.0) Gecko/20100101 Firefox/134.0", # Firefox on macOS (v134) "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:134.0) Gecko/20100101 Firefox/134.0", # Edge on Windows (v133) "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36 Edg/133.0.0.0", # Safari on macOS (v18.2) "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.2 Safari/605.1.15", ] # Supported European countries with active Tor exit relays EUROPEAN_COUNTRIES: Dict[str, Dict[str, str]] = { "de": {"name": "Germany", "flag": "🇩🇪"}, "nl": {"name": "Netherlands", "flag": "🇳🇱"}, "fr": {"name": "France", "flag": "🇫🇷"}, "ch": {"name": "Switzerland", "flag": "🇨🇭"}, "at": {"name": "Austria", "flag": "🇦🇹"}, "se": {"name": "Sweden", "flag": "🇸🇪"}, "no": {"name": "Norway", "flag": "🇳🇴"}, "fi": {"name": "Finland", "flag": "🇫🇮"}, "dk": {"name": "Denmark", "flag": "🇩🇰"}, "gb": {"name": "United Kingdom", "flag": "🇬🇧"}, "pl": {"name": "Poland", "flag": "🇵🇱"}, "it": {"name": "Italy", "flag": "🇮🇹"}, "es": {"name": "Spain", "flag": "🇪🇸"}, "be": {"name": "Belgium", "flag": "🇧🇪"}, "cz": {"name": "Czech Republic", "flag": "🇨🇿"}, "ro": {"name": "Romania", "flag": "🇷🇴"}, "ie": {"name": "Ireland", "flag": "🇮🇪"}, "pt": {"name": "Portugal", "flag": "🇵🇹"}, "hu": {"name": "Hungary", "flag": "🇭🇺"}, "is": {"name": "Iceland", "flag": "🇮🇸"}, "lu": {"name": "Luxembourg", "flag": "🇱🇺"}, "gr": {"name": "Greece", "flag": "🇬🇷"}, "bg": {"name": "Bulgaria", "flag": "🇧🇬"}, "ee": {"name": "Estonia", "flag": "🇪🇪"}, "lv": {"name": "Latvia", "flag": "🇱🇻"}, "lt": {"name": "Lithuania", "flag": "🇱🇹"}, "sk": {"name": "Slovakia", "flag": "🇸🇰"}, "si": {"name": "Slovenia", "flag": "🇸🇮"}, "hr": {"name": "Croatia", "flag": "🇭🇷"}, } @dataclass class CrawlerConfig: """Configuration settings for the crawler.""" start_url: str use_tor: bool = False tor_circuit_mode: str = "iteration" # "iteration" (new circuit per run) or "pages" (new circuit per N pages) tor_circuit_pages: int = 10 # Number of pages to crawl before renewing circuit if mode is "pages" tor_max_consecutive_timeouts: int = 5 # Number of consecutive ERR_TIMED_OUT failures before fallback circuit renewal tor_exit_nodes: List[str] = field(default_factory=list) # ISO country codes (empty = any global exit node) num_tabs: int = 5 num_iterations: int = 1 iteration_mode: str = "all" # "all" (crawl all discovered websites) or "visits" (cap visits per iteration) max_visits_per_iteration: int = 10 # Max website visits before starting a new iteration if mode is "visits" randomize_queue: bool = True # Randomly pop URLs from queue instead of strict FIFO order headless: bool = False accept_cookies: bool = True slow_mo: int = 100 timeout_ms: int = 35000 tor_socks_port: int = 9050 tor_control_port: int = 9051 user_agent: str = "random" # "random" (auto-rotate modern desktop UAs), "default" (native browser), or custom string @property def visits_per_iteration(self) -> int: """Alias for max_visits_per_iteration.""" return self.max_visits_per_iteration @property def start_domain(self) -> str: """Extracts the hostname/netloc from the start URL.""" return urlparse(self.start_url).netloc @property def tor_exit_nodes_formatted(self) -> Optional[str]: """Formats exit node country codes into Tor syntax (e.g. '{de},{fr}'), or None if empty.""" if not self.tor_exit_nodes: return None valid_codes = [c.strip().lower() for c in self.tor_exit_nodes if c.strip()] if not valid_codes: return None return ",".join(f"{{{c}}}" for c in valid_codes) def get_user_agent(self) -> Optional[str]: """ Resolves the user agent string based on the configuration setting: - "random" or empty: randomly selects a modern real-world desktop User-Agent. - "default": returns None to instruct Playwright to use its native browser User-Agent. - Custom string: returns the specified string. """ if not self.user_agent or self.user_agent.strip().lower() in ["random", "rotate"]: return random.choice(DEFAULT_USER_AGENTS) if self.user_agent.strip().lower() == "default": return None return self.user_agent.strip() @classmethod def from_env_or_prompt(cls) -> "CrawlerConfig": """ Loads configuration values from environment variables / .env file, falling back to interactive CLI prompts for any unset options. """ load_dotenv() # 1. Starting URL start_url = os.getenv("CRAWLER_START_URL") if not start_url: start_url = input("Enter the starting URL (e.g., https://example.com): ").strip() if not start_url.startswith("http://") and not start_url.startswith("https://"): start_url = "https://" + start_url start_domain = urlparse(start_url).netloc norm_start, _ = normalize_url(start_url, start_domain) if norm_start: start_url = norm_start # 2. Tor Proxy use_tor_env = os.getenv("CRAWLER_USE_TOR") if use_tor_env is not None: use_tor = use_tor_env.strip().lower() in ["true", "1", "y", "yes"] else: use_tor_input = input( "Use Tor proxy (starts/connects to headless Tor on 127.0.0.1:9050)? (y/n): " ).strip().lower() use_tor = (use_tor_input == "y") # 2.1 Tor Circuit Rotation Strategy & Exit Nodes (if Tor is enabled) tor_circuit_mode = "iteration" tor_circuit_pages = 10 tor_max_consecutive_timeouts = 5 tor_exit_nodes = [] if use_tor: mode_env = os.getenv("CRAWLER_TOR_CIRCUIT_MODE") pages_env = os.getenv("CRAWLER_TOR_CIRCUIT_PAGES") timeouts_env = os.getenv("CRAWLER_TOR_MAX_CONSECUTIVE_TIMEOUTS") nodes_env = os.getenv("CRAWLER_TOR_EXIT_NODES") if mode_env is not None: tor_circuit_mode = "pages" if mode_env.strip().lower() in ["pages", "page", "per_pages"] else "iteration" else: print("\nTor Circuit Renewal Options:") print(" [1] Renew circuit before every iteration (default)") print(" [2] Renew circuit after every N crawled pages") tor_mode_choice = input("Select Tor renewal mode (1/2, default 1): ").strip() tor_circuit_mode = "pages" if tor_mode_choice == "2" else "iteration" if tor_circuit_mode == "pages": if pages_env and pages_env.isdigit(): tor_circuit_pages = max(1, int(pages_env)) else: pages_input = input("How many crawled pages before renewing Tor circuit? (default 10): ").strip() tor_circuit_pages = int(pages_input) if pages_input and pages_input.isdigit() else 10 if timeouts_env and timeouts_env.isdigit(): tor_max_consecutive_timeouts = max(1, int(timeouts_env)) if nodes_env is not None: tor_exit_nodes = [c.strip().lower() for c in nodes_env.split(",") if c.strip()] else: nodes_input = input( "Select European Tor Exit Node countries (comma-separated, e.g. de,nl,fr,ch) [Enter for any]: " ).strip() if nodes_input: tor_exit_nodes = [c.strip().lower() for c in nodes_input.split(",") if c.strip()] # 3. Concurrent Tabs num_tabs_str = os.getenv("CRAWLER_NUM_TABS") if not num_tabs_str: num_tabs_str = input("How many concurrent tabs would you like to use? (default 5): ").strip() num_tabs = int(num_tabs_str) if num_tabs_str and num_tabs_str.isdigit() else 5 # 4. Iterations num_iterations_str = os.getenv("CRAWLER_NUM_ITERATIONS") if not num_iterations_str: num_iterations_str = input("How many times would you like to run the crawler? (default 1): ").strip() num_iterations = int(num_iterations_str) if num_iterations_str and num_iterations_str.isdigit() else 1 # 4.1 Iteration Scope & Website Visit Cap iter_mode_env = os.getenv("CRAWLER_ITERATION_MODE") max_visits_env = os.getenv("CRAWLER_MAX_VISITS_PER_ITERATION") or os.getenv("CRAWLER_VISITS_PER_ITERATION") if iter_mode_env is not None: iteration_mode = "visits" if iter_mode_env.strip().lower() in ["visits", "limit", "pages", "capped"] else "all" else: print("\nIteration Scope Options:") print(" [1] Crawl all reachable websites per iteration (default)") print(" [2] Limit website visits per iteration before starting a new iteration") iter_choice = input("Select iteration scope mode (1/2, default 1): ").strip() iteration_mode = "visits" if iter_choice == "2" else "all" max_visits_per_iteration = 10 if iteration_mode == "visits": if max_visits_env and max_visits_env.isdigit(): max_visits_per_iteration = max(1, int(max_visits_env)) else: visits_input = input("How many website visits before starting a new iteration? (default 10): ").strip() max_visits_per_iteration = int(visits_input) if visits_input and visits_input.isdigit() else 10 # 5. Headless Mode use_headless_env = os.getenv("CRAWLER_HEADLESS") if use_headless_env is not None: headless = use_headless_env.strip().lower() in ["true", "1", "y", "yes"] else: headless = input("Run browser in headless mode? (y/n, default n): ").strip().lower() == "y" # 6. Accept Cookies accept_cookies_env = os.getenv("CRAWLER_ACCEPT_COOKIES") if accept_cookies_env is not None: accept_cookies = accept_cookies_env.strip().lower() in ["true", "1", "y", "yes"] else: accept_cookies_input = input("Automatically accept cookies on pages? (y/n, default y): ").strip().lower() accept_cookies = (accept_cookies_input != "n") # 7. User Agent user_agent = os.getenv("CRAWLER_USER_AGENT", "random").strip() # 8. Randomize Queue randomize_queue_env = os.getenv("CRAWLER_RANDOMIZE_QUEUE") if randomize_queue_env is not None: randomize_queue = randomize_queue_env.strip().lower() in ["true", "1", "y", "yes"] else: randomize_queue = True return cls( start_url=start_url, use_tor=use_tor, tor_circuit_mode=tor_circuit_mode, tor_circuit_pages=tor_circuit_pages, tor_max_consecutive_timeouts=tor_max_consecutive_timeouts, tor_exit_nodes=tor_exit_nodes, num_tabs=num_tabs, num_iterations=num_iterations, iteration_mode=iteration_mode, max_visits_per_iteration=max_visits_per_iteration, randomize_queue=randomize_queue, headless=headless, accept_cookies=accept_cookies, user_agent=user_agent, ) def to_dict(self) -> dict: """Serializes the configuration to a dictionary.""" return { "start_url": self.start_url, "use_tor": self.use_tor, "tor_circuit_mode": self.tor_circuit_mode, "tor_circuit_pages": self.tor_circuit_pages, "tor_max_consecutive_timeouts": self.tor_max_consecutive_timeouts, "tor_exit_nodes": self.tor_exit_nodes, "num_tabs": self.num_tabs, "num_iterations": self.num_iterations, "iteration_mode": self.iteration_mode, "max_visits_per_iteration": self.max_visits_per_iteration, "randomize_queue": self.randomize_queue, "headless": self.headless, "accept_cookies": self.accept_cookies, "slow_mo": self.slow_mo, "timeout_ms": self.timeout_ms, "tor_socks_port": self.tor_socks_port, "tor_control_port": self.tor_control_port, "user_agent": self.user_agent, } @classmethod def from_dict(cls, data: dict) -> "CrawlerConfig": """Creates a CrawlerConfig from a dictionary with validation and defaults.""" start_url = str(data.get("start_url", "")).strip() if start_url and not start_url.startswith("http://") and not start_url.startswith("https://"): start_url = "https://" + start_url if start_url: start_domain = urlparse(start_url).netloc norm_start, _ = normalize_url(start_url, start_domain) if norm_start: start_url = norm_start circuit_mode = str(data.get("tor_circuit_mode", "iteration")).strip().lower() if circuit_mode not in ["iteration", "pages"]: circuit_mode = "iteration" try: circuit_pages = max(1, int(data.get("tor_circuit_pages", 10))) except (ValueError, TypeError): circuit_pages = 10 try: max_consecutive_timeouts = max(1, int(data.get("tor_max_consecutive_timeouts", 5))) except (ValueError, TypeError): max_consecutive_timeouts = 5 raw_exit_nodes = data.get("tor_exit_nodes", []) if isinstance(raw_exit_nodes, str): exit_nodes_list = [c.strip().lower() for c in raw_exit_nodes.split(",") if c.strip()] elif isinstance(raw_exit_nodes, list): exit_nodes_list = [str(c).strip().lower() for c in raw_exit_nodes if str(c).strip()] else: exit_nodes_list = [] iter_mode = str(data.get("iteration_mode", "all")).strip().lower() if iter_mode in ["visits", "limit", "pages", "capped"]: iteration_mode = "visits" else: iteration_mode = "all" raw_visits = data.get("max_visits_per_iteration") or data.get("visits_per_iteration") or 10 try: val = int(raw_visits) max_visits_per_iteration = val if val >= 1 else 10 except (ValueError, TypeError): max_visits_per_iteration = 10 return cls( start_url=start_url, use_tor=bool(data.get("use_tor", False)), tor_circuit_mode=circuit_mode, tor_circuit_pages=circuit_pages, tor_max_consecutive_timeouts=max_consecutive_timeouts, tor_exit_nodes=exit_nodes_list, num_tabs=int(data.get("num_tabs", 5)), num_iterations=int(data.get("num_iterations", 1)), iteration_mode=iteration_mode, max_visits_per_iteration=max_visits_per_iteration, randomize_queue=bool(data.get("randomize_queue", True)), headless=bool(data.get("headless", False)), accept_cookies=bool(data.get("accept_cookies", True)), slow_mo=int(data.get("slow_mo", 100)), timeout_ms=int(data.get("timeout_ms", 35000)), tor_socks_port=int(data.get("tor_socks_port", 9050)), tor_control_port=int(data.get("tor_control_port", 9051)), user_agent=str(data.get("user_agent", "random")).strip(), ) @classmethod def get_env_dict(cls, env_path: str = ".env") -> dict: """Reads configuration from .env file or environment variables.""" load_dotenv(dotenv_path=env_path, override=True) raw_mode = os.getenv("CRAWLER_TOR_CIRCUIT_MODE", "iteration").strip().lower() tor_circuit_mode = "pages" if raw_mode in ["pages", "page", "per_pages"] else "iteration" raw_pages = os.getenv("CRAWLER_TOR_CIRCUIT_PAGES", "10").strip() tor_circuit_pages = int(raw_pages) if raw_pages.isdigit() else 10 raw_timeouts = os.getenv("CRAWLER_TOR_MAX_CONSECUTIVE_TIMEOUTS", "5").strip() tor_max_consecutive_timeouts = int(raw_timeouts) if raw_timeouts.isdigit() else 5 raw_nodes = os.getenv("CRAWLER_TOR_EXIT_NODES", "") tor_exit_nodes = [c.strip().lower() for c in raw_nodes.split(",") if c.strip()] raw_iter_mode = os.getenv("CRAWLER_ITERATION_MODE", "all").strip().lower() iteration_mode = "visits" if raw_iter_mode in ["visits", "limit", "pages", "capped"] else "all" raw_max_visits = ( os.getenv("CRAWLER_MAX_VISITS_PER_ITERATION") or os.getenv("CRAWLER_VISITS_PER_ITERATION") or "10" ).strip() max_visits_per_iteration = int(raw_max_visits) if raw_max_visits.isdigit() else 10 return { "start_url": os.getenv("CRAWLER_START_URL", "https://example.com"), "use_tor": os.getenv("CRAWLER_USE_TOR", "false").strip().lower() in ["true", "1", "y", "yes"], "tor_circuit_mode": tor_circuit_mode, "tor_circuit_pages": max(1, tor_circuit_pages), "tor_max_consecutive_timeouts": max(1, tor_max_consecutive_timeouts), "tor_exit_nodes": tor_exit_nodes, "num_tabs": int(os.getenv("CRAWLER_NUM_TABS", "5")) if os.getenv("CRAWLER_NUM_TABS", "").isdigit() else 5, "num_iterations": int(os.getenv("CRAWLER_NUM_ITERATIONS", "1")) if os.getenv("CRAWLER_NUM_ITERATIONS", "").isdigit() else 1, "iteration_mode": iteration_mode, "max_visits_per_iteration": max(1, max_visits_per_iteration), "randomize_queue": os.getenv("CRAWLER_RANDOMIZE_QUEUE", "true").strip().lower() not in ["false", "0", "n", "no"], "headless": os.getenv("CRAWLER_HEADLESS", "false").strip().lower() in ["true", "1", "y", "yes"], "accept_cookies": os.getenv("CRAWLER_ACCEPT_COOKIES", "true").strip().lower() not in ["false", "0", "n", "no"], "slow_mo": int(os.getenv("CRAWLER_SLOW_MO", "100")) if os.getenv("CRAWLER_SLOW_MO", "").isdigit() else 100, "timeout_ms": int(os.getenv("CRAWLER_TIMEOUT_MS", "35000")) if os.getenv("CRAWLER_TIMEOUT_MS", "").isdigit() else 35000, "tor_socks_port": int(os.getenv("CRAWLER_TOR_SOCKS_PORT", "9050")) if os.getenv("CRAWLER_TOR_SOCKS_PORT", "").isdigit() else 9050, "tor_control_port": int(os.getenv("CRAWLER_TOR_CONTROL_PORT", "9051")) if os.getenv("CRAWLER_TOR_CONTROL_PORT", "").isdigit() else 9051, "user_agent": os.getenv("CRAWLER_USER_AGENT", "random").strip(), } @classmethod def save_env_file(cls, data: dict, env_path: str = ".env") -> None: """Writes configuration key-values to a .env file.""" raw_exit = data.get("tor_exit_nodes", []) if isinstance(raw_exit, list): exit_str = ",".join(raw_exit) else: exit_str = str(raw_exit) lines = [ f"CRAWLER_START_URL={data.get('start_url', '')}", f"CRAWLER_USE_TOR={'true' if data.get('use_tor') else 'false'}", f"CRAWLER_TOR_CIRCUIT_MODE={data.get('tor_circuit_mode', 'iteration')}", f"CRAWLER_TOR_CIRCUIT_PAGES={data.get('tor_circuit_pages', 10)}", f"CRAWLER_TOR_MAX_CONSECUTIVE_TIMEOUTS={data.get('tor_max_consecutive_timeouts', 5)}", f"CRAWLER_TOR_EXIT_NODES={exit_str}", f"CRAWLER_NUM_TABS={data.get('num_tabs', 5)}", f"CRAWLER_NUM_ITERATIONS={data.get('num_iterations', 1)}", f"CRAWLER_ITERATION_MODE={data.get('iteration_mode', 'all')}", f"CRAWLER_MAX_VISITS_PER_ITERATION={data.get('max_visits_per_iteration', 10)}", f"CRAWLER_RANDOMIZE_QUEUE={'true' if data.get('randomize_queue', True) else 'false'}", f"CRAWLER_HEADLESS={'true' if data.get('headless') else 'false'}", f"CRAWLER_ACCEPT_COOKIES={'true' if data.get('accept_cookies') else 'false'}", f"CRAWLER_SLOW_MO={data.get('slow_mo', 100)}", f"CRAWLER_TIMEOUT_MS={data.get('timeout_ms', 35000)}", f"CRAWLER_TOR_SOCKS_PORT={data.get('tor_socks_port', 9050)}", f"CRAWLER_TOR_CONTROL_PORT={data.get('tor_control_port', 9051)}", f"CRAWLER_USER_AGENT={data.get('user_agent', 'random')}", "", ] with open(env_path, "w", encoding="utf-8") as f: f.write("\n".join(lines))