pdf-quiz-generator/backend/app/cli.py

"""CLI management commands. Run inside the backend container:

  docker compose exec backend python -m app.cli <command> [args]

Commands:
  list-users                          List all users
  reset-password <email> <password>   Reset a user's password
  set-role <email> <role>             Change user role (admin/moderator/user)
  verify-email <email>                Force-verify a user's email
  delete-user <email>                 Delete a user and their data
  stats                               Show platform statistics
  list-quizzes                        List quizzes (--all includes deleted)
  list-docs                           List all documents with status
  fix-stuck-docs                      Reset stuck 'processing' docs to 'ready'
  reprocess-doc <id>                  Requeue a document for processing
  cleanup-redis                       Clear expired/orphaned Redis keys
  export-users                        Export users to CSV
"""
import sys
import csv
import io

from app.database import SessionLocal
from app.models.user import User
from app.utils.auth import get_password_hash


def reset_password(email: str, new_password: str):
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email.lower().strip()).first()
        if not user:
            print(f"Error: No user found with email '{email}'")
            return False
        if len(new_password) < 6:
            print("Error: Password must be at least 6 characters")
            return False
        user.hashed_password = get_password_hash(new_password)
        db.commit()
        print(f"Password reset for {user.name} ({user.email}) — role: {user.role}")
        return True
    finally:
        db.close()


def list_users(args=None):
    db = SessionLocal()
    try:
        users = db.query(User).order_by(User.id).all()
        print(f"{'ID':<5} {'Name':<25} {'Email':<35} {'Role':<12}")
        print("-" * 77)
        for u in users:
            print(f"{u.id:<5} {(u.name or '-'):<25} {u.email:<35} {u.role:<12}")
        print(f"\n{len(users)} user(s)")
    finally:
        db.close()


def set_role(email: str, role: str):
    if role not in ("admin", "moderator", "user"):
        print(f"Error: Invalid role '{role}'. Must be: admin, moderator, user")
        return False
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email.lower().strip()).first()
        if not user:
            print(f"Error: No user found with email '{email}'")
            return False
        old_role = user.role
        user.role = role
        db.commit()
        print(f"{user.name} ({user.email}): {old_role} -> {role}")
        return True
    finally:
        db.close()


def verify_email(email: str):
    from datetime import datetime
    from app.models.email_verification import EmailVerification
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email.lower().strip()).first()
        if not user:
            print(f"Error: No user found with email '{email}'")
            return False
        v = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
        if not v:
            print(f"{user.email} has no verification record (already verified or legacy user)")
            return True
        if v.verified_at:
            print(f"{user.email} is already verified (at {v.verified_at})")
            return True
        v.verified_at = datetime.utcnow()
        db.commit()
        print(f"Email verified for {user.name} ({user.email})")
        return True
    finally:
        db.close()


def delete_user(email: str):
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email.lower().strip()).first()
        if not user:
            print(f"Error: No user found with email '{email}'")
            return False
        confirm = input(f"Delete {user.name} ({user.email}, role={user.role})? Type 'yes' to confirm: ")
        if confirm.strip().lower() != "yes":
            print("Cancelled.")
            return False
        db.delete(user)
        db.commit()
        print(f"Deleted user {user.name} ({user.email})")
        return True
    finally:
        db.close()


def stats(args=None):
    from app.models.quiz import Quiz
    from app.models.question import Question
    from app.models.attempt import QuizAttempt
    from app.models.flashcard import FlashcardDeck, Flashcard
    from app.models.pdf_document import PDFDocument
    db = SessionLocal()
    try:
        users = db.query(User).count()
        admins = db.query(User).filter(User.role == "admin").count()
        mods = db.query(User).filter(User.role == "moderator").count()
        quizzes = db.query(Quiz).filter(Quiz.deleted_at.is_(None)).count()
        questions = db.query(Question).count()
        attempts = db.query(QuizAttempt).filter(QuizAttempt.completed_at.isnot(None)).count()
        in_progress = db.query(QuizAttempt).filter(QuizAttempt.completed_at.is_(None)).count()
        decks = db.query(FlashcardDeck).count()
        cards = db.query(Flashcard).count()
        docs = db.query(PDFDocument).count()

        print("Platform Statistics")
        print("=" * 40)
        print(f"Users:            {users} ({admins} admin, {mods} moderator)")
        print(f"Documents:        {docs}")
        print(f"Questions:        {questions}")
        print(f"Quizzes:          {quizzes}")
        print(f"Attempts:         {attempts} completed, {in_progress} in-progress")
        print(f"Flashcard decks:  {decks} ({cards} cards)")
    finally:
        db.close()


def list_quizzes(args=None):
    from app.models.quiz import Quiz
    db = SessionLocal()
    try:
        q = db.query(Quiz)
        if not (args and "--all" in args):
            q = q.filter(Quiz.deleted_at.is_(None))
        quizzes = q.order_by(Quiz.id).all()
        print(f"{'ID':<5} {'Title':<35} {'Qs':<5} {'Mode':<10} {'Status':<10}")
        print("-" * 65)
        for qz in quizzes:
            status = "deleted" if qz.deleted_at else ("shared" if qz.is_shared else "private")
            print(f"{qz.id:<5} {(qz.title or '-')[:34]:<35} {qz.questions_count:<5} {qz.mode:<10} {status:<10}")
        print(f"\n{len(quizzes)} quiz(zes)")
    finally:
        db.close()


