validate.py

"""
Phase 12 Validation — offline simulation using synthetic fetcher.

This replaces the network fetcher with a mock that simulates:
- 200 OK first crawl
- 304 Not Modified second crawl (same ETags)
- 200 OK with changed content (third crawl, one page updated)
- Interrupted + resumed crawl
"""

import asyncio
import logging
import sqlite3
import time
from dataclasses import replace
from pathlib import Path

from src.config import ScraperConfig, load_config
from src.crawler import CrawlEvent, FetchFailure, FetchResult, crawl_pending_urls
from src.frontier import Frontier
from src.indexer import PageIndexer
from src.scraper import Scraper
from src.storage import PageStorage
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Synthetic pages (URL-keyed)
# ---------------------------------------------------------------------------
SITE = {
    "http://crawltest.local/": {
        "body": "<html><body><a href='/page2'>Page 2</a><a href='/page3'>Page 3</a><p>Home content for testing incremental crawl pipeline. This is long enough text to pass the min_text_length check in the scraper configuration.</p></body></html>",
        "etag": '"etag-home-v1"',
        "lm": "Wed, 05 Aug 2026 12:00:00 GMT",
    },
    "http://crawltest.local/page2": {
        "body": "<html><body><p>Page two content for testing. This text body is also long enough to pass the min text length threshold for extraction configured in the scraper.</p></body></html>",
        "etag": '"etag-p2-v1"',
        "lm": "Wed, 05 Aug 2026 12:00:01 GMT",
    },
    "http://crawltest.local/page3": {
        "body": "<html><body><p>Page three content for testing. This text body is also long enough to pass the min text length threshold for extraction configured in the scraper.</p></body></html>",
        "etag": '"etag-p3-v1"',
        "lm": "Wed, 05 Aug 2026 12:00:02 GMT",
    },
}


class MockFetcher:
    """Synthetic fetcher that reads from SITE dict and honours ETags."""

    def __init__(self, site, user_agent="validator"):
        self.site = site
        self.user_agent = user_agent
        self._fetch_count: dict[str, int] = {}
        self._304_count: dict[str, int] = {}

    async def start(self):
        pass

    async def close(self):
        pass

    async def fetch(
        self, url: str, *, conditional_headers: dict[str, str] | None = None
    ) -> FetchResult:
        if url not in self.site:
            raise FetchFailure(
                category="http_error",
                url=url,
                status_code=404,
                message=f"Not found: {url}",
            )
        data = self.site[url]
        self._fetch_count[url] = self._fetch_count.get(url, 0) + 1

        if conditional_headers:
            inm = conditional_headers.get("If-None-Match")
            if inm and data.get("etag") == inm:
                logger.info("HTTP 304")
                logger.info("Skipping download")
                self._304_count[url] = self._304_count.get(url, 0) + 1
                return FetchResult(
                    original_url=url,
                    final_url=url,
                    http_status=304,
                    headers={
                        "etag": data.get("etag", ""),
                        "last-modified": data.get("lm", ""),
                    },
                    content_type="text/html",
                    encoding="utf-8",
                    html_text="",
                    fetched_at=_utc_now(),
                    is_not_modified=True,
                )

        headers = {"content-type": "text/html; charset=utf-8"}
        if "etag" in data:
            headers["etag"] = data["etag"]
        if "lm" in data:
            headers["last-modified"] = data["lm"]
        return FetchResult(
            original_url=url,
            final_url=url,
            http_status=200,
            headers=headers,
            content_type="text/html",
            encoding="utf-8",
            html_text=data["body"],
            fetched_at=_utc_now(),
        )

    @property
    def fetch_count(self):
        return dict(self._fetch_count)

    @property
    def not_modified_count(self):
        return dict(self._304_count)


def _utc_now():
    from datetime import UTC, datetime

    return datetime.now(UTC).replace(microsecond=0).isoformat()


