feat(admin_routes): add admin routes for scheduler and sound management; refactor main routes
This commit is contained in:
@@ -68,10 +68,11 @@ def create_app():
|
|||||||
scheduler_service.start()
|
scheduler_service.start()
|
||||||
|
|
||||||
# Register blueprints
|
# 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(main.bp, url_prefix="/api")
|
||||||
app.register_blueprint(auth.bp, url_prefix="/api/auth")
|
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
|
# Shutdown scheduler when app is torn down
|
||||||
@app.teardown_appcontext
|
@app.teardown_appcontext
|
||||||
|
|||||||
95
app/routes/admin.py
Normal file
95
app/routes/admin.py
Normal file
@@ -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/<int:sound_id>", 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()
|
||||||
@@ -1,16 +1,12 @@
|
|||||||
"""Main routes for the application."""
|
"""Main routes for the application."""
|
||||||
|
|
||||||
from flask import Blueprint, request
|
from flask import Blueprint
|
||||||
|
|
||||||
from app.services.decorators import (
|
from app.services.decorators import (
|
||||||
get_current_user,
|
get_current_user,
|
||||||
require_auth,
|
require_auth,
|
||||||
require_credits,
|
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__)
|
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")
|
@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/<int:sound_id>", 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()
|
|
||||||
|
|||||||
Reference in New Issue
Block a user