#!/usr/bin/env python3
"""
neopig - Neo Python Image Grabber

Full domain crawler that:
1. Crawls entire domain (no depth limit)
2. Extracts all images/videos via multiple methods
3. Uses HEAD checks for extensionless URLs
4. Stores in vault (MD5 dedupe)
5. Queues for Qwen 3 VL analysis
6. Indexes metadata in SQLite

Full-domain async media crawler with deduplication
https://russell.ballestrini.net/python-image-grabber-pig-py/

Usage:
    python neopig.py https://example.com "rick and morty" --mode images
    python neopig.py https://example.com --mode all --depth -1  # Full domain slurp
"""

import argparse
import asyncio
import atexit
import hashlib
import json
import logging
import multiprocessing
import os
import signal
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any, Optional, Set, Tuple

from miniuri import Uri

from async_web_fetcher import (
    AsyncWebFetcher,
    CrawlMode,
    MediaItem,
    extract_media_from_html,
    get_media_type_from_extension,
    get_media_type_from_mime,
)
from filevault import AsyncVault, hash_to_path
from database import Database, SCORE_SCREENSHOT, SCORE_OG_IMAGE, SCORE_THUMBNAIL, SCORE_FULL_RES
from screenshot import ScreenshotCapture, ScreenshotConfig
from domain_vault import VaultManager, DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault, extract_media_urls
from repo import detect_vcs, clone_repo_async, pull_repo_async, get_repo_path, walk_files, get_commit_hash, get_file_language, is_binary_file
from tqdm import tqdm

logger = logging.getLogger(__name__)

# Track child processes for cleanup on exit
_CHILD_PROCESSES: List = []

def _cleanup_children():
    """Kill all tracked child processes."""
    for proc in _CHILD_PROCESSES:
        if proc and proc.poll() is None:
            try:
                os.kill(proc.pid, signal.SIGTERM)
                proc.wait(timeout=2)
            except:
                try:
                    os.kill(proc.pid, signal.SIGKILL)
                except:
                    pass

# atexit runs even when asyncio overrides signal handlers
atexit.register(_cleanup_children)

# Live media event queue - push here after disk save, SSE reads from here
# Format: {'md5_hash': str, 'media_type': str, 'file_size': int, 'alt_text': str, ...}
LIVE_MEDIA_QUEUE: asyncio.Queue = None  # Initialized lazily

def get_live_queue() -> asyncio.Queue:
    """Get or create the live media queue."""
    global LIVE_MEDIA_QUEUE
    if LIVE_MEDIA_QUEUE is None:
        LIVE_MEDIA_QUEUE = asyncio.Queue(maxsize=1000)
    return LIVE_MEDIA_QUEUE

def emit_live_media(media_info: Dict[str, Any]):
    """Emit media to live feed (non-blocking)."""
    try:
        queue = get_live_queue()
        queue.put_nowait(media_info)
    except asyncio.QueueFull:
        pass  # Drop if queue is full (live feed will catch up from DB)

# Per-job log handlers
LOGS_PATH = Path("data/logs")
JOB_LOG_HANDLERS: Dict[int, logging.FileHandler] = {}

