engineering/pig.py
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
pig.py/neopig.py
Russell Ballestrini 027f5afc57 Add public domain license and headers to all source files 
- Add LICENSE file (public domain, permacomputer values)
- Add license headers to all 27 Python source files
- Restore full about_ch9_p2 with "terminated 🟣" content
- Translate about_ch9_p2 to all 26 languages

The permacomputer is community-owned infrastructure optimized around:
TRUTH, FREEDOM, HARMONY, LOVE
2026-01-08 15:38:28 -05:00

2962 lines
114 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# PUBLIC DOMAIN - NO LICENSE, NO WARRANTY
#
# This is free public domain software for the public good of a permacomputer hosted
# at permacomputer.com - an always-on computer by the people, for the people. One
# which is durable, easy to repair, and distributed like tap water for machine
# learning intelligence.
#
# The permacomputer is community-owned infrastructure optimized around four values:
#
# TRUTH - First principles, math & science, open source code freely distributed
# FREEDOM - Voluntary partnerships, freedom from tyranny & corporate control
# HARMONY - Minimal waste, self-renewing systems with diverse thriving connections
# LOVE - Be yourself without hurting others, cooperation through natural law
#
# Anyone is free to copy, modify, publish, use, compile, sell, or distribute this
# software, either in source code form or as a compiled binary, for any purpose,
# commercial or non-commercial, and by any means.
#
# NO WARRANTY. THE SOFTWARE IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
#
# Copyright 2025 TimeHexOn & foxhop & russell@unturf
# https://git.unturf.com/engineering/unturf/pig.py
# Side quest 9/21: The pig eats everything. Everything.
"""
neopig - Neo Python Image Grabber
Full domain crawler that:
1. Crawls entire domain (no depth limit)
2. Extracts all images/videos via multiple methods
3. Uses HEAD checks for extensionless URLs
4. Stores in vault (MD5 dedupe)
5. Queues for Qwen 3 VL analysis
6. Indexes metadata in SQLite
Full-domain async media crawler with deduplication
https://russell.ballestrini.net/python-image-grabber-pig-py/
Usage:
python neopig.py https://example.com "rick and morty" --mode images
python neopig.py https://example.com --mode all --depth -1 # Full domain slurp
# grow food not lawn
# - The Sign Maker
"""
import argparse
import asyncio
import atexit
import hashlib
import json
import logging
import multiprocessing
import os
import signal
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any, Optional, Set, Tuple
from miniuri import Uri
from neopig.async_web_fetcher import (
AsyncWebFetcher,
CrawlMode,
MediaItem,
extract_media_from_html,
get_media_type_from_extension,
get_media_type_from_mime,
)
from neopig.filevault import AsyncVault, hash_to_path
from neopig.database import Database, SCORE_SCREENSHOT, SCORE_OG_IMAGE, SCORE_THUMBNAIL, SCORE_FULL_RES
from neopig.screenshot import ScreenshotCapture, ScreenshotConfig
from neopig.domain_vault import VaultManager, DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault, extract_media_urls
from neopig.repo import detect_vcs, clone_repo_async, pull_repo_async, get_repo_path, walk_files, get_commit_hash, get_file_language, is_binary_file
from tqdm import tqdm
logger = logging.getLogger(__name__)
# Environment configuration
NO_VCS = os.environ.get("NEOPIG_NO_VCS", "").lower() in ("1", "true", "yes")
# Track child processes for cleanup on exit
_CHILD_PROCESSES: List = []
def _cleanup_children():
"""Kill all tracked child processes."""
for proc in _CHILD_PROCESSES:
if proc and proc.poll() is None:
try:
os.kill(proc.pid, signal.SIGTERM)
proc.wait(timeout=2)
except:
try:
os.kill(proc.pid, signal.SIGKILL)
except:
pass
# atexit runs even when asyncio overrides signal handlers
atexit.register(_cleanup_children)
# Live media event queue - push here after disk save, SSE reads from here
# Format: {'md5_hash': str, 'media_type': str, 'file_size': int, 'alt_text': str, ...}
LIVE_MEDIA_QUEUE: asyncio.Queue = None # Initialized lazily
def get_live_queue() -> asyncio.Queue:
"""Get or create the live media queue."""
global LIVE_MEDIA_QUEUE
if LIVE_MEDIA_QUEUE is None:
LIVE_MEDIA_QUEUE = asyncio.Queue(maxsize=1000)
return LIVE_MEDIA_QUEUE
def emit_live_media(media_info: Dict[str, Any]):
"""Emit media to live feed (non-blocking)."""
try:
queue = get_live_queue()
queue.put_nowait(media_info)
except asyncio.QueueFull:
pass # Drop if queue is full (live feed will catch up from DB)
# Per-job log handlers
LOGS_PATH = Path("data/logs")
JOB_LOG_HANDLERS: Dict[int, logging.FileHandler] = {}
def start_job_logging(job_id: int) -> None:
"""Start capturing logs for a crawl job to file."""
LOGS_PATH.mkdir(parents=True, exist_ok=True)
log_file = LOGS_PATH / f"{job_id}.log"
handler = logging.FileHandler(log_file, mode='w', encoding='utf-8')
handler.setLevel(logging.INFO)
handler.setFormatter(logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s', datefmt='%H:%M:%S'))
logging.getLogger().addHandler(handler)
JOB_LOG_HANDLERS[job_id] = handler
def stop_job_logging(job_id: int) -> None:
"""Stop capturing logs for a crawl job."""
handler = JOB_LOG_HANDLERS.pop(job_id, None)
if handler:
handler.close()
logging.getLogger().removeHandler(handler)
class AppendOnlyStateLog:
"""Append-only state log for fast crawl state tracking.
Format (one record per line):
P <url> # page seen
M <md5_hash> <url> # media downloaded
S <url> # screenshot taken
D <domain> # skip domain
X <key> <json> # stats checkpoint
Benefits:
- No JSON parsing on write (just append)
- No locking needed for single writer
- Fast resume by scanning lines
- Works with tail -f for monitoring
"""
def __init__(self, path: Path):
self.path = path
self._file = None
def open(self):
"""Open log file for appending."""
self.path.parent.mkdir(parents=True, exist_ok=True)
self._file = open(self.path, 'a', encoding='utf-8', buffering=1) # line buffered
def close(self):
"""Close log file."""
if self._file:
self._file.close()
self._file = None
def __enter__(self):
self.open()
return self
def __exit__(self, *args):
self.close()
def page(self, url: str):
"""Record page as seen."""
if self._file:
self._file.write(f"P {url}\n")
def media(self, md5_hash: str, url: str):
"""Record media as downloaded."""
if self._file:
self._file.write(f"M {md5_hash} {url}\n")
def screenshot(self, url: str):
"""Record screenshot as taken."""
if self._file:
self._file.write(f"S {url}\n")
def skip_domain(self, domain: str):
"""Record domain to skip."""
if self._file:
self._file.write(f"D {domain}\n")
def stats(self, stats_dict: Dict[str, Any]):
"""Record stats checkpoint."""
if self._file:
self._file.write(f"X stats {json.dumps(stats_dict)}\n")
def load(self) -> Dict[str, Any]:
"""Load state from log file.
Returns dict with:
- seen_pages: set of URLs
- seen_media: dict of url -> md5_hash
- seen_screenshots: set of URLs
- skip_domains: set of domains
- stats: last stats checkpoint (or empty dict)
"""
result = {
'seen_pages': set(),
'seen_media': {},
'seen_screenshots': set(),
'skip_domains': set(),
'stats': {},
}
if not self.path.exists():
return result
with open(self.path, 'r', encoding='utf-8') as f:
for line in f:
line = line.rstrip('\n')
if not line:
continue
parts = line.split(' ', 2)
if len(parts) < 2:
continue
record_type = parts[0]
if record_type == 'P':
result['seen_pages'].add(parts[1])
elif record_type == 'M' and len(parts) >= 3:
result['seen_media'][parts[2]] = parts[1] # url -> hash
elif record_type == 'S':
result['seen_screenshots'].add(parts[1])
elif record_type == 'D':
result['skip_domains'].add(parts[1])
elif record_type == 'X' and len(parts) >= 3:
try:
result['stats'] = json.loads(parts[2])
except json.JSONDecodeError:
pass
return result
def get_state_log_path(domain: str) -> Path:
"""Get append-only state log path for a domain."""
safe_domain = domain.replace('://', '-').replace('/', '-').replace('.', '-')
return Path(f"data/{safe_domain}.log")
def get_state_file_path(domain: str) -> Path:
"""Get unified state file path for a domain.
Args:
domain: Domain name (e.g., 'example.com')
Returns:
Path like data/{domain}.state
"""
safe_domain = domain.replace('://', '-').replace('/', '-').replace('.', '-')
return Path(f"data/{safe_domain}.state")
def rotate_state_file(state_path: Path, preserve_keys: List[str] = None) -> Optional[Path]:
"""Rotate state file with optional selective preservation.
Args:
state_path: Path to state file
preserve_keys: If provided, rotate then copy back these keys from rotated file.
If None, just rotate (full fresh start).
Returns:
Path to rotated file, or None if no rotation needed.
"""
if not state_path.exists():
return None
# Find next available rotation number
i = 1
while Path(f"{state_path}.{i}").exists():
i += 1
rotated = Path(f"{state_path}.{i}")
state_path.rename(rotated)
logger.info(f"Rotated state file to {rotated}")
# If preserve_keys specified, copy back those keys from rotated file
if preserve_keys:
try:
import json
old_state = json.loads(rotated.read_text())
new_state = {k: v for k, v in old_state.items() if k in preserve_keys}
if new_state:
state_path.write_text(json.dumps(new_state, indent=2))
logger.info(f"Preserved keys: {list(new_state.keys())}")
except Exception as e:
logger.warning(f"Could not preserve state keys: {e}")
return rotated
class TqdmLoggingHandler(logging.Handler):
"""Logging handler that writes through tqdm to avoid progress bar corruption."""
def emit(self, record):
try:
msg = self.format(record)
tqdm.write(msg)
except Exception:
self.handleError(record)
def setup_logging(level=logging.INFO):
"""Setup logging to work with tqdm progress bars."""
handler = TqdmLoggingHandler()
handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
root = logging.getLogger()
root.handlers = [handler]
root.setLevel(level)
class NeoPig:
"""
Neo Python Image Grabber - async media crawler with deduplication.
"""
def __init__(
self,
db_path: str = "neopig.db",
vault_path: str = "vault",
user_agent: str = "neopig/1.0 (ethical image crawler)",
screenshot_config: ScreenshotConfig = None,
fast_mode: bool = False,
trim_wrapper: bool = False,
):
self.db = Database(db_path)
self.vault = AsyncVault(vault_path)
self.trim_wrapper = trim_wrapper
# Triple filevault system: html_vault/, media_vault/, and linkpeek_vault/
self.domain_vaults = VaultManager(
html_vault_base=f"{vault_path}/html_vault",
media_vault_base=f"{vault_path}/media_vault",
linkpeek_vault_base=f"{vault_path}/linkpeek_vault",
media_base_url='/media',
linkpeek_base_url='/linkpeek',
)
# Fast mode: no crawl delay, short timeouts (for sites without robots.txt)
self.fast_mode = fast_mode
crawl_delay = 0.0 if fast_mode else 2.0
self.fetcher = AsyncWebFetcher(user_agent=user_agent, default_crawl_delay=crawl_delay, fast_mode=fast_mode)
self.screenshot = ScreenshotCapture(screenshot_config or ScreenshotConfig())
self.screenshot_config = screenshot_config or ScreenshotConfig()
self.vault_path = vault_path
# Track stats
self.stats = {
'pages_crawled': 0,
'pages_pending': 0, # Queue size for ETA calculation
'pages_changed': 0, # Pages with content changes (for git commit)
'media_found': 0,
'media_downloaded': 0,
'media_new': 0, # New media (for git commit)
'duplicates_skipped': 0,
'content_exists': 0, # Same content (MD5) already in vault
'screenshots_taken': 0,
'errors': 0,
'bytes_downloaded': 0, # Total bytes fetched from network
'bytes_stored': 0, # Unique bytes stored in vault
'crawl_started': 0, # Unix timestamp for rate calculation
}
# Track seen media URLs -> md5_hash (to add page context without re-download)
self.seen_media: Dict[str, str] = {}
# Track screenshotted pages to avoid duplicates
self.seen_screenshots: Set[str] = set()
# Track crawled page URLs for resume support
self.seen_pages: Set[str] = set()
# Track per-domain stats for vault commits
self._domain_stats: Dict[str, Dict[str, int]] = {} # domain -> {pages_changed, media_new, screenshots_new}
# Progress bar
self.pbar: Optional[tqdm] = None
# Resume support - state files go in data/
self._state_dir = Path("data")
self._state_dir.mkdir(exist_ok=True)
self._state_save_interval = 10
self._items_since_save = 0
# Append-only state log (faster than JSON)
self._state_log: Optional[AppendOnlyStateLog] = None
def _get_state_file(self, target_url: str) -> Path:
"""Get unified state file path for domain."""
parsed = Uri(target_url)
domain = parsed.hostname.lower()
return get_state_file_path(domain)
def _get_state_log(self, target_url: str) -> AppendOnlyStateLog:
"""Get append-only state log for domain."""
parsed = Uri(target_url)
domain = parsed.hostname.lower()
return AppendOnlyStateLog(get_state_log_path(domain))
def _save_state(self, target_url: str):
"""Save crawl state for resume."""
self._items_since_save += 1
if self._items_since_save < self._state_save_interval:
return
self._items_since_save = 0
state_file = self._get_state_file(target_url)
# Load existing state to preserve other keys (e.g., backfill state)
try:
full_state = json.loads(state_file.read_text()) if state_file.exists() else {}
except Exception:
full_state = {}
# Update crawl section
full_state['crawl'] = {
'target_url': target_url,
'seen_media': list(self.seen_media),
'seen_screenshots': list(self.seen_screenshots),
'seen_pages': list(self.seen_pages),
'skip_domains': list(self.fetcher.skip_domains),
'stats': self.stats,
'timestamp': datetime.now(timezone.utc).isoformat(),
}
try:
self._state_dir.mkdir(parents=True, exist_ok=True)
with open(state_file, 'w') as f:
json.dump(full_state, f, indent=2)
except Exception as e:
logger.debug(f"Failed to save state: {e}")
def _get_hydra_state_file(self, domain: str) -> Path:
"""Get path to hydra state file for a domain (never rotated)."""
safe_domain = domain.replace('/', '_').replace(':', '_')
return self._state_dir / f"hydra-{safe_domain}.json"
def _load_hydra_state(self, domain: str) -> Dict[str, Any]:
"""Load hydra state - feeds and URLs seen (never cleared)."""
state_file = self._get_hydra_state_file(domain)
if not state_file.exists():
return {'seen_urls': {}, 'feeds': {}}
try:
return json.loads(state_file.read_text())
except Exception as e:
logger.warning(f"Failed to load hydra state: {e}")
return {'seen_urls': {}, 'feeds': {}}
def _save_hydra_state(self, domain: str, state: Dict[str, Any]):
"""Save hydra state (persists across crawls, never rotated)."""
state_file = self._get_hydra_state_file(domain)
try:
self._state_dir.mkdir(parents=True, exist_ok=True)
state['last_updated'] = datetime.now(timezone.utc).isoformat()
with open(state_file, 'w') as f:
json.dump(state, f, indent=2)
except Exception as e:
logger.warning(f"Failed to save hydra state: {e}")
def _get_known_feeds(self, domain: str) -> List[str]:
"""Get list of known feed URLs for a domain."""
state = self._load_hydra_state(domain)
return list(state.get('feeds', {}).keys())
def _add_hydra_feeds(self, domain: str, feed_urls: List[str]):
"""Add newly discovered feed URLs to hydra state."""
if not feed_urls:
return
state = self._load_hydra_state(domain)
feeds = state.get('feeds', {})
now = datetime.now(timezone.utc).isoformat()
new_count = 0
for url in feed_urls:
if url not in feeds:
feeds[url] = {'discovered': now}
new_count += 1
if new_count:
state['feeds'] = feeds
self._save_hydra_state(domain, state)
logger.info(f"Hydra: Added {new_count} new feeds (total: {len(feeds)})")
def _filter_new_hydra_urls(self, domain: str, feed_items: List) -> List:
"""Filter feed items to only return NEW URLs not seen before.
Also updates hydra state with newly seen URLs.
Returns list of new FeedItem objects.
"""
state = self._load_hydra_state(domain)
seen_urls = state.get('seen_urls', {})
new_items = []
now = datetime.now(timezone.utc).isoformat()
for item in feed_items:
url = item.url
if url not in seen_urls:
new_items.append(item)
seen_urls[url] = {
'first_seen': now,
'title': item.title,
'published': item.published,
}
if new_items:
state['seen_urls'] = seen_urls
self._save_hydra_state(domain, state)
logger.info(f"Hydra: {len(new_items)} NEW URLs (of {len(feed_items)} total)")
else:
logger.info(f"Hydra: No new URLs found (all {len(feed_items)} already seen)")
return new_items
async def _fetch_hydra_urls(self, target_uri: str, discover_new: bool = False) -> Optional[List[str]]:
"""Fetch URLs from known feeds (self-healing) and optionally discover new feeds.
Args:
target_uri: Target URI being crawled
discover_new: If True, also discover new feeds from target page
Returns:
List of NEW URLs to inject into crawl, or None if no feeds known
"""
domain = Uri(target_uri).hostname
known_feeds = self._get_known_feeds(domain)
# Discover new feeds if requested (--hydra mode)
if discover_new:
logger.info("=== HYDRA MODE: Discovering feeds/sitemaps ===")
# fetch_feeds will discover from common locations
feed_items = await self.fetcher.fetch_feeds(target_uri)
if feed_items:
# Extract discovered feed URLs and persist them
# The fetcher tracks which feeds it found
discovered_feeds = getattr(self.fetcher, '_last_discovered_feeds', [])
if discovered_feeds:
self._add_hydra_feeds(domain, discovered_feeds)
known_feeds = self._get_known_feeds(domain) # Refresh
# If we have known feeds, fetch them for new URLs
if known_feeds:
logger.info(f"Hydra: Checking {len(known_feeds)} known feeds for new content")
feed_items = await self.fetcher.fetch_feeds(
target_uri,
feed_urls=known_feeds, # Pass known feeds directly
)
if feed_items:
new_items = self._filter_new_hydra_urls(domain, feed_items)
if new_items:
return [item.url for item in new_items]
return None
def _load_state(self, target_url: str) -> bool:
"""Load saved crawl state. Returns True if state was loaded.
Tries append-only log first (faster), falls back to legacy JSON.
Note: seen_media and seen_screenshots are NOT loaded from state file -
they come from the database (source of truth for successful downloads).
Only seen_pages (for fast mode) and stats are loaded from state.
"""
# Try append-only log first (preferred)
state_log = self._get_state_log(target_url)
if state_log.path.exists():
state = state_log.load()
if self.fast_mode:
self.seen_pages = state['seen_pages']
self.fetcher.skip_domains = state['skip_domains']
saved_stats = state.get('stats', {})
for key in self.stats:
if key in saved_stats:
self.stats[key] = saved_stats[key]
if self.fast_mode:
logger.info(f"Fast resume from log: skipping {len(self.seen_pages)} pages")
else:
logger.info(f"Resume from log: loading stats")
return True
# Fall back to legacy JSON state file
state_file = self._get_state_file(target_url)
if not state_file.exists():
return False
try:
full_state = json.loads(state_file.read_text())
state = full_state.get('crawl', {})
if not state:
return False
# In fast mode, skip already-crawled pages for speed
if self.fast_mode:
self.seen_pages = set(state.get('seen_pages', []))
saved_stats = state.get('stats', {})
for key in self.stats:
if key in saved_stats:
self.stats[key] = saved_stats[key]
if self.fast_mode:
logger.info(f"Fast resume: skipping {len(self.seen_pages)} pages (media/screenshots from DB)")
else:
logger.info(f"Resume: loading stats from state (media/screenshots from DB)")
return True
except Exception as e:
logger.warning(f"Could not load state: {e}")
return False
def _clear_state(self, target_url: str, preserve_keys: List[str] = None):
"""Rotate state file for fresh start.
Args:
preserve_keys: Keys to preserve (e.g., ['crawl'] for backfill-only fresh)
"""
try:
state_file = self._get_state_file(target_url)
rotate_state_file(state_file, preserve_keys=preserve_keys)
except Exception:
pass
async def purge_job(self, job_id: int) -> Dict[str, Any]:
"""Completely delete a crawl job and all associated data.
Order of operations (critical for data integrity):
1. Get job info and collect hashes/URIs to delete
2. Delete MediaSource records (breaks references)
3. Delete Page records
4. Delete orphan Media records (no longer referenced)
5. Delete orphan media files from vault
6. Delete screenshot files for deleted pages
7. Delete CrawlJob record
8. Delete state files
Returns:
Dict with deletion stats
"""
stats = {
'deleted': False,
'media_files_deleted': 0,
'screenshots_deleted': 0,
'pages_deleted': 0,
'media_sources_deleted': 0,
'state_files_deleted': 0,
}
# Step 1-6: Database deletion (returns info for file deletion)
db_result = await self.db.delete_crawl_job(job_id, purge_data=True)
if not db_result['deleted']:
logger.warning(f"Job {job_id} not found")
return stats
stats['deleted'] = True
stats['pages_deleted'] = len(db_result['page_uris'])
target_uri = db_result['target_uri']
domain = self._get_domain(target_uri) if target_uri else None
# Step 5: Delete orphan media files from vault
for md5_hash in db_result['orphan_media']:
# Try common extensions
for ext in ['jpg', 'jpeg', 'png', 'gif', 'webp', 'mp4', 'webm']:
file_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext)
if file_path.exists():
try:
file_path.unlink()
stats['media_files_deleted'] += 1
logger.debug(f"Deleted media file: {file_path}")
break
except Exception as e:
logger.warning(f"Failed to delete {file_path}: {e}")
# Step 6: Delete screenshot files for deleted pages
if domain:
linkpeek_dir = Path(self.vault_path) / 'linkpeek_vault' / domain
for page_uri in db_result['page_uris']:
parsed = Uri(page_uri)
url_path = (parsed.path or '').lstrip('/') or 'index'
url_path = url_path.replace('/', '_')
# Try common screenshot patterns
for pattern in [f'{url_path}.png', f'{url_path}.jpg', f'{url_path}_0.jpg']:
ss_path = linkpeek_dir / pattern
if ss_path.exists():
try:
ss_path.unlink()
stats['screenshots_deleted'] += 1
logger.debug(f"Deleted screenshot: {ss_path}")
except Exception as e:
logger.warning(f"Failed to delete {ss_path}: {e}")
# Step 7: Already done in db.delete_crawl_job
# Step 8: Delete state files
if target_uri:
# Main state file
state_file = self._get_state_file(target_uri)
if state_file.exists():
try:
state_file.unlink()
stats['state_files_deleted'] += 1
logger.debug(f"Deleted state file: {state_file}")
except Exception as e:
logger.warning(f"Failed to delete state file: {e}")
# State log file
state_log = self._get_state_log(target_uri)
if state_log.path.exists():
try:
state_log.path.unlink()
stats['state_files_deleted'] += 1
logger.debug(f"Deleted state log: {state_log.path}")
except Exception as e:
logger.warning(f"Failed to delete state log: {e}")
# Hydra state file (per-domain, keep it - other jobs may use it)
# Don't delete hydra state as it's domain-wide, not job-specific
logger.info(f"Purged job {job_id}: {stats['media_files_deleted']} media, "
f"{stats['screenshots_deleted']} screenshots, {stats['pages_deleted']} pages")
return stats
def _get_domain(self, url: str) -> str:
"""Extract domain from URL."""
parsed = Uri(url)
return parsed.hostname.lower()
def _track_domain_stat(self, domain: str, stat: str, increment: int = 1):
"""Track per-domain stats for vault commits."""
if domain not in self._domain_stats:
self._domain_stats[domain] = {'pages_changed': 0, 'media_new': 0, 'screenshots_new': 0}
self._domain_stats[domain][stat] += increment
def _extract_text_from_html(self, html: str, base_url: str = None) -> tuple:
"""Extract title, text content, and markdown from HTML.
Args:
html: Raw HTML content
base_url: Base URL for resolving relative links/images
Returns: (title, text_content, markdown)
"""
try:
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
# Extract title
title = ''
title_tag = soup.find('title')
if title_tag:
title = title_tag.get_text(strip=True)
# Convert to markdown using smart converter
markdown = ''
try:
from html2md import html_to_markdown
markdown = html_to_markdown(html, base_url=base_url)[:200000]
except Exception:
# Fallback to html2text
try:
import html2text
h = html2text.HTML2Text()
h.ignore_links = False
h.ignore_images = False
h.body_width = 0
if base_url:
h.baseurl = base_url
markdown = h.handle(html)[:200000]
except ImportError:
pass
# Remove script and style elements for plain text
for tag in soup(['script', 'style', 'nav', 'header', 'footer']):
tag.decompose()
# Get text content
text = soup.get_text(separator=' ', strip=True)
# Clean up whitespace
import re
text = re.sub(r'\s+', ' ', text)
return title, text[:100000], markdown
except Exception:
return '', '', ''
async def _archive_page_to_vault(
self,
uri: str,
html: str,
media_mappings: Dict[str, str] = None,
crawl_job_id: int = None,
):
"""Archive a page to the HTML vault and store for search."""
domain = self._get_domain(uri)
html_vault = self.domain_vaults.get_html_vault(domain)
is_changed, _ = await html_vault.archive_page(uri, html, media_mappings)
self.stats['bytes_downloaded'] += len(html.encode('utf-8'))
if is_changed:
self._track_domain_stat(domain, 'pages_changed')
self.stats['pages_changed'] += 1
# Store page content for full-text search and phantom site recreation (in thread to not block)
title, content, markdown = await asyncio.to_thread(self._extract_text_from_html, html, uri)
parsed = Uri(uri)
path = parsed.path or '/'
await self.db.store_page(
uri=uri,
title=title,
content=content,
path=path,
markdown=markdown,
raw_html=html,
crawl_job_id=crawl_job_id,
)
async def _archive_media_to_vault(
self,
url: str,
md5_hash: str,
ext: str,
page_url: str = '',
):
"""Create symlink in domain media vault pointing to hash vault."""
domain = self._get_domain(url)
# Domain media path: vault/media_vault/{domain}/{url_path}
parsed = Uri(url)
url_path = (parsed.path or '').lstrip('/') or 'index'
if not url_path.endswith(ext):
url_path = f"{url_path}{ext}"
domain_media_dir = Path(self.vault_path) / 'media_vault' / domain
domain_media_path = domain_media_dir / url_path
# Hash vault path: vault/{9-deep-path}/{hash}.{ext}
hash_vault_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext.lstrip('.'))
# Create symlink if not exists
if not domain_media_path.exists():
domain_media_path.parent.mkdir(parents=True, exist_ok=True)
# Calculate relative path from domain media to hash vault
rel_path = os.path.relpath(hash_vault_path, domain_media_path.parent)
try:
domain_media_path.symlink_to(rel_path)
self._track_domain_stat(domain, 'media_new')
self.stats['media_new'] += 1
except FileExistsError:
pass # Already exists
async def _archive_screenshot_to_vault(
self,
url: str,
md5_hash: str,
ext: str = 'jpg',
suffix: str = '',
):
"""Create symlink in domain linkpeek vault pointing to hash vault."""
domain = self._get_domain(url)
# Domain screenshot path: vault/linkpeek_vault/{domain}/{url_path}{suffix}.{ext}
parsed = Uri(url)
url_path = (parsed.path or '').lstrip('/') or 'index'
url_path = url_path.replace('/', '_') + f'{suffix}.{ext}'
domain_ss_dir = Path(self.vault_path) / 'linkpeek_vault' / domain
domain_ss_path = domain_ss_dir / url_path
# Hash vault path: vault/{9-deep-path}/{hash}.{ext}
hash_vault_path = Path(self.vault_path) / hash_to_path(md5_hash, ext=ext)
# Create symlink if not exists
if not domain_ss_path.exists():
domain_ss_path.parent.mkdir(parents=True, exist_ok=True)
rel_path = os.path.relpath(hash_vault_path, domain_ss_path.parent)
try:
domain_ss_path.symlink_to(rel_path)
self._track_domain_stat(domain, 'screenshots_new')
except FileExistsError:
pass
async def _finish_domain_vaults(self, keywords: List[str] = None):
"""Commit changes to all domain vaults that have diffs."""
for domain, stats in self._domain_stats.items():
if stats['pages_changed'] > 0 or stats['media_new'] > 0 or stats['screenshots_new'] > 0:
html_vault = self.domain_vaults.get_html_vault(domain)
media_vault = self.domain_vaults.get_media_vault(domain)
linkpeek_vault = self.domain_vaults.get_linkpeek_vault(domain)
html_commit = await html_vault.finish_crawl(stats)
media_commit = await media_vault.finish_crawl(stats)
linkpeek_commit = await linkpeek_vault.finish_crawl(stats)
if html_commit:
logger.info(f"Committed HTML vault for {domain}: {html_commit[:8]}")
if media_commit:
logger.info(f"Committed media vault for {domain}: {media_commit[:8]}")
if linkpeek_commit:
logger.info(f"Committed linkpeek vault for {domain}: {linkpeek_commit[:8]}")
async def init(self):
"""Initialize database and vault."""
await self.db.init()
await self.vault.init()
async def crawl(
self,
target_uri: str,
keywords: List[str] = None,
mode: CrawlMode = CrawlMode.IMAGES,
depth: int = -1, # -1 = unlimited
max_pages: int = -1, # -1 = unlimited
download_media: bool = True,
job_id: int = None, # Optional existing job ID (skip creation if provided)
quiet: bool = False, # Disable progress bar (for UI-initiated crawls)
hydra: bool = False, # Parse RSS/Atom/Sitemap for fast discovery
) -> Dict[str, Any]:
"""
Crawl a domain for media.
Args:
target_uri: Starting URL
keywords: Keywords to tag media with (e.g., ["rick and morty"])
mode: CrawlMode - what to collect (IMAGES, VIDEOS, MEDIA, ALL)
depth: Crawl depth (-1 = unlimited)
max_pages: Max pages to crawl (-1 = unlimited)
download_media: Whether to download media or just index URLs
job_id: Optional existing job ID (skip creation if provided)
quiet: Disable progress bar (for UI-initiated crawls)
hydra: Parse RSS/Atom/Sitemap feeds for fast URL discovery
Returns:
Crawl statistics
"""
keywords = keywords or []
# Check if target is a VCS repository (git, hg, svn, etc.)
# Skip if NEOPIG_NO_VCS=1
if not NO_VCS:
vcs_type, clone_url = detect_vcs(target_uri)
if vcs_type:
logger.info(f"Detected {vcs_type} repository: {clone_url}")
return await self.clone_and_index(
target_uri=target_uri,
clone_url=clone_url,
vcs_type=vcs_type,
keywords=keywords,
job_id=job_id,
quiet=quiet,
)
# Create crawl job (unless one was provided)
if job_id is None:
job_id = await self.db.create_crawl_job(
target_uri=target_uri,
keywords=keywords,
mode=mode.value,
)
# Always log to file for this job
start_job_logging(job_id)
logger.info(f"Starting neopig crawl job {job_id}")
logger.info(f"Target: {target_uri}")
logger.info(f"Mode: {mode.value}")
logger.info(f"Keywords: {keywords}")
logger.info(f"Depth: {'unlimited' if depth == -1 else depth}")
if hydra:
logger.info(f"Hydra mode: enabled (RSS/Atom/Sitemap discovery)")
# Load saved state if exists (resume support)
self._load_state(target_uri)
# Track timing for stats
start_time = datetime.now(timezone.utc)
self.stats['crawl_started'] = int(start_time.timestamp())
def format_size(b: int) -> str:
if b < 1024:
return f"{b}B"
elif b < 1024 * 1024:
return f"{b/1024:.1f}KB"
elif b < 1024 * 1024 * 1024:
return f"{b/(1024*1024):.1f}MB"
else:
return f"{b/(1024*1024*1024):.1f}GB"
# Track total URIs discovered for progress bar
uris_total = [0] # Use list for mutability in closure
def on_uris_total(total: int):
uris_total[0] = total
self.stats['pages_pending'] = max(0, total - self.stats['pages_crawled'])
if self.pbar and total > 0:
self.pbar.total = total
self.pbar.refresh()
# Create progress bar with dynamic total
initial_pages = len(self.seen_pages)
self.pbar = tqdm(
total=1, # Start with 1, will be updated as links are discovered
initial=initial_pages,
unit="pages",
dynamic_ncols=True,
bar_format='{n_fmt}/{total_fmt} pages [{elapsed}] {postfix}',
mininterval=0.1,
disable=quiet,
)
self.pbar.set_postfix_str(
f"new: {self.stats['media_downloaded']}, "
f"skip: {self.stats['duplicates_skipped']}, "
f"dup: {self.stats['content_exists']}, "
f"found: {self.stats['media_found']}, "
f"ss: {self.stats['screenshots_taken']}, "
f"err: {self.stats['errors']}, "
f"{format_size(self.stats['bytes_stored'])} stored"
)
def update_pbar():
self.pbar.set_postfix_str(
f"new: {self.stats['media_downloaded']}, "
f"skip: {self.stats['duplicates_skipped']}, "
f"dup: {self.stats['content_exists']}, "
f"found: {self.stats['media_found']}, "
f"ss: {self.stats['screenshots_taken']}, "
f"err: {self.stats['errors']}, "
f"{format_size(self.stats['bytes_stored'])} stored"
)
self.pbar.refresh()
# Open append-only state log for fast writes
self._state_log = self._get_state_log(target_uri)
self._state_log.open()
# Background task to push stats to DB every second (for UI visibility)
stats_stop = asyncio.Event()
async def stats_pusher():
while not stats_stop.is_set():
await asyncio.sleep(1)
if not stats_stop.is_set():
try:
await self.db.update_crawl_job_stats(job_id, self.stats)
except Exception:
pass # Ignore DB errors in background task
stats_task = asyncio.create_task(stats_pusher())
# Progress callback
async def on_progress(msg: str):
self.stats['pages_crawled'] += 1
self.pbar.update(1)
update_pbar()
# STREAMING ETL: HTML → Media + Screenshots run concurrently
max_workers = multiprocessing.cpu_count() * 3 if self.fast_mode else multiprocessing.cpu_count()
max_media_workers = max_workers
max_ss_workers = max_workers
media_semaphore = asyncio.Semaphore(max_media_workers)
ss_semaphore = asyncio.Semaphore(max_ss_workers)
# Track active tasks for graceful completion
active_media_tasks = set()
active_ss_tasks = set()
downloaded_count = [0]
async def process_media_item(item):
"""Download a single media item."""
async with media_semaphore:
url = item.get('url')
try:
success, md5_hash = await self._process_media_item(item, job_id, keywords)
if success and md5_hash:
self.seen_media[url] = md5_hash
self._state_log.media(md5_hash, url)
downloaded_count[0] += 1
if downloaded_count[0] % 10 == 0:
update_pbar()
return success
except Exception as e:
logger.warning(f"Media download failed {url}: {e}")
return False
async def process_screenshot(uri, content_length):
"""Take a single screenshot."""
async with ss_semaphore:
try:
await self._capture_page_screenshot(uri, job_id, page_title='', content_length=content_length)
self._state_log.screenshot(uri)
return True
except Exception as e:
logger.warning(f"Screenshot failed {uri}: {e}")
return False
# STREAMING callbacks - immediately spawn work
async def on_media_streaming(item: Dict[str, Any]):
"""Queue media for immediate parallel download."""
url = item.get('url')
if not url:
return
if url in self.seen_media:
md5_hash = self.seen_media[url]
page_uri = item.get('source_page', '')
existing = await self.db.check_media_uri_exists(url, page_uri)
if not existing:
await self.db.add_media_source(
md5_hash=md5_hash, media_uri=url, page_uri=page_uri,
page_title=item.get('page_title', ''),
page_description=item.get('page_description', ''),
page_keywords=item.get('page_keywords', ''),
page_content=item.get('page_content', ''),
alt_text=item.get('alt_text', ''),
link_text=item.get('link_text', ''),
detail_page_uri=item.get('detail_page_url', ''),
detail_title=item.get('detail_title', ''),
detail_content=item.get('detail_content', ''),
searchable_text=item.get('searchable_text', ''),
crawl_job_id=job_id,
)
self.stats['content_exists'] += 1
return
self.stats['media_found'] += 1
if download_media:
# Spawn download task immediately (don't await)
task = asyncio.create_task(process_media_item(item))
active_media_tasks.add(task)
task.add_done_callback(active_media_tasks.discard)
update_pbar()
# Track archive tasks too
active_archive_tasks = set()
async def on_page_streaming(uri: str, html: str):
"""Archive HTML and spawn screenshot immediately - all non-blocking."""
self.seen_pages.add(uri)
self._state_log.page(uri)
# Spawn archive task (don't await - let it run in background)
async def do_archive():
await self._archive_page_to_vault(uri, html, media_mappings=None, crawl_job_id=job_id)
archive_task = asyncio.create_task(do_archive())
active_archive_tasks.add(archive_task)
archive_task.add_done_callback(active_archive_tasks.discard)
if self.screenshot_config.enabled:
# Spawn screenshot task immediately (don't await)
task = asyncio.create_task(process_screenshot(uri, len(html)))
active_ss_tasks.add(task)
task.add_done_callback(active_ss_tasks.discard)
# Hydra: ALWAYS check known feeds (self-healing), discover new if --hydra
# This runs on every crawl to catch new content from persisted feeds
hydra_urls = await self._fetch_hydra_urls(target_uri, discover_new=hydra)
if hydra_urls:
logger.info(f"Hydra: Injecting {len(hydra_urls)} NEW URLs into crawl queue")
logger.info("=== STREAMING ETL: HTML + Media + Screenshots ===")
pages = await self.fetcher.fetch_with_depth(
start_url=target_uri,
depth=depth,
max_pages=max_pages,
query_keywords=keywords,
mode=mode,
media_callback=on_media_streaming,
progress_callback=on_progress,
page_callback=on_page_streaming,
uris_total_callback=on_uris_total,
initial_visited=self.seen_pages if self.seen_pages else None,
hydra_urls=hydra_urls,
)
self.stats['pages_crawled'] = len(pages)
logger.info(f"HTML complete: {len(pages)} pages. Waiting for {len(active_media_tasks)} media + {len(active_ss_tasks)} screenshots...")
# Wait for all spawned tasks to complete
if active_media_tasks:
await asyncio.gather(*active_media_tasks, return_exceptions=True)
if active_ss_tasks:
await asyncio.gather(*active_ss_tasks, return_exceptions=True)
if active_archive_tasks:
await asyncio.gather(*active_archive_tasks, return_exceptions=True)
update_pbar()
logger.info(f"ETL complete: {downloaded_count[0]} media downloaded, {self.stats['screenshots_taken']} screenshots")
# Stop stats pusher
stats_stop.set()
stats_task.cancel()
try:
await stats_task
except asyncio.CancelledError:
pass
# Close state log
self._state_log.stats(self.stats) # Final stats checkpoint
self._state_log.close()
# Close progress bar
self.pbar.close()
# Calculate final stats
elapsed = (datetime.now(timezone.utc) - start_time).total_seconds()
rate = self.stats['media_downloaded'] / elapsed * 60 if elapsed > 0 else 0
# Update job status
await self.db.complete_crawl_job(job_id, self.stats)
# Commit domain vaults if there were changes
await self._finish_domain_vaults(keywords)
logger.info(f"=== CRAWL COMPLETE ({elapsed:.0f}s) ===")
logger.info(f" Pages: {self.stats['pages_crawled']} | "
f"Found: {self.stats['media_found']} | "
f"Downloaded: {self.stats['media_downloaded']} | "
f"Dupes: {self.stats['duplicates_skipped']} | "
f"Screenshots: {self.stats['screenshots_taken']} | "
f"Errors: {self.stats['errors']}")
logger.info(f" Rate: {rate:.1f}/min | Total time: {elapsed:.1f}s")
# Stop logging to file
stop_job_logging(job_id)
return self.stats
async def _process_media_item(
self,
item: Dict[str, Any],
job_id: int,
keywords: List[str]
) -> Tuple[bool, Optional[str]]:
"""Download and store a media item with page context. Returns (success, md5_hash).
Implements the skeleton key approach:
- If detail_page_url is set, resolves canonical image URL
- Collects both embedding_title (page_title) and detail_title
- Prefers canonical URL for download, falls back to original
"""
media_uri = item['url']
page_uri = item.get('source_page', '')
media_type = item.get('media_type', 'unknown')
detail_page_url = item.get('detail_page_url')
# Extract page context for searchability (embedding context)
page_title = item.get('page_title', '')
page_description = item.get('page_description', '')
page_keywords = item.get('page_keywords', '')
page_content = item.get('page_content', '') # Blog post text for full-text search
alt_text = item.get('alt_text', '')
link_text = item.get('link_text', '')
searchable_text = item.get('searchable_text', '') # Combined metadata from accumulator
# Skeleton key: detail context (from detail page if Pinterest-style gallery)
detail_page_uri = ''
detail_title = ''
detail_content = ''
# Score tracking - higher = better quality
original_media_uri = media_uri
is_canonical = False # Track if we resolved to a canonical URL
try:
# Crystal algorithm: resolve canonical image if this looks like a gallery thumbnail
if detail_page_url:
canonical_result = await self.fetcher.resolve_canonical_image(
detail_page_url=detail_page_url,
thumbnail_url=media_uri,
embedding_title=page_title, # Pass listing page title
)
if canonical_result:
detail_page_uri = detail_page_url
detail_title = canonical_result.get('detail_title', '')
detail_content = canonical_result.get('detail_content', '')
# Use canonical URL if found, otherwise keep thumbnail
if canonical_result.get('canonical_url'):
logger.debug(f"Canonical resolution: {media_uri} -> {canonical_result['canonical_url']}")
media_uri = canonical_result['canonical_url']
is_canonical = True # Resolved to full-res
# Enrich metadata from detail page
if not alt_text and canonical_result.get('description'):
alt_text = canonical_result['description']
# Check if this exact media+page combo was already crawled
existing_hash = await self.db.check_media_uri_exists(media_uri, page_uri)
if existing_hash:
self.stats['duplicates_skipped'] += 1
logger.debug(f"Already crawled: {media_uri} from {page_uri}")
return True, existing_hash # Already have it, return existing hash
# Check if content already in vault (same MD5 = same content)
# We still need to add the new page context even if content exists
if media_uri in self.seen_media:
# Already processed this media_uri, just add context
# We need to fetch to get MD5, but we can skip if we track it
pass
# Fetch the media
result = await self.fetcher.fetch_media(media_uri)
if not result:
self.stats['errors'] += 1
return False, None
md5_hash = result['md5_hash']
self.stats['bytes_downloaded'] += result.get('size', 0)
# Check if content already in vault
if await self.vault.exists(md5_hash):
# Content exists, but add this new page context with skeleton key
await self.db.add_media_source(
md5_hash=md5_hash,
media_uri=media_uri,
page_uri=page_uri,
page_title=page_title,
page_description=page_description,
page_keywords=page_keywords,
page_content=page_content,
alt_text=alt_text,
link_text=link_text,
detail_page_uri=detail_page_uri,
detail_title=detail_title,
detail_content=detail_content,
searchable_text=searchable_text,
crawl_job_id=job_id,
)
self.stats['content_exists'] += 1
logger.debug(f"Content exists, added context: {md5_hash} from {page_uri}")
return True, md5_hash
# Store in vault (new content)
ext = self._get_extension(media_uri, result.get('mime_type', ''))
await self.vault.store(md5_hash, result['data'], ext)
self.stats['bytes_stored'] += len(result['data'])
# Emit to live feed immediately after disk save (before DB insert)
emit_live_media({
'md5_hash': md5_hash,
'media_type': media_type,
'mime_type': result.get('mime_type', ''),
'file_size': len(result['data']),
'alt_text': alt_text,
'title': page_title,
'media_uri': media_uri,
'page_uri': page_uri,
})
# Create symlink in domain media vault pointing to hash vault
await self._archive_media_to_vault(media_uri, md5_hash, ext, page_uri)
# Determine score based on resolution
score = SCORE_FULL_RES if is_canonical else SCORE_THUMBNAIL
# Record in database with full context and skeleton key
await self.db.create_media_record(
md5_hash=md5_hash,
media_uri=media_uri,
page_uri=page_uri,
crawl_job_id=job_id,
media_type=media_type,
mime_type=result.get('mime_type', ''),
file_size=result.get('size', 0),
page_title=page_title,
page_description=page_description,
page_keywords=page_keywords,
page_content=page_content,
alt_text=alt_text,
link_text=link_text,
detail_page_uri=detail_page_uri,
detail_title=detail_title,
detail_content=detail_content,
searchable_text=searchable_text,
score=score,
)
self.stats['media_downloaded'] += 1
logger.debug(f"Stored: {md5_hash} ({media_uri})")
return True, md5_hash
except Exception as e:
logger.warning(f"Failed to process {media_uri}: {e}")
self.stats['errors'] += 1
return False, None
async def _capture_page_screenshot(
self,
page_uri: str,
job_id: int,
page_title: str = '',
content_length: int = 0,
):
"""Capture and store a screenshot of a page.
Respects robots.txt crawl-delay by coordinating with the fetcher's
per-domain delay tracking. Screenshots use a headless browser which
makes its own HTTP request, so we must enforce delay before capture.
Args:
page_uri: URL of the page to screenshot
job_id: Crawl job ID for database tracking
page_title: Optional page title for metadata
content_length: Raw HTML content length for dynamic delay calculation
"""
if not self.screenshot_config.enabled:
logger.debug(f"Screenshots disabled, skipping {page_uri}")
return
if page_uri in self.seen_screenshots:
logger.debug(f"Screenshot already exists for {page_uri}")
return
logger.info(f"Taking screenshot: {page_uri}")
try:
# Enforce crawl delay before screenshot (headless browser makes HTTP request)
domain = self._get_domain(page_uri)
await self.fetcher._enforce_crawl_delay(domain)
# Pass content_length for dynamic delay calculation on long pages
result = await self.screenshot.capture(page_uri, content_length=content_length)
if not result:
return
# Normalize to list (oversized images return multiple chunks)
chunks = result if isinstance(result, list) else [result]
total_size = 0
for i, chunk in enumerate(chunks):
md5_hash = chunk['md5_hash']
screenshot_data = chunk['data']
screenshot_size = len(screenshot_data)
screenshot_ext = chunk.get('format', 'png')
screenshot_mime = chunk.get('mime_type', 'image/png')
total_size += screenshot_size
# Store in MD5 vault (for deduplication)
if not await self.vault.exists(md5_hash):
await self.vault.store(md5_hash, screenshot_data, screenshot_ext)
self.stats['bytes_stored'] += screenshot_size
# Create symlink in linkpeek vault (with suffix for chunks)
suffix = f'_{i}' if len(chunks) > 1 else ''
await self._archive_screenshot_to_vault(page_uri, md5_hash, screenshot_ext, suffix=suffix)
# Record in database (all chunks, with chunk index in alt_text)
chunk_label = f" (part {i+1}/{len(chunks)})" if len(chunks) > 1 else ""
await self.db.create_media_record(
md5_hash=md5_hash,
media_uri=f"screenshot:{page_uri}{suffix}",
page_uri=page_uri,
crawl_job_id=job_id,
media_type='screenshot',
mime_type=screenshot_mime,
file_size=screenshot_size,
page_title=page_title,
page_description='',
page_keywords='',
alt_text=f"Screenshot of {page_uri}{chunk_label}",
link_text='',
score=SCORE_SCREENSHOT,
)
# Screenshots are fetched by headless browser (network traffic)
self.stats['bytes_downloaded'] += total_size
self.stats['screenshots_taken'] += 1
self.seen_screenshots.add(page_uri) # Only mark seen after success
logger.debug(f"Screenshot captured: {page_uri} -> {chunks[0]['md5_hash']} ({len(chunks)} chunk(s))")
except Exception as e:
logger.warning(f"Screenshot failed for {page_uri}: {e}")
async def clone_and_index(
self,
target_uri: str,
clone_url: str,
vcs_type: str,
keywords: List[str] = None,
job_id: int = None,
quiet: bool = False,
pull: bool = False,
) -> Dict[str, Any]:
"""
Clone a VCS repository and index its files.
Like Hydra mode for feeds, VCS detection is a "smart source" that
bypasses slow HTTP crawling for high-priority ingestion.
Args:
target_uri: Original URI
clone_url: Clone URL (may differ from target_uri)
vcs_type: 'git', 'hg', 'svn', or 'fossil'
keywords: Keywords to tag files with
job_id: Optional existing job ID
quiet: Disable progress output
pull: Update existing clone instead of fresh clone
Returns:
Indexing statistics
"""
from filevault import content_hash
keywords = keywords or []
repo_vault_base = Path(self.vault.path).parent / 'repo_vault'
# Create job
if job_id is None:
job_id = await self.db.create_crawl_job(
target_uri=target_uri,
keywords=keywords,
mode='code',
)
start_job_logging(job_id)
logger.info(f"Starting VCS clone job {job_id}")
logger.info(f"Repository: {clone_url}")
logger.info(f"VCS type: {vcs_type}")
logger.info(f"Keywords: {keywords}")
start_time = datetime.now(timezone.utc)
stats = {
'vcs_type': vcs_type,
'clone_url': clone_url,
'files_indexed': 0,
'files_skipped': 0,
'bytes_stored': 0,
'errors': 0,
'commit_hash': None,
}
# Determine repo path
repo_path = get_repo_path(clone_url, repo_vault_base)
logger.info(f"Repo path: {repo_path}")
# Clone or pull
if repo_path.exists() and (repo_path / '.git').exists():
logger.info(f"Repository exists, pulling updates...")
success, msg = await pull_repo_async(repo_path)
if not success:
logger.error(f"Pull failed: {msg}")
stats['errors'] += 1
else:
logger.info(f"Pull: {msg}")
else:
logger.info(f"Cloning repository...")
success, msg = await clone_repo_async(clone_url, repo_path, vcs_type, shallow=True)
if not success:
logger.error(f"Clone failed: {msg}")
await self.db.complete_crawl_job(job_id, stats)
return stats
logger.info(f"Clone: {msg}")
# Get commit hash
stats['commit_hash'] = get_commit_hash(repo_path)
logger.info(f"Commit: {stats['commit_hash']}")
# Walk files and index
files = list(walk_files(repo_path, include_binary=True))
logger.info(f"Found {len(files)} files to index")
pbar = tqdm(files, unit="files", disable=quiet)
for file_path in pbar:
try:
content = file_path.read_bytes()
h = content_hash(content)
ext = file_path.suffix or '.txt'
file_size = len(content)
# Store in vault
vault_path = await self.vault.store(h, content, ext)
stats['bytes_stored'] += file_size
# Replace file with symlink
rel_path = os.path.relpath(vault_path, file_path.parent)
file_path.unlink()
file_path.symlink_to(rel_path)
# Get file metadata
repo_rel_path = str(file_path.relative_to(repo_path))
language = get_file_language(file_path)
is_binary = is_binary_file(Path(vault_path))
# Create media record with repo metadata
await self.db.create_media_record(
md5_hash=h,
media_uri=f"{clone_url}/blob/HEAD/{repo_rel_path}",
page_uri=target_uri,
page_title=repo_rel_path,
media_type='code',
mime_type=f"text/{language}" if language and not is_binary else "application/octet-stream",
file_size=file_size,
alt_text=language or '',
keywords=keywords,
crawl_job_id=job_id,
score=SCORE_FULL_RES,
# VCS repo metadata
repo_uri=clone_url,
repo_path=repo_rel_path,
commit_hash=stats['commit_hash'],
vcs_type=vcs_type,
)
stats['files_indexed'] += 1
pbar.set_postfix_str(f"indexed: {stats['files_indexed']}, {stats['bytes_stored'] // 1024}KB")
except Exception as e:
logger.warning(f"Error indexing {file_path}: {e}")
stats['errors'] += 1
stats['files_skipped'] += 1
pbar.close()
# Complete job
duration = (datetime.now(timezone.utc) - start_time).total_seconds()
stats['duration_seconds'] = duration
await self.db.complete_crawl_job(job_id, stats)
logger.info(f"VCS indexing complete: {stats['files_indexed']} files, {stats['bytes_stored'] // 1024}KB stored")
return stats
async def backfill_missing_screenshots(self, domain: str = None, crawl_job_id: int = None, fast_mode: bool = False, quiet: bool = False, create_job: bool = True):
"""Capture screenshots for pages that don't have them yet.
Args:
domain: Domain to backfill (e.g., 'example.com'), or None for all pages
crawl_job_id: Optional crawl job ID for tracking
fast_mode: Skip crawl delay, use max parallel workers
quiet: Disable progress bar
create_job: If True, create a backfill_jobs record (False when called from crawl)
"""
import aiosqlite
from datetime import datetime, timezone
if not self.screenshot_config.enabled:
return
# Get pages without screenshots (optionally filtered by domain)
pages = await self.db.get_pages_without_screenshots(domain)
# Filter out already-seen
pages = [p for p in pages if p not in self.seen_screenshots]
if not pages:
logger.debug(f"No pages need screenshots{f' for {domain}' if domain else ''}")
return
total = len(pages)
logger.info(f"Backfilling {total} missing screenshots{f' for {domain}' if domain else ''}")
# Create backfill job record for progress tracking (skip when part of a crawl)
db_path = self.db.db_path
job_id = None
if create_job:
async with aiosqlite.connect(db_path) as db:
await db.execute("PRAGMA journal_mode=WAL")
# Ensure table exists
await db.execute("""
CREATE TABLE IF NOT EXISTS backfill_jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_type TEXT NOT NULL,
domain_filter TEXT,
status TEXT DEFAULT 'running',
total_records INTEGER DEFAULT 0,
processed_records INTEGER DEFAULT 0,
error_count INTEGER DEFAULT 0,
started_at TEXT NOT NULL,
completed_at TEXT
)
""")
now = datetime.now(timezone.utc).isoformat()
cursor = await db.execute(
"INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, processed_records, error_count, started_at) VALUES (?, ?, 'running', ?, 0, 0, ?)",
('screenshots', domain, total, now)
)
job_id = cursor.lastrowid
await db.commit()
logger.info(f"Created screenshot backfill job {job_id}")
# Concurrent workers: 3x CPU in fast mode, 1x CPU otherwise
import multiprocessing
max_workers = multiprocessing.cpu_count() * 3 if fast_mode else multiprocessing.cpu_count()
semaphore = asyncio.Semaphore(max_workers)
logger.info(f"{'Fast' if fast_mode else 'Normal'} mode: {max_workers} concurrent workers")
pbar = tqdm(total=total, desc="Screenshots", unit="pages", disable=quiet, smoothing=0.1)
captured = 0
failed = 0
async def update_progress(success: bool):
"""Atomically update job progress in DB."""
nonlocal captured, failed
if success:
captured += 1
else:
failed += 1
if job_id:
async with aiosqlite.connect(db_path) as db:
if success:
await db.execute(
"UPDATE backfill_jobs SET processed_records = processed_records + 1 WHERE id = ?",
(job_id,)
)
else:
await db.execute(
"UPDATE backfill_jobs SET error_count = error_count + 1 WHERE id = ?",
(job_id,)
)
await db.commit()
async def capture_one(page_uri: str):
async with semaphore:
try:
# Get content length for dynamic delay
content_length = 0
async with aiosqlite.connect(db_path) as conn:
cursor = await conn.execute(
"SELECT LENGTH(raw_html) FROM pages WHERE uri = ?",
(page_uri,)
)
row = await cursor.fetchone()
if row and row[0]:
content_length = row[0]
await self._capture_page_screenshot(page_uri, crawl_job_id or 0, content_length=content_length)
# Check if screenshot was actually captured (added to seen_screenshots)
success = page_uri in self.seen_screenshots
except Exception as e:
logger.debug(f"Screenshot failed {page_uri}: {e}")
success = False
await update_progress(success)
pbar.update(1)
# Run all captures concurrently (semaphore limits parallelism)
await asyncio.gather(*[capture_one(uri) for uri in pages])
pbar.close()
# Mark job completed (only if tracking)
if job_id:
async with aiosqlite.connect(db_path) as db:
now = datetime.now(timezone.utc).isoformat()
await db.execute(
"UPDATE backfill_jobs SET status = 'completed', completed_at = ? WHERE id = ?",
(now, job_id)
)
await db.commit()
logger.info(f"Screenshot backfill complete: {captured} captured, {failed} failed")
def _get_extension(self, url: str, mime_type: str) -> str:
"""Determine file extension from URL or MIME type."""
# Try from URL path
path = Uri(url).path.lower()
for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.svg', '.bmp',
'.mp4', '.webm', '.mov', '.avi', '.mkv']:
if path.endswith(ext):
return ext.lstrip('.')
# Try from MIME type
mime_map = {
'image/jpeg': 'jpg',
'image/png': 'png',
'image/gif': 'gif',
'image/webp': 'webp',
'image/svg+xml': 'svg',
'video/mp4': 'mp4',
'video/webm': 'webm',
}
for mt, ext in mime_map.items():
if mt in mime_type:
return ext
return 'bin'
def trim_html_wrapper(html: str) -> str:
"""Strip nav, header, footer, sidebar, and logo elements from HTML.
Useful for cleaning up Discourse and similar sites before markdown conversion.
"""
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
# Remove common wrapper elements
selectors_to_remove = [
'nav', 'header', 'footer', 'aside',
'.sidebar', '.nav', '.navigation', '.menu',
'.header', '.footer', '.logo', '.site-logo',
'#header', '#footer', '#nav', '#sidebar',
'.d-header', '.d-footer', # Discourse specific
'.header-wrapper', '.footer-wrapper',
'[role="banner"]', '[role="navigation"]', '[role="contentinfo"]',
]
for selector in selectors_to_remove:
for tag in soup.select(selector):
tag.decompose()
# Remove site logo images (be specific to avoid removing content images)
for img in soup.find_all('img'):
src = img.get('src', '').lower()
alt = img.get('alt', '').lower()
cls = ' '.join(img.get('class', [])).lower()
# Only remove if it's clearly a site logo, not general icons
is_logo = 'logo' in cls or 'brand' in cls or 'site-logo' in src
is_logo = is_logo or (alt and ('logo' in alt or 'brand' in alt))
if is_logo:
img.decompose()
return str(soup)
def extract_meta_from_html(html: str) -> tuple[str, list[str]]:
"""Extract meta description and keywords from HTML.
Returns:
Tuple of (description, keywords_list)
"""
from bs4 import BeautifulSoup
import re
soup = BeautifulSoup(html, 'html.parser')
description = ""
keywords = []
# Extract meta description
meta_desc = soup.find('meta', attrs={'name': re.compile(r'^description$', re.I)})
if meta_desc and meta_desc.get('content'):
description = meta_desc['content'].strip()[:500]
# Extract meta keywords
meta_kw = soup.find('meta', attrs={'name': re.compile(r'^keywords$', re.I)})
if meta_kw and meta_kw.get('content'):
raw_kw = meta_kw['content']
keywords = [k.strip().lower() for k in raw_kw.split(',') if k.strip()]
# Also check og:description as fallback
if not description:
og_desc = soup.find('meta', attrs={'property': 'og:description'})
if og_desc and og_desc.get('content'):
description = og_desc['content'].strip()[:500]
# Extract from article:tag meta tags (common in blogs)
for tag_meta in soup.find_all('meta', attrs={'property': 'article:tag'}):
if tag_meta.get('content'):
keywords.append(tag_meta['content'].strip().lower())
# Dedupe keywords
keywords = list(dict.fromkeys(keywords))[:20]
return description, keywords
def _process_single_page(args: tuple) -> tuple:
"""Process a single page for markdown conversion (runs in thread within process)."""
from html2md import html_to_markdown
import json
page_id, uri, raw_html, trim_wrapper = args
try:
description, keywords = extract_meta_from_html(raw_html)
raw = raw_html
if trim_wrapper:
raw = trim_html_wrapper(raw)
new_markdown = html_to_markdown(raw, base_url=uri)[:200000]
return (page_id, new_markdown, description, json.dumps(keywords) if keywords else None, None)
except Exception as e:
return (page_id, None, None, None, str(e))
def _process_chunk(args: tuple) -> dict:
"""Process a chunk of pages in a subprocess with thread workers.
Each process gets a chunk and spawns threads to burn through it.
Each thread has its own DB connection - WAL mode allows concurrent writes.
Background flusher thread syncs progress every 100ms (workers never block).
"""
import sqlite3
import threading
import time
from concurrent.futures import ThreadPoolExecutor
chunk, db_path, trim_wrapper, threads_per_process, job_id = args
# Thread-local storage for DB connections
thread_local = threading.local()
progress = {'updated': 0, 'errors': 0}
progress_lock = threading.Lock()
stop_flusher = threading.Event()
def get_conn():
"""Get thread-local DB connection for page writes."""
if not hasattr(thread_local, 'conn'):
thread_local.conn = sqlite3.connect(db_path, timeout=60.0)
thread_local.conn.execute("PRAGMA journal_mode=WAL")
thread_local.conn.execute("PRAGMA synchronous=NORMAL")
thread_local.conn.execute("PRAGMA busy_timeout=60000")
return thread_local.conn
def flusher_thread():
"""Background thread that syncs progress to DB every 100ms."""
conn = sqlite3.connect(db_path, timeout=60.0)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=60000")
flushed_updated, flushed_errors = 0, 0
while not stop_flusher.is_set():
time.sleep(0.05)
with progress_lock:
curr_updated, curr_errors = progress['updated'], progress['errors']
delta_updated = curr_updated - flushed_updated
delta_errors = curr_errors - flushed_errors
if delta_updated > 0 or delta_errors > 0:
conn.execute(
"UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
(delta_updated, delta_errors, job_id)
)
conn.commit()
flushed_updated, flushed_errors = curr_updated, curr_errors
# Final flush - get final counts and flush any remaining delta
with progress_lock:
final_updated, final_errors = progress['updated'], progress['errors']
delta_updated = final_updated - flushed_updated
delta_errors = final_errors - flushed_errors
if delta_updated > 0 or delta_errors > 0:
conn.execute(
"UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
(delta_updated, delta_errors, job_id)
)
conn.commit()
# Store final counts for verification
progress['_flushed_updated'] = final_updated
progress['_flushed_errors'] = final_errors
conn.close()
def process_and_write(page_data):
"""Process one page and write immediately with thread-local connection."""
page_id, uri, raw_html = page_data
result = _process_single_page((page_id, uri, raw_html, trim_wrapper))
page_id, markdown, description, keywords_json, error = result
conn = get_conn()
if error:
with progress_lock:
progress['errors'] += 1
else:
conn.execute(
"UPDATE pages SET markdown = ?, description = ?, keywords = ? WHERE id = ?",
(markdown, description, keywords_json, page_id)
)
conn.commit()
with progress_lock:
progress['updated'] += 1
# Start background flusher
flusher = threading.Thread(target=flusher_thread, daemon=True)
flusher.start()
# Fan out to thread workers
with ThreadPoolExecutor(max_workers=threads_per_process) as thread_executor:
list(thread_executor.map(process_and_write, chunk))
# Stop flusher and wait for final flush
stop_flusher.set()
flusher.join(timeout=5.0)
# Return verified counts (what was actually flushed to DB)
return {
'updated': progress.get('_flushed_updated', progress['updated']),
'errors': progress.get('_flushed_errors', progress['errors']),
'total': len(chunk)
}
async def backfill_markdown(db_path: str, domain_filter: str = None, trim_wrapper: bool = False, quiet: bool = False, fresh: bool = False):
"""Re-process stored HTML to regenerate markdown with smart structure detection.
ETL-style parallel processing:
- Divides work into chunks (1 per CPU core)
- Each process spawns 6 threads to burn through its chunk
- Immediate DB writes with WAL mode (thread-safe)
- Progress tracked via backfill_jobs table (atomic increments)
Args:
db_path: Path to SQLite database
domain_filter: Only process pages matching this domain (e.g., 'example.com')
trim_wrapper: Strip nav/header/footer/logo before conversion
quiet: Disable progress bar
fresh: Re-process all pages, even those that already have markdown
"""
import aiosqlite
import sqlite3
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
from datetime import datetime, timezone
import time
num_cpus = multiprocessing.cpu_count()
threads_per_process = 6
logger.info(f"ETL mode: {num_cpus} processes x {threads_per_process} threads = {num_cpus * threads_per_process} workers")
if domain_filter:
pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%"
logger.info(f"Backfilling markdown for domain: {domain_filter}")
else:
pattern = None
logger.info("Backfilling markdown for ALL pages")
if trim_wrapper:
logger.info("Trim wrapper enabled")
# Setup: ensure tables/columns exist, create job record, get pages
async with aiosqlite.connect(db_path) as db:
await db.execute("PRAGMA journal_mode=WAL")
# Ensure backfill_jobs table exists
await db.execute("""
CREATE TABLE IF NOT EXISTS backfill_jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_type TEXT NOT NULL,
domain_filter TEXT,
status TEXT DEFAULT 'running',
total_records INTEGER DEFAULT 0,
processed_records INTEGER DEFAULT 0,
error_count INTEGER DEFAULT 0,
started_at TEXT NOT NULL,
completed_at TEXT
)
""")
# Ensure page columns exist
for col in ['description', 'keywords']:
try:
await db.execute(f"ALTER TABLE pages ADD COLUMN {col} TEXT")
except Exception:
pass
# Get pages to process
# fresh=True: re-process all pages
# fresh=False: only pages without markdown
base_condition = "raw_html IS NOT NULL"
if not fresh:
base_condition += " AND (markdown IS NULL OR markdown = '')"
logger.info("Incremental mode: only processing pages without markdown")
else:
logger.info("Fresh mode: re-processing all pages")
if pattern:
cursor = await db.execute(
f"SELECT id, uri, raw_html FROM pages WHERE {base_condition} AND uri LIKE ?",
(pattern,)
)
else:
cursor = await db.execute(
f"SELECT id, uri, raw_html FROM pages WHERE {base_condition}"
)
rows = await cursor.fetchall()
total = len(rows)
logger.info(f"Found {total} pages to process")
# Create job record (explicitly set processed_records/error_count to 0 for atomic increments)
now = datetime.now(timezone.utc).isoformat()
cursor = await db.execute(
"INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, processed_records, error_count, started_at) VALUES (?, ?, 'running', ?, 0, 0, ?)",
('markdown', domain_filter, total, now)
)
job_id = cursor.lastrowid
await db.commit()
logger.info(f"Created backfill job {job_id}")
# Convert to list of tuples
all_pages = [(row[0], row[1], row[2]) for row in rows if row[2]]
# Divide into exactly num_cpus chunks
chunks = []
chunk_size = len(all_pages) // num_cpus if num_cpus > 0 else len(all_pages)
for i in range(num_cpus):
start = i * chunk_size
end = (i + 1) * chunk_size if i < num_cpus - 1 else len(all_pages)
chunks.append(all_pages[start:end])
logger.info(f"Split into {len(chunks)} chunks of ~{chunk_size} pages each")
# Process chunks in parallel, poll job record for progress
with ProcessPoolExecutor(max_workers=num_cpus) as executor:
chunk_args = [(chunk, db_path, trim_wrapper, threads_per_process, job_id) for chunk in chunks]
futures = [executor.submit(_process_chunk, args) for args in chunk_args]
# Poll backfill_jobs table for progress
# Must reconnect each poll to see WAL commits from other processes
with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet, smoothing=0.1) as pbar:
last_count = 0
while True:
# Fresh connection each poll to see latest WAL commits
conn = sqlite3.connect(db_path, timeout=60.0)
conn.execute("PRAGMA journal_mode=WAL")
cursor = conn.execute(
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
(job_id,)
)
row = cursor.fetchone()
current = row[0] if row and row[0] is not None else 0
conn.close()
if current > last_count:
pbar.update(current - last_count)
last_count = current
if all(f.done() for f in futures):
# Final count with fresh connection
conn = sqlite3.connect(db_path, timeout=60.0)
cursor = conn.execute(
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
(job_id,)
)
row = cursor.fetchone()
current = row[0] if row and row[0] is not None else 0
conn.close()
if current > last_count:
pbar.update(current - last_count)
break
time.sleep(0.1)
# Collect results
total_updated = 0
total_errors = 0
for f in futures:
result = f.result()
total_updated += result['updated']
total_errors += result['errors']
# Mark job completed
async with aiosqlite.connect(db_path) as db:
now = datetime.now(timezone.utc).isoformat()
await db.execute(
"UPDATE backfill_jobs SET status = 'completed', completed_at = ? WHERE id = ?",
(now, job_id)
)
await db.commit()
logger.info(f"Backfill complete: {total_updated} updated, {total_errors} errors")
async def backfill_screenshots(
db_path: str,
vault_path: str = "vault",
domain_filter: str = None,
delete_old: bool = True,
fast_mode: bool = False,
quiet: bool = False,
):
"""Re-capture screenshots as JPEG to replace old PNGs.
Args:
db_path: Path to SQLite database
vault_path: Path to vault directory
domain_filter: Only process pages matching this domain
delete_old: Delete old PNG files after successful JPEG capture
fast_mode: Skip delay between captures (for sites without robots.txt)
quiet: Disable progress bar
"""
from screenshot import ScreenshotCapture, ScreenshotConfig
from filevault import AsyncVault, hash_to_path
from database import Database, Media, MediaSource, Page
from sqlalchemy import select, update
from datetime import datetime, timezone
import time
# Load completed URIs from unified state file
completed_uris = set()
state_path = get_state_file_path(domain_filter) if domain_filter else None
if state_path and state_path.exists():
try:
full_state = json.loads(state_path.read_text())
completed_uris = set(full_state.get('backfill_screenshots', []))
if completed_uris:
logger.info(f"Resuming: {len(completed_uris)} already backfilled")
except Exception as e:
logger.warning(f"Could not load state: {e}")
config = ScreenshotConfig(enabled=True, full_page=True)
capture = ScreenshotCapture(config)
vault = AsyncVault(vault_path)
await vault.init()
db = Database(db_path)
await db.init()
async with db.session() as session:
# Find all screenshot records using ORM, excluding already-completed URIs
if domain_filter:
pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%"
stmt = (
select(MediaSource.page_uri, Media.md5_hash)
.join(Media, MediaSource.md5_hash == Media.md5_hash)
.where(Media.media_type == 'screenshot')
.where(MediaSource.page_uri.like(pattern))
.where(MediaSource.page_uri.notin_(completed_uris) if completed_uris else True)
.distinct()
)
logger.info(f"Backfilling screenshots for domain: {domain_filter}")
else:
stmt = (
select(MediaSource.page_uri, Media.md5_hash)
.join(Media, MediaSource.md5_hash == Media.md5_hash)
.where(Media.media_type == 'screenshot')
.where(MediaSource.page_uri.notin_(completed_uris) if completed_uris else True)
.distinct()
)
logger.info("Backfilling screenshots for ALL pages")
result = await session.execute(stmt)
rows = result.fetchall()
to_process = len(rows)
skipped_count = len(completed_uris)
total = to_process + skipped_count
logger.info(f"Found {to_process} to process ({skipped_count} already completed)")
domain_last_fetched = {} # Track last fetch time per domain
crawl_delay = 2.0 # Default crawl delay in seconds
# Concurrent workers in fast mode (2x CPU count since screenshot is I/O bound)
import multiprocessing
max_workers = multiprocessing.cpu_count() * 2 if fast_mode else 1
semaphore = asyncio.Semaphore(max_workers)
if fast_mode and max_workers > 1:
logger.info(f"Fast mode: {max_workers} concurrent workers")
pbar = tqdm(total=total, initial=skipped_count, desc="Screenshots", unit="pages", disable=quiet, smoothing=0.1)
# Thread-safe counters
stats_lock = asyncio.Lock()
stats = {'captured': 0, 'failed': 0, 'bytes_saved': 0}
async def capture_one(page_uri: str, old_hash: str):
"""Capture screenshot for one page. Returns (success, result_data) or (False, None)."""
async with semaphore:
try:
# Get content length for dynamic delay
async with db.session() as local_session:
from sqlalchemy import func as sqlfunc
result = await local_session.execute(
select(sqlfunc.length(Page.raw_html)).where(Page.uri == page_uri)
)
content_length = result.scalar() or 0
# Capture screenshot
result = await capture.capture(page_uri, content_length=content_length)
if not result:
return False, None
return True, (page_uri, old_hash, result)
except Exception as e:
logger.debug(f"Capture error {page_uri}: {e}")
return False, None
async def process_result(success, data):
"""Process a captured screenshot - store and update DB."""
nonlocal completed_uris
if not success or not data:
async with stats_lock:
stats['failed'] += 1
pbar.update(1)
return
page_uri, old_hash, result = data
domain = Uri(page_uri).hostname.lower()
try:
chunks = result if isinstance(result, list) else [result]
old_path = Path(vault_path) / hash_to_path(old_hash, ext='png')
old_size = old_path.stat().st_size if old_path.exists() else 0
# Store chunks
total_new_size = 0
for i, chunk in enumerate(chunks):
new_hash = chunk['md5_hash']
new_data = chunk['data']
new_ext = chunk.get('format', 'jpg')
total_new_size += len(new_data)
if not await vault.exists(new_hash):
await vault.store(new_hash, new_data, new_ext)
# Symlink
parsed = Uri(page_uri)
url_path = (parsed.path or '').lstrip('/') or 'index'
suffix = f'_{i}' if len(chunks) > 1 else ''
new_symlink = Path(vault_path) / 'linkpeek_vault' / domain / (url_path.replace('/', '_') + f'{suffix}.{new_ext}')
new_symlink.parent.mkdir(parents=True, exist_ok=True)
hash_vault_path = Path(vault_path) / hash_to_path(new_hash, ext=new_ext)
rel_path = os.path.relpath(hash_vault_path, new_symlink.parent)
if not new_symlink.exists():
new_symlink.symlink_to(rel_path)
# Delete old
if delete_old and old_path.exists():
old_path.unlink()
async with stats_lock:
stats['bytes_saved'] += old_size - total_new_size
# Remove old symlink
parsed = Uri(page_uri)
url_path = (parsed.path or '').lstrip('/') or 'index'
old_symlink = Path(vault_path) / 'linkpeek_vault' / domain / (url_path.replace('/', '_') + '.png')
if old_symlink.exists():
old_symlink.unlink()
# Update database
new_hash = chunks[0]['md5_hash']
new_ext = chunks[0].get('format', 'jpg')
new_mime = chunks[0].get('mime_type', 'image/jpeg')
new_size = total_new_size
async with db.session() as local_session:
from sqlalchemy import delete as sqldelete
old_media_result = await local_session.execute(
select(Media).where(Media.md5_hash == old_hash)
)
old_media = old_media_result.scalar_one_or_none()
if old_media and old_hash != new_hash:
# Check if new hash already exists (duplicate screenshot from another page)
existing = await local_session.execute(
select(Media).where(Media.md5_hash == new_hash)
)
if not existing.scalar_one_or_none():
new_media = Media(
md5_hash=new_hash,
media_type='screenshot',
mime_type=new_mime,
file_size=new_size,
keywords=old_media.keywords,
alt_text=old_media.alt_text,
title=old_media.title,
first_seen_at=datetime.now(timezone.utc).isoformat(),
analysis_status=old_media.analysis_status,
analysis_result=old_media.analysis_result,
score=SCORE_SCREENSHOT,
)
local_session.add(new_media)
await local_session.flush()
# Update sources to point to new hash and delete old media
await local_session.execute(
update(MediaSource).where(MediaSource.md5_hash == old_hash).values(md5_hash=new_hash)
)
await local_session.execute(sqldelete(Media).where(Media.md5_hash == old_hash))
elif old_media:
old_media.mime_type = new_mime
old_media.file_size = new_size
old_media.first_seen_at = datetime.now(timezone.utc).isoformat()
await local_session.commit()
async with stats_lock:
stats['captured'] += 1
completed_uris.add(page_uri)
# Save state
if state_path:
async with stats_lock:
try:
full_state = json.loads(state_path.read_text()) if state_path.exists() else {}
except Exception:
full_state = {}
full_state['backfill_screenshots'] = list(completed_uris)
state_path.write_text(json.dumps(full_state, indent=2))
pbar.update(1)
except Exception as e:
err_msg = str(e).split('\n')[0][:60]
logger.warning(f"Failed: {page_uri} - {err_msg}")
async with stats_lock:
stats['failed'] += 1
pbar.update(1)
# Process - each worker flushes immediately when done
if fast_mode and max_workers > 1:
async def capture_and_process(row):
"""Capture and immediately process/flush to DB."""
success, data = await capture_one(row.page_uri, row.md5_hash)
await process_result(success, data)
# Launch all tasks, they'll complete and flush independently
tasks = [asyncio.create_task(capture_and_process(row)) for row in rows]
# Wait for all to complete
await asyncio.gather(*tasks)
else:
# Sequential mode with crawl delay
for row in rows:
if not fast_mode:
domain = Uri(row.page_uri).hostname.lower()
last_fetched = domain_last_fetched.get(domain, 0)
elapsed = time.time() - last_fetched
if elapsed < crawl_delay:
await asyncio.sleep(crawl_delay - elapsed)
domain_last_fetched[domain] = time.time()
success, data = await capture_one(row.page_uri, row.md5_hash)
await process_result(success, data)
pbar.close()
logger.info(f"Backfill complete: {stats['captured']} captured, {skipped_count} skipped, {stats['failed']} failed")
if stats['bytes_saved'] > 0:
logger.info(f"Space saved: {stats['bytes_saved'] / 1024 / 1024:.1f} MB")
async def main():
parser = argparse.ArgumentParser(
description="neopig - Neo Python Image Grabber",
epilog="Full-domain async media crawler with deduplication"
)
parser.add_argument(
"targets",
nargs="*",
help="Target URI(s) to crawl (e.g., https://example.com https://other.com)"
)
parser.add_argument(
"-k", "--keywords",
nargs="*",
default=[],
help="Keywords to tag media with (e.g., -k 'rick' 'morty')"
)
parser.add_argument(
"-m", "--mode",
choices=["text", "images", "videos", "media", "all"],
default="images",
help="Crawl mode (default: images)"
)
parser.add_argument(
"-d", "--depth",
type=int,
default=-1,
help="Crawl depth: 0=single page, 1=page+links, -1=unlimited (default: -1)"
)
parser.add_argument(
"-p", "--max-pages",
type=int,
default=-1,
help="Maximum pages to crawl (-1 = unlimited, default: -1)"
)
parser.add_argument(
"--db",
default="data/neopig.db",
help="Database path (default: data/neopig.db)"
)
parser.add_argument(
"--vault",
default="data/vault",
help="Vault storage path (default: data/vault)"
)
parser.add_argument(
"--no-download",
action="store_true",
help="Don't download media, just index URLs"
)
parser.add_argument(
"-v", "--verbose",
action="store_true",
help="Verbose output"
)
# Screenshot options (on by default)
parser.add_argument(
"--no-screenshot",
action="store_true",
help="Disable page screenshots"
)
parser.add_argument(
"--screenshot-width",
type=int,
default=1280,
help="Screenshot viewport width in pixels (default: 1280)"
)
parser.add_argument(
"--screenshot-height",
type=int,
default=1024,
help="Screenshot viewport height in pixels (default: 1024)"
)
# Hydra mode - feed/sitemap discovery
parser.add_argument(
"--hydra",
action="store_true",
help="Parse RSS/Atom/Sitemap feeds for fast URL discovery"
)
parser.add_argument(
"--screenshot-delay",
type=int,
default=1000,
help="Delay after page load in ms (default: 1000)"
)
parser.add_argument(
"--screenshot-engine",
type=str,
default=None,
help="Screenshot engine: wkhtmltoimage, cutycapt, playwright-webkit, etc. (default: auto-detect lightest)"
)
parser.add_argument(
"--list-engines",
action="store_true",
help="List available screenshot engines and exit"
)
parser.add_argument(
"--fresh",
action="store_true",
help="Start fresh (rotates state file; backfills preserve crawl state)"
)
parser.add_argument(
"--fast",
action="store_true",
help="Fast mode: no crawl delay (use for sites without robots.txt)"
)
parser.add_argument(
"--backfill-markdown",
metavar="DOMAIN",
help="Re-process stored HTML for DOMAIN to regenerate markdown with absolute URLs (e.g., example.com)"
)
parser.add_argument(
"--trim-wrapper",
action="store_true",
help="With --backfill-markdown: strip nav/header/footer/logo before conversion"
)
parser.add_argument(
"-q", "--quiet",
action="store_true",
help="Disable progress bars"
)
parser.add_argument(
"--backfill-screenshots",
metavar="DOMAIN",
help="Re-capture screenshots as JPEG for DOMAIN (e.g., example.com)"
)
parser.add_argument(
"--keep-old-screenshots",
action="store_true",
help="With --backfill-screenshots: keep old PNG files instead of deleting them"
)
parser.add_argument(
"--serve",
action="store_true",
help="Start SERP server to watch crawl live"
)
parser.add_argument(
"--port",
type=int,
default=31337,
help="Port for SERP server (default: 31337)"
)
parser.add_argument(
"--host",
default="0.0.0.0",
help="Host for SERP server (default: 0.0.0.0)"
)
parser.add_argument(
"--archive",
action="store_true",
help="Package crawl results into a distributable tar.gz archive"
)
parser.add_argument(
"--package-only",
action="store_true",
help="With --archive: skip crawling, just package existing data from vault"
)
parser.add_argument(
"-o", "--output",
default=".",
help="Output directory for archive tar.gz (default: current directory)"
)
parser.add_argument(
"--upgrade-neopig",
metavar="TARBALL",
help="Upgrade neopig inside an existing archive"
)
parser.add_argument(
"--purge-job",
type=int,
metavar="JOB_ID",
help="Purge a crawl job and ALL associated data (media, screenshots, pages, state)"
)
args = parser.parse_args()
# Validate depth - max 18 before suggesting unlimited
if args.depth > 18:
print(f"Error: depth {args.depth} is too high. Use --depth -1 for unlimited crawling.")
return
# Create data directory if needed
from pathlib import Path
db_dir = Path(args.db).parent
if db_dir and str(db_dir) != '.':
db_dir.mkdir(parents=True, exist_ok=True)
# Setup logging to work with tqdm progress bars
setup_logging(level=logging.DEBUG if args.verbose else logging.INFO)
# Handle --list-engines
if args.list_engines:
from screenshot import list_available_engines
engines = await list_available_engines()
print("Available screenshot engines:")
for e in engines:
print(f" {e['name']}: {e['description']}")
print("\nPreference order: wkhtmltoimage > cutycapt > playwright-webkit > playwright")
print("Install lightweight: apt install wkhtmltopdf OR apt install cutycapt")
return
# Handle --upgrade-neopig
if args.upgrade_neopig:
from archive import upgrade_neopig_in_archive
archive_path = Path(args.upgrade_neopig)
output_dir = Path(args.output) if args.output != "." else None
logger.info(f"Upgrading neopig in {archive_path}...")
upgrade_neopig_in_archive(archive_path, output_dir=output_dir)
logger.info(f"Done! Archive updated: {archive_path}")
return
# Handle --purge-job
if args.purge_job:
pig = NeoPig(db_path=args.db, vault_path=args.vault)
await pig.init()
job_id = args.purge_job
# Get job info first for confirmation
job = await pig.db.get_crawl_job(job_id)
if not job:
print(f"Error: Job {job_id} not found")
return
print(f"Purging job {job_id}: {job['target_uri']}")
print(f" Status: {job['status']}")
print(f" Started: {job['started_at']}")
result = await pig.purge_job(job_id)
if result['deleted']:
print(f"\nPurged successfully:")
print(f" Media files deleted: {result['media_files_deleted']}")
print(f" Screenshots deleted: {result['screenshots_deleted']}")
print(f" Pages deleted: {result['pages_deleted']}")
print(f" State files deleted: {result['state_files_deleted']}")
else:
print(f"Error: Failed to purge job {job_id}")
return
# Handle --backfill-markdown
if args.backfill_markdown:
serp_process = None
if args.serve:
import subprocess, sys
serp_script = Path(__file__).parent / 'serp.py'
if serp_script.exists():
serp_cmd = [
sys.executable, str(serp_script),
'--host', args.host,
'--port', str(args.port),
'--db', args.db,
'--vault', args.vault,
]
serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
logger.info(f"SERP server started at http://{args.host}:{args.port}")
await backfill_markdown(args.db, domain_filter=args.backfill_markdown, trim_wrapper=args.trim_wrapper, quiet=args.quiet, fresh=args.fresh)
if serp_process:
logger.info(f"Backfill complete. SERP server still running at http://localhost:{args.port}")
logger.info("Press Ctrl+C to stop...")
try:
serp_process.wait()
except KeyboardInterrupt:
serp_process.terminate()
return
# Handle --backfill-screenshots
if args.backfill_screenshots:
# Handle --fresh: rotate state but preserve crawl data
if args.fresh:
state_path = get_state_file_path(args.backfill_screenshots)
rotate_state_file(state_path, preserve_keys=['crawl'])
serp_process = None
if args.serve:
import subprocess, sys
serp_script = Path(__file__).parent / 'serp.py'
if serp_script.exists():
serp_cmd = [
sys.executable, str(serp_script),
'--host', args.host,
'--port', str(args.port),
'--db', args.db,
'--vault', args.vault,
]
serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
logger.info(f"SERP server started at http://{args.host}:{args.port}")
await backfill_screenshots(
args.db,
vault_path=args.vault,
domain_filter=args.backfill_screenshots,
delete_old=not args.keep_old_screenshots,
fast_mode=args.fast,
quiet=args.quiet,
)
if serp_process:
logger.info(f"Backfill complete. SERP server still running at http://localhost:{args.port}")
logger.info("Press Ctrl+C to stop...")
try:
serp_process.wait()
except KeyboardInterrupt:
serp_process.terminate()
return
# Handle --serve alone (just run the server)
if args.serve and not args.targets:
import subprocess
import sys
import signal
serp_script = Path(__file__).parent / 'serp.py'
if serp_script.exists():
serp_cmd = [
sys.executable, str(serp_script),
'--host', args.host,
'--port', str(args.port),
'--db', args.db,
'--vault', args.vault,
]
logger.info(f"Starting SERP server at http://{args.host}:{args.port}")
try:
proc = subprocess.Popen(serp_cmd)
proc.wait() # Block until server exits
except KeyboardInterrupt:
proc.terminate()
proc.wait(timeout=5)
else:
logger.error("serp.py not found")
return
# Handle --archive: use SiteArchiver from archive.py
if args.archive:
if not args.targets:
parser.error("URL required for --archive")
from archive import SiteArchiver
screenshot_config = ScreenshotConfig(
enabled=not args.no_screenshot,
width=args.screenshot_width,
height=args.screenshot_height,
delay=args.screenshot_delay,
engine=args.screenshot_engine,
full_page=True,
)
archiver = SiteArchiver(
output_dir=args.output,
include_screenshots=not args.no_screenshot,
include_markdown=True,
screenshot_config=screenshot_config,
fast_mode=args.fast,
trim_wrapper=args.trim_wrapper,
show_progress=not args.quiet,
fresh_start=args.fresh,
)
# Start SERP server if requested
serp_process = None
if args.serve:
import subprocess
serp_script = Path(__file__).parent / 'serp.py'
if serp_script.exists():
serp_cmd = [
sys.executable, str(serp_script),
'--host', args.host,
'--port', str(args.port),
'--db', args.db,
'--vault', args.vault,
]
serp_process = subprocess.Popen(serp_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
logger.info(f"SERP server started at http://{args.host}:{args.port}")
try:
archive_path = await archiver.archive(
target_url=args.targets[0],
depth=args.depth,
max_pages=args.max_pages,
db_path=args.db,
vault_path=args.vault,
package_only=args.package_only,
)
print(f"\nArchive created: {archive_path}")
print(f"Extract with: tar -xzf {archive_path.name}")
finally:
if serp_process:
serp_process.terminate()
return
# Require targets for crawling
if not args.targets:
parser.error("targets required (use --list-engines to see available screenshot engines)")
# Map mode string to enum
mode_map = {
"text": CrawlMode.TEXT,
"images": CrawlMode.IMAGES,
"videos": CrawlMode.VIDEOS,
"media": CrawlMode.MEDIA,
"all": CrawlMode.ALL,
}
mode = mode_map[args.mode]
# Create screenshot config from args
screenshot_config = ScreenshotConfig(
enabled=not args.no_screenshot,
width=args.screenshot_width,
height=args.screenshot_height,
delay=args.screenshot_delay,
engine=args.screenshot_engine,
full_page=True, # Always capture full page height
)
if not args.no_screenshot:
engine_info = f", engine={args.screenshot_engine}" if args.screenshot_engine else " (auto-detect)"
logger.info(f"Screenshots enabled: {screenshot_config.width}x{screenshot_config.height}, delay={screenshot_config.delay}ms{engine_info}")
# Initialize and run
pig = NeoPig(
db_path=args.db,
vault_path=args.vault,
screenshot_config=screenshot_config,
fast_mode=args.fast,
trim_wrapper=args.trim_wrapper,
)
await pig.init()
if args.fast:
logger.info("Fast mode: no crawl delay (ignoring robots.txt)")
# Handle --fresh: clear state files for all targets
if args.fresh:
for target in args.targets:
pig._clear_state(target)
logger.info("Starting fresh (state files cleared)")
else:
# Load previously crawled media/screenshots from DB (source of truth)
crawled_media = await pig.db.get_crawled_media_uris()
crawled_screenshots = await pig.db.get_crawled_screenshot_uris()
if crawled_media or crawled_screenshots:
logger.info(f"Resuming: {len(crawled_media)} media, {len(crawled_screenshots)} screenshots in database")
pig.seen_media = crawled_media
pig.seen_screenshots = crawled_screenshots
# Start SERP server if requested
serp_process = None
if args.serve:
import subprocess
import sys
serp_script = Path(__file__).parent / 'serp.py'
if serp_script.exists():
serp_cmd = [
sys.executable, str(serp_script),
'--host', args.host,
'--port', str(args.port),
'--db', args.db,
'--vault', args.vault,
]
# preexec_fn sets child to die when parent dies (Linux)
def _set_pdeathsig():
try:
import ctypes
libc = ctypes.CDLL("libc.so.6", use_errno=True)
PR_SET_PDEATHSIG = 1
libc.prctl(PR_SET_PDEATHSIG, signal.SIGTERM)
except:
pass # Not Linux or prctl unavailable
serp_process = subprocess.Popen(
serp_cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
preexec_fn=_set_pdeathsig,
)
_CHILD_PROCESSES.append(serp_process) # Track for cleanup
logger.info(f"SERP server started at http://{args.host}:{args.port}")
logger.info("Watch the crawl live - pages appear as they're indexed!")
else:
logger.warning("serp.py not found - --serve disabled")
try:
# Crawl all targets concurrently
async def crawl_target(target: str):
logger.info(f"=== Starting crawl: {target} ===")
return await pig.crawl(
target_uri=target,
keywords=args.keywords,
mode=mode,
depth=args.depth,
max_pages=args.max_pages,
download_media=not args.no_download,
hydra=args.hydra,
)
await asyncio.gather(*[crawl_target(t) for t in args.targets])
# Backfill missing screenshots for ALL pages in DB (not just crawled domain)
# Part of crawl, so don't create separate job
if pig.screenshot_config.enabled:
await pig.backfill_missing_screenshots(domain=None, fast_mode=args.fast, quiet=args.quiet, create_job=False)
# Keep SERP server running after crawl
if serp_process:
logger.info(f"Crawl complete. SERP server still running at http://localhost:{args.port}")
logger.info("Press Ctrl+C to stop...")
try:
serp_process.wait()
except KeyboardInterrupt:
pass
finally:
# Cleanup SERP server
if serp_process and serp_process.poll() is None:
logger.info("Stopping SERP server...")
serp_process.terminate()
try:
serp_process.wait(timeout=5)
except subprocess.TimeoutExpired:
serp_process.kill()
if __name__ == "__main__":
asyncio.run(main())
Reference in a new issue View git blame Copy permalink