diff --git a/database.py b/database.py index 86743f2..b608340 100644 --- a/database.py +++ b/database.py @@ -8,6 +8,7 @@ Stores: - Pages for full-text search """ +import hashlib import json import logging from datetime import datetime, timezone @@ -105,8 +106,11 @@ class Page(Base): id = Column(Integer, primary_key=True, autoincrement=True) uri = Column(Text, nullable=False, unique=True) + uri_hash = Column(String(32), unique=True) # MD5 of URI for clean URLs path = Column(Text) title = Column(Text) + description = Column(Text) # Meta description + keywords = Column(Text) # JSON array of keywords content = Column(Text) markdown = Column(Text) raw_html = Column(Text) @@ -117,6 +121,7 @@ class Page(Base): __table_args__ = ( Index('idx_pages_uri', 'uri'), + Index('idx_pages_uri_hash', 'uri_hash'), ) @@ -163,39 +168,75 @@ class Database: async with self._engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + # Add new columns to existing databases (ignore if already exist) + for col in ['description', 'keywords']: + try: + await conn.execute(text(f"ALTER TABLE pages ADD COLUMN {col} TEXT")) + except Exception: + pass # Column already exists + + # Add uri_hash column for clean URLs + try: + await conn.execute(text("ALTER TABLE pages ADD COLUMN uri_hash TEXT")) + await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_pages_uri_hash ON pages(uri_hash)")) + except Exception: + pass # Column already exists + # Create FTS5 virtual table (SQLAlchemy doesn't handle virtual tables) + # Includes description and keywords for better search scoring + # Check if FTS table needs rebuilding (old schema didn't have description/keywords) + try: + result = await conn.execute(text("SELECT sql FROM sqlite_master WHERE name='pages_fts'")) + row = result.fetchone() + if row and 'description' not in (row[0] or ''): + # Old schema - drop and recreate + await conn.execute(text("DROP TABLE IF EXISTS pages_fts")) + logger.info("Rebuilding FTS5 index with new schema") + except Exception: + pass + await conn.execute(text(""" CREATE VIRTUAL TABLE IF NOT EXISTS pages_fts USING fts5( - title, content, uri, path, + title, description, keywords, content, uri, path, content='pages', content_rowid='id' ) """)) - # FTS triggers + # FTS triggers - drop and recreate to ensure they match current schema + await conn.execute(text("DROP TRIGGER IF EXISTS pages_ai")) + await conn.execute(text("DROP TRIGGER IF EXISTS pages_ad")) + await conn.execute(text("DROP TRIGGER IF EXISTS pages_au")) + await conn.execute(text(""" - CREATE TRIGGER IF NOT EXISTS pages_ai AFTER INSERT ON pages BEGIN - INSERT INTO pages_fts(rowid, title, content, uri, path) - VALUES (new.id, new.title, new.content, new.uri, new.path); + CREATE TRIGGER pages_ai AFTER INSERT ON pages BEGIN + INSERT INTO pages_fts(rowid, title, description, keywords, content, uri, path) + VALUES (new.id, new.title, new.description, new.keywords, new.content, new.uri, new.path); END """)) await conn.execute(text(""" - CREATE TRIGGER IF NOT EXISTS pages_ad AFTER DELETE ON pages BEGIN - INSERT INTO pages_fts(pages_fts, rowid, title, content, uri, path) - VALUES ('delete', old.id, old.title, old.content, old.uri, old.path); + CREATE TRIGGER pages_ad AFTER DELETE ON pages BEGIN + INSERT INTO pages_fts(pages_fts, rowid, title, description, keywords, content, uri, path) + VALUES ('delete', old.id, old.title, old.description, old.keywords, old.content, old.uri, old.path); END """)) await conn.execute(text(""" - CREATE TRIGGER IF NOT EXISTS pages_au AFTER UPDATE ON pages BEGIN - INSERT INTO pages_fts(pages_fts, rowid, title, content, uri, path) - VALUES ('delete', old.id, old.title, old.content, old.uri, old.path); - INSERT INTO pages_fts(rowid, title, content, uri, path) - VALUES (new.id, new.title, new.content, new.uri, new.path); + CREATE TRIGGER pages_au AFTER UPDATE ON pages BEGIN + INSERT INTO pages_fts(pages_fts, rowid, title, description, keywords, content, uri, path) + VALUES ('delete', old.id, old.title, old.description, old.keywords, old.content, old.uri, old.path); + INSERT INTO pages_fts(rowid, title, description, keywords, content, uri, path) + VALUES (new.id, new.title, new.description, new.keywords, new.content, new.uri, new.path); END """)) + # Rebuild FTS index from existing pages (idempotent - FTS5 handles dupes) + await conn.execute(text(""" + INSERT OR IGNORE INTO pages_fts(rowid, title, description, keywords, content, uri, path) + SELECT id, title, description, keywords, content, uri, path FROM pages + """)) + self._initialized = True logger.info(f"Database initialized: {self.db_path}") @@ -412,10 +453,12 @@ class Database: crawl_job_id: int = None ) -> None: """Store a page for full-text search.""" + uri_hash = hashlib.md5(uri.encode()).hexdigest() async with self.session() as session: # Use SQLite upsert (INSERT OR REPLACE) stmt = sqlite_insert(Page).values( uri=uri, + uri_hash=uri_hash, path=path, title=title, content=content[:100000], @@ -428,6 +471,7 @@ class Database: stmt = stmt.on_conflict_do_update( index_elements=['uri'], set_={ + 'uri_hash': stmt.excluded.uri_hash, 'path': stmt.excluded.path, 'title': stmt.excluded.title, 'content': stmt.excluded.content, @@ -453,7 +497,7 @@ class Database: fts_query = ' '.join(f'"{word}"*' for word in query.split()) result = await session.execute( text(""" - SELECT p.uri, p.path, p.title, + SELECT p.uri, p.uri_hash, p.path, p.title, snippet(pages_fts, 1, '', '', '...', 40) as snippet FROM pages_fts JOIN pages p ON pages_fts.rowid = p.id @@ -470,7 +514,7 @@ class Database: if not results: like_q = f'%{query}%' stmt = ( - select(Page.uri, Page.path, Page.title, + select(Page.uri, Page.uri_hash, Page.path, Page.title, func.substr(Page.content, 1, 200).label('snippet')) .where(or_( Page.title.ilike(like_q), @@ -548,6 +592,27 @@ class Database: row = result.scalar_one_or_none() return self._model_to_dict(row) if row else None + async def get_page_by_hash(self, uri_hash: str) -> Optional[Dict[str, Any]]: + """Get page by URI hash (for clean URLs).""" + async with self.session() as session: + stmt = select(Page).where(Page.uri_hash == uri_hash) + result = await session.execute(stmt) + row = result.scalar_one_or_none() + return self._model_to_dict(row) if row else None + + async def backfill_page_hashes(self) -> int: + """Backfill uri_hash for pages that don't have one. Returns count updated.""" + async with self.session() as session: + stmt = select(Page).where(Page.uri_hash == None) + result = await session.execute(stmt) + pages = result.scalars().all() + count = 0 + for page in pages: + page.uri_hash = hashlib.md5(page.uri.encode()).hexdigest() + count += 1 + await session.commit() + return count + async def get_crawl_jobs(self, limit: int = 50) -> List[Dict[str, Any]]: """Get recent crawl jobs.""" async with self.session() as session: diff --git a/html2md.py b/html2md.py index d5734b9..7215bf1 100644 --- a/html2md.py +++ b/html2md.py @@ -426,8 +426,12 @@ class SmartMarkdownConverter: src = self._resolve_url(child.get('src', '')) alt = child.get('alt', '') if src: - # Images should be block-level, not inline - parts.append(f"\n\n![{alt}]({src})\n\n") + # Emoji images (alt like :smile:) stay inline + if alt.startswith(':') and alt.endswith(':'): + parts.append(f"![{alt}]({src})") + else: + # Regular images are block-level + parts.append(f"\n\n![{alt}]({src})\n\n") elif tag == 'br': parts.append('\n') diff --git a/neopig.py b/neopig.py index c7af4a8..64bd4c9 100644 --- a/neopig.py +++ b/neopig.py @@ -915,24 +915,156 @@ def trim_html_wrapper(html: str) -> str: return str(soup) +def extract_meta_from_html(html: str) -> tuple[str, list[str]]: + """Extract meta description and keywords from HTML. + + Returns: + Tuple of (description, keywords_list) + """ + from bs4 import BeautifulSoup + import re + + soup = BeautifulSoup(html, 'html.parser') + description = "" + keywords = [] + + # Extract meta description + meta_desc = soup.find('meta', attrs={'name': re.compile(r'^description$', re.I)}) + if meta_desc and meta_desc.get('content'): + description = meta_desc['content'].strip()[:500] + + # Extract meta keywords + meta_kw = soup.find('meta', attrs={'name': re.compile(r'^keywords$', re.I)}) + if meta_kw and meta_kw.get('content'): + raw_kw = meta_kw['content'] + keywords = [k.strip().lower() for k in raw_kw.split(',') if k.strip()] + + # Also check og:description as fallback + if not description: + og_desc = soup.find('meta', attrs={'property': 'og:description'}) + if og_desc and og_desc.get('content'): + description = og_desc['content'].strip()[:500] + + # Extract from article:tag meta tags (common in blogs) + for tag_meta in soup.find_all('meta', attrs={'property': 'article:tag'}): + if tag_meta.get('content'): + keywords.append(tag_meta['content'].strip().lower()) + + # Dedupe keywords + keywords = list(dict.fromkeys(keywords))[:20] + + return description, keywords + + +def _process_single_page(args: tuple) -> tuple: + """Process a single page for markdown conversion (runs in thread within process).""" + from html2md import html_to_markdown + import json + + page_id, uri, raw_html, trim_wrapper = args + try: + description, keywords = extract_meta_from_html(raw_html) + raw = raw_html + if trim_wrapper: + raw = trim_html_wrapper(raw) + new_markdown = html_to_markdown(raw, base_url=uri)[:200000] + return (page_id, new_markdown, description, json.dumps(keywords) if keywords else None, None) + except Exception as e: + return (page_id, None, None, None, str(e)) + + +def _process_chunk(args: tuple) -> dict: + """Process a chunk of pages in a subprocess with thread workers. + + Each process gets a chunk and spawns threads to burn through it. + Each thread has its own DB connection - WAL mode allows concurrent writes. + """ + import sqlite3 + import threading + from concurrent.futures import ThreadPoolExecutor + + chunk, db_path, trim_wrapper, threads_per_process, progress_counter = args + + # Thread-local storage for DB connections + thread_local = threading.local() + results = {'updated': 0, 'errors': 0} + results_lock = threading.Lock() + + def get_conn(): + """Get thread-local DB connection.""" + if not hasattr(thread_local, 'conn'): + thread_local.conn = sqlite3.connect(db_path, timeout=60.0) + thread_local.conn.execute("PRAGMA journal_mode=WAL") + thread_local.conn.execute("PRAGMA synchronous=NORMAL") + thread_local.conn.execute("PRAGMA busy_timeout=60000") + return thread_local.conn + + def process_and_write(page_data): + """Process one page and write immediately with thread-local connection.""" + page_id, uri, raw_html = page_data + result = _process_single_page((page_id, uri, raw_html, trim_wrapper)) + page_id, markdown, description, keywords_json, error = result + + if error: + with results_lock: + results['errors'] += 1 + else: + # Each thread uses its own connection + conn = get_conn() + conn.execute( + "UPDATE pages SET markdown = ?, description = ?, keywords = ? WHERE id = ?", + (markdown, description, keywords_json, page_id) + ) + conn.commit() + with results_lock: + results['updated'] += 1 + + # Update shared progress counter (Manager proxy) + if progress_counter is not None: + progress_counter.value += 1 + + # Fan out to thread workers within this process + with ThreadPoolExecutor(max_workers=threads_per_process) as thread_executor: + list(thread_executor.map(process_and_write, chunk)) + + return {'updated': results['updated'], 'errors': results['errors'], 'total': len(chunk)} + + async def backfill_markdown(db_path: str, domain_filter: str = None, trim_wrapper: bool = False, quiet: bool = False): """Re-process stored HTML to regenerate markdown with smart structure detection. + ETL-style parallel processing: + - Divides work into chunks (1 per CPU core) + - Each process spawns 6 threads to burn through its chunk + - Immediate DB writes with WAL mode (thread-safe) + Args: db_path: Path to SQLite database domain_filter: Only process pages matching this domain (e.g., 'example.com') - trim_wrapper: Strip nav/header/footer/logo before conversion (applied before smart conversion) + trim_wrapper: Strip nav/header/footer/logo before conversion quiet: Disable progress bar """ - from html2md import html_to_markdown import aiosqlite + import multiprocessing + from concurrent.futures import ProcessPoolExecutor + import time + num_cpus = multiprocessing.cpu_count() + threads_per_process = 6 + logger.info(f"ETL mode: {num_cpus} processes x {threads_per_process} threads = {num_cpus * threads_per_process} workers") + + # Ensure WAL mode and columns exist async with aiosqlite.connect(db_path) as db: - db.row_factory = aiosqlite.Row + await db.execute("PRAGMA journal_mode=WAL") + for col in ['description', 'keywords']: + try: + await db.execute(f"ALTER TABLE pages ADD COLUMN {col} TEXT") + except Exception: + pass + await db.commit() # Build query with optional domain filter if domain_filter: - # Match domain in URI pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%" select_sql = "SELECT id, uri, raw_html FROM pages WHERE raw_html IS NOT NULL AND uri LIKE ?" params = (pattern,) @@ -950,27 +1082,54 @@ async def backfill_markdown(db_path: str, domain_filter: str = None, trim_wrappe total = len(rows) logger.info(f"Found {total} pages to process") - updated = 0 - errors = 0 - for row in tqdm(rows, desc="Markdown", unit="pages", disable=quiet): - if not row['raw_html']: - continue + # Convert to list of tuples + all_pages = [(row[0], row[1], row[2]) for row in rows if row[2]] - try: - raw = row['raw_html'] - if trim_wrapper: - raw = trim_html_wrapper(raw) - new_markdown = html_to_markdown(raw, base_url=row['uri'])[:200000] - await db.execute("UPDATE pages SET markdown = ? WHERE id = ?", (new_markdown, row['id'])) - updated += 1 - if updated % 100 == 0: - await db.commit() - except Exception as e: - errors += 1 - logger.debug(f"Error processing {row['uri']}: {e}") + # Divide into exactly num_cpus chunks (last chunk may be slightly larger) + chunks = [] + chunk_size = len(all_pages) // num_cpus + for i in range(num_cpus): + start = i * chunk_size + end = (i + 1) * chunk_size if i < num_cpus - 1 else len(all_pages) + chunks.append(all_pages[start:end]) + logger.info(f"Split into {len(chunks)} chunks of ~{chunk_size} pages each") - await db.commit() - logger.info(f"Backfill complete: {updated} updated, {errors} errors") + # Manager for cross-process shared state + with multiprocessing.Manager() as manager: + # Shared counter for real-time progress across all processes + progress_counter = manager.Value('i', 0) + + # Process chunks in parallel with per-page progress + with ProcessPoolExecutor(max_workers=num_cpus) as executor: + chunk_args = [(chunk, db_path, trim_wrapper, threads_per_process, progress_counter) for chunk in chunks] + futures = [executor.submit(_process_chunk, args) for args in chunk_args] + + # Poll shared counter for real-time progress + with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet) as pbar: + last_count = 0 + while True: + current = progress_counter.value + if current > last_count: + pbar.update(current - last_count) + last_count = current + # Check if all done + if all(f.done() for f in futures): + # Final update + current = progress_counter.value + if current > last_count: + pbar.update(current - last_count) + break + time.sleep(0.05) + + # Collect results + total_updated = 0 + total_errors = 0 + for f in futures: + result = f.result() + total_updated += result['updated'] + total_errors += result['errors'] + + logger.info(f"Backfill complete: {total_updated} updated, {total_errors} errors") async def backfill_screenshots( @@ -1048,78 +1207,72 @@ async def backfill_screenshots( total = to_process + skipped_count logger.info(f"Found {to_process} to process ({skipped_count} already completed)") - captured = 0 - failed = 0 - bytes_saved = 0 domain_last_fetched = {} # Track last fetch time per domain crawl_delay = 2.0 # Default crawl delay in seconds - # Concurrent workers in fast mode (based on CPU count) + # Concurrent workers in fast mode (2x CPU count since screenshot is I/O bound) import multiprocessing - max_workers = multiprocessing.cpu_count() if fast_mode else 1 + max_workers = multiprocessing.cpu_count() * 2 if fast_mode else 1 semaphore = asyncio.Semaphore(max_workers) if fast_mode and max_workers > 1: logger.info(f"Fast mode: {max_workers} concurrent workers") - pbar = tqdm(total=total, initial=skipped_count, desc="Screenshots", unit="pages", disable=quiet) + pbar = tqdm(total=total, initial=skipped_count, desc="Screenshots", unit="pages", disable=quiet, smoothing=0.1) - # Lock for thread-safe updates - state_lock = asyncio.Lock() + # Thread-safe counters + stats_lock = asyncio.Lock() stats = {'captured': 0, 'failed': 0, 'bytes_saved': 0} - async def process_row(row): - nonlocal completed_uris + async def capture_one(page_uri: str, old_hash: str): + """Capture screenshot for one page. Returns (success, result_data) or (False, None).""" async with semaphore: - page_uri = row.page_uri - old_hash = row.md5_hash + try: + # Get content length for dynamic delay + async with db.session() as local_session: + from sqlalchemy import func as sqlfunc + result = await local_session.execute( + select(sqlfunc.length(Page.raw_html)).where(Page.uri == page_uri) + ) + content_length = result.scalar() or 0 + + # Capture screenshot + result = await capture.capture(page_uri, content_length=content_length) + if not result: + return False, None + return True, (page_uri, old_hash, result) + except Exception as e: + logger.debug(f"Capture error {page_uri}: {e}") + return False, None + + async def process_result(success, data): + """Process a captured screenshot - store and update DB.""" + nonlocal completed_uris + if not success or not data: + async with stats_lock: + stats['failed'] += 1 + pbar.update(1) + return + + page_uri, old_hash, result = data domain = Uri(page_uri).hostname.lower() - # Log current page (debug level to reduce noise) - logger.debug(f"Capturing: {page_uri}") - try: - # Enforce crawl delay (skip in fast mode) - if not fast_mode: - last_fetched = domain_last_fetched.get(domain, 0) - elapsed = time.time() - last_fetched - if elapsed < crawl_delay: - await asyncio.sleep(crawl_delay - elapsed) - domain_last_fetched[domain] = time.time() - - # Get content length for dynamic delay calculation - from sqlalchemy import func - content_length_result = await session.execute( - select(func.length(Page.raw_html)).where(Page.uri == page_uri) - ) - content_length = content_length_result.scalar() or 0 - - # Capture new screenshot (will be JPEG) with dynamic delay for large pages - result = await capture.capture(page_uri, content_length=content_length) - if not result: - failed += 1 - pbar.update(1) - continue - - # Normalize to list (oversized images return multiple chunks) chunks = result if isinstance(result, list) else [result] - - # Get old file size for comparison old_path = Path(vault_path) / old_hash[:2] / f"{old_hash}.png" old_size = old_path.stat().st_size if old_path.exists() else 0 - # Store all chunks + # Store chunks total_new_size = 0 for i, chunk in enumerate(chunks): new_hash = chunk['md5_hash'] new_data = chunk['data'] new_ext = chunk.get('format', 'jpg') - new_mime = chunk.get('mime_type', 'image/jpeg') total_new_size += len(new_data) if not await vault.exists(new_hash): await vault.store(new_hash, new_data, new_ext) - # Create symlink for each chunk + # Symlink parsed = Uri(page_uri) url_path = parsed.path.lstrip('/') or 'index' suffix = f'_{i}' if len(chunks) > 1 else '' @@ -1130,10 +1283,11 @@ async def backfill_screenshots( if not new_symlink.exists(): new_symlink.symlink_to(rel_path) - # Delete old PNG file FIRST (before metadata update) + # Delete old if delete_old and old_path.exists(): old_path.unlink() - bytes_saved += old_size - total_new_size + async with stats_lock: + stats['bytes_saved'] += old_size - total_new_size # Remove old symlink parsed = Uri(page_uri) @@ -1142,84 +1296,96 @@ async def backfill_screenshots( if old_symlink.exists(): old_symlink.unlink() - # Use first chunk for database record + # Update database new_hash = chunks[0]['md5_hash'] new_ext = chunks[0].get('format', 'jpg') new_mime = chunks[0].get('mime_type', 'image/jpeg') new_size = total_new_size - # Update database using ORM - # md5_hash is PRIMARY KEY, so we need to: insert new -> update refs -> delete old - from sqlalchemy import delete - - # Get old record data - old_media_result = await session.execute( - select(Media).where(Media.md5_hash == old_hash) - ) - old_media = old_media_result.scalar_one_or_none() - - if old_media and old_hash != new_hash: - # Insert new media record - new_media = Media( - md5_hash=new_hash, - media_type='screenshot', - mime_type=new_mime, - file_size=new_size, - keywords=old_media.keywords, - alt_text=old_media.alt_text, - title=old_media.title, - first_seen_at=datetime.now(timezone.utc).isoformat(), - analysis_status=old_media.analysis_status, - analysis_result=old_media.analysis_result, + async with db.session() as local_session: + from sqlalchemy import delete as sqldelete + old_media_result = await local_session.execute( + select(Media).where(Media.md5_hash == old_hash) ) - session.add(new_media) - await session.flush() + old_media = old_media_result.scalar_one_or_none() - # Update MediaSource references - await session.execute( - update(MediaSource) - .where(MediaSource.md5_hash == old_hash) - .values(md5_hash=new_hash) - ) + if old_media and old_hash != new_hash: + new_media = Media( + md5_hash=new_hash, + media_type='screenshot', + mime_type=new_mime, + file_size=new_size, + keywords=old_media.keywords, + alt_text=old_media.alt_text, + title=old_media.title, + first_seen_at=datetime.now(timezone.utc).isoformat(), + analysis_status=old_media.analysis_status, + analysis_result=old_media.analysis_result, + ) + local_session.add(new_media) + await local_session.flush() + await local_session.execute( + update(MediaSource).where(MediaSource.md5_hash == old_hash).values(md5_hash=new_hash) + ) + await local_session.execute(sqldelete(Media).where(Media.md5_hash == old_hash)) + elif old_media: + old_media.mime_type = new_mime + old_media.file_size = new_size + old_media.first_seen_at = datetime.now(timezone.utc).isoformat() + await local_session.commit() - # Delete old media record - await session.execute( - delete(Media).where(Media.md5_hash == old_hash) - ) - elif old_media: - # Same hash, just update metadata - old_media.mime_type = new_mime - old_media.file_size = new_size - old_media.first_seen_at = datetime.now(timezone.utc).isoformat() - - captured += 1 - # Commit after each capture to preserve progress - await session.commit() - - # Save state after each successful capture - if state_path: + async with stats_lock: + stats['captured'] += 1 completed_uris.add(page_uri) - try: - full_state = json.loads(state_path.read_text()) if state_path.exists() else {} - except Exception: - full_state = {} - full_state['backfill_screenshots'] = list(completed_uris) - state_path.write_text(json.dumps(full_state, indent=2)) + + # Save state + if state_path: + async with stats_lock: + try: + full_state = json.loads(state_path.read_text()) if state_path.exists() else {} + except Exception: + full_state = {} + full_state['backfill_screenshots'] = list(completed_uris) + state_path.write_text(json.dumps(full_state, indent=2)) pbar.update(1) except Exception as e: - # Truncate error message for cleaner logs err_msg = str(e).split('\n')[0][:60] logger.warning(f"Failed: {page_uri} - {err_msg}") - failed += 1 + async with stats_lock: + stats['failed'] += 1 pbar.update(1) + # Process - each worker flushes immediately when done + if fast_mode and max_workers > 1: + async def capture_and_process(row): + """Capture and immediately process/flush to DB.""" + success, data = await capture_one(row.page_uri, row.md5_hash) + await process_result(success, data) + + # Launch all tasks, they'll complete and flush independently + tasks = [asyncio.create_task(capture_and_process(row)) for row in rows] + # Wait for all to complete + await asyncio.gather(*tasks) + else: + # Sequential mode with crawl delay + for row in rows: + if not fast_mode: + domain = Uri(row.page_uri).hostname.lower() + last_fetched = domain_last_fetched.get(domain, 0) + elapsed = time.time() - last_fetched + if elapsed < crawl_delay: + await asyncio.sleep(crawl_delay - elapsed) + domain_last_fetched[domain] = time.time() + + success, data = await capture_one(row.page_uri, row.md5_hash) + await process_result(success, data) + pbar.close() - await session.commit() - logger.info(f"Backfill complete: {captured} captured, {skipped_count} skipped, {failed} failed") - if bytes_saved > 0: - logger.info(f"Space saved: {bytes_saved / 1024 / 1024:.1f} MB") + logger.info(f"Backfill complete: {stats['captured']} captured, {skipped_count} skipped, {stats['failed']} failed") + if stats['bytes_saved'] > 0: + logger.info(f"Space saved: {stats['bytes_saved'] / 1024 / 1024:.1f} MB") async def main(): @@ -1441,6 +1607,34 @@ async def main(): finally: if serp_process: serp_process.terminate() + try: + serp_process.wait(timeout=5) + except subprocess.TimeoutExpired: + serp_process.kill() + return + + # Handle --serve alone (just run the server) + if args.serve and not args.targets: + import subprocess + import sys + import signal + serp_script = Path(__file__).parent / 'serp.py' + if serp_script.exists(): + serp_cmd = [ + sys.executable, str(serp_script), + '--port', str(args.port), + '--db', args.db, + '--vault', args.vault, + ] + logger.info(f"Starting SERP server at http://localhost:{args.port}") + try: + proc = subprocess.Popen(serp_cmd) + proc.wait() # Block until server exits + except KeyboardInterrupt: + proc.terminate() + proc.wait(timeout=5) + else: + logger.error("serp.py not found") return # Require targets for crawling diff --git a/screenshot.py b/screenshot.py index 0ff41a2..73a9920 100644 --- a/screenshot.py +++ b/screenshot.py @@ -262,8 +262,23 @@ class ScreenshotCapture: ) if not result.success: - # Extract just the first line of error, skip wkhtmltoimage verbosity - error_msg = (result.error or 'unknown error').split('\n')[0][:80] + # Extract meaningful error from wkhtmltoimage output + raw_error = result.error or 'unknown error' + # Filter out wkhtmltoimage progress output (progress bars, loading messages) + lines = raw_error.split('\n') + # Skip lines that are progress bars or loading messages + meaningful = [l.strip() for l in lines if l.strip() + and not l.strip().startswith('[') # Progress bars like [> + and not l.strip().startswith('Loading') + and '%' not in l] # Percentage indicators + # Look for actual error/fail lines first + error_lines = [l for l in meaningful if 'error' in l.lower() or 'fail' in l.lower()] + if error_lines: + error_msg = error_lines[0][:60] + elif meaningful: + error_msg = meaningful[-1][:60] + else: + error_msg = 'capture failed' logger.warning(f"Screenshot failed: {uri} - {error_msg}") return None diff --git a/serp.py b/serp.py index a323c4c..eacc6a5 100644 --- a/serp.py +++ b/serp.py @@ -82,7 +82,7 @@ body { .nav a { color: #ff6b6b; text-decoration: none; } .nav a:hover { text-decoration: underline; } .nav .brand { font-weight: bold; font-size: 18px; } -.container { padding: 20px; max-width: 1200px; margin: 0 auto; } +.container { padding: 20px; } h1 { color: #ff6b6b; font-size: 20px; margin: 0 0 10px 0; } h3 { color: #ff6b6b; font-size: 16px; margin: 30px 0 15px 0; border-bottom: 1px solid #333; padding-bottom: 8px; } a { color: #ff6b6b; } @@ -103,9 +103,11 @@ a { color: #ff6b6b; } width: 40px; height: 40px; border-radius: 50%; margin: 0 10px 0 0; object-fit: cover; } -.content-rendered img.emoji { +.content-rendered img.emoji, +.content-rendered img[alt^=":"][alt$=":"], +.content-rendered img[src*="emoji"] { display: inline; width: 20px; height: 20px; - margin: 0 2px; vertical-align: text-bottom; + max-width: 20px; margin: 0 2px; vertical-align: text-bottom; } .content-rendered a { color: #ff6b6b; } .content-rendered pre, .content-rendered code { @@ -144,6 +146,254 @@ a { color: #ff6b6b; } """ +def render_detail_page( + title: str, + hero_html: str, + meta_rows: list, + media_items: list, + content_html: str, + screenshot_hashes: list, + source_domain: str, + page_uri: str, + download_btn_html: str = "", + noai: bool = False, +) -> str: + """ + Shared template for media view and page view. + + Args: + title: Page title + hero_html: HTML for hero section (media element or empty) + meta_rows: List of (label, value_html) tuples for metadata + media_items: List of media dicts with md5_hash, media_type + content_html: Rendered markdown/content HTML + screenshot_hashes: List of screenshot md5 hashes + source_domain: Domain for AI prompt + page_uri: Page URI for AI prompt + download_btn_html: Optional download button HTML + noai: Disable AI assistant + """ + import html as html_module + + # Build metadata section + meta_html = ''.join( + f'
{label}:{value}
' + for label, value in meta_rows + ) + + # Build media gallery + media_grid = "" + if media_items: + media_cards = [] + for m in media_items: + is_video = m.get("media_type") == "video" + if is_video: + el = f'' + else: + el = f'' + media_cards.append(f'''{el}''') + media_grid = f''' +
+

Media ({len(media_items)})

+
{''.join(media_cards)}
+
''' + + escaped_title = html_module.escape(title) + escaped_uri = html_module.escape(page_uri or "") + + return f""" + + + {escaped_title} - neopig + + + + + +
+ 🐷 neopig + Search + Live + Random + Crawl + Phantom +
+
+
+ + +
+
+ {f'
{hero_html}
' if hero_html else ''} +
+

{escaped_title}

+
{meta_html}
+ {download_btn_html} +
+
+ {media_grid} +
+
+ {f'

Content

{content_html}
' if content_html else ''} +
+ {f'''
+

+ ▼ Screenshot +

+
+ {''.join(f'Screenshot' for h in screenshot_hashes)} +
+
''' if screenshot_hashes else ''} +
+ + {'' if noai else f''' + '''} + +""" + + def read_from_tarball(path: str) -> bytes: """Read a file from the tarball. Path is relative to archive root.""" if not TAR_FILE or not ARCHIVE_ROOT: @@ -203,6 +453,10 @@ async def startup_event(): # Normal mode: use full async database db = Database(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") VAULT_PATH.mkdir(parents=True, exist_ok=True) logger.info(f"Vault directory ready: {VAULT_PATH}") @@ -553,7 +807,7 @@ SEARCH_HTML = """ searchColumns.classList.add('has-pages'); pageContainer.innerHTML = pageResults.map(p => `
- ${p.title || p.uri} + ${p.title || p.uri}
${p.path || p.uri}
${p.snippet || ''}
@@ -1228,8 +1482,12 @@ async def live_page(): @app.get("/view/{md5_hash}", response_class=HTMLResponse) -async def view_media_page(md5_hash: str): +async def view_media_page(md5_hash: str, noai: bool = Query(False)): """Detail view page for a single media item.""" + 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"), @@ -1250,602 +1508,133 @@ async def view_media_page(md5_hash: str): 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) - # Generate download filename - # Priority: alt_text -> title -> (page_title + consistent index) - name_source = media.get('alt_text') or media.get('title') - if not name_source and sources: - page_title = sources[0].get('page_title', '') - # Use first 4 hex chars of md5 as consistent index (0-65535) - media_idx = int(md5_hash[:4], 16) - name_source = f"{page_title}-{media_idx}" if page_title else f"media-{media_idx}" + # 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" - download_name = slugify(name_source or f"media-{md5_hash[:8]}") - - # Get extension from mime type - ext_map = { - 'image/jpeg': '.jpg', 'image/png': '.png', 'image/gif': '.gif', - 'image/webp': '.webp', 'image/svg+xml': '.svg', 'image/bmp': '.bmp', - 'video/mp4': '.mp4', 'video/webm': '.webm', 'video/quicktime': '.mov', - 'audio/mpeg': '.mp3', 'audio/wav': '.wav', 'audio/ogg': '.ogg', - } - ext = ext_map.get(media.get('mime_type', ''), '.bin') - download_filename = f"{download_name}{ext}" - - # Display title: alt_text -> title -> page_title -> hash + # 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' - if is_video: - media_html = f'' + hero_html = f'' elif is_audio: - media_html = f'' + hero_html = f'' else: - media_html = f'{media.get(' + hero_html = f'{html_module.escape(media.get(' - # Build sources - show all pages that embed this image - # Prioritize more specific pages (longer paths) over generic ones - sorted_sources = sorted(sources, key=lambda s: len(s["page_uri"] or ""), reverse=True) + # Metadata rows + 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 Media URI", f'{neopig_media_uri}'), + ("Source Page URI", f'{page_uri}' if page_uri else '-'), + ("Neopig Page URI", f'{neopig_page_uri}' if neopig_page_uri else '-'), + ("MD5", f'{md5_hash}'), + ("Type", media['media_type']), + ("MIME", media.get('mime_type') or 'unknown'), + ("Size", f"{media.get('file_size') or 0:,} bytes"), + ("Alt", media.get('alt_text') or '-'), + ("Keywords", keywords_html), + ] - sources_html = ''.join([ - f'
  • {s["page_uri"]}
    • ' - for s in sorted_sources - ]) + # Download button + name_source = media.get('alt_text') or media.get('title') + if not name_source 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_name = slugify(name_source or f"media-{md5_hash[:8]}") + 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'} + ext = ext_map.get(media.get('mime_type', ''), '.bin') + download_btn = f'Download ({download_name}{ext})' - # Also show the direct media URL(s) separately - media_urls = list(set(s["media_uri"] for s in sources)) - media_urls_html = ''.join([ - f'
  • {url}
    • ' - for url in media_urls - ]) - - keywords_html = ''.join([f'{k}' for k in keywords]) - - # Get page URI for content lookup - page_uri = sorted_sources[0]["page_uri"] if sorted_sources else None - - # Extract domain for AI system prompt - 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" - - # Get page content - prefer markdown, fallback to raw HTML or text - page_content_html = "" + # 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: - import html as html_module - import re - page_title = page_row.get("title") or "" - - # Render markdown to HTML with our stylesheet - if page_row.get("markdown"): - # Render markdown to HTML - try: - import markdown - md_converter = markdown.Markdown(extensions=['fenced_code', 'tables', 'nl2br']) - rendered = md_converter.convert(page_row["markdown"][:100000]) - - # Hydrate: rewrite image URLs to use our vault - img_pattern = re.compile(r']+src=["\']([^"\']+)["\']', re.IGNORECASE) - img_urls = img_pattern.findall(rendered) - - if img_urls: - from urllib.parse import urljoin - resolved_urls = {} - for url in img_urls: - if url.startswith(('http://', 'https://', '//')): - resolved_urls[url] = url - else: - resolved_urls[url] = urljoin(page_uri, url) - - # Look up md5_hash for resolved URLs - all_urls = list(set(resolved_urls.values())) - resolved_to_hash = await db.lookup_media_by_uris(all_urls) - - # Fallback: for URLs not found, try matching by filename - missing_urls = [u for u in all_urls if u not in resolved_to_hash] - if missing_urls: + 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} + url_to_hash = await db.lookup_media_by_uris(list(set(resolved.values()))) + # Fallback for imgur and other CDNs - lookup by filename/ID + for orig, res in resolved.items(): + if res not in url_to_hash: from pathlib import Path as P - for murl in missing_urls: - fname = P(murl).stem - if len(fname) >= 20: - result = await db.lookup_media_by_filename(fname) - if result: - resolved_to_hash[murl] = result[1] + fname = P(res.split('?')[0]).stem # Remove query params, get 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}"') + 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])}
    " - # Replace original URLs with vault paths - for orig_url, resolved_url in resolved_urls.items(): - if resolved_url in resolved_to_hash: - md5 = resolved_to_hash[resolved_url] - rendered = rendered.replace(f'src="{orig_url}"', f'src="/media/{md5}"') - rendered = rendered.replace(f"src='{orig_url}'", f'src="/media/{md5}"') + # 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] - # Look for screenshot of this page - screenshot_html = "" - screenshot_hash = await db.get_page_screenshot(page_uri, exclude_hash=md5_hash) - if screenshot_hash: - screenshot_html = f''' -
    - - Page screenshot - -
    ''' + return render_detail_page( + 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, + ) - if screenshot_html: - page_content_html = f''' -
    -

    Page Context{(' - ' + html_module.escape(page_title)) if page_title else ''}

    -
    -
    {rendered}
    - {screenshot_html} -
    -
    -''' - else: - page_content_html = f''' -
    -

    Page Context{(' - ' + html_module.escape(page_title)) if page_title else ''}

    -
    {rendered}
    -
    -''' - except ImportError: - escaped = html_module.escape(page_row["markdown"][:50000]) - page_content_html = f''' -
    -

    Page Context{(' - ' + html_module.escape(page_title)) if page_title else ''}

    - 
    {escaped}
    -
    -''' - elif page_row.get("content"): - content = page_row["content"][:50000] - escaped = html_module.escape(content) - paragraphs = escaped.split('\n\n') - formatted = ''.join(f'

    {p.replace(chr(10), "
    ")}

    ' for p in paragraphs if p.strip()) - page_content_html = f''' -
    -

    Page Context{(' - ' + html_module.escape(page_title)) if page_title else ''}

    -
    {formatted}
    -
    -''' - return f""" - - - - {display_title} - neopig - - - - - -
    - 🐷 neopig - Search - Live - Random - Crawl - Phantom -
    -
    -
    - - - -
    -
    -
    -
    - {media_html} -
    -

    {display_title}

    -
    -
    -
    -
    MD5 Hash:{md5_hash}
    -
    Type:{media['media_type']}
    -
    MIME:{media.get('mime_type') or 'unknown'}
    -
    Size:{media.get('file_size') or 0:,} bytes
    -
    First seen:{media.get('first_seen_at')}
    -
    Alt text:{media.get('alt_text') or '-'}
    -
    Title:{media.get('title') or '-'}
    -
    Keywords:{keywords_html or '-'}
    -
    - - ⬇ Download ({download_filename}) - -
    -

    Source pages ({len(sources)})

    -
      {sources_html}
    -
    -
    -
    -{page_content_html} -
    - - - - - -""" +@app.get("/page/{uri_hash}", response_class=HTMLResponse) +async def view_page_by_hash( + uri_hash: str, + noai: bool = Query(False, description="Disable AI assistant") +): + """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") + # Redirect to the URI-based view (reuses same logic) + return await view_page(uri=page['uri'], noai=noai) @app.get("/page/view", response_class=HTMLResponse) @@ -1856,6 +1645,7 @@ async def view_page( """View an archived page with markdown and screenshot.""" import html as html_module import re + from urllib.parse import urljoin, quote source_domain = Uri(uri).hostname @@ -1873,157 +1663,65 @@ async def view_page( md_converter = markdown.Markdown(extensions=['fenced_code', 'tables', 'nl2br']) content_html = md_converter.convert(page["markdown"][:100000]) - # Hydrate images from vault - img_pattern = re.compile(r']+src=["\']([^"\']+)["\']', re.IGNORECASE) - img_urls = img_pattern.findall(content_html) - if img_urls: - from urllib.parse import urljoin - resolved_urls = {} - for url in img_urls: - if url.startswith(('http://', 'https://', '//')): - resolved_urls[url] = url - else: - resolved_urls[url] = urljoin(uri, url) - - # Look up md5_hash for resolved URLs - all_urls = list(set(resolved_urls.values())) - resolved_to_hash = await db.lookup_media_by_uris(all_urls) - - # Fallback: for URLs not found, try matching by filename - missing_urls = [u for u in all_urls if u not in resolved_to_hash] - if missing_urls: + # 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: + resolved = {u: u if u.startswith(('http://', 'https://', '//')) else urljoin(uri, u) for u in all_urls} + url_to_hash = await db.lookup_media_by_uris(list(set(resolved.values()))) + # Fallback for imgur and other CDNs - lookup by filename/ID + for murl in [u for u in resolved.values() if u not in url_to_hash]: from pathlib import Path as P - for murl in missing_urls: - fname = P(murl).stem - if len(fname) >= 20: - result = await db.lookup_media_by_filename(fname) - if result: - resolved_to_hash[murl] = result[1] - - # Replace original URLs with vault paths - for orig_url, resolved_url in resolved_urls.items(): - if resolved_url in resolved_to_hash: - md5 = resolved_to_hash[resolved_url] - content_html = content_html.replace(f'src="{orig_url}"', f'src="/media/{md5}"') - content_html = content_html.replace(f"src='{orig_url}'", f'src="/media/{md5}"') + fname = P(murl.split('?')[0]).stem # Remove query params, get stem + if fname and len(fname) >= 5: + result = await db.lookup_media_by_filename(fname) + if result: + url_to_hash[murl] = result[1] + 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}"') + # Rewrite a href to neopig media view page + 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.get('markdown', '')[:50000])}
    " elif page.get("content"): escaped = html_module.escape(page["content"][:50000]) content_html = f"
    {escaped}
    " - # Find all screenshots (may be chunked for tall pages) + # Get screenshots and media screenshot_hashes = await db.get_page_screenshots(uri) - screenshot_html = "" - if screenshot_hashes: - imgs = ''.join(f'Page screenshot chunk' for h in screenshot_hashes) - screenshot_html = f'
    {imgs}
    ' - - # Find media from this page media_items = await db.get_page_media(uri) - media_grid = "" - if media_items: - media_cards = [] - for m in media_items: - is_video = m.get("media_type") == "video" - if is_video: - el = f'' - else: - el = f'' - media_cards.append(f''' - - {el} - ''') - media_grid = f''' -
    -

    Media from this page ({len(media_items)})

    -
    {''.join(media_cards)}
    -
    ''' - - has_screenshot = bool(screenshot_html) - return f""" - - - {html_module.escape(page_title)} - neopig - - - - - -
    - 🐷 neopig - Search - Live - Random -
    -
    -

    {html_module.escape(page_title)}

    -
    -
    URL:{uri}
    -
    Media:{len(media_items)} items
    -
    -
    -
    - {media_grid} -

    Page Content

    -
    {content_html or '

    No content available

    '}
    -
    - {f'''
    - ◀ Screenshot - {screenshot_html} -
    ''' if has_screenshot else ''} -
    -
    - - {'' if noai else f''' - '''} - -""" - + # Metadata rows for page view + 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", page.get('description') or '-'), + ("Keywords", keywords_html), + ("Media", f"{len(media_items)} items"), + ] + return render_detail_page( + title=page_title, + hero_html="", # No hero media for page view + meta_rows=meta_rows, + media_items=media_items, + content_html=content_html, + screenshot_hashes=screenshot_hashes, + source_domain=source_domain, + page_uri=uri, + noai=noai, + ) @app.get("/phantom/export") async def phantom_export(domain: str = Query(None, description="Filter by domain")): """ @@ -2327,10 +2025,10 @@ async def random_item(type: str = Query(None, description="Type: media or page ( async with db.session() as session: if type == "page": # Get a random page - stmt = select(Page.uri).order_by(func.random()).limit(1) + stmt = select(Page.uri_hash).order_by(func.random()).limit(1) result = await session.execute(stmt) row = result.fetchone() - if row: + if row and row[0]: return RedirectResponse(url=f"/page/{row[0]}", status_code=302) else: # Get a random media item (excluding screenshots)