def start_job_logging(job_id: int) -> None:
    """Start capturing logs for a crawl job to file."""
    LOGS_PATH.mkdir(parents=True, exist_ok=True)
    log_file = LOGS_PATH / f"{job_id}.log"
    handler = logging.FileHandler(log_file, mode='w', encoding='utf-8')
    handler.setLevel(logging.INFO)
    handler.setFormatter(logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s', datefmt='%H:%M:%S'))
    logging.getLogger().addHandler(handler)
    JOB_LOG_HANDLERS[job_id] = handler

def stop_job_logging(job_id: int) -> None:
    """Stop capturing logs for a crawl job."""
    handler = JOB_LOG_HANDLERS.pop(job_id, None)
    if handler:
        handler.close()
        logging.getLogger().removeHandler(handler)


class AppendOnlyStateLog:
    """Append-only state log for fast crawl state tracking.

    Format (one record per line):
        P <url>              # page seen
        M <md5_hash> <url>   # media downloaded
        S <url>              # screenshot taken
        D <domain>           # skip domain
        X <key> <json>       # stats checkpoint

    Benefits:
        - No JSON parsing on write (just append)
        - No locking needed for single writer
        - Fast resume by scanning lines
        - Works with tail -f for monitoring
    """

    def __init__(self, path: Path):
        self.path = path
        self._file = None

    def open(self):
        """Open log file for appending."""
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._file = open(self.path, 'a', encoding='utf-8', buffering=1)  # line buffered

    def close(self):
        """Close log file."""
        if self._file:
            self._file.close()
            self._file = None

    def __enter__(self):
        self.open()
        return self

    def __exit__(self, *args):
        self.close()

    def page(self, url: str):
        """Record page as seen."""
        if self._file:
            self._file.write(f"P {url}\n")

    def media(self, md5_hash: str, url: str):
        """Record media as downloaded."""
        if self._file:
            self._file.write(f"M {md5_hash} {url}\n")

    def screenshot(self, url: str):
        """Record screenshot as taken."""
        if self._file:
            self._file.write(f"S {url}\n")

    def skip_domain(self, domain: str):
        """Record domain to skip."""
        if self._file:
            self._file.write(f"D {domain}\n")

    def stats(self, stats_dict: Dict[str, Any]):
        """Record stats checkpoint."""
        if self._file:
            self._file.write(f"X stats {json.dumps(stats_dict)}\n")

    def load(self) -> Dict[str, Any]:
        """Load state from log file.

        Returns dict with:
            - seen_pages: set of URLs
            - seen_media: dict of url -> md5_hash
            - seen_screenshots: set of URLs
            - skip_domains: set of domains
            - stats: last stats checkpoint (or empty dict)
        """
        result = {
            'seen_pages': set(),
            'seen_media': {},
            'seen_screenshots': set(),
            'skip_domains': set(),
            'stats': {},
        }

        if not self.path.exists():
            return result

        with open(self.path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.rstrip('\n')
                if not line:
                    continue

                parts = line.split(' ', 2)
                if len(parts) < 2:
                    continue

                record_type = parts[0]

                if record_type == 'P':
                    result['seen_pages'].add(parts[1])
                elif record_type == 'M' and len(parts) >= 3:
                    result['seen_media'][parts[2]] = parts[1]  # url -> hash
                elif record_type == 'S':
                    result['seen_screenshots'].add(parts[1])
                elif record_type == 'D':
                    result['skip_domains'].add(parts[1])
                elif record_type == 'X' and len(parts) >= 3:
                    try:
                        result['stats'] = json.loads(parts[2])
                    except json.JSONDecodeError:
                        pass

        return result


def get_state_log_path(domain: str) -> Path:
    """Get append-only state log path for a domain."""
    safe_domain = domain.replace('://', '-').replace('/', '-').replace('.', '-')
    return Path(f"data/{safe_domain}.log")


def get_state_file_path(domain: str) -> Path:
    """Get unified state file path for a domain.

    Args:
        domain: Domain name (e.g., 'example.com')

    Returns:
        Path like data/{domain}.state
    """
    safe_domain = domain.replace('://', '-').replace('/', '-').replace('.', '-')
    return Path(f"data/{safe_domain}.state")


def rotate_state_file(state_path: Path, preserve_keys: List[str] = None) -> Optional[Path]:
    """Rotate state file with optional selective preservation.

    Args:
        state_path: Path to state file
        preserve_keys: If provided, rotate then copy back these keys from rotated file.
                      If None, just rotate (full fresh start).

    Returns:
        Path to rotated file, or None if no rotation needed.
    """
    if not state_path.exists():
        return None

    # Find next available rotation number
    i = 1
    while Path(f"{state_path}.{i}").exists():
        i += 1
    rotated = Path(f"{state_path}.{i}")
    state_path.rename(rotated)
    logger.info(f"Rotated state file to {rotated}")

    # If preserve_keys specified, copy back those keys from rotated file
    if preserve_keys:
        try:
            import json
            old_state = json.loads(rotated.read_text())
            new_state = {k: v for k, v in old_state.items() if k in preserve_keys}
            if new_state:
                state_path.write_text(json.dumps(new_state, indent=2))
                logger.info(f"Preserved keys: {list(new_state.keys())}")
        except Exception as e:
            logger.warning(f"Could not preserve state keys: {e}")

    return rotated


class TqdmLoggingHandler(logging.Handler):
    """Logging handler that writes through tqdm to avoid progress bar corruption."""

    def emit(self, record):
        try:
            msg = self.format(record)
            tqdm.write(msg)
        except Exception:
            self.handleError(record)


def setup_logging(level=logging.INFO):
    """Setup logging to work with tqdm progress bars."""
    handler = TqdmLoggingHandler()
    handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(level)


class NeoPig:
    """
    Neo Python Image Grabber - async media crawler with deduplication.
    """

    def __init__(
        self,
        db_path: str = "neopig.db",
        vault_path: str = "vault",
        user_agent: str = "neopig/1.0 (ethical image crawler)",
        screenshot_config: ScreenshotConfig = None,
        fast_mode: bool = False,
        trim_wrapper: bool = False,
    ):
        self.db = Database(db_path)
        self.vault = AsyncVault(vault_path)
        self.trim_wrapper = trim_wrapper
        # Triple filevault system: html_vault/, media_vault/, and linkpeek_vault/
        self.domain_vaults = VaultManager(
            html_vault_base=f"{vault_path}/html_vault",
            media_vault_base=f"{vault_path}/media_vault",
            linkpeek_vault_base=f"{vault_path}/linkpeek_vault",
            media_base_url='/media',
            linkpeek_base_url='/linkpeek',
        )
        # Fast mode: no crawl delay, short timeouts (for sites without robots.txt)
        self.fast_mode = fast_mode
        crawl_delay = 0.0 if fast_mode else 2.0
        self.fetcher = AsyncWebFetcher(user_agent=user_agent, default_crawl_delay=crawl_delay, fast_mode=fast_mode)
        self.screenshot = ScreenshotCapture(screenshot_config or ScreenshotConfig())
        self.screenshot_config = screenshot_config or ScreenshotConfig()
        self.vault_path = vault_path

        # Track stats
        self.stats = {
            'pages_crawled': 0,
            'pages_pending': 0,  # Queue size for ETA calculation
            'pages_changed': 0,  # Pages with content changes (for git commit)
            'media_found': 0,
            'media_downloaded': 0,
            'media_new': 0,  # New media (for git commit)
            'duplicates_skipped': 0,
            'content_exists': 0,  # Same content (MD5) already in vault
            'screenshots_taken': 0,
            'errors': 0,
            'bytes_downloaded': 0,  # Total bytes fetched from network
            'bytes_stored': 0,  # Unique bytes stored in vault
            'crawl_started': 0,  # Unix timestamp for rate calculation
        }

        # Track seen media URLs -> md5_hash (to add page context without re-download)
        self.seen_media: Dict[str, str] = {}
        # Track screenshotted pages to avoid duplicates
        self.seen_screenshots: Set[str] = set()
        # Track crawled page URLs for resume support
        self.seen_pages: Set[str] = set()
        # Track per-domain stats for vault commits
        self._domain_stats: Dict[str, Dict[str, int]] = {}  # domain -> {pages_changed, media_new, screenshots_new}

        # Progress bar
        self.pbar: Optional[tqdm] = None

        # Resume support - state files go in data/
        self._state_dir = Path("data")
        self._state_dir.mkdir(exist_ok=True)
        self._state_save_interval = 10
        self._items_since_save = 0

        # Append-only state log (faster than JSON)
        self._state_log: Optional[AppendOnlyStateLog] = None

    def _get_state_file(self, target_url: str) -> Path:
        """Get unified state file path for domain."""
        parsed = Uri(target_url)
        domain = parsed.hostname.lower()
        return get_state_file_path(domain)

    def _get_state_log(self, target_url: str) -> AppendOnlyStateLog:
        """Get append-only state log for domain."""
        parsed = Uri(target_url)
        domain = parsed.hostname.lower()
        return AppendOnlyStateLog(get_state_log_path(domain))

    def _save_state(self, target_url: str):
        """Save crawl state for resume."""
        self._items_since_save += 1
        if self._items_since_save < self._state_save_interval:
            return

        self._items_since_save = 0
        state_file = self._get_state_file(target_url)

        # Load existing state to preserve other keys (e.g., backfill state)
        try:
            full_state = json.loads(state_file.read_text()) if state_file.exists() else {}
        except Exception:
            full_state = {}

        # Update crawl section
        full_state['crawl'] = {
            'target_url': target_url,
            'seen_media': list(self.seen_media),
            'seen_screenshots': list(self.seen_screenshots),
            'seen_pages': list(self.seen_pages),
            'skip_domains': list(self.fetcher.skip_domains),
            'stats': self.stats,
            'timestamp': datetime.now(timezone.utc).isoformat(),
        }

        try:
            self._state_dir.mkdir(parents=True, exist_ok=True)
            with open(state_file, 'w') as f:
                json.dump(full_state, f, indent=2)
        except Exception as e:
            logger.debug(f"Failed to save state: {e}")

    def _get_hydra_state_file(self, domain: str) -> Path:
        """Get path to hydra state file for a domain (never rotated)."""
        safe_domain = domain.replace('/', '_').replace(':', '_')
        return self._state_dir / f"hydra-{safe_domain}.json"

    def _load_hydra_state(self, domain: str) -> Dict[str, Any]:
        """Load hydra state - feeds and URLs seen (never cleared)."""
        state_file = self._get_hydra_state_file(domain)
        if not state_file.exists():
            return {'seen_urls': {}, 'feeds': {}}
        try:
            return json.loads(state_file.read_text())
        except Exception as e:
            logger.warning(f"Failed to load hydra state: {e}")
            return {'seen_urls': {}, 'feeds': {}}

    def _save_hydra_state(self, domain: str, state: Dict[str, Any]):
        """Save hydra state (persists across crawls, never rotated)."""
        state_file = self._get_hydra_state_file(domain)
        try:
            self._state_dir.mkdir(parents=True, exist_ok=True)
            state['last_updated'] = datetime.now(timezone.utc).isoformat()
            with open(state_file, 'w') as f:
                json.dump(state, f, indent=2)
        except Exception as e:
            logger.warning(f"Failed to save hydra state: {e}")

    def _get_known_feeds(self, domain: str) -> List[str]:
        """Get list of known feed URLs for a domain."""
        state = self._load_hydra_state(domain)
        return list(state.get('feeds', {}).keys())

    def _add_hydra_feeds(self, domain: str, feed_urls: List[str]):
        """Add newly discovered feed URLs to hydra state."""
        if not feed_urls:
            return
        state = self._load_hydra_state(domain)
        feeds = state.get('feeds', {})
        now = datetime.now(timezone.utc).isoformat()
        new_count = 0
        for url in feed_urls:
            if url not in feeds:
                feeds[url] = {'discovered': now}
                new_count += 1
        if new_count:
            state['feeds'] = feeds
            self._save_hydra_state(domain, state)
            logger.info(f"Hydra: Added {new_count} new feeds (total: {len(feeds)})")

    def _filter_new_hydra_urls(self, domain: str, feed_items: List) -> List:
        """Filter feed items to only return NEW URLs not seen before.

        Also updates hydra state with newly seen URLs.
        Returns list of new FeedItem objects.
        """
        state = self._load_hydra_state(domain)
        seen_urls = state.get('seen_urls', {})

        new_items = []
        now = datetime.now(timezone.utc).isoformat()

        for item in feed_items:
            url = item.url
            if url not in seen_urls:
                new_items.append(item)
                seen_urls[url] = {
                    'first_seen': now,
                    'title': item.title,
                    'published': item.published,
                }

        if new_items:
            state['seen_urls'] = seen_urls
            self._save_hydra_state(domain, state)
            logger.info(f"Hydra: {len(new_items)} NEW URLs (of {len(feed_items)} total)")
        else:
            logger.info(f"Hydra: No new URLs found (all {len(feed_items)} already seen)")

        return new_items

    async def _fetch_hydra_urls(self, target_uri: str, discover_new: bool = False) -> Optional[List[str]]:
        """Fetch URLs from known feeds (self-healing) and optionally discover new feeds.

        Args:
            target_uri: Target URI being crawled
            discover_new: If True, also discover new feeds from target page

        Returns:
            List of NEW URLs to inject into crawl, or None if no feeds known
        """
        domain = Uri(target_uri).hostname
        known_feeds = self._get_known_feeds(domain)

        # Discover new feeds if requested (--hydra mode)
        if discover_new:
            logger.info("=== HYDRA MODE: Discovering feeds/sitemaps ===")
            # fetch_feeds will discover from common locations
            feed_items = await self.fetcher.fetch_feeds(target_uri)
            if feed_items:
                # Extract discovered feed URLs and persist them
                # The fetcher tracks which feeds it found
                discovered_feeds = getattr(self.fetcher, '_last_discovered_feeds', [])
                if discovered_feeds:
                    self._add_hydra_feeds(domain, discovered_feeds)
                    known_feeds = self._get_known_feeds(domain)  # Refresh

        # If we have known feeds, fetch them for new URLs
        if known_feeds:
            logger.info(f"Hydra: Checking {len(known_feeds)} known feeds for new content")
            feed_items = await self.fetcher.fetch_feeds(
                target_uri,
                feed_urls=known_feeds,  # Pass known feeds directly
            )
            if feed_items:
                new_items = self._filter_new_hydra_urls(domain, feed_items)
                if new_items:
                    return [item.url for item in new_items]

        return None

    def _load_state(self, target_url: str) -> bool:
        """Load saved crawl state. Returns True if state was loaded.

        Tries append-only log first (faster), falls back to legacy JSON.
        Note: seen_media and seen_screenshots are NOT loaded from state file -
        they come from the database (source of truth for successful downloads).
        Only seen_pages (for fast mode) and stats are loaded from state.
        """
        # Try append-only log first (preferred)
        state_log = self._get_state_log(target_url)
        if state_log.path.exists():
            state = state_log.load()
            if self.fast_mode:
                self.seen_pages = state['seen_pages']
                self.fetcher.skip_domains = state['skip_domains']
            saved_stats = state.get('stats', {})
            for key in self.stats:
                if key in saved_stats:
                    self.stats[key] = saved_stats[key]
            if self.fast_mode:
                logger.info(f"Fast resume from log: skipping {len(self.seen_pages)} pages")
            else:
                logger.info(f"Resume from log: loading stats")
            return True

        # Fall back to legacy JSON state file
        state_file = self._get_state_file(target_url)
        if not state_file.exists():
            return False

        try:
            full_state = json.loads(state_file.read_text())
            state = full_state.get('crawl', {})
            if not state:
                return False

            # In fast mode, skip already-crawled pages for speed
            if self.fast_mode:
                self.seen_pages = set(state.get('seen_pages', []))
            saved_stats = state.get('stats', {})
            for key in self.stats:
                if key in saved_stats:
                    self.stats[key] = saved_stats[key]
            if self.fast_mode:
                logger.info(f"Fast resume: skipping {len(self.seen_pages)} pages (media/screenshots from DB)")
            else:
                logger.info(f"Resume: loading stats from state (media/screenshots from DB)")
            return True
        except Exception as e:
            logger.warning(f"Could not load state: {e}")
            return False

    def _clear_state(self, target_url: str, preserve_keys: List[str] = None):
        """Rotate state file for fresh start.

        Args:
            preserve_keys: Keys to preserve (e.g., ['crawl'] for backfill-only fresh)
        """
        try:
            state_file = self._get_state_file(target_url)
            rotate_state_file(state_file, preserve_keys=preserve_keys)
        except Exception:
            pass

    async def purge_job(self, job_id: int) -> Dict[str, Any]:
        """Completely delete a crawl job and all associated data.

        Order of operations (critical for data integrity):
        1. Get job info and collect hashes/URIs to delete
        2. Delete MediaSource records (breaks references)
        3. Delete Page records
        4. Delete orphan Media records (no longer referenced)
        5. Delete orphan media files from vault
        6. Delete screenshot files for deleted pages
        7. Delete CrawlJob record
        8. Delete state files

        Returns:
            Dict with deletion stats
        """
        stats = {
            'deleted': False,
            'media_files_deleted': 0,
            'screenshots_deleted': 0,
            'pages_deleted': 0,
            'media_sources_deleted': 0,
            'state_files_deleted': 0,
        }

        # Step 1-6: Database deletion (returns info for file deletion)
        db_result = await self.db.delete_crawl_job(job_id, purge_data=True)

        if not db_result['deleted']:
            logger.warning(f"Job {job_id} not found")
            return stats

        stats['deleted'] = True
        stats['pages_deleted'] = len(db_result['page_uris'])

        target_uri = db_result['target_uri']
        domain = self._get_domain(target_uri) if target_uri else None

        # Step 5: Delete orphan media files from vault
        for md5_hash in db_result['orphan_media']:
            # Try common extensions
            for ext in ['jpg', 'jpeg', 'png', 'gif', 'webp', 'mp4', 'webm']:
                file_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext)
                if file_path.exists():
                    try:
                        file_path.unlink()
                        stats['media_files_deleted'] += 1
                        logger.debug(f"Deleted media file: {file_path}")
                        break
                    except Exception as e:
                        logger.warning(f"Failed to delete {file_path}: {e}")

        # Step 6: Delete screenshot files for deleted pages
        if domain:
            linkpeek_dir = Path(self.vault_path) / 'linkpeek_vault' / domain
            for page_uri in db_result['page_uris']:
                parsed = Uri(page_uri)
                url_path = (parsed.path or '').lstrip('/') or 'index'
                url_path = url_path.replace('/', '_')
                # Try common screenshot patterns
                for pattern in [f'{url_path}.png', f'{url_path}.jpg', f'{url_path}_0.jpg']:
                    ss_path = linkpeek_dir / pattern
                    if ss_path.exists():
                        try:
                            ss_path.unlink()
                            stats['screenshots_deleted'] += 1
                            logger.debug(f"Deleted screenshot: {ss_path}")
                        except Exception as e:
                            logger.warning(f"Failed to delete {ss_path}: {e}")

        # Step 7: Already done in db.delete_crawl_job

        # Step 8: Delete state files
        if target_uri:
            # Main state file
            state_file = self._get_state_file(target_uri)
            if state_file.exists():
                try:
                    state_file.unlink()
                    stats['state_files_deleted'] += 1
                    logger.debug(f"Deleted state file: {state_file}")
                except Exception as e:
                    logger.warning(f"Failed to delete state file: {e}")

            # State log file
            state_log = self._get_state_log(target_uri)
            if state_log.path.exists():
                try:
                    state_log.path.unlink()
                    stats['state_files_deleted'] += 1
                    logger.debug(f"Deleted state log: {state_log.path}")
                except Exception as e:
                    logger.warning(f"Failed to delete state log: {e}")

            # Hydra state file (per-domain, keep it - other jobs may use it)
            # Don't delete hydra state as it's domain-wide, not job-specific

        logger.info(f"Purged job {job_id}: {stats['media_files_deleted']} media, "
                   f"{stats['screenshots_deleted']} screenshots, {stats['pages_deleted']} pages")

        return stats

    def _get_domain(self, url: str) -> str:
        """Extract domain from URL."""
        parsed = Uri(url)
        return parsed.hostname.lower()

    def _track_domain_stat(self, domain: str, stat: str, increment: int = 1):
        """Track per-domain stats for vault commits."""
        if domain not in self._domain_stats:
            self._domain_stats[domain] = {'pages_changed': 0, 'media_new': 0, 'screenshots_new': 0}
        self._domain_stats[domain][stat] += increment

    def _extract_text_from_html(self, html: str, base_url: str = None) -> tuple:
        """Extract title, text content, and markdown from HTML.

        Args:
            html: Raw HTML content
            base_url: Base URL for resolving relative links/images

        Returns: (title, text_content, markdown)
        """
        try:
            from bs4 import BeautifulSoup
            soup = BeautifulSoup(html, 'html.parser')

            # Extract title
            title = ''
            title_tag = soup.find('title')
            if title_tag:
                title = title_tag.get_text(strip=True)

            # Convert to markdown using smart converter
            markdown = ''
            try:
                from html2md import html_to_markdown
                markdown = html_to_markdown(html, base_url=base_url)[:200000]
            except Exception:
                # Fallback to html2text
                try:
                    import html2text
                    h = html2text.HTML2Text()
                    h.ignore_links = False
                    h.ignore_images = False
                    h.body_width = 0
                    if base_url:
                        h.baseurl = base_url
                    markdown = h.handle(html)[:200000]
                except ImportError:
                    pass

            # Remove script and style elements for plain text
            for tag in soup(['script', 'style', 'nav', 'header', 'footer']):
                tag.decompose()

            # Get text content
            text = soup.get_text(separator=' ', strip=True)
            # Clean up whitespace
            import re
            text = re.sub(r'\s+', ' ', text)

            return title, text[:100000], markdown
        except Exception:
            return '', '', ''

    async def _archive_page_to_vault(
        self,
        uri: str,
        html: str,
        media_mappings: Dict[str, str] = None,
        crawl_job_id: int = None,
    ):
        """Archive a page to the HTML vault and store for search."""
        domain = self._get_domain(uri)
        html_vault = self.domain_vaults.get_html_vault(domain)
        is_changed, _ = await html_vault.archive_page(uri, html, media_mappings)
        self.stats['bytes_downloaded'] += len(html.encode('utf-8'))
        if is_changed:
            self._track_domain_stat(domain, 'pages_changed')
            self.stats['pages_changed'] += 1

        # Store page content for full-text search and phantom site recreation (in thread to not block)
        title, content, markdown = await asyncio.to_thread(self._extract_text_from_html, html, uri)
        parsed = Uri(uri)
        path = parsed.path or '/'
        await self.db.store_page(
            uri=uri,
            title=title,
            content=content,
            path=path,
            markdown=markdown,
            raw_html=html,
            crawl_job_id=crawl_job_id,
        )

    async def _archive_media_to_vault(
        self,
        url: str,
        md5_hash: str,
        ext: str,
        page_url: str = '',
    ):
        """Create symlink in domain media vault pointing to hash vault."""
        domain = self._get_domain(url)
        # Domain media path: vault/media_vault/{domain}/{url_path}
        parsed = Uri(url)
        url_path = (parsed.path or '').lstrip('/') or 'index'
        if not url_path.endswith(ext):
            url_path = f"{url_path}{ext}"
        domain_media_dir = Path(self.vault_path) / 'media_vault' / domain
        domain_media_path = domain_media_dir / url_path

        # Hash vault path: vault/{9-deep-path}/{hash}.{ext}
        hash_vault_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext.lstrip('.'))

        # Create symlink if not exists
        if not domain_media_path.exists():
            domain_media_path.parent.mkdir(parents=True, exist_ok=True)
            # Calculate relative path from domain media to hash vault
            rel_path = os.path.relpath(hash_vault_path, domain_media_path.parent)
            try:
                domain_media_path.symlink_to(rel_path)
                self._track_domain_stat(domain, 'media_new')
                self.stats['media_new'] += 1
            except FileExistsError:
                pass  # Already exists

    async def _archive_screenshot_to_vault(
        self,
        url: str,
        md5_hash: str,
        ext: str = 'jpg',
        suffix: str = '',
    ):
        """Create symlink in domain linkpeek vault pointing to hash vault."""
        domain = self._get_domain(url)
        # Domain screenshot path: vault/linkpeek_vault/{domain}/{url_path}{suffix}.{ext}
        parsed = Uri(url)
        url_path = (parsed.path or '').lstrip('/') or 'index'
        url_path = url_path.replace('/', '_') + f'{suffix}.{ext}'
        domain_ss_dir = Path(self.vault_path) / 'linkpeek_vault' / domain
        domain_ss_path = domain_ss_dir / url_path

        # Hash vault path: vault/{9-deep-path}/{hash}.{ext}
        hash_vault_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext)

        # Create symlink if not exists
        if not domain_ss_path.exists():
            domain_ss_path.parent.mkdir(parents=True, exist_ok=True)
            rel_path = os.path.relpath(hash_vault_path, domain_ss_path.parent)
            try:
                domain_ss_path.symlink_to(rel_path)
                self._track_domain_stat(domain, 'screenshots_new')
            except FileExistsError:
                pass

    async def _finish_domain_vaults(self, keywords: List[str] = None):
        """Commit changes to all domain vaults that have diffs."""
        for domain, stats in self._domain_stats.items():
            if stats['pages_changed'] > 0 or stats['media_new'] > 0 or stats['screenshots_new'] > 0:
                html_vault = self.domain_vaults.get_html_vault(domain)
                media_vault = self.domain_vaults.get_media_vault(domain)
                linkpeek_vault = self.domain_vaults.get_linkpeek_vault(domain)

                html_commit = await html_vault.finish_crawl(stats)
                media_commit = await media_vault.finish_crawl(stats)
                linkpeek_commit = await linkpeek_vault.finish_crawl(stats)

                if html_commit:
                    logger.info(f"Committed HTML vault for {domain}: {html_commit[:8]}")
                if media_commit:
                    logger.info(f"Committed media vault for {domain}: {media_commit[:8]}")
                if linkpeek_commit:
                    logger.info(f"Committed linkpeek vault for {domain}: {linkpeek_commit[:8]}")

    async def init(self):
        """Initialize database and vault."""
        await self.db.init()
        await self.vault.init()

    async def crawl(
        self,
        target_uri: str,
        keywords: List[str] = None,
        mode: CrawlMode = CrawlMode.IMAGES,
        depth: int = -1,  # -1 = unlimited
        max_pages: int = -1,  # -1 = unlimited
        download_media: bool = True,
        job_id: int = None,  # Optional existing job ID (skip creation if provided)
        quiet: bool = False,  # Disable progress bar (for UI-initiated crawls)
        hydra: bool = False,  # Parse RSS/Atom/Sitemap for fast discovery
    ) -> Dict[str, Any]:
        """
        Crawl a domain for media.

        Args:
            target_uri: Starting URL
            keywords: Keywords to tag media with (e.g., ["rick and morty"])
            mode: CrawlMode - what to collect (IMAGES, VIDEOS, MEDIA, ALL)
            depth: Crawl depth (-1 = unlimited)
            max_pages: Max pages to crawl (-1 = unlimited)
            download_media: Whether to download media or just index URLs
            job_id: Optional existing job ID (skip creation if provided)
            quiet: Disable progress bar (for UI-initiated crawls)
            hydra: Parse RSS/Atom/Sitemap feeds for fast URL discovery

        Returns:
            Crawl statistics
        """
        keywords = keywords or []

        # Check if target is a VCS repository (git, hg, svn, etc.)
        vcs_type, clone_url = detect_vcs(target_uri)
        if vcs_type:
            logger.info(f"Detected {vcs_type} repository: {clone_url}")
            return await self.clone_and_index(
                target_uri=target_uri,
                clone_url=clone_url,
                vcs_type=vcs_type,
                keywords=keywords,
                job_id=job_id,
                quiet=quiet,
            )

        # Create crawl job (unless one was provided)
        if job_id is None:
            job_id = await self.db.create_crawl_job(
                target_uri=target_uri,
                keywords=keywords,
                mode=mode.value,
            )

        # Always log to file for this job
        start_job_logging(job_id)

        logger.info(f"Starting neopig crawl job {job_id}")
        logger.info(f"Target: {target_uri}")
        logger.info(f"Mode: {mode.value}")
        logger.info(f"Keywords: {keywords}")
        logger.info(f"Depth: {'unlimited' if depth == -1 else depth}")
        if hydra:
            logger.info(f"Hydra mode: enabled (RSS/Atom/Sitemap discovery)")

        # Load saved state if exists (resume support)
        self._load_state(target_uri)

        # Track timing for stats
        start_time = datetime.now(timezone.utc)
        self.stats['crawl_started'] = int(start_time.timestamp())

        def format_size(b: int) -> str:
            if b < 1024:
                return f"{b}B"
            elif b < 1024 * 1024:
                return f"{b/1024:.1f}KB"
            elif b < 1024 * 1024 * 1024:
                return f"{b/(1024*1024):.1f}MB"
            else:
                return f"{b/(1024*1024*1024):.1f}GB"

        # Track total URIs discovered for progress bar
        uris_total = [0]  # Use list for mutability in closure

        def on_uris_total(total: int):
            uris_total[0] = total
            self.stats['pages_pending'] = max(0, total - self.stats['pages_crawled'])
            if self.pbar and total > 0:
                self.pbar.total = total
                self.pbar.refresh()

        # Create progress bar with dynamic total
        initial_pages = len(self.seen_pages)
        self.pbar = tqdm(
            total=1,  # Start with 1, will be updated as links are discovered
            initial=initial_pages,
            unit="pages",
            dynamic_ncols=True,
            bar_format='{n_fmt}/{total_fmt} pages [{elapsed}] {postfix}',
            mininterval=0.1,
            disable=quiet,
        )
        self.pbar.set_postfix_str(
            f"new: {self.stats['media_downloaded']}, "
            f"skip: {self.stats['duplicates_skipped']}, "
            f"dup: {self.stats['content_exists']}, "
            f"found: {self.stats['media_found']}, "
            f"ss: {self.stats['screenshots_taken']}, "
            f"err: {self.stats['errors']}, "
            f"{format_size(self.stats['bytes_stored'])} stored"
        )

        def update_pbar():
            self.pbar.set_postfix_str(
                f"new: {self.stats['media_downloaded']}, "
                f"skip: {self.stats['duplicates_skipped']}, "
                f"dup: {self.stats['content_exists']}, "
                f"found: {self.stats['media_found']}, "
                f"ss: {self.stats['screenshots_taken']}, "
                f"err: {self.stats['errors']}, "
                f"{format_size(self.stats['bytes_stored'])} stored"
            )
            self.pbar.refresh()

        # Open append-only state log for fast writes
        self._state_log = self._get_state_log(target_uri)
        self._state_log.open()

        # Background task to push stats to DB every second (for UI visibility)
        stats_stop = asyncio.Event()

        async def stats_pusher():
            while not stats_stop.is_set():
                await asyncio.sleep(1)
                if not stats_stop.is_set():
                    try:
                        await self.db.update_crawl_job_stats(job_id, self.stats)
                    except Exception:
                        pass  # Ignore DB errors in background task

        stats_task = asyncio.create_task(stats_pusher())

        # Progress callback
        async def on_progress(msg: str):
            self.stats['pages_crawled'] += 1
            self.pbar.update(1)
            update_pbar()

        # STREAMING ETL: HTML → Media + Screenshots run concurrently
        max_workers = multiprocessing.cpu_count() * 3 if self.fast_mode else multiprocessing.cpu_count()
        max_media_workers = max_workers
        max_ss_workers = max_workers

        media_semaphore = asyncio.Semaphore(max_media_workers)
        ss_semaphore = asyncio.Semaphore(max_ss_workers)

        # Track active tasks for graceful completion
        active_media_tasks = set()
        active_ss_tasks = set()
        downloaded_count = [0]

        async def process_media_item(item):
            """Download a single media item."""
            async with media_semaphore:
                url = item.get('url')
                try:
                    success, md5_hash = await self._process_media_item(item, job_id, keywords)
                    if success and md5_hash:
                        self.seen_media[url] = md5_hash
                        self._state_log.media(md5_hash, url)
                        downloaded_count[0] += 1
                        if downloaded_count[0] % 10 == 0:
                            update_pbar()
                    return success
                except Exception as e:
                    logger.warning(f"Media download failed {url}: {e}")
                    return False

        async def process_screenshot(uri, content_length):
            """Take a single screenshot."""
            async with ss_semaphore:
                try:
                    await self._capture_page_screenshot(uri, job_id, page_title='', content_length=content_length)
                    self._state_log.screenshot(uri)
                    return True
                except Exception as e:
                    logger.warning(f"Screenshot failed {uri}: {e}")
                    return False

        # STREAMING callbacks - immediately spawn work
        async def on_media_streaming(item: Dict[str, Any]):
            """Queue media for immediate parallel download."""
            url = item.get('url')
            if not url:
                return

            if url in self.seen_media:
                md5_hash = self.seen_media[url]
                page_uri = item.get('source_page', '')
                existing = await self.db.check_media_uri_exists(url, page_uri)
                if not existing:
                    await self.db.add_media_source(
                        md5_hash=md5_hash, media_uri=url, page_uri=page_uri,
                        page_title=item.get('page_title', ''),
                        page_description=item.get('page_description', ''),
                        page_keywords=item.get('page_keywords', ''),
                        page_content=item.get('page_content', ''),
                        alt_text=item.get('alt_text', ''),
                        link_text=item.get('link_text', ''),
                        detail_page_uri=item.get('detail_page_url', ''),
                        detail_title=item.get('detail_title', ''),
                        detail_content=item.get('detail_content', ''),
                        searchable_text=item.get('searchable_text', ''),
                        crawl_job_id=job_id,
                    )
                    self.stats['content_exists'] += 1
                return

            self.stats['media_found'] += 1
            if download_media:
                # Spawn download task immediately (don't await)
                task = asyncio.create_task(process_media_item(item))
                active_media_tasks.add(task)
                task.add_done_callback(active_media_tasks.discard)
            update_pbar()

        # Track archive tasks too
        active_archive_tasks = set()

        async def on_page_streaming(uri: str, html: str):
            """Archive HTML and spawn screenshot immediately - all non-blocking."""
            self.seen_pages.add(uri)
            self._state_log.page(uri)

            # Spawn archive task (don't await - let it run in background)
            async def do_archive():
                await self._archive_page_to_vault(uri, html, media_mappings=None, crawl_job_id=job_id)
            archive_task = asyncio.create_task(do_archive())
            active_archive_tasks.add(archive_task)
            archive_task.add_done_callback(active_archive_tasks.discard)

            if self.screenshot_config.enabled:
                # Spawn screenshot task immediately (don't await)
                task = asyncio.create_task(process_screenshot(uri, len(html)))
                active_ss_tasks.add(task)
                task.add_done_callback(active_ss_tasks.discard)

        # Hydra: ALWAYS check known feeds (self-healing), discover new if --hydra
        # This runs on every crawl to catch new content from persisted feeds
        hydra_urls = await self._fetch_hydra_urls(target_uri, discover_new=hydra)
        if hydra_urls:
            logger.info(f"Hydra: Injecting {len(hydra_urls)} NEW URLs into crawl queue")

        logger.info("=== STREAMING ETL: HTML + Media + Screenshots ===")
        pages = await self.fetcher.fetch_with_depth(
            start_url=target_uri,
            depth=depth,
            max_pages=max_pages,
            query_keywords=keywords,
            mode=mode,
            media_callback=on_media_streaming,
            progress_callback=on_progress,
            page_callback=on_page_streaming,
            uris_total_callback=on_uris_total,
            initial_visited=self.seen_pages if self.seen_pages else None,
            hydra_urls=hydra_urls,
        )

        self.stats['pages_crawled'] = len(pages)
        logger.info(f"HTML complete: {len(pages)} pages. Waiting for {len(active_media_tasks)} media + {len(active_ss_tasks)} screenshots...")

        # Wait for all spawned tasks to complete
        if active_media_tasks:
            await asyncio.gather(*active_media_tasks, return_exceptions=True)
        if active_ss_tasks:
            await asyncio.gather(*active_ss_tasks, return_exceptions=True)
        if active_archive_tasks:
            await asyncio.gather(*active_archive_tasks, return_exceptions=True)

        update_pbar()
        logger.info(f"ETL complete: {downloaded_count[0]} media downloaded, {self.stats['screenshots_taken']} screenshots")

        # Stop stats pusher
        stats_stop.set()
        stats_task.cancel()
        try:
            await stats_task
        except asyncio.CancelledError:
            pass

        # Close state log
        self._state_log.stats(self.stats)  # Final stats checkpoint
        self._state_log.close()

        # Close progress bar
        self.pbar.close()

        # Calculate final stats
        elapsed = (datetime.now(timezone.utc) - start_time).total_seconds()
        rate = self.stats['media_downloaded'] / elapsed * 60 if elapsed > 0 else 0

        # Update job status
        await self.db.complete_crawl_job(job_id, self.stats)

        # Commit domain vaults if there were changes
        await self._finish_domain_vaults(keywords)

        logger.info(f"=== CRAWL COMPLETE ({elapsed:.0f}s) ===")
        logger.info(f"  Pages: {self.stats['pages_crawled']} | "
                   f"Found: {self.stats['media_found']} | "
                   f"Downloaded: {self.stats['media_downloaded']} | "
                   f"Dupes: {self.stats['duplicates_skipped']} | "
                   f"Screenshots: {self.stats['screenshots_taken']} | "
                   f"Errors: {self.stats['errors']}")
        logger.info(f"  Rate: {rate:.1f}/min | Total time: {elapsed:.1f}s")

        # Stop logging to file
        stop_job_logging(job_id)

        return self.stats

    async def _process_media_item(
        self,
        item: Dict[str, Any],
        job_id: int,
        keywords: List[str]
    ) -> Tuple[bool, Optional[str]]:
        """Download and store a media item with page context. Returns (success, md5_hash).

        Implements the skeleton key approach:
        - If detail_page_url is set, resolves canonical image URL
        - Collects both embedding_title (page_title) and detail_title
        - Prefers canonical URL for download, falls back to original
        """
        media_uri = item['url']
        page_uri = item.get('source_page', '')
        media_type = item.get('media_type', 'unknown')
        detail_page_url = item.get('detail_page_url')

        # Extract page context for searchability (embedding context)
        page_title = item.get('page_title', '')
        page_description = item.get('page_description', '')
        page_keywords = item.get('page_keywords', '')
        page_content = item.get('page_content', '')  # Blog post text for full-text search
        alt_text = item.get('alt_text', '')
        link_text = item.get('link_text', '')
        searchable_text = item.get('searchable_text', '')  # Combined metadata from accumulator

        # Skeleton key: detail context (from detail page if Pinterest-style gallery)
        detail_page_uri = ''
        detail_title = ''
        detail_content = ''

        # Score tracking - higher = better quality
        original_media_uri = media_uri
        is_canonical = False  # Track if we resolved to a canonical URL

        try:
            # Crystal algorithm: resolve canonical image if this looks like a gallery thumbnail
            if detail_page_url:
                canonical_result = await self.fetcher.resolve_canonical_image(
                    detail_page_url=detail_page_url,
                    thumbnail_url=media_uri,
                    embedding_title=page_title,  # Pass listing page title
                )
                if canonical_result:
                    detail_page_uri = detail_page_url
                    detail_title = canonical_result.get('detail_title', '')
                    detail_content = canonical_result.get('detail_content', '')
                    # Use canonical URL if found, otherwise keep thumbnail
                    if canonical_result.get('canonical_url'):
                        logger.debug(f"Canonical resolution: {media_uri} -> {canonical_result['canonical_url']}")
                        media_uri = canonical_result['canonical_url']
                        is_canonical = True  # Resolved to full-res
                    # Enrich metadata from detail page
                    if not alt_text and canonical_result.get('description'):
                        alt_text = canonical_result['description']
            # Check if this exact media+page combo was already crawled
            existing_hash = await self.db.check_media_uri_exists(media_uri, page_uri)
            if existing_hash:
                self.stats['duplicates_skipped'] += 1
                logger.debug(f"Already crawled: {media_uri} from {page_uri}")
                return True, existing_hash  # Already have it, return existing hash

            # Check if content already in vault (same MD5 = same content)
            # We still need to add the new page context even if content exists
            if media_uri in self.seen_media:
                # Already processed this media_uri, just add context
                # We need to fetch to get MD5, but we can skip if we track it
                pass

            # Fetch the media
            result = await self.fetcher.fetch_media(media_uri)
            if not result:
                self.stats['errors'] += 1
                return False, None

            md5_hash = result['md5_hash']
            self.stats['bytes_downloaded'] += result.get('size', 0)

            # Check if content already in vault
            if await self.vault.exists(md5_hash):
                # Content exists, but add this new page context with skeleton key
                await self.db.add_media_source(
                    md5_hash=md5_hash,
                    media_uri=media_uri,
                    page_uri=page_uri,
                    page_title=page_title,
                    page_description=page_description,
                    page_keywords=page_keywords,
                    page_content=page_content,
                    alt_text=alt_text,
                    link_text=link_text,
                    detail_page_uri=detail_page_uri,
                    detail_title=detail_title,
                    detail_content=detail_content,
                    searchable_text=searchable_text,
                    crawl_job_id=job_id,
                )
                self.stats['content_exists'] += 1
                logger.debug(f"Content exists, added context: {md5_hash} from {page_uri}")
                return True, md5_hash

            # Store in vault (new content)
            ext = self._get_extension(media_uri, result.get('mime_type', ''))
            await self.vault.store(md5_hash, result['data'], ext)
            self.stats['bytes_stored'] += len(result['data'])

            # Emit to live feed immediately after disk save (before DB insert)
            emit_live_media({
                'md5_hash': md5_hash,
                'media_type': media_type,
                'mime_type': result.get('mime_type', ''),
                'file_size': len(result['data']),
                'alt_text': alt_text,
                'title': page_title,
                'media_uri': media_uri,
                'page_uri': page_uri,
            })

            # Create symlink in domain media vault pointing to hash vault
            await self._archive_media_to_vault(media_uri, md5_hash, ext, page_uri)

            # Determine score based on resolution
            score = SCORE_FULL_RES if is_canonical else SCORE_THUMBNAIL

            # Record in database with full context and skeleton key
            await self.db.create_media_record(
                md5_hash=md5_hash,
                media_uri=media_uri,
                page_uri=page_uri,
                crawl_job_id=job_id,
                media_type=media_type,
                mime_type=result.get('mime_type', ''),
                file_size=result.get('size', 0),
                page_title=page_title,
                page_description=page_description,
                page_keywords=page_keywords,
                page_content=page_content,
                alt_text=alt_text,
                link_text=link_text,
                detail_page_uri=detail_page_uri,
                detail_title=detail_title,
                detail_content=detail_content,
                searchable_text=searchable_text,
                score=score,
            )

            self.stats['media_downloaded'] += 1
            logger.debug(f"Stored: {md5_hash} ({media_uri})")
            return True, md5_hash

        except Exception as e:
            logger.warning(f"Failed to process {media_uri}: {e}")
            self.stats['errors'] += 1
            return False, None

    async def _capture_page_screenshot(
        self,
        page_uri: str,
        job_id: int,
        page_title: str = '',
        content_length: int = 0,
    ):
        """Capture and store a screenshot of a page.

        Respects robots.txt crawl-delay by coordinating with the fetcher's
        per-domain delay tracking. Screenshots use a headless browser which
        makes its own HTTP request, so we must enforce delay before capture.

        Args:
            page_uri: URL of the page to screenshot
            job_id: Crawl job ID for database tracking
            page_title: Optional page title for metadata
            content_length: Raw HTML content length for dynamic delay calculation
        """
        if not self.screenshot_config.enabled:
            logger.debug(f"Screenshots disabled, skipping {page_uri}")
            return

        if page_uri in self.seen_screenshots:
            logger.debug(f"Screenshot already exists for {page_uri}")
            return

        logger.info(f"Taking screenshot: {page_uri}")

        try:
            # Enforce crawl delay before screenshot (headless browser makes HTTP request)
            domain = self._get_domain(page_uri)
            await self.fetcher._enforce_crawl_delay(domain)

            # Pass content_length for dynamic delay calculation on long pages
            result = await self.screenshot.capture(page_uri, content_length=content_length)
            if not result:
                return

            # Normalize to list (oversized images return multiple chunks)
            chunks = result if isinstance(result, list) else [result]
            total_size = 0

            for i, chunk in enumerate(chunks):
                md5_hash = chunk['md5_hash']
                screenshot_data = chunk['data']
                screenshot_size = len(screenshot_data)
                screenshot_ext = chunk.get('format', 'png')
                screenshot_mime = chunk.get('mime_type', 'image/png')
                total_size += screenshot_size

                # Store in MD5 vault (for deduplication)
                if not await self.vault.exists(md5_hash):
                    await self.vault.store(md5_hash, screenshot_data, screenshot_ext)
                    self.stats['bytes_stored'] += screenshot_size

                # Create symlink in linkpeek vault (with suffix for chunks)
                suffix = f'_{i}' if len(chunks) > 1 else ''
                await self._archive_screenshot_to_vault(page_uri, md5_hash, screenshot_ext, suffix=suffix)

                # Record in database (all chunks, with chunk index in alt_text)
                chunk_label = f" (part {i+1}/{len(chunks)})" if len(chunks) > 1 else ""
                await self.db.create_media_record(
                    md5_hash=md5_hash,
                    media_uri=f"screenshot:{page_uri}{suffix}",
                    page_uri=page_uri,
                    crawl_job_id=job_id,
                    media_type='screenshot',
                    mime_type=screenshot_mime,
                    file_size=screenshot_size,
                    page_title=page_title,
                    page_description='',
                    page_keywords='',
                    alt_text=f"Screenshot of {page_uri}{chunk_label}",
                    link_text='',
                    score=SCORE_SCREENSHOT,
                )

            # Screenshots are fetched by headless browser (network traffic)
            self.stats['bytes_downloaded'] += total_size
            self.stats['screenshots_taken'] += 1
            self.seen_screenshots.add(page_uri)  # Only mark seen after success
            logger.debug(f"Screenshot captured: {page_uri} -> {chunks[0]['md5_hash']} ({len(chunks)} chunk(s))")

        except Exception as e:
            logger.warning(f"Screenshot failed for {page_uri}: {e}")

    async def clone_and_index(
        self,
        target_uri: str,
        clone_url: str,
        vcs_type: str,
        keywords: List[str] = None,
        job_id: int = None,
        quiet: bool = False,
        pull: bool = False,
    ) -> Dict[str, Any]:
        """
        Clone a VCS repository and index its files.

        Like Hydra mode for feeds, VCS detection is a "smart source" that
        bypasses slow HTTP crawling for high-priority ingestion.

        Args:
            target_uri: Original URI
            clone_url: Clone URL (may differ from target_uri)
            vcs_type: 'git', 'hg', 'svn', or 'fossil'
            keywords: Keywords to tag files with
            job_id: Optional existing job ID
            quiet: Disable progress output
            pull: Update existing clone instead of fresh clone

        Returns:
            Indexing statistics
        """
        from filevault import content_hash

        keywords = keywords or []
        repo_vault_base = Path(self.vault.path).parent / 'repo_vault'

        # Create job
        if job_id is None:
            job_id = await self.db.create_crawl_job(
                target_uri=target_uri,
                keywords=keywords,
                mode='code',
            )

        start_job_logging(job_id)
        logger.info(f"Starting VCS clone job {job_id}")
        logger.info(f"Repository: {clone_url}")
        logger.info(f"VCS type: {vcs_type}")
        logger.info(f"Keywords: {keywords}")

        start_time = datetime.now(timezone.utc)
        stats = {
            'vcs_type': vcs_type,
            'clone_url': clone_url,
            'files_indexed': 0,
            'files_skipped': 0,
            'bytes_stored': 0,
            'errors': 0,
            'commit_hash': None,
        }

        # Determine repo path
        repo_path = get_repo_path(clone_url, repo_vault_base)
        logger.info(f"Repo path: {repo_path}")

        # Clone or pull
        if repo_path.exists() and (repo_path / '.git').exists():
            logger.info(f"Repository exists, pulling updates...")
            success, msg = await pull_repo_async(repo_path)
            if not success:
                logger.error(f"Pull failed: {msg}")
                stats['errors'] += 1
            else:
                logger.info(f"Pull: {msg}")
        else:
            logger.info(f"Cloning repository...")
            success, msg = await clone_repo_async(clone_url, repo_path, vcs_type, shallow=True)
            if not success:
                logger.error(f"Clone failed: {msg}")
                await self.db.complete_crawl_job(job_id, stats)
                return stats
            logger.info(f"Clone: {msg}")

        # Get commit hash
        stats['commit_hash'] = get_commit_hash(repo_path)
        logger.info(f"Commit: {stats['commit_hash']}")

        # Walk files and index
        files = list(walk_files(repo_path, include_binary=True))
        logger.info(f"Found {len(files)} files to index")

        pbar = tqdm(files, unit="files", disable=quiet)
        for file_path in pbar:
            try:
                content = file_path.read_bytes()
                h = content_hash(content)
                ext = file_path.suffix or '.txt'
                file_size = len(content)

                # Store in vault
                vault_path = await self.vault.store(h, content, ext)
                stats['bytes_stored'] += file_size

                # Replace file with symlink
                rel_path = os.path.relpath(vault_path, file_path.parent)
                file_path.unlink()
                file_path.symlink_to(rel_path)

                # Get file metadata
                repo_rel_path = str(file_path.relative_to(repo_path))
                language = get_file_language(file_path)
                is_binary = is_binary_file(Path(vault_path))

                # Create media record with repo metadata
                await self.db.create_media_record(
                    md5_hash=h,
                    media_uri=f"{clone_url}/blob/HEAD/{repo_rel_path}",
                    page_uri=target_uri,
                    page_title=repo_rel_path,
                    media_type='code',
                    mime_type=f"text/{language}" if language and not is_binary else "application/octet-stream",
                    file_size=file_size,
                    alt_text=language or '',
                    keywords=keywords,
                    crawl_job_id=job_id,
                    score=SCORE_FULL_RES,
                    # VCS repo metadata
                    repo_uri=clone_url,
                    repo_path=repo_rel_path,
                    commit_hash=stats['commit_hash'],
                    vcs_type=vcs_type,
                )

                stats['files_indexed'] += 1
                pbar.set_postfix_str(f"indexed: {stats['files_indexed']}, {stats['bytes_stored'] // 1024}KB")

            except Exception as e:
                logger.warning(f"Error indexing {file_path}: {e}")
                stats['errors'] += 1
                stats['files_skipped'] += 1

        pbar.close()

        # Complete job
        duration = (datetime.now(timezone.utc) - start_time).total_seconds()
        stats['duration_seconds'] = duration
        await self.db.complete_crawl_job(job_id, stats)

        logger.info(f"VCS indexing complete: {stats['files_indexed']} files, {stats['bytes_stored'] // 1024}KB stored")
        return stats

    async def backfill_missing_screenshots(self, domain: str = None, crawl_job_id: int = None, fast_mode: bool = False, quiet: bool = False, create_job: bool = True):
        """Capture screenshots for pages that don't have them yet.

        Args:
            domain: Domain to backfill (e.g., 'example.com'), or None for all pages
            crawl_job_id: Optional crawl job ID for tracking
            fast_mode: Skip crawl delay, use max parallel workers
            quiet: Disable progress bar
            create_job: If True, create a backfill_jobs record (False when called from crawl)
        """
        import aiosqlite
        from datetime import datetime, timezone

        if not self.screenshot_config.enabled:
            return

        # Get pages without screenshots (optionally filtered by domain)
        pages = await self.db.get_pages_without_screenshots(domain)
        # Filter out already-seen
        pages = [p for p in pages if p not in self.seen_screenshots]
        if not pages:
            logger.debug(f"No pages need screenshots{f' for {domain}' if domain else ''}")
            return

        total = len(pages)
        logger.info(f"Backfilling {total} missing screenshots{f' for {domain}' if domain else ''}")

        # Create backfill job record for progress tracking (skip when part of a crawl)
        db_path = self.db.db_path
        job_id = None
        if create_job:
            async with aiosqlite.connect(db_path) as db:
                await db.execute("PRAGMA journal_mode=WAL")
                # Ensure table exists
                await db.execute("""
                    CREATE TABLE IF NOT EXISTS backfill_jobs (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        job_type TEXT NOT NULL,
                        domain_filter TEXT,
                        status TEXT DEFAULT 'running',
                        total_records INTEGER DEFAULT 0,
                        processed_records INTEGER DEFAULT 0,
                        error_count INTEGER DEFAULT 0,
                        started_at TEXT NOT NULL,
                        completed_at TEXT
                    )
                """)
                now = datetime.now(timezone.utc).isoformat()
                cursor = await db.execute(
                    "INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, processed_records, error_count, started_at) VALUES (?, ?, 'running', ?, 0, 0, ?)",
                    ('screenshots', domain, total, now)
                )
                job_id = cursor.lastrowid
                await db.commit()
            logger.info(f"Created screenshot backfill job {job_id}")

        # Concurrent workers: 3x CPU in fast mode, 1x CPU otherwise
        import multiprocessing
        max_workers = multiprocessing.cpu_count() * 3 if fast_mode else multiprocessing.cpu_count()
        semaphore = asyncio.Semaphore(max_workers)
        logger.info(f"{'Fast' if fast_mode else 'Normal'} mode: {max_workers} concurrent workers")

        pbar = tqdm(total=total, desc="Screenshots", unit="pages", disable=quiet, smoothing=0.1)
        captured = 0
        failed = 0

        async def update_progress(success: bool):
            """Atomically update job progress in DB."""
            nonlocal captured, failed
            if success:
                captured += 1
            else:
                failed += 1
            if job_id:
                async with aiosqlite.connect(db_path) as db:
                    if success:
                        await db.execute(
                            "UPDATE backfill_jobs SET processed_records = processed_records + 1 WHERE id = ?",
                            (job_id,)
                        )
                    else:
                        await db.execute(
                            "UPDATE backfill_jobs SET error_count = error_count + 1 WHERE id = ?",
                            (job_id,)
                        )
                    await db.commit()

        async def capture_one(page_uri: str):
            async with semaphore:
                try:
                    # Get content length for dynamic delay
                    content_length = 0
                    async with aiosqlite.connect(db_path) as conn:
                        cursor = await conn.execute(
                            "SELECT LENGTH(raw_html) FROM pages WHERE uri = ?",
                            (page_uri,)
                        )
                        row = await cursor.fetchone()
                        if row and row[0]:
                            content_length = row[0]

                    await self._capture_page_screenshot(page_uri, crawl_job_id or 0, content_length=content_length)
                    # Check if screenshot was actually captured (added to seen_screenshots)
                    success = page_uri in self.seen_screenshots
                except Exception as e:
                    logger.debug(f"Screenshot failed {page_uri}: {e}")
                    success = False
                await update_progress(success)
                pbar.update(1)

        # Run all captures concurrently (semaphore limits parallelism)
        await asyncio.gather(*[capture_one(uri) for uri in pages])
        pbar.close()

        # Mark job completed (only if tracking)
        if job_id:
            async with aiosqlite.connect(db_path) as db:
                now = datetime.now(timezone.utc).isoformat()
                await db.execute(
                    "UPDATE backfill_jobs SET status = 'completed', completed_at = ? WHERE id = ?",
                    (now, job_id)
                )
                await db.commit()

        logger.info(f"Screenshot backfill complete: {captured} captured, {failed} failed")

    def _get_extension(self, url: str, mime_type: str) -> str:
        """Determine file extension from URL or MIME type."""
        # Try from URL path
        path = Uri(url).path.lower()
        for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.svg', '.bmp',
                    '.mp4', '.webm', '.mov', '.avi', '.mkv']:
            if path.endswith(ext):
                return ext.lstrip('.')

        # Try from MIME type
        mime_map = {
            'image/jpeg': 'jpg',
            'image/png': 'png',
            'image/gif': 'gif',
            'image/webp': 'webp',
            'image/svg+xml': 'svg',
            'video/mp4': 'mp4',
            'video/webm': 'webm',
        }
        for mt, ext in mime_map.items():
            if mt in mime_type:
                return ext

        return 'bin'


