scripts/common_crawl_import.py

#!/usr/bin/env python3
"""
Eesha Search - Common Crawl Data Import Script
================================================
Imports high-quality web pages into our OpenSearch index using the Tranco
top-1M domains list as a seed source.

Strategy:
1. Download Tranco top-1M domains list (~6MB zip)
2. Select top N domains (configurable, default 5000)
3. For each domain, crawl the homepage + linked pages (depth 1)
4. Bulk index to OpenSearch in batches of 500
5. Track inlink counts as a "PageRank-lite" signal
6. Resumable via state file, respectful crawl delays

Common Crawl's full WAT/WARC data is petabyte-scale — impossible to import
entirely. This script takes the practical approach of crawling the most
important domains directly, yielding ~50K high-quality pages.

Uses ONLY Python stdlib + urllib (no pip install needed).
"""

import argparse
import gzip
import hashlib
import io
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from collections import defaultdict
from datetime import datetime
from html.parser import HTMLParser

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

OPENSEARCH_INDEX = os.environ.get('OPENSEARCH_INDEX', 'nutch')
USER_AGENT = 'EeshaSearch/1.0 (Eesha Browser Search; +https://eesha.search)'
TRANCO_URL = 'https://tranco-list.eu/top-1m.csv.zip'
STATE_FILE = os.environ.get('COMMON_CRAWL_STATE', 'common_crawl_state.json')
BULK_BATCH_SIZE = 500
CRAWL_DELAY = 1.0  # seconds between requests to same domain
REQUEST_TIMEOUT = 15  # seconds
MAX_CONTENT_LENGTH = 2 * 1024 * 1024  # 2MB max page size
MAX_LINKS_PER_PAGE = 100
MAX_TEXT_LENGTH = 10000

# High-value bonus seed domains — always include these even if not in Tranco top N
BONUS_SEED_DOMAINS = [
    # Major news
    'bbc.com', 'reuters.com', 'apnews.com', 'aljazeera.com',
    'nytimes.com', 'theguardian.com',
    # Reference
    'wikipedia.org', 'britannica.com', 'dictionary.com', 'wiktionary.org',
    # Tech
    'github.com', 'stackoverflow.com', 'arstechnica.com', 'techcrunch.com',
    # Science
    'nature.com', 'scientificamerican.com', 'nasa.gov', 'arxiv.org',
    # Education
    'khanacademy.org', 'mit.edu', 'coursera.org', 'edx.org',
]

# Domains to skip (login walls, social media, low-content)
SKIP_DOMAINS = {
    'facebook.com', 'instagram.com', 'twitter.com', 'x.com',
    'tiktok.com', 'linkedin.com', 'pinterest.com', 'reddit.com',
    'snapchat.com', 'whatsapp.com', 'telegram.org', 'discord.com',
    'paypal.com', 'bankofamerica.com', 'chase.com', 'wellsfargo.com',
    'netflix.com', 'hulu.com', 'primevideo.com', 'disneyplus.com',
    'spotify.com', 'apple.com', 'google.com', 'youtube.com',
    'amazon.com', 'amazon.co.uk', 'amazon.de', 'amazon.co.jp',
    'ebay.com', 'etsy.com', 'aliexpress.com',
    'login.microsoftonline.com', 'accounts.google.com',
}

# File extensions to skip (non-HTML content)
SKIP_EXTENSIONS = {
    '.pdf', '.zip', '.tar', '.gz', '.bz2', '.7z', '.rar',
    '.mp3', '.mp4', '.avi', '.mov', '.wmv', '.flv', '.webm',
    '.jpg', '.jpeg', '.png', '.gif', '.svg', '.ico', '.webp',
    '.css', '.js', '.json', '.xml', '.rss', '.atom',
    '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx',
    '.exe', '.dmg', '.deb', '.rpm', '.apk',
}

# ---------------------------------------------------------------------------
# HTML Parser (reused from lightweight_crawler.py)
# ---------------------------------------------------------------------------

