Russell Ballestrini
ee9c3cfc74
- backfill_jobs table tracks ETL progress with atomic increments - Background flusher thread syncs progress every 50ms (workers never block) - Thread-local DB connections for page writes (WAL mode) - tqdm smoothing=0.1 for stable rate display (30-40 pages/sec) - 4 processes x 6 threads = 24 workers burning all CPUs - Zero lost counts - final flush verified against return values
1853 lines
70 KiB
Python
1853 lines
70 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
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
|
|
|
|
Based on pig.py by Russell Ballestrini
|
|
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
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone, timezone
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Optional, Set
|
|
|
|
from miniuri import Uri
|
|
|
|
from async_web_fetcher import (
|
|
AsyncWebFetcher,
|
|
CrawlMode,
|
|
MediaItem,
|
|
extract_media_from_html,
|
|
get_media_type_from_extension,
|
|
get_media_type_from_mime,
|
|
)
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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 = ImageVault(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_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
|
|
}
|
|
|
|
# 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 unified state file path for domain."""
|
|
parsed = Uri(target_url)
|
|
domain = parsed.hostname.lower()
|
|
return get_state_file_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 _load_state(self, target_url: str) -> bool:
|
|
"""Load saved crawl state. Returns True if state was loaded.
|
|
|
|
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.
|
|
"""
|
|
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
|
|
|
|
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
|
|
title, content, markdown = self._extract_text_from_html(html, base_url=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.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,
|
|
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.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/{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_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,
|
|
) -> 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
|
|
|
|
Returns:
|
|
Crawl statistics
|
|
"""
|
|
keywords = keywords or []
|
|
|
|
# Create crawl job
|
|
job_id = await self.db.create_crawl_job(
|
|
target_uri=target_uri,
|
|
keywords=keywords,
|
|
mode=mode.value,
|
|
)
|
|
|
|
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}")
|
|
|
|
# Load saved state if exists (resume support)
|
|
self._load_state(target_uri)
|
|
|
|
# Track timing for stats
|
|
start_time = datetime.now(timezone.utc)
|
|
|
|
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
|
|
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,
|
|
)
|
|
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()
|
|
|
|
# Media callback - called for each discovered media item
|
|
async def on_media_discovered(item: Dict[str, Any]):
|
|
url = item.get('url')
|
|
if not url:
|
|
return
|
|
if url in self.seen_media:
|
|
# Debug: log when skipping
|
|
if self.stats['media_found'] < 5: # Only first few
|
|
logger.debug(f"SKIP (in seen_media): {url[:80]}")
|
|
return
|
|
|
|
self.stats['media_found'] += 1
|
|
update_pbar()
|
|
|
|
if download_media:
|
|
success = await self._process_media_item(item, job_id, keywords)
|
|
if success:
|
|
self.seen_media.add(url) # Only mark seen after success
|
|
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
|
|
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(uri: str, html: str):
|
|
# Track this page as crawled for resume support
|
|
self.seen_pages.add(uri)
|
|
|
|
# Archive to vault and store for search (with job_id for tracking)
|
|
await self._archive_page_to_vault(uri, html, media_mappings=None, crawl_job_id=job_id)
|
|
|
|
# Capture screenshot for every page (honors crawl delay as a UNIT with page fetch)
|
|
if self.screenshot_config.enabled:
|
|
await self._capture_page_screenshot(uri, job_id, page_title='', content_length=len(html))
|
|
|
|
# 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,
|
|
depth=depth,
|
|
max_pages=max_pages,
|
|
query_keywords=keywords,
|
|
mode=mode,
|
|
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)
|
|
|
|
# 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")
|
|
|
|
# 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(
|
|
self,
|
|
item: Dict[str, Any],
|
|
job_id: int,
|
|
keywords: List[str]
|
|
) -> bool:
|
|
"""Download and store a media item with page context. Returns True on success.
|
|
|
|
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 = ''
|
|
|
|
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']
|
|
# 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 # Already have it, consider success
|
|
|
|
# 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
|
|
|
|
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
|
|
|
|
# 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'])
|
|
|
|
# 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(
|
|
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,
|
|
)
|
|
|
|
self.stats['media_downloaded'] += 1
|
|
logger.debug(f"Stored: {md5_hash} ({media_uri})")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to process {media_uri}: {e}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
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='',
|
|
)
|
|
|
|
# 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}")
|
|
|
|
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):
|
|
"""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
|
|
"""
|
|
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
|
|
if pattern:
|
|
cursor = await db.execute(
|
|
"SELECT id, uri, raw_html FROM pages WHERE raw_html IS NOT NULL AND uri LIKE ?",
|
|
(pattern,)
|
|
)
|
|
else:
|
|
cursor = await db.execute(
|
|
"SELECT id, uri, raw_html FROM pages WHERE raw_html IS NOT NULL"
|
|
)
|
|
rows = await cursor.fetchall()
|
|
total = len(rows)
|
|
logger.info(f"Found {total} pages to process")
|
|
|
|
# Create job record
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
cursor = await db.execute(
|
|
"INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, started_at) VALUES (?, ?, 'running', ?, ?)",
|
|
('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
|
|
conn = sqlite3.connect(db_path, timeout=60.0)
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
|
|
with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet, smoothing=0.1) as pbar:
|
|
last_count = 0
|
|
while True:
|
|
cursor = conn.execute(
|
|
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
|
|
(job_id,)
|
|
)
|
|
current = cursor.fetchone()[0]
|
|
if current > last_count:
|
|
pbar.update(current - last_count)
|
|
last_count = current
|
|
if all(f.done() for f in futures):
|
|
# Final count
|
|
cursor = conn.execute(
|
|
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
|
|
(job_id,)
|
|
)
|
|
current = cursor.fetchone()[0]
|
|
if current > last_count:
|
|
pbar.update(current - last_count)
|
|
break
|
|
time.sleep(0.05)
|
|
|
|
conn.close()
|
|
|
|
# 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 storage import ImageVault
|
|
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 = ImageVault(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) / old_hash[:2] / f"{old_hash}.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.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) / new_hash[:2] / f"{new_hash}.{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.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:
|
|
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,
|
|
)
|
|
local_session.add(new_media)
|
|
await local_session.flush()
|
|
await local_session.execute(
|
|
update(MediaSource).where(MediaSource.md5_hash == old_hash).values(md5_hash=new_hash)
|
|
)
|
|
await local_session.execute(sqldelete(Media).where(Media.md5_hash == old_hash))
|
|
elif old_media:
|
|
old_media.mime_type = new_mime
|
|
old_media.file_size = new_size
|
|
old_media.first_seen_at = datetime.now(timezone.utc).isoformat()
|
|
await local_session.commit()
|
|
|
|
async with stats_lock:
|
|
stats['captured'] += 1
|
|
completed_uris.add(page_uri)
|
|
|
|
# Save state
|
|
if state_path:
|
|
async with stats_lock:
|
|
try:
|
|
full_state = json.loads(state_path.read_text()) if state_path.exists() else {}
|
|
except Exception:
|
|
full_state = {}
|
|
full_state['backfill_screenshots'] = list(completed_uris)
|
|
state_path.write_text(json.dumps(full_state, indent=2))
|
|
|
|
pbar.update(1)
|
|
|
|
except Exception as e:
|
|
err_msg = str(e).split('\n')[0][:60]
|
|
logger.warning(f"Failed: {page_uri} - {err_msg}")
|
|
async with stats_lock:
|
|
stats['failed'] += 1
|
|
pbar.update(1)
|
|
|
|
# Process - 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="Based on pig.py by Russell Ballestrini"
|
|
)
|
|
|
|
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 (-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 (all off by default)
|
|
parser.add_argument(
|
|
"--screenshot",
|
|
action="store_true",
|
|
help="Enable page screenshots (requires uri2png)"
|
|
)
|
|
|
|
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)"
|
|
)
|
|
|
|
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=8000,
|
|
help="Port for SERP server (default: 8000)"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 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 --backfill-markdown
|
|
if args.backfill_markdown:
|
|
await backfill_markdown(args.db, domain_filter=args.backfill_markdown, trim_wrapper=args.trim_wrapper, quiet=args.quiet)
|
|
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),
|
|
'--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://localhost:{args.port}")
|
|
try:
|
|
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,
|
|
)
|
|
finally:
|
|
if serp_process:
|
|
serp_process.terminate()
|
|
try:
|
|
serp_process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
serp_process.kill()
|
|
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),
|
|
'--port', str(args.port),
|
|
'--db', args.db,
|
|
'--vault', args.vault,
|
|
]
|
|
logger.info(f"Starting SERP server at http://localhost:{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
|
|
|
|
# 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=args.screenshot,
|
|
width=args.screenshot_width,
|
|
height=args.screenshot_height,
|
|
delay=args.screenshot_delay,
|
|
engine=args.screenshot_engine,
|
|
)
|
|
|
|
if args.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),
|
|
'--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://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:
|
|
# 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,
|
|
)
|
|
|
|
await asyncio.gather(*[crawl_target(t) for t in args.targets])
|
|
|
|
# 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())
|