def trim_html_wrapper(html: str) -> str:
    """Strip nav, header, footer, sidebar, and logo elements from HTML.

    Useful for cleaning up Discourse and similar sites before markdown conversion.
    """
    from bs4 import BeautifulSoup
    soup = BeautifulSoup(html, 'html.parser')

    # Remove common wrapper elements
    selectors_to_remove = [
        'nav', 'header', 'footer', 'aside',
        '.sidebar', '.nav', '.navigation', '.menu',
        '.header', '.footer', '.logo', '.site-logo',
        '#header', '#footer', '#nav', '#sidebar',
        '.d-header', '.d-footer',  # Discourse specific
        '.header-wrapper', '.footer-wrapper',
        '[role="banner"]', '[role="navigation"]', '[role="contentinfo"]',
    ]

    for selector in selectors_to_remove:
        for tag in soup.select(selector):
            tag.decompose()

    # Remove site logo images (be specific to avoid removing content images)
    for img in soup.find_all('img'):
        src = img.get('src', '').lower()
        alt = img.get('alt', '').lower()
        cls = ' '.join(img.get('class', [])).lower()
        # Only remove if it's clearly a site logo, not general icons
        is_logo = 'logo' in cls or 'brand' in cls or 'site-logo' in src
        is_logo = is_logo or (alt and ('logo' in alt or 'brand' in alt))
        if is_logo:
            img.decompose()

    return str(soup)


