diff --git a/app/__init__.py b/app/__init__.py index 81d2c50..511cce3 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -68,10 +68,11 @@ def create_app(): scheduler_service.start() # Register blueprints - from app.routes import auth, main + from app.routes import admin, auth, main app.register_blueprint(main.bp, url_prefix="/api") app.register_blueprint(auth.bp, url_prefix="/api/auth") + app.register_blueprint(admin.bp, url_prefix="/api/admin") # Shutdown scheduler when app is torn down @app.teardown_appcontext diff --git a/app/routes/admin.py b/app/routes/admin.py new file mode 100644 index 0000000..85751b9 --- /dev/null +++ b/app/routes/admin.py @@ -0,0 +1,95 @@ +"""Admin routes for the application.""" + +from flask import Blueprint, request + +from app.services.decorators import ( + get_current_user, + require_auth, + require_role, +) +from app.services.scheduler_service import scheduler_service +from app.services.sound_normalizer_service import SoundNormalizerService +from app.services.sound_scanner_service import SoundScannerService + +bp = Blueprint("admin", __name__) + + +@bp.route("/") +@require_auth +@require_role("admin") +def admin_only() -> dict[str, str]: + """Admin-only endpoint to demonstrate role-based access.""" + user = get_current_user() + return { + "message": f"Hello admin {user['name']}, you have admin access!", + "user": user, + "admin_info": "This endpoint is only accessible to admin users", + } + + +@bp.route("/scheduler/status") +@require_auth +@require_role("admin") +def scheduler_status() -> dict: + """Get scheduler status (admin only).""" + return scheduler_service.get_scheduler_status() + + +@bp.route("/credits/refill", methods=["POST"]) +@require_auth +@require_role("admin") +def manual_credit_refill() -> dict: + """Manually trigger credit refill for all users (admin only).""" + return scheduler_service.trigger_credit_refill_now() + + +@bp.route("/sounds/scan", methods=["POST"]) +@require_auth +@require_role("admin") +def manual_sound_scan() -> dict: + """Manually trigger sound directory scan (admin only).""" + return scheduler_service.trigger_sound_scan_now() + + +@bp.route("/sounds/stats") +@require_auth +@require_role("admin") +def sound_statistics() -> dict: + """Get sound database statistics (admin only).""" + return SoundScannerService.get_scan_statistics() + + +@bp.route("/sounds/normalize/", methods=["POST"]) +@require_auth +@require_role("admin") +def normalize_sound(sound_id: int) -> dict: + """Normalize a specific sound file (admin only).""" + overwrite = request.args.get("overwrite", "false").lower() == "true" + return SoundNormalizerService.normalize_sound(sound_id, overwrite) + + +@bp.route("/sounds/normalize-all", methods=["POST"]) +@require_auth +@require_role("admin") +def normalize_all_sounds() -> dict: + """Normalize all soundboard files (admin only).""" + overwrite = request.args.get("overwrite", "false").lower() == "true" + limit_str = request.args.get("limit") + limit = int(limit_str) if limit_str else None + return SoundNormalizerService.normalize_all_sounds(overwrite, limit) + + +@bp.route("/sounds/normalization-status") +@require_auth +@require_role("admin") +def normalization_status() -> dict: + """Get normalization status statistics (admin only).""" + return SoundNormalizerService.get_normalization_status() + + +@bp.route("/sounds/ffmpeg-check") +@require_auth +@require_role("admin") +def ffmpeg_check() -> dict: + """Check ffmpeg availability and capabilities (admin only).""" + return SoundNormalizerService.check_ffmpeg_availability() \ No newline at end of file diff --git a/app/routes/main.py b/app/routes/main.py index 26bf73a..20d503a 100644 --- a/app/routes/main.py +++ b/app/routes/main.py @@ -1,16 +1,12 @@ """Main routes for the application.""" -from flask import Blueprint, request +from flask import Blueprint from app.services.decorators import ( get_current_user, require_auth, require_credits, - require_role, ) -from app.services.scheduler_service import scheduler_service -from app.services.sound_normalizer_service import SoundNormalizerService -from app.services.sound_scanner_service import SoundScannerService bp = Blueprint("main", __name__) @@ -43,17 +39,6 @@ def api_protected() -> dict[str, str]: } -@bp.route("/admin") -@require_auth -@require_role("admin") -def admin_only() -> dict[str, str]: - """Admin-only endpoint to demonstrate role-based access.""" - user = get_current_user() - return { - "message": f"Hello admin {user['name']}, you have admin access!", - "user": user, - "admin_info": "This endpoint is only accessible to admin users", - } @bp.route("/health") @@ -89,69 +74,3 @@ def expensive_operation() -> dict[str, str]: } -@bp.route("/admin/scheduler/status") -@require_auth -@require_role("admin") -def scheduler_status() -> dict: - """Get scheduler status (admin only).""" - return scheduler_service.get_scheduler_status() - - -@bp.route("/admin/credits/refill", methods=["POST"]) -@require_auth -@require_role("admin") -def manual_credit_refill() -> dict: - """Manually trigger credit refill for all users (admin only).""" - return scheduler_service.trigger_credit_refill_now() - - -@bp.route("/admin/sounds/scan", methods=["POST"]) -@require_auth -@require_role("admin") -def manual_sound_scan() -> dict: - """Manually trigger sound directory scan (admin only).""" - return scheduler_service.trigger_sound_scan_now() - - -@bp.route("/admin/sounds/stats") -@require_auth -@require_role("admin") -def sound_statistics() -> dict: - """Get sound database statistics (admin only).""" - return SoundScannerService.get_scan_statistics() - - -@bp.route("/admin/sounds/normalize/", methods=["POST"]) -@require_auth -@require_role("admin") -def normalize_sound(sound_id: int) -> dict: - """Normalize a specific sound file (admin only).""" - overwrite = request.args.get("overwrite", "false").lower() == "true" - return SoundNormalizerService.normalize_sound(sound_id, overwrite) - - -@bp.route("/admin/sounds/normalize-all", methods=["POST"]) -@require_auth -@require_role("admin") -def normalize_all_sounds() -> dict: - """Normalize all soundboard files (admin only).""" - overwrite = request.args.get("overwrite", "false").lower() == "true" - limit_str = request.args.get("limit") - limit = int(limit_str) if limit_str else None - return SoundNormalizerService.normalize_all_sounds(overwrite, limit) - - -@bp.route("/admin/sounds/normalization-status") -@require_auth -@require_role("admin") -def normalization_status() -> dict: - """Get normalization status statistics (admin only).""" - return SoundNormalizerService.get_normalization_status() - - -@bp.route("/admin/sounds/ffmpeg-check") -@require_auth -@require_role("admin") -def ffmpeg_check() -> dict: - """Check ffmpeg availability and capabilities (admin only).""" - return SoundNormalizerService.check_ffmpeg_availability()