Files
sdb-back/app/routes/soundboard.py

239 lines
6.5 KiB
Python

"""Soundboard routes."""
from flask import Blueprint, jsonify, request
from app.models.sound import Sound, SoundType
from app.models.sound_played import SoundPlayed
from app.services.decorators import (
get_current_user,
require_auth,
require_credits,
)
from app.services.vlc_service import vlc_service
bp = Blueprint("soundboard", __name__, url_prefix="/api/soundboard")
@bp.route("/sounds", methods=["GET"])
@require_auth
def get_sounds():
"""Get all soundboard sounds."""
try:
# Get filter parameters
sound_type = request.args.get("type", "SDB")
# Validate sound type
if sound_type not in [t.value for t in SoundType]:
return jsonify({"error": "Invalid sound type"}), 400
# Get sounds from database
sounds = Sound.find_by_type(sound_type)
# Convert to dict format
sounds_data = [sound.to_dict() for sound in sounds]
return jsonify(
{
"sounds": sounds_data,
"total": len(sounds_data),
"type": sound_type,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/sounds/<int:sound_id>/play", methods=["POST"])
@require_auth
@require_credits(1)
def play_sound(sound_id: int):
"""Play a specific sound."""
try:
# Get current user for tracking
user = get_current_user()
user_id = int(user["id"]) if user else None
success = vlc_service.play_sound(
sound_id=sound_id,
user_id=user_id,
)
if success:
return jsonify({"message": "Sound playing", "sound_id": sound_id})
return (
jsonify({"error": "Sound not found or cannot be played"}),
404,
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/stop-all", methods=["POST"])
@require_auth
def stop_all_sounds():
"""Stop all currently playing sounds."""
try:
# Try normal stop first
vlc_service.stop_all()
# Wait a moment and check if any are still playing
import time
time.sleep(0.2)
# If there are still instances, force stop them
if vlc_service.get_playing_count() > 0:
stopped_count = vlc_service.force_stop_all()
return jsonify(
{
"message": f"Force stopped {stopped_count} sounds",
"forced": True,
},
)
return jsonify({"message": "All sounds stopped"})
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/force-stop", methods=["POST"])
@require_auth
def force_stop_all_sounds():
"""Force stop all sounds with aggressive cleanup."""
try:
stopped_count = vlc_service.force_stop_all()
return jsonify(
{
"message": f"Force stopped {stopped_count} sound instances",
"stopped_count": stopped_count,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/status", methods=["GET"])
@require_auth
def get_status():
"""Get current playback status."""
try:
playing_count = vlc_service.get_playing_count()
# Get detailed process information
with vlc_service.lock:
processes = []
for process_id, process in vlc_service.processes.items():
processes.append(
{
"id": process_id,
"pid": process.pid,
"running": process.poll() is None,
},
)
return jsonify(
{
"playing_count": playing_count,
"is_playing": playing_count > 0,
"processes": processes,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/history", methods=["GET"])
@require_auth
def get_play_history():
"""Get recent play history."""
try:
page = int(request.args.get("page", 1))
per_page = min(int(request.args.get("per_page", 50)), 100)
offset = (page - 1) * per_page
recent_plays = SoundPlayed.get_recent_plays(
limit=per_page,
offset=offset,
)
return jsonify(
{
"plays": [play.to_dict() for play in recent_plays],
"page": page,
"per_page": per_page,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/my-history", methods=["GET"])
@require_auth
def get_my_play_history():
"""Get current user's play history."""
try:
user = get_current_user()
if not user:
return jsonify({"error": "User not found"}), 404
user_id = int(user["id"])
page = int(request.args.get("page", 1))
per_page = min(int(request.args.get("per_page", 50)), 100)
offset = (page - 1) * per_page
user_plays = SoundPlayed.get_user_plays(
user_id=user_id,
limit=per_page,
offset=offset,
)
return jsonify(
{
"plays": [play.to_dict() for play in user_plays],
"page": page,
"per_page": per_page,
"user_id": user_id,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/my-stats", methods=["GET"])
@require_auth
def get_my_stats():
"""Get current user's play statistics."""
try:
user = get_current_user()
if not user:
return jsonify({"error": "User not found"}), 404
user_id = int(user["id"])
stats = SoundPlayed.get_user_stats(user_id)
return jsonify(stats)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/popular", methods=["GET"])
@require_auth
def get_popular_sounds():
"""Get most popular sounds."""
try:
limit = min(int(request.args.get("limit", 10)), 50)
days = request.args.get("days")
days = int(days) if days and days.isdigit() else None
popular_sounds = SoundPlayed.get_popular_sounds(limit=limit, days=days)
return jsonify(
{
"popular_sounds": popular_sounds,
"limit": limit,
"days": days,
},
)
except Exception as e:
return jsonify({"error": str(e)}), 500