diff --git a/.gitignore b/.gitignore index 2746ffe..fbc8025 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,4 @@ vendor/ # OS .DS_Store Thumbs.db +data/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..1b28ecf --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,293 @@ +# CLAUDE.md - neopig Project Reference + +## Commit Rules + +**NEVER attribute Claude in commit messages.** No co-author tags, no "Generated with Claude" footers, no AI mentions. Keep commits clean and human-authored in appearance. + +## Quick Start + +```bash +# Setup +python3 -m venv .venv && source .venv/bin/activate +pip install -r requirements.txt + +# Run crawler +python neopig.py https://example.com --mode images + +# Run SERP web interface +python serp.py --host 0.0.0.0 --port 8000 + +# Run tests +pytest tests/ -v +``` + +## Project Overview + +**neopig** (Neo Python Image Grabber) is a full-domain async media crawler with: +- MD5 content-addressable deduplication +- SQLite metadata indexing with full-text search +- Triple git-tracked vault system (HTML, media, screenshots) +- FastAPI SERP web interface for search/browse +- Optional page screenshot capture via uri2png + +Based on [pig.py](http://russell.ballestrini.net/python-image-grabber-pig-py/) by Russell Ballestrini. + +## Architecture + +``` +neopig.py # Main CLI crawler entry point (NeoPig class) +archive.py # Site archiver - creates distributable tar.gz packages +async_web_fetcher.py # Async HTTP client, CrawlMode enum, robots.txt handling +database.py # SQLite schema: crawl_jobs, media, media_sources tables +storage.py # ImageVault - content-addressed storage by MD5 hash +domain_vault.py # Triple filevault: DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault +screenshot.py # ScreenshotCapture wrapper for uri2png +serp.py # FastAPI SERP server with search, live feed, crawl UI +``` + +## Key Classes + +### NeoPig (neopig.py) +Main crawler orchestrator. Initializes database, vault, fetcher, and screenshot modules. + +```python +pig = NeoPig(db_path="neopig.db", vault_path="vault") +await pig.init() +stats = await pig.crawl( + target_uri="https://example.com", + keywords=["tag1", "tag2"], + mode=CrawlMode.IMAGES, + depth=-1, # unlimited + max_pages=-1, +) +``` + +### CrawlMode (async_web_fetcher.py) +```python +class CrawlMode(Enum): + TEXT = "text" # Extract text content + IMAGES = "images" # Images only + VIDEOS = "videos" # Videos only + MEDIA = "media" # All media (images + videos + audio) + ALL = "all" # Full domain slurp +``` + +### Database Schema (database.py) +- **crawl_jobs**: id, target_uri, keywords (JSON), mode, status, started_at, completed_at, stats (JSON) +- **media**: md5_hash (PK), media_type, mime_type, file_size, keywords, alt_text, title, first_seen_at, analysis_status, analysis_result +- **media_sources**: Tracks all contexts where media was found (page_uri, page_title, page_description, page_content, alt_text, link_text, detail_page_uri, detail_title, detail_content, searchable_text) + +### Triple Vault System (domain_vault.py) + +Three separate git-tracked vaults per domain with 9-layer deep hash paths: + +1. **HTML Vault** (`html_vault/`): Stores page HTML + - `{path}/index.html.og` - Original HTML + - `{path}/index.html` - Rewritten with neopig media paths + +2. **Media Vault** (`media_vault/`): Stores images/videos with git LFS + - Files mirror original URL paths + +3. **Linkpeek Vault** (`linkpeek_vault/`): Stores page screenshots with git LFS + - One PNG per page, named after URL path + +Environment: `NEOPIG_VAULT_SALT` - secret salt for domain hashing (privacy) + +### ImageVault (storage.py) +Content-addressed storage. Files stored by MD5 hash in 256-bucket directory structure: +``` +vault/ + ab/abcd1234...5678.jpg + cd/cdef5678...1234.png +``` + +## Site Archiver (archive.py) + +Archive sunset sites into distributable tar.gz packages: + +```bash +# Basic usage - archive entire site +python archive.py https://discourse-urho3d.github.io/ + +# Custom output directory +python archive.py https://example.com -o ./archives/ + +# Limit depth and pages +python archive.py https://example.com --depth 5 --max-pages 500 + +# Disable screenshots (faster) +python archive.py https://example.com --no-screenshots + +# Disable markdown conversion +python archive.py https://example.com --no-markdown +``` + +Output structure: +``` +{domain}-{date}/ + index.html # Archive index with sitemap + html/ # Original HTML pages + markdown/ # Converted markdown (optional) + media/ # Images, videos, audio + screenshots/ # Page screenshots (optional) + archive.db # SQLite FTS5 search database + serve.py # Embedded Pyramid search server + requirements.txt # Server dependencies (pyramid) + metadata.json # Crawl metadata and statistics +``` + +**Embedded Search Server:** +```bash +# Option 1: Extract and run +tar -xzf example.com-20251229.tar.gz +cd example.com-20251229 +pip install -r requirements.txt +python serve.py +# Open http://localhost:6543 + +# Option 2: Serve directly from tar.gz (no extraction) +python serve.py example.com-20251229.tar.gz +``` + +**Self-Extracting Executable (.run):** +```bash +# Create self-extracting archive (builds bootstrap if needed) +make run TARBALL=example.com-20251229.tar.gz + +# Run it (just needs python3 + pyramid on target) +./example.com-20251229.run +# Opens http://localhost:6543 +``` + +The `.run` file is a single executable containing: +- C bootstrap (~14KB) +- Full tar.gz archive +- NEOPIG trailer with offset + +When run, it extracts serve.py to /tmp and launches the search server. + +## Screenshot Engines + +neopig supports multiple screenshot backends via uri2png. Auto-detects the lightest available: + +| Engine | Install | Speed | Notes | +|--------|---------|-------|-------| +| wkhtmltoimage | `apt install wkhtmltopdf` | Fast | Native Qt WebKit, no browser download | +| cutycapt | `apt install cutycapt` | Fast | Native Qt WebKit, no browser download | +| playwright-webkit | `pip install playwright && playwright install webkit` | Medium | Lighter than Chromium | +| playwright-chromium | `pip install playwright && playwright install chromium` | Slow | Most compatible, heaviest | + +```bash +# List available engines +python neopig.py --list-engines + +# Use specific engine +python neopig.py https://example.com --screenshot --screenshot-engine wkhtmltoimage +python archive.py https://example.com --screenshot-engine cutycapt + +# Auto-detect (default) - picks lightest available +python neopig.py https://example.com --screenshot +``` + +**Recommendation:** Install `wkhtmltopdf` for fast, lightweight screenshots without browser downloads. + +## CLI Usage (neopig.py) + +```bash +# Single target +python neopig.py https://example.com --mode images + +# Multiple targets concurrently +python neopig.py https://site1.com https://site2.com --mode media + +# With keywords for tagging +python neopig.py https://example.com -k "tag1" "tag2" --mode images + +# Limit depth and pages +python neopig.py https://example.com --depth 5 --max-pages 500 + +# Index only (no download) +python neopig.py https://example.com --no-download + +# Enable screenshots (requires uri2png) +python neopig.py https://example.com --screenshot --screenshot-width 1920 --screenshot-height 1080 +``` + +## SERP API Endpoints (serp.py) + +- `GET /` - Search UI +- `GET /crawl` - Crawler UI +- `GET /live` - Live feed (watch images appear) +- `GET /view/{md5_hash}` - Media detail page +- `GET /media/{md5_hash}` - Serve media file +- `GET /api/stats` - Database statistics +- `GET /api/search?q=&type=&limit=` - Search media +- `GET /api/media/{md5_hash}` - Media info JSON +- `POST /api/crawl` - Start crawl job +- `GET /api/crawl/jobs` - List crawl jobs +- `GET /api/crawl/jobs/{id}` - Get job status +- `GET /health` - Health check + +## Dependencies + +Core: +- aiohttp, aiofiles - async HTTP/file operations +- beautifulsoup4, html5lib - HTML parsing +- aiosqlite - async SQLite +- miniuri - URI parsing + +SERP: +- fastapi, uvicorn - web server +- python-multipart - form handling + +Optional: +- filevault - content-addressed storage backend +- pillow - image processing +- uri2png - page screenshots (wkhtmltoimage, cutycapt, or playwright backends) + +## Makefile Targets + +```bash +make install # Create venv, install deps +make test # Run pytest +make serp # Start basic SERP server +make server # Start SERP + screenshot server +make crawl ARGS="..." # Run crawler with args +make vendor-install # Install uri2png with playwright +make clean # Remove venv and test artifacts +``` + +## Testing + +```bash +pytest tests/ -v --tb=short +``` + +Tests use pytest-asyncio. Config in `tests/conftest.py` and `pytest.ini`. + +## Key Patterns + +### Skeleton Key Approach +Media records track both: +- **Embedding context**: page_title, page_content from listing page +- **Detail context**: detail_title, detail_content from detail page (Pinterest-style galleries) + +This enables finding images by ANY associated text. + +### MediaMetadata Accumulator (async_web_fetcher.py) +"Never clobber, always append" - collects ALL metadata: +- img.alt, img.title, a.title, a.text, figcaption, nearby headings, page title +- Produces combined `searchable_text` for full-text search + +### Deduplication +- Content: MD5 hash of file bytes +- Context: UNIQUE(md5_hash, media_uri, page_uri) - same content from different pages tracked separately + +### Robots.txt Compliance +AsyncWebFetcher respects robots.txt with configurable crawl delay (default 2s). + +## File Extensions + +Images: .jpg .jpeg .png .gif .webp .svg .bmp .ico .tiff .avif +Videos: .mp4 .webm .mov .avi .mkv .m4v .ogv .flv .wmv +Audio: .mp3 .wav .ogg .m4a .flac .aac .wma diff --git a/Makefile b/Makefile index dd6b921..63c01ff 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: venv install test crawl serp clean vendor-uri2png vendor-install +.PHONY: venv install test crawl serp clean vendor-uri2png vendor-install archive bootstrap VENV := .venv PYTHON := $(VENV)/bin/python @@ -17,12 +17,35 @@ test: install crawl: install $(PYTHON) neopig.py $(ARGS) +archive: install + $(PYTHON) archive.py $(ARGS) + +# Build bootstrap binary for self-extracting archives +bootstrap: bootstrap.c + gcc -O2 -Wall -o bootstrap bootstrap.c -lz + @echo "Built: bootstrap ($$(stat -c%s bootstrap 2>/dev/null || stat -f%z bootstrap) bytes)" + +# Create self-extracting .run from a tarball +# Usage: make run TARBALL=example.tar.gz +run: bootstrap +ifndef TARBALL + $(error TARBALL not set. Usage: make run TARBALL=path/to/archive.tar.gz) +endif + @OUTNAME=$$(basename "$(TARBALL)" .tar.gz).run; \ + BOOTSTRAP_SIZE=$$(stat -c%s bootstrap 2>/dev/null || stat -f%z bootstrap); \ + cat bootstrap "$(TARBALL)" > "$$OUTNAME"; \ + echo -n "NEOPIG" >> "$$OUTNAME"; \ + printf '%016x' "$$BOOTSTRAP_SIZE" >> "$$OUTNAME"; \ + chmod +x "$$OUTNAME"; \ + echo "Created: $$OUTNAME ($$(stat -c%s $$OUTNAME 2>/dev/null || stat -f%z $$OUTNAME) bytes)" + serp: install $(PYTHON) serp.py --host 0.0.0.0 --port 8000 clean: rm -rf $(VENV) __pycache__ *.pyc rm -rf test_vault test_neopig.db + rm -f bootstrap *.run # Vendor dependencies vendor-uri2png: @@ -48,6 +71,13 @@ server: vendor-install # make serp - start basic SERP server # make server - start combined server (SERP + screenshot) # make crawl ARGS="https://example.com rick morty --mode images" +# make archive ARGS="https://discourse-urho3d.github.io/" +# make bootstrap - build C bootstrap for self-extracting archives # make clean - remove venv and test artifacts # make vendor-uri2png - fetch uri2png into vendor/ # make vendor-install - install uri2png Python package with screenshot support +# +# Self-extracting archive: +# make archive ARGS="https://example.com" +# make run TARBALL=example.com-20251229.tar.gz +# ./example.com-20251229.run diff --git a/archive.py b/archive.py new file mode 100644 index 0000000..64fb0a3 --- /dev/null +++ b/archive.py @@ -0,0 +1,1041 @@ +#!/usr/bin/env python3 +""" +archive.py - Site Archiver for Sunset Sites + +Thin wrapper around neopig that packages crawl results into a distributable +tar.gz archive. Uses neopig for the actual crawling. + +Output structure: + {domain}-{date}/ + html/ # Original HTML pages + markdown/ # Converted markdown (optional) + media/ # Images, videos, audio + screenshots/ # Page screenshots (optional) + archive.db # SQLite search database + serve.py # Embedded search server + metadata.json # Crawl metadata + +Usage: + python archive.py https://discourse-urho3d.github.io/ + python archive.py https://example.com --no-screenshots --no-markdown +""" + +import argparse +import asyncio +import json +import logging +import os +import re +import shutil +import signal +import sqlite3 +import subprocess +import sys +import tarfile +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, Any, List, Optional +from urllib.parse import urlparse + +import aiofiles +import aiofiles.os +from bs4 import BeautifulSoup +from tqdm import tqdm + +from neopig import NeoPig, setup_logging +from async_web_fetcher import CrawlMode +from screenshot import ScreenshotConfig + +# Optional markdown conversion +try: + import html2text + HAS_HTML2TEXT = True +except ImportError: + HAS_HTML2TEXT = False + +logger = logging.getLogger(__name__) + + +def sanitize_filename(name: str) -> str: + """Sanitize a string for use as filename. Uses - as separator.""" + name = re.sub(r'[<>:"/\\|?*.]', '-', name) + name = re.sub(r'-+', '-', name) # collapse multiple dashes + name = name.strip('- ') + return name[:200] if name else 'unnamed' + + +def url_to_path(url: str) -> str: + """Convert URL to filesystem path.""" + parsed = urlparse(url) + path = parsed.path.strip('/') + if not path: + return 'index.html' + if path.endswith('.html') or path.endswith('.htm'): + return path + if '.' in path.split('/')[-1]: + return path + return f"{path}/index.html" + + +def html_to_markdown(html: str, base_url: str = '') -> str: + """Convert HTML to markdown.""" + if not HAS_HTML2TEXT: + return html + h = html2text.HTML2Text() + h.ignore_links = False + h.ignore_images = False + h.body_width = 0 + h.unicode_snob = True + if base_url: + h.baseurl = base_url + return h.handle(html) + + +class SiteArchiver: + """ + Packages neopig crawl results into a distributable tar.gz archive. + + Uses neopig for crawling, then reads from its vaults to build the archive. + """ + + def __init__( + self, + output_dir: str = '.', + include_screenshots: bool = True, + include_markdown: bool = True, + screenshot_config: ScreenshotConfig = None, + fast_mode: bool = False, + ): + self.output_dir = Path(output_dir) + self.include_screenshots = include_screenshots + self.include_markdown = include_markdown and HAS_HTML2TEXT + self.screenshot_config = screenshot_config or ScreenshotConfig(enabled=include_screenshots) + self.fast_mode = fast_mode + + async def archive( + self, + target_url: str, + depth: int = -1, + max_pages: int = -1, + db_path: str = None, + vault_path: str = None, + ) -> Path: + """ + Archive a site using neopig and package into tar.gz. + """ + parsed = urlparse(target_url) + domain = parsed.netloc.lower() + date_str = datetime.now().strftime('%Y%m%d') + archive_name = f"{sanitize_filename(domain)}-{date_str}" + + # Use standard neopig data paths - one vault, multiple domains + if not db_path: + db_path = "data/neopig.db" + if not vault_path: + vault_path = "data/vault" + + # Ensure data directory exists + Path("data").mkdir(exist_ok=True) + + logger.info(f"Starting archive of {target_url}") + logger.info(f"Archive name: {archive_name}") + + # Create neopig instance and crawl + pig = NeoPig( + db_path=db_path, + vault_path=vault_path, + screenshot_config=self.screenshot_config, + fast_mode=self.fast_mode, + ) + await pig.init() + + # Run the crawl + stats = await pig.crawl( + target_uri=target_url, + mode=CrawlMode.ALL, + depth=depth, + max_pages=max_pages, + download_media=True, + ) + + # Now package the results + logger.info("Packaging archive...") + + with tempfile.TemporaryDirectory() as tmpdir: + archive_root = Path(tmpdir) / archive_name + archive_root.mkdir(parents=True) + + # Create subdirectories + html_dir = archive_root / 'html' + media_dir = archive_root / 'media' + html_dir.mkdir() + media_dir.mkdir() + + if self.include_markdown: + md_dir = archive_root / 'markdown' + md_dir.mkdir() + + if self.include_screenshots: + screenshots_dir = archive_root / 'screenshots' + screenshots_dir.mkdir() + + # Read pages from neopig's html vault + html_vault_path = Path(vault_path) / 'html_vault' / domain + sitemap = [] + + if html_vault_path.exists(): + for html_file in html_vault_path.rglob('*.html'): + rel_path = html_file.relative_to(html_vault_path) + html_content = html_file.read_text(encoding='utf-8', errors='replace') + + # Write HTML + dest_path = html_dir / rel_path + await aiofiles.os.makedirs(dest_path.parent, exist_ok=True) + async with aiofiles.open(dest_path, 'w', encoding='utf-8') as f: + await f.write(html_content) + + # Write markdown + if self.include_markdown: + md_path = md_dir / str(rel_path).replace('.html', '.md') + await aiofiles.os.makedirs(md_path.parent, exist_ok=True) + md_content = html_to_markdown(html_content) + async with aiofiles.open(md_path, 'w', encoding='utf-8') as f: + await f.write(md_content) + + # Extract title for sitemap + title = self._extract_title(html_content) or str(rel_path) + sitemap.append({ + 'path': f'html/{rel_path}', + 'title': title, + }) + + # Copy media from neopig's media vault (follows symlinks to hash vault) + media_vault_path = Path(vault_path) / 'media_vault' / domain + if media_vault_path.exists(): + for media_file in media_vault_path.rglob('*'): + if media_file.is_file() or media_file.is_symlink(): + try: + # Follow symlinks to get actual content + if media_file.is_symlink(): + target = media_file.resolve() + if not target.exists(): + logger.debug(f"Skipping broken symlink: {media_file}") + continue + content = target.read_bytes() + else: + content = media_file.read_bytes() + rel_path = media_file.relative_to(media_vault_path) + dest_path = media_dir / rel_path + await aiofiles.os.makedirs(dest_path.parent, exist_ok=True) + async with aiofiles.open(dest_path, 'wb') as f: + await f.write(content) + except Exception as e: + logger.debug(f"Error copying media {media_file}: {e}") + + # Copy screenshots from neopig's linkpeek vault (follows symlinks to hash vault) + if self.include_screenshots: + linkpeek_vault_path = Path(vault_path) / 'linkpeek_vault' / domain + if linkpeek_vault_path.exists(): + for screenshot_file in linkpeek_vault_path.rglob('*.png'): + try: + # Follow symlinks to get actual content + if screenshot_file.is_symlink(): + target = screenshot_file.resolve() + if not target.exists(): + logger.debug(f"Skipping broken symlink: {screenshot_file}") + continue + content = target.read_bytes() + else: + content = screenshot_file.read_bytes() + rel_path = screenshot_file.relative_to(linkpeek_vault_path) + dest_path = screenshots_dir / rel_path + await aiofiles.os.makedirs(dest_path.parent, exist_ok=True) + async with aiofiles.open(dest_path, 'wb') as f: + await f.write(content) + except Exception as e: + logger.debug(f"Error copying screenshot {screenshot_file}: {e}") + + # Create search database + await self._create_search_database(archive_root, sitemap, domain, html_dir) + + # Write embedded serve.py + self._write_serve_py(archive_root) + + # Write metadata + metadata = { + 'domain': domain, + 'target_url': target_url, + 'created': datetime.now(timezone.utc).isoformat(), + 'stats': stats, + 'include_screenshots': self.include_screenshots, + 'include_markdown': self.include_markdown, + } + async with aiofiles.open(archive_root / 'metadata.json', 'w') as f: + await f.write(json.dumps(metadata, indent=2)) + + # Copy state file into archive for future delta crawls + state_domain = domain.replace('.', '-').replace(':', '-') + state_file = Path("data") / f"crawl-state-{state_domain}.json" + if state_file.exists(): + shutil.copy(state_file, archive_root / 'crawl_state.json') + logger.info(f"Included crawl state for future delta crawls") + + # Create tar.gz + tar_path = self.output_dir / f"{archive_name}.tar.gz" + + def create_tarball(): + with tarfile.open(tar_path, 'w:gz') as tar: + tar.add(archive_root, arcname=archive_name) + + await asyncio.to_thread(create_tarball) + + final_size = tar_path.stat().st_size + logger.info(f"Archive complete: {tar_path} ({final_size / 1024 / 1024:.1f} MB)") + + return tar_path + + def _extract_title(self, html: str) -> Optional[str]: + """Extract title from HTML.""" + try: + soup = BeautifulSoup(html, 'html.parser') + if soup.title and soup.title.string: + return soup.title.string.strip() + h1 = soup.find('h1') + if h1: + return h1.get_text().strip() + except Exception: + pass + return None + + def _extract_text(self, html: str) -> str: + """Extract readable text from HTML for search indexing.""" + try: + soup = BeautifulSoup(html, 'html.parser') + for tag in soup(['script', 'style', 'nav', 'footer', 'header']): + tag.decompose() + text = soup.get_text(separator=' ', strip=True) + text = re.sub(r'\s+', ' ', text) + return text[:50000] + except Exception: + return '' + + async def _create_search_database( + self, + archive_root: Path, + sitemap: List[Dict[str, str]], + domain: str, + html_dir: Path, + ): + """Create SQLite database with searchable page content.""" + db_path = archive_root / 'archive.db' + + def create_db(): + conn = sqlite3.connect(db_path) + c = conn.cursor() + + c.execute(''' + CREATE TABLE pages ( + id INTEGER PRIMARY KEY, + url TEXT NOT NULL, + path TEXT NOT NULL, + title TEXT, + content TEXT + ) + ''') + + c.execute(''' + CREATE VIRTUAL TABLE pages_fts USING fts5( + title, content, url, path, + content='pages', + content_rowid='id' + ) + ''') + + c.execute(''' + CREATE TRIGGER pages_ai AFTER INSERT ON pages BEGIN + INSERT INTO pages_fts(rowid, title, content, url, path) + VALUES (new.id, new.title, new.content, new.url, new.path); + END + ''') + + for item in sitemap: + path = item['path'] + title = item['title'] + + html_path = archive_root / path + if html_path.exists(): + html_content = html_path.read_text(encoding='utf-8', errors='replace') + text_content = self._extract_text(html_content) + else: + text_content = '' + + c.execute( + 'INSERT INTO pages (url, path, title, content) VALUES (?, ?, ?, ?)', + (path, path, title, text_content) + ) + + conn.commit() + conn.close() + + await asyncio.to_thread(create_db) + logger.info(f"Created search database: {db_path}") + + def _write_serve_py(self, archive_root: Path): + """Write embedded search server (stdlib only, no dependencies).""" + serve_py = '''#!/usr/bin/env python3 +""" +neopig Archive Server - Browse and search archived sites. + +Zero dependencies - uses only Python stdlib. + +Usage: + python serve.py # Serve from extracted archive + python serve.py archive.tar.gz # Serve directly from tarball + python serve.py -p 8080 # Custom port + ./archive.run # Self-extracting archive +""" + +import argparse +import html +import json +import mimetypes +import os +import re +import sqlite3 +import sys +import tarfile +import tempfile +from http.server import HTTPServer, BaseHTTPRequestHandler +from pathlib import Path +from urllib.parse import parse_qs, urlparse, unquote + +# Globals set at startup +ARCHIVE_ROOT = None +TAR_FILE = None +TAR_MEMBERS = {} +DB_PATH = None +METADATA = {} + + +def get_archive_source(): + """Determine if we are in a tarball, .run, or extracted directory.""" + exe_path = Path(sys.argv[0]).resolve() + + # Check for .run format (has NEOPIG trailer) + if exe_path.suffix == '.run' or (len(sys.argv) == 1 and exe_path.stat().st_size > 1000000): + try: + with open(exe_path, 'rb') as f: + f.seek(-22, 2) + trailer = f.read(22) + if trailer[:6] == b'NEOPIG': + offset = int(trailer[6:22].decode(), 16) + return ('run', exe_path, offset) + except Exception: + pass + + # Check command line for tarball argument + for arg in sys.argv[1:]: + if not arg.startswith('-'): + p = Path(arg) + if p.exists() and p.suffix in ('.gz', '.tar', '.tgz'): + return ('tarball', p, 0) + + # Must be extracted directory + return ('directory', Path(__file__).parent, 0) + + +def init_archive(): + """Initialize archive access.""" + global ARCHIVE_ROOT, TAR_FILE, TAR_MEMBERS, DB_PATH, METADATA + + source_type, source_path, offset = get_archive_source() + + if source_type == 'run': + print(f"Serving from self-extracting archive: {source_path}") + f = open(source_path, 'rb') + f.seek(offset) + TAR_FILE = tarfile.open(fileobj=f, mode='r:gz') + elif source_type == 'tarball': + print(f"Serving from tarball: {source_path}") + TAR_FILE = tarfile.open(source_path, 'r:gz') + else: + print(f"Serving from directory: {source_path}") + ARCHIVE_ROOT = source_path + + if TAR_FILE: + # Build member lookup and find archive root + for member in TAR_FILE.getmembers(): + TAR_MEMBERS[member.name] = member + first = list(TAR_MEMBERS.keys())[0] + archive_name = first.split('/')[0] + ARCHIVE_ROOT = Path(archive_name) + + # Extract database to temp for searching + db_member = f"{archive_name}/archive.db" + if db_member in TAR_MEMBERS: + temp_dir = tempfile.mkdtemp() + TAR_FILE.extract(TAR_MEMBERS[db_member], temp_dir) + DB_PATH = Path(temp_dir) / db_member + else: + DB_PATH = ARCHIVE_ROOT / 'archive.db' + + # Load metadata + meta_path = ARCHIVE_ROOT / 'metadata.json' if not TAR_FILE else None + if meta_path and meta_path.exists(): + METADATA = json.loads(meta_path.read_text()) + elif TAR_FILE: + meta_member = f"{ARCHIVE_ROOT}/metadata.json" + if meta_member in TAR_MEMBERS: + f = TAR_FILE.extractfile(TAR_MEMBERS[meta_member]) + if f: + METADATA = json.loads(f.read().decode()) + + +def read_file(path: str) -> tuple: + """Read file from archive. Returns (content_bytes, mime_type) or (None, None).""" + if TAR_FILE: + # Normalize path for tarball + tar_path = f"{ARCHIVE_ROOT}/{path}".lstrip('/') + if tar_path in TAR_MEMBERS: + f = TAR_FILE.extractfile(TAR_MEMBERS[tar_path]) + if f: + mime, _ = mimetypes.guess_type(path) + return f.read(), mime or 'application/octet-stream' + return None, None + else: + file_path = ARCHIVE_ROOT / path + if file_path.exists() and file_path.is_file(): + # Security: prevent path traversal + try: + file_path.resolve().relative_to(ARCHIVE_ROOT.resolve()) + except ValueError: + return None, None + mime, _ = mimetypes.guess_type(str(file_path)) + return file_path.read_bytes(), mime or 'application/octet-stream' + return None, None + + +def list_files(subdir: str, pattern: str = '*') -> list: + """List files in a subdirectory.""" + files = [] + if TAR_FILE: + prefix = f"{ARCHIVE_ROOT}/{subdir}/" + for name in TAR_MEMBERS: + if name.startswith(prefix) and not name.endswith('/'): + rel = name[len(prefix):] + if pattern == '*' or rel.endswith(pattern.replace('*', '')): + files.append(rel) + else: + dir_path = ARCHIVE_ROOT / subdir + if dir_path.exists(): + for f in dir_path.rglob(pattern): + if f.is_file(): + files.append(str(f.relative_to(dir_path))) + return sorted(files) + + +def search_pages(query: str, limit: int = 50) -> list: + """Search pages using FTS5.""" + if not DB_PATH or not DB_PATH.exists(): + return [] + + conn = sqlite3.connect(DB_PATH) + c = conn.cursor() + try: + c.execute(""" + SELECT p.path, p.title, snippet(pages_fts, 1, '', '', '...', 40) + FROM pages_fts + JOIN pages p ON pages_fts.rowid = p.id + WHERE pages_fts MATCH ? + ORDER BY rank + LIMIT ? + """, (query, limit)) + results = [{'path': r[0], 'title': r[1], 'snippet': r[2]} for r in c.fetchall()] + except sqlite3.OperationalError: + results = [] + finally: + conn.close() + return results + + +def get_stats() -> dict: + """Get archive statistics.""" + stats = { + 'pages': len(list_files('html', '*.html')), + 'media': len(list_files('media')), + 'screenshots': len(list_files('screenshots', '*.png')), + 'domain': METADATA.get('domain', 'unknown'), + 'created': METADATA.get('created', 'unknown'), + } + if METADATA.get('stats'): + stats.update(METADATA['stats']) + return stats + + +# HTML Templates +INDEX_HTML = """ + + + + + {domain} - neopig Archive + + + +
+

