scripts-learning-db-query

#!/usr/bin/env python3 """

title: "Database path" component_type: script version: "1.0.0" audience: contributor status: stable summary: "CODITECT Learning Management System Query Interface" keywords: ['database', 'learning', 'query', 'review'] tokens: ~500 created: 2025-12-22 updated: 2025-12-22 script_name: "learning_db_query.py" language: python executable: true usage: "python3 scripts/learning_db_query.py [options]" python_version: "3.10+" dependencies: [] modifies_files: false network_access: false requires_auth: false

CODITECT Learning Management System Query Interface

Query learning progress, modules, badges, and achievements from the LMS database. Integrates with /cxq via the --onboarding and --learning flags.

Usage: python3 scripts/learning-db-query.py --progress # Show learning progress python3 scripts/learning-db-query.py --badges # Show earned badges python3 scripts/learning-db-query.py --modules # List available modules python3 scripts/learning-db-query.py --next # Recommend next module python3 scripts/learning-db-query.py --stats # Learning statistics python3 scripts/learning-db-query.py --search "commands" # Search learning content

Version: 1.0.0 """

import argparse import json import sqlite3 import sys from datetime import datetime from pathlib import Path from typing import Dict, List, Optional

ADR-114 & ADR-118: Use centralized path discovery for user data

SCRIPT_DIR = Path(file).parent sys.path.insert(0, str(SCRIPT_DIR / "core"))

try: from paths import ( get_context_storage_dir, get_sessions_db_path, SESSIONS_DB, ) DB_PATH = SESSIONS_DB # Learning data is TIER 3 (regenerable session data) except ImportError: # Fallback for backward compatibility _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): DB_PATH = _user_data / "sessions.db" else: DB_PATH = SCRIPT_DIR.parent / "context-storage" / "sessions.db"

NOTE: context.db is DEPRECATED - NO FALLBACK per ADR-118

def get_db() -> sqlite3.Connection: """Get database connection (ADR-118 four-tier architecture).

NOTE: context.db is DEPRECATED - NO FALLBACK per ADR-118.
"""
if not DB_PATH.exists():
    print(f"Error: sessions.db not found at {DB_PATH}")
    print("Run: python3 scripts/learning-db-migrate.py --all or /cx")
    print("NOTE: context.db is DEPRECATED - NO FALLBACK per ADR-118")
    sys.exit(1)

conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
return conn

def get_learning_progress(user_id: str = "default") -> List[Dict]: """Get user's learning progress across all modules.""" conn = get_db()

results = conn.execute("""
    SELECT
        m.module_key,
        m.title,
        m.category,
        m.estimated_duration_minutes,
        COALESCE(p.status, 'not_started') as status,
        p.quiz_score as score,
        p.completed_at,
        p.time_spent_seconds,
        b.icon as badge_icon,
        b.title as badge_title
    FROM learning_modules m
    LEFT JOIN learning_progress p ON m.id = p.module_id AND p.user_id = ?
    LEFT JOIN learning_badges b ON m.badge_id = b.badge_key
    WHERE m.is_active = 1
    ORDER BY m.sort_order
""", (user_id,)).fetchall()

conn.close()
return [dict(r) for r in results]

def get_earned_badges(user_id: str = "default") -> List[Dict]: """Get user's earned badges.""" conn = get_db()

results = conn.execute("""
    SELECT
        b.badge_key,
        b.icon,
        b.title,
        b.description,
        b.badge_category,
        a.earned_at,
        a.notes as earned_context
    FROM learning_achievements a
    JOIN learning_badges b ON b.id = a.badge_id
    WHERE a.user_id = ?
    ORDER BY a.earned_at DESC
""", (user_id,)).fetchall()

conn.close()
return [dict(r) for r in results]

def get_available_badges() -> List[Dict]: """Get all available badges.""" conn = get_db()

results = conn.execute("""
    SELECT badge_key, icon, title, description, badge_category
    FROM learning_badges
    WHERE is_active = 1
    ORDER BY sort_order
""").fetchall()

conn.close()
return [dict(r) for r in results]

def get_available_modules(category: str = None) -> List[Dict]: """Get available learning modules.""" conn = get_db()