async def run_crawl_synthetic(config, db_path, raw_dir, site_data, *, max_pages=50):
    """Synthetic crawl using MockFetcher."""
    frontier = Frontier(config, db_path)
    frontier.mark_done_to_pending()
    frontier.add("http://crawltest.local/", depth=0)

    fetcher = MockFetcher(site_data)
    scraper = Scraper(config)
    storage = PageStorage(db_path, raw_dir)

    events = []

    async def on_event(ev: CrawlEvent):
        events.append(ev)

    report = await crawl_pending_urls(
        frontier,
        fetcher,
        scraper,
        storage,
        max_pages=max_pages,
        progress_interval=100,
        on_event=on_event,
    )

    indexer = PageIndexer(config, db_path, db_path.parent.parent / "chroma_db")
    idx_report = indexer.index_incremental()

    return report, events, idx_report, fetcher


def show_db(db_path):
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    print("\n  Pages table:")
    for row in conn.execute(
        "SELECT url, content_hash, fetched_at, etag, last_modified FROM pages ORDER BY url"
    ).fetchall():
        print(f"    {row['url']}")
        print(
            f"      hash={row['content_hash'][:16]}...  etag={row['etag']}  lm={row['last_modified']}"
        )
    total = conn.execute("SELECT COUNT(*) FROM pages").fetchone()[0]
    print(f"  Total pages: {total}")
    conn.close()


