Archive improvements: venv support, package-only mode, resume fixes

- bootstrap.c: Extract requirements.txt, create venv, pip install deps
- archive.py: Add --package-only flag, use neopig vault directly
- archive.py: Resume now loads crawled pages from DB
- database.py: Add close() and get_crawled_page_uris()
- domain_vault.py: Fix None path handling in url_to_filepath
- Makefile: Add test-alpha target for testing archives
This commit is contained in:
Russell Ballestrini 2026-01-01 15:57:23 -05:00
parent e00034c39b
commit 17efcf9a27
5 changed files with 183 additions and 209 deletions

View file

@ -1,4 +1,4 @@
.PHONY: venv install test crawl serp clean vendor-uri2png vendor-install archive bootstrap
.PHONY: venv install test crawl serp clean vendor-uri2png vendor-install archive bootstrap test-alpha
VENV := .venv
PYTHON := $(VENV)/bin/python
@ -65,6 +65,21 @@ vendor-install: vendor-uri2png venv
server: vendor-install
$(PYTHON) serp.py --host 0.0.0.0 --port 8000
# Test tarball extraction and run embedded neopig server
# Usage: make test-alpha TARBALL=example.tar.gz
test-alpha:
ifndef TARBALL
$(error TARBALL not set. Usage: make test-alpha TARBALL=path/to/archive.tar.gz)
endif
rm -rf test-alpha
mkdir -p test-alpha
tar -xzf "$(TARBALL)" -C test-alpha --strip-components=1
@echo "Extracted to test-alpha/"
@echo "Vault structure:"
@find test-alpha/vault -type f 2>/dev/null | head -5
@echo "Starting neopig server..."
cd test-alpha && PYTHONPATH=neopig ../$(PYTHON) neopig/serp.py --db neopig.db --vault vault --host 0.0.0.0 --port 8001
# Examples:
# make install - create venv and install deps
# make test - run functional test

View file