sql = """
    SELECT
        module_key,
        title,
        description,
        category,
        estimated_duration_minutes,
        difficulty_level,
        has_quiz,
        has_exercise,
        badge_id
    FROM learning_modules
    WHERE is_active = 1
"""
params = []

if category:
    sql += " AND category = ?"
    params.append(category)

sql += " ORDER BY sort_order"

results = conn.execute(sql, params).fetchall()
conn.close()
return [dict(r) for r in results]

def get_next_module(user_id: str = "default") -> Optional[Dict]: """Get recommended next module based on progress.""" conn = get_db()

# Find first incomplete module in order
result = conn.execute("""
    SELECT
        m.module_key,
        m.title,
        m.description,
        m.category,
        m.estimated_duration_minutes,
        m.learning_objectives
    FROM learning_modules m
    LEFT JOIN learning_progress p ON m.id = p.module_id AND p.user_id = ?
    WHERE m.is_active = 1
      AND (p.status IS NULL OR p.status NOT IN ('completed', 'skipped'))
    ORDER BY m.sort_order
    LIMIT 1
""", (user_id,)).fetchone()

conn.close()
return dict(result) if result else None

def get_learning_stats(user_id: str = "default") -> Dict: """Get learning statistics.""" conn = get_db()

# Module stats
module_stats = conn.execute("""
    SELECT
        COUNT(*) as total_modules,
        SUM(CASE WHEN p.status = 'completed' THEN 1 ELSE 0 END) as completed,
        SUM(CASE WHEN p.status = 'in_progress' THEN 1 ELSE 0 END) as in_progress,
        SUM(CASE WHEN p.status = 'skipped' THEN 1 ELSE 0 END) as skipped,
        SUM(COALESCE(p.time_spent_seconds, 0)) as total_time_seconds,
        AVG(CASE WHEN p.quiz_score IS NOT NULL THEN p.quiz_score END) as avg_score
    FROM learning_modules m
    LEFT JOIN learning_progress p ON m.id = p.module_id AND p.user_id = ?
    WHERE m.is_active = 1
""", (user_id,)).fetchone()

# Badge stats
badge_stats = conn.execute("""
    SELECT
        COUNT(*) as total_badges,
        (SELECT COUNT(*) FROM learning_achievements WHERE user_id = ?) as earned_badges
    FROM learning_badges
    WHERE is_active = 1
""", (user_id,)).fetchone()

# Recent activity
recent = conn.execute("""
    SELECT event_type, COUNT(*) as count
    FROM learning_analytics_events
    WHERE user_id = ?
    GROUP BY event_type
    ORDER BY count DESC
    LIMIT 5
""", (user_id,)).fetchall()

conn.close()

total = module_stats['total_modules'] or 1
completed = module_stats['completed'] or 0

return {
    'modules': {
        'total': total,
        'completed': completed,
        'in_progress': module_stats['in_progress'] or 0,
        'skipped': module_stats['skipped'] or 0,
        'completion_percent': round(100 * completed / total, 1)
    },
    'badges': {
        'total': badge_stats['total_badges'] or 0,
        'earned': badge_stats['earned_badges'] or 0
    },
    'time': {
        'total_seconds': module_stats['total_time_seconds'] or 0,
        'total_minutes': round((module_stats['total_time_seconds'] or 0) / 60, 1)
    },
    'performance': {
        'avg_score': round(module_stats['avg_score'] or 0, 1)
    },
    'activity': [dict(r) for r in recent]
}

def search_learning_content(query: str, limit: int = 20) -> List[Dict]: """Search learning modules and resources.""" conn = get_db()

# Search modules
modules = conn.execute("""
    SELECT
        'module' as content_type,
        module_key as id,
        title,
        description,
        category
    FROM learning_modules
    WHERE is_active = 1
      AND (title LIKE ? OR description LIKE ? OR category LIKE ?)
    LIMIT ?
""", (f'%{query}%', f'%{query}%', f'%{query}%', limit)).fetchall()

# Search badges
badges = conn.execute("""
    SELECT
        'badge' as content_type,
        badge_key as id,
        title,
        description,
        badge_category as category
    FROM learning_badges
    WHERE is_active = 1
      AND (title LIKE ? OR description LIKE ?)
    LIMIT ?
""", (f'%{query}%', f'%{query}%', limit)).fetchall()

