diff --git a/lib/acquisition/__init__.py b/lib/acquisition/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lib/dispatcher.py b/lib/dispatcher.py new file mode 100644 index 0000000..0abb314 --- /dev/null +++ b/lib/dispatcher.py @@ -0,0 +1,121 @@ +""" +RECON Dispatcher + +Watches configured acquired// 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: + .txt (or other content extension) + .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 diff --git a/lib/embedder.py b/lib/embedder.py index b4ed162..35fcb58 100644 --- a/lib/embedder.py +++ b/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): diff --git a/lib/enricher.py b/lib/enricher.py index fe18d06..d9540aa 100644 --- a/lib/enricher.py +++ b/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'] diff --git a/lib/processors/__init__.py b/lib/processors/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lib/processors/transcript_processor.py b/lib/processors/transcript_processor.py new file mode 100644 index 0000000..76998f7 --- /dev/null +++ b/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 diff --git a/lib/utils.py b/lib/utils.py index a98359c..1aec793 100644 --- a/lib/utils.py +++ b/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)