local-deep-research/scripts/check_metrics.py

#!/usr/bin/env python3
"""Check if metrics are being saved in the database."""

import sys
from pathlib import Path

# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))

from src.local_deep_research.database.encrypted_db import db_manager
from src.local_deep_research.database.models import TokenUsage, SearchCall


def check_user_metrics(username: str, password: str):
    """Check metrics for a specific user."""
    print(f"\n🔍 Checking metrics for user: {username}")
    print("=" * 60)

    try:
        # Open database
        if not db_manager.open_user_database(username, password):
            print("❌ Failed to open database")
            return

        # Get session
        session = db_manager.get_session(username)
        if not session:
            print("❌ Failed to get session")
            return

        # Check token usage
        token_count = session.query(TokenUsage).count()
        print(f"\n📊 Token Usage Records: {token_count}")

        if token_count > 0:
            # Get recent token usage
            recent_tokens = (
                session.query(TokenUsage)
                .order_by(TokenUsage.created_at.desc())
                .limit(5)
                .all()
            )
            print("\n   Recent token usage:")
            for token in recent_tokens:
                print(
                    f"   - {token.created_at}: {token.model_name} - {token.total_tokens} tokens"
                )
                print(
                    f"     Phase: {token.research_phase}, Status: {token.success_status}"
                )

        # Check search calls
        search_count = session.query(SearchCall).count()
        print(f"\n🔎 Search Call Records: {search_count}")

        if search_count > 0:
            # Get recent searches
            recent_searches = (
                session.query(SearchCall)
                .order_by(SearchCall.created_at.desc())
                .limit(5)
                .all()
            )
            print("\n   Recent searches:")
            for search in recent_searches:
                print(
                    f"   - {search.created_at}: {search.search_engine} - {search.query[:50]}..."
                )
                print(f"     Results: {search.results_returned}")

        session.close()

    except Exception as e:
        print(f"❌ Error: {e}")
        import traceback

        traceback.print_exc()
    finally:
        if username in db_manager.connections:
            db_manager.connections.pop(username)


def main():
    """Check metrics for test users."""
    # Check for a specific test user (modify as needed)
    test_users = [
        ("simple_1751323627595", "password"),  # Latest test user
        # Add more test users as needed
    ]

    for username, password in test_users:
        try:
            check_user_metrics(username, password)
        except Exception as e:
            print(f"Failed to check {username}: {e}")

    print("\n" + "=" * 60)
    print("✅ Metrics check complete")


if __name__ == "__main__":
    main()