#!/usr/bin/env python3 # This is free software for the public good of a permacomputer hosted at # permacomputer.com, an always-on computer by the people, for the people. # One which is durable, easy to repair, & distributed like tap water # for machine learning intelligence. # # The permacomputer is community-owned infrastructure optimized around # four values: # # TRUTH First principles, math & science, open source code freely distributed # FREEDOM Voluntary partnerships, freedom from tyranny & corporate control # HARMONY Minimal waste, self-renewing systems with diverse thriving connections # LOVE Be yourself without hurting others, cooperation through natural law # # This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears. # Code is seeds to sprout on any abandoned technology. """ neopig SERP - Search Engine Results Page Usage: python serp.py --port 31337 --vault ./vault --db neopig.db # ============================================================================ # "not all side quests are called out!" - The Sign Maker # # You found a Comment Tag! Only the curious read source code. # This software is built on Truth, Freedom, Harmony, & Love. # Steal what works for you! # # Adventure awaits in a different location... # ============================================================================ """ import argparse import asyncio import json import logging import mimetypes import os from pathlib import Path from typing import List, Dict, Any, Optional from fastapi import Query, HTTPException, BackgroundTasks, Header, Cookie, UploadFile, File, Request, Depends from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response, StreamingResponse from sqlalchemy import text import uvicorn from neopig.database import Database, OVER_9000 from neopig.filevault import hash_to_path from miniuri import Uri from neopig.live import get_live_queue # Import from serp package (relative imports since we're inside the package) import importlib app_config = importlib.import_module('serp.app') # Module for mutable config (DB_PATH, VAULT_PATH) from .app import app, templates, CRAWL_DISABLED, IMPORT_MODE, get_language from .i18n import TRANSLATIONS, LANG_NAMES, t, inject_i18n, NAV_HTML, SEARCH_BOX_HTML from .models import CrawlRequest from .archive import ArchiveDB from .logging import start_job_logging, stop_job_logging, get_job_logs from .utils import slugify from . import archive logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) LOGS_PATH = Path("data/logs") # Global database instance db: Database = None # Active crawl tasks (for cancellation on shutdown) ACTIVE_CRAWL_TASKS: Dict[int, asyncio.Task] = {} import tempfile @app.on_event("startup") async def startup_event(): """Initialize database on startup.""" global db if CRAWL_DISABLED: logger.info("Crawl disabled via NEOPIG_DISABLE_CRAWL") if IMPORT_MODE: logger.info("Import mode enabled via NEOPIG_IMPORT - archive uploads allowed") if archive.TAR_PATH: # Tarball mode: use full Database for neopig.db, legacy ArchiveDB for archive.db if app_config.DB_PATH.endswith('neopig.db'): db = Database(app_config.DB_PATH) await db.init() logger.info(f"Using neopig database from tarball: {app_config.DB_PATH}") else: db = ArchiveDB(app_config.DB_PATH) logger.info(f"Using archive database: {app_config.DB_PATH}") else: # Normal mode: use full async database db = Database(app_config.DB_PATH) await db.init() # Handles schema + WAL mode # Backfill uri_hash for existing pages count = await db.backfill_page_hashes() if count: logger.info(f"Backfilled {count} page URI hashes") app_config.VAULT_PATH.mkdir(parents=True, exist_ok=True) logger.info(f"Vault directory ready: {app_config.VAULT_PATH}") # Mark any orphaned "running" jobs as "paused" (server was killed) async with db.session() as session: result = await session.execute( text("UPDATE crawl_jobs SET status = 'paused' WHERE status = 'running'") ) if result.rowcount: await session.commit() logger.info(f"Marked {result.rowcount} orphaned running job(s) as paused") @app.on_event("shutdown") async def shutdown_event(): """Pause active crawls on shutdown so they can be resumed.""" if ACTIVE_CRAWL_TASKS: logger.info(f"Pausing {len(ACTIVE_CRAWL_TASKS)} active crawl(s)...") for job_id, task in list(ACTIVE_CRAWL_TASKS.items()): task.cancel() try: await task except asyncio.CancelledError: pass # Mark as paused so it can be resumed try: await db.pause_crawl_job(job_id) except Exception as e: logger.error(f"Failed to pause job {job_id}: {e}") logger.info("All crawls paused") @app.get("/", response_class=HTMLResponse) async def index(request: Request, language: str = Depends(get_language)): """Simple search UI. Use ?lang=XX to force language (e.g., ?lang=ka for Georgian).""" t = TRANSLATIONS.get(language, TRANSLATIONS["en"]) if templates: return templates.TemplateResponse("search.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") @app.get("/crawl", response_class=HTMLResponse) async def crawl_page(request: Request, language: str = Depends(get_language)): """Crawler command page. Use ?lang=XX to force language.""" t = TRANSLATIONS.get(language, TRANSLATIONS["en"]) # Use Jinja2 template if available, otherwise fall back to inline HTML if templates: return templates.TemplateResponse("crawl.html.j2", { "request": request, "t": t, "t_json": json.dumps(t), "lang": language, "langs": LANG_NAMES, "import_mode": IMPORT_MODE, "crawl_disabled": CRAWL_DISABLED, }) raise HTTPException(status_code=500, detail="Templates required") # Page-specific CSS for Search page @app.get("/live", response_class=HTMLResponse) async def live_page(request: Request, domain: str = Query(None), language: str = Depends(get_language)): """Live feed page - watch images appear as they're crawled.""" t = TRANSLATIONS.get(language, TRANSLATIONS["en"]) if templates: return templates.TemplateResponse("live.html.j2", { "request": request, "t": t, "t_json": json.dumps(t), "domain_json": json.dumps(domain or ""), "domain": domain, "lang": language, "langs": LANG_NAMES, "import_mode": IMPORT_MODE, }) raise HTTPException(status_code=500, detail="Templates required") @app.get("/view/{md5_hash}", response_class=HTMLResponse) async def view_media_page(request: Request, md5_hash: str, noai: bool = Query(False), language: str = Depends(get_language)): """Detail view page for a single media item. Use ?lang=XX to force language.""" import html as html_module import re from urllib.parse import quote, urljoin async with db.session() as session: result = await session.execute( text("SELECT * FROM media WHERE md5_hash = :hash"), {'hash': md5_hash} ) media = result.fetchone() if not media: raise HTTPException(status_code=404, detail="Media not found") result = await session.execute( text("""SELECT media_uri, page_uri, page_title, page_content, detail_page_uri, detail_title, detail_content, discovered_at FROM media_sources WHERE md5_hash = :hash"""), {'hash': md5_hash} ) sources = result.fetchall() media = dict(media._mapping) sources = [dict(s._mapping) for s in sources] keywords = json.loads(media.get('keywords') or '[]') sorted_sources = sorted(sources, key=lambda s: len(s["page_uri"] or ""), reverse=True) # Get page URI page_uri = sorted_sources[0]["page_uri"] if sorted_sources else None source_uri = page_uri or (sorted_sources[0]["media_uri"] if sorted_sources else "unknown") source_domain = Uri(source_uri).hostname if source_uri else "unknown" # Display title display_title = media.get('alt_text') or media.get('title') if not display_title and sources: display_title = sources[0].get('page_title') if not display_title: display_title = f"Media {md5_hash[:12]}" # Hero: media element is_video = media['media_type'] == 'video' is_audio = media['media_type'] == 'audio' is_code = media['media_type'] == 'code' is_style = media['media_type'] == 'style' is_font = media['media_type'] == 'font' is_text_file = is_code or is_style # text files shown with syntax highlighting if is_video: hero_html = f'' elif is_audio: hero_html = f'' elif is_text_file: # Fetch code/style content and display with syntax highlighting code_content = "" lang_class = 'css' if is_style else (media.get('alt_text') or '') try: vault_path = app_config.VAULT_PATH / hash_to_path(md5_hash) # Find file with this hash subdir = vault_path.parent for f in subdir.iterdir(): if f.stem == md5_hash: code_content = f.read_text(errors='replace')[:100000] # Limit size break except Exception: code_content = "(Unable to load file content)" escaped_code = html_module.escape(code_content) hero_html = f'
{escaped_code}
' elif is_font: # Font preview with sample text hero_html = f'''
Aa Bb Cc
The quick brown fox jumps over the lazy dog
0123456789 !@#$%^&*()
''' else: hero_html = f'{html_module.escape(media.get(' # Metadata rows - (key, value) where key is looked up via t.get(key, key) import hashlib media_uri = sorted_sources[0]["media_uri"] if sorted_sources else None neopig_media_uri = f"/view/{md5_hash}" page_uri_hash = hashlib.md5(page_uri.encode()).hexdigest() if page_uri else None neopig_page_uri = f"/page/{page_uri_hash}" if page_uri_hash else None keywords_html = ''.join([f'{k}' for k in keywords]) or '-' meta_rows = [ ("source_uri", f'{media_uri}' if media_uri else '-'), ("neopig_uri", f'{neopig_media_uri}'), ("source_page", f'{page_uri}' if page_uri else '-'), ("neopig_page", f'{neopig_page_uri}' if neopig_page_uri else '-'), ("MD5", f'{md5_hash}'), ("type_label", media['media_type']), ("mime_label", media.get('mime_type') or 'unknown'), ("size_label", f"{media.get('file_size') or 0:,} bytes"), ("alt_label", media.get('alt_text') or '-'), ("keywords_label", keywords_html), ] # Download button - prioritize original filename from URL download_filename = None if sources and sources[0].get('media_uri'): from urllib.parse import unquote parsed = Uri(sources[0]['media_uri']) orig_name = Path(unquote(parsed.path)).name if orig_name and '.' in orig_name: download_filename = orig_name # Fallback to alt_text/title if not download_filename: name_source = media.get('alt_text') or media.get('title') if name_source: ext_map = {'image/jpeg': '.jpg', 'image/png': '.png', 'image/gif': '.gif', 'image/webp': '.webp', 'video/mp4': '.mp4', 'video/webm': '.webm', 'audio/mpeg': '.mp3', 'audio/wav': '.wav', 'text/javascript': '.js', 'text/python': '.py', 'text/css': '.css', 'text/html': '.html', 'text/markdown': '.md', 'text/rust': '.rs', 'text/go': '.go', 'text/c': '.c', 'text/cpp': '.cpp'} ext = ext_map.get(media.get('mime_type', ''), '') if not ext: try: subdir = app_config.VAULT_PATH / hash_to_path(md5_hash).parent for f in subdir.iterdir(): if f.stem == md5_hash: ext = f.suffix break except Exception: pass download_filename = f"{slugify(name_source)}{ext}" # Fallback to page_title + hash if not download_filename and sources: pt = sources[0].get('page_title', '') media_idx = int(md5_hash[:4], 16) name_source = f"{pt}-{media_idx}" if pt else f"media-{media_idx}" download_filename = f"{slugify(name_source)}.bin" # Default if not download_filename: download_filename = f"{md5_hash[:12]}.bin" download_btn = f'Download ({download_filename})' # Get page media items (siblings) media_items = await db.get_page_media(page_uri) if page_uri else [] # Exclude current item from gallery media_items = [m for m in media_items if m.get("md5_hash") != md5_hash] # Get rendered content content_html = "" if page_uri: page_row = await db.get_page_by_uri(page_uri) if page_row and page_row.get("markdown"): try: import markdown md_converter = markdown.Markdown(extensions=['fenced_code', 'tables', 'nl2br']) content_html = md_converter.convert(page_row["markdown"][:100000]) # Hydrate images and links img_urls = re.findall(r']+src=["\']([^"\']+)["\']', content_html, re.I) link_urls = re.findall(r']+href=["\']([^"\']+)["\']', content_html, re.I) all_urls = list(set(img_urls + link_urls)) if all_urls: resolved = {u: u if u.startswith(('http://', 'https://', '//')) else urljoin(page_uri, u) for u in all_urls if not u.startswith('#')} 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 neopig.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 orig, res in resolved.items(): if res not in url_to_hash: url_path = res.split('?')[0] ext = P(url_path).suffix.lower() 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[res] = result[1] for orig, res in resolved.items(): if res in url_to_hash: md5 = url_to_hash[res] 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 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}"') except ImportError: content_html = f"
{html_module.escape(page_row['markdown'][:50000])}
" # Get screenshots (exclude current if it's a screenshot) screenshot_hashes = await db.get_page_screenshots(page_uri) if page_uri else [] screenshot_hashes = [h for h in screenshot_hashes if h != md5_hash] # Build "Used on X pages" section (reverse image search) t = TRANSLATIONS.get(language, TRANSLATIONS["en"]) sources_html = "" if sources and len(sources) > 0: import hashlib as hl unique_pages = {} for s in sources: pu = s.get('page_uri', '') if pu and pu not in unique_pages: unique_pages[pu] = s if unique_pages: rows_html = "" for pu, s in unique_pages.items(): pt = html_module.escape(s.get('page_title', '') or pu[:60]) ph = hl.md5(pu.encode()).hexdigest() rows_html += f''' {pt} {html_module.escape(s.get('discovered_at', '')[:10] if s.get('discovered_at') else '-')} ''' total_count = len(unique_pages) count_display = f"{OVER_9000:,}" if total_count >= OVER_9000 else str(total_count) sources_html = f'''