def list_docs(args=None):
    from app.models.pdf_document import PDFDocument
    db = SessionLocal()
    try:
        docs = db.query(PDFDocument).order_by(PDFDocument.id).all()
        print(f"{'ID':<5} {'Filename':<45} {'Pages':<7} {'Status':<12} {'Uploaded':<12}")
        print("-" * 81)
        for d in docs:
            pages = str(d.total_pages) if d.total_pages else "-"
            date = d.uploaded_at.strftime("%Y-%m-%d") if d.uploaded_at else "-"
            fname = (d.original_filename or "-")[:44]
            print(f"{d.id:<5} {fname:<45} {pages:<7} {d.status:<12} {date:<12}")
            if d.status == "error" and d.error_message:
                print(f"      Error: {d.error_message[:70]}")
        print(f"\n{len(docs)} document(s)")
    finally:
        db.close()


def fix_stuck_docs(args=None):
    from datetime import datetime, timedelta
    from app.models.pdf_document import PDFDocument
    db = SessionLocal()
    try:
        # Documents stuck in 'processing' for more than 30 minutes
        cutoff = datetime.utcnow() - timedelta(minutes=30)
        stuck = db.query(PDFDocument).filter(
            PDFDocument.status == "processing",
            PDFDocument.uploaded_at < cutoff,
        ).all()
        if not stuck:
            print("No stuck documents found.")
            return
        for d in stuck:
            d.status = "ready"
            d.error_message = "Reset by admin — was stuck in processing"
            print(f"  Reset doc {d.id}: {d.original_filename}")
        db.commit()
        print(f"\nReset {len(stuck)} stuck document(s) to 'ready'")
    finally:
        db.close()


def reprocess_doc(doc_id: str):
    from app.models.pdf_document import PDFDocument
    db = SessionLocal()
    try:
        doc = db.query(PDFDocument).filter(PDFDocument.id == int(doc_id)).first()
        if not doc:
            print(f"Error: No document with id {doc_id}")
            return False
        print(f"Document: {doc.original_filename} (status: {doc.status})")
        if doc.status == "processing":
            print("Warning: Document is currently processing. Resetting first.")
        doc.status = "processing"
        doc.error_message = None
        db.commit()
        # Queue the Celery task
        from app.tasks.pdf_tasks import process_pdf
        process_pdf.delay(doc.id)
        print(f"Queued for reprocessing (Celery task dispatched)")
        return True
    finally:
        db.close()


def cleanup_redis(args=None):
    from app.config import settings
    import redis as redis_lib
    r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True)
    keys = r.keys("quiz_progress:*")
    expired = 0
    for key in keys:
        ttl = r.ttl(key)
        if ttl == -1:  # no expiry set — orphaned
            r.delete(key)
            expired += 1
    session_keys = r.keys("quiz_session:*")
    for key in session_keys:
        ttl = r.ttl(key)
        if ttl == -1:
            r.delete(key)
            expired += 1
    print(f"Cleaned {expired} orphaned keys out of {len(keys) + len(session_keys)} total")


def export_users(args=None):
    db = SessionLocal()
    try:
        users = db.query(User).order_by(User.id).all()
        out = io.StringIO()
        writer = csv.writer(out)
        writer.writerow(["id", "name", "email", "role", "created_at"])
        for u in users:
            writer.writerow([u.id, u.name, u.email, u.role, u.created_at])
        print(out.getvalue())
    finally:
        db.close()


COMMANDS = {
    "list-users": (list_users, 0),
    "reset-password": (lambda args: reset_password(args[0], args[1]), 2),
    "set-role": (lambda args: set_role(args[0], args[1]), 2),
    "verify-email": (lambda args: verify_email(args[0]), 1),
    "delete-user": (lambda args: delete_user(args[0]), 1),
    "stats": (stats, 0),
    "list-quizzes": (list_quizzes, 0),
    "list-docs": (list_docs, 0),
    "fix-stuck-docs": (fix_stuck_docs, 0),
    "reprocess-doc": (lambda args: reprocess_doc(args[0]), 1),
    "cleanup-redis": (cleanup_redis, 0),
    "export-users": (export_users, 0),
}


if __name__ == "__main__":
    args = sys.argv[1:]
    if not args or args[0] in ("-h", "--help", "help"):
        print(__doc__)
        sys.exit(0)

    cmd = args[0]
    rest = args[1:]

    if cmd not in COMMANDS:
        print(f"Unknown command: {cmd}")
        print("Run with --help to see available commands")
        sys.exit(1)

    fn, min_args = COMMANDS[cmd]
    if len(rest) < min_args:
        print(f"Error: '{cmd}' requires {min_args} argument(s), got {len(rest)}")
        sys.exit(1)

    result = fn(rest) if min_args > 0 else fn(rest)
    if result is False:
        sys.exit(1)