def extract_meta_from_html(html: str) -> tuple[str, list[str]]:
    """Extract meta description and keywords from HTML.

    Returns:
        Tuple of (description, keywords_list)
    """
    from bs4 import BeautifulSoup
    import re

    soup = BeautifulSoup(html, 'html.parser')
    description = ""
    keywords = []

    # Extract meta description
    meta_desc = soup.find('meta', attrs={'name': re.compile(r'^description$', re.I)})
    if meta_desc and meta_desc.get('content'):
        description = meta_desc['content'].strip()[:500]

    # Extract meta keywords
    meta_kw = soup.find('meta', attrs={'name': re.compile(r'^keywords$', re.I)})
    if meta_kw and meta_kw.get('content'):
        raw_kw = meta_kw['content']
        keywords = [k.strip().lower() for k in raw_kw.split(',') if k.strip()]

    # Also check og:description as fallback
    if not description:
        og_desc = soup.find('meta', attrs={'property': 'og:description'})
        if og_desc and og_desc.get('content'):
            description = og_desc['content'].strip()[:500]

    # Extract from article:tag meta tags (common in blogs)
    for tag_meta in soup.find_all('meta', attrs={'property': 'article:tag'}):
        if tag_meta.get('content'):
            keywords.append(tag_meta['content'].strip().lower())

    # Dedupe keywords
    keywords = list(dict.fromkeys(keywords))[:20]

    return description, keywords


