n0mad1k
943baa3c02
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
1257 lines
47 KiB
Python
Executable File
1257 lines
47 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Mosaic — Intelligence Extraction Platform CLI."""
|
|
import sys
|
|
import os
|
|
import logging
|
|
import hashlib
|
|
import configparser
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.panel import Panel
|
|
from rich.prompt import Prompt
|
|
from rich.markdown import Markdown
|
|
from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn
|
|
from rich import box
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths & constants
|
|
# ---------------------------------------------------------------------------
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
CONFIG_PATH = BASE_DIR / "config" / "mosaic.conf"
|
|
PROFILES_DIR = BASE_DIR / "collectors" / "profiles"
|
|
DEFAULT_DB = "mosaic.db"
|
|
|
|
console = Console()
|
|
logger = logging.getLogger("mosaic")
|
|
|
|
DOMAIN_CHOICES = ["tools", "ttps", "tradecraft", "surveillance", "infrastructure", "entities"]
|
|
EXPORT_CATEGORIES = ["tools", "ttps", "tradecraft", "entities", "infrastructure"]
|
|
QUERY_MODES = ["default", "emulate", "compare", "brief", "scenario"]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_config() -> configparser.ConfigParser:
|
|
"""Load mosaic.conf; return a ConfigParser (empty sections on missing file)."""
|
|
cfg = configparser.ConfigParser()
|
|
if CONFIG_PATH.exists():
|
|
cfg.read(str(CONFIG_PATH))
|
|
else:
|
|
logger.debug("Config file not found at %s — using defaults", CONFIG_PATH)
|
|
return cfg
|
|
|
|
|
|
def get_api_key(cfg: configparser.ConfigParser) -> Optional[str]:
|
|
"""ANTHROPIC_API_KEY env var takes priority over config file."""
|
|
key = os.environ.get("ANTHROPIC_API_KEY")
|
|
if key:
|
|
return key
|
|
return cfg.get("api", "api_key", fallback=None)
|
|
|
|
|
|
def get_db_path(cfg: configparser.ConfigParser, cli_override: Optional[str] = None) -> str:
|
|
"""Resolve database path: CLI flag > config > default."""
|
|
if cli_override:
|
|
return cli_override
|
|
rel = cfg.get("general", "database_path", fallback=DEFAULT_DB)
|
|
return str(BASE_DIR / rel)
|
|
|
|
|
|
def setup_logging(verbose: bool = False):
|
|
"""Configure root logger."""
|
|
level = logging.DEBUG if verbose else logging.INFO
|
|
fmt = "%(asctime)s %(name)-18s %(levelname)-7s %(message)s"
|
|
logging.basicConfig(level=level, format=fmt, datefmt="%H:%M:%S")
|
|
|
|
|
|
def get_output_dir(cfg: configparser.ConfigParser) -> str:
|
|
"""Resolve the output directory from config."""
|
|
rel = cfg.get("general", "output_dir", fallback="output")
|
|
return str(BASE_DIR / rel)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lazy DB / store helpers (avoid opening DB at import time)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def open_db(db_path: str):
|
|
"""Import and return a MosaicDB instance."""
|
|
from storage.db import MosaicDB
|
|
return MosaicDB(db_path)
|
|
|
|
|
|
def open_jsonl_store(db, output_dir: str):
|
|
"""Import and return a JSONLStore instance."""
|
|
from storage.jsonl_store import JSONLStore
|
|
return JSONLStore(db, output_dir=os.path.join(output_dir, "extracted"))
|
|
|
|
|
|
def open_cache(output_dir: str):
|
|
"""Import and return a DownloadCache instance."""
|
|
from utils.cache import DownloadCache
|
|
return DownloadCache(base_dir=os.path.join(output_dir, "raw"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Profile helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_profiles() -> list[dict]:
|
|
"""Scan collectors/profiles/*.yaml and return name+description dicts."""
|
|
profiles = []
|
|
if not PROFILES_DIR.is_dir():
|
|
return profiles
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
# Fallback: parse name/description with basic string matching
|
|
return _load_profiles_fallback()
|
|
|
|
for yf in sorted(PROFILES_DIR.glob("*.yaml")):
|
|
try:
|
|
data = yaml.safe_load(yf.read_text())
|
|
if isinstance(data, dict):
|
|
profiles.append({
|
|
"file": yf.name,
|
|
"name": data.get("name", yf.stem),
|
|
"description": data.get("description", ""),
|
|
"base_url": data.get("base_url", ""),
|
|
"expected_min_count": data.get("expected_min_count", ""),
|
|
})
|
|
except Exception as exc:
|
|
logger.warning("Failed to parse %s: %s", yf.name, exc)
|
|
return profiles
|
|
|
|
|
|
def _load_profiles_fallback() -> list[dict]:
|
|
"""Parse YAML profiles without the yaml library (best-effort)."""
|
|
profiles = []
|
|
for yf in sorted(PROFILES_DIR.glob("*.yaml")):
|
|
name = yf.stem
|
|
description = ""
|
|
base_url = ""
|
|
for line in yf.read_text().splitlines():
|
|
line = line.strip()
|
|
if line.startswith("name:"):
|
|
name = line.split(":", 1)[1].strip().strip('"').strip("'")
|
|
elif line.startswith("description:"):
|
|
description = line.split(":", 1)[1].strip().strip('"').strip("'")
|
|
elif line.startswith("base_url:"):
|
|
base_url = line.split(":", 1)[1].strip().strip('"').strip("'")
|
|
profiles.append({
|
|
"file": yf.name,
|
|
"name": name,
|
|
"description": description,
|
|
"base_url": base_url,
|
|
"expected_min_count": "",
|
|
})
|
|
return profiles
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Click CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@click.group(invoke_without_command=True)
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--url", type=str, default=None, help="Ad-hoc source URL.")
|
|
@click.option("--profile", "profile_path", type=click.Path(exists=True), default=None,
|
|
help="Custom YAML profile path.")
|
|
@click.option("--dry-run", is_flag=True, default=False, help="Estimate without running.")
|
|
@click.option("--retry-failed", is_flag=True, default=False, help="Re-attempt failed docs.")
|
|
@click.option("--since", type=str, default=None, help="Only docs after DATE (YYYY-MM-DD).")
|
|
@click.option("--date-range", nargs=2, type=str, default=None,
|
|
help="Date window: START END (YYYY-MM-DD).")
|
|
@click.option("--since-last-run", is_flag=True, default=False,
|
|
help="Only new/changed since last run.")
|
|
@click.option("--update", is_flag=True, default=False, help="Re-check cached URLs.")
|
|
@click.option("--db", "db_path", type=click.Path(), default=None, help="Database path.")
|
|
@click.option("--verbose", "-v", is_flag=True, default=False, help="Enable debug logging.")
|
|
@click.pass_context
|
|
def cli(ctx, source, url, profile_path, dry_run, retry_failed, since, date_range,
|
|
since_last_run, update, db_path, verbose):
|
|
"""Mosaic -- Intelligence Extraction Platform."""
|
|
setup_logging(verbose)
|
|
cfg = load_config()
|
|
|
|
ctx.ensure_object(dict)
|
|
ctx.obj["cfg"] = cfg
|
|
ctx.obj["source"] = source
|
|
ctx.obj["url"] = url
|
|
ctx.obj["profile_path"] = profile_path
|
|
ctx.obj["dry_run"] = dry_run
|
|
ctx.obj["retry_failed"] = retry_failed
|
|
ctx.obj["since"] = since
|
|
ctx.obj["date_range"] = date_range
|
|
ctx.obj["since_last_run"] = since_last_run
|
|
ctx.obj["update"] = update
|
|
ctx.obj["db_path"] = get_db_path(cfg, db_path)
|
|
ctx.obj["verbose"] = verbose
|
|
ctx.obj["output_dir"] = get_output_dir(cfg)
|
|
|
|
if ctx.invoked_subcommand is None:
|
|
interactive_menu(ctx)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# collect
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--url", type=str, default=None, help="Ad-hoc source URL.")
|
|
@click.option("--name", type=str, default=None, help="Name for ad-hoc source.")
|
|
@click.option("--profile", "profile_path", type=click.Path(exists=True), default=None,
|
|
help="Custom YAML profile path.")
|
|
@click.option("--limit", type=int, default=None, help="Maximum documents to fetch.")
|
|
@click.option("--since", type=str, default=None, help="Only docs after DATE (YYYY-MM-DD).")
|
|
@click.option("--date-range", nargs=2, type=str, default=None,
|
|
help="Date window: START END.")
|
|
@click.option("--update", is_flag=True, default=False, help="Re-check cached URLs.")
|
|
@click.pass_context
|
|
def collect(ctx, source, url, name, profile_path, limit, since, date_range, update):
|
|
"""Download documents from a source."""
|
|
source = source or ctx.obj.get("source")
|
|
url = url or ctx.obj.get("url")
|
|
|
|
if not source and not url and not profile_path:
|
|
console.print("[yellow]No source specified.[/yellow] Use --source, --url, or --profile.")
|
|
raise SystemExit(1)
|
|
|
|
from collectors.profile_loader import load_profile, create_adhoc_profile
|
|
from collectors.custom import CustomSource
|
|
from utils.cache import DownloadCache
|
|
from utils.http_limiter import HTTPRateLimiter
|
|
|
|
# Load profile
|
|
try:
|
|
if profile_path:
|
|
profile = load_profile(profile_path)
|
|
elif url:
|
|
pname = name or url.split('/')[2].replace('.', '_')
|
|
profile = create_adhoc_profile(url, pname)
|
|
else:
|
|
profile = load_profile(source)
|
|
except (FileNotFoundError, ValueError) as e:
|
|
console.print(f"[red]Profile error: {e}[/red]")
|
|
raise SystemExit(1)
|
|
|
|
label = profile['name']
|
|
console.print(Panel(
|
|
f"[bold cyan]Collect[/bold cyan]\n\n"
|
|
f" Source : {label} — {profile.get('description', '')}\n"
|
|
f" Limit : {limit or 'all'}\n"
|
|
f" Since : {since or 'n/a'}\n"
|
|
f" Range : {' to '.join(date_range) if date_range else 'n/a'}\n"
|
|
f" Update : {'yes' if update else 'no'}",
|
|
title="[bold]mosaic collect[/bold]",
|
|
border_style="cyan",
|
|
))
|
|
|
|
db = open_db(ctx.obj["db_path"])
|
|
try:
|
|
cache = DownloadCache(os.path.join(ctx.obj["output_dir"], "raw"))
|
|
cfg = ctx.obj["cfg"]
|
|
http_cfg = {
|
|
'timeout': cfg.get('http', 'timeout', fallback='30'),
|
|
'max_retries': cfg.get('http', 'max_retries', fallback='3'),
|
|
'user_agent': cfg.get('http', 'user_agent', fallback='Mozilla/5.0 (compatible; research-crawler/1.0)'),
|
|
}
|
|
collector = CustomSource(profile, db, cache, config=http_cfg)
|
|
|
|
since_date = since
|
|
until_date = None
|
|
if date_range:
|
|
since_date, until_date = date_range
|
|
|
|
stats = collector.collect(
|
|
since=since_date, until=until_date,
|
|
limit=limit, update=update,
|
|
)
|
|
|
|
table = Table(title="Collection Results", box=box.ROUNDED)
|
|
table.add_column("Metric", style="bold")
|
|
table.add_column("Count", justify="right")
|
|
table.add_row("Discovered", str(stats.get('discovered', 0)))
|
|
table.add_row("Downloaded", str(stats.get('downloaded', 0)))
|
|
table.add_row("Already cached", str(stats.get('cached', 0)))
|
|
table.add_row("Failed", str(stats.get('failed', 0)))
|
|
if update:
|
|
table.add_row("Updated", str(stats.get('updated', 0)))
|
|
console.print(table)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--retry-failed", is_flag=True, default=False, help="Re-attempt failed docs.")
|
|
@click.pass_context
|
|
def parse(ctx, source, retry_failed):
|
|
"""Parse downloaded documents into structured text."""
|
|
source = source or ctx.obj.get("source")
|
|
retry_failed = retry_failed or ctx.obj.get("retry_failed", False)
|
|
|
|
from parsers.base import ParserManager
|
|
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
|
|
|
|
console.print(Panel(
|
|
f"[bold green]Parse[/bold green]\n\n"
|
|
f" Source : {source or 'all'}\n"
|
|
f" Retry failed: {'yes' if retry_failed else 'no'}",
|
|
title="[bold]mosaic parse[/bold]",
|
|
border_style="green",
|
|
))
|
|
|
|
db = open_db(ctx.obj["db_path"])
|
|
try:
|
|
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')),
|
|
min_chars_per_page=int(cfg.get('parser', 'min_chars_per_page', fallback='50')),
|
|
max_pdf_pages=int(cfg.get('parser', 'max_pdf_pages', fallback='200')),
|
|
)
|
|
|
|
# Get documents to parse
|
|
if source:
|
|
docs = db.get_documents_by_source(source, status='CACHED')
|
|
else:
|
|
docs = db.get_documents_by_source('', status='CACHED')
|
|
# Get all cached docs across sources
|
|
cursor = db.conn.execute(
|
|
"SELECT * FROM documents WHERE status = 'CACHED' AND parse_status IN ('PENDING', ?)",
|
|
('PARSE_FAILED' if retry_failed else '__NONE__',)
|
|
)
|
|
from storage.models import Document
|
|
docs = [Document.from_row(row) for row in cursor.fetchall()]
|
|
|
|
if not docs:
|
|
console.print("[yellow]No documents to parse.[/yellow]")
|
|
return
|
|
|
|
parsed = 0
|
|
failed = 0
|
|
needs_ocr = 0
|
|
|
|
with Progress(
|
|
SpinnerColumn(), TextColumn("{task.description}"),
|
|
BarColumn(), TaskProgressColumn(),
|
|
) as progress:
|
|
task = progress.add_task(f"Parsing {len(docs)} docs", total=len(docs))
|
|
for doc in docs:
|
|
if doc.parse_status in ('PARSED', 'NEEDS_OCR') and not retry_failed:
|
|
progress.advance(task)
|
|
continue
|
|
|
|
db.update_document_status(doc.id, parse_status='IN_PROGRESS')
|
|
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
|
|
if result.metadata:
|
|
doc.metadata.update(result.metadata)
|
|
ps = 'NEEDS_OCR' if result.needs_ocr else 'PARSED'
|
|
doc.parse_status = ps
|
|
db.insert_document(doc)
|
|
db.update_document_status(doc.id, parse_status=ps)
|
|
if result.needs_ocr:
|
|
needs_ocr += 1
|
|
else:
|
|
parsed += 1
|
|
else:
|
|
db.update_document_status(doc.id, parse_status='PARSE_FAILED')
|
|
failed += 1
|
|
|
|
progress.advance(task)
|
|
|
|
table = Table(title="Parse Results", box=box.ROUNDED)
|
|
table.add_column("Metric", style="bold")
|
|
table.add_column("Count", justify="right")
|
|
table.add_row("Parsed", str(parsed))
|
|
table.add_row("Needs OCR", str(needs_ocr))
|
|
table.add_row("Failed", str(failed))
|
|
console.print(table)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# extract
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--domain", "-d", multiple=True,
|
|
type=click.Choice(DOMAIN_CHOICES, case_sensitive=False),
|
|
help="Extraction domain(s). Repeat for multiple.")
|
|
@click.option("--dry-run", is_flag=True, default=False, help="Estimate only.")
|
|
@click.option("--since", type=str, default=None, help="Only docs after DATE.")
|
|
@click.option("--date-range", nargs=2, type=str, default=None, help="Date window.")
|
|
@click.option("--since-last-run", is_flag=True, default=False,
|
|
help="Only new/changed since last run.")
|
|
@click.option("--retry-failed", is_flag=True, default=False, help="Re-attempt failed.")
|
|
@click.pass_context
|
|
def extract(ctx, source, domain, dry_run, since, date_range, since_last_run, retry_failed):
|
|
"""Run AI-powered extraction on parsed documents."""
|
|
source = source or ctx.obj.get("source")
|
|
dry_run = dry_run or ctx.obj.get("dry_run", False)
|
|
domains = list(domain) if domain else DOMAIN_CHOICES
|
|
|
|
if dry_run:
|
|
_extract_dry_run(ctx.obj["db_path"], source, domains, since, date_range, since_last_run)
|
|
return
|
|
|
|
console.print(Panel(
|
|
f"[bold magenta]Extract[/bold magenta]\n\n"
|
|
f" Source : {source or 'all'}\n"
|
|
f" Domains : {', '.join(domains)}\n"
|
|
f" Since : {since or 'n/a'}\n"
|
|
f" Range : {' to '.join(date_range) if date_range else 'n/a'}\n"
|
|
f" Since last run : {'yes' if since_last_run else 'no'}\n"
|
|
f" Retry failed : {'yes' if retry_failed else 'no'}",
|
|
title="[bold]mosaic extract[/bold]",
|
|
border_style="magenta",
|
|
))
|
|
console.print(
|
|
"\n[dim]Extraction not yet implemented — see extractors/[/dim]"
|
|
)
|
|
|
|
|
|
def _extract_dry_run(db_path: str, source: Optional[str], domains: list[str],
|
|
since: Optional[str], date_range: Optional[tuple],
|
|
since_last_run: bool):
|
|
"""Estimate extraction scope without running anything."""
|
|
from utils.text_utils import estimate_tokens
|
|
|
|
db = open_db(db_path)
|
|
try:
|
|
if date_range:
|
|
docs = db.get_documents_by_date_range(source=source,
|
|
since=date_range[0],
|
|
until=date_range[1])
|
|
elif since:
|
|
docs = db.get_documents_by_date_range(source=source, since=since)
|
|
elif since_last_run:
|
|
docs = db.get_unextracted_documents(source=source, since_last_run=True)
|
|
else:
|
|
docs = db.get_unextracted_documents(source=source)
|
|
|
|
total_chars = sum(d.char_count for d in docs)
|
|
total_tokens = sum(estimate_tokens(d.text) for d in docs if d.text)
|
|
|
|
table = Table(title="Extraction Dry-Run Estimate", box=box.ROUNDED)
|
|
table.add_column("Metric", style="bold")
|
|
table.add_column("Value", justify="right")
|
|
table.add_row("Documents", str(len(docs)))
|
|
table.add_row("Total characters", f"{total_chars:,}")
|
|
table.add_row("Estimated tokens", f"{total_tokens:,}")
|
|
table.add_row("Domains", ", ".join(domains))
|
|
table.add_row("Est. API calls", str(len(docs) * len(domains)))
|
|
console.print(table)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# analyze
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--limit", type=int, default=None, help="Max documents per stage.")
|
|
@click.pass_context
|
|
def analyze(ctx, source, limit):
|
|
"""Full pipeline: collect -> parse -> extract -> report."""
|
|
source = source or ctx.obj.get("source")
|
|
|
|
if not source:
|
|
console.print("[yellow]No source specified.[/yellow] Use --source.")
|
|
raise SystemExit(1)
|
|
|
|
console.print(Panel(
|
|
f"[bold yellow]Analyze — Full Pipeline[/bold yellow]\n\n"
|
|
f" Source : {source}\n"
|
|
f" Limit : {limit or 'all'}\n\n"
|
|
f" Stages:\n"
|
|
f" 1. Collect — download documents\n"
|
|
f" 2. Parse — extract structured text\n"
|
|
f" 3. Extract — AI-powered extraction\n"
|
|
f" 4. Report — generate output",
|
|
title="[bold]mosaic analyze[/bold]",
|
|
border_style="yellow",
|
|
))
|
|
console.print(
|
|
"\n[dim]Full pipeline not yet implemented — run individual stages.[/dim]"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# query
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.option("--mode", type=click.Choice(QUERY_MODES, case_sensitive=False),
|
|
default="default", help="Query mode.")
|
|
@click.option("--actor", type=str, default=None, help="Filter to actor/group name.")
|
|
@click.option("--topic", type=str, default=None, help="Filter to topic.")
|
|
@click.pass_context
|
|
def query(ctx, source, mode, actor, topic):
|
|
"""Interactive Q&A against the intelligence database."""
|
|
source = source or ctx.obj.get("source")
|
|
db_path = ctx.obj["db_path"]
|
|
|
|
db = open_db(db_path)
|
|
try:
|
|
_query_loop(db, source=source, mode=mode, actor=actor, topic=topic)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _query_loop(db, source: Optional[str] = None, mode: str = "default",
|
|
actor: Optional[str] = None, topic: Optional[str] = None):
|
|
"""Rich-based interactive query loop with FTS5 search."""
|
|
console.print(Panel(
|
|
"[bold cyan]Mosaic Query Console[/bold cyan]\n\n"
|
|
" Type a search query to find documents via full-text search.\n"
|
|
" Special commands:\n"
|
|
" [bold]help[/bold] — show this message\n"
|
|
" [bold]exit[/bold] / [bold]quit[/bold] — leave the console\n"
|
|
" [bold]mode <name>[/bold] — switch query mode\n"
|
|
" [bold]source <name>[/bold] — filter by source",
|
|
title="[bold]mosaic query[/bold]",
|
|
border_style="cyan",
|
|
))
|
|
|
|
if source:
|
|
console.print(f" [dim]Source filter:[/dim] {source}")
|
|
if mode != "default":
|
|
console.print(f" [dim]Mode:[/dim] {mode}")
|
|
if actor:
|
|
console.print(f" [dim]Actor:[/dim] {actor}")
|
|
if topic:
|
|
console.print(f" [dim]Topic:[/dim] {topic}")
|
|
console.print()
|
|
|
|
current_source = source
|
|
current_mode = mode
|
|
|
|
while True:
|
|
try:
|
|
user_input = Prompt.ask("[bold cyan]mosaic[/bold cyan]")
|
|
except (EOFError, KeyboardInterrupt):
|
|
console.print("\n[dim]Exiting.[/dim]")
|
|
break
|
|
|
|
text = user_input.strip()
|
|
if not text:
|
|
continue
|
|
|
|
# --- special commands ---
|
|
lower = text.lower()
|
|
if lower in ("exit", "quit", "q"):
|
|
console.print("[dim]Exiting.[/dim]")
|
|
break
|
|
|
|
if lower == "help":
|
|
console.print(
|
|
" [bold]<query>[/bold] — full-text search\n"
|
|
" [bold]mode <name>[/bold] — switch mode (default/emulate/compare/brief/scenario)\n"
|
|
" [bold]source <name>[/bold] — set source filter (blank to clear)\n"
|
|
" [bold]exit / quit[/bold] — leave"
|
|
)
|
|
continue
|
|
|
|
if lower.startswith("mode "):
|
|
new_mode = text[5:].strip().lower()
|
|
if new_mode in QUERY_MODES:
|
|
current_mode = new_mode
|
|
console.print(f" [dim]Mode set to:[/dim] {current_mode}")
|
|
else:
|
|
console.print(f" [red]Unknown mode.[/red] Choose from: {', '.join(QUERY_MODES)}")
|
|
continue
|
|
|
|
if lower.startswith("source "):
|
|
new_source = text[7:].strip()
|
|
current_source = new_source if new_source else None
|
|
console.print(f" [dim]Source filter:[/dim] {current_source or '(none)'}")
|
|
continue
|
|
|
|
# --- FTS search ---
|
|
results = db.search_documents(text, source=current_source, limit=20)
|
|
if not results:
|
|
console.print(" [dim]No results found.[/dim]")
|
|
continue
|
|
|
|
console.print(f"\n [bold]{len(results)} result(s)[/bold]\n")
|
|
for i, doc in enumerate(results, 1):
|
|
title = doc.title or doc.source_url or doc.id[:16]
|
|
date_str = doc.date or ""
|
|
src = doc.source or ""
|
|
|
|
console.print(
|
|
f" [bold cyan]{i}.[/bold cyan] {title}",
|
|
markup=True, highlight=False,
|
|
)
|
|
console.print(
|
|
f" [dim]{src} {date_str} chars={doc.char_count}[/dim]"
|
|
)
|
|
# Show a text snippet (first 300 chars)
|
|
if doc.text:
|
|
snippet = doc.text[:300].replace("\n", " ")
|
|
console.print(f" {snippet}", markup=False, highlight=False)
|
|
console.print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# export
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command(name="export")
|
|
@click.option("--category", "-c",
|
|
type=click.Choice(EXPORT_CATEGORIES, case_sensitive=False),
|
|
default=None, help="Category to export (default: all).")
|
|
@click.option("--format", "fmt", type=str, default="jsonl",
|
|
help="Export format (default: jsonl).")
|
|
@click.pass_context
|
|
def export_cmd(ctx, category, fmt):
|
|
"""Export extracted intelligence to JSONL files."""
|
|
db_path = ctx.obj["db_path"]
|
|
output_dir = ctx.obj["output_dir"]
|
|
|
|
if fmt != "jsonl":
|
|
console.print(f"[yellow]Format '{fmt}' not supported yet. Using jsonl.[/yellow]")
|
|
|
|
db = open_db(db_path)
|
|
try:
|
|
store = open_jsonl_store(db, output_dir)
|
|
with console.status("[bold green]Exporting..."):
|
|
created = store.export(category=category)
|
|
|
|
if created:
|
|
table = Table(title="Exported Files", box=box.ROUNDED)
|
|
table.add_column("File", style="bold")
|
|
table.add_column("Path")
|
|
for fpath in created:
|
|
table.add_row(Path(fpath).name, fpath)
|
|
console.print(table)
|
|
else:
|
|
console.print("[dim]Nothing to export — database is empty for the requested category.[/dim]")
|
|
finally:
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.pass_context
|
|
def status(ctx, source):
|
|
"""Show progress dashboard with collection, parse, and extraction stats."""
|
|
source = source or ctx.obj.get("source")
|
|
db_path = ctx.obj["db_path"]
|
|
|
|
db = open_db(db_path)
|
|
try:
|
|
_show_status(db, source)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _show_status(db, source: Optional[str] = None):
|
|
"""Render status tables."""
|
|
counts = db.get_status_counts(source=source)
|
|
ext_counts = db.get_extraction_counts(source=source)
|
|
tokens = db.get_total_tokens(source=source)
|
|
|
|
title_suffix = f" [{source}]" if source else ""
|
|
|
|
# --- Collection status ---
|
|
coll_table = Table(title=f"Collection Status{title_suffix}", box=box.ROUNDED)
|
|
coll_table.add_column("Status", style="bold")
|
|
coll_table.add_column("Count", justify="right")
|
|
coll_statuses = ["PENDING", "DOWNLOADING", "CACHED", "DOWNLOAD_FAILED"]
|
|
for s in coll_statuses:
|
|
cnt = counts["collection"].get(s, 0)
|
|
style = "green" if s == "CACHED" else ("red" if "FAIL" in s else "")
|
|
coll_table.add_row(s, f"[{style}]{cnt}[/{style}]" if style else str(cnt))
|
|
coll_table.add_row("[bold]Total[/bold]", f"[bold]{counts['total']}[/bold]")
|
|
|
|
# --- Parse status ---
|
|
parse_table = Table(title=f"Parse Status{title_suffix}", box=box.ROUNDED)
|
|
parse_table.add_column("Status", style="bold")
|
|
parse_table.add_column("Count", justify="right")
|
|
parse_statuses = ["PENDING", "IN_PROGRESS", "PARSED", "PARSE_FAILED", "NEEDS_OCR"]
|
|
for s in parse_statuses:
|
|
cnt = counts["parsing"].get(s, 0)
|
|
style = "green" if s == "PARSED" else ("red" if "FAIL" in s else
|
|
("yellow" if s == "NEEDS_OCR" else ""))
|
|
parse_table.add_row(s, f"[{style}]{cnt}[/{style}]" if style else str(cnt))
|
|
|
|
# --- Extraction status per extractor ---
|
|
ext_table = Table(title=f"Extraction Status{title_suffix}", box=box.ROUNDED)
|
|
ext_table.add_column("Extractor", style="bold")
|
|
ext_table.add_column("PENDING", justify="right")
|
|
ext_table.add_column("IN_PROGRESS", justify="right")
|
|
ext_table.add_column("DONE", justify="right", style="green")
|
|
ext_table.add_column("FAILED", justify="right", style="red")
|
|
if ext_counts:
|
|
for ext_name, statuses in sorted(ext_counts.items()):
|
|
ext_table.add_row(
|
|
ext_name,
|
|
str(statuses.get("PENDING", 0)),
|
|
str(statuses.get("IN_PROGRESS", 0)),
|
|
str(statuses.get("DONE", 0)),
|
|
str(statuses.get("FAILED", 0)),
|
|
)
|
|
else:
|
|
ext_table.add_row("[dim]no extractions yet[/dim]", "", "", "", "")
|
|
|
|
# --- Token usage ---
|
|
token_table = Table(title=f"Token Usage{title_suffix}", box=box.ROUNDED)
|
|
token_table.add_column("Metric", style="bold")
|
|
token_table.add_column("Value", justify="right")
|
|
token_table.add_row("Input tokens", f"{tokens['input_tokens']:,}")
|
|
token_table.add_row("Output tokens", f"{tokens['output_tokens']:,}")
|
|
token_table.add_row("[bold]Total tokens[/bold]",
|
|
f"[bold]{tokens['total_tokens']:,}[/bold]")
|
|
|
|
console.print()
|
|
console.print(coll_table)
|
|
console.print()
|
|
console.print(parse_table)
|
|
console.print()
|
|
console.print(ext_table)
|
|
console.print()
|
|
console.print(token_table)
|
|
console.print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# verify
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--source", type=str, default=None, help="Source profile name.")
|
|
@click.pass_context
|
|
def verify(ctx, source):
|
|
"""Re-hash cached files and report integrity status."""
|
|
source = source or ctx.obj.get("source")
|
|
db_path = ctx.obj["db_path"]
|
|
output_dir = ctx.obj["output_dir"]
|
|
|
|
db = open_db(db_path)
|
|
try:
|
|
_run_verify(db, output_dir, source)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _run_verify(db, output_dir: str, source: Optional[str] = None):
|
|
"""Verify cached files against stored hashes."""
|
|
cache = open_cache(output_dir)
|
|
doc_hashes = db.get_all_document_hashes()
|
|
|
|
if not doc_hashes:
|
|
console.print("[dim]No cached documents to verify.[/dim]")
|
|
return
|
|
|
|
verified = 0
|
|
mismatched = 0
|
|
missing = 0
|
|
bad_files: list[str] = []
|
|
|
|
with Progress(
|
|
SpinnerColumn(),
|
|
TextColumn("[progress.description]{task.description}"),
|
|
BarColumn(),
|
|
TextColumn("{task.completed}/{task.total}"),
|
|
console=console,
|
|
) as progress:
|
|
task = progress.add_task("Verifying files...", total=len(doc_hashes))
|
|
|
|
for doc_id, raw_path, expected_hash in doc_hashes:
|
|
if source:
|
|
# Filter by source if provided — look up in DB
|
|
doc = db.get_document(doc_id)
|
|
if doc and doc.source != source:
|
|
progress.advance(task)
|
|
continue
|
|
|
|
if not raw_path or not Path(raw_path).exists():
|
|
missing += 1
|
|
progress.advance(task)
|
|
continue
|
|
|
|
actual_hash = cache.compute_hash(raw_path)
|
|
if actual_hash == expected_hash:
|
|
verified += 1
|
|
else:
|
|
mismatched += 1
|
|
bad_files.append(raw_path)
|
|
|
|
progress.advance(task)
|
|
|
|
# Results table
|
|
table = Table(title="Integrity Verification Results", box=box.ROUNDED)
|
|
table.add_column("Status", style="bold")
|
|
table.add_column("Count", justify="right")
|
|
table.add_row("[green]Verified[/green]", str(verified))
|
|
table.add_row("[red]Mismatched[/red]", str(mismatched))
|
|
table.add_row("[yellow]Missing[/yellow]", str(missing))
|
|
table.add_row("[bold]Total checked[/bold]", str(verified + mismatched + missing))
|
|
console.print()
|
|
console.print(table)
|
|
|
|
if bad_files:
|
|
console.print("\n[red bold]Mismatched files:[/red bold]")
|
|
for bf in bad_files:
|
|
console.print(f" [red]- {bf}[/red]", markup=False)
|
|
elif mismatched == 0 and missing == 0:
|
|
console.print("\n[green]All cached files passed integrity checks.[/green]")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# manual
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.option("--output", "-o", type=click.Path(), default="output/FIELD_MANUAL.md",
|
|
help="Output file path.")
|
|
@click.option("--title", type=str, default="Operator's Field Manual",
|
|
help="Manual title.")
|
|
@click.pass_context
|
|
def manual(ctx, output, title):
|
|
"""Generate Markdown field manual from extracted intelligence."""
|
|
from analysis.field_manual import generate_field_manual
|
|
|
|
db = open_db(ctx.obj["db_path"])
|
|
try:
|
|
path = generate_field_manual(db, output_path=output, title=title)
|
|
console.print(f"\n[bold green]Field manual generated:[/bold green] {path}")
|
|
|
|
# Show stats
|
|
import os
|
|
size = os.path.getsize(path)
|
|
with open(path) as f:
|
|
lines = sum(1 for _ in f)
|
|
console.print(f" Size: {size:,} bytes ({lines:,} lines)")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sources
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@cli.command()
|
|
@click.pass_context
|
|
def sources(ctx):
|
|
"""List available source profiles."""
|
|
profiles = load_profiles()
|
|
|
|
table = Table(title="Source Profiles", box=box.ROUNDED)
|
|
table.add_column("Name", style="bold cyan")
|
|
table.add_column("Description")
|
|
table.add_column("Base URL", style="dim")
|
|
table.add_column("Expected Docs", justify="right")
|
|
table.add_column("File", style="dim")
|
|
|
|
if profiles:
|
|
for p in profiles:
|
|
table.add_row(
|
|
p["name"],
|
|
p["description"],
|
|
p["base_url"],
|
|
str(p["expected_min_count"]) if p["expected_min_count"] else "",
|
|
p["file"],
|
|
)
|
|
else:
|
|
table.add_row("[dim]no profiles found[/dim]", "", "", "", "")
|
|
|
|
console.print()
|
|
console.print(table)
|
|
console.print(f"\n [dim]Profile directory: {PROFILES_DIR}[/dim]")
|
|
console.print(" [dim]Use --profile to load a custom YAML file from any path.[/dim]\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interactive menu
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MENU_ITEMS = [
|
|
("1", "Collect", "Download documents", "collect"),
|
|
("2", "Parse", "Extract structured text", "parse"),
|
|
("3", "Extract", "AI-powered extraction", "extract"),
|
|
("4", "Analyze", "Full pipeline", "analyze"),
|
|
("5", "Query", "Interactive Q&A", "query"),
|
|
("6", "Export", "Export to JSONL", "export"),
|
|
("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),
|
|
]
|
|
|
|
|
|
def interactive_menu(ctx: click.Context):
|
|
"""Display the main interactive menu and dispatch commands."""
|
|
while True:
|
|
lines = []
|
|
for key, label, desc, _ in MENU_ITEMS:
|
|
if key == "0":
|
|
lines.append(f" [bold red][{key}][/bold red] {label}")
|
|
else:
|
|
lines.append(f" [bold cyan][{key}][/bold cyan] {label:<12} [dim]{desc}[/dim]")
|
|
|
|
panel_text = "\n".join(lines)
|
|
console.print()
|
|
console.print(Panel(
|
|
panel_text,
|
|
title="[bold white]MOSAIC[/bold white] [dim]Intelligence Extraction Platform[/dim]",
|
|
border_style="bright_blue",
|
|
padding=(1, 2),
|
|
))
|
|
|
|
try:
|
|
choice = Prompt.ask("\n[bold bright_blue]Select[/bold bright_blue]",
|
|
choices=[m[0] for m in MENU_ITEMS],
|
|
show_choices=False)
|
|
except (EOFError, KeyboardInterrupt):
|
|
console.print("\n[dim]Exiting.[/dim]")
|
|
break
|
|
|
|
if choice == "0":
|
|
console.print("[dim]Exiting.[/dim]")
|
|
break
|
|
|
|
# Find the matching menu item
|
|
for key, label, desc, cmd_name in MENU_ITEMS:
|
|
if key == choice and cmd_name:
|
|
_dispatch_interactive(ctx, cmd_name)
|
|
break
|
|
|
|
|
|
def _dispatch_interactive(ctx: click.Context, cmd_name: str):
|
|
"""Invoke a subcommand from the interactive menu."""
|
|
cmd = cli.get_command(ctx, cmd_name)
|
|
if cmd is None:
|
|
console.print(f"[red]Unknown command: {cmd_name}[/red]")
|
|
return
|
|
|
|
# Some commands benefit from prompting for a source first
|
|
if cmd_name in ("collect", "parse", "extract", "analyze", "status", "verify"):
|
|
source = _prompt_source_optional()
|
|
if source:
|
|
ctx.obj["source"] = source
|
|
|
|
try:
|
|
sub_ctx = click.Context(cmd, parent=ctx, info_name=cmd_name)
|
|
sub_ctx.ensure_object(dict)
|
|
sub_ctx.obj = ctx.obj
|
|
cmd.invoke(sub_ctx)
|
|
except SystemExit:
|
|
pass
|
|
except Exception as exc:
|
|
console.print(f"[red]Error: {exc}[/red]")
|
|
logger.exception("Command '%s' failed", cmd_name)
|
|
|
|
|
|
def _prompt_source_optional() -> Optional[str]:
|
|
"""Optionally prompt for a source profile name."""
|
|
profiles = load_profiles()
|
|
if profiles:
|
|
names = [p["name"] for p in profiles]
|
|
console.print(f"\n [dim]Available sources: {', '.join(names)}[/dim]")
|
|
|
|
source = Prompt.ask(
|
|
" [dim]Source (blank for all)[/dim]",
|
|
default="",
|
|
show_default=False,
|
|
)
|
|
return source.strip() or None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
"""Entry point for mosaic CLI."""
|
|
# Ensure we can import project modules regardless of cwd
|
|
if str(BASE_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BASE_DIR))
|
|
cli(standalone_mode=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|