1023 lines
40 KiB
Python
1023 lines
40 KiB
Python
import asyncio
|
|
import inspect
|
|
import random
|
|
import time
|
|
from typing import Dict, List, Set, Tuple, Optional, Callable, Any
|
|
from playwright.async_api import async_playwright
|
|
|
|
from .config import CrawlerConfig
|
|
from .cookies import auto_accept_cookies
|
|
from .tor import TorManager
|
|
from .url_utils import normalize_url
|
|
|
|
|
|
async def _dispatch_event(callback: Optional[Callable[[Dict[str, Any]], Any]], event: Dict[str, Any]):
|
|
"""Safely dispatches an event dictionary to an async or sync callback."""
|
|
if callback is None:
|
|
return
|
|
try:
|
|
if inspect.iscoroutinefunction(callback):
|
|
await callback(event)
|
|
else:
|
|
callback(event)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def crawl_level(
|
|
page,
|
|
url: str,
|
|
source_url: Optional[str],
|
|
visited: Set[str],
|
|
errors: List[Tuple[str, Optional[str], str]],
|
|
start_domain: str,
|
|
accept_cookies_default: bool = True,
|
|
stats: Optional[Dict[str, int]] = None,
|
|
timeout_ms: int = 35000,
|
|
event_callback: Optional[Callable[[Dict[str, Any]], Any]] = None,
|
|
worker_name: str = "Worker",
|
|
page_result_callback: Optional[Callable[[str, Optional[str], bool, Optional[str]], Any]] = None,
|
|
) -> Set[str]:
|
|
"""Navigates to a URL and returns a set of valid HTTP/HTTPS links found on that page."""
|
|
log_msg = f"[{worker_name}] Browsing: {url}"
|
|
print(log_msg)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": log_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
start_time = time.time()
|
|
try:
|
|
# Go to the URL and wait for the page to load
|
|
response = await page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms)
|
|
visited.add(url)
|
|
|
|
if response and response.status >= 400:
|
|
retry_after = response.headers.get("retry-after")
|
|
status_desc = ""
|
|
if response.status == 503:
|
|
status_desc = " (Service Temporarily Unavailable / Rate Limited)"
|
|
elif response.status == 429:
|
|
status_desc = " (Too Many Requests / Rate Limited)"
|
|
elif response.status == 403:
|
|
status_desc = " (Forbidden / Access Denied)"
|
|
|
|
retry_info = f" [Retry-After: {retry_after}s]" if retry_after else ""
|
|
err_text = f"HTTP {response.status}{status_desc}{retry_info}"
|
|
errors.append((url, source_url, err_text))
|
|
err_log = f" [{worker_name}] [!] HTTP {response.status}{status_desc}{retry_info} returned for {url}"
|
|
print(err_log)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": err_log,
|
|
"level": "error",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(event_callback, {
|
|
"type": "error",
|
|
"url": url,
|
|
"source_url": source_url,
|
|
"error": err_text,
|
|
"timestamp": time.time(),
|
|
})
|
|
if page_result_callback:
|
|
try:
|
|
res = page_result_callback(url, source_url, False, err_text)
|
|
if inspect.iscoroutine(res):
|
|
await res
|
|
except Exception:
|
|
pass
|
|
return set()
|
|
|
|
except Exception as e:
|
|
duration = time.time() - start_time
|
|
visited.add(url)
|
|
error_msg = str(e).split("\n")[0]
|
|
errors.append((url, source_url, error_msg))
|
|
err_log = f" [{worker_name}] [!] Failed to load {url} after {duration:.2f}s: {error_msg}"
|
|
print(err_log)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": err_log,
|
|
"level": "error",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(event_callback, {
|
|
"type": "error",
|
|
"url": url,
|
|
"source_url": source_url,
|
|
"error": error_msg,
|
|
"timestamp": time.time(),
|
|
})
|
|
if page_result_callback:
|
|
try:
|
|
res = page_result_callback(url, source_url, False, error_msg)
|
|
if inspect.iscoroutine(res):
|
|
await res
|
|
except Exception:
|
|
pass
|
|
return set()
|
|
|
|
duration = time.time() - start_time
|
|
ok_log = f" [{worker_name}] [OK] Loaded in {duration:.2f} seconds"
|
|
print(ok_log)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": ok_log,
|
|
"level": "success",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
# Automatically accept cookies to ensure tracking and scripts are activated
|
|
if accept_cookies_default:
|
|
accepted = await auto_accept_cookies(page)
|
|
if accepted and stats is not None:
|
|
stats["cookies_accepted"] += 1
|
|
cookie_log = f" [{worker_name}] [COOKIE] Accepted cookie banner on {url}"
|
|
print(cookie_log)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": cookie_log,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
if page_result_callback:
|
|
try:
|
|
res = page_result_callback(url, source_url, True, None)
|
|
if inspect.iscoroutine(res):
|
|
await res
|
|
except Exception:
|
|
pass
|
|
|
|
# Extract all 'href' attributes from 'a' tags
|
|
try:
|
|
links = await page.eval_on_selector_all("a[href]", "elements => elements.map(el => el.href)")
|
|
except Exception as e:
|
|
extract_err = f" [{worker_name}] [!] Failed to extract links from {url}: {e}"
|
|
print(extract_err)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": extract_err,
|
|
"level": "warn",
|
|
"timestamp": time.time(),
|
|
})
|
|
return set()
|
|
|
|
# Filter and normalize valid HTTP/HTTPS links
|
|
valid_links = set()
|
|
for link in links:
|
|
norm_url, _ = normalize_url(link, start_domain)
|
|
if norm_url:
|
|
valid_links.add(norm_url)
|
|
|
|
return valid_links
|
|
|
|
|
|
class RandomQueue(asyncio.Queue):
|
|
"""An asyncio Queue that pops items uniformly at random instead of strict FIFO order."""
|
|
|
|
def _init(self, maxsize):
|
|
self._queue = []
|
|
|
|
def _put(self, item):
|
|
self._queue.append(item)
|
|
|
|
def _get(self):
|
|
idx = random.randrange(len(self._queue))
|
|
self._queue[idx], self._queue[-1] = self._queue[-1], self._queue[idx]
|
|
return self._queue.pop()
|
|
|
|
|
|
class IterationVisitTracker:
|
|
"""Coordinates and limits website visits within a single crawl iteration."""
|
|
|
|
def __init__(self, max_visits: Optional[int]):
|
|
self.max_visits = max_visits
|
|
self.started_visits = 0
|
|
self.completed_visits = 0
|
|
self._lock = asyncio.Lock()
|
|
self.limit_reached_event = asyncio.Event()
|
|
|
|
async def acquire_visit_slot(self) -> bool:
|
|
"""
|
|
Attempts to reserve a visit slot.
|
|
Returns True if reservation was successful (or visits unlimited),
|
|
False if the max visit limit for this iteration has been reached.
|
|
"""
|
|
if self.max_visits is None or self.max_visits <= 0:
|
|
return True
|
|
async with self._lock:
|
|
if self.started_visits >= self.max_visits:
|
|
return False
|
|
self.started_visits += 1
|
|
return True
|
|
|
|
async def record_visit_completed(self) -> None:
|
|
"""Records that a visit attempt has finished, signalling limit_reached_event if target reached."""
|
|
if self.max_visits is None or self.max_visits <= 0:
|
|
return
|
|
async with self._lock:
|
|
self.completed_visits += 1
|
|
if self.completed_visits >= self.max_visits:
|
|
self.limit_reached_event.set()
|
|
|
|
|
|
async def worker(
|
|
name: str,
|
|
context,
|
|
queue: asyncio.Queue,
|
|
visited: Set[str],
|
|
enqueued: Set[str],
|
|
errors: List[Tuple[str, Optional[str], str]],
|
|
start_domain: str,
|
|
stats: Dict[str, int],
|
|
stop_event: asyncio.Event,
|
|
iteration_stop_event: Optional[asyncio.Event] = None,
|
|
visit_tracker: Optional[IterationVisitTracker] = None,
|
|
accept_cookies_default: bool = True,
|
|
timeout_ms: int = 35000,
|
|
event_callback: Optional[Callable[[Dict[str, Any]], Any]] = None,
|
|
page_done_callback: Optional[Callable[[], Any]] = None,
|
|
page_result_callback: Optional[Callable[[str, Optional[str], bool, Optional[str]], Any]] = None,
|
|
prior_visited: int = 0,
|
|
prior_errors: int = 0,
|
|
):
|
|
"""Worker task that pulls URLs from the queue and crawls them in its own dedicated tab."""
|
|
page = None
|
|
try:
|
|
if stop_event.is_set() or (iteration_stop_event and iteration_stop_event.is_set()):
|
|
return
|
|
|
|
page = await context.new_page()
|
|
|
|
# Automatically close any popups or new tabs spawned by links or scripts
|
|
page.on("popup", lambda popup: asyncio.create_task(popup.close()))
|
|
|
|
while not stop_event.is_set() and (iteration_stop_event is None or not iteration_stop_event.is_set()):
|
|
# Wait for a URL from the queue with a short timeout to check stop conditions periodically
|
|
try:
|
|
item = await asyncio.wait_for(queue.get(), timeout=1.0)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
|
|
current_url, source_url, is_external = item
|
|
|
|
try:
|
|
if stop_event.is_set() or (iteration_stop_event and iteration_stop_event.is_set()):
|
|
break
|
|
|
|
if current_url in visited:
|
|
continue
|
|
|
|
if visit_tracker:
|
|
can_visit = await visit_tracker.acquire_visit_slot()
|
|
if not can_visit:
|
|
if iteration_stop_event:
|
|
iteration_stop_event.set()
|
|
break
|
|
|
|
prefix = f"[EXTERNAL|{name}]" if is_external else f"[INTERNAL|{name}]"
|
|
queue_log = f"\n{prefix} (Queue approx: {queue.qsize()})"
|
|
print(queue_log)
|
|
|
|
try:
|
|
found_links = await crawl_level(
|
|
page=page,
|
|
url=current_url,
|
|
source_url=source_url,
|
|
visited=visited,
|
|
errors=errors,
|
|
start_domain=start_domain,
|
|
accept_cookies_default=accept_cookies_default,
|
|
stats=stats,
|
|
timeout_ms=timeout_ms,
|
|
event_callback=event_callback,
|
|
worker_name=name,
|
|
page_result_callback=page_result_callback,
|
|
)
|
|
finally:
|
|
if visit_tracker:
|
|
await visit_tracker.record_visit_completed()
|
|
|
|
if is_external:
|
|
stats["external"] += 1
|
|
else:
|
|
stats["internal"] += 1
|
|
|
|
# Emit updated stats
|
|
await _dispatch_event(event_callback, {
|
|
"type": "stats",
|
|
"stats": stats,
|
|
"queue_size": queue.qsize(),
|
|
"visited_count": len(visited),
|
|
"error_count": len(errors),
|
|
"total_visited_all": prior_visited + len(visited),
|
|
"total_errors_all": prior_errors + len(errors),
|
|
})
|
|
|
|
# Trigger legacy page crawled callback if page_result_callback was not supplied
|
|
if page_done_callback and not page_result_callback and not stop_event.is_set():
|
|
try:
|
|
res = page_done_callback()
|
|
if inspect.iscoroutine(res):
|
|
await res
|
|
except Exception:
|
|
pass
|
|
|
|
# If it's an internal link, add its discovered links to the queue
|
|
if not is_external and not stop_event.is_set() and (iteration_stop_event is None or not iteration_stop_event.is_set()):
|
|
new_links = []
|
|
for link in found_links:
|
|
norm_link, link_is_internal = normalize_url(link, start_domain)
|
|
if norm_link and norm_link not in enqueued and norm_link not in visited:
|
|
new_links.append((norm_link, current_url, not link_is_internal))
|
|
enqueued.add(norm_link)
|
|
|
|
# Shuffle new links to maintain a degree of randomness in crawling order
|
|
if new_links:
|
|
random.shuffle(new_links)
|
|
for n_item in new_links:
|
|
queue.put_nowait(n_item)
|
|
elif is_external:
|
|
ext_log = f" [{name}] -> Reached external level 1. Not following further links."
|
|
print(ext_log)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": ext_log,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
except Exception as e:
|
|
err_str = f" [!] Unexpected error in worker {name} processing {current_url}: {e}"
|
|
print(err_str)
|
|
await _dispatch_event(event_callback, {
|
|
"type": "log",
|
|
"message": err_str,
|
|
"level": "error",
|
|
"timestamp": time.time(),
|
|
})
|
|
finally:
|
|
queue.task_done()
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
finally:
|
|
if page:
|
|
try:
|
|
await page.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def fetch_tor_identity(context, timeout_ms: int = 10000) -> Dict[str, Any]:
|
|
"""Fetches active exit IP and geolocation details using the browser's Tor proxy context."""
|
|
# 1. Try ipapi.co
|
|
try:
|
|
res = await context.request.get("https://ipapi.co/json/", timeout=timeout_ms)
|
|
if res.ok:
|
|
data = await res.json()
|
|
if "ip" in data:
|
|
return {
|
|
"ip": data.get("ip", "Unknown"),
|
|
"city": data.get("city", "Unknown"),
|
|
"region": data.get("region", "Unknown"),
|
|
"country": data.get("country_name", "Unknown"),
|
|
"country_code": data.get("country_code", ""),
|
|
"org": data.get("org", "Unknown"),
|
|
"is_tor": True,
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Try ip-api.com fallback
|
|
try:
|
|
res = await context.request.get("http://ip-api.com/json/", timeout=timeout_ms)
|
|
if res.ok:
|
|
data = await res.json()
|
|
if data.get("status") == "success":
|
|
return {
|
|
"ip": data.get("query", "Unknown"),
|
|
"city": data.get("city", "Unknown"),
|
|
"region": data.get("regionName", "Unknown"),
|
|
"country": data.get("country", "Unknown"),
|
|
"country_code": data.get("countryCode", ""),
|
|
"org": data.get("isp", "Unknown"),
|
|
"is_tor": True,
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Try official check.torproject.org fallback
|
|
try:
|
|
res = await context.request.get("https://check.torproject.org/api/ip", timeout=timeout_ms)
|
|
if res.ok:
|
|
data = await res.json()
|
|
return {
|
|
"ip": data.get("IP", "Unknown"),
|
|
"city": "Unknown",
|
|
"region": "Unknown",
|
|
"country": "Unknown",
|
|
"country_code": "",
|
|
"org": "Tor Exit Relay",
|
|
"is_tor": bool(data.get("IsTor", True)),
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"ip": "Unknown",
|
|
"city": "Unknown",
|
|
"region": "Unknown",
|
|
"country": "Unknown",
|
|
"country_code": "",
|
|
"org": "Tor Network",
|
|
"is_tor": True,
|
|
}
|
|
|
|
|
|
class Crawler:
|
|
"""Core crawler engine orchestrating browser instances, tabs, and queue scheduling."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: CrawlerConfig,
|
|
event_callback: Optional[Callable[[Dict[str, Any]], Any]] = None,
|
|
):
|
|
self.config = config
|
|
self.event_callback = event_callback
|
|
self.tor_manager = TorManager(
|
|
socks_port=config.tor_socks_port,
|
|
control_port=config.tor_control_port,
|
|
exit_nodes=config.tor_exit_nodes_formatted,
|
|
) if config.use_tor else None
|
|
self._stop_event = asyncio.Event()
|
|
self._is_running = False
|
|
self._circuit_lock = asyncio.Lock()
|
|
self._pages_in_current_circuit = 0
|
|
self._consecutive_timeouts = 0
|
|
self._total_circuit_rotations = 0
|
|
self._total_visited_all_iterations = 0
|
|
self._total_errors_all_iterations = 0
|
|
|
|
def stop(self) -> None:
|
|
"""Signals the crawler and active workers to stop immediately."""
|
|
self._stop_event.set()
|
|
|
|
@property
|
|
def is_running(self) -> bool:
|
|
return self._is_running
|
|
|
|
@property
|
|
def total_visited_all_iterations(self) -> int:
|
|
return self._total_visited_all_iterations
|
|
|
|
@property
|
|
def total_errors_all_iterations(self) -> int:
|
|
return self._total_errors_all_iterations
|
|
|
|
async def _rotate_tor_circuit(self, context=None, reason: str = "") -> None:
|
|
"""Safely triggers NEWNYM via TorManager and queries fresh exit node identity."""
|
|
if not self.tor_manager or not self.config.use_tor:
|
|
return
|
|
|
|
rot_msg = f"[Tor] Renewing Tor circuit (NEWNYM)... (Reason: {reason})"
|
|
print(f"\n{rot_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": rot_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
await asyncio.to_thread(self.tor_manager.renew_circuit)
|
|
self._total_circuit_rotations += 1
|
|
self._pages_in_current_circuit = 0
|
|
self._consecutive_timeouts = 0
|
|
|
|
if context:
|
|
try:
|
|
tor_id = await fetch_tor_identity(context)
|
|
tor_loc_str = f"{tor_id['city']}, {tor_id['country']}" if tor_id['city'] != "Unknown" else tor_id['country']
|
|
tor_id_msg = f"[Tor] Circuit #{self._total_circuit_rotations} active! Exit Node IP: {tor_id['ip']} | Location: {tor_loc_str} | ISP: {tor_id['org']}"
|
|
print(tor_id_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": tor_id_msg,
|
|
"level": "success",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "tor_identity",
|
|
"data": {
|
|
**tor_id,
|
|
"rotation_count": self._total_circuit_rotations,
|
|
"exit_nodes": self.config.tor_exit_nodes,
|
|
},
|
|
})
|
|
except Exception as e:
|
|
print(f" [!] Failed to query Tor identity after rotation: {e}")
|
|
|
|
async def _on_page_result(
|
|
self,
|
|
context,
|
|
url: str,
|
|
source_url: Optional[str],
|
|
success: bool,
|
|
error_msg: Optional[str] = None,
|
|
) -> None:
|
|
"""Handles Tor circuit rotation on N crawled pages as well as consecutive timeout fallback."""
|
|
if not self.config.use_tor:
|
|
return
|
|
|
|
async with self._circuit_lock:
|
|
if success:
|
|
self._consecutive_timeouts = 0
|
|
if self.config.tor_circuit_mode == "pages":
|
|
self._pages_in_current_circuit += 1
|
|
if self._pages_in_current_circuit >= self.config.tor_circuit_pages:
|
|
await self._rotate_tor_circuit(
|
|
context=context,
|
|
reason=f"Reached {self.config.tor_circuit_pages} crawled pages threshold",
|
|
)
|
|
else:
|
|
# Check if this error is net::ERR_TIMED_OUT, connection timeout, or rate-limit / WAF block (503, 429, 403)
|
|
is_timeout = False
|
|
is_rate_limited = False
|
|
if error_msg:
|
|
err_lower = error_msg.lower()
|
|
if (
|
|
"err_timed_out" in err_lower
|
|
or "timed out" in err_lower
|
|
or "timed_out" in err_lower
|
|
or "timeout" in err_lower
|
|
):
|
|
is_timeout = True
|
|
elif (
|
|
"http 503" in err_lower
|
|
or "http 429" in err_lower
|
|
or "http 403" in err_lower
|
|
):
|
|
is_rate_limited = True
|
|
|
|
if is_timeout or is_rate_limited:
|
|
self._consecutive_timeouts += 1
|
|
err_type = "Rate limit / WAF block" if is_rate_limited else "Timeout"
|
|
timeout_log = (
|
|
f" [Tor] Consecutive {err_type.lower()} ({self._consecutive_timeouts}/"
|
|
f"{self.config.tor_max_consecutive_timeouts}) on {url}: {error_msg}"
|
|
)
|
|
print(timeout_log)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": timeout_log,
|
|
"level": "warn",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
if self._consecutive_timeouts >= self.config.tor_max_consecutive_timeouts:
|
|
fallback_reason = (
|
|
f"Fallback: {self.config.tor_max_consecutive_timeouts} consecutive pages failed "
|
|
f"({err_type.lower()}: {error_msg})"
|
|
)
|
|
fallback_log = (
|
|
f" [Tor] Fallback triggered! {self._consecutive_timeouts} consecutive pages failed "
|
|
f"with {err_type.lower()}. Acquiring new Tor circuit..."
|
|
)
|
|
print(f"\n{fallback_log}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": fallback_log,
|
|
"level": "warn",
|
|
"timestamp": time.time(),
|
|
})
|
|
self._consecutive_timeouts = 0
|
|
await self._rotate_tor_circuit(
|
|
context=context,
|
|
reason=fallback_reason,
|
|
)
|
|
else:
|
|
# Other non-transient errors (e.g. 404) reset consecutive failure counter
|
|
self._consecutive_timeouts = 0
|
|
|
|
async def _on_page_crawled(self, context) -> None:
|
|
"""Checks whether Tor circuit renewal is due in 'pages' mode (legacy compatibility)."""
|
|
await self._on_page_result(context, "", None, True, None)
|
|
|
|
async def run_iteration(self, playwright_instance, iteration: int) -> Dict[str, any]:
|
|
"""Runs a single crawl iteration."""
|
|
if self._stop_event.is_set():
|
|
return {
|
|
"duration": 0,
|
|
"stats": {},
|
|
"errors": [],
|
|
"visited_count": 0,
|
|
"total_visited_all": self._total_visited_all_iterations,
|
|
"total_errors_all": self._total_errors_all_iterations,
|
|
}
|
|
|
|
# Handle Tor circuit rotation at iteration boundary
|
|
if self.config.use_tor and self.tor_manager:
|
|
if self.config.tor_circuit_mode == "iteration":
|
|
await self._rotate_tor_circuit(
|
|
context=None,
|
|
reason=f"Starting Iteration {iteration}",
|
|
)
|
|
elif self.config.tor_circuit_mode == "pages" and self._total_circuit_rotations == 0:
|
|
await self._rotate_tor_circuit(
|
|
context=None,
|
|
reason="Initial Crawl Setup",
|
|
)
|
|
|
|
iteration_start_time = time.time()
|
|
|
|
launch_args = {
|
|
"headless": self.config.headless,
|
|
"slow_mo": self.config.slow_mo,
|
|
"args": [
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--no-sandbox",
|
|
"--disable-dev-shm-usage",
|
|
],
|
|
}
|
|
|
|
if self.config.use_tor:
|
|
tor_msg = f"[INFO] Routing traffic through Tor (socks5://127.0.0.1:{self.config.tor_socks_port})..."
|
|
print(f"\n{tor_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": tor_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
launch_args["proxy"] = {"server": f"socks5://127.0.0.1:{self.config.tor_socks_port}"}
|
|
launch_args["args"].append("--host-resolver-rules=MAP * ~NOTFOUND , EXCLUDE 127.0.0.1")
|
|
|
|
chosen_ua = self.config.get_user_agent()
|
|
context_kwargs = {
|
|
"viewport": {"width": 1920, "height": 1080},
|
|
"locale": "de-DE",
|
|
"timezone_id": "Europe/Berlin",
|
|
}
|
|
if chosen_ua:
|
|
context_kwargs["user_agent"] = chosen_ua
|
|
|
|
browser = await playwright_instance.chromium.launch(**launch_args)
|
|
context = await browser.new_context(**context_kwargs)
|
|
|
|
ua_msg = f"[INFO] Active Browser User-Agent: {chosen_ua or 'Native Browser Default'}"
|
|
print(ua_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": ua_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
# If Tor is enabled, discover exit IP and region for the current circuit
|
|
if self.config.use_tor:
|
|
tor_id = await fetch_tor_identity(context)
|
|
tor_loc_str = f"{tor_id['city']}, {tor_id['country']}" if tor_id['city'] != "Unknown" else tor_id['country']
|
|
tor_id_msg = f"[Tor] Exit Node IP: {tor_id['ip']} | Location: {tor_loc_str} | ISP: {tor_id['org']}"
|
|
print(tor_id_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": tor_id_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "tor_identity",
|
|
"data": {
|
|
**tor_id,
|
|
"rotation_count": self._total_circuit_rotations,
|
|
"exit_nodes": self.config.tor_exit_nodes,
|
|
},
|
|
})
|
|
|
|
visited: Set[str] = set()
|
|
enqueued: Set[str] = {self.config.start_url}
|
|
errors: List[Tuple[str, Optional[str], str]] = []
|
|
stats: Dict[str, int] = {"internal": 0, "external": 0, "cookies_accepted": 0}
|
|
|
|
prior_visited = self._total_visited_all_iterations
|
|
prior_errors = self._total_errors_all_iterations
|
|
|
|
queue: asyncio.Queue = (
|
|
RandomQueue() if self.config.randomize_queue else asyncio.Queue()
|
|
)
|
|
queue.put_nowait((self.config.start_url, None, False))
|
|
|
|
max_visits = (
|
|
self.config.max_visits_per_iteration
|
|
if self.config.iteration_mode == "visits" and self.config.max_visits_per_iteration > 0
|
|
else None
|
|
)
|
|
visit_tracker = IterationVisitTracker(max_visits=max_visits) if max_visits else None
|
|
iteration_stop_event = asyncio.Event()
|
|
|
|
start_domain = self.config.start_domain
|
|
start_msg = f"=== Starting Crawl on {start_domain} with {self.config.num_tabs} concurrent workers ==="
|
|
print(f"\n{start_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": start_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
if max_visits:
|
|
mode_msg = f"[INFO] Iteration visit mode: Limit to {max_visits} website visits before starting next iteration."
|
|
else:
|
|
mode_msg = "[INFO] Iteration visit mode: Crawling all reachable website links."
|
|
print(mode_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": mode_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
queue_msg = (
|
|
"[INFO] Crawl queue strategy: RANDOM (URLs are picked randomly from discovered links)."
|
|
if self.config.randomize_queue
|
|
else "[INFO] Crawl queue strategy: FIFO (Standard breadth-first search order)."
|
|
)
|
|
print(queue_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": queue_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
if self.config.accept_cookies:
|
|
cookie_enabled_msg = "[INFO] Automatic cookie acceptance is ENABLED."
|
|
print(cookie_enabled_msg)
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": cookie_enabled_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
# Emit initial stats
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "stats",
|
|
"stats": stats,
|
|
"queue_size": queue.qsize(),
|
|
"visited_count": len(visited),
|
|
"error_count": len(errors),
|
|
"total_visited_all": prior_visited + len(visited),
|
|
"total_errors_all": prior_errors + len(errors),
|
|
})
|
|
|
|
# Create worker tasks
|
|
workers = []
|
|
for i in range(self.config.num_tabs):
|
|
task = asyncio.create_task(
|
|
worker(
|
|
name=f"Tab-{i+1}",
|
|
context=context,
|
|
queue=queue,
|
|
visited=visited,
|
|
enqueued=enqueued,
|
|
errors=errors,
|
|
start_domain=start_domain,
|
|
stats=stats,
|
|
stop_event=self._stop_event,
|
|
iteration_stop_event=iteration_stop_event,
|
|
visit_tracker=visit_tracker,
|
|
accept_cookies_default=self.config.accept_cookies,
|
|
timeout_ms=self.config.timeout_ms,
|
|
event_callback=self.event_callback,
|
|
page_result_callback=lambda u, s, ok, err: self._on_page_result(context, u, s, ok, err),
|
|
prior_visited=prior_visited,
|
|
prior_errors=prior_errors,
|
|
)
|
|
)
|
|
workers.append(task)
|
|
|
|
# Wait until queue is fully processed or stop signal / limit is reached
|
|
join_task = asyncio.create_task(queue.join())
|
|
stop_wait_task = asyncio.create_task(self._stop_event.wait())
|
|
wait_tasks = [join_task, stop_wait_task]
|
|
|
|
limit_wait_task = None
|
|
if visit_tracker and visit_tracker.max_visits:
|
|
limit_wait_task = asyncio.create_task(visit_tracker.limit_reached_event.wait())
|
|
wait_tasks.append(limit_wait_task)
|
|
|
|
await asyncio.wait(
|
|
wait_tasks,
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
|
|
if not join_task.done():
|
|
join_task.cancel()
|
|
if not stop_wait_task.done():
|
|
stop_wait_task.cancel()
|
|
if limit_wait_task and not limit_wait_task.done():
|
|
limit_wait_task.cancel()
|
|
|
|
# Signal workers in this iteration to stop
|
|
iteration_stop_event.set()
|
|
|
|
# Cancel worker tasks
|
|
for task in workers:
|
|
task.cancel()
|
|
|
|
# Wait for workers to cleanly exit
|
|
await asyncio.gather(*workers, return_exceptions=True)
|
|
|
|
iteration_duration = time.time() - iteration_start_time
|
|
|
|
if visit_tracker and visit_tracker.limit_reached_event.is_set():
|
|
limit_summary = (
|
|
f"[INFO] Reached configured website visit limit "
|
|
f"({visit_tracker.completed_visits}/{max_visits} visits) for iteration {iteration}."
|
|
)
|
|
print(f"\n{limit_summary}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": limit_summary,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
fin_msg = f"Finished iteration {iteration}! Execution time: {iteration_duration:.2f} seconds."
|
|
print(f"\n{fin_msg}")
|
|
print(f"Total internal pages visited: {stats['internal']}")
|
|
print(f"Total external pages visited: {stats['external']}")
|
|
if self.config.accept_cookies:
|
|
print(f"Total cookie banners accepted: {stats['cookies_accepted']}")
|
|
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": fin_msg,
|
|
"level": "success",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
if errors:
|
|
err_summary = f"=== Websites with Errors ({len(errors)}) ==="
|
|
print(f"\n{err_summary}")
|
|
for err_url, src_url, err_msg in errors:
|
|
src_display = src_url if src_url else "User Input"
|
|
print(f"- {err_url}\n Error: {err_msg}\n Found on: {src_display}")
|
|
|
|
self._total_visited_all_iterations = prior_visited + len(visited)
|
|
self._total_errors_all_iterations = prior_errors + len(errors)
|
|
|
|
await browser.close()
|
|
return {
|
|
"duration": iteration_duration,
|
|
"stats": stats,
|
|
"errors": errors,
|
|
"visited_count": len(visited),
|
|
"total_visited_all": self._total_visited_all_iterations,
|
|
"total_errors_all": self._total_errors_all_iterations,
|
|
}
|
|
|
|
async def run(self) -> None:
|
|
"""Main execution loop running across all configured iterations."""
|
|
self._is_running = True
|
|
self._stop_event.clear()
|
|
self._total_visited_all_iterations = 0
|
|
self._total_errors_all_iterations = 0
|
|
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "status",
|
|
"status": "running",
|
|
"message": "Crawl job started",
|
|
"total_visited_all": 0,
|
|
"total_errors_all": 0,
|
|
})
|
|
|
|
if self.tor_manager:
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
def _on_tor_log(msg: str):
|
|
if self.event_callback:
|
|
asyncio.run_coroutine_threadsafe(
|
|
_dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
}),
|
|
loop
|
|
)
|
|
self.tor_manager.log_callback = _on_tor_log
|
|
|
|
# Launch or connect to Tor in background thread without blocking event loop
|
|
await asyncio.to_thread(self.tor_manager.start)
|
|
|
|
if self.config.tor_exit_nodes:
|
|
countries_display = ", ".join(self.config.tor_exit_nodes).upper()
|
|
filter_msg = f"[Tor] Exit node country filter active: {countries_display} ({len(self.config.tor_exit_nodes)} selected)"
|
|
else:
|
|
filter_msg = "[Tor] Exit node country filter: Global / Any (No restrictions)"
|
|
print(f"[INFO] {filter_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": filter_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
except Exception as e:
|
|
err_msg = f"[ERROR] Could not initialize Tor: {e}"
|
|
print(f"\n{err_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": err_msg,
|
|
"level": "error",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "status",
|
|
"status": "error",
|
|
"message": str(e),
|
|
})
|
|
self._is_running = False
|
|
return
|
|
|
|
try:
|
|
global_start_time = time.time()
|
|
|
|
for iteration in range(1, self.config.num_iterations + 1):
|
|
if self._stop_event.is_set():
|
|
break
|
|
|
|
if self.config.num_iterations > 1:
|
|
iter_msg = f"=== Starting Iteration {iteration}/{self.config.num_iterations} ==="
|
|
print(f"\n{'='*40}\n{iter_msg}\n{'='*40}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": iter_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "iteration",
|
|
"iteration": iteration,
|
|
"total": self.config.num_iterations,
|
|
"iteration_mode": self.config.iteration_mode,
|
|
"max_visits_per_iteration": self.config.max_visits_per_iteration if self.config.iteration_mode == "visits" else None,
|
|
"total_visited_all": self._total_visited_all_iterations,
|
|
"total_errors_all": self._total_errors_all_iterations,
|
|
})
|
|
|
|
async with async_playwright() as p:
|
|
await self.run_iteration(p, iteration)
|
|
|
|
total_duration = time.time() - global_start_time
|
|
done_msg = (
|
|
f"Finished all {self.config.num_iterations} iterations! "
|
|
f"Total execution time: {total_duration:.2f} seconds. "
|
|
f"Total websites visited: {self._total_visited_all_iterations}, "
|
|
f"total errors: {self._total_errors_all_iterations}."
|
|
)
|
|
if self._stop_event.is_set():
|
|
done_msg = (
|
|
f"Crawl stopped by user after {total_duration:.2f} seconds. "
|
|
f"Total websites visited: {self._total_visited_all_iterations}, "
|
|
f"total errors: {self._total_errors_all_iterations}."
|
|
)
|
|
|
|
print(f"\n{done_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": done_msg,
|
|
"level": "info",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "status",
|
|
"status": "stopped" if self._stop_event.is_set() else "finished",
|
|
"message": done_msg,
|
|
"total_duration": total_duration,
|
|
"total_visited_all": self._total_visited_all_iterations,
|
|
"total_errors_all": self._total_errors_all_iterations,
|
|
})
|
|
except Exception as e:
|
|
err_msg = f"Fatal crawler error: {e}"
|
|
print(f"\n[FATAL] {err_msg}")
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "log",
|
|
"message": err_msg,
|
|
"level": "error",
|
|
"timestamp": time.time(),
|
|
})
|
|
await _dispatch_event(self.event_callback, {
|
|
"type": "status",
|
|
"status": "error",
|
|
"message": str(e),
|
|
})
|
|
finally:
|
|
self._is_running = False
|
|
if self.tor_manager:
|
|
self.tor_manager.stop()
|
|
|