def _process_single_page(args: tuple) -> tuple:
    """Process a single page for markdown conversion (runs in thread within process)."""
    from html2md import html_to_markdown
    import json

    page_id, uri, raw_html, trim_wrapper = args
    try:
        description, keywords = extract_meta_from_html(raw_html)
        raw = raw_html
        if trim_wrapper:
            raw = trim_html_wrapper(raw)
        new_markdown = html_to_markdown(raw, base_url=uri)[:200000]
        return (page_id, new_markdown, description, json.dumps(keywords) if keywords else None, None)
    except Exception as e:
        return (page_id, None, None, None, str(e))


def _process_chunk(args: tuple) -> dict:
    """Process a chunk of pages in a subprocess with thread workers.

    Each process gets a chunk and spawns threads to burn through it.
    Each thread has its own DB connection - WAL mode allows concurrent writes.
    Background flusher thread syncs progress every 100ms (workers never block).
    """
    import sqlite3
    import threading
    import time
    from concurrent.futures import ThreadPoolExecutor

    chunk, db_path, trim_wrapper, threads_per_process, job_id = args

    # Thread-local storage for DB connections
    thread_local = threading.local()
    progress = {'updated': 0, 'errors': 0}
    progress_lock = threading.Lock()
    stop_flusher = threading.Event()

    def get_conn():
        """Get thread-local DB connection for page writes."""
        if not hasattr(thread_local, 'conn'):
            thread_local.conn = sqlite3.connect(db_path, timeout=60.0)
            thread_local.conn.execute("PRAGMA journal_mode=WAL")
            thread_local.conn.execute("PRAGMA synchronous=NORMAL")
            thread_local.conn.execute("PRAGMA busy_timeout=60000")
        return thread_local.conn

    def flusher_thread():
        """Background thread that syncs progress to DB every 100ms."""
        conn = sqlite3.connect(db_path, timeout=60.0)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA busy_timeout=60000")
        flushed_updated, flushed_errors = 0, 0

        while not stop_flusher.is_set():
            time.sleep(0.05)
            with progress_lock:
                curr_updated, curr_errors = progress['updated'], progress['errors']

            delta_updated = curr_updated - flushed_updated
            delta_errors = curr_errors - flushed_errors

            if delta_updated > 0 or delta_errors > 0:
                conn.execute(
                    "UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
                    (delta_updated, delta_errors, job_id)
                )
                conn.commit()
                flushed_updated, flushed_errors = curr_updated, curr_errors

        # Final flush - get final counts and flush any remaining delta
        with progress_lock:
            final_updated, final_errors = progress['updated'], progress['errors']
        delta_updated = final_updated - flushed_updated
        delta_errors = final_errors - flushed_errors
        if delta_updated > 0 or delta_errors > 0:
            conn.execute(
                "UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
                (delta_updated, delta_errors, job_id)
            )
            conn.commit()

        # Store final counts for verification
        progress['_flushed_updated'] = final_updated
        progress['_flushed_errors'] = final_errors
        conn.close()

    def process_and_write(page_data):
        """Process one page and write immediately with thread-local connection."""
        page_id, uri, raw_html = page_data
        result = _process_single_page((page_id, uri, raw_html, trim_wrapper))
        page_id, markdown, description, keywords_json, error = result

        conn = get_conn()
        if error:
            with progress_lock:
                progress['errors'] += 1
        else:
            conn.execute(
                "UPDATE pages SET markdown = ?, description = ?, keywords = ? WHERE id = ?",
                (markdown, description, keywords_json, page_id)
            )
            conn.commit()
            with progress_lock:
                progress['updated'] += 1

    # Start background flusher
    flusher = threading.Thread(target=flusher_thread, daemon=True)
    flusher.start()

    # Fan out to thread workers
    with ThreadPoolExecutor(max_workers=threads_per_process) as thread_executor:
        list(thread_executor.map(process_and_write, chunk))

    # Stop flusher and wait for final flush
    stop_flusher.set()
    flusher.join(timeout=5.0)

    # Return verified counts (what was actually flushed to DB)
    return {
        'updated': progress.get('_flushed_updated', progress['updated']),
        'errors': progress.get('_flushed_errors', progress['errors']),
        'total': len(chunk)
    }


