339 lines
12 KiB
Python
339 lines
12 KiB
Python
"""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-courses List all courses with status
|
|
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), Quiz.course_id.is_(None)).count()
|
|
course_quizzes = db.query(Quiz).filter(Quiz.deleted_at.is_(None), Quiz.course_id.isnot(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()
|
|
|
|
try:
|
|
from app.models.course import Course, CourseEnrollment
|
|
courses = db.query(Course).count()
|
|
enrollments = db.query(CourseEnrollment).count()
|
|
except Exception:
|
|
courses = enrollments = "N/A"
|
|
|
|
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} (+{course_quizzes} course)")
|
|
print(f"Attempts: {attempts} completed, {in_progress} in-progress")
|
|
print(f"Flashcard decks: {decks} ({cards} cards)")
|
|
print(f"Courses: {courses}")
|
|
print(f"Enrollments: {enrollments}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def list_courses(args=None):
|
|
from app.models.course import Course, CourseEnrollment
|
|
from sqlalchemy import func
|
|
db = SessionLocal()
|
|
try:
|
|
counts = dict(db.query(CourseEnrollment.course_id, func.count()).group_by(CourseEnrollment.course_id).all())
|
|
courses = db.query(Course).order_by(Course.id).all()
|
|
print(f"{'ID':<5} {'Title':<35} {'Status':<12} {'Creator':<20} {'Enrolled':<10}")
|
|
print("-" * 82)
|
|
for c in courses:
|
|
creator = db.query(User).filter(User.id == c.user_id).first()
|
|
name = creator.name if creator else "?"
|
|
enrolled = counts.get(c.id, 0)
|
|
print(f"{c.id:<5} {(c.title or '-')[:34]:<35} {c.status:<12} {name[:19]:<20} {enrolled:<10}")
|
|
print(f"\n{len(courses)} course(s)")
|
|
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} {'Type':<10} {'Status':<10}")
|
|
print("-" * 75)
|
|
for qz in quizzes:
|
|
qtype = "course" if qz.course_id else "standalone"
|
|
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} {qtype:<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-courses": (list_courses, 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)
|