From 943baa3c0237657445c0defa61bfc3240b58caff Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Fri, 10 Apr 2026 06:57:54 -0400 Subject: [PATCH] Add ingest command for local file and dump import New 'mosaic ingest' CLI command bypasses HTTP collection to import raw documents (PDF, HTML, text) or pre-extracted JSONL data directly into the pipeline. Modes: - Single/multiple files: -f file1.pdf -f file2.html -s source_name - Directory bulk import: -d /path/to/dump/ -s source_name -r - JSONL extracted data: -f data.jsonl -s name -c tools - Real-time watch: -d /incoming/ -s name -w (polls every 2s) - Auto-parse: --parse flag runs parser on newly ingested docs Files are hashed, deduplicated, and stored in the cache using the same hash-sharded scheme as HTTP-collected documents. DevTrack: mosaic #498 --- mosaic.py | 260 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 257 insertions(+), 3 deletions(-) diff --git a/mosaic.py b/mosaic.py index bb45409..027ef49 100755 --- a/mosaic.py +++ b/mosaic.py @@ -654,6 +654,259 @@ def export_cmd(ctx, category, fmt): db.close() +# --------------------------------------------------------------------------- +# ingest +# --------------------------------------------------------------------------- + +MIME_TO_DOCTYPE = { + 'application/pdf': 'pdf', + 'text/html': 'html', + 'text/plain': 'text', + 'message/rfc822': 'email', +} + + +def _detect_doc_type(file_path: str) -> str: + """Detect document type from file content.""" + from utils.sanitize import validate_content_type + mime = validate_content_type(file_path) + return MIME_TO_DOCTYPE.get(mime, 'text') + + +def _hash_file(file_path: str) -> str: + """Compute SHA256 of a file.""" + sha = hashlib.sha256() + with open(file_path, 'rb') as f: + for chunk in iter(lambda: f.read(8192), b''): + sha.update(chunk) + return sha.hexdigest() + + +def _ingest_raw_file(db, cache, file_path: str, source_name: str) -> bool: + """Ingest a single raw document file into the DB and cache. + + Returns True if the document was new, False if already exists. + """ + from storage.models import Document + + fpath = Path(file_path) + content_hash = _hash_file(str(fpath)) + + existing = db.get_document(content_hash) + if existing: + return False + + doc_type = _detect_doc_type(str(fpath)) + ext = fpath.suffix.lstrip('.') + + # Store in cache + cached_path = cache.get_shard_path(content_hash, source_name, ext=ext) + cached_path.parent.mkdir(parents=True, exist_ok=True) + import shutil + shutil.copy2(str(fpath), str(cached_path)) + + doc = Document( + id=content_hash, + source=source_name, + source_url=str(fpath.resolve()), + doc_type=doc_type, + title=fpath.stem, + raw_path=str(cached_path), + content_hash=content_hash, + fetch_date=datetime.now().isoformat(), + char_count=fpath.stat().st_size, + status='CACHED', + parse_status='PENDING', + ) + db.insert_document(doc) + return True + + +@cli.command() +@click.option("--file", "-f", "files", multiple=True, type=click.Path(exists=True), + help="File(s) to ingest. Repeat for multiple files.") +@click.option("--dir", "-d", "directory", type=click.Path(exists=True, file_okay=False), + help="Directory of files to ingest.") +@click.option("--source", "-s", type=str, required=True, help="Source name to tag ingested docs.") +@click.option("--category", "-c", type=click.Choice(EXPORT_CATEGORIES, case_sensitive=False), + default=None, help="JSONL category (for importing pre-extracted data).") +@click.option("--watch", "-w", is_flag=True, default=False, + help="Watch directory for new files (real-time mode).") +@click.option("--recursive", "-r", is_flag=True, default=False, + help="Recurse into subdirectories.") +@click.option("--parse", "auto_parse", is_flag=True, default=False, + help="Automatically parse ingested documents.") +@click.pass_context +def ingest(ctx, files, directory, source, category, watch, recursive, auto_parse): + """Ingest local files or data dumps into the Mosaic pipeline. + + Import raw documents (PDF, HTML, text) or pre-extracted JSONL data + without HTTP collection. Supports single files, directories, and + real-time directory watching. + + \b + Examples: + mosaic ingest -f dump.pdf -s wikileaks_dump + mosaic ingest -d /path/to/cables/ -s cablegate -r --parse + mosaic ingest -f extracted.jsonl -s vault7 -c tools + mosaic ingest -d /incoming/ -s live_feed -w + """ + db_path = ctx.obj["db_path"] + output_dir = ctx.obj["output_dir"] + db = open_db(db_path) + + try: + # JSONL import mode + if category: + store = open_jsonl_store(db, output_dir) + total = 0 + targets = list(files) + if directory: + p = Path(directory) + targets.extend(str(f) for f in p.glob("*.jsonl")) + + if not targets: + console.print("[yellow]No JSONL files to import.[/yellow]") + return + + for fpath in targets: + with console.status(f"[bold green]Importing {Path(fpath).name}..."): + count = store.import_jsonl(fpath, category) + console.print(f" [green]+{count}[/green] {category} from {Path(fpath).name}") + total += count + + console.print(f"\n[bold green]Imported {total} {category} records total.[/bold green]") + return + + # Raw document ingest mode + cache = open_cache(output_dir) + file_list = list(files) + + if directory: + p = Path(directory) + pattern = "**/*" if recursive else "*" + file_list.extend( + str(f) for f in p.glob(pattern) + if f.is_file() and not f.name.startswith('.') + ) + + if watch and directory: + _watch_directory(db, cache, directory, source, recursive, auto_parse, ctx) + return + + if not file_list: + console.print("[yellow]No files to ingest.[/yellow] Use --file, --dir, or --watch.") + return + + console.print(Panel( + f"[bold green]Ingest[/bold green]\n\n" + f" Source : {source}\n" + f" Files : {len(file_list)}\n" + f" Parse : {'yes' if auto_parse else 'no'}", + title="[bold]mosaic ingest[/bold]", + border_style="green", + )) + + new_count = 0 + skip_count = 0 + + with Progress( + SpinnerColumn(), TextColumn("{task.description}"), + BarColumn(), TextColumn("{task.completed}/{task.total}"), + ) as progress: + task = progress.add_task("Ingesting files", total=len(file_list)) + for fpath in file_list: + is_new = _ingest_raw_file(db, cache, fpath, source) + if is_new: + new_count += 1 + else: + skip_count += 1 + progress.advance(task) + + table = Table(title="Ingest Results", box=box.ROUNDED) + table.add_column("Metric", style="bold") + table.add_column("Count", justify="right") + table.add_row("[green]New documents[/green]", str(new_count)) + table.add_row("[dim]Already cached[/dim]", str(skip_count)) + table.add_row("[bold]Total processed[/bold]", str(new_count + skip_count)) + console.print(table) + + # Auto-parse if requested + if auto_parse and new_count > 0: + console.print("\n[bold]Running parser on new documents...[/bold]") + ctx.invoke(parse, source=source) + + finally: + db.close() + + +def _watch_directory(db, cache, directory: str, source: str, + recursive: bool, auto_parse: bool, ctx): + """Watch a directory for new files and ingest them in real-time.""" + import time as _time + + watched = Path(directory) + seen = set() + + # Seed with existing files + pattern = "**/*" if recursive else "*" + for f in watched.glob(pattern): + if f.is_file() and not f.name.startswith('.'): + seen.add(str(f.resolve())) + + console.print(Panel( + f"[bold cyan]Watch Mode[/bold cyan]\n\n" + f" Directory : {directory}\n" + f" Source : {source}\n" + f" Recursive : {'yes' if recursive else 'no'}\n" + f" Auto-parse: {'yes' if auto_parse else 'no'}\n\n" + f" [dim]Press Ctrl+C to stop watching.[/dim]", + title="[bold]mosaic ingest --watch[/bold]", + border_style="cyan", + )) + console.print(f" [dim]Watching {len(seen)} existing files (skipped)...[/dim]") + + try: + while True: + new_files = [] + for f in watched.glob(pattern): + resolved = str(f.resolve()) + if f.is_file() and not f.name.startswith('.') and resolved not in seen: + new_files.append(f) + seen.add(resolved) + + for fpath in new_files: + is_new = _ingest_raw_file(db, cache, str(fpath), source) + if is_new: + console.print(f" [green]+[/green] {fpath.name}") + if auto_parse: + from parsers.base import ParserManager + from storage.models import Document + cfg = ctx.obj["cfg"] + parser = ParserManager( + memory_limit_mb=int(cfg.get('parser', 'subprocess_memory_limit_mb', fallback='512')), + timeout_seconds=int(cfg.get('parser', 'subprocess_timeout_seconds', fallback='60')), + ) + doc = db.get_document(_hash_file(str(fpath))) + if doc and doc.parse_status == 'PENDING': + result = parser.parse(doc.raw_path, doc.doc_type, doc.content_hash) + if result.success: + doc.text = result.text + doc.title = result.title or doc.title + doc.char_count = result.char_count + ps = 'NEEDS_OCR' if result.needs_ocr else 'PARSED' + db.insert_document(doc) + db.update_document_status(doc.id, parse_status=ps) + console.print(f" [dim]parsed → {ps}[/dim]") + else: + db.update_document_status(doc.id, parse_status='PARSE_FAILED') + console.print(f" [red]parse failed[/red]") + + _time.sleep(2) + except KeyboardInterrupt: + console.print("\n[dim]Watch stopped.[/dim]") + + # --------------------------------------------------------------------------- # status # --------------------------------------------------------------------------- @@ -901,9 +1154,10 @@ MENU_ITEMS = [ ("4", "Analyze", "Full pipeline", "analyze"), ("5", "Query", "Interactive Q&A", "query"), ("6", "Export", "Export to JSONL", "export"), - ("7", "Status", "Progress dashboard", "status"), - ("8", "Verify", "Integrity check", "verify"), - ("9", "Sources", "List profiles", "sources"), + ("7", "Ingest", "Import local files/dumps", "ingest"), + ("8", "Status", "Progress dashboard", "status"), + ("9", "Verify", "Integrity check", "verify"), + ("s", "Sources", "List profiles", "sources"), ("0", "Exit", "", None), ]