async def backfill_markdown(db_path: str, domain_filter: str = None, trim_wrapper: bool = False, quiet: bool = False, fresh: bool = False):
    """Re-process stored HTML to regenerate markdown with smart structure detection.

    ETL-style parallel processing:
    - Divides work into chunks (1 per CPU core)
    - Each process spawns 6 threads to burn through its chunk
    - Immediate DB writes with WAL mode (thread-safe)
    - Progress tracked via backfill_jobs table (atomic increments)

    Args:
        db_path: Path to SQLite database
        domain_filter: Only process pages matching this domain (e.g., 'example.com')
        trim_wrapper: Strip nav/header/footer/logo before conversion
        quiet: Disable progress bar
        fresh: Re-process all pages, even those that already have markdown
    """
    import aiosqlite
    import sqlite3
    import multiprocessing
    from concurrent.futures import ProcessPoolExecutor
    from datetime import datetime, timezone
    import time

    num_cpus = multiprocessing.cpu_count()
    threads_per_process = 6
    logger.info(f"ETL mode: {num_cpus} processes x {threads_per_process} threads = {num_cpus * threads_per_process} workers")

    if domain_filter:
        pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%"
        logger.info(f"Backfilling markdown for domain: {domain_filter}")
    else:
        pattern = None
        logger.info("Backfilling markdown for ALL pages")

    if trim_wrapper:
        logger.info("Trim wrapper enabled")

    # Setup: ensure tables/columns exist, create job record, get pages
    async with aiosqlite.connect(db_path) as db:
        await db.execute("PRAGMA journal_mode=WAL")

        # Ensure backfill_jobs table exists
        await db.execute("""
            CREATE TABLE IF NOT EXISTS backfill_jobs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                job_type TEXT NOT NULL,
                domain_filter TEXT,
                status TEXT DEFAULT 'running',
                total_records INTEGER DEFAULT 0,
                processed_records INTEGER DEFAULT 0,
                error_count INTEGER DEFAULT 0,
                started_at TEXT NOT NULL,
                completed_at TEXT
            )
        """)

        # Ensure page columns exist
        for col in ['description', 'keywords']:
            try:
                await db.execute(f"ALTER TABLE pages ADD COLUMN {col} TEXT")
            except Exception:
                pass

        # Get pages to process
        # fresh=True: re-process all pages
        # fresh=False: only pages without markdown
        base_condition = "raw_html IS NOT NULL"
        if not fresh:
            base_condition += " AND (markdown IS NULL OR markdown = '')"
            logger.info("Incremental mode: only processing pages without markdown")
        else:
            logger.info("Fresh mode: re-processing all pages")

        if pattern:
            cursor = await db.execute(
                f"SELECT id, uri, raw_html FROM pages WHERE {base_condition} AND uri LIKE ?",
                (pattern,)
            )
        else:
            cursor = await db.execute(
                f"SELECT id, uri, raw_html FROM pages WHERE {base_condition}"
            )
        rows = await cursor.fetchall()
        total = len(rows)
        logger.info(f"Found {total} pages to process")

        # Create job record (explicitly set processed_records/error_count to 0 for atomic increments)
        now = datetime.now(timezone.utc).isoformat()
        cursor = await db.execute(
            "INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, processed_records, error_count, started_at) VALUES (?, ?, 'running', ?, 0, 0, ?)",
            ('markdown', domain_filter, total, now)
        )
        job_id = cursor.lastrowid
        await db.commit()
        logger.info(f"Created backfill job {job_id}")

    # Convert to list of tuples
    all_pages = [(row[0], row[1], row[2]) for row in rows if row[2]]

    # Divide into exactly num_cpus chunks
    chunks = []
    chunk_size = len(all_pages) // num_cpus if num_cpus > 0 else len(all_pages)
    for i in range(num_cpus):
        start = i * chunk_size
        end = (i + 1) * chunk_size if i < num_cpus - 1 else len(all_pages)
        chunks.append(all_pages[start:end])
    logger.info(f"Split into {len(chunks)} chunks of ~{chunk_size} pages each")

    # Process chunks in parallel, poll job record for progress
    with ProcessPoolExecutor(max_workers=num_cpus) as executor:
        chunk_args = [(chunk, db_path, trim_wrapper, threads_per_process, job_id) for chunk in chunks]
        futures = [executor.submit(_process_chunk, args) for args in chunk_args]

        # Poll backfill_jobs table for progress
        # Must reconnect each poll to see WAL commits from other processes
        with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet, smoothing=0.1) as pbar:
            last_count = 0
            while True:
                # Fresh connection each poll to see latest WAL commits
                conn = sqlite3.connect(db_path, timeout=60.0)
                conn.execute("PRAGMA journal_mode=WAL")
                cursor = conn.execute(
                    "SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
                    (job_id,)
                )
                row = cursor.fetchone()
                current = row[0] if row and row[0] is not None else 0
                conn.close()

                if current > last_count:
                    pbar.update(current - last_count)
                    last_count = current
                if all(f.done() for f in futures):
                    # Final count with fresh connection
                    conn = sqlite3.connect(db_path, timeout=60.0)
                    cursor = conn.execute(
                        "SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
                        (job_id,)
                    )
                    row = cursor.fetchone()
                    current = row[0] if row and row[0] is not None else 0
                    conn.close()
                    if current > last_count:
                        pbar.update(current - last_count)
                    break
                time.sleep(0.1)

        # Collect results
        total_updated = 0
        total_errors = 0
        for f in futures:
            result = f.result()
            total_updated += result['updated']
            total_errors += result['errors']

    # Mark job completed
    async with aiosqlite.connect(db_path) as db:
        now = datetime.now(timezone.utc).isoformat()
        await db.execute(
            "UPDATE backfill_jobs SET status = 'completed', completed_at = ? WHERE id = ?",
            (now, job_id)
        )
        await db.commit()

    logger.info(f"Backfill complete: {total_updated} updated, {total_errors} errors")