class PageParser(HTMLParser):
    """Simple HTML parser that extracts title, text, links, and meta tags."""

    def __init__(self):
        super().__init__()
        self.title = ''
        self.in_title = False
        self.text_parts = []
        self.links = []
        self.description = ''
        self.in_script = False
        self.in_style = False

    def handle_starttag(self, tag, attrs):
        tag_lower = tag.lower()
        attrs_dict = dict(attrs)

        if tag_lower == 'title':
            self.in_title = True
        elif tag_lower == 'script':
            self.in_script = True
        elif tag_lower == 'style':
            self.in_style = True
        elif tag_lower == 'a' and 'href' in attrs_dict:
            href = attrs_dict['href']
            if href and not href.startswith(('javascript:', 'mailto:', 'tel:', '#')):
                self.links.append(href)
        elif tag_lower == 'meta':
            name = attrs_dict.get('name', attrs_dict.get('property', '')).lower()
            if name in ('description', 'og:description'):
                content = attrs_dict.get('content', '')
                if content and not self.description:
                    self.description = content

    def handle_endtag(self, tag):
        tag_lower = tag.lower()
        if tag_lower == 'title':
            self.in_title = False
        elif tag_lower == 'script':
            self.in_script = False
        elif tag_lower == 'style':
            self.in_style = False

    def handle_data(self, data):
        if self.in_script or self.in_style:
            return
        if self.in_title:
            self.title += data
        else:
            stripped = data.strip()
            if stripped:
                self.text_parts.append(stripped)

    def get_text(self, max_length=MAX_TEXT_LENGTH):
        text = ' '.join(self.text_parts)
        # Collapse multiple spaces
        text = re.sub(r'\s+', ' ', text).strip()
        return text[:max_length]


# ---------------------------------------------------------------------------
# Utility Functions
# ---------------------------------------------------------------------------

def is_valid_url(url):
    """Check if URL is crawlable (HTTP/HTTPS, non-skipped domain, non-binary)."""
    try:
        parsed = urllib.parse.urlparse(url)
        if parsed.scheme not in ('http', 'https'):
            return False
        if not parsed.hostname:
            return False
        # Check skip domains (exact or subdomain)
        hostname_lower = parsed.hostname.lower()
        for skip in SKIP_DOMAINS:
            if hostname_lower == skip or hostname_lower.endswith('.' + skip):
                return False
        # Check file extension
        path_lower = parsed.path.lower()
        for ext in SKIP_EXTENSIONS:
            if path_lower.endswith(ext):
                return False
        return True
    except Exception:
        return False


def resolve_url(base_url, href):
    """Resolve a relative URL against a base URL."""
    try:
        return urllib.parse.urljoin(base_url, href)
    except Exception:
        return None


def domain_from_url(url):
    """Extract the registered domain (e.g. 'en.wikipedia.org' → 'wikipedia.org')."""
    try:
        hostname = urllib.parse.urlparse(url).hostname or ''
        parts = hostname.split('.')
        # Simple heuristic: last two parts for normal domains, three for co.uk etc.
        if len(parts) >= 2:
            return '.'.join(parts[-2:])
        return hostname
    except Exception:
        return ''


def fetch_page(url, timeout=REQUEST_TIMEOUT):
    """Fetch a web page and return its HTML content. Returns None on failure."""
    try:
        headers = {
            'User-Agent': USER_AGENT,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'identity',  # avoid compressed responses for simplicity
        }
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            # Check content type
            content_type = resp.headers.get('Content-Type', '')
            if 'text/html' not in content_type and 'application/xhtml' not in content_type:
                return None
            # Check content length
            content_length = resp.headers.get('Content-Length')
            if content_length and int(content_length) > MAX_CONTENT_LENGTH:
                return None
            # Read with size limit
            data = resp.read(MAX_CONTENT_LENGTH + 1)
            if len(data) > MAX_CONTENT_LENGTH:
                return None
            return data.decode('utf-8', errors='replace')
    except Exception as e:
        return None


