Phase 3: dispatcher, transcript processor, text_dir resolution

- lib/dispatcher.py: one-shot dispatcher that scans acquired/<type>/
  for content+sidecar pairs and routes to registered processors
- lib/processors/transcript_processor.py: pre_flight() for transcripts
  (hash, dedupe, split into pages, register in DB, set text_dir)
- lib/utils.py: resolve_text_dir() helper for text_dir column fallback
- lib/enricher.py: use resolve_text_dir() instead of hardcoded path
- lib/embedder.py: use resolve_text_dir() instead of hardcoded path
- lib/processors/__init__.py, lib/acquisition/__init__.py: package inits

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

lib/dispatcher.py

@@ -0,0 +1,121 @@
+"""
+RECON Dispatcher
+
+Watches configured acquired/<subfolder>/ directories for content+sidecar pairs
+that have been stable (mtime unchanged) for the configured threshold, then
+hands them to the appropriate processor's pre_flight().
+
+Phase 3: importable one-shot dispatcher. Service-loop integration in Phase 5.
+"""
+import importlib
+import logging
+import os
+import time
+
+from .utils import get_config
+from .status import StatusDB
+
+logger = logging.getLogger("recon.dispatcher")
+
+
+def _load_processor(processor_name):
+    """Dynamically import a processor module from lib.processors."""
+    module_path = f"lib.processors.{processor_name}"
+    try:
+        return importlib.import_module(module_path)
+    except ImportError as e:
+        logger.error("Cannot load processor %s: %s", processor_name, e)
+        return None
+
+
+def _find_pairs(subfolder_path):
+    """Find content+sidecar pairs in a subfolder.
+
+    A pair is two files sharing a basename:
+      <basename>.txt  (or other content extension)
+      <basename>.meta.json  (sidecar)
+
+    Returns list of (content_path, meta_path, basename) tuples.
+    """
+    if not os.path.isdir(subfolder_path):
+        return []
+
+    files = set(os.listdir(subfolder_path))
+    pairs = []
+
+    for fname in sorted(files):
+        if fname.endswith('.meta.json'):
+            basename = fname[:-len('.meta.json')]
+            # Look for matching content file (try common extensions)
+            for ext in ['.txt', '.vtt', '.html', '.pdf']:
+                content_name = basename + ext
+                if content_name in files:
+                    pairs.append((
+                        os.path.join(subfolder_path, content_name),
+                        os.path.join(subfolder_path, fname),
+                        basename,
+                    ))
+                    break
+
+    return pairs
+
+
+def _is_stable(filepath, stability_seconds):
+    """Check if a file's mtime is older than stability_seconds ago."""
+    try:
+        mtime = os.path.getmtime(filepath)
+        return (time.time() - mtime) >= stability_seconds
+    except OSError:
+        return False
+
+
+def dispatch_once():
+    """One-shot dispatch: scan all configured acquired/ subfolders once.
+
+    Returns list of result dicts from processor pre_flight calls.
+    """
+    config = get_config()
+    pipeline_cfg = config.get('pipeline', {})
+    acquired_root = pipeline_cfg.get('acquired_root', '/opt/recon/data/acquired')
+    dispatch_map = pipeline_cfg.get('dispatch', {})
+    stability_seconds = pipeline_cfg.get('mtime_stability_seconds', 10)
+
+    db = StatusDB(config['paths']['db'])
+    results = []
+
+    for subfolder_name, processor_name in dispatch_map.items():
+        subfolder_path = os.path.join(acquired_root, subfolder_name)
+
+        processor = _load_processor(processor_name)
+        if processor is None:
+            continue
+
+        if not hasattr(processor, 'pre_flight'):
+            logger.error("Processor %s has no pre_flight function", processor_name)
+            continue
+
+        pairs = _find_pairs(subfolder_path)
+        if not pairs:
+            continue
+
+        for content_path, meta_path, basename in pairs:
+            # Both files must be stable
+            if not (_is_stable(content_path, stability_seconds) and
+                    _is_stable(meta_path, stability_seconds)):
+                logger.debug("Pair %s not yet stable, skipping", basename)
+                continue
+
+            logger.info("Dispatching %s/%s to %s", subfolder_name, basename, processor_name)
+            try:
+                result = processor.pre_flight(content_path, meta_path, db, config)
+                results.append(result)
+                logger.info("Result for %s: %s", basename, result.get('action', 'unknown'))
+            except Exception as e:
+                logger.error("Processor %s crashed on %s: %s", processor_name, basename, e)
+                results.append({
+                    'action': 'error',
+                    'error': str(e),
+                    'basename': basename,
+                })
+
+    return results