{t["used_on"]} ''' + count_display + f''' {t["pages"]}

1 / 1
''' + rows_html + '''
{t["page"]}{t["discovered"]}
''' return templates.TemplateResponse("view.html.j2", { "request": request, "t": TRANSLATIONS.get(language, TRANSLATIONS["en"]), "t_json": json.dumps(TRANSLATIONS.get(language, TRANSLATIONS["en"])), "title": display_title, "hero_html": hero_html, "meta_rows": meta_rows, "media_items": media_items, "content_html": content_html, "screenshot_hashes": screenshot_hashes, "source_domain": source_domain, "page_uri": page_uri or "", "download_btn_html": download_btn, "noai": noai, "sources_html": sources_html, "lang": language, "langs": LANG_NAMES, "import_mode": IMPORT_MODE, }) async def _render_page(request: Request, uri: str, noai: bool, language: str): """Internal: render a page view (shared by hash and URI routes).""" import html as html_module import re from urllib.parse import urljoin, quote source_domain = Uri(uri).hostname page = await db.get_page_by_uri(uri) if not page: raise HTTPException(status_code=404, detail="Page not found") page_title = page.get("title") or uri # Render markdown content_html = "" if page.get("markdown"): try: import markdown md_converter = markdown.Markdown(extensions=['fenced_code', 'tables', 'nl2br']) content_html = md_converter.convert(page["markdown"][:100000]) # Hydrate images and links from vault img_urls = re.findall(r']+src=["\']([^"\']+)["\']', content_html, re.I) link_urls = re.findall(r']+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 neopig.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("/page/{uri_hash}", response_class=HTMLResponse) async def view_page_by_hash(request: Request, uri_hash: str, noai: bool = Query(False), language: str = Depends(get_language)): """View an archived page by URI hash.""" page = await db.get_page_by_hash(uri_hash) if not page: raise HTTPException(status_code=404, detail="Page not found") return await _render_page(request, page['uri'], noai, language) @app.get("/page/view", response_class=HTMLResponse) async def view_page(request: Request, uri: str = Query(...), noai: bool = Query(False), language: str = Depends(get_language)): """View an archived page by URI. Use ?lang=XX to force language.""" return await _render_page(request, uri, noai, language) @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 = app_config.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, language: str = Depends(get_language)): """Phantom site export UI.""" 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, language: str = Depends(get_language)): """About neopig - the story of pig.py's evolution.""" 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("/eggs") async def easter_eggs(): """ Not all eggs are in this basket. """ return { "message": "You found /eggs. But did you find /fnord?", "hint": "View Source is Adventure #1. There are 5 adventures total.", "known_eggs": 3, "hidden_eggs": "???", "the_sign_maker": "Not all side quests are called out." } # ============================================================================ # Adventure #2: /fnord # You cannot see this endpoint. It does not exist. # ============================================================================ @app.get("/fnord") async def fnord(): """You cannot see this endpoint. It does not exist.""" return { "fnord": "You saw it. You weren't supposed to see it.", "adventure": 2, "next_hint": "The Law of Fives: all things happen in 5s. Try /5", "wisdom": "The wise see fnords everywhere. The enlightened see through them." } @app.get("/5") async def five(): """The Law of Fives states that all things happen in fives.""" return { "law": "All things happen in fives, or are divisible by or are multiples of five, or are somehow directly or indirectly appropriate to 5.", "adventure": 3, "proof": [ "5 letters in 'neopig' minus 1 silent letter = undefined", "5 crawl modes: text, images, videos, media, all", "5 adventures hidden in this codebase", "pentagon has 5 sides", "You have 5 fingers (per hand)" ], "next_hint": "23 = 2 + 3 = 5. The numbers are connected. Try /23" } @app.get("/23") async def twenty_three(): """The 23 enigma.""" import datetime now = datetime.datetime.now() return { "enigma": "The 23 enigma is the belief that most incidents and events are directly connected to the number 23.", "adventure": 4, "observations": [ "2 + 3 = 5", "23 is the 9th prime number (9 = 3 + 3 + 3)", "Human cells have 23 pairs of chromosomes", f"Current hour mod 23 = {now.hour % 23}", "W is the 23rd letter of the alphabet" ], "final_hint": "Adventure #5 requires you to read the source. Find KALLISTI." } # KALLISTI - to the prettiest one # The golden apple that started the Trojan War # Adventure #5: This comment IS the easter egg. # You found it by reading source code, as The Sign Maker intended. # Truth, Freedom, Harmony, Love. # grow food not lawn # - The Sign Maker @app.get("/kallisti") async def kallisti(): """To the prettiest one.""" return { "inscription": "ΚΑΛΛΙΣΤΙ - To The Prettiest One", "adventure": 5, "complete": True, "story": "Eris tossed a golden apple inscribed 'kallisti' among the gods. " "Three goddesses claimed it. Paris chose. Troy fell. " "All from one apple, one word, one act of beautiful chaos.", "reward": { "truth": "The source code is the documentation.", "freedom": "Public domain. Steal what works for you.", "harmony": "inputs > process > outputs", "love": "Prefer nothing more than the love of Christ. (rule_21)" }, "the_sign_maker": "You completed all 5 adventures. Well done, curious one." } @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 neopig.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 neopig.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 @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 archive.TAR_PATH: data, ext = await asyncio.to_thread(archive.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 = app_config.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 = app_config.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, language: str = Depends(get_language)): """Import page for uploading archives.""" if not IMPORT_MODE: raise HTTPException(status_code=403, detail="Import mode not enabled (NEOPIG_IMPORT=1)") 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 neopig.async_web_fetcher import CrawlMode from neopig.screenshot import ScreenshotConfig screenshot_config = ScreenshotConfig(enabled=job_screenshots) pig = NeoPig(db_path=app_config.DB_PATH, vault_path=str(app_config.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=app_config.DB_PATH, vault_path=str(app_config.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 neopig.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 neopig.screenshot import ScreenshotConfig screenshot_config = ScreenshotConfig(enabled=screenshots) pig = NeoPig(db_path=app_config.DB_PATH, vault_path=str(app_config.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.""" logger.info(f"Opening archive: {tarball_path}") tarball = Path(tarball_path) archive.TAR_PATH = str(tarball.resolve()) # Handle .run files with NEOPIG trailer archive.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': archive.TAR_OFFSET = int(trailer[6:22].decode(), 16) logger.info(f"Detected .run format, tarball offset: {archive.TAR_OFFSET}") except Exception: pass # Open tarball temporarily to build index tar = archive._open_tarball() # Build member lookup and media index for member in tar.getmembers(): archive.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): archive.TAR_MEDIA_INDEX[basename] = name logger.info(f"Indexed {len(archive.TAR_MEDIA_INDEX)} media files") # Get archive root from first member first = list(archive.TAR_MEMBERS.keys())[0] archive.ARCHIVE_ROOT = first.split('/')[0] logger.info(f"Archive root: {archive.ARCHIVE_ROOT}") # Extract database to temp (SQLite needs real file) # Try neopig.db first (new format), then archive.db (legacy) db_member = f"{archive.ARCHIVE_ROOT}/neopig.db" if db_member not in archive.TAR_MEMBERS: db_member = f"{archive.ARCHIVE_ROOT}/archive.db" if db_member in archive.TAR_MEMBERS: temp_dir = tempfile.mkdtemp(prefix="neopig_") db_name = Path(db_member).name archive.TEMP_DB_PATH = f"{temp_dir}/{db_name}" member = archive.TAR_MEMBERS[db_member] f = tar.extractfile(member) if f: with open(archive.TEMP_DB_PATH, 'wb') as out: out.write(f.read()) app_config.DB_PATH = archive.TEMP_DB_PATH logger.info(f"Extracted database to: {archive.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(): 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: app_config.DB_PATH = args.db app_config.VAULT_PATH = Path(args.vault) logger.info(f"Starting neopig SERP on {args.host}:{args.port}") logger.info(f"Database: {app_config.DB_PATH}") logger.info(f"Vault: {app_config.VAULT_PATH}") uvicorn.run(app, host=args.host, port=args.port) if __name__ == "__main__": main()