def parse_page(html, base_url):
    """Parse HTML and extract structured data including outgoing links."""
    parser = PageParser()
    try:
        parser.feed(html)
    except Exception:
        pass

    # Resolve relative links
    resolved_links = []
    for link in parser.links:
        resolved = resolve_url(base_url, link)
        if resolved and is_valid_url(resolved):
            resolved_links.append(resolved)

    return {
        'title': parser.title.strip() or 'Untitled',
        'content': parser.get_text(),
        'description': parser.description[:500] if parser.description else '',
        'url': base_url,
        'links': resolved_links[:MAX_LINKS_PER_PAGE],
        'host': urllib.parse.urlparse(base_url).hostname or '',
    }


# ---------------------------------------------------------------------------
# Tranco List Downloader
# ---------------------------------------------------------------------------

def download_tranco_list(target_path='top-1m.csv'):
    """
    Download and extract the Tranco top-1M domains list.
    Returns the path to the extracted CSV file.
    """
    if os.path.exists(target_path):
        print(f"[OK] Tranco list already exists: {target_path}")
        return target_path

    print(f"[...] Downloading Tranco top-1M list from {TRANCO_URL}...")
    try:
        req = urllib.request.Request(TRANCO_URL, headers={'User-Agent': USER_AGENT})
        with urllib.request.urlopen(req, timeout=120) as resp:
            zip_data = resp.read()
        print(f"[OK] Downloaded {len(zip_data):,} bytes")
    except Exception as e:
        print(f"[ERROR] Failed to download Tranco list: {e}")
        print("[INFO] Will use bonus seed domains only.")
        return None

    # Extract CSV from zip
    try:
        with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
            # Find the CSV file inside
            csv_names = [n for n in zf.namelist() if n.endswith('.csv')]
            if not csv_names:
                print(f"[ERROR] No CSV file found in zip. Contents: {zf.namelist()}")
                return None
            csv_name = csv_names[0]
            with zf.open(csv_name) as csv_file:
                csv_data = csv_file.read()
            with open(target_path, 'wb') as f:
                f.write(csv_data)
        print(f"[OK] Extracted {csv_name} → {target_path} ({len(csv_data):,} bytes)")
        return target_path
    except Exception as e:
        print(f"[ERROR] Failed to extract Tranco zip: {e}")
        return None


def load_domains(csv_path, max_domains):
    """
    Load top N domains from the Tranco CSV.
    CSV format: rank,domain
    Returns a list of domain strings.
    """
    domains = []
    if not csv_path or not os.path.exists(csv_path):
        print("[WARN] No Tranco CSV available, using bonus seeds only.")
        return list(BONUS_SEED_DOMAINS)

    try:
        with open(csv_path, 'r') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                parts = line.split(',', 1)
                if len(parts) < 2:
                    continue
                domain = parts[1].strip().lower()
                if not domain:
                    continue
                # Skip domains in our skip list
                if domain in SKIP_DOMAINS:
                    continue
                # Basic validation
                if '.' not in domain:
                    continue
                domains.append(domain)
                if len(domains) >= max_domains:
                    break
    except Exception as e:
        print(f"[ERROR] Failed to read Tranco CSV: {e}")
        return list(BONUS_SEED_DOMAINS)

    # Merge in bonus seeds at the front (deduplicated)
    seen = set(domains)
    bonus = [d for d in BONUS_SEED_DOMAINS if d not in seen]
    domains = bonus + domains

    print(f"[OK] Loaded {len(domains)} domains (incl. {len(bonus)} bonus seeds)")
    return domains


# ---------------------------------------------------------------------------
# OpenSearch Operations
# ---------------------------------------------------------------------------