lib/embedder.py

@@ -21,6 +21,7 @@ from qdrant_client.models import PointStruct, SparseVector
 
 from .utils import get_config, concept_id, generate_download_url, setup_logging
 from .status import StatusDB
+from .utils import resolve_text_dir
 
 logger = setup_logging('recon.embedder')
 
@@ -274,7 +275,7 @@ def embed_single(file_hash, db, config):
         source_type = 'web' if is_web else 'document'
 
         # Check meta.json for explicit source_type (e.g. 'transcript')
-        text_dir = os.path.join(config['paths']['text'], file_hash)
+        text_dir = resolve_text_dir(file_hash, config, db)
         meta_path = os.path.join(text_dir, 'meta.json')
         page_timestamps = {}
         if os.path.exists(meta_path):

lib/enricher.py

@@ -25,6 +25,7 @@ import google.generativeai as genai
 
 from .utils import get_config, setup_logging
 from .status import StatusDB
+from .utils import resolve_text_dir
 
 logger = setup_logging('recon.enricher')
 
@@ -345,7 +346,7 @@ def enrich_single(file_hash, db, config, key_rotator):
     if not doc:
         return False
 
-    text_dir = os.path.join(config['paths']['text'], file_hash)
+    text_dir = resolve_text_dir(file_hash, config, db)
     concepts_dir = os.path.join(config['paths']['concepts'], file_hash)
     window_size = config['processing']['enrich_window_size']
     delay = config['processing']['rate_limit_delay']

lib/processors/transcript_processor.py

