#!/usr/bin/env python3 """Shadow Mode CLI — Management and validation tools. Commands: status Show shadow mode status and recent stats export Export daily data for review validate Mark extractions as validated/rejected metrics Calculate precision/recall metrics review-queue Show extractions awaiting validation weekly-report Generate weekly summary report init-db Initialize shadow database Usage: python -m icarus.core.observer.shadow_cli [options] """ import os import sys import json import argparse from pathlib import Path from datetime import datetime, timezone # Ensure we can import from parent sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from icarus.core.observer import ShadowDatabase, ShadowValidator from icarus.core.config import shadow as shadow_config def cmd_status(args): """Show shadow mode status.""" print("=" * 60) print("SHADOW MODE STATUS") print("=" * 60) print(f"ENV: {shadow_config.ICARUS_ENV}") print(f"SPEAK_ENABLED: {shadow_config.SPEAK_ENABLED}") print(f"SHADOW_MODE: {shadow_config.SHADOW_MODE}") print(f"DATA_DIR: {shadow_config.DATA_DIR}") print(f"DB_PATH: {shadow_config.SHADOW_DB_PATH}") print(f"EXPORT_DIR: {shadow_config.EXPORT_DIR}") print(f"FAMILY_CONFIG: {shadow_config.FAMILY_CONFIG_PATH}") print() # Check database if shadow_config.SHADOW_DB_PATH.exists(): db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) stats = db.get_daily_stats() print("TODAY'S STATS:") print(f" Messages: {stats.get('total_messages', 0)}") print(f" Tripwire: {stats.get('tripwire_fired_count', 0)}") print(f" Extractions: {stats.get('extractions_created', 0)}") else: print("Database: NOT INITIALIZED (run init-db)") print() def cmd_export(args): """Export daily data.""" date = args.date or datetime.now(timezone.utc).strftime("%Y-%m-%d") db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) validator = ShadowValidator(db, shadow_config.EXPORT_DIR) path = validator.export_daily(date) if path: print(f"Exported to: {path}") else: print(f"No data for {date}") def cmd_validate(args): """Validate an extraction.""" db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) db.update_validation( extraction_id=args.id, status=args.status, validated_by=args.by, notes=args.notes, ground_truth_type=args.ground_truth_type, ground_truth_who=args.ground_truth_who.split(",") if args.ground_truth_who else None, ) print(f"Updated extraction {args.id}: status={args.status}") def cmd_metrics(args): """Calculate metrics.""" date = args.date or datetime.now(timezone.utc).strftime("%Y-%m-%d") db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) validator = ShadowValidator(db, shadow_config.EXPORT_DIR) metrics = validator.calculate_metrics(date) print(json.dumps(metrics, indent=2)) def cmd_review_queue(args): """Show review queue.""" db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) validator = ShadowValidator(db, shadow_config.EXPORT_DIR) queue = validator.get_review_queue(args.limit) print(f"REVIEW QUEUE ({len(queue)} items)") print("=" * 80) for item in queue: print(f"\nID: {item.get('id')}") print(f" Message: {item.get('message_text', '')[:100]}...") print(f" From: {item.get('sender_name')}") print(f" Type: {item.get('extraction_type')}") print(f" Confidence: {item.get('confidence')}") print(f" Who: {item.get('extracted_who')}") print(f" What: {item.get('extracted_what', '')[:50]}") print("-" * 80) def cmd_weekly_report(args): """Generate weekly report.""" db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) validator = ShadowValidator(db, shadow_config.EXPORT_DIR) report = validator.generate_weekly_report(args.days) if args.json: print(json.dumps(report, indent=2)) else: print("=" * 60) print("WEEKLY SHADOW MODE REPORT") print("=" * 60) print(f"Period: {report['period']['start']} to {report['period']['end']}") print() print("SUMMARY:") print(f" Total Messages: {report['summary']['total_messages']}") print(f" Tripwire Fired: {report['summary']['tripwire_fired']}") print(f" Extractions: {report['summary']['extractions']}") print(f" Tripwire Rate: {report['summary']['tripwire_rate']:.1%}") print() print("DAILY BREAKDOWN:") for day in report['daily']: print(f" {day.get('date')}: {day.get('total_messages', 0)} messages, " f"{day.get('tripwire_fired_count', 0)} tripwire, " f"{day.get('extractions_created', 0)} extractions") def cmd_init_db(args): """Initialize shadow database.""" db = ShadowDatabase(shadow_config.SHADOW_DB_PATH) print(f"Initialized database: {shadow_config.SHADOW_DB_PATH}") # Verify if shadow_config.SHADOW_DB_PATH.exists(): size = shadow_config.SHADOW_DB_PATH.stat().st_size print(f"Database size: {size} bytes") def main(): parser = argparse.ArgumentParser( description="Shadow Mode CLI", formatter_class=argparse.RawDescriptionHelpFormatter, ) subparsers = parser.add_subparsers(dest="command", help="Commands") # status subparsers.add_parser("status", help="Show shadow mode status") # export export_parser = subparsers.add_parser("export", help="Export daily data") export_parser.add_argument("--date", help="Date to export (YYYY-MM-DD), default: today") # validate validate_parser = subparsers.add_parser("validate", help="Validate an extraction") validate_parser.add_argument("id", type=int, help="Extraction ID") validate_parser.add_argument("status", choices=["validated", "rejected", "unclear"], help="Validation status") validate_parser.add_argument("--by", default="manual", help="Who validated") validate_parser.add_argument("--notes", help="Validation notes") validate_parser.add_argument("--ground-truth-type", help="Actual type") validate_parser.add_argument("--ground-truth-who", help="Actual who (comma-separated)") # metrics metrics_parser = subparsers.add_parser("metrics", help="Calculate metrics") metrics_parser.add_argument("--date", help="Date (YYYY-MM-DD), default: today") # review-queue queue_parser = subparsers.add_parser("review-queue", help="Show review queue") queue_parser.add_argument("--limit", type=int, default=50, help="Max items") # weekly-report report_parser = subparsers.add_parser("weekly-report", help="Generate weekly report") report_parser.add_argument("--days", type=int, default=7, help="Days to include") report_parser.add_argument("--json", action="store_true", help="Output as JSON") # init-db subparsers.add_parser("init-db", help="Initialize shadow database") args = parser.parse_args() if not args.command: parser.print_help() return # Dispatch commands = { "status": cmd_status, "export": cmd_export, "validate": cmd_validate, "metrics": cmd_metrics, "review-queue": cmd_review_queue, "weekly-report": cmd_weekly_report, "init-db": cmd_init_db, } func = commands.get(args.command) if func: func(args) else: parser.print_help() if __name__ == "__main__": main()