def opensearch_request(url, data=None, method='GET', opensearch_url='http://localhost:9200'):
    """Make a request to OpenSearch. Returns response body or None on failure."""
    full_url = opensearch_url + url
    try:
        body = json.dumps(data).encode('utf-8') if data else None
        headers = {'Content-Type': 'application/json'} if body else {}
        req = urllib.request.Request(full_url, data=body, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        # Read error body for debugging
        try:
            err_body = e.read().decode('utf-8', errors='replace')
            if data and len(err_body) < 500:
                print(f"  [DEBUG] OpenSearch error: {err_body}")
        except Exception:
            pass
        return None
    except Exception as e:
        return None


def ensure_index(opensearch_url, index_name=OPENSEARCH_INDEX):
    """Create the OpenSearch index with proper mapping if it doesn't exist."""
    resp = opensearch_request(f'/{index_name}', method='HEAD', opensearch_url=opensearch_url)
    if resp is not None:
        print(f"[OK] Index '{index_name}' already exists")
        return True

    mapping = {
        "mappings": {
            "properties": {
                "title": {"type": "text", "analyzer": "english"},
                "url": {"type": "keyword"},
                "content": {"type": "text", "analyzer": "english"},
                "description": {"type": "text", "analyzer": "english"},
                "host": {"type": "keyword"},
                "inlink_count": {"type": "integer"},
                "crawlDate": {"type": "date", "format": "strict_date_optional_time||epoch_millis"},
                "title_suggest": {
                    "type": "text",
                    "analyzer": "edge_ngram_analyzer",
                    "search_analyzer": "standard"
                }
            }
        },
        "settings": {
            "analysis": {
                "analyzer": {
                    "edge_ngram_analyzer": {
                        "type": "custom",
                        "tokenizer": "standard",
                        "filter": ["lowercase", "edge_ngram_filter"]
                    }
                },
                "filter": {
                    "edge_ngram_filter": {
                        "type": "edge_ngram",
                        "min_gram": 2,
                        "max_gram": 20
                    }
                }
            }
        }
    }

    resp = opensearch_request(f'/{index_name}', data=mapping, method='PUT',
                              opensearch_url=opensearch_url)
    if resp:
        print(f"[OK] Created index '{index_name}' with mapping (incl. inlink_count, title_suggest)")
        return True
    else:
        print(f"[WARN] Could not create index (may already exist)")
        return False


def bulk_index(docs, opensearch_url, index_name=OPENSEARCH_INDEX):
    """
    Bulk index documents to OpenSearch using the /_bulk API.
    docs: list of dicts with at least 'url' field.
    Returns number of successfully indexed documents.
    """
    if not docs:
        return 0

    # Build NDJSON bulk body
    lines = []
    for doc in docs:
        doc_id = hashlib.md5(doc['url'].encode()).hexdigest()
        # Action line
        action = {"index": {"_index": index_name, "_id": doc_id}}
        lines.append(json.dumps(action, ensure_ascii=False))
        # Document line
        lines.append(json.dumps(doc, ensure_ascii=False))

    body = '\n'.join(lines) + '\n'
    body_bytes = body.encode('utf-8')

    full_url = f"{opensearch_url}/{index_name}/_bulk"
    try:
        req = urllib.request.Request(
            full_url,
            data=body_bytes,
            headers={'Content-Type': 'application/x-ndjson'},
            method='POST'
        )
        with urllib.request.urlopen(req, timeout=60) as resp:
            result = json.loads(resp.read().decode('utf-8'))

        # Count errors
        errors = result.get('errors', False)
        if errors:
            error_count = sum(1 for item in result.get('items', [])
                              if item.get('index', {}).get('error'))
            print(f"  [WARN] Bulk index: {error_count} errors out of {len(docs)} docs")
            return len(docs) - error_count
        return len(docs)
    except Exception as e:
        print(f"  [ERROR] Bulk index failed: {e}")
        return 0


def update_inlink_counts(inlinks, opensearch_url, index_name=OPENSEARCH_INDEX):
    """
    Update inlink_count for all URLs that have incoming links.
    Uses bulk update for efficiency.
    """
    if not inlinks:
        print("[INFO] No inlinks to update")
        return

    print(f"\n[...] Updating inlink counts for {len(inlinks):,} URLs...")

    # Process in batches of 500
    batch = []
    updated = 0
    for url, count in inlinks.items():
        doc_id = hashlib.md5(url.encode()).hexdigest()
        action = {"update": {"_index": index_name, "_id": doc_id}}
        doc = {"doc": {"inlink_count": count}}
        batch.append(json.dumps(action, ensure_ascii=False))
        batch.append(json.dumps(doc, ensure_ascii=False))

        if len(batch) >= BULK_BATCH_SIZE * 2:  # 2 lines per doc
            _send_bulk_update(batch, opensearch_url, index_name)
            updated += len(batch) // 2
            batch = []
            # Progress
            print(f"  Updated {updated:,}/{len(inlinks):,} inlink counts...")

    if batch:
        _send_bulk_update(batch, opensearch_url, index_name)
        updated += len(batch) // 2

    print(f"[OK] Updated inlink counts for {updated:,} URLs")


def _send_bulk_update(lines, opensearch_url, index_name=OPENSEARCH_INDEX):
    """Send a bulk update request to OpenSearch."""
    body = '\n'.join(lines) + '\n'
    body_bytes = body.encode('utf-8')

    full_url = f"{opensearch_url}/{index_name}/_bulk"
    try:
        req = urllib.request.Request(
            full_url,
            data=body_bytes,
            headers={'Content-Type': 'application/x-ndjson'},
            method='POST'
        )
        with urllib.request.urlopen(req, timeout=60) as resp:
            result = json.loads(resp.read().decode('utf-8'))
        if result.get('errors'):
            error_count = sum(1 for item in result.get('items', [])
                              if item.get('update', {}).get('error'))
            if error_count > 5:
                print(f"  [WARN] Bulk update: {error_count} errors")
    except Exception as e:
        print(f"  [ERROR] Bulk update failed: {e}")


# ---------------------------------------------------------------------------
# State Management (Resumability)
# ---------------------------------------------------------------------------

def load_state(state_path):
    """Load resume state from file. Returns state dict."""
    if not os.path.exists(state_path):
        return {'completed_domains': [], 'indexed_urls': [], 'pages_indexed': 0,
                'domains_processed': 0}
    try:
        with open(state_path, 'r') as f:
            state = json.load(f)
        print(f"[OK] Resuming from state: {state.get('domains_processed', 0)} domains, "
              f"{state.get('pages_indexed', 0)} pages indexed")
        return state
    except Exception as e:
        print(f"[WARN] Could not load state file: {e}. Starting fresh.")
        return {'completed_domains': [], 'indexed_urls': [], 'pages_indexed': 0,
                'domains_processed': 0}


def save_state(state, state_path):
    """Save resume state to file."""
    try:
        with open(state_path, 'w') as f:
            json.dump(state, f)
    except Exception as e:
        print(f"[WARN] Could not save state: {e}")


# ---------------------------------------------------------------------------
# Progress Reporting
# ---------------------------------------------------------------------------

class ProgressReporter:
    """Tracks and displays crawl progress with rate and ETA."""

    def __init__(self, total_domains, page_limit):
        self.total_domains = total_domains
        self.page_limit = page_limit
        self.domains_done = 0
        self.pages_indexed = 0
        self.pages_failed = 0
        self.start_time = time.time()
        self.last_report_time = self.start_time

    def report(self, domain=None, force=False):
        """Print progress if enough time has passed or force=True."""
        now = time.time()
        if not force and (now - self.last_report_time) < 5:
            return
        self.last_report_time = now

        elapsed = now - self.start_time
        if elapsed < 1:
            elapsed = 1
        rate = self.pages_indexed / elapsed

        # ETA calculation
        if self.page_limit:
            remaining_pages = max(0, self.page_limit - self.pages_indexed)
            eta_secs = remaining_pages / rate if rate > 0 else 0
        else:
            # Estimate based on domains remaining
            if self.domains_done > 0:
                avg_pages = self.pages_indexed / self.domains_done
                remaining_domains = self.total_domains - self.domains_done
                remaining_pages = remaining_domains * avg_pages
                eta_secs = remaining_pages / rate if rate > 0 else 0
            else:
                eta_secs = 0

        eta_str = self._format_duration(eta_secs)
        elapsed_str = self._format_duration(elapsed)

        domain_info = f" | Domain: {domain}" if domain else ""
        print(f"  [PROGRESS] Domains: {self.domains_done}/{self.total_domains} | "
              f"Pages: {self.pages_indexed}" +
              (f"/{self.page_limit}" if self.page_limit else "") +
              f" | Rate: {rate:.1f} pg/s | Elapsed: {elapsed_str} | ETA: {eta_str}"
              f"{domain_info}")

    def final_report(self):
        """Print final summary."""
        elapsed = time.time() - self.start_time
        if elapsed < 1:
            elapsed = 1
        rate = self.pages_indexed / elapsed
        elapsed_str = self._format_duration(elapsed)

        print(f"\n{'='*60}")
        print(f"  IMPORT COMPLETE")
        print(f"  Domains processed: {self.domains_done:,}")
        print(f"  Pages indexed:     {self.pages_indexed:,}")
        print(f"  Pages failed:      {self.pages_failed:,}")
        print(f"  Time elapsed:      {elapsed_str}")
        print(f"  Average rate:      {rate:.1f} pages/sec")
        print(f"{'='*60}")

    @staticmethod
    def _format_duration(seconds):
        """Format seconds into human-readable duration."""
        if seconds < 60:
            return f"{seconds:.0f}s"
        elif seconds < 3600:
            return f"{seconds/60:.1f}min"
        else:
            return f"{seconds/3600:.1f}h"


# ---------------------------------------------------------------------------
# Main Crawl Logic
# ---------------------------------------------------------------------------

def crawl_domain(domain, depth=1, max_pages_per_domain=15, crawl_delay=CRAWL_DELAY):
    """
    Crawl a single domain: homepage + linked pages at depth 1.
    Returns (list_of_docs, dict_of_outgoing_links).
    """
    # Construct homepage URL
    if domain.startswith(('http://', 'https://')):
        homepage = domain
    else:
        homepage = f'https://{domain}/'

    docs = []
    all_outgoing_links = []  # all links found on pages from this domain

    # Fetch homepage
    html = fetch_page(homepage)
    if html is None:
        # Try HTTP as fallback
        if homepage.startswith('https://'):
            homepage = homepage.replace('https://', 'http://', 1)
            html = fetch_page(homepage)
    if html is None:
        return docs, all_outgoing_links

    page_data = parse_page(html, homepage)
    page_data['crawlDate'] = datetime.utcnow().isoformat() + 'Z'
    page_data['title_suggest'] = page_data['title']
    page_data['inlink_count'] = 0  # will be updated later

    # Only index pages with meaningful content
    if len(page_data['content']) > 50:
        docs.append(page_data)
        all_outgoing_links.extend(page_data['links'])

    if depth < 1:
        return docs, all_outgoing_links

    # Crawl linked pages (depth 1)
    # Filter: only follow links on the same domain or subdomain
    base_domain = domain_from_url(homepage)
    seen_urls = {homepage}
    pages_crawled = 1

    for link in page_data['links'][:max_pages_per_domain]:
        if pages_crawled >= max_pages_per_domain:
            break
        if link in seen_urls:
            continue
        # Only follow same-domain links
        link_domain = domain_from_url(link)
        if link_domain != base_domain:
            continue
        if not is_valid_url(link):
            continue

        seen_urls.add(link)

        # Respect crawl delay
        time.sleep(crawl_delay)

        link_html = fetch_page(link)
        if link_html is None:
            continue

        link_data = parse_page(link_html, link)
        link_data['crawlDate'] = datetime.utcnow().isoformat() + 'Z'
        link_data['title_suggest'] = link_data['title']
        link_data['inlink_count'] = 0

        if len(link_data['content']) > 50:
            docs.append(link_data)
            all_outgoing_links.extend(link_data['links'])
            pages_crawled += 1

    return docs, all_outgoing_links


def run_import(domains, opensearch_url, page_limit, state_path,
               index_name=OPENSEARCH_INDEX, crawl_delay=CRAWL_DELAY):
    """
    Main import loop: crawl each domain, collect docs + inlinks,
    bulk index to OpenSearch, update inlink counts at the end.
    """
    # Load state for resumability
    state = load_state(state_path)
    completed_domains = set(state.get('completed_domains', []))
    indexed_urls = set(state.get('indexed_urls', []))
    total_indexed = state.get('pages_indexed', 0)

    # Filter out already-completed domains
    remaining_domains = [d for d in domains if d not in completed_domains]

    print(f"\n{'='*60}")
    print(f"  Eesha Search — Common Crawl Import")
    print(f"  Domains: {len(remaining_domains)} remaining "
          f"({len(completed_domains)} already done)")
    print(f"  Pages indexed so far: {total_indexed:,}")
    if page_limit:
        print(f"  Page limit: {page_limit:,}")
    print(f"  OpenSearch: {opensearch_url}")
    print(f"  Index: {index_name}")
    print(f"{'='*60}\n")

    # Initialize progress reporter
    progress = ProgressReporter(len(remaining_domains), page_limit)
    progress.pages_indexed = total_indexed

    # Inlink tracking: url → count of pages that link to it
    inlinks = defaultdict(int)

    # Bulk indexing buffer
    bulk_buffer = []
    last_save_time = time.time()

    for domain in remaining_domains:
        # Check page limit
        if page_limit and total_indexed >= page_limit:
            print(f"\n[INFO] Page limit reached ({page_limit:,}). Stopping.")
            break

        progress.report(domain=domain)

        # Crawl the domain
        try:
            docs, outgoing_links = crawl_domain(domain, depth=1, max_pages_per_domain=15,
                                                 crawl_delay=crawl_delay)
        except Exception as e:
            print(f"  [ERROR] Crawl failed for {domain}: {e}")
            docs, outgoing_links = [], []

        # Track inlinks from outgoing links
        for link in outgoing_links:
            if is_valid_url(link):
                inlinks[link] += 1

        # Add docs to bulk buffer
        for doc in docs:
            if doc['url'] in indexed_urls:
                continue
            if page_limit and total_indexed >= page_limit:
                break

            indexed_urls.add(doc['url'])
            bulk_buffer.append(doc)
            total_indexed += 1
            progress.pages_indexed = total_indexed

        # Flush bulk buffer when full
        if len(bulk_buffer) >= BULK_BATCH_SIZE:
            count = bulk_index(bulk_buffer, opensearch_url, index_name)
            print(f"  [BULK] Indexed {count}/{len(bulk_buffer)} pages")
            bulk_buffer = []

        # Update state
        completed_domains.add(domain)
        progress.domains_done += 1
        progress.pages_indexed = total_indexed

        # Save state periodically (every 30 seconds)
        now = time.time()
        if (now - last_save_time) >= 30:
            state = {
                'completed_domains': list(completed_domains),
                'indexed_urls': list(indexed_urls)[-50000:],  # keep last 50K to limit size
                'pages_indexed': total_indexed,
                'domains_processed': len(completed_domains),
                'last_updated': datetime.utcnow().isoformat() + 'Z',
            }
            save_state(state, state_path)
            last_save_time = now

        # Respect crawl delay between domains
        time.sleep(crawl_delay)

    # Flush remaining buffer
    if bulk_buffer:
        count = bulk_index(bulk_buffer, opensearch_url, index_name)
        print(f"  [BULK] Final batch: indexed {count}/{len(bulk_buffer)} pages")

    # Update inlink counts
    print(f"\n[INFO] Total inlinks tracked: {len(inlinks):,} unique URLs")
    update_inlink_counts(inlinks, opensearch_url, index_name)

    # Save final state
    state = {
        'completed_domains': list(completed_domains),
        'indexed_urls': list(indexed_urls)[-50000:],
        'pages_indexed': total_indexed,
        'domains_processed': len(completed_domains),
        'last_updated': datetime.utcnow().isoformat() + 'Z',
        'status': 'complete',
    }
    save_state(state, state_path)

    # Final report
    progress.final_report()

    return total_indexed


# ---------------------------------------------------------------------------
# CLI Entry Point
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description='Eesha Search — Common Crawl Data Import',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Import top 5000 domains (default)
  python common_crawl_import.py

  # Import top 1000 domains, limit to 10000 pages
  python common_crawl_import.py --domains 1000 --limit 10000

  # Custom OpenSearch URL
  python common_crawl_import.py --url http://opensearch:9200

  # Resume interrupted import
  python common_crawl_import.py --resume
"""
    )
    parser.add_argument('--domains', type=int, default=5000,
                        help='Number of top domains to import (default: 5000)')
    parser.add_argument('--limit', type=int, default=None,
                        help='Maximum number of pages to import (default: unlimited)')
    parser.add_argument('--url', type=str, default='http://localhost:9200',
                        help='OpenSearch URL (default: http://localhost:9200)')
    parser.add_argument('--index', type=str, default=OPENSEARCH_INDEX,
                        help=f'OpenSearch index name (default: {OPENSEARCH_INDEX})')
    parser.add_argument('--state-file', type=str, default=STATE_FILE,
                        help=f'State file path (default: {STATE_FILE})')
    parser.add_argument('--resume', action='store_true',
                        help='Resume from previous state file')
    parser.add_argument('--download-only', action='store_true',
                        help='Only download Tranco list, do not crawl')
    parser.add_argument('--dry-run', action='store_true',
                        help='Show what would be done without indexing')
    parser.add_argument('--delay', type=float, default=CRAWL_DELAY,
                        help=f'Delay between requests in seconds (default: {CRAWL_DELAY})')

    args = parser.parse_args()

    # Capture overrides from CLI args
    index_name = args.index
    crawl_delay = args.delay
    state_file = args.state_file

    opensearch_url = args.url.rstrip('/')
    csv_path = 'top-1m.csv'

    # Download Tranco list
    tranco_path = download_tranco_list(csv_path)

    if args.download_only:
        if tranco_path:
            print(f"[DONE] Tranco list downloaded to: {tranco_path}")
        else:
            print("[ERROR] Download failed")
            sys.exit(1)
        return

    # Load domains
    domains = load_domains(tranco_path, args.domains)

    if not domains:
        print("[ERROR] No domains to crawl. Check Tranco list download.")
        sys.exit(1)

    if args.dry_run:
        print(f"\n[DRY RUN] Would crawl {len(domains)} domains:")
        for i, d in enumerate(domains[:20]):
            print(f"  {i+1}. {d}")
        if len(domains) > 20:
            print(f"  ... and {len(domains) - 20} more")
        print(f"\nEstimated pages: ~{len(domains) * 10:,} (assuming ~10 pages/domain)")
        return

    # Ensure OpenSearch index exists
    ensure_index(opensearch_url, index_name)

    # Run the import
    if not args.resume:
        # Fresh start — remove old state if exists
        if os.path.exists(state_file):
            os.remove(state_file)
            print("[INFO] Removed old state file (fresh start)")

    total = run_import(domains, opensearch_url, args.limit, state_file,
                       index_name=index_name, crawl_delay=crawl_delay)
    print(f"\n[DONE] Import complete. Total pages indexed: {total:,}")


if __name__ == '__main__':
    main()