@ -9,10 +9,9 @@ Output structure:
{domain}-{date}/
html/ # Original HTML pages
markdown/ # Converted markdown (optional)
media/ # Images, videos, audio
screenshots/ # Page screenshots (optional)
archive.db # SQLite search database
serve.py # Embedded search server
vault/ # Media vault (9-layer hash paths, neopig-compatible)
neopig.db # Full neopig database
neopig/ # Embedded neopig server
metadata.json # Crawl metadata
Usage:
@ -28,7 +27,6 @@ import os
import re
import shutil
import signal
import sqlite3
import subprocess
import sys
import tarfile
@ -46,6 +44,7 @@ from tqdm import tqdm
from neopig import NeoPig, setup_logging, rotate_state_file, get_state_file_path
from async_web_fetcher import CrawlMode
from screenshot import ScreenshotConfig
from filevault import hash_to_path
# Optional markdown conversion
try:
@ -130,6 +129,7 @@ class SiteArchiver:
max_pages: int = -1,
db_path: str = None,
vault_path: str = None,
package_only: bool = False,
) -> Path:
"""
Archive a site using neopig and package into tar.gz.
@ -166,32 +166,36 @@ class SiteArchiver:
pig._clear_state(target_url)
logger.info("Fresh start: state file rotated")
else:
# Load previously crawled media/screenshots from DB (source of truth for resume)
# Load previously crawled pages/media/screenshots from DB (source of truth for resume)
crawled_pages = await pig.db.get_crawled_page_uris()
crawled_media = await pig.db.get_crawled_media_uris()
crawled_screenshots = await pig.db.get_crawled_screenshot_uris()
logger.info(f"Resume state: {len(crawled_media)} media URIs, {len(crawled_screenshots)} screenshotted pages in DB")
logger.info(f"Resume state: {len(crawled_pages)} pages, {len(crawled_media)} media URIs, {len(crawled_screenshots)} screenshotted pages in DB")
if crawled_pages:
pig.seen_pages = crawled_pages
if crawled_media:
pig.seen_media = crawled_media
if crawled_screenshots:
pig.seen_screenshots = crawled_screenshots
logger.info(f"Screenshots enabled: {pig.screenshot_config.enabled}")
# Run the crawl
stats = await pig.crawl(
target_uri=target_url,
mode=CrawlMode.ALL,
depth=depth,
max_pages=max_pages,
download_media=True,
)
# Run the crawl (unless package_only)
stats = {}
if package_only:
logger.info("Package-only mode: skipping crawl, packaging existing data")
else:
stats = await pig.crawl(
target_uri=target_url,
mode=CrawlMode.ALL,
depth=depth,
max_pages=max_pages,
download_media=True,
)
# Now package the results
logger.info("Packaging archive (streaming mode)...")
tar_path = self.output_dir / f"{archive_name}.tar.gz"
html_vault_path = Path(vault_path) / 'html_vault' / domain
media_vault_path = Path(vault_path) / 'media_vault' / domain
linkpeek_vault_path = Path(vault_path) / 'linkpeek_vault' / domain
# Get URL-to-hash mapping for rewriting external URLs to local copies
logger.info("Loading media URL mappings...")
@ -216,23 +220,37 @@ class SiteArchiver:
replace_url, html_content, flags=re.IGNORECASE)
return html_content
# Step 1: Scan HTML files to build sitemap and search index (no copying)
logger.info("Building search index...")
# Step 1: Load pages from database (has both raw_html and markdown)
logger.info("Loading pages from database...")
sitemap = []
html_contents = {}
markdown_contents = {}
if html_vault_path.exists():
for html_file in html_vault_path.rglob('*.html'):
rel_path = html_file.relative_to(html_vault_path)
try:
content = html_file.read_text(encoding='utf-8', errors='replace')
# Rewrite external URLs to local copies
content = rewrite_urls(content)
title = self._extract_title(content) or str(rel_path)
sitemap.append({'path': f'html/{rel_path}', 'title': title})
html_contents[str(rel_path)] = content
except Exception:
pass
pages = await pig.db.get_pages_by_domain(domain)
for page in pages:
if not page.get('raw_html'):
continue
# Convert URI to path
uri_path = page.get('path') or ''
if not uri_path or uri_path == '/':
rel_path = 'index.html'
elif uri_path.endswith('.html'):
rel_path = uri_path.lstrip('/')
else:
rel_path = uri_path.strip('/') + '/index.html'
try:
content = page['raw_html']
# Rewrite external URLs to local copies
content = rewrite_urls(content)
title = page.get('title') or self._extract_title(content) or rel_path
sitemap.append({'path': f'html/{rel_path}', 'title': title})
html_contents[rel_path] = content
# Store markdown if available
if page.get('markdown'):
markdown_contents[rel_path] = page['markdown']
except Exception:
pass
# Step 2: Create generated files in small temp dir
local_tmpdir = self.output_dir / '.tmp'
@ -241,9 +259,6 @@ class SiteArchiver:
with tempfile.TemporaryDirectory(dir=local_tmpdir) as tmpdir:
tmpdir_path = Path(tmpdir)
# Create search database
await self._create_search_database_streaming(tmpdir_path, sitemap, domain, html_contents)
# Write metadata
metadata = {
'domain': domain,
@ -273,11 +288,6 @@ class SiteArchiver:
trim_wrapper = self.trim_wrapper
include_screenshots = self.include_screenshots
# Pre-count screenshots for progress bar
screenshot_files = []
if include_screenshots and linkpeek_vault_path.exists():
screenshot_files = list(linkpeek_vault_path.rglob('*.png')) + list(linkpeek_vault_path.rglob('*.jpg'))
def stream_to_tar():
with tarfile.open(tar_path, 'w:gz') as tar:
# Add generated files first (from temp)
@ -294,9 +304,11 @@ class SiteArchiver:
info.size = len(html_bytes)
tar.addfile(info, io.BytesIO(html_bytes))
# Generate markdown on the fly
# Use stored markdown (or generate if not available)
if include_markdown:
md_content = html_to_markdown(content, trim_wrapper=trim_wrapper)
md_content = markdown_contents.get(rel_path_str)
if not md_content:
md_content = html_to_markdown(content, trim_wrapper=trim_wrapper)
md_bytes = md_content.encode('utf-8')
md_rel = rel_path_str.replace('.html', '.md')
md_arcname = f"{archive_name}/markdown/{md_rel}"
@ -313,36 +325,25 @@ class SiteArchiver:
for url, md5 in media_iter:
if md5 in added_hashes:
continue
# Find file in hash vault: vault/xx/hash.ext
bucket = md5[:2]
bucket_dir = hash_vault / bucket
if bucket_dir.exists():
for f in bucket_dir.iterdir():
# Skip screenshots if not included
if not include_screenshots and url.startswith('screenshot:'):
continue
# Find file in hash vault using 9-layer deep path: vault/ab/cd/.../hash.ext
hash_path = hash_vault / hash_to_path(md5)
hash_dir = hash_path.parent
if hash_dir.exists():
for f in hash_dir.iterdir():
if f.stem == md5:
try:
arcname = f"{archive_name}/media/{f.name}"
# Preserve vault structure for neopig compatibility
rel_path = f.relative_to(hash_vault)
arcname = f"{archive_name}/vault/{rel_path}"
tar.add(f, arcname=arcname)
added_hashes.add(md5)
except Exception as e:
logger.debug(f"Error adding media {f}: {e}")
break
# Stream screenshots (follow symlinks)
if include_screenshots and screenshot_files:
screen_iter = tqdm(screenshot_files, desc="Screenshots", unit="files", disable=not show_progress)
for screenshot_file in screen_iter:
try:
rel_path = screenshot_file.relative_to(linkpeek_vault_path)
arcname = f"{archive_name}/screenshots/{rel_path}"
if screenshot_file.is_symlink():
target = screenshot_file.resolve()
if target.exists():
tar.add(target, arcname=arcname)
else:
tar.add(screenshot_file, arcname=arcname)
except Exception as e:
logger.debug(f"Error adding screenshot {screenshot_file}: {e}")
# Bundle neopig source files for self-contained crawling
neopig_src = Path(__file__).parent
neopig_files = [
@ -360,72 +361,24 @@ class SiteArchiver:
if req_path.exists():
tar.add(req_path, arcname=f"{archive_name}/requirements.txt")
# Add full neopig database for self-containment
db_file = Path(db_path)
if db_file.exists():
tar.add(db_file, arcname=f"{archive_name}/neopig.db")
await asyncio.to_thread(stream_to_tar)
# Clear html_contents to free memory
html_contents.clear()
# Close database connection
await pig.db.close()
final_size = tar_path.stat().st_size
logger.info(f"Archive complete: {tar_path} ({final_size / 1024 / 1024:.1f} MB)")
return tar_path
async def _create_search_database_streaming(self, tmpdir: Path, sitemap: list, domain: str, html_contents: dict):
"""Create FTS5 search database from collected html contents."""
import sqlite3
db_path = tmpdir / 'archive.db'
def create_db():
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute('''
CREATE TABLE pages (
id INTEGER PRIMARY KEY,
path TEXT UNIQUE,
title TEXT,
content TEXT
)
''')
c.execute('''
CREATE VIRTUAL TABLE pages_fts USING fts5(
title, content, path,
content='pages',
content_rowid='id'
)
''')
for item in sitemap:
html_path = item['path'].replace('html/', '', 1)
content = html_contents.get(html_path, '')
if content:
try:
soup = BeautifulSoup(content, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'header', 'footer']):
tag.decompose()
text = soup.get_text(separator=' ', strip=True)[:50000]
except Exception:
text = ''
else:
text = ''
try:
c.execute('INSERT INTO pages (path, title, content) VALUES (?, ?, ?)',
(item['path'], item['title'], text))
except Exception:
pass
c.execute('''
INSERT INTO pages_fts(rowid, title, content, path)
SELECT id, title, content, path FROM pages
''')
conn.commit()
conn.close()
await asyncio.to_thread(create_db)
def _write_index_html(self, tmpdir: Path, sitemap: list, domain: str):
"""Write index.html with sitemap."""
html = f'''<!DOCTYPE html>
@ -468,85 +421,6 @@ class SiteArchiver:
pass
return None
def _extract_text(self, html: str) -> str:
"""Extract readable text from HTML for search indexing."""
try:
soup = BeautifulSoup(html, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
tag.decompose()
text = soup.get_text(separator=' ', strip=True)
text = re.sub(r'\s+', ' ', text)
return text[:50000]
except Exception:
return ''
async def _create_search_database(
self,
archive_root: Path,
sitemap: List[Dict[str, str]],
domain: str,
html_dir: Path,
):
"""Create SQLite database with searchable page content."""
db_path = archive_root / 'archive.db'
def create_db():
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute('''
CREATE TABLE pages (
id INTEGER PRIMARY KEY,
url TEXT NOT NULL,
path TEXT NOT NULL,
title TEXT,
content TEXT
)
''')
c.execute('''
CREATE VIRTUAL TABLE pages_fts USING fts5(
title, content, url, path,
content='pages',
content_rowid='id'
)
''')
c.execute('''
CREATE TRIGGER pages_ai AFTER INSERT ON pages BEGIN
INSERT INTO pages_fts(rowid, title, content, url, path)
VALUES (new.id, new.title, new.content, new.url, new.path);
END
''')
for item in sitemap:
path = item['path']
title = item['title']
html_path = archive_root / path
if html_path.exists():
html_content = html_path.read_text(encoding='utf-8', errors='replace')
text_content = self._extract_text(html_content)
else:
text_content = ''
c.execute(
'INSERT INTO pages (url, path, title, content) VALUES (?, ?, ?, ?)',
(path, path, title, text_content)
)
conn.commit()
conn.close()
await asyncio.to_thread(create_db)
logger.info(f"Created search database: {db_path}")
# NOTE: Embedded serve.py removed - archives now use serp.py from bundled neopig/
# The bootstrap.c extracts neopig/*.py and runs serp.py with the tarball as argument.
# See: make run TARBALL=archive.tar.gz
def upgrade_neopig_in_archive(archive_path: Path, output_dir: Path = None) -> Path:
"""Replace neopig/ directory in existing archive with current source.
@ -670,13 +544,14 @@ async def main():
parser.add_argument("-o", "--output", default=".", help="Output directory for tar.gz")
parser.add_argument("-d", "--depth", type=int, default=-1, help="Crawl depth: 0=single page, 1=page+links, -1=unlimited")
parser.add_argument("-p", "--max-pages", type=int, default=-1, help="Max pages (-1 = unlimited)")
parser.add_argument("--no-screenshot", action="store_true", help="Disable screenshots")
parser.add_argument("--no-screenshot", "--no-screenshots", action="store_true", help="Disable screenshots")
parser.add_argument("--screenshot-width", type=int, default=1280, help="Screenshot width")
parser.add_argument("--screenshot-height", type=int, default=1024, help="Screenshot height")
parser.add_argument("--screenshot-engine", type=str, default=None, help="Screenshot engine")
parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
parser.add_argument("--fresh", action="store_true", help="Start fresh (rotates old state files instead of resuming)")
parser.add_argument("--fast", action="store_true", help="Fast mode: no crawl delay (for sites without robots.txt)")
parser.add_argument("--package-only", action="store_true", help="Skip crawling, just package existing data from vault")
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)")
parser.add_argument("--backfill-markdown", metavar="DOMAIN", help="Re-process stored HTML for DOMAIN to regenerate markdown with absolute URLs")
@ -761,6 +636,7 @@ async def main():
target_url=args.url,
depth=args.depth,
max_pages=args.max_pages,
package_only=args.package_only,
)
print(f"\nArchive created: {archive_path}")

