Building a Site Explorer Agent: Autonomous Website Architecture Mapping

# Hybrid explorer with intelligent wait strategy
from playwright.async_api import async_playwright
import asyncio
from collections import deque
import re

class SiteExplorerAgent:
    def __init__(self, base_url, max_depth=5):
        self.base_url = base_url
        self.max_depth = max_depth
        self.visited = set()
        self.queue = deque([(base_url, 0)])  # (url, depth)
        self.site_map = {}

    async def explore_with_wait(self, page, url):
        await page.goto(url, wait_until='networkidle')
        # Wait for DOM stability
        await page.wait_for_function(
            "(() => { const last = performance.getEntriesByType('navigation')[0]; "
            "return Date.now() - last.loadEventEnd > 1000; })()"
        )

        # Extract all navigation-relevant links
        links = await page.eval_on_selector_all('a', """
            elements => elements.map(el => ({
                href: el.href,
                text: el.textContent.trim(),
                is_nav: el.closest('nav') !== null,
                is_footer: el.closest('footer') !== null
            }))
        """)

        return [l['href'] for l in links if self._is_valid_link(l)]

    async def run(self):
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context(
                user_agent='Mozilla/5.0 (compatible; SiteExplorer/1.0)'
            )
            page = await context.new_page()

            while self.queue:
                url, depth = self.queue.popleft()
                if url in self.visited or depth >= self.max_depth:
                    continue

                self.visited.add(url)
                try:
                    links = await self.explore_with_wait(page, url)
                    self.site_map[url] = links

                    for link in links:
                        if link not in self.visited:
                            self.queue.append((link, depth + 1))
                except Exception as e:
                    print(f"Failed to explore {url}: {e}")

            await browser.close()
        return self.site_map

import re
from urllib.parse import urlparse, parse_qs
import hashlib

class URLNormalizer:
    def __init__(self):
        # Parameters that typically don't change page content
        self.template_params = {'sort', 'order', 'direction', 'view', 'display'}
        # Parameters that typically indicate unique content
        self.content_params = {'id', 'slug', 'product', 'article', 'page'}

    def normalize(self, url):
        parsed = urlparse(url)
        params = parse_qs(parsed.query)

        # Check if URL has content-defining parameters
        has_content_param = any(p in self.content_params for p in params)

        if has_content_param:
            # Keep all parameters for unique content
            return url

        # Remove template parameters for canonical URL
        filtered_params = {
            k: v for k, v in params.items()
            if k not in self.template_params
        }

        # Reconstruct URL
        normalized = parsed._replace(
            query='&'.join(f"{k}={v[0]}" for k, v in filtered_params.items())
        ).geturl()

        return normalized

    def get_fingerprint(self, url):
        """Create fingerprint for duplicate detection"""
        normalized = self.normalize(url)
        # Hash the path + key parameters
        path_hash = hashlib.md5(normalized.encode()).hexdigest()[:12]
        return path_hash

import time
import asyncio
from collections import deque
from urllib.parse import urlparse

class AdaptiveRateLimiter:
    def __init__(self, initial_rate=2.0, max_rate=10.0):
        self.rate = initial_rate  # requests per second
        self.max_rate = max_rate
        self.min_rate = 0.5
        self.tokens = initial_rate
        self.last_update = time.time()
        self.response_times = deque(maxlen=20)
        self.domain_limits = {}  # per-domain rate limits

    async def acquire(self, url):
        domain = urlparse(url).netloc
        domain_limit = self.domain_limits.get(domain, self.rate)

        while True:
            now = time.time()
            elapsed = now - self.last_update
            self.tokens = min(
                self.max_rate,
                self.tokens + elapsed * domain_limit
            )
            self.last_update = now

            if self.tokens >= 1:
                self.tokens -= 1
                return

            # Wait for next token
            wait_time = (1 - self.tokens) / domain_limit
            await asyncio.sleep(wait_time)

    def record_response(self, url, response_time, status_code):
        domain = urlparse(url).netloc

        # Adjust rate based on response time
        self.response_times.append(response_time)
        avg_response = sum(self.response_times) / len(self.response_times)

        if avg_response > 2.0:  # Server struggling
            self.domain_limits[domain] = max(
                self.min_rate,
                self.domain_limits.get(domain, self.rate) * 0.8
            )
        elif avg_response < 0.5 and status_code == 200:
            # Server fast, can increase rate
            self.domain_limits[domain] = min(
                self.max_rate,
                self.domain_limits.get(domain, self.rate) * 1.1
            )

        if status_code == 429:
            # Immediate backoff
            self.domain_limits[domain] = self.min_rate