async def main():
    project_root = Path(".")
    base_config = load_config(project_root / "config.yaml")
    config = replace(
        base_config,
        seeds=["http://crawltest.local/"],
        allowed_domains=["crawltest.local"],
        max_depth=1,
        max_pages=50,
        scraper=ScraperConfig(min_text_length=10),
    )

    db_path = project_root / "data" / "db.sqlite"
    raw_dir = project_root / "data" / "raw"

    # Clean slate for validation
    conn = sqlite3.connect(str(db_path))
    for tbl in ["pages", "frontier", "page_index_state"]:
        conn.execute(f"DROP TABLE IF EXISTS {tbl}")
    for tbl in [
        "pages_fts",
        "pages_fts_data",
        "pages_fts_idx",
        "pages_fts_docsize",
        "pages_fts_config",
    ]:
        conn.execute(f"DROP TABLE IF EXISTS {tbl}")
    for trg in ["pages_ai", "pages_ad", "pages_au"]:
        conn.execute(f"DROP TRIGGER IF EXISTS {trg}")
    conn.commit()
    conn.close()

    site = dict(SITE)

    # ============ FIRST CRAWL ============
    print("=" * 60)
    print("FIRST CRAWL (brand new, all pages are new)")
    print("=" * 60)
    t0 = time.perf_counter()
    report1, events1, idx1, fetcher1 = await run_crawl_synthetic(
        config, db_path, raw_dir, site
    )
    elapsed1 = time.perf_counter() - t0

    print(f"  Processed: {report1.processed_count}")
    print(f"  New pages: {report1.new_pages}")
    print(f"  Updated pages: {report1.updated_pages}")
    print(f"  Unchanged pages: {report1.unchanged_pages}")
    print(f"  Duplicates: {report1.duplicate_count}")
    print(f"  Failed: {report1.failed_count}")
    print(f"  Embedding updates: {idx1.embedded_pages}")
    print(f"  Fetch counts: {fetcher1.fetch_count}")
    print(f"  304 count: {fetcher1.not_modified_count}")
    print(f"  Elapsed: {elapsed1:.2f}s")
    event_kinds1 = [e.kind for e in events1]
    print(f"  Event kinds: {event_kinds1}")
    show_db(db_path)

    # ============ SECOND CRAWL (same content, should be 304s) ============
    print("\n" + "=" * 60)
    print("SECOND CRAWL (same content — expect 304 / unchanged)")
    print("=" * 60)
    t0 = time.perf_counter()
    report2, events2, idx2, fetcher2 = await run_crawl_synthetic(
        config, db_path, raw_dir, site
    )
    elapsed2 = time.perf_counter() - t0

    print(f"  Processed: {report2.processed_count}")
    print(f"  New pages: {report2.new_pages}")
    print(f"  Updated pages: {report2.updated_pages}")
    print(f"  Unchanged pages: {report2.unchanged_pages}")
    print(f"  Duplicates: {report2.duplicate_count}")
    print(f"  Embedding updates: {idx2.embedded_pages}")
    print(f"  Fetch counts: {fetcher2.fetch_count}")
    print(f"  304 count: {fetcher2.not_modified_count}")
    print(f"  Elapsed: {elapsed2:.2f}s")
    event_kinds2 = [e.kind for e in events2]
    print(f"  Event kinds: {event_kinds2}")
    show_db(db_path)

    # ============ THIRD CRAWL — simulated content change ============
    print("\n" + "=" * 60)
    print("THIRD CRAWL (home page content updated)")
    print("=" * 60)
    site_v3 = dict(site)
    site_v3["http://crawltest.local/"] = {
        "body": "<html><body><a href='/page2'>Page 2</a><a href='/page3'>Page 3</a><p>Home content UPDATED for incremental crawl validation. This change triggers re-indexing and embedding regeneration for the home page only.</p></body></html>",
        "etag": '"etag-home-v2"',
        "lm": "Wed, 06 Aug 2026 09:00:00 GMT",
    }

    t0 = time.perf_counter()
    report3, events3, idx3, fetcher3 = await run_crawl_synthetic(
        config, db_path, raw_dir, site_v3
    )
    elapsed3 = time.perf_counter() - t0

    print(f"  Processed: {report3.processed_count}")
    print(f"  New pages: {report3.new_pages}")
    print(f"  Updated pages: {report3.updated_pages}")
    print(f"  Unchanged pages: {report3.unchanged_pages}")
    print(f"  Duplicates: {report3.duplicate_count}")
    print(f"  Embedding updates: {idx3.embedded_pages}")
    print(f"  Fetch counts: {fetcher3.fetch_count}")
    print(f"  304 count: {fetcher3.not_modified_count}")
    print(f"  Elapsed: {elapsed3:.2f}s")
    event_kinds3 = [e.kind for e in events3]
    print(f"  Event kinds: {event_kinds3}")
    show_db(db_path)

    # ============ INTERRUPTED + RESUMED CRAWL ============
    print("\n" + "=" * 60)
    print("INTERRUPTED + RESUMED CRAWL")
    print("=" * 60)
    site_v4 = dict(site_v3)

    # Interrupt: reset to pending, limit to 1 page
    frontier = Frontier(config, db_path)
    frontier.mark_done_to_pending()
    frontier.add("http://crawltest.local/", depth=0)

    fetcher_int = MockFetcher(site_v4)
    scraper_int = Scraper(config)
    storage_int = PageStorage(db_path, raw_dir)

    report_int = await crawl_pending_urls(
        frontier,
        fetcher_int,
        scraper_int,
        storage_int,
        max_pages=1,
        progress_interval=100,
    )
    statuses_int = frontier.count_statuses()
    print("  After interrupt (max_pages=1):")
    print(f"    Processed: {report_int.processed_count}")
    print(f"    Frontier pending: {statuses_int['pending']}")
    print(f"    Frontier done: {statuses_int['done']}")
    print(f"    Frontier skipped: {statuses_int['skipped']}")
    print(f"    Unchanged: {report_int.unchanged_pages}, New: {report_int.new_pages}")

    # Resume: continue without resetting frontier
    fetcher_res = MockFetcher(site_v4)
    scraper_res = Scraper(config)
    storage_res = PageStorage(db_path, raw_dir)

    t0 = time.perf_counter()
    report_res = await crawl_pending_urls(
        frontier,
        fetcher_res,
        scraper_res,
        storage_res,
        max_pages=50,
        progress_interval=100,
    )
    elapsed_res = time.perf_counter() - t0

    statuses_res = frontier.count_statuses()
    print("  After resume:")
    print(f"    Processed: {report_res.processed_count}")
    print(f"    New: {report_res.new_pages}")
    print(f"    Updated: {report_res.updated_pages}")
    print(f"    Unchanged: {report_res.unchanged_pages}")
    print(f"    Frontier pending: {statuses_res['pending']}")
    print(f"    Frontier done: {statuses_res['done']}")
    print(f"    Elapsed: {elapsed_res:.2f}s")

    # ============ PERFORMANCE COMPARISON ============
    print("\n" + "=" * 60)
    print("PERFORMANCE COMPARISON")
    print("=" * 60)
    print(
        f"  First crawl:  {elapsed1:.2f}s  new={report1.new_pages}  embedding_updates={idx1.embedded_pages}"
    )
    print(
        f"  Second crawl: {elapsed2:.2f}s  unchanged={report2.unchanged_pages}  embedding_updates={idx2.embedded_pages}  (304s: {sum(fetcher2.not_modified_count.values())})"
    )
    print(
        f"  Third crawl:  {elapsed3:.2f}s  updated={report3.updated_pages}  unchanged={report3.unchanged_pages}  embedding_updates={idx3.embedded_pages}"
    )

    # ============ DATABASE EVIDENCE ============
    print("\n" + "=" * 60)
    print("DATABASE EVIDENCE")
    print("=" * 60)
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    for row in conn.execute(
        "SELECT url, content_hash, fetched_at, etag, last_modified FROM pages ORDER BY url"
    ).fetchall():
        print(f"  URL: {row['url']}")
        print(f"    content_hash:  {row['content_hash']}")
        print(f"    fetched_at:    {row['fetched_at']}")
        print(f"    etag:          {row['etag']}")
        print(f"    last_modified: {row['last_modified']}")

    pis_rows = conn.execute(
        "SELECT url, content_hash, indexed_at FROM page_index_state ORDER BY url"
    ).fetchall()
    print(f"\n  page_index_state rows: {len(pis_rows)}")
    for row in pis_rows:
        print(f"    {row['url']}")
        print(
            f"      hash={row['content_hash'][:16]}...  indexed_at={row['indexed_at']}"
        )
    conn.close()

    # ============ ASSERTIONS ============
    print("\n" + "=" * 60)
    print("ASSERTIONS")
    print("=" * 60)

    checks = []

    def check(label, condition):
        status = "PASS" if condition else "FAIL"
        checks.append((label, status))
        print(f"  [{status}] {label}")

    check(
        "First crawl: all pages new",
        report1.new_pages == 3 and report1.unchanged_pages == 0,
    )
    check("First crawl: embeddings generated", idx1.embedded_pages == 3)
    check("First crawl: no 304s", sum(fetcher1.not_modified_count.values()) == 0)
    check(
        "Second crawl: all pages unchanged (304)",
        report2.unchanged_pages == 3
        and report2.new_pages == 0
        and report2.updated_pages == 0,
    )
    check("Second crawl: no embedding updates", idx2.embedded_pages == 0)
    check(
        "Second crawl: all 304 responses",
        sum(fetcher2.not_modified_count.values()) == 3,
    )
    check("Third crawl: exactly 1 page updated", report3.updated_pages == 1)
    check("Third crawl: 2 pages unchanged (304)", report3.unchanged_pages == 2)
    check("Third crawl: only 1 embedding update", idx3.embedded_pages == 1)
    check("Resume: first leg processes 1 page", report_int.processed_count == 1)
    check(
        "Resume: remaining pages pending after interrupt", statuses_int["pending"] > 0
    )
    check(
        "Resume: second leg completes remaining pages", report_res.processed_count < 3
    )
    check("Resume: all done after complete", statuses_res["pending"] == 0)

    passed = sum(1 for _, s in checks if s == "PASS")
    total = len(checks)
    print(f"\n  {passed}/{total} checks passed")

    if passed == total:
        print("\n  ALL CHECKS PASSED — Phase 12 validation successful!")
    else:
        failed = [(l, s) for l, s in checks if s == "FAIL"]
        print("\n  FAILED CHECKS:")
        for label, _ in failed:
            print(f"    - {label}")

    print("\nValidation complete.")


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.WARNING, format="%(levelname)s - %(name)s - %(message)s"
    )
    asyncio.run(main())