neopig Archive

+

{domain} - archived {created}

+ + + +
+
{pages}
Pages
+
{media}
Media
+
{screenshots}
Screenshots
+
+ +
+ + +
+ +
+
+ + + +""" + +BROWSE_HTML = """ + + + + + Browse - {domain} + + + +
+

Browse Pages

+ +
+ {items} +
+
+ +""" + +MEDIA_HTML = """ + + + + + Media - {domain} + + + +
+

{title}

+ +
+ {items} +
+
+ +""" + + +class ArchiveHandler(BaseHTTPRequestHandler): + """HTTP request handler for the archive.""" + + def log_message(self, format, *args): + print(f"[{self.log_date_time_string()}] {args[0]}") + + def send_html(self, content: str, status: int = 200): + self.send_response(status) + self.send_header('Content-Type', 'text/html; charset=utf-8') + self.send_header('Content-Length', len(content.encode())) + self.end_headers() + self.wfile.write(content.encode()) + + def send_json(self, data, status: int = 200): + content = json.dumps(data) + self.send_response(status) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', len(content.encode())) + self.end_headers() + self.wfile.write(content.encode()) + + def send_file(self, content: bytes, mime: str): + self.send_response(200) + self.send_header('Content-Type', mime) + self.send_header('Content-Length', len(content)) + self.send_header('Cache-Control', 'public, max-age=86400') + self.end_headers() + self.wfile.write(content) + + def send_404(self): + self.send_html('

404 Not Found

', 404) + + def do_GET(self): + parsed = urlparse(self.path) + path = unquote(parsed.path) + query = parse_qs(parsed.query) + + # API endpoints + if path == '/api/search': + q = query.get('q', [''])[0] + results = search_pages(q) if q else [] + self.send_json(results) + return + + if path == '/api/stats': + self.send_json(get_stats()) + return + + # Pages + if path == '/': + stats = get_stats() + content = INDEX_HTML.format(**stats) + self.send_html(content) + return + + if path == '/browse': + pages = list_files('html', '*.html') + items = ''.join([ + f'
{html.escape(p)}
' + for p in pages[:500] + ]) + if not items: + items = '
No pages found
' + content = BROWSE_HTML.format(domain=METADATA.get('domain', ''), items=items) + self.send_html(content) + return + + if path == '/media': + files = list_files('media') + items = [] + for f in files[:200]: + ext = Path(f).suffix.lower() + if ext in ('.mp4', '.webm', '.mov'): + media_el = f'' + else: + media_el = f'' + items.append( + '
' + f'{media_el}' + f'
{html.escape(f)}
' + '
' + ) + content = MEDIA_HTML.format( + domain=METADATA.get('domain', ''), + title='Media', + items=''.join(items) if items else '
No media found
' + ) + self.send_html(content) + return + + if path == '/screenshots': + files = list_files('screenshots', '*.png') + items = [] + for f in files[:200]: + items.append( + '
' + f'' + f'' + '' + f'
{html.escape(f)}
' + '
' + ) + content = MEDIA_HTML.format( + domain=METADATA.get('domain', ''), + title='Screenshots', + items=''.join(items) if items else '
No screenshots found
' + ) + self.send_html(content) + return + + # Serve static files + file_path = path.lstrip('/') + content, mime = read_file(file_path) + if content is not None: + self.send_file(content, mime) + else: + self.send_404() + + +def main(): + parser = argparse.ArgumentParser(description='neopig Archive Server') + parser.add_argument('archive', nargs='?', help='Path to archive.tar.gz (optional)') + parser.add_argument('-p', '--port', type=int, default=8000, help='Port to listen on') + parser.add_argument('--host', default='0.0.0.0', help='Host to bind to') + args = parser.parse_args() + + init_archive() + + server = HTTPServer((args.host, args.port), ArchiveHandler) + print(f"Starting neopig archive server at http://{args.host}:{args.port}") + print(f"Archive: {METADATA.get('domain', 'unknown')} ({get_stats()['pages']} pages)") + print("Press Ctrl+C to stop") + + try: + server.serve_forever() + except KeyboardInterrupt: + print("\\nShutting down...") + server.shutdown() + + +if __name__ == '__main__': + main() +''' + (archive_root / 'serve.py').write_text(serve_py) + + +async def main(): + parser = argparse.ArgumentParser( + description="Archive a website for preservation (uses neopig for crawling)", + epilog="Example: python archive.py https://discourse-urho3d.github.io/" + ) + + parser.add_argument("url", help="URL of the site to archive") + parser.add_argument("-o", "--output", default=".", help="Output directory for tar.gz") + parser.add_argument("-d", "--depth", type=int, default=-1, help="Crawl depth (-1 = unlimited)") + parser.add_argument("-p", "--max-pages", type=int, default=-1, help="Max pages (-1 = unlimited)") + parser.add_argument("--no-screenshots", action="store_true", help="Disable screenshots") + parser.add_argument("--no-markdown", action="store_true", help="Disable markdown conversion") + parser.add_argument("--screenshot-width", type=int, default=1280, help="Screenshot width") + parser.add_argument("--screenshot-height", type=int, default=1024, help="Screenshot height") + parser.add_argument("--screenshot-engine", type=str, default=None, help="Screenshot engine") + parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output") + parser.add_argument("--fresh", action="store_true", help="Start fresh, ignore resume state") + parser.add_argument("--fast", action="store_true", help="Fast mode: no crawl delay (for sites without robots.txt)") + parser.add_argument("--serve", action="store_true", help="Start SERP server to watch crawl live") + parser.add_argument("--port", type=int, default=8000, help="Port for SERP server (default: 8000)") + + args = parser.parse_args() + + setup_logging(level=logging.DEBUG if args.verbose else logging.INFO) + + if args.no_markdown and not HAS_HTML2TEXT: + pass + elif not HAS_HTML2TEXT: + logger.warning("html2text not installed - markdown disabled") + + screenshot_config = ScreenshotConfig( + enabled=not args.no_screenshots, + width=args.screenshot_width, + height=args.screenshot_height, + engine=args.screenshot_engine, + ) + + archiver = SiteArchiver( + output_dir=args.output, + include_screenshots=not args.no_screenshots, + include_markdown=not args.no_markdown, + screenshot_config=screenshot_config, + fast_mode=args.fast, + ) + + # Start SERP server if requested + serp_process = None + if args.serve: + serp_script = Path(__file__).parent / 'serp.py' + if serp_script.exists(): + serp_cmd = [ + sys.executable, str(serp_script), + '--port', str(args.port), + '--db', 'data/neopig.db', + '--vault', 'data/vault', + ] + serp_process = subprocess.Popen( + serp_cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + logger.info(f"SERP server started at http://localhost:{args.port}") + logger.info("Watch the crawl live - pages appear as they're indexed!") + else: + logger.warning("serp.py not found - --serve disabled") + + try: + archive_path = await archiver.archive( + target_url=args.url, + depth=args.depth, + max_pages=args.max_pages, + ) + + print(f"\nArchive created: {archive_path}") + print(f"Extract with: tar -xzf {archive_path.name}") + finally: + # Cleanup SERP server + if serp_process: + 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()) diff --git a/async_web_fetcher.py b/async_web_fetcher.py index 5740363..8a3e24b 100644 --- a/async_web_fetcher.py +++ b/async_web_fetcher.py @@ -802,10 +802,16 @@ class AsyncWebFetcher: def __init__( self, user_agent: str = "uncloseai.com/1.42 (ethical web crawler; +https://uncloseai.com)", - default_crawl_delay: float = DEFAULT_CRAWL_DELAY + default_crawl_delay: float = DEFAULT_CRAWL_DELAY, + fast_mode: bool = False ): self.user_agent = user_agent self.default_crawl_delay = default_crawl_delay + self.fast_mode = fast_mode + + # Timeouts: 5s in fast mode, 60s normally + self.media_timeout = 5 if fast_mode else 60 + self.page_timeout = 5 if fast_mode else 15 # Caches for robots.txt and crawl delays per domain self.robot_parsers: Dict[str, Optional[RobotFileParser]] = {} @@ -815,6 +821,11 @@ class AsyncWebFetcher: # Page cache: {url: (html, links, timestamp)} self.page_cache: Dict[str, Tuple[str, List, float]] = {} + # Domain skip list: domains with too many consecutive timeouts + self.skip_domains: Set[str] = set() + self.domain_timeout_counts: Dict[str, int] = {} + self.MAX_CONSECUTIVE_TIMEOUTS = 5 + logger.info(f"AsyncWebFetcher initialized with user-agent: {self.user_agent}") def _get_domain(self, url: str) -> str: @@ -1173,6 +1184,18 @@ class AsyncWebFetcher: logger.debug(f"Failed to resolve canonical image from {detail_page_url}: {e}") return None + def _record_timeout(self, domain: str): + """Record a timeout for a domain. After MAX_CONSECUTIVE_TIMEOUTS, add to skip list.""" + self.domain_timeout_counts[domain] = self.domain_timeout_counts.get(domain, 0) + 1 + if self.domain_timeout_counts[domain] >= self.MAX_CONSECUTIVE_TIMEOUTS: + if domain not in self.skip_domains: + self.skip_domains.add(domain) + logger.warning(f"Skipping domain {domain} after {self.MAX_CONSECUTIVE_TIMEOUTS} consecutive timeouts") + + def _record_success(self, domain: str): + """Record a successful fetch, resetting timeout count.""" + self.domain_timeout_counts[domain] = 0 + async def fetch_media( self, url: str, @@ -1192,13 +1215,19 @@ class AsyncWebFetcher: """ global LAST_FETCH_ERROR + # Check if domain is in skip list + domain = self._get_domain(url) + if domain in self.skip_domains: + LAST_FETCH_ERROR = {'type': 'skip_domain', 'details': f'Domain {domain} skipped (too many timeouts)', 'url': url} + logger.debug(f"Skipping {url}: domain {domain} in skip list") + return None + # Check robots.txt if not await self._can_fetch(url): LAST_FETCH_ERROR = {'type': 'robots_txt', 'details': 'Blocked by robots.txt', 'url': url} return None # Enforce crawl delay - domain = self._get_domain(url) await self._enforce_crawl_delay(domain) try: @@ -1210,7 +1239,7 @@ class AsyncWebFetcher: async with session.get( url, headers={"User-Agent": self.user_agent}, - timeout=aiohttp.ClientTimeout(total=60), # Longer timeout for media + timeout=aiohttp.ClientTimeout(total=self.media_timeout), allow_redirects=True ) as response: if response.status != 200: @@ -1240,6 +1269,9 @@ class AsyncWebFetcher: # Determine media type media_type = get_media_type_from_mime(mime_type) or get_media_type_from_extension(url) + # Success - reset timeout count + self._record_success(domain) + logger.info(f"Fetched media {url}: {len(data)} bytes, MD5: {md5_hash}, type: {media_type}") return { @@ -1256,8 +1288,9 @@ class AsyncWebFetcher: await session.close() except asyncio.TimeoutError: - LAST_FETCH_ERROR = {'type': 'timeout', 'details': 'Download timeout', 'url': url} - logger.error(f"Timeout fetching media {url}") + self._record_timeout(domain) + LAST_FETCH_ERROR = {'type': 'timeout', 'details': f'Download timeout ({self.media_timeout}s)', 'url': url} + logger.error(f"Timeout fetching media {url} ({self.media_timeout}s)") return None except Exception as e: LAST_FETCH_ERROR = {'type': 'unknown', 'details': str(e), 'url': url} @@ -1764,6 +1797,8 @@ class AsyncWebFetcher: mode: CrawlMode = CrawlMode.TEXT, media_callback = None, # Callback for discovered media: async fn(media_item: Dict) -> None page_callback = None, # Callback for page HTML: async fn(url: str, html: str) -> None + uris_total_callback = None, # Callback for URI total updates: fn(total: int) -> None + initial_visited: Optional[Set[str]] = None, # Pre-visited URLs for resume support ) -> List[Dict[str, str]]: """ Intelligent keyword-driven crawl strategy with domain prioritization. @@ -1816,9 +1851,13 @@ class AsyncWebFetcher: if unlimited_pages: max_pages = 999999 # Effectively unlimited all_pages = [] - visited = set() + visited = set(initial_visited) if initial_visited else set() base_domain = self._get_domain(start_url) + # Log resume info + if initial_visited: + logger.info(f"Resuming with {len(initial_visited)} previously visited pages") + # Track all discovered links: {url: {'anchor_texts': [str], 'seen_count': int, 'total_link_score': float, 'domain': str, 'is_same_domain': bool}} link_registry = {} @@ -1945,6 +1984,8 @@ class AsyncWebFetcher: link_registry[link_url]['total_link_score'] += link_score logger.info(f"Registered {len(link_registry)} unique links from target page (same-domain: {same_domain_count}, cross-domain: {cross_domain_count})") + if uris_total_callback: + uris_total_callback(len(link_registry) + 1) # +1 for target page # PHASE 3: Decide whether to crawl deeper # If depth=0 or we're at max_pages, stop @@ -2151,6 +2192,10 @@ class AsyncWebFetcher: link_score = self._score_link(child_url, anchor_text, query_keywords, keyword_variations) link_registry[child_url]['total_link_score'] += link_score + # Update total after processing child links + if uris_total_callback: + uris_total_callback(len(link_registry) + 1) + logger.info(f"Depth {current_depth} complete: crawled {pages_at_this_depth} pages") return self._finalize_results(all_pages) diff --git a/neopig.py b/neopig.py index 18fadf4..d7acb8e 100644 --- a/neopig.py +++ b/neopig.py @@ -21,7 +21,9 @@ Usage: import argparse import asyncio import hashlib +import json import logging +import os import sys from datetime import datetime, timezone from pathlib import Path @@ -41,14 +43,31 @@ from storage import ImageVault from database import Database from screenshot import ScreenshotCapture, ScreenshotConfig from domain_vault import VaultManager, DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault, extract_media_urls +from tqdm import tqdm -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s' -) logger = logging.getLogger(__name__) +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. @@ -60,6 +79,7 @@ class NeoPig: vault_path: str = "vault", user_agent: str = "neopig/1.0 (ethical image crawler)", screenshot_config: ScreenshotConfig = None, + fast_mode: bool = False, ): self.db = Database(db_path) self.vault = ImageVault(vault_path) @@ -71,7 +91,10 @@ class NeoPig: media_base_url='/media', linkpeek_base_url='/linkpeek', ) - self.fetcher = AsyncWebFetcher(user_agent=user_agent) + # 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 @@ -86,15 +109,98 @@ class NeoPig: 'duplicates_skipped': 0, 'screenshots_taken': 0, 'errors': 0, + 'bytes_downloaded': 0, # Total bytes fetched from network + 'bytes_stored': 0, # Unique bytes stored in vault } # Track seen media URLs to avoid re-processing self.seen_media: Set[str] = set() # 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 + + def _get_state_file(self, target_url: str) -> Path: + """Get path to state file for resume support.""" + parsed = urlparse(target_url) + domain = parsed.netloc.lower().replace('.', '-').replace(':', '-') + return self._state_dir / f"crawl-state-{domain}.json" + + 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 = { + 'target_url': target_url, + 'seen_media': list(self.seen_media), + 'seen_screenshots': list(self.seen_screenshots), + 'seen_pages': list(self.seen_pages), + # Note: skip_domains NOT persisted - domains may come back online + # It's saved during session for resume, but cleared on fresh runs + '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) + state_file = self._get_state_file(target_url) + with open(state_file, 'w') as f: + json.dump(state, f) + except Exception as e: + logger.debug(f"Failed to save state: {e}") + + def _load_state(self, target_url: str) -> bool: + """Load saved crawl state. Returns True if state was loaded.""" + state_file = self._get_state_file(target_url) + if not state_file.exists(): + return False + + try: + with open(state_file, 'r') as f: + state = json.load(f) + self.seen_media = set(state.get('seen_media', [])) + self.seen_screenshots = set(state.get('seen_screenshots', [])) + # In fast mode, skip already-crawled pages for speed + # In normal mode, re-fetch pages to detect content changes (git handles versioning) + if self.fast_mode: + self.seen_pages = set(state.get('seen_pages', [])) + # Note: skip_domains NOT loaded - domains may have come back online + 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, {len(self.seen_media)} media, {len(self.seen_screenshots)} screenshots") + else: + logger.info(f"Resuming: {len(self.seen_media)} media, {len(self.seen_screenshots)} screenshots (pages will be re-checked)") + return True + except Exception as e: + logger.warning(f"Could not load state: {e}") + return False + + def _clear_state(self, target_url: str): + """Clear state file after successful completion.""" + try: + state_file = self._get_state_file(target_url) + if state_file.exists(): + state_file.unlink() + except Exception: + pass + def _get_domain(self, url: str) -> str: """Extract domain from URL.""" parsed = urlparse(url) @@ -116,6 +222,7 @@ class NeoPig: domain = self._get_domain(url) html_vault = self.domain_vaults.get_html_vault(domain) is_changed, _ = await html_vault.archive_page(url, 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 @@ -123,28 +230,61 @@ class NeoPig: async def _archive_media_to_vault( self, url: str, - content: bytes, + md5_hash: str, + ext: str, page_url: str = '', ): - """Archive media to the media vault.""" + """Create symlink in domain media vault pointing to hash vault.""" domain = self._get_domain(url) - media_vault = self.domain_vaults.get_media_vault(domain) - is_new, _, _ = await media_vault.archive_media(url, content, page_url) - if is_new: - self._track_domain_stat(domain, 'media_new') - self.stats['media_new'] += 1 + # Domain media path: vault/media_vault/{domain}/{url_path} + parsed = urlparse(url) + url_path = parsed.path.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/{hash[:2]}/{hash}.{ext} + hash_vault_path = Path(self.vault_path) / md5_hash[:2] / f"{md5_hash}{ext}" + + # 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, - screenshot_data: bytes, + md5_hash: str, ): - """Archive screenshot to the linkpeek vault.""" + """Create symlink in domain linkpeek vault pointing to hash vault.""" domain = self._get_domain(url) - linkpeek_vault = self.domain_vaults.get_linkpeek_vault(domain) - is_new, _, _ = await linkpeek_vault.archive_screenshot(url, screenshot_data) - if is_new: - self._track_domain_stat(domain, 'screenshots_new') + # Domain screenshot path: vault/linkpeek_vault/{domain}/{url_path}.png + parsed = urlparse(url) + url_path = parsed.path.lstrip('/') or 'index' + url_path = url_path.replace('/', '_') + '.png' + domain_ss_dir = Path(self.vault_path) / 'linkpeek_vault' / domain + domain_ss_path = domain_ss_dir / url_path + + # Hash vault path: vault/{hash[:2]}/{hash}.png + hash_vault_path = Path(self.vault_path) / md5_hash[:2] / f"{md5_hash}.png" + + # 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.""" @@ -208,31 +348,56 @@ class NeoPig: logger.info(f"Keywords: {keywords}") logger.info(f"Depth: {'unlimited' if depth == -1 else depth}") + # Load saved state if exists (resume support) + self._load_state(target_uri) + # Track timing for stats start_time = datetime.now(timezone.utc) - last_stats_time = start_time - stats_running = True - # Background task to emit stats every 15 seconds - async def stats_reporter(): - nonlocal last_stats_time - while stats_running: - await asyncio.sleep(15) - if not stats_running: - break - elapsed = (datetime.now(timezone.utc) - start_time).total_seconds() - rate = self.stats['media_downloaded'] / elapsed * 60 if elapsed > 0 else 0 - logger.info(f"=== CRAWL STATS ({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 | " - f"Vault size: {len(self.seen_media)}") + 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" - stats_task = asyncio.create_task(stats_reporter()) + # Track total URIs discovered for progress bar + uris_total = [1] # Start with 1 for target page, use list for mutability in closure + + def on_uris_total(total: int): + uris_total[0] = total + self.pbar.total = total + self.pbar.refresh() + + # Create progress bar with actual bar display + # Start with previous progress if resuming + initial_pages = len(self.seen_pages) + self.pbar = tqdm( + total=max(1, initial_pages), + initial=initial_pages, + unit="pages", + dynamic_ncols=True, + bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}] {postfix}', + mininterval=0.1, + ) + self.pbar.set_postfix_str( + f"media: {self.stats['media_downloaded']}/{self.stats['media_found']}, " + f"ss: {self.stats['screenshots_taken']}, " + f"err: {self.stats['errors']}, " + f"{format_size(self.stats['bytes_stored'])} stored ({format_size(self.stats['bytes_downloaded'])} fetched)" + ) + + def update_pbar(): + self.pbar.set_postfix_str( + f"media: {self.stats['media_downloaded']}/{self.stats['media_found']}, " + f"ss: {self.stats['screenshots_taken']}, " + f"err: {self.stats['errors']}, " + f"{format_size(self.stats['bytes_stored'])} stored ({format_size(self.stats['bytes_downloaded'])} fetched)" + ) + self.pbar.refresh() # Media callback - called for each discovered media item async def on_media_discovered(item: Dict[str, Any]): @@ -242,23 +407,27 @@ class NeoPig: self.seen_media.add(url) self.stats['media_found'] += 1 + update_pbar() if download_media: await self._process_media_item(item, job_id, keywords) + update_pbar() + self._save_state(target_uri) # Note: Screenshots are now captured per-page in on_page_fetched, # not per-media-item, to honor crawl delay as a unit # Progress callback async def on_progress(msg: str): self.stats['pages_crawled'] += 1 - if self.stats['pages_crawled'] % 10 == 0: - logger.info(f"Progress: {self.stats['pages_crawled']} pages, " - f"{self.stats['media_found']} media found, " - f"{self.stats['media_downloaded']} downloaded") + self.pbar.update(1) + update_pbar() # Page callback - archive raw HTML to vault and capture screenshot # Screenshot happens here (same crawl delay window as page fetch) async def on_page_fetched(url: str, html: str): + # Track this page as crawled for resume support + self.seen_pages.add(url) + # For now, archive without media URL rewriting (we'd need to download media first) # TODO: Build media_mappings after media is downloaded await self._archive_page_to_vault(url, html, media_mappings=None) @@ -267,6 +436,9 @@ class NeoPig: if self.screenshot_config.enabled: await self._capture_page_screenshot(url, job_id, page_title='') + # Save state periodically for resume support + self._save_state(target_uri) + # Run the crawl pages = await self.fetcher.fetch_with_depth( start_url=target_uri, @@ -277,17 +449,14 @@ class NeoPig: media_callback=on_media_discovered, progress_callback=on_progress, page_callback=on_page_fetched, + uris_total_callback=on_uris_total, + initial_visited=self.seen_pages if self.seen_pages else None, ) self.stats['pages_crawled'] = len(pages) - # Stop the stats reporter - stats_running = False - stats_task.cancel() - try: - await stats_task - except asyncio.CancelledError: - pass + # Close progress bar + self.pbar.close() # Calculate final stats elapsed = (datetime.now(timezone.utc) - start_time).total_seconds() @@ -308,6 +477,11 @@ class NeoPig: f"Errors: {self.stats['errors']}") logger.info(f" Rate: {rate:.1f}/min | Total time: {elapsed:.1f}s") + # Keep state file for future delta crawls (don't clear) + # Force a final save to ensure latest state is persisted + self._items_since_save = self._state_save_interval # Force save + self._save_state(target_uri) + return self.stats async def _process_media_item( @@ -382,6 +556,7 @@ class NeoPig: return 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): @@ -409,9 +584,10 @@ class NeoPig: # 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']) - # Archive to domain media vault (git-tracked) - await self._archive_media_to_vault(media_uri, result['data'], 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) # Record in database with full context and skeleton key await self.db.create_media_record( @@ -447,7 +623,12 @@ class NeoPig: job_id: int, page_title: str = '', ): - """Capture and store a screenshot of a page.""" + """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. + """ if not self.screenshot_config.enabled: return @@ -457,19 +638,28 @@ class NeoPig: self.seen_screenshots.add(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) + result = await self.screenshot.capture(page_uri) if not result: return md5_hash = result['md5_hash'] screenshot_data = result['data'] + screenshot_size = len(screenshot_data) + + # Screenshots are fetched by headless browser (network traffic) + self.stats['bytes_downloaded'] += screenshot_size # Store in MD5 vault (for deduplication) if not await self.vault.exists(md5_hash): await self.vault.store(md5_hash, screenshot_data, 'png') + self.stats['bytes_stored'] += screenshot_size - # Archive to linkpeek vault (git-tracked by URL path) - await self._archive_screenshot_to_vault(page_uri, screenshot_data) + # Create symlink in linkpeek vault pointing to hash vault + await self._archive_screenshot_to_vault(page_uri, md5_hash) # Record in database as screenshot type await self.db.create_media_record( @@ -529,7 +719,7 @@ async def main(): parser.add_argument( "targets", - nargs="+", + nargs="*", help="Target URI(s) to crawl (e.g., https://example.com https://other.com)" ) @@ -613,10 +803,50 @@ async def main(): 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, ignoring any saved resume state" + ) + + parser.add_argument( + "--fast", + action="store_true", + help="Fast mode: no crawl delay (use for sites without robots.txt)" + ) + args = parser.parse_args() - if args.verbose: - logging.getLogger().setLevel(logging.DEBUG) + # 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 + + # 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 = { @@ -634,25 +864,36 @@ async def main(): width=args.screenshot_width, height=args.screenshot_height, delay=args.screenshot_delay, + engine=args.screenshot_engine, ) if args.screenshot: - logger.info(f"Screenshots enabled: {screenshot_config.width}x{screenshot_config.height}, delay={screenshot_config.delay}ms") + 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, ) await pig.init() - # Load previously crawled media URIs to enable resume - crawled_media = await pig.db.get_crawled_media_uris() + if args.fast: + logger.info("Fast mode: no crawl delay (ignoring robots.txt)") - if crawled_media: - logger.info(f"Resuming: {len(crawled_media)} media already crawled") - pig.seen_media = crawled_media + # 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 URIs from DB to enable resume + crawled_media = await pig.db.get_crawled_media_uris() + if crawled_media: + logger.info(f"Resuming: {len(crawled_media)} media already in database") + pig.seen_media = crawled_media # Crawl all targets concurrently async def crawl_target(target: str): diff --git a/requirements.txt b/requirements.txt index 8545a8a..1095087 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,5 +16,9 @@ fastapi>=0.104.0 uvicorn>=0.24.0 python-multipart>=0.0.6 +# Progress bar +tqdm>=4.66.0 + # Optional pillow>=10.0.0 +html2text>=2024.2.26 diff --git a/screenshot.py b/screenshot.py index deb5a9c..540d133 100644 --- a/screenshot.py +++ b/screenshot.py @@ -4,19 +4,46 @@ Screenshot capture module for neopig. Wraps uri2png for async-compatible page screenshots. Screenshots are stored in vault with MD5 hash like other media. + +Supported engines (in order of preference): + - wkhtmltoimage: Fast, lightweight, uses Qt WebKit. Install: apt install wkhtmltopdf + - cutycapt: Fast, lightweight, uses Qt WebKit. Install: apt install cutycapt + - playwright-webkit: WebKit via Playwright (lighter than Chromium) + - playwright-firefox: Firefox via Playwright + - playwright-chromium: Chromium via Playwright (heaviest, but most compatible) + - selenium-*: Various Selenium drivers + +The module auto-detects available engines and picks the lightest one, +or you can specify an engine explicitly. """ import asyncio import hashlib import logging -import subprocess -import tempfile -from dataclasses import dataclass +import shutil +from dataclasses import dataclass, field from pathlib import Path -from typing import Optional +from typing import Optional, List, Dict, Any +import tempfile logger = logging.getLogger(__name__) +# Engine preference order - lightest/fastest first +ENGINE_PREFERENCE = [ + 'wkhtmltoimage', # Native Qt WebKit - very fast, no browser download + 'cutycapt', # Native Qt WebKit - very fast, no browser download + 'playwright-webkit', # WebKit via Playwright - lighter than Chromium + 'playwright-firefox', + 'playwright-chromium', + 'playwright', # Default Playwright (Chromium) + 'selenium-chrome', + 'selenium-firefox', + 'selenium', +] + +# Native tools that don't require browser downloads +NATIVE_ENGINES = {'wkhtmltoimage', 'cutycapt'} + @dataclass class ScreenshotConfig: @@ -25,48 +52,134 @@ class ScreenshotConfig: width: int = 1280 height: int = 1024 delay: int = 1000 # ms after DOM load + timeout: int = 30000 # ms total timeout user_agent: Optional[str] = None + engine: Optional[str] = None # None = auto-detect best available + full_page: bool = False class ScreenshotCapture: """ Async screenshot capture using uri2png. - Since uri2png uses GTK main loop, we run it in a subprocess - to avoid blocking the async event loop. + Supports multiple backends with automatic selection of the lightest + available engine. Native tools (wkhtmltoimage, cutycapt) are preferred + over browser-based solutions. + + Usage: + config = ScreenshotConfig(enabled=True, engine='wkhtmltoimage') + capture = ScreenshotCapture(config) + result = await capture.capture('https://example.com') """ def __init__(self, config: ScreenshotConfig = None): self.config = config or ScreenshotConfig() - self._uri2png_available = None + self._engine = None + self._engine_name = None + self._available_engines: Optional[List[Dict[str, str]]] = None + self._initialized = False + + async def _get_available_engines(self) -> List[Dict[str, str]]: + """Get list of available screenshot engines.""" + if self._available_engines is not None: + return self._available_engines + + def _check(): + try: + from uri2png import get_available_engines + return get_available_engines() + except ImportError: + return [] + + self._available_engines = await asyncio.to_thread(_check) + return self._available_engines + + async def _select_engine(self) -> Optional[str]: + """Select the best available engine based on preference order.""" + available = await self._get_available_engines() + available_names = {e['name'] for e in available} + + # If user specified an engine, try to use it + if self.config.engine: + if self.config.engine in available_names: + return self.config.engine + else: + logger.warning(f"Requested engine '{self.config.engine}' not available") + logger.info(f"Available engines: {', '.join(available_names)}") + + # Check native tools first (they're fast and don't need browser downloads) + for engine in ENGINE_PREFERENCE: + if engine in available_names: + # For native engines, verify the binary exists + if engine in NATIVE_ENGINES: + binary = 'wkhtmltoimage' if engine == 'wkhtmltoimage' else 'cutycapt' + if shutil.which(binary): + return engine + else: + logger.debug(f"Engine {engine} listed but binary not found") + continue + return engine + + return None + + async def initialize(self) -> bool: + """Initialize the screenshot engine.""" + if self._initialized: + return self._engine is not None + + engine_name = await self._select_engine() + if not engine_name: + logger.warning("No screenshot engine available") + logger.info("Install one of: wkhtmltopdf, cutycapt, or playwright") + self._initialized = True + return False + + def _create_engine(): + try: + from uri2png import create_engine + # Pass options as kwargs to create_engine + return create_engine( + engine_name, + width=self.config.width, + height=self.config.height, + delay=self.config.delay, + timeout=self.config.timeout, + full_page=self.config.full_page, + user_agent=self.config.user_agent, + ) + except Exception as e: + logger.warning(f"Failed to create engine '{engine_name}': {e}") + return None + + self._engine = await asyncio.to_thread(_create_engine) + self._engine_name = engine_name + self._initialized = True + + if self._engine: + logger.info(f"Screenshot engine: {engine_name}") + return True + return False async def is_available(self) -> bool: - """Check if uri2png is installed and available.""" - if self._uri2png_available is not None: - return self._uri2png_available + """Check if screenshots are available.""" + if not self._initialized: + await self.initialize() + return self._engine is not None - try: - proc = await asyncio.create_subprocess_exec( - 'python', '-c', 'from uri2png import Uri2Png', - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.DEVNULL, - ) - await proc.wait() - self._uri2png_available = proc.returncode == 0 - except Exception: - self._uri2png_available = False + def get_engine_name(self) -> Optional[str]: + """Get the name of the active engine.""" + return self._engine_name - if not self._uri2png_available: - logger.warning("uri2png not available - screenshots disabled") - - return self._uri2png_available + async def list_engines(self) -> List[Dict[str, str]]: + """List all available screenshot engines.""" + return await self._get_available_engines() async def capture(self, uri: str) -> Optional[dict]: """ Capture screenshot of a URI. Returns: - Dict with 'data', 'md5_hash', 'mime_type' or None on failure + Dict with 'data', 'md5_hash', 'mime_type', 'engine' or None on failure """ if not self.config.enabled: return None @@ -74,94 +187,55 @@ class ScreenshotCapture: if not await self.is_available(): return None - # Create temp file for screenshot - with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp: - tmp_path = tmp.name - try: - # Build uri2png command - cmd = [ - 'python', '-c', - f''' -from uri2png import Uri2Png -Uri2Png( - uri="{uri}", - filepath="{tmp_path}", - width={self.config.width}, - height={self.config.height}, - delay={self.config.delay}, - user_agent={repr(self.config.user_agent)}, -).capture() -''' - ] - - # Run with timeout (uri2png can hang on bad URLs) - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) + # Create temp file for output + with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as f: + output_path = f.name try: - stdout, stderr = await asyncio.wait_for( - proc.communicate(), - timeout=30.0 # 30 second timeout + # Capture - API is capture(url, output_path) + capture_coro = self._engine.capture(uri, output_path) + + # All uri2png engines return coroutines + result = await asyncio.wait_for( + capture_coro, + timeout=self.config.timeout / 1000 + 5 ) - except asyncio.TimeoutError: - proc.kill() - await proc.wait() - logger.warning(f"Screenshot timeout: {uri}") - return None - if proc.returncode != 0: - logger.warning(f"Screenshot failed ({proc.returncode}): {uri}") - if stderr: - logger.debug(f"stderr: {stderr.decode()}") - return None - - # Read screenshot data - path = Path(tmp_path) - - def _check_file(): - if not path.exists(): + if not result.success: + logger.warning(f"Screenshot failed: {uri} - {result.error}") return None - size = path.stat().st_size - if size == 0: + + # Read bytes from output file + data = Path(output_path).read_bytes() + if not data: + logger.warning(f"Screenshot empty: {uri}") return None - return path.read_bytes() - data = await asyncio.to_thread(_check_file) - if data is None: - logger.warning(f"Screenshot empty: {uri}") - return None - md5_hash = hashlib.md5(data).hexdigest() + md5_hash = hashlib.md5(data).hexdigest() + logger.debug(f"Screenshot captured ({self._engine_name}): {uri} -> {md5_hash}") - logger.debug(f"Screenshot captured: {uri} -> {md5_hash}") - - return { - 'data': data, - 'md5_hash': md5_hash, - 'mime_type': 'image/png', - 'size': len(data), - 'source_uri': uri, - } - - except Exception as e: - logger.warning(f"Screenshot error for {uri}: {e}") - return None - - finally: - # Cleanup temp file (sync unlink is fine in finally - small operation) - def _cleanup(): + return { + 'data': data, + 'md5_hash': md5_hash, + 'mime_type': 'image/png', + 'size': len(data), + 'source_uri': uri, + 'engine': self._engine_name, + } + finally: + # Clean up temp file try: - Path(tmp_path).unlink(missing_ok=True) + Path(output_path).unlink(missing_ok=True) except Exception: pass - try: - await asyncio.to_thread(_cleanup) - except Exception: - pass + except asyncio.TimeoutError: + logger.warning(f"Screenshot timeout: {uri}") + return None + except Exception as e: + logger.warning(f"Screenshot error for {uri}: {e}") + return None async def capture_to_file(self, uri: str, output_path: str) -> bool: """ @@ -180,3 +254,27 @@ Uri2Png( except Exception as e: logger.warning(f"Failed to write screenshot: {e}") return False + + async def cleanup(self): + """Cleanup engine resources.""" + if self._engine: + try: + def _cleanup(): + if hasattr(self._engine, 'cleanup'): + self._engine.cleanup() + await asyncio.to_thread(_cleanup) + except Exception as e: + logger.debug(f"Engine cleanup error: {e}") + + +async def list_available_engines() -> List[Dict[str, str]]: + """List all available screenshot engines (utility function).""" + capture = ScreenshotCapture() + return await capture.list_engines() + + +async def get_best_engine() -> Optional[str]: + """Get the best available engine name (utility function).""" + capture = ScreenshotCapture(ScreenshotConfig(enabled=True)) + await capture.initialize() + return capture.get_engine_name()