Russell Ballestrini
7883ae0eba
Phase 1 of module restructuring as outlined in docs/REFACTOR.md: New neopig/ package modules: - live.py: Live media queue (get_live_queue, emit_live_media) - state.py: AppendOnlyStateLog, state file helpers - html_utils.py: trim_html_wrapper, extract_meta_from_html - logging.py: TqdmLoggingHandler, job logging functions - backfill/: markdown and screenshot backfill operations Package features: - Lazy import of NeoPig/main from neopig.py via __getattr__ - Full backwards compatibility with existing imports - 42 new unit tests for extracted modules Total: 458 tests passing
300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""
|
|
Markdown backfill operations for neopig.
|
|
|
|
Re-processes stored HTML to regenerate markdown with smart structure detection.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import multiprocessing
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
|
|
from datetime import datetime, timezone
|
|
|
|
from tqdm import tqdm
|
|
|
|
from ..html_utils import trim_html_wrapper, extract_meta_from_html
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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
|
|
|
|
page_id, uri, raw_html, trim_wrapper_flag = args
|
|
try:
|
|
description, keywords = extract_meta_from_html(raw_html)
|
|
raw = raw_html
|
|
if trim_wrapper_flag:
|
|
raw = trim_html_wrapper(raw)
|
|
new_markdown = html_to_markdown(raw, base_url=uri)[:200000]
|
|
return (page_id, new_markdown, description, json.dumps(keywords) if keywords else None, None)
|
|
except Exception as e:
|
|
return (page_id, None, None, None, str(e))
|
|
|
|
|
|
def _process_chunk(args: tuple) -> dict:
|
|
"""Process a chunk of pages in a subprocess with thread workers.
|
|
|
|
Each process gets a chunk and spawns threads to burn through it.
|
|
Each thread has its own DB connection - WAL mode allows concurrent writes.
|
|
Background flusher thread syncs progress every 100ms (workers never block).
|
|
"""
|
|
chunk, db_path, trim_wrapper_flag, threads_per_process, job_id = args
|
|
|
|
# Thread-local storage for DB connections
|
|
thread_local = threading.local()
|
|
progress = {'updated': 0, 'errors': 0}
|
|
progress_lock = threading.Lock()
|
|
stop_flusher = threading.Event()
|
|
|
|
def get_conn():
|
|
"""Get thread-local DB connection for page writes."""
|
|
if not hasattr(thread_local, 'conn'):
|
|
thread_local.conn = sqlite3.connect(db_path, timeout=60.0)
|
|
thread_local.conn.execute("PRAGMA journal_mode=WAL")
|
|
thread_local.conn.execute("PRAGMA synchronous=NORMAL")
|
|
thread_local.conn.execute("PRAGMA busy_timeout=60000")
|
|
return thread_local.conn
|
|
|
|
def flusher_thread():
|
|
"""Background thread that syncs progress to DB every 100ms."""
|
|
conn = sqlite3.connect(db_path, timeout=60.0)
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA busy_timeout=60000")
|
|
flushed_updated, flushed_errors = 0, 0
|
|
|
|
while not stop_flusher.is_set():
|
|
time.sleep(0.05)
|
|
with progress_lock:
|
|
curr_updated, curr_errors = progress['updated'], progress['errors']
|
|
|
|
delta_updated = curr_updated - flushed_updated
|
|
delta_errors = curr_errors - flushed_errors
|
|
|
|
if delta_updated > 0 or delta_errors > 0:
|
|
conn.execute(
|
|
"UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
|
|
(delta_updated, delta_errors, job_id)
|
|
)
|
|
conn.commit()
|
|
flushed_updated, flushed_errors = curr_updated, curr_errors
|
|
|
|
# Final flush
|
|
with progress_lock:
|
|
final_updated, final_errors = progress['updated'], progress['errors']
|
|
delta_updated = final_updated - flushed_updated
|
|
delta_errors = final_errors - flushed_errors
|
|
if delta_updated > 0 or delta_errors > 0:
|
|
conn.execute(
|
|
"UPDATE backfill_jobs SET processed_records = processed_records + ?, error_count = error_count + ? WHERE id = ?",
|
|
(delta_updated, delta_errors, job_id)
|
|
)
|
|
conn.commit()
|
|
|
|
progress['_flushed_updated'] = final_updated
|
|
progress['_flushed_errors'] = final_errors
|
|
conn.close()
|
|
|
|
def process_and_write(page_data):
|
|
"""Process one page and write immediately with thread-local connection."""
|
|
page_id, uri, raw_html = page_data
|
|
result = _process_single_page((page_id, uri, raw_html, trim_wrapper_flag))
|
|
page_id, markdown, description, keywords_json, error = result
|
|
|
|
conn = get_conn()
|
|
if error:
|
|
with progress_lock:
|
|
progress['errors'] += 1
|
|
else:
|
|
conn.execute(
|
|
"UPDATE pages SET markdown = ?, description = ?, keywords = ? WHERE id = ?",
|
|
(markdown, description, keywords_json, page_id)
|
|
)
|
|
conn.commit()
|
|
with progress_lock:
|
|
progress['updated'] += 1
|
|
|
|
# Start background flusher
|
|
flusher = threading.Thread(target=flusher_thread, daemon=True)
|
|
flusher.start()
|
|
|
|
# Fan out to thread workers
|
|
with ThreadPoolExecutor(max_workers=threads_per_process) as thread_executor:
|
|
list(thread_executor.map(process_and_write, chunk))
|
|
|
|
# Stop flusher and wait for final flush
|
|
stop_flusher.set()
|
|
flusher.join(timeout=5.0)
|
|
|
|
return {
|
|
'updated': progress.get('_flushed_updated', progress['updated']),
|
|
'errors': progress.get('_flushed_errors', progress['errors']),
|
|
'total': len(chunk)
|
|
}
|
|
|
|
|
|
async def backfill_markdown(
|
|
db_path: str,
|
|
domain_filter: str = None,
|
|
trim_wrapper: bool = False,
|
|
quiet: bool = False,
|
|
fresh: bool = False
|
|
):
|
|
"""Re-process stored HTML to regenerate markdown with smart structure detection.
|
|
|
|
ETL-style parallel processing:
|
|
- Divides work into chunks (1 per CPU core)
|
|
- Each process spawns 6 threads to burn through its chunk
|
|
- Immediate DB writes with WAL mode (thread-safe)
|
|
- Progress tracked via backfill_jobs table (atomic increments)
|
|
|
|
Args:
|
|
db_path: Path to SQLite database
|
|
domain_filter: Only process pages matching this domain (e.g., 'example.com')
|
|
trim_wrapper: Strip nav/header/footer/logo before conversion
|
|
quiet: Disable progress bar
|
|
fresh: Re-process all pages, even those that already have markdown
|
|
"""
|
|
import aiosqlite
|
|
|
|
num_cpus = multiprocessing.cpu_count()
|
|
threads_per_process = 6
|
|
logger.info(f"ETL mode: {num_cpus} processes x {threads_per_process} threads = {num_cpus * threads_per_process} workers")
|
|
|
|
if domain_filter:
|
|
pattern = f"%://{domain_filter}%" if '://' not in domain_filter else f"%{domain_filter}%"
|
|
logger.info(f"Backfilling markdown for domain: {domain_filter}")
|
|
else:
|
|
pattern = None
|
|
logger.info("Backfilling markdown for ALL pages")
|
|
|
|
if trim_wrapper:
|
|
logger.info("Trim wrapper enabled")
|
|
|
|
# Setup: ensure tables/columns exist, create job record, get pages
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await db.execute("PRAGMA journal_mode=WAL")
|
|
|
|
# Ensure backfill_jobs table exists
|
|
await db.execute("""
|
|
CREATE TABLE IF NOT EXISTS backfill_jobs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
job_type TEXT NOT NULL,
|
|
domain_filter TEXT,
|
|
status TEXT DEFAULT 'running',
|
|
total_records INTEGER DEFAULT 0,
|
|
processed_records INTEGER DEFAULT 0,
|
|
error_count INTEGER DEFAULT 0,
|
|
started_at TEXT NOT NULL,
|
|
completed_at TEXT
|
|
)
|
|
""")
|
|
|
|
# Ensure page columns exist
|
|
for col in ['description', 'keywords']:
|
|
try:
|
|
await db.execute(f"ALTER TABLE pages ADD COLUMN {col} TEXT")
|
|
except Exception:
|
|
pass
|
|
|
|
# Get pages to process
|
|
base_condition = "raw_html IS NOT NULL"
|
|
if not fresh:
|
|
base_condition += " AND (markdown IS NULL OR markdown = '')"
|
|
logger.info("Incremental mode: only processing pages without markdown")
|
|
else:
|
|
logger.info("Fresh mode: re-processing all pages")
|
|
|
|
if pattern:
|
|
cursor = await db.execute(
|
|
f"SELECT id, uri, raw_html FROM pages WHERE {base_condition} AND uri LIKE ?",
|
|
(pattern,)
|
|
)
|
|
else:
|
|
cursor = await db.execute(
|
|
f"SELECT id, uri, raw_html FROM pages WHERE {base_condition}"
|
|
)
|
|
rows = await cursor.fetchall()
|
|
total = len(rows)
|
|
logger.info(f"Found {total} pages to process")
|
|
|
|
# Create job record
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
cursor = await db.execute(
|
|
"INSERT INTO backfill_jobs (job_type, domain_filter, status, total_records, processed_records, error_count, started_at) VALUES (?, ?, 'running', ?, 0, 0, ?)",
|
|
('markdown', domain_filter, total, now)
|
|
)
|
|
job_id = cursor.lastrowid
|
|
await db.commit()
|
|
logger.info(f"Created backfill job {job_id}")
|
|
|
|
# Convert to list of tuples
|
|
all_pages = [(row[0], row[1], row[2]) for row in rows if row[2]]
|
|
|
|
# Divide into exactly num_cpus chunks
|
|
chunks = []
|
|
chunk_size = len(all_pages) // num_cpus if num_cpus > 0 else len(all_pages)
|
|
for i in range(num_cpus):
|
|
start = i * chunk_size
|
|
end = (i + 1) * chunk_size if i < num_cpus - 1 else len(all_pages)
|
|
chunks.append(all_pages[start:end])
|
|
logger.info(f"Split into {len(chunks)} chunks of ~{chunk_size} pages each")
|
|
|
|
# Process chunks in parallel, poll job record for progress
|
|
with ProcessPoolExecutor(max_workers=num_cpus) as executor:
|
|
chunk_args = [(chunk, db_path, trim_wrapper, threads_per_process, job_id) for chunk in chunks]
|
|
futures = [executor.submit(_process_chunk, args) for args in chunk_args]
|
|
|
|
# Poll backfill_jobs table for progress
|
|
with tqdm(total=len(all_pages), desc="Pages", unit="page", disable=quiet, smoothing=0.1) as pbar:
|
|
last_count = 0
|
|
while True:
|
|
conn = sqlite3.connect(db_path, timeout=60.0)
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
cursor = conn.execute(
|
|
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
|
|
(job_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
current = row[0] if row and row[0] is not None else 0
|
|
conn.close()
|
|
|
|
if current > last_count:
|
|
pbar.update(current - last_count)
|
|
last_count = current
|
|
if all(f.done() for f in futures):
|
|
conn = sqlite3.connect(db_path, timeout=60.0)
|
|
cursor = conn.execute(
|
|
"SELECT processed_records + error_count FROM backfill_jobs WHERE id = ?",
|
|
(job_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
current = row[0] if row and row[0] is not None else 0
|
|
conn.close()
|
|
if current > last_count:
|
|
pbar.update(current - last_count)
|
|
break
|
|
time.sleep(0.1)
|
|
|
|
# Collect results
|
|
total_updated = 0
|
|
total_errors = 0
|
|
for f in futures:
|
|
result = f.result()
|
|
total_updated += result['updated']
|
|
total_errors += result['errors']
|
|
|
|
# Mark job completed
|
|
async with aiosqlite.connect(db_path) as db:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
await db.execute(
|
|
"UPDATE backfill_jobs SET status = 'completed', completed_at = ? WHERE id = ?",
|
|
(now, job_id)
|
|
)
|
|
await db.commit()
|
|
|
|
logger.info(f"Backfill complete: {total_updated} updated, {total_errors} errors")
|