async def backfill_screenshots(
    db_path: str,
    vault_path: str = "vault",
    domain_filter: str = None,
    delete_old: bool = True,
    fast_mode: bool = False,
    quiet: bool = False,
):
    """Re-capture screenshots as JPEG to replace old PNGs.

    Args:
        db_path: Path to SQLite database
        vault_path: Path to vault directory
        domain_filter: Only process pages matching this domain
        delete_old: Delete old PNG files after successful JPEG capture
        fast_mode: Skip delay between captures (for sites without robots.txt)
        quiet: Disable progress bar
    """
    from screenshot import ScreenshotCapture, ScreenshotConfig
    from filevault import AsyncVault, hash_to_path
    from database import Database, Media, MediaSource, Page
    from sqlalchemy import select, update
    from datetime import datetime, timezone
    import time

    # Load completed URIs from unified state file
    completed_uris = set()
    state_path = get_state_file_path(domain_filter) if domain_filter else None
    if state_path and state_path.exists():
        try:
            full_state = json.loads(state_path.read_text())
            completed_uris = set(full_state.get('backfill_screenshots', []))
            if completed_uris:
                logger.info(f"Resuming: {len(completed_uris)} already backfilled")
        except Exception as e:
            logger.warning(f"Could not load state: {e}")

    config = ScreenshotConfig(enabled=True, full_page=True)
    capture = ScreenshotCapture(config)
    vault = AsyncVault(vault_path)
    await vault.init()

    db = Database(db_path)
    await db.init()

    async with db.session() as session:
        # Find all screenshot records using ORM, excluding already-completed URIs
        if domain_filter:
            pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%"
            stmt = (
                select(MediaSource.page_uri, Media.md5_hash)
                .join(Media, MediaSource.md5_hash == Media.md5_hash)
                .where(Media.media_type == 'screenshot')
                .where(MediaSource.page_uri.like(pattern))
                .where(MediaSource.page_uri.notin_(completed_uris) if completed_uris else True)
                .distinct()
            )
            logger.info(f"Backfilling screenshots for domain: {domain_filter}")
        else:
            stmt = (
                select(MediaSource.page_uri, Media.md5_hash)
                .join(Media, MediaSource.md5_hash == Media.md5_hash)
                .where(Media.media_type == 'screenshot')
                .where(MediaSource.page_uri.notin_(completed_uris) if completed_uris else True)
                .distinct()
            )
            logger.info("Backfilling screenshots for ALL pages")

        result = await session.execute(stmt)
        rows = result.fetchall()
        to_process = len(rows)
        skipped_count = len(completed_uris)
        total = to_process + skipped_count
        logger.info(f"Found {to_process} to process ({skipped_count} already completed)")

        domain_last_fetched = {}  # Track last fetch time per domain
        crawl_delay = 2.0  # Default crawl delay in seconds

        # Concurrent workers in fast mode (2x CPU count since screenshot is I/O bound)
        import multiprocessing
        max_workers = multiprocessing.cpu_count() * 2 if fast_mode else 1
        semaphore = asyncio.Semaphore(max_workers)
        if fast_mode and max_workers > 1:
            logger.info(f"Fast mode: {max_workers} concurrent workers")

        pbar = tqdm(total=total, initial=skipped_count, desc="Screenshots", unit="pages", disable=quiet, smoothing=0.1)

        # Thread-safe counters
        stats_lock = asyncio.Lock()
        stats = {'captured': 0, 'failed': 0, 'bytes_saved': 0}

        async def capture_one(page_uri: str, old_hash: str):
            """Capture screenshot for one page. Returns (success, result_data) or (False, None)."""
            async with semaphore:
                try:
                    # Get content length for dynamic delay
                    async with db.session() as local_session:
                        from sqlalchemy import func as sqlfunc
                        result = await local_session.execute(
                            select(sqlfunc.length(Page.raw_html)).where(Page.uri == page_uri)
                        )
                        content_length = result.scalar() or 0

                    # Capture screenshot
                    result = await capture.capture(page_uri, content_length=content_length)
                    if not result:
                        return False, None
                    return True, (page_uri, old_hash, result)
                except Exception as e:
                    logger.debug(f"Capture error {page_uri}: {e}")
                    return False, None

        async def process_result(success, data):
            """Process a captured screenshot - store and update DB."""
            nonlocal completed_uris
            if not success or not data:
                async with stats_lock:
                    stats['failed'] += 1
                pbar.update(1)
                return

            page_uri, old_hash, result = data
            domain = Uri(page_uri).hostname.lower()

            try:
                chunks = result if isinstance(result, list) else [result]
                old_path = Path(vault_path) / hash_to_path(old_hash, ext='png')
                old_size = old_path.stat().st_size if old_path.exists() else 0

                # Store chunks
                total_new_size = 0
                for i, chunk in enumerate(chunks):
                    new_hash = chunk['md5_hash']
                    new_data = chunk['data']
                    new_ext = chunk.get('format', 'jpg')
                    total_new_size += len(new_data)

                    if not await vault.exists(new_hash):
                        await vault.store(new_hash, new_data, new_ext)

                    # Symlink
                    parsed = Uri(page_uri)
                    url_path = (parsed.path or '').lstrip('/') or 'index'
                    suffix = f'_{i}' if len(chunks) > 1 else ''
                    new_symlink = Path(vault_path) / 'linkpeek_vault' / domain / (url_path.replace('/', '_') + f'{suffix}.{new_ext}')
                    new_symlink.parent.mkdir(parents=True, exist_ok=True)
                    hash_vault_path = Path(vault_path) / hash_to_path(new_hash, ext=new_ext)
                    rel_path = os.path.relpath(hash_vault_path, new_symlink.parent)
                    if not new_symlink.exists():
                        new_symlink.symlink_to(rel_path)

                # Delete old
                if delete_old and old_path.exists():
                    old_path.unlink()
                    async with stats_lock:
                        stats['bytes_saved'] += old_size - total_new_size

                # Remove old symlink
                parsed = Uri(page_uri)
                url_path = (parsed.path or '').lstrip('/') or 'index'
                old_symlink = Path(vault_path) / 'linkpeek_vault' / domain / (url_path.replace('/', '_') + '.png')
                if old_symlink.exists():
                    old_symlink.unlink()

                # Update database
                new_hash = chunks[0]['md5_hash']
                new_ext = chunks[0].get('format', 'jpg')
                new_mime = chunks[0].get('mime_type', 'image/jpeg')
                new_size = total_new_size

                async with db.session() as local_session:
                    from sqlalchemy import delete as sqldelete
                    old_media_result = await local_session.execute(
                        select(Media).where(Media.md5_hash == old_hash)
                    )
                    old_media = old_media_result.scalar_one_or_none()

                    if old_media and old_hash != new_hash:
                        # Check if new hash already exists (duplicate screenshot from another page)
                        existing = await local_session.execute(
                            select(Media).where(Media.md5_hash == new_hash)
                        )
                        if not existing.scalar_one_or_none():
                            new_media = Media(
                                md5_hash=new_hash,
                                media_type='screenshot',
                                mime_type=new_mime,
                                file_size=new_size,
                                keywords=old_media.keywords,
                                alt_text=old_media.alt_text,
                                title=old_media.title,
                                first_seen_at=datetime.now(timezone.utc).isoformat(),
                                analysis_status=old_media.analysis_status,
                                analysis_result=old_media.analysis_result,
                                score=SCORE_SCREENSHOT,
                            )
                            local_session.add(new_media)
                            await local_session.flush()
                        # Update sources to point to new hash and delete old media
                        await local_session.execute(
                            update(MediaSource).where(MediaSource.md5_hash == old_hash).values(md5_hash=new_hash)
                        )
                        await local_session.execute(sqldelete(Media).where(Media.md5_hash == old_hash))
                    elif old_media:
                        old_media.mime_type = new_mime
                        old_media.file_size = new_size
                        old_media.first_seen_at = datetime.now(timezone.utc).isoformat()
                    await local_session.commit()

                async with stats_lock:
                    stats['captured'] += 1
                    completed_uris.add(page_uri)

                # Save state
                if state_path:
                    async with stats_lock:
                        try:
                            full_state = json.loads(state_path.read_text()) if state_path.exists() else {}
                        except Exception:
                            full_state = {}
                        full_state['backfill_screenshots'] = list(completed_uris)
                        state_path.write_text(json.dumps(full_state, indent=2))

                pbar.update(1)

            except Exception as e:
                err_msg = str(e).split('\n')[0][:60]
                logger.warning(f"Failed: {page_uri} - {err_msg}")
                async with stats_lock:
                    stats['failed'] += 1
                pbar.update(1)

        # Process - each worker flushes immediately when done
        if fast_mode and max_workers > 1:
            async def capture_and_process(row):
                """Capture and immediately process/flush to DB."""
                success, data = await capture_one(row.page_uri, row.md5_hash)
                await process_result(success, data)

            # Launch all tasks, they'll complete and flush independently
            tasks = [asyncio.create_task(capture_and_process(row)) for row in rows]
            # Wait for all to complete
            await asyncio.gather(*tasks)
        else:
            # Sequential mode with crawl delay
            for row in rows:
                if not fast_mode:
                    domain = Uri(row.page_uri).hostname.lower()
                    last_fetched = domain_last_fetched.get(domain, 0)
                    elapsed = time.time() - last_fetched
                    if elapsed < crawl_delay:
                        await asyncio.sleep(crawl_delay - elapsed)
                    domain_last_fetched[domain] = time.time()

                success, data = await capture_one(row.page_uri, row.md5_hash)
                await process_result(success, data)

        pbar.close()
        logger.info(f"Backfill complete: {stats['captured']} captured, {skipped_count} skipped, {stats['failed']} failed")
        if stats['bytes_saved'] > 0:
            logger.info(f"Space saved: {stats['bytes_saved'] / 1024 / 1024:.1f} MB")