View file

@ -173,15 +173,23 @@ static char *extract_neopig(const char *self_path, long tar_offset) {
size_str[12] = '\0';
long filesize = strtol(size_str, NULL, 8);
/* Check if this is a neopig/*.py file */
/* Check if this is a neopig/*.py file or requirements.txt */
char *neopig_pos = strstr(filename, "/neopig/");
char *py_ext = strstr(filename, ".py");
char *req_file = strstr(filename, "/requirements.txt");
if (neopig_pos && py_ext && py_ext > neopig_pos) {
int is_neopig_py = (neopig_pos && py_ext && py_ext > neopig_pos);
int is_requirements = (req_file != NULL);
if (is_neopig_py || is_requirements) {
/* Extract this file */
char *basename = neopig_pos + 8; /* Skip "/neopig/" */
char out_path[4096];
snprintf(out_path, sizeof(out_path), "%s/%s", temp_dir, basename);
if (is_neopig_py) {
char *basename = neopig_pos + 8; /* Skip "/neopig/" */
snprintf(out_path, sizeof(out_path), "%s/%s", temp_dir, basename);
} else {
snprintf(out_path, sizeof(out_path), "%s/requirements.txt", temp_dir);
}
int fd = open(out_path, O_WRONLY | O_CREAT | O_TRUNC, 0644);
if (fd >= 0) {
@ -197,7 +205,12 @@ static char *extract_neopig(const char *self_path, long tar_offset) {
}
close(fd);
found_any = 1;
printf(" Extracted: %s\n", basename);
if (is_neopig_py) {
char *basename = neopig_pos + 8;
printf(" Extracted: %s\n", basename);
} else {
printf(" Extracted: requirements.txt\n");
}
/* Skip padding to 512 boundary */
int pad = (512 - (filesize % 512)) % 512;
@ -273,8 +286,66 @@ int main(int argc, char *argv[]) {
/* Find python */
const char *python = find_python();
/* Build command: python3 serp.py <self_path> */
printf("\nStarting server...\n");
/* Create virtualenv and install deps */
char venv_path[4096];
snprintf(venv_path, sizeof(venv_path), "%s/.venv", neopig_dir);
char req_path[4096];
snprintf(req_path, sizeof(req_path), "%s/requirements.txt", neopig_dir);
/* Check if requirements.txt exists */
if (access(req_path, F_OK) == 0) {
printf("\nCreating virtualenv...\n");
char venv_cmd[4096];
snprintf(venv_cmd, sizeof(venv_cmd), "%s -m venv %s", python, venv_path);
if (system(venv_cmd) != 0) {
fprintf(stderr, "Failed to create virtualenv\n");
free(neopig_dir);
return 1;
}
printf("Installing dependencies...\n");
char pip_cmd[8192];
snprintf(pip_cmd, sizeof(pip_cmd),
"%s/.venv/bin/pip install -r %s", neopig_dir, req_path);
if (system(pip_cmd) != 0) {
fprintf(stderr, "Failed to install dependencies\n");
free(neopig_dir);
return 1;
}
/* Use venv python */
char venv_python[4096];
snprintf(venv_python, sizeof(venv_python), "%s/.venv/bin/python", neopig_dir);
printf("\nStarting server...\n");
printf("Command: %s %s %s\n\n", venv_python, serp_path, self_path);
/* Fork and exec with venv python */
pid_t pid = fork();
if (pid < 0) {
perror("fork");
free(neopig_dir);
return 1;
}
if (pid == 0) {
execl(venv_python, "python", serp_path, self_path, NULL);
perror("execl venv python");
_exit(1);
}
int status;
waitpid(pid, &status, 0);
printf("\nCleaning up %s...\n", neopig_dir);
char rm_cmd[4096];
snprintf(rm_cmd, sizeof(rm_cmd), "rm -rf %s", neopig_dir);
system(rm_cmd);
free(neopig_dir);
return WIFEXITED(status) ? WEXITSTATUS(status) : 1;
}
/* No requirements.txt - use system python */
printf("\nStarting server (no venv)...\n");
printf("Command: %s %s %s\n\n", python, serp_path, self_path);
/* Fork and exec */

View file

@ -297,6 +297,11 @@ class Database:
"""Get a new async session."""
return self._session_factory()
async def close(self):
"""Close database connections."""
if self._engine:
await self._engine.dispose()
async def create_crawl_job(
self,
target_uri: str,
@ -1022,6 +1027,13 @@ class Database:
result = await session.execute(stmt)
return {row[0] for row in result.fetchall()}
async def get_crawled_page_uris(self) -> Set[str]:
"""Get all page URIs that have been crawled (for resume support)."""
async with self.session() as session:
stmt = select(Page.uri)
result = await session.execute(stmt)
return {row[0] for row in result.fetchall()}
async def get_pages_without_screenshots(self, domain: str = None) -> List[str]:
"""Get page URIs that don't have screenshots yet.

View file

@ -110,7 +110,7 @@ def url_to_filepath(url: str) -> str:
https://example.com/images/logo.png -> images/logo.png
"""
parsed = Uri(url)
path = parsed.path.strip('/')
path = (parsed.path or '').strip('/')
if not path:
return 'index.html'