!/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
"""
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()