async def main():
    parser = argparse.ArgumentParser(
        description="neopig - Neo Python Image Grabber",
        epilog="Full-domain async media crawler with deduplication"
    )

    parser.add_argument(
        "targets",
        nargs="*",
        help="Target URI(s) to crawl (e.g., https://example.com https://other.com)"
    )

    parser.add_argument(
        "-k", "--keywords",
        nargs="*",
        default=[],
        help="Keywords to tag media with (e.g., -k 'rick' 'morty')"
    )

    parser.add_argument(
        "-m", "--mode",
        choices=["text", "images", "videos", "media", "all"],
        default="images",
        help="Crawl mode (default: images)"
    )

    parser.add_argument(
        "-d", "--depth",
        type=int,
        default=-1,
        help="Crawl depth: 0=single page, 1=page+links, -1=unlimited (default: -1)"
    )

    parser.add_argument(
        "-p", "--max-pages",
        type=int,
        default=-1,
        help="Maximum pages to crawl (-1 = unlimited, default: -1)"
    )

    parser.add_argument(
        "--db",
        default="data/neopig.db",
        help="Database path (default: data/neopig.db)"
    )

    parser.add_argument(
        "--vault",
        default="data/vault",
        help="Vault storage path (default: data/vault)"
    )

    parser.add_argument(
        "--no-download",
        action="store_true",
        help="Don't download media, just index URLs"
    )

    parser.add_argument(
        "-v", "--verbose",
        action="store_true",
        help="Verbose output"
    )

    # Screenshot options (on by default)
    parser.add_argument(
        "--no-screenshot",
        action="store_true",
        help="Disable page screenshots"
    )

    parser.add_argument(
        "--screenshot-width",
        type=int,
        default=1280,
        help="Screenshot viewport width in pixels (default: 1280)"
    )

    parser.add_argument(
        "--screenshot-height",
        type=int,
        default=1024,
        help="Screenshot viewport height in pixels (default: 1024)"
    )

    # Hydra mode - feed/sitemap discovery
    parser.add_argument(
        "--hydra",
        action="store_true",
        help="Parse RSS/Atom/Sitemap feeds for fast URL discovery"
    )

    parser.add_argument(
        "--screenshot-delay",
        type=int,
        default=1000,
        help="Delay after page load in ms (default: 1000)"
    )

    parser.add_argument(
        "--screenshot-engine",
        type=str,
        default=None,
        help="Screenshot engine: wkhtmltoimage, cutycapt, playwright-webkit, etc. (default: auto-detect lightest)"
    )

    parser.add_argument(
        "--list-engines",
        action="store_true",
        help="List available screenshot engines and exit"
    )

    parser.add_argument(
        "--fresh",
        action="store_true",
        help="Start fresh (rotates state file; backfills preserve crawl state)"
    )

    parser.add_argument(
        "--fast",
        action="store_true",
        help="Fast mode: no crawl delay (use for sites without robots.txt)"
    )

    parser.add_argument(
        "--backfill-markdown",
        metavar="DOMAIN",
        help="Re-process stored HTML for DOMAIN to regenerate markdown with absolute URLs (e.g., example.com)"
    )

    parser.add_argument(
        "--trim-wrapper",
        action="store_true",
        help="With --backfill-markdown: strip nav/header/footer/logo before conversion"
    )

    parser.add_argument(
        "-q", "--quiet",
        action="store_true",
        help="Disable progress bars"
    )

    parser.add_argument(
        "--backfill-screenshots",
        metavar="DOMAIN",
        help="Re-capture screenshots as JPEG for DOMAIN (e.g., example.com)"
    )

    parser.add_argument(
        "--keep-old-screenshots",
        action="store_true",
        help="With --backfill-screenshots: keep old PNG files instead of deleting them"
    )

    parser.add_argument(
        "--serve",
        action="store_true",
        help="Start SERP server to watch crawl live"
    )

    parser.add_argument(
        "--port",
        type=int,
        default=31337,
        help="Port for SERP server (default: 31337)"
    )

    parser.add_argument(
        "--host",
        default="0.0.0.0",
        help="Host for SERP server (default: 0.0.0.0)"
    )

    parser.add_argument(
        "--archive",
        action="store_true",
        help="Package crawl results into a distributable tar.gz archive"
    )

    parser.add_argument(
        "--package-only",
        action="store_true",
        help="With --archive: skip crawling, just package existing data from vault"
    )

    parser.add_argument(
        "-o", "--output",
        default=".",
        help="Output directory for archive tar.gz (default: current directory)"
    )

    parser.add_argument(
        "--upgrade-neopig",
        metavar="TARBALL",
        help="Upgrade neopig inside an existing archive"
    )

    parser.add_argument(
        "--purge-job",
        type=int,
        metavar="JOB_ID",
        help="Purge a crawl job and ALL associated data (media, screenshots, pages, state)"
    )

    args = parser.parse_args()

    # Validate depth - max 18 before suggesting unlimited
    if args.depth > 18:
        print(f"Error: depth {args.depth} is too high. Use --depth -1 for unlimited crawling.")
        return

    # Create data directory if needed
    from pathlib import Path
    db_dir = Path(args.db).parent
    if db_dir and str(db_dir) != '.':
        db_dir.mkdir(parents=True, exist_ok=True)

    # Setup logging to work with tqdm progress bars
    setup_logging(level=logging.DEBUG if args.verbose else logging.INFO)

    # Handle --list-engines
    if args.list_engines:
        from screenshot import list_available_engines
        engines = await list_available_engines()
        print("Available screenshot engines:")
        for e in engines:
            print(f"  {e['name']}: {e['description']}")
        print("\nPreference order: wkhtmltoimage > cutycapt > playwright-webkit > playwright")
        print("Install lightweight: apt install wkhtmltopdf  OR  apt install cutycapt")
        return

    # Handle --upgrade-neopig
    if args.upgrade_neopig:
        from archive import upgrade_neopig_in_archive
        archive_path = Path(args.upgrade_neopig)
        output_dir = Path(args.output) if args.output != "." else None
        logger.info(f"Upgrading neopig in {archive_path}...")
        upgrade_neopig_in_archive(archive_path, output_dir=output_dir)
        logger.info(f"Done! Archive updated: {archive_path}")
        return

    # Handle --purge-job
    if args.purge_job:
        pig = NeoPig(db_path=args.db, vault_path=args.vault)
        await pig.init()
        job_id = args.purge_job

        # Get job info first for confirmation
        job = await pig.db.get_crawl_job(job_id)
        if not job:
            print(f"Error: Job {job_id} not found")
            return

        print(f"Purging job {job_id}: {job['target_uri']}")
        print(f"  Status: {job['status']}")
        print(f"  Started: {job['started_at']}")

        result = await pig.purge_job(job_id)
        if result['deleted']:
            print(f"\nPurged successfully:")
            print(f"  Media files deleted: {result['media_files_deleted']}")
            print(f"  Screenshots deleted: {result['screenshots_deleted']}")
            print(f"  Pages deleted: {result['pages_deleted']}")
            print(f"  State files deleted: {result['state_files_deleted']}")
        else:
            print(f"Error: Failed to purge job {job_id}")
        return

    # Handle --backfill-markdown
    if args.backfill_markdown:
        serp_process = None
        if args.serve:
            import subprocess, sys
            serp_script = Path(__file__).parent / 'serp.py'
            if serp_script.exists():
                serp_cmd = [
                    sys.executable, str(serp_script),
                    '--host', args.host,
                    '--port', str(args.port),
                    '--db', args.db,
                    '--vault', args.vault,
                ]
                serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                logger.info(f"SERP server started at http://{args.host}:{args.port}")

        await backfill_markdown(args.db, domain_filter=args.backfill_markdown, trim_wrapper=args.trim_wrapper, quiet=args.quiet, fresh=args.fresh)

        if serp_process:
            logger.info(f"Backfill complete. SERP server still running at http://localhost:{args.port}")
            logger.info("Press Ctrl+C to stop...")
            try:
                serp_process.wait()
            except KeyboardInterrupt:
                serp_process.terminate()
        return

    # Handle --backfill-screenshots
    if args.backfill_screenshots:
        # Handle --fresh: rotate state but preserve crawl data
        if args.fresh:
            state_path = get_state_file_path(args.backfill_screenshots)
            rotate_state_file(state_path, preserve_keys=['crawl'])

        serp_process = None
        if args.serve:
            import subprocess, sys
            serp_script = Path(__file__).parent / 'serp.py'
            if serp_script.exists():
                serp_cmd = [
                    sys.executable, str(serp_script),
                    '--host', args.host,
                    '--port', str(args.port),
                    '--db', args.db,
                    '--vault', args.vault,
                ]
                serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                logger.info(f"SERP server started at http://{args.host}:{args.port}")
        await backfill_screenshots(
            args.db,
            vault_path=args.vault,
            domain_filter=args.backfill_screenshots,
            delete_old=not args.keep_old_screenshots,
            fast_mode=args.fast,
            quiet=args.quiet,
        )

        if serp_process:
            logger.info(f"Backfill complete. SERP server still running at http://localhost:{args.port}")
            logger.info("Press Ctrl+C to stop...")
            try:
                serp_process.wait()
            except KeyboardInterrupt:
                serp_process.terminate()
        return

    # Handle --serve alone (just run the server)
    if args.serve and not args.targets:
        import subprocess
        import sys
        import signal
        serp_script = Path(__file__).parent / 'serp.py'
        if serp_script.exists():
            serp_cmd = [
                sys.executable, str(serp_script),
                '--host', args.host,
                '--port', str(args.port),
                '--db', args.db,
                '--vault', args.vault,
            ]
            logger.info(f"Starting SERP server at http://{args.host}:{args.port}")
            try:
                proc = subprocess.Popen(serp_cmd)
                proc.wait()  # Block until server exits
            except KeyboardInterrupt:
                proc.terminate()
                proc.wait(timeout=5)
        else:
            logger.error("serp.py not found")
        return

    # Handle --archive: use SiteArchiver from archive.py
    if args.archive:
        if not args.targets:
            parser.error("URL required for --archive")

        from archive import SiteArchiver
        screenshot_config = ScreenshotConfig(
            enabled=not args.no_screenshot,
            width=args.screenshot_width,
            height=args.screenshot_height,
            delay=args.screenshot_delay,
            engine=args.screenshot_engine,
            full_page=True,
        )

        archiver = SiteArchiver(
            output_dir=args.output,
            include_screenshots=not args.no_screenshot,
            include_markdown=True,
            screenshot_config=screenshot_config,
            fast_mode=args.fast,
            trim_wrapper=args.trim_wrapper,
            show_progress=not args.quiet,
            fresh_start=args.fresh,
        )

        # Start SERP server if requested
        serp_process = None
        if args.serve:
            import subprocess
            serp_script = Path(__file__).parent / 'serp.py'
            if serp_script.exists():
                serp_cmd = [
                    sys.executable, str(serp_script),
                    '--host', args.host,
                    '--port', str(args.port),
                    '--db', args.db,
                    '--vault', args.vault,
                ]
                serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                logger.info(f"SERP server started at http://{args.host}:{args.port}")

        try:
            archive_path = await archiver.archive(
                target_url=args.targets[0],
                depth=args.depth,
                max_pages=args.max_pages,
                db_path=args.db,
                vault_path=args.vault,
                package_only=args.package_only,
            )
            print(f"\nArchive created: {archive_path}")
            print(f"Extract with: tar -xzf {archive_path.name}")
        finally:
            if serp_process:
                serp_process.terminate()
        return

    # Require targets for crawling
    if not args.targets:
        parser.error("targets required (use --list-engines to see available screenshot engines)")

    # Map mode string to enum
    mode_map = {
        "text": CrawlMode.TEXT,
        "images": CrawlMode.IMAGES,
        "videos": CrawlMode.VIDEOS,
        "media": CrawlMode.MEDIA,
        "all": CrawlMode.ALL,
    }
    mode = mode_map[args.mode]

    # Create screenshot config from args
    screenshot_config = ScreenshotConfig(
        enabled=not args.no_screenshot,
        width=args.screenshot_width,
        height=args.screenshot_height,
        delay=args.screenshot_delay,
        engine=args.screenshot_engine,
        full_page=True,  # Always capture full page height
    )

    if not args.no_screenshot:
        engine_info = f", engine={args.screenshot_engine}" if args.screenshot_engine else " (auto-detect)"
        logger.info(f"Screenshots enabled: {screenshot_config.width}x{screenshot_config.height}, delay={screenshot_config.delay}ms{engine_info}")

    # Initialize and run
    pig = NeoPig(
        db_path=args.db,
        vault_path=args.vault,
        screenshot_config=screenshot_config,
        fast_mode=args.fast,
        trim_wrapper=args.trim_wrapper,
    )
    await pig.init()

    if args.fast:
        logger.info("Fast mode: no crawl delay (ignoring robots.txt)")

    # Handle --fresh: clear state files for all targets
    if args.fresh:
        for target in args.targets:
            pig._clear_state(target)
        logger.info("Starting fresh (state files cleared)")
    else:
        # Load previously crawled media/screenshots from DB (source of truth)
        crawled_media = await pig.db.get_crawled_media_uris()
        crawled_screenshots = await pig.db.get_crawled_screenshot_uris()
        if crawled_media or crawled_screenshots:
            logger.info(f"Resuming: {len(crawled_media)} media, {len(crawled_screenshots)} screenshots in database")
            pig.seen_media = crawled_media
            pig.seen_screenshots = crawled_screenshots

    # Start SERP server if requested
    serp_process = None
    if args.serve:
        import subprocess
        import sys
        serp_script = Path(__file__).parent / 'serp.py'
        if serp_script.exists():
            serp_cmd = [
                sys.executable, str(serp_script),
                '--host', args.host,
                '--port', str(args.port),
                '--db', args.db,
                '--vault', args.vault,
            ]
            # preexec_fn sets child to die when parent dies (Linux)
            def _set_pdeathsig():
                try:
                    import ctypes
                    libc = ctypes.CDLL("libc.so.6", use_errno=True)
                    PR_SET_PDEATHSIG = 1
                    libc.prctl(PR_SET_PDEATHSIG, signal.SIGTERM)
                except:
                    pass  # Not Linux or prctl unavailable

            serp_process = subprocess.Popen(
                serp_cmd,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                preexec_fn=_set_pdeathsig,
            )
            _CHILD_PROCESSES.append(serp_process)  # Track for cleanup
            logger.info(f"SERP server started at http://{args.host}:{args.port}")
            logger.info("Watch the crawl live - pages appear as they're indexed!")
        else:
            logger.warning("serp.py not found - --serve disabled")

    try:
        # Crawl all targets concurrently
        async def crawl_target(target: str):
            logger.info(f"=== Starting crawl: {target} ===")
            return await pig.crawl(
                target_uri=target,
                keywords=args.keywords,
                mode=mode,
                depth=args.depth,
                max_pages=args.max_pages,
                download_media=not args.no_download,
                hydra=args.hydra,
            )

        await asyncio.gather(*[crawl_target(t) for t in args.targets])

        # Backfill missing screenshots for ALL pages in DB (not just crawled domain)
        # Part of crawl, so don't create separate job
        if pig.screenshot_config.enabled:
            await pig.backfill_missing_screenshots(domain=None, fast_mode=args.fast, quiet=args.quiet, create_job=False)

        # Keep SERP server running after crawl
        if serp_process:
            logger.info(f"Crawl complete. SERP server still running at http://localhost:{args.port}")
            logger.info("Press Ctrl+C to stop...")
            try:
                serp_process.wait()
            except KeyboardInterrupt:
                pass
    finally:
        # Cleanup SERP server
        if serp_process and serp_process.poll() is None:
            logger.info("Stopping SERP server...")
            serp_process.terminate()
            try:
                serp_process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                serp_process.kill()


if __name__ == "__main__":
    asyncio.run(main())