@@ -0,0 +1,152 @@
+"""
+RECON Transcript Processor
+
+Handles pre_flight for transcript content arriving in acquired/stream/.
+Reads a raw text file + meta.json sidecar, hashes, dedupes, splits into
+page_NNNN.txt files, and registers in the database.
+
+Phase 3: first processor implementation.
+"""
+import hashlib
+import json
+import logging
+import os
+import shutil
+
+from lib.web_scraper import chunk_text
+from lib.utils import content_hash
+
+logger = logging.getLogger("recon.processors.transcript")
+
+# Words per page for transcript chunking (matches existing pipeline)
+WORDS_PER_PAGE = 2000
+
+
+def pre_flight(content_path, meta_path, db, config):
+    """Process a transcript pair from acquired/stream/.
+
+    Args:
+        content_path: Path to the raw transcript .txt file
+        meta_path: Path to the .meta.json sidecar
+        db: StatusDB instance
+        config: RECON config dict
+
+    Returns:
+        dict with keys: hash, action, source_path, error
+        Actions: 'extracted', 'duplicate', 'error'
+    """
+    result = {
+        'hash': None,
+        'action': 'error',
+        'source_path': content_path,
+        'error': None,
+    }
+
+    # Read and hash the content file
+    try:
+        file_hash = content_hash(content_path)
+        result['hash'] = file_hash
+    except Exception as e:
+        result['error'] = f"Cannot hash content file: {e}"
+        return result
+
+    # Hash dedupe: if hash exists in catalogue, delete the pair and return
+    conn = db._get_conn()
+    existing = conn.execute(
+        "SELECT hash FROM catalogue WHERE hash = ?", (file_hash,)
+    ).fetchone()
+    if existing:
+        logger.info("Duplicate hash %s, removing pair", file_hash[:8])
+        try:
+            os.remove(content_path)
+            os.remove(meta_path)
+        except OSError as e:
+            logger.warning("Failed to remove duplicate pair: %s", e)
+        result['action'] = 'duplicate'
+        return result
+
+    # Read meta.json sidecar
+    try:
+        with open(meta_path, encoding='utf-8') as f:
+            meta = json.load(f)
+    except Exception as e:
+        result['error'] = f"Cannot read meta.json: {e}"
+        return result
+
+    # Read raw transcript text
+    try:
+        with open(content_path, encoding='utf-8') as f:
+            raw_text = f.read()
+    except Exception as e:
+        result['error'] = f"Cannot read content file: {e}"
+        return result
+
+    if not raw_text.strip():
+        result['error'] = "Empty transcript"
+        return result
+
+    # Set up processing directory
+    processing_root = config.get('pipeline', {}).get(
+        'processing_root', '/opt/recon/data/processing'
+    )
+    proc_dir = os.path.join(processing_root, file_hash)
+    try:
+        os.makedirs(proc_dir, exist_ok=True)
+    except Exception as e:
+        result['error'] = f"Cannot create processing dir: {e}"
+        return result
+
+    # Move the pair to processing/{hash}/
+    try:
+        shutil.move(content_path, os.path.join(proc_dir, 'transcript.txt'))
+        shutil.move(meta_path, os.path.join(proc_dir, 'meta.json'))
+    except Exception as e:
+        result['error'] = f"Cannot move pair to processing: {e}"
+        return result
+
+    # Split raw text into page_NNNN.txt files
+    pages = chunk_text(raw_text, WORDS_PER_PAGE)
+    for i, page_text in enumerate(pages, start=1):
+        page_path = os.path.join(proc_dir, f"page_{i:04d}.txt")
+        with open(page_path, 'w', encoding='utf-8') as f:
+            f.write(page_text)
+
+    # Update meta.json with processing metadata
+    meta['hash'] = file_hash
+    meta['source_type'] = meta.get('source_type', 'transcript')
+    meta['page_count'] = len(pages)
+    meta['text_length'] = len(raw_text)
+    meta_out_path = os.path.join(proc_dir, 'meta.json')
+    with open(meta_out_path, 'w', encoding='utf-8') as f:
+        json.dump(meta, f, indent=2)
+
+    # Register in catalogue
+    title = meta.get('title', os.path.basename(content_path))
+    source_url = meta.get('source_url', meta.get('url', ''))
+    source = 'stream.echo6.co'
+    category = meta.get('category', 'Transcript')
+    size_bytes = os.path.getsize(os.path.join(proc_dir, 'transcript.txt'))
+
+    db.add_to_catalogue(file_hash, title, source_url, size_bytes, source, category)
+
+    # Queue and advance to extracted
+    db.queue_document(file_hash)
+
+    # Set text_dir on the documents row
+    conn = db._get_conn()
+    conn.execute(
+        "UPDATE documents SET text_dir = ? WHERE hash = ?",
+        (proc_dir, file_hash)
+    )
+    conn.commit()
+
+    # Update status to extracted with page count
+    db.update_status(file_hash, 'extracted', pages_extracted=len(pages))
+
+    logger.info(
+        "Transcript pre_flight complete: %s (%s) -> %d pages in %s",
+        file_hash[:8], title, len(pages), proc_dir,
+    )
+
+    result['action'] = 'extracted'
+    return result

lib/utils.py

@@ -388,3 +388,19 @@ def generate_download_url(filepath, library_root='/mnt/library', base_url='https
     parts = rel.split(os.sep)
     encoded = '/'.join(quote(p) for p in parts)
     return f"{base_url}/{encoded}"
+
+
+def resolve_text_dir(file_hash, config, db=None):
+    """Resolve the text directory for a document.
+
+    If db is provided and documents.text_dir is set for this hash, use that.
+    Otherwise fall back to the legacy location: config['paths']['text']/{hash}/
+    """
+    if db is not None:
+        conn = db._get_conn()
+        row = conn.execute(
+            "SELECT text_dir FROM documents WHERE hash = ?", (file_hash,)
+        ).fetchone()
+        if row and row['text_dir']:
+            return row['text_dir']
+    return os.path.join(config['paths']['text'], file_hash)