conn.close()

results = [dict(r) for r in modules] + [dict(r) for r in badges]
return results[:limit]

def format_progress_display(progress: List[Dict]) -> str: """Format progress for display.""" lines = [] lines.append("") lines.append("=" * 60) lines.append("CODITECT LEARNING PROGRESS") lines.append("=" * 60) lines.append("")

completed = sum(1 for p in progress if p['status'] == 'completed')
total = len(progress)
pct = round(100 * completed / total, 1) if total > 0 else 0

# Progress bar
bar_width = 30
filled = int(bar_width * completed / total) if total > 0 else 0
bar = "#" * filled + "-" * (bar_width - filled)
lines.append(f"Overall: [{bar}] {pct}% ({completed}/{total})")
lines.append("")

# Module list
status_icons = {
    'completed': '[x]',
    'in_progress': '[>]',
    'skipped': '[-]',
    'not_started': '[ ]'
}

for p in progress:
    icon = status_icons.get(p['status'], '[ ]')
    badge = f" {p['badge_icon']}" if p['badge_icon'] and p['status'] == 'completed' else ""
    duration = f"({p['estimated_duration_minutes']}min)" if p['estimated_duration_minutes'] else ""
    lines.append(f"  {icon} {p['title']:<40} {duration}{badge}")

lines.append("")
return "\n".join(lines)

def format_badges_display(badges: List[Dict], all_badges: List[Dict]) -> str: """Format badges for display.""" lines = [] lines.append("") lines.append("=" * 60) lines.append("CODITECT BADGES") lines.append("=" * 60) lines.append("")

earned_keys = {b['badge_key'] for b in badges}

lines.append(f"Earned: {len(badges)} / {len(all_badges)}")
lines.append("")

# Earned badges
if badges:
    lines.append("EARNED:")
    for b in badges:
        earned_date = b['earned_at'][:10] if b['earned_at'] else ""
        lines.append(f"  {b['icon']} {b['title']:<30} {earned_date}")
    lines.append("")

# Available badges
available = [b for b in all_badges if b['badge_key'] not in earned_keys]
if available:
    lines.append("AVAILABLE:")
    for b in available:
        lines.append(f"  [ ] {b['title']:<30} - {b['description']}")
    lines.append("")

return "\n".join(lines)

def format_modules_display(modules: List[Dict]) -> str: """Format modules for display.""" lines = [] lines.append("") lines.append("=" * 60) lines.append("AVAILABLE LEARNING MODULES") lines.append("=" * 60) lines.append("")

# Group by category
categories = {}
for m in modules:
    cat = m['category'] or 'general'
    if cat not in categories:
        categories[cat] = []
    categories[cat].append(m)

for cat, mods in categories.items():
    lines.append(f"{cat.upper()}:")
    for m in mods:
        duration = f"({m['estimated_duration_minutes']}min)" if m['estimated_duration_minutes'] else ""
        quiz = "[Q]" if m['has_quiz'] else ""
        exercise = "[E]" if m['has_exercise'] else ""
        lines.append(f"  - {m['title']:<40} {duration} {quiz}{exercise}")
    lines.append("")

return "\n".join(lines)

def format_stats_display(stats: Dict) -> str: """Format statistics for display.""" lines = [] lines.append("") lines.append("=" * 60) lines.append("LEARNING STATISTICS") lines.append("=" * 60) lines.append("")

m = stats['modules']
lines.append("MODULES:")
lines.append(f"  Completed:   {m['completed']}/{m['total']} ({m['completion_percent']}%)")
lines.append(f"  In Progress: {m['in_progress']}")
lines.append(f"  Skipped:     {m['skipped']}")
lines.append("")

b = stats['badges']
lines.append("BADGES:")
lines.append(f"  Earned: {b['earned']}/{b['total']}")
lines.append("")

t = stats['time']
lines.append("TIME:")
lines.append(f"  Total: {t['total_minutes']} minutes")
lines.append("")

p = stats['performance']
if p['avg_score'] > 0:
    lines.append("PERFORMANCE:")
    lines.append(f"  Average Score: {p['avg_score']}%")
    lines.append("")

return "\n".join(lines)

def format_next_display(next_module: Optional[Dict]) -> str: """Format next module recommendation.""" lines = [] lines.append("") lines.append("=" * 60) lines.append("RECOMMENDED NEXT MODULE") lines.append("=" * 60) lines.append("")

if next_module:
    lines.append(f"Title: {next_module['title']}")
    lines.append(f"Category: {next_module['category']}")
    if next_module['estimated_duration_minutes']:
        lines.append(f"Duration: {next_module['estimated_duration_minutes']} minutes")
    if next_module['description']:
        lines.append(f"Description: {next_module['description']}")
    if next_module.get('learning_objectives'):
        try:
            objectives = json.loads(next_module['learning_objectives'])
            lines.append("Learning Objectives:")
            for obj in objectives:
                lines.append(f"  - {obj}")
        except (json.JSONDecodeError, TypeError):
            pass
    lines.append("")
    lines.append(f"Start with: /onboard-full --from {next_module['module_key']}")
else:
    lines.append("Congratulations! You've completed all available modules!")
    lines.append("")
    lines.append("Next steps:")
    lines.append("  - Review the CODITECT Cookbook: docs/02-user-guides/CODITECT-COOKBOOK.md")
    lines.append("  - Explore certification: /training-status")
    lines.append("  - Start a project: /new-project")

lines.append("")
return "\n".join(lines)

def main(): parser = argparse.ArgumentParser( description='CODITECT Learning Management System Query Interface', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 scripts/learning-db-query.py --progress # Show your progress python3 scripts/learning-db-query.py --badges # Show earned badges python3 scripts/learning-db-query.py --modules # List all modules python3 scripts/learning-db-query.py --next # What to learn next python3 scripts/learning-db-query.py --stats # Learning statistics python3 scripts/learning-db-query.py --search agents # Search content

Integration with /cxq: These commands are also available via /cxq: /cxq --onboarding --progress /cxq --onboarding --badges /cxq --learning --search "topic" """ )

# Query options
parser.add_argument('--progress', '-p', action='store_true',
                   help='Show learning progress')
parser.add_argument('--badges', '-b', action='store_true',
                   help='Show earned and available badges')
parser.add_argument('--modules', '-m', action='store_true',
                   help='List available learning modules')
parser.add_argument('--next', '-n', action='store_true',
                   help='Recommend next module to learn')
parser.add_argument('--stats', '-s', action='store_true',
                   help='Show learning statistics')
parser.add_argument('--search', metavar='QUERY',
                   help='Search learning content')

# Filters
parser.add_argument('--category', '-c',
                   help='Filter modules by category')
parser.add_argument('--user', '-u', default='default',
                   help='User ID (default: default)')

# Output
parser.add_argument('--json', '-j', action='store_true',
                   help='Output as JSON')
parser.add_argument('--limit', '-l', type=int, default=20,
                   help='Result limit for search (default: 20)')

args = parser.parse_args()

# Default to --progress if no option specified
if not any([args.progress, args.badges, args.modules, args.next, args.stats, args.search]):
    args.progress = True

# Progress
if args.progress:
    results = get_learning_progress(args.user)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_progress_display(results))
    return

# Badges
if args.badges:
    earned = get_earned_badges(args.user)
    all_badges = get_available_badges()
    if args.json:
        print(json.dumps({'earned': earned, 'available': all_badges}, indent=2))
    else:
        print(format_badges_display(earned, all_badges))
    return

# Modules
if args.modules:
    results = get_available_modules(args.category)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_modules_display(results))
    return

# Next recommendation
if args.next:
    result = get_next_module(args.user)
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print(format_next_display(result))
    return

# Stats
if args.stats:
    results = get_learning_stats(args.user)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_stats_display(results))
    return

# Search
if args.search:
    results = search_learning_content(args.search, args.limit)
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(f"\nFound {len(results)} results for '{args.search}':\n")
        for r in results:
            print(f"  [{r['content_type']}] {r['title']}")
            if r['description']:
                print(f"         {r['description'][:60]}...")
        print()
    return

if name == 'main': main()