Refactor: Extract neopig package from neopig.py

Phase 1 of module restructuring as outlined in docs/REFACTOR.md:

New neopig/ package modules:
- live.py: Live media queue (get_live_queue, emit_live_media)
- state.py: AppendOnlyStateLog, state file helpers
- html_utils.py: trim_html_wrapper, extract_meta_from_html
- logging.py: TqdmLoggingHandler, job logging functions
- backfill/: markdown and screenshot backfill operations

Package features:
- Lazy import of NeoPig/main from neopig.py via __getattr__
- Full backwards compatibility with existing imports
- 42 new unit tests for extracted modules

Total: 458 tests passing

neopig/__init__.py

@@ -0,0 +1,65 @@
+"""
+neopig - Neo Python Image Grabber package.
+
+Full-domain async media crawler with content-addressed deduplication.
+"""
+
+from .live import get_live_queue, emit_live_media
+from .state import (
+    AppendOnlyStateLog,
+    get_state_log_path,
+    get_state_file_path,
+    rotate_state_file,
+)
+from .html_utils import trim_html_wrapper, extract_meta_from_html
+from .logging import (
+    TqdmLoggingHandler,
+    setup_logging,
+    start_job_logging,
+    stop_job_logging,
+    get_job_logs,
+)
+from .backfill import backfill_markdown, backfill_screenshots
+
+# Lazy imports for items still in neopig.py (to avoid circular imports)
+_lazy_imports = {'NeoPig', 'main'}
+
+def __getattr__(name):
+    """Lazy import NeoPig and main from neopig.py to avoid circular imports."""
+    if name in _lazy_imports:
+        import importlib.util
+        import os
+        spec = importlib.util.spec_from_file_location(
+            "neopig_main",
+            os.path.join(os.path.dirname(os.path.dirname(__file__)), "neopig.py")
+        )
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return getattr(module, name)
+    raise AttributeError(f"module 'neopig' has no attribute {name!r}")
+
+__all__ = [
+    # Live queue
+    'get_live_queue',
+    'emit_live_media',
+    # State management
+    'AppendOnlyStateLog',
+    'get_state_log_path',
+    'get_state_file_path',
+    'rotate_state_file',
+    # HTML utilities
+    'trim_html_wrapper',
+    'extract_meta_from_html',
+    # Logging
+    'TqdmLoggingHandler',
+    'setup_logging',
+    'start_job_logging',
+    'stop_job_logging',
+    'get_job_logs',
+    # Backfill
+    'backfill_markdown',
+    'backfill_screenshots',
+    # Main (lazy imports)
+    'NeoPig',
+    'main',
+]

neopig/backfill/__init__.py

@@ -0,0 +1,13 @@
+"""
+Backfill operations for neopig.
+
+Re-processes stored content to regenerate or update derived data.
+"""
+
+from .markdown import backfill_markdown
+from .screenshots import backfill_screenshots
+
+__all__ = [
+    'backfill_markdown',
+    'backfill_screenshots',
+]

neopig/backfill/markdown.py

@@ -0,0 +1,300 @@
+"""
+Markdown backfill operations for neopig.
+
+Re-processes stored HTML to regenerate markdown with smart structure detection.
+"""
+
+import json
+import logging
+import multiprocessing
+import sqlite3
+import threading
+import time
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
+from datetime import datetime, timezone
+
+from tqdm import tqdm
+
+from ..html_utils import trim_html_wrapper, extract_meta_from_html
+
+logger = logging.getLogger(__name__)
+
+
+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
+
+    page_id, uri, raw_html, trim_wrapper_flag = args
+    try:
+        description, keywords = extract_meta_from_html(raw_html)
+        raw = raw_html
+        if trim_wrapper_flag:
+            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).
+    """
+    chunk, db_path, trim_wrapper_flag, 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
+        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()
+
+        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_flag))
+        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 {
+        '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
+
+    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
+        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
+        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
+        with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet, smoothing=0.1) as pbar:
+            last_count = 0
+            while True:
+                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):
+                    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")

neopig/backfill/screenshots.py

@@ -0,0 +1,273 @@
+"""
+Screenshot backfill operations for neopig.
+
+Re-captures screenshots as JPEG to replace old PNGs.
+"""
+
+import asyncio
+import json
+import logging
+import multiprocessing
+import os
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from miniuri import Uri
+from tqdm import tqdm
+
+from ..state import get_state_file_path
+
+logger = logging.getLogger(__name__)
+
+
+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, SCORE_SCREENSHOT
+    from sqlalchemy import select, update
+
+    # 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
+        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."""
+            async with semaphore:
+                try:
+                    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
+
+                    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:
+                        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()
+                        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
+        if fast_mode and max_workers > 1:
+            async def capture_and_process(row):
+                success, data = await capture_one(row.page_uri, row.md5_hash)
+                await process_result(success, data)
+
+            tasks = [asyncio.create_task(capture_and_process(row)) for row in rows]
+            await asyncio.gather(*tasks)
+        else:
+            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")

neopig/html_utils.py

@@ -0,0 +1,83 @@
+"""
+HTML processing utilities for neopig.
+
+Functions for cleaning and extracting metadata from HTML content.
+"""
+
+
+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

neopig/live.py

@@ -0,0 +1,29 @@
+"""
+Live media event queue for real-time SSE streaming.
+
+Pushed to after disk save, SSE endpoint reads from here.
+"""
+
+import asyncio
+from typing import Dict, Any
+
+# 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)

neopig/logging.py

@@ -0,0 +1,75 @@
+"""
+Logging utilities for neopig.
+
+Provides tqdm-safe logging handlers and job-specific log capture.
+"""
+
+import logging
+from pathlib import Path
+from typing import Dict
+
+from tqdm import tqdm
+
+# Per-job log handlers (job_id -> handler)
+LOGS_PATH = Path("data/logs")
+JOB_LOG_HANDLERS: Dict[int, logging.FileHandler] = {}
+
+
+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)
+
+
+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)
+
+
+def get_job_logs(job_id: int, tail: int = 0) -> str:
+    """Read logs for a crawl job.
+
+    Args:
+        job_id: Job ID to get logs for
+        tail: If > 0, return only last N lines
+
+    Returns:
+        Log contents as string
+    """
+    log_file = LOGS_PATH / f"{job_id}.log"
+    if not log_file.exists():
+        return ""
+
+    content = log_file.read_text(encoding='utf-8')
+    if tail > 0:
+        lines = content.splitlines()
+        return '\n'.join(lines[-tail:])
+    return content

neopig/state.py

@@ -0,0 +1,182 @@
+"""
+State management for crawl operations.
+
+Provides append-only state logging for fast, reliable crawl state tracking
+that survives interruptions and enables easy resume.
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Dict, Any, Optional, List
+
+logger = logging.getLogger(__name__)
+
+
+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:
+            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