]+href=["\']([^"\']+)["\']', content_html, re.I)
all_urls = list(set(img_urls + link_urls))
if all_urls:
# Keep anchor-only links (#foo) as-is, resolve others
resolved = {}
for u in all_urls:
if u.startswith('#'):
continue # Skip anchor-only links, they stay internal
elif u.startswith(('http://', 'https://', '//')):
resolved[u] = u
else:
resolved[u] = urljoin(uri, u)
# Lookup media (images, etc) - track which came from exact match vs fallback
url_to_hash = await db.lookup_media_by_uris(list(set(resolved.values())))
exact_matches = set(url_to_hash.keys()) # These are reliable
# Fallback for imgur and other CDNs - lookup by filename/ID
# Only apply to URLs that look like media files (have media extension)
from async_web_fetcher import IMAGE_EXTENSIONS, VIDEO_EXTENSIONS, AUDIO_EXTENSIONS
from pathlib import Path as P
media_exts = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
for murl in [u for u in resolved.values() if u not in url_to_hash]:
url_path = murl.split('?')[0]
ext = P(url_path).suffix.lower()
# Only do filename fallback for URLs with media extensions
if ext in media_exts:
fname = P(url_path).stem
if fname and len(fname) >= 5:
result = await db.lookup_media_by_filename(fname)
if result:
url_to_hash[murl] = result[1]
# Lookup pages for internal link rewriting
page_links = [res for res in resolved.values() if source_domain and source_domain in res]
uri_to_page_hash = await db.lookup_pages_by_uris(page_links) if page_links else {}
for orig, res in resolved.items():
if res in url_to_hash:
md5 = url_to_hash[res]
# Rewrite img src to serve media directly
content_html = content_html.replace(f'src="{orig}"', f'src="/media/{md5}"')
content_html = content_html.replace(f"src='{orig}'", f'src="/media/{md5}"')
# Only rewrite hrefs for exact matches or URLs with media extensions
# (avoid rewriting external links that happen to match by filename)
url_path = res.split('?')[0]
ext = P(url_path).suffix.lower()
if res in exact_matches or ext in media_exts:
content_html = content_html.replace(f'href="{orig}"', f'href="/view/{md5}"')
content_html = content_html.replace(f"href='{orig}'", f'href="/view/{md5}"')
elif res in uri_to_page_hash:
# Rewrite internal page links to archived versions
page_hash = uri_to_page_hash[res]
content_html = content_html.replace(f'href="{orig}"', f'href="/page/{page_hash}"')
content_html = content_html.replace(f"href='{orig}'", f'href="/page/{page_hash}"')
except ImportError:
content_html = f"{html_module.escape(page.get('markdown', '')[:50000])}"
elif page.get("content"):
escaped = html_module.escape(page["content"][:50000])
content_html = f"{escaped}"
# Get screenshots and media
screenshot_hashes = await db.get_page_screenshots(uri)
media_items = await db.get_page_media(uri)
# Metadata rows - (key, value) where key is looked up via t.get(key, key)
import hashlib
uri_hash = hashlib.md5(uri.encode()).hexdigest()
neopig_page_uri = f"/page/{uri_hash}"
keywords = json.loads(page.get('keywords') or '[]') if page.get('keywords') else []
keywords_html = ''.join([f'{k}' for k in keywords]) or '-'
meta_rows = [
("source_uri", f'{uri}'),
("neopig_uri", f'{neopig_page_uri}'),
("description_label", page.get('description') or '-'),
("keywords_label", keywords_html),
("media", f"{len(media_items)}"),
]
return templates.TemplateResponse("view.html.j2", {
"request": request,
"t": TRANSLATIONS.get(language, TRANSLATIONS["en"]),
"t_json": json.dumps(TRANSLATIONS.get(language, TRANSLATIONS["en"])),
"title": page_title,
"hero_html": "",
"meta_rows": meta_rows,
"media_items": media_items,
"content_html": content_html,
"screenshot_hashes": screenshot_hashes,
"source_domain": source_domain,
"page_uri": uri,
"download_btn_html": "",
"noai": noai,
"sources_html": "",
"lang": language,
"langs": LANG_NAMES,
"import_mode": IMPORT_MODE,
})
@app.get("/phantom/export")
async def phantom_export(domain: str = Query(None, description="Filter by domain")):
"""
Export a phantom HTML site - original HTML with media URLs rewritten to vault.
Creates a downloadable zip of the phantom site ready for static hosting.
"""
import io
import re
import zipfile
from urllib.parse import urljoin
from fastapi.responses import StreamingResponse
# Get all pages with raw_html
pages = await db.get_pages_by_domain(domain)
if not pages:
raise HTTPException(status_code=404, detail="No pages with raw HTML found")
# Get all media URL to hash mappings
url_to_hash = await db.get_all_media_uri_mappings()
# Create zip in memory
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
pages_written = 0
media_hashes = set()
for page in pages:
uri = page['uri']
raw_html = page.get('raw_html')
if not raw_html:
continue
# Parse URI to get path
parsed = Uri(uri)
site_domain = parsed.hostname
path = parsed.path.strip('/') or 'index'
if not path.endswith('.html') and '.' not in path.split('/')[-1]:
path = f"{path}/index.html" if path else "index.html"
# Rewrite media URLs to local paths
html = raw_html
# Find all src and href attributes pointing to media
patterns = [
(r'src=["\']([^"\']+)["\']', 'src'),
(r'href=["\']([^"\']+\.(jpg|jpeg|png|gif|webp|mp4|webm|svg|ico))["\']', 'href'),
]
for pattern, attr in patterns:
matches = re.findall(pattern, html, re.IGNORECASE)
for match in matches:
url = match[0] if isinstance(match, tuple) else match
# Resolve relative URLs
full_url = urljoin(uri, url)
# Check if we have this media
if full_url in url_to_hash:
md5 = url_to_hash[full_url]
media_hashes.add(md5)
# Replace with local path
ext = Path(url).suffix or '.bin'
local_path = f"media/{md5}{ext}"
html = html.replace(f'{attr}="{url}"', f'{attr}="{local_path}"')
html = html.replace(f"{attr}='{url}'", f'{attr}="{local_path}"')
elif url in url_to_hash:
md5 = url_to_hash[url]
media_hashes.add(md5)
ext = Path(url).suffix or '.bin'
local_path = f"media/{md5}{ext}"
html = html.replace(f'{attr}="{url}"', f'{attr}="{local_path}"')
html = html.replace(f"{attr}='{url}'", f'{attr}="{local_path}"')
# Write HTML file
zf.writestr(f"site/{path}", html.encode('utf-8'))
pages_written += 1
# Copy media files from vault
media_copied = 0
for md5 in media_hashes:
subdir = VAULT_PATH / md5[:2]
if subdir.exists():
for f in subdir.iterdir():
if f.name.startswith(md5):
ext = f.suffix or '.bin'
zf.write(f, f"site/media/{md5}{ext}")
media_copied += 1
break
# Write index
index_html = f"""
Phantom Site - {site_domain}
Phantom Site Archive
Domain: {site_domain}
Pages: {pages_written}
Media: {media_copied}
Pages
"""
for page in pages[:100]:
parsed = Uri(page['uri'])
path = parsed.path.strip('/') or 'index'
if not path.endswith('.html') and '.' not in path.split('/')[-1]:
path = f"{path}/index.html" if path else "index.html"
title = page.get('title') or path
index_html += f' - {title}
\n'
index_html += """
"""
zf.writestr("site/phantom_index.html", index_html.encode('utf-8'))
# Return zip
zip_buffer.seek(0)
return StreamingResponse(
zip_buffer,
media_type="application/zip",
headers={"Content-Disposition": f"attachment; filename=phantom_{site_domain or 'site'}.zip"}
)
@app.get("/phantom", response_class=HTMLResponse)
async def phantom_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
"""Phantom site export UI."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
domains = await db.get_domains_with_pages()
if templates:
return templates.TemplateResponse("phantom.html.j2", {
"request": request,
"t": t,
"lang": language,
"langs": LANG_NAMES,
"import_mode": IMPORT_MODE,
"domains": domains,
})
raise HTTPException(status_code=500, detail="Templates required")
@app.get("/about", response_class=HTMLResponse)
async def about_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
"""About neopig - the story of pig.py's evolution."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates:
return templates.TemplateResponse("about.html.j2", {
"request": request,
"t": t,
"lang": language,
"langs": LANG_NAMES,
"import_mode": IMPORT_MODE,
})
raise HTTPException(status_code=500, detail="Templates required")
@app.get("/health")
async def health():
"""Health check endpoint."""
has_screenshot = False
try:
from uri2png import get_available_engines
has_screenshot = True
except ImportError:
pass
return {
"status": "healthy",
"features": {
"search": True,
"crawl": True,
"screenshot": has_screenshot
}
}
@app.get("/api/stats")
async def get_stats_endpoint():
"""Get database statistics."""
stats = await db.get_stats()
# Add page count (handled separately since table may not exist)
try:
from sqlalchemy import select, func
from database import Page
async with db.session() as session:
result = await session.execute(select(func.count()).select_from(Page))
stats['total_pages'] = result.scalar() or 0
except Exception:
stats['total_pages'] = 0
return stats
@app.get("/random")
async def random_item(type: str = Query(None, description="Type: media or page (random if not specified)")):
"""Redirect to a random media item or page."""
from sqlalchemy import select, func
from database import Media, Page
import random
# If no type specified, randomly pick between media and page
if type is None:
type = random.choice(["media", "page"])
async with db.session() as session:
if type == "page":
# Get a random page
stmt = select(Page.uri_hash).order_by(func.random()).limit(1)
result = await session.execute(stmt)
row = result.fetchone()
if row and row[0]:
return RedirectResponse(url=f"/page/{row[0]}", status_code=302)
else:
# Get a random media item (excluding screenshots)
stmt = (
select(Media.md5_hash)
.where(Media.media_type != 'screenshot')
.order_by(func.random())
.limit(1)
)
result = await session.execute(stmt)
row = result.fetchone()
if row:
return RedirectResponse(url=f"/view/{row[0]}", status_code=302)
return RedirectResponse(url="/", status_code=302)
@app.get("/api/search")
async def search(
q: str = Query("", description="Search query"),
type: Optional[str] = Query(None, description="Filter by media type"),
status: Optional[str] = Query(None, description="Filter by analysis status"),
limit: int = Query(OVER_9000, le=OVER_9000),
offset: int = Query(0)
):
"""
Search media by text query.
Searches across: keywords, alt_text, title, source URLs, analysis results.
"""
results = await db.search_media_advanced(
q=q if q else None,
media_type=type,
limit=limit,
offset=offset
)
return results
@app.get("/api/live/stream")
async def live_stream():
"""
SSE endpoint for live media feed.
Media appears here immediately after being saved to disk,
before DB insert completes. Use EventSource to connect.
"""
async def event_generator():
queue = get_live_queue()
while True:
try:
# Wait for next media item with timeout
media = await asyncio.wait_for(queue.get(), timeout=30)
yield f"data: {json.dumps(media)}\n\n"
except asyncio.TimeoutError:
# Send keepalive
yield ": keepalive\n\n"
except Exception as e:
logger.warning(f"SSE error: {e}")
break
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable nginx buffering
}
)
@app.get("/api/search/pages")
async def search_pages_endpoint(
q: str = Query("", description="Search query"),
limit: int = Query(50, le=500),
):
"""
Search pages by text query using FTS5.
"""
if not q:
return []
return await db.search_pages(q, limit)
@app.get("/api/upgraded")
async def get_upgraded_media(limit: int = Query(50, le=200)):
"""
Get media that was recently upgraded (found higher quality version).
Used by live feed to show when thumbnails get replaced by full-res.
"""
return await db.get_recently_upgraded(limit)
@app.get("/api/media/{md5_hash}")
async def get_media_info(md5_hash: str):
"""Get full media info including all source URLs."""
media = await db.get_media_by_hash(md5_hash)
if not media:
raise HTTPException(status_code=404, detail="Media not found")
media['sources'] = await db.get_media_sources(md5_hash)
return media
def slugify(text: str, max_len: int = 60) -> str:
"""Convert text to a safe filename slug."""
import re
import unicodedata
# Normalize unicode
text = unicodedata.normalize('NFKD', text).encode('ascii', 'ignore').decode('ascii')
# Lowercase and replace spaces/special chars with hyphens
text = re.sub(r'[^\w\s-]', '', text.lower())
text = re.sub(r'[-\s]+', '-', text).strip('-')
return text[:max_len] if text else ""
@app.get("/media/{md5_hash}")
async def serve_media(md5_hash: str, download: bool = False):
"""
Serve media file from vault or tarball.
Use ?download=1 for attachment mode with smart filename.
Caddy should be configured to cache these responses.
"""
# Tarball mode: serve from tar.gz (run in thread to avoid blocking)
if TAR_PATH:
data, ext = await asyncio.to_thread(find_media_in_tarball, md5_hash)
if data:
mime_type, _ = mimetypes.guess_type(f"file{ext}")
if not mime_type:
mime_type = "application/octet-stream"
filename = f"{md5_hash[:12]}{ext}"
headers = {
"Cache-Control": "public, max-age=31536000, immutable",
"X-Content-Hash": md5_hash,
}
if download:
headers["Content-Disposition"] = f'attachment; filename="{filename}"'
return Response(content=data, media_type=mime_type, headers=headers)
raise HTTPException(status_code=404, detail="Media not found in archive")
# Filesystem mode: find file in vault (9-deep path)
subdir = VAULT_PATH / hash_to_path(md5_hash).parent
if not subdir.exists():
raise HTTPException(status_code=404, detail="Media not found")
# Find file with this hash prefix
for f in subdir.iterdir():
if f.name.startswith(md5_hash):
# Guess content type from filename
mime_type, _ = mimetypes.guess_type(f.name)
if not mime_type:
mime_type = "application/octet-stream"
# Inline mode: skip DB queries, just serve the file fast
if not download:
return FileResponse(
f,
media_type=mime_type,
content_disposition_type="inline",
headers={
"Cache-Control": "public, max-age=31536000, immutable",
"X-Content-Hash": md5_hash,
}
)
# Download mode: generate smart filename from metadata
ext = f.suffix or ""
filename = None
# Priority 1: Original filename from URL
sources = await db.get_media_sources(md5_hash)
if sources:
row2 = sources[0]
if row2.get("media_uri"):
from urllib.parse import unquote
parsed = Uri(row2["media_uri"])
orig_name = Path(unquote(parsed.path)).name
if orig_name and '.' in orig_name:
filename = orig_name
# Priority 2: Generate from alt_text or title
if not filename:
media_record = await db.get_media_by_hash(md5_hash)
if media_record:
if not mime_type and media_record.get("mime_type"):
mime_type = media_record["mime_type"]
name_source = media_record.get("alt_text") or media_record.get("title")
if name_source:
slug = slugify(name_source)
if slug:
filename = f"{slug}{ext}"
# Priority 3: page_title + hash index
if not filename and sources:
row2 = sources[0]
if row2.get("page_title"):
media_idx = int(md5_hash[:4], 16)
slug = slugify(f"{row2['page_title']}-{media_idx}")
if slug:
filename = f"{slug}{ext}"
# Default filename if nothing else
if not filename:
filename = f"{md5_hash[:12]}{ext}"
return FileResponse(
f,
media_type=mime_type,
filename=filename,
content_disposition_type="attachment",
headers={
"Cache-Control": "public, max-age=31536000, immutable",
"X-Content-Hash": md5_hash,
}
)
raise HTTPException(status_code=404, detail="Media not found")
# ============================================================================
# Import Mode - Upload and serve archives
# ============================================================================
IMPORT_UPLOAD_DIR = Path(tempfile.gettempdir()) / "neopig_import"
# Resumable upload tracking: upload_id -> {filename, total_size, created_at}
PENDING_UPLOADS: Dict[str, dict] = {}
def generate_upload_id() -> str:
"""Generate a unique upload ID."""
import secrets
return f"upload-{secrets.token_hex(8)}"
@app.post("/api/import/upload/init")
async def init_resumable_upload(filename: str = Query(...), size: int = Query(...)):
"""Initialize a resumable upload session.
Returns upload_id that client uses for subsequent chunk uploads.
Client can resume from any disconnect by checking /api/import/upload/{upload_id}/status
"""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled")
if not filename.endswith(('.tar.gz', '.tgz', '.run')):
raise HTTPException(status_code=400, detail="File must be .tar.gz, .tgz, or .run")
upload_id = generate_upload_id()
IMPORT_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
# Create empty file for this upload
upload_path = IMPORT_UPLOAD_DIR / f"{upload_id}.part"
upload_path.touch()
PENDING_UPLOADS[upload_id] = {
"filename": filename,
"total_size": size,
"created_at": asyncio.get_event_loop().time(),
"path": str(upload_path)
}
logger.info(f"Import: initialized resumable upload {upload_id} for {filename} ({size / 1024 / 1024:.1f} MB)")
return {
"upload_id": upload_id,
"filename": filename,
"total_size": size,
"bytes_received": 0
}
@app.get("/api/import/upload/{upload_id}/status")
async def get_upload_status(upload_id: str):
"""Get status of a resumable upload. Use this to resume after disconnect."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled")
if upload_id not in PENDING_UPLOADS:
raise HTTPException(status_code=404, detail="Upload not found or expired")
info = PENDING_UPLOADS[upload_id]
upload_path = Path(info["path"])
if not upload_path.exists():
raise HTTPException(status_code=404, detail="Upload file not found")
bytes_received = upload_path.stat().st_size
return {
"upload_id": upload_id,
"filename": info["filename"],
"total_size": info["total_size"],
"bytes_received": bytes_received,
"complete": bytes_received >= info["total_size"]
}
@app.patch("/api/import/upload/{upload_id}")
async def upload_chunk(
upload_id: str,
request: Request,
content_range: str = Header(None)
):
"""Upload a chunk of data for resumable upload.
Use Content-Range header: bytes START-END/TOTAL
Example: Content-Range: bytes 0-1048575/10485760
Or use X-Upload-Offset header for simpler resumption.
"""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled")
if upload_id not in PENDING_UPLOADS:
raise HTTPException(status_code=404, detail="Upload not found or expired")
info = PENDING_UPLOADS[upload_id]
upload_path = Path(info["path"])
# Parse offset from Content-Range or X-Upload-Offset
offset = 0
if content_range:
# Parse "bytes START-END/TOTAL"
try:
range_spec = content_range.replace("bytes ", "")
range_part = range_spec.split("/")[0]
offset = int(range_part.split("-")[0])
except Exception:
raise HTTPException(status_code=400, detail="Invalid Content-Range header")
else:
# Use X-Upload-Offset if no Content-Range
offset_header = request.headers.get("x-upload-offset")
if offset_header:
offset = int(offset_header)
# Verify offset matches current file size (no gaps)
current_size = upload_path.stat().st_size if upload_path.exists() else 0
if offset != current_size:
raise HTTPException(
status_code=409,
detail=f"Offset mismatch: expected {current_size}, got {offset}. Resume from byte {current_size}."
)
# Read and append chunk
chunk_data = await request.body()
chunk_size = len(chunk_data)
with open(upload_path, 'ab') as f:
f.write(chunk_data)
new_size = upload_path.stat().st_size
logger.info(f"Import: {upload_id} received chunk {chunk_size} bytes, total {new_size}/{info['total_size']}")
return {
"upload_id": upload_id,
"bytes_received": new_size,
"total_size": info["total_size"],
"complete": new_size >= info["total_size"]
}
@app.post("/api/import/upload/{upload_id}/complete")
async def complete_resumable_upload(upload_id: str):
"""Finalize upload and start import job."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled")
if upload_id not in PENDING_UPLOADS:
raise HTTPException(status_code=404, detail="Upload not found or expired")
info = PENDING_UPLOADS[upload_id]
upload_path = Path(info["path"])
if not upload_path.exists():
raise HTTPException(status_code=404, detail="Upload file not found")
bytes_received = upload_path.stat().st_size
if bytes_received < info["total_size"]:
raise HTTPException(
status_code=400,
detail=f"Upload incomplete: {bytes_received}/{info['total_size']} bytes"
)
# Rename to final filename
final_path = IMPORT_UPLOAD_DIR / info["filename"]
upload_path.rename(final_path)
# Remove from pending
del PENDING_UPLOADS[upload_id]
logger.info(f"Import: {upload_id} completed, saved as {info['filename']}")
# Create import job
job_id = await db.create_crawl_job(
target_uri=f"import://{info['filename']}",
keywords=["import"],
mode="import"
)
# Run import in background
async def run_import():
try:
await db.set_crawl_job_status(job_id, "running")
stats = await import_archive_to_db(final_path, job_id)
await db.complete_crawl_job(job_id, stats)
final_path.unlink(missing_ok=True)
logger.info(f"Import complete: {stats}")
except Exception as e:
logger.error(f"Import failed: {e}")
await db.fail_crawl_job(job_id, str(e))
task = asyncio.create_task(run_import())
ACTIVE_CRAWL_TASKS[job_id] = task
task.add_done_callback(lambda t, jid=job_id: ACTIVE_CRAWL_TASKS.pop(jid, None))
return {
"status": "importing",
"job_id": job_id,
"filename": info["filename"],
"size": bytes_received
}
@app.post("/api/import/upload")
async def upload_archive(file: UploadFile = File(...)):
"""Upload a tar.gz archive and import into local database as a job."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled (NEOPIG_IMPORT=1)")
if not file.filename.endswith(('.tar.gz', '.tgz', '.run')):
raise HTTPException(status_code=400, detail="File must be .tar.gz, .tgz, or .run")
# Save uploaded file
IMPORT_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
upload_path = IMPORT_UPLOAD_DIR / file.filename
logger.info(f"Import: receiving upload {file.filename}")
try:
with open(upload_path, 'wb') as f:
while chunk := await file.read(1024 * 1024):
f.write(chunk)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Upload failed: {e}")
file_size = upload_path.stat().st_size
logger.info(f"Import: saved {file.filename} ({file_size / 1024 / 1024:.1f} MB)")
# Create import job
job_id = await db.create_crawl_job(
target_uri=f"import://{file.filename}",
keywords=["import"],
mode="import"
)
# Run import in background
async def run_import():
try:
await db.set_crawl_job_status(job_id, "running")
stats = await import_archive_to_db(upload_path, job_id)
await db.complete_crawl_job(job_id, stats)
upload_path.unlink(missing_ok=True)
logger.info(f"Import complete: {stats}")
except Exception as e:
logger.error(f"Import failed: {e}")
await db.fail_crawl_job(job_id, str(e))
task = asyncio.create_task(run_import())
ACTIVE_CRAWL_TASKS[job_id] = task
task.add_done_callback(lambda t, jid=job_id: ACTIVE_CRAWL_TASKS.pop(jid, None))
return {"status": "importing", "job_id": job_id, "filename": file.filename, "size": file_size}
async def import_archive_to_db(archive_path: Path, job_id: int) -> dict:
"""Import archive's database and vault into local database."""
import sqlite3
stats = {"media_imported": 0, "pages_imported": 0, "sources_imported": 0}
# Detect .run offset
offset = 0
try:
with open(archive_path, 'rb') as f:
f.seek(-22, 2)
trailer = f.read(22)
if trailer[:6] == b'NEOPIG':
offset = int(trailer[6:22].decode(), 16)
except Exception:
pass
# Open tarball
if offset > 0:
f = open(archive_path, 'rb')
f.seek(offset)
tar = tarfile.open(fileobj=f, mode='r:gz')
else:
tar = tarfile.open(archive_path, 'r:gz')
try:
members = tar.getmembers()
if not members:
raise ValueError("Empty archive")
archive_root = members[0].name.split('/')[0]
# Extract neopig.db to temp
db_member = f"{archive_root}/neopig.db"
temp_db = IMPORT_UPLOAD_DIR / f"import_{job_id}.db"
for m in members:
if m.name == db_member:
f_db = tar.extractfile(m)
if f_db:
with open(temp_db, 'wb') as out:
out.write(f_db.read())
break
# Merge database
if not temp_db.exists():
raise ValueError("Malformed archive: missing neopig.db")
src = sqlite3.connect(temp_db)
src.row_factory = sqlite3.Row
try:
# Check what tables exist in source
tables = [r[0] for r in src.execute("SELECT name FROM sqlite_master WHERE type='table'")]
logger.info(f"Import: source database has tables: {tables}")
# Validate archive has required tables
required_tables = {'pages', 'media', 'media_sources'}
if not required_tables.intersection(tables):
raise ValueError(f"Malformed archive: neopig.db has no data tables (found: {tables})")
# Import media (if table exists)
if 'media' in tables:
for row in src.execute("SELECT * FROM media"):
try:
async with db.session() as session:
await session.execute(text(
"""INSERT OR IGNORE INTO media
(md5_hash, media_type, mime_type, file_size, keywords, alt_text, title,
first_seen_at, last_seen_at, score)
VALUES (:md5_hash, :media_type, :mime_type, :file_size, :keywords, :alt_text, :title,
:first_seen_at, :last_seen_at, :score)"""), dict(row))
await session.commit()
stats["media_imported"] += 1
except Exception as e:
logger.debug(f"Skip media row: {e}")
else:
logger.warning("Import: source has no 'media' table")
# Import media_sources (if table exists)
if 'media_sources' in tables:
for row in src.execute("SELECT * FROM media_sources"):
try:
row_dict = dict(row)
row_dict['crawl_job_id'] = job_id # Link to import job for deletion
async with db.session() as session:
await session.execute(text(
"""INSERT OR IGNORE INTO media_sources
(md5_hash, media_uri, page_uri, page_title, alt_text, searchable_text, discovered_at, crawl_job_id)
VALUES (:md5_hash, :media_uri, :page_uri, :page_title, :alt_text, :searchable_text, :discovered_at, :crawl_job_id)"""), row_dict)
await session.commit()
stats["sources_imported"] += 1
except Exception as e:
logger.debug(f"Skip media_source row: {e}")
else:
logger.warning("Import: source has no 'media_sources' table")
# Import pages (if table exists)
if 'pages' in tables:
for row in src.execute("SELECT * FROM pages"):
try:
row_dict = dict(row)
row_dict['crawl_job_id'] = job_id # Link to import job for deletion
async with db.session() as session:
await session.execute(text(
"""INSERT OR REPLACE INTO pages
(uri, uri_hash, path, title, description, keywords, content, markdown, raw_html, crawled_at, crawl_job_id)
VALUES (:uri, :uri_hash, :path, :title, :description, :keywords, :content, :markdown, :raw_html, :crawled_at, :crawl_job_id)"""), row_dict)
await session.commit()
stats["pages_imported"] += 1
except Exception as e:
logger.debug(f"Skip page row: {e}")
else:
logger.warning("Import: source has no 'pages' table")
finally:
src.close()
temp_db.unlink(missing_ok=True)
# Extract vault files
vault_prefix = f"{archive_root}/vault/"
for m in members:
if m.name.startswith(vault_prefix) and m.isfile():
rel_path = m.name[len(vault_prefix):]
dest_path = VAULT_PATH / rel_path
if not dest_path.exists():
dest_path.parent.mkdir(parents=True, exist_ok=True)
f_media = tar.extractfile(m)
if f_media:
with open(dest_path, 'wb') as out:
out.write(f_media.read())
stats["media_imported"] += 1
finally:
tar.close()
return stats
@app.get("/api/import/status")
async def import_status():
"""Get current import status."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled")
# Get import job stats
stats = await db.get_stats()
import_jobs = await db.get_crawl_jobs(limit=10)
imports = [j for j in import_jobs if j.get("mode") == "import"]
return {
"import_mode": True,
"media_count": stats.get("media", 0),
"pages_count": stats.get("pages", 0),
"recent_imports": len(imports)
}
@app.get("/import", response_class=HTMLResponse)
async def import_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
"""Import page for uploading archives."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled (NEOPIG_IMPORT=1)")
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates:
return templates.TemplateResponse("import.html.j2", {
"request": request,
"t": t,
"t_json": json.dumps(t),
"lang": language,
"langs": LANG_NAMES,
"import_mode": IMPORT_MODE,
})
raise HTTPException(status_code=500, detail="Templates required")
# ============================================================================
# Crawler API
# ============================================================================
@app.get("/api/crawl/jobs")
async def get_crawl_jobs_endpoint(limit: int = Query(50, le=200)):
"""Get recent jobs (crawl + backfill)."""
try:
# Get both crawl jobs and backfill jobs
crawl_jobs = await db.get_crawl_jobs(limit)
backfill_jobs = await db.get_backfill_jobs(limit)
# Add job_kind to distinguish them
for job in crawl_jobs:
job['job_kind'] = 'crawl'
for job in backfill_jobs:
job['job_kind'] = 'backfill'
# Merge and sort by started_at descending
all_jobs = crawl_jobs + backfill_jobs
all_jobs.sort(key=lambda j: j.get('started_at', ''), reverse=True)
return all_jobs[:limit]
except Exception as e:
logger.error(f"Failed to get jobs: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/crawl/jobs/{job_id}/pause")
async def pause_crawl_job(job_id: int):
"""Pause a running crawl job."""
task = ACTIVE_CRAWL_TASKS.get(job_id)
if not task:
raise HTTPException(status_code=404, detail="Job not running")
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
await db.pause_crawl_job(job_id)
ACTIVE_CRAWL_TASKS.pop(job_id, None)
return {"status": "paused", "job_id": job_id}
@app.post("/api/crawl/jobs/{job_id}/resume")
async def resume_crawl_job(job_id: int):
"""Resume a paused crawl job."""
job = await db.get_crawl_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="Job not found")
if job['status'] not in ('paused', 'cancelled'):
raise HTTPException(status_code=400, detail=f"Job is {job['status']}, cannot resume")
if job_id in ACTIVE_CRAWL_TASKS:
raise HTTPException(status_code=400, detail="Job is already running")
# Mark as running
async with db.session() as session:
await session.execute(
text("UPDATE crawl_jobs SET status = 'running' WHERE id = :id"),
{"id": job_id}
)
await session.commit()
# Start crawl task (resume from state files with original settings)
target_uri = job['target_uri']
keywords = json.loads(job.get('keywords') or '[]')
mode_str = job.get('mode', 'all')
# Use stored settings (with fallbacks for old jobs)
job_depth = job.get('depth', 15) or 15
job_max_pages = job.get('max_pages', -1) if job.get('max_pages') is not None else -1
job_fast = bool(job.get('fast', 0))
job_screenshots = bool(job.get('screenshots', 1))
async def run_crawl(jid=job_id, uri=target_uri):
try:
from neopig import NeoPig
from async_web_fetcher import CrawlMode
from screenshot import ScreenshotConfig
screenshot_config = ScreenshotConfig(enabled=job_screenshots)
pig = NeoPig(db_path=DB_PATH, vault_path=str(VAULT_PATH), fast_mode=job_fast, screenshot_config=screenshot_config)
await pig.init()
# Resume from state (don't clear state files)
crawled_media = await pig.db.get_crawled_media_uris()
crawled_screenshots = await pig.db.get_crawled_screenshot_uris()
if crawled_media:
pig.seen_media = crawled_media
if crawled_screenshots:
pig.seen_screenshots = crawled_screenshots
mode = CrawlMode(mode_str) if mode_str else CrawlMode.ALL
# Progress updater
stop_progress = asyncio.Event()
async def update_progress():
while not stop_progress.is_set():
await asyncio.sleep(2)
if not stop_progress.is_set():
await db.update_crawl_job_stats(jid, pig.stats)
progress_task = asyncio.create_task(update_progress())
try:
stats = await pig.crawl(
target_uri=uri,
keywords=keywords,
mode=mode,
depth=job_depth,
max_pages=job_max_pages,
job_id=jid,
quiet=True,
)
finally:
stop_progress.set()
progress_task.cancel()
try:
await progress_task
except asyncio.CancelledError:
pass
await db.complete_crawl_job(jid, stats)
except Exception as e:
logger.error(f"Resumed crawl job {jid} failed: {e}")
await db.complete_crawl_job(jid, {"error": str(e), "status": "failed"})
task = asyncio.create_task(run_crawl())
ACTIVE_CRAWL_TASKS[job_id] = task
task.add_done_callback(lambda t, jid=job_id: ACTIVE_CRAWL_TASKS.pop(jid, None))
return {"status": "running", "job_id": job_id}
@app.delete("/api/crawl/jobs/{job_id}")
async def delete_crawl_job(job_id: int, purge: bool = Query(True, description="Purge all data (media, screenshots, pages)")):
"""Delete a crawl job and optionally purge all associated data.
With purge=True (default):
- Deletes all MediaSource records for this job
- Deletes all Page records for this job
- Deletes orphan Media records (not referenced by other jobs)
- Deletes orphan media files from vault
- Deletes screenshot files for deleted pages
- Deletes state files
"""
# Can't delete running jobs
if job_id in ACTIVE_CRAWL_TASKS:
raise HTTPException(status_code=400, detail="Cannot delete running job. Pause it first.")
if purge:
# Full purge using NeoPig
from neopig import NeoPig
pig = NeoPig(db_path=DB_PATH, vault_path=str(VAULT_PATH))
await pig.init()
result = await pig.purge_job(job_id)
if not result['deleted']:
raise HTTPException(status_code=404, detail="Job not found")
return {
"status": "purged",
"job_id": job_id,
"media_files_deleted": result['media_files_deleted'],
"screenshots_deleted": result['screenshots_deleted'],
"pages_deleted": result['pages_deleted'],
"state_files_deleted": result['state_files_deleted'],
}
else:
# Just delete job record
result = await db.delete_crawl_job(job_id, purge_data=False)
if not result['deleted']:
raise HTTPException(status_code=404, detail="Job not found")
return {"status": "deleted", "job_id": job_id}
@app.get("/api/crawl/jobs/{job_id}")
async def get_crawl_job_endpoint(job_id: int):
"""Get a specific crawl job."""
job = await db.get_crawl_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="Job not found")
return job
@app.get("/api/crawl/jobs/{job_id}/logs")
async def get_crawl_job_logs(job_id: int, tail: int = Query(0, description="Return only last N lines")):
"""Get logs for a crawl job."""
logs = get_job_logs(job_id, tail=tail)
return Response(content=logs, media_type="text/plain")
@app.get("/api/crawl/jobs/{job_id}/replay")
async def replay_crawl_job(job_id: int, speed: float = Query(1.0, ge=0.1, le=100.0)):
"""
Replay a completed crawl job as SSE stream.
Streams media items at the original discovery pace (adjusted by speed multiplier).
This is a read-only operation - no network requests or DB modifications.
Connect with EventSource to watch the replay in the /live feed.
Args:
job_id: The crawl job to replay
speed: Playback speed multiplier (1.0 = realtime, 2.0 = 2x faster, 0.5 = half speed)
"""
from datetime import datetime
# Verify job exists
job = await db.get_crawl_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="Job not found")
async def replay_generator():
prev_time = None
total_count = 0
batch_size = 100
offset = 0
while True:
# Fetch next batch
batch = await db.get_job_media_timeline(job_id, limit=batch_size, offset=offset)
if not batch:
break
for item in batch:
total_count += 1
# Parse discovery time
discovered_at = item.get('discovered_at')
if discovered_at:
try:
curr_time = datetime.fromisoformat(discovered_at.replace('Z', '+00:00'))
# Calculate delay from previous item
if prev_time:
delta = (curr_time - prev_time).total_seconds()
# Apply speed multiplier and cap delay at 5 seconds max
delay = min(delta / speed, 5.0)
if delay > 0:
await asyncio.sleep(delay)
prev_time = curr_time
except (ValueError, TypeError):
pass
# Format media item for SSE (same format as live stream)
media_event = {
'md5_hash': item['md5_hash'],
'media_type': item.get('media_type', 'image'),
'mime_type': item.get('mime_type', ''),
'page_uri': item.get('page_uri', ''),
'page_title': item.get('page_title', ''),
'replay': True,
'job_id': job_id,
}
yield f"data: {json.dumps(media_event)}\n\n"
offset += batch_size
# Send end-of-replay marker
yield f"data: {json.dumps({'type': 'replay_end', 'job_id': job_id, 'total': total_count})}\n\n"
return StreamingResponse(
replay_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
)
@app.post("/api/crawl")
async def start_crawl(request: CrawlRequest, background_tasks: BackgroundTasks):
"""
Start new crawl job(s).
Supports multiple targets - creates one job per target.
The crawls run in the background. Poll /api/crawl/jobs/{id} for status.
"""
if CRAWL_DISABLED:
raise HTTPException(status_code=403, detail="Crawling is disabled (NEOPIG_DISABLE_CRAWL=1)")
# Import neopig here to avoid circular imports
from neopig import NeoPig
from async_web_fetcher import CrawlMode
# 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.get(request.mode, CrawlMode.IMAGES)
# Get targets (support both new 'targets' array and old 'target_uri' single value)
targets = request.targets if request.targets else [request.target_uri] if request.target_uri else []
if not targets:
raise HTTPException(status_code=400, detail="No target URIs provided")
job_ids = []
# Create a job for each target (store all settings for resume)
depth = request.depth if 1 <= request.depth <= 15 else 15
for target_uri in targets:
job_id = await db.create_crawl_job(
target_uri, request.keywords, request.mode,
depth=depth, max_pages=request.max_pages,
fast=request.fast, screenshots=request.screenshots
)
job_ids.append(job_id)
# Run crawl in background - capture ALL values to avoid closure issues
is_fresh = request.fresh
is_fast = request.fast
has_screenshots = request.screenshots
is_hydra = request.hydra
crawl_keywords = list(request.keywords) # Copy list
crawl_max_pages = request.max_pages
crawl_depth = request.depth if 1 <= request.depth <= 15 else 15
async def run_crawl(jid=job_id, uri=target_uri, fresh=is_fresh, fast=is_fast, screenshots=has_screenshots, hydra=is_hydra, depth=crawl_depth, keywords=crawl_keywords, max_pages=crawl_max_pages):
try:
from screenshot import ScreenshotConfig
screenshot_config = ScreenshotConfig(enabled=screenshots)
pig = NeoPig(db_path=DB_PATH, vault_path=str(VAULT_PATH), fast_mode=fast, screenshot_config=screenshot_config)
await pig.init()
# Handle fresh vs resume - NEVER fresh on resume
if fresh:
pig._clear_state(uri)
else:
# Load existing seen media for resume capability
crawled_media = await pig.db.get_crawled_media_uris()
crawled_screenshots = await pig.db.get_crawled_screenshot_uris()
if crawled_media:
pig.seen_media = crawled_media
if crawled_screenshots:
pig.seen_screenshots = crawled_screenshots
# Progress updater - runs alongside crawl
stop_progress = asyncio.Event()
async def update_progress():
while not stop_progress.is_set():
await asyncio.sleep(2) # Update every 2 seconds
if not stop_progress.is_set():
await db.update_crawl_job_stats(jid, pig.stats)
progress_task = asyncio.create_task(update_progress())
try:
stats = await pig.crawl(
target_uri=uri,
keywords=keywords,
mode=mode,
depth=depth,
max_pages=max_pages,
job_id=jid, # Use existing job, don't create another
quiet=True, # No progress bar for UI-initiated crawls
hydra=hydra, # Feed/sitemap discovery mode
)
finally:
stop_progress.set()
progress_task.cancel()
try:
await progress_task
except asyncio.CancelledError:
pass
# Update job as completed
await db.complete_crawl_job(jid, stats)
except Exception as e:
logger.error(f"Crawl job {jid} failed: {e}")
await db.complete_crawl_job(jid, {"error": str(e), "status": "failed"})
# Schedule async task on the event loop (not BackgroundTasks which runs in threadpool)
task = asyncio.create_task(run_crawl())
ACTIVE_CRAWL_TASKS[job_id] = task
# Clean up when done
task.add_done_callback(lambda t, jid=job_id: ACTIVE_CRAWL_TASKS.pop(jid, None))
return {"job_ids": job_ids, "status": "running", "count": len(job_ids)}
def init_tarball_mode(tarball_path: str):
"""Initialize serving from a tar.gz archive."""
global TAR_PATH, TAR_OFFSET, TAR_MEMBERS, TAR_MEDIA_INDEX, ARCHIVE_ROOT, DB_PATH, TEMP_DB_PATH
logger.info(f"Opening archive: {tarball_path}")
tarball = Path(tarball_path)
TAR_PATH = str(tarball.resolve())
# Handle .run files with NEOPIG trailer
TAR_OFFSET = 0
if tarball.suffix == '.run' or tarball.stat().st_size > 100000:
try:
with open(tarball, 'rb') as f:
f.seek(-22, 2)
trailer = f.read(22)
if trailer[:6] == b'NEOPIG':
TAR_OFFSET = int(trailer[6:22].decode(), 16)
logger.info(f"Detected .run format, tarball offset: {TAR_OFFSET}")
except Exception:
pass
# Open tarball temporarily to build index
tar = _open_tarball()
# Build member lookup and media index
for member in tar.getmembers():
TAR_MEMBERS[member.name] = member
# Index media files by hash for O(1) lookup
name = member.name
if '/vault/' in name or '/media/' in name:
# Extract hash from filename (hash.ext)
basename = Path(name).stem # removes extension
if len(basename) == 32 and all(c in '0123456789abcdef' for c in basename):
TAR_MEDIA_INDEX[basename] = name
logger.info(f"Indexed {len(TAR_MEDIA_INDEX)} media files")
# Get archive root from first member
first = list(TAR_MEMBERS.keys())[0]
ARCHIVE_ROOT = first.split('/')[0]
logger.info(f"Archive root: {ARCHIVE_ROOT}")
# Extract database to temp (SQLite needs real file)
# Try neopig.db first (new format), then archive.db (legacy)
db_member = f"{ARCHIVE_ROOT}/neopig.db"
if db_member not in TAR_MEMBERS:
db_member = f"{ARCHIVE_ROOT}/archive.db"
if db_member in TAR_MEMBERS:
temp_dir = tempfile.mkdtemp(prefix="neopig_")
db_name = Path(db_member).name
TEMP_DB_PATH = f"{temp_dir}/{db_name}"
member = TAR_MEMBERS[db_member]
f = tar.extractfile(member)
if f:
with open(TEMP_DB_PATH, 'wb') as out:
out.write(f.read())
DB_PATH = TEMP_DB_PATH
logger.info(f"Extracted database to: {TEMP_DB_PATH}")
else:
logger.warning("No neopig.db or archive.db found in tarball")
# Close the temporary tar handle (requests will open their own)
tar.close()
def main():
global DB_PATH, VAULT_PATH
parser = argparse.ArgumentParser(description="neopig SERP")
parser.add_argument("tarball", nargs='?', help="Path to archive.tar.gz or .run file")
parser.add_argument("--port", type=int, default=31337)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--db", default="data/neopig.db")
parser.add_argument("--vault", default="data/vault")
args = parser.parse_args()
# Tarball mode
if args.tarball:
init_tarball_mode(args.tarball)
logger.info(f"Starting neopig SERP (archive mode) on {args.host}:{args.port}")
else:
DB_PATH = args.db
VAULT_PATH = Path(args.vault)
logger.info(f"Starting neopig SERP on {args.host}:{args.port}")
logger.info(f"Database: {DB_PATH}")
logger.info(f"Vault: {VAULT_PATH}")
uvicorn.run(app, host=args.host, port=args.port)
if __name__ == "__main__":
main()