feat: Replace pydub with ffmpeg for audio duration and metadata extraction in sound services

app/__init__.py

@@ -100,6 +100,7 @@ def create_app():
         auth,
         main,
         player,
+        referential,
         soundboard,
         sounds,
         stream,
@@ -109,6 +110,7 @@ def create_app():
     app.register_blueprint(auth.bp, url_prefix="/api/auth")
     app.register_blueprint(admin.bp, url_prefix="/api/admin")
     app.register_blueprint(admin_sounds.bp, url_prefix="/api/admin/sounds")
+    app.register_blueprint(referential.bp, url_prefix="/api/referential")
     app.register_blueprint(soundboard.bp, url_prefix="/api/soundboard")
     app.register_blueprint(sounds.bp, url_prefix="/api/sounds")
     app.register_blueprint(stream.bp, url_prefix="/api/stream")

app/routes/admin.py

@@ -37,3 +37,139 @@ def manual_credit_refill() -> dict:
     return scheduler_service.trigger_credit_refill_now()
 
 
+@bp.route("/users")
+@require_auth
+@require_role("admin")
+def list_users() -> dict:
+    """List all users (admin only)."""
+    from app.models.user import User
+    
+    users = User.query.order_by(User.created_at.desc()).all()
+    return {
+        "users": [user.to_dict() for user in users],
+        "total": len(users)
+    }
+
+
+@bp.route("/users/<int:user_id>", methods=["PATCH"])
+@require_auth
+@require_role("admin")
+def update_user(user_id: int) -> dict:
+    """Update user information (admin only)."""
+    from flask import request
+    from app.database import db
+    from app.models.user import User
+    from app.models.plan import Plan
+    
+    data = request.get_json()
+    if not data:
+        return {"error": "No data provided"}, 400
+    
+    user = User.query.get(user_id)
+    if not user:
+        return {"error": "User not found"}, 404
+    
+    # Validate and update fields
+    try:
+        if "name" in data:
+            name = data["name"].strip()
+            if not name:
+                return {"error": "Name cannot be empty"}, 400
+            if len(name) > 100:
+                return {"error": "Name too long (max 100 characters)"}, 400
+            user.name = name
+        
+        if "credits" in data:
+            credits = data["credits"]
+            if not isinstance(credits, int) or credits < 0:
+                return {"error": "Credits must be a non-negative integer"}, 400
+            user.credits = credits
+        
+        if "plan_id" in data:
+            plan_id = data["plan_id"]
+            if not isinstance(plan_id, int):
+                return {"error": "Plan ID must be an integer"}, 400
+            
+            plan = Plan.query.get(plan_id)
+            if not plan:
+                return {"error": "Plan not found"}, 404
+            
+            user.plan_id = plan_id
+        
+        if "is_active" in data:
+            is_active = data["is_active"]
+            if not isinstance(is_active, bool):
+                return {"error": "is_active must be a boolean"}, 400
+            user.is_active = is_active
+        
+        db.session.commit()
+        
+        return {
+            "message": "User updated successfully",
+            "user": user.to_dict()
+        }
+        
+    except Exception as e:
+        db.session.rollback()
+        return {"error": f"Failed to update user: {str(e)}"}, 500
+
+
+@bp.route("/users/<int:user_id>/deactivate", methods=["POST"])
+@require_auth
+@require_role("admin")
+def deactivate_user(user_id: int) -> dict:
+    """Deactivate a user (admin only)."""
+    from app.database import db
+    from app.models.user import User
+    
+    user = User.query.get(user_id)
+    if not user:
+        return {"error": "User not found"}, 404
+    
+    # Prevent admin from deactivating themselves
+    current_user = get_current_user()
+    if str(user.id) == current_user["id"]:
+        return {"error": "Cannot deactivate your own account"}, 400
+    
+    try:
+        user.deactivate()
+        db.session.commit()
+        
+        return {
+            "message": "User deactivated successfully",
+            "user": user.to_dict()
+        }
+        
+    except Exception as e:
+        db.session.rollback()
+        return {"error": f"Failed to deactivate user: {str(e)}"}, 500
+
+
+@bp.route("/users/<int:user_id>/activate", methods=["POST"])
+@require_auth
+@require_role("admin")
+def activate_user(user_id: int) -> dict:
+    """Activate a user (admin only)."""
+    from app.database import db
+    from app.models.user import User
+    
+    user = User.query.get(user_id)
+    if not user:
+        return {"error": "User not found"}, 404
+    
+    try:
+        user.activate()
+        db.session.commit()
+        
+        return {
+            "message": "User activated successfully",
+            "user": user.to_dict()
+        }
+        
+    except Exception as e:
+        db.session.rollback()
+        return {"error": f"Failed to activate user: {str(e)}"}, 500
+
+
+
+

app/routes/main.py

@@ -1,6 +1,17 @@
 """Main routes for the application."""
 
-from flask import Blueprint
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from flask import Blueprint, request
+from sqlalchemy import desc, func
+
+from app.database import db
+from app.models.playlist import Playlist
+from app.models.sound import Sound
+from app.models.sound_played import SoundPlayed
+from app.models.user import User
+from app.services.decorators import require_auth
 
 bp = Blueprint("main", __name__)
 
@@ -9,3 +20,213 @@ bp = Blueprint("main", __name__)
 def health() -> dict[str, str]:
     """Health check endpoint."""
     return {"status": "ok"}
+
+
+def get_period_filter(period: str) -> datetime | None:
+    """Get the start date for the specified period."""
+    now = datetime.now(tz=ZoneInfo("UTC"))
+
+    if period == "today":
+        return now.replace(hour=0, minute=0, second=0, microsecond=0)
+    if period == "week":
+        return now - timedelta(days=7)
+    if period == "month":
+        return now - timedelta(days=30)
+    if period == "year":
+        return now - timedelta(days=365)
+    if period == "all":
+        return None
+    # Default to all time
+    return None
+
+
+@bp.route("/dashboard/stats")
+@require_auth
+def dashboard_stats() -> dict:
+    """Get dashboard statistics."""
+    # Count soundboard sounds (type = SDB)
+    soundboard_count = Sound.query.filter_by(type="SDB").count()
+
+    # Count tracks (type = STR)
+    track_count = Sound.query.filter_by(type="STR").count()
+
+    # Count playlists
+    playlist_count = Playlist.query.count()
+
+    # Calculate total size of all sounds (original + normalized)
+    total_size_result = db.session.query(
+        func.sum(Sound.size).label("original_size"),
+        func.sum(Sound.normalized_size).label("normalized_size"),
+    ).first()
+
+    original_size = getattr(total_size_result, "original_size", 0) or 0
+    normalized_size = getattr(total_size_result, "normalized_size", 0) or 0
+    total_size = original_size + normalized_size
+
+    return {
+        "soundboard_sounds": soundboard_count,
+        "tracks": track_count,
+        "playlists": playlist_count,
+        "total_size": total_size,
+        "original_size": original_size,
+        "normalized_size": normalized_size,
+    }
+
+
+@bp.route("/dashboard/top-sounds")
+@require_auth
+def top_sounds() -> dict:
+    """Get top played sounds for a specific period."""
+    period = request.args.get("period", "all")
+    limit = int(request.args.get("limit", 5))
+
+    period_start = get_period_filter(period)
+
+    # Base query for soundboard sounds with play counts
+    query = (
+        db.session.query(
+            Sound.id,
+            Sound.name,
+            Sound.filename,
+            Sound.thumbnail,
+            Sound.type,
+            func.count(SoundPlayed.id).label("play_count"),
+        )
+        .outerjoin(SoundPlayed, Sound.id == SoundPlayed.sound_id)
+        .filter(Sound.type == "SDB")  # Only soundboard sounds
+        .group_by(
+            Sound.id,
+            Sound.name,
+            Sound.filename,
+            Sound.thumbnail,
+            Sound.type,
+        )
+    )
+
+    # Apply period filter if specified
+    if period_start:
+        query = query.filter(SoundPlayed.played_at >= period_start)
+
+    # Order by play count and limit results
+    results = query.order_by(desc("play_count")).limit(limit).all()
+
+    # Convert to list of dictionaries
+    top_sounds_list = [
+        {
+            "id": result.id,
+            "name": result.name,
+            "filename": result.filename,
+            "thumbnail": result.thumbnail,
+            "type": result.type,
+            "play_count": result.play_count,
+        }
+        for result in results
+    ]
+
+    return {
+        "period": period,
+        "sounds": top_sounds_list,
+    }
+
+
+@bp.route("/dashboard/top-tracks")
+@require_auth
+def top_tracks() -> dict:
+    """Get top played tracks for a specific period."""
+    period = request.args.get("period", "all")
+    limit = int(request.args.get("limit", 10))
+
+    period_start = get_period_filter(period)
+
+    # Base query for tracks with play counts
+    query = (
+        db.session.query(
+            Sound.id,
+            Sound.name,
+            Sound.filename,
+            Sound.thumbnail,
+            Sound.type,
+            func.count(SoundPlayed.id).label("play_count"),
+        )
+        .outerjoin(SoundPlayed, Sound.id == SoundPlayed.sound_id)
+        .filter(Sound.type == "STR")  # Only tracks
+        .group_by(
+            Sound.id,
+            Sound.name,
+            Sound.filename,
+            Sound.thumbnail,
+            Sound.type,
+        )
+    )
+
+    # Apply period filter if specified
+    if period_start:
+        query = query.filter(SoundPlayed.played_at >= period_start)
+
+    # Order by play count and limit results
+    results = query.order_by(desc("play_count")).limit(limit).all()
+
+    # Convert to list of dictionaries
+    top_tracks_list = [
+        {
+            "id": result.id,
+            "name": result.name,
+            "filename": result.filename,
+            "thumbnail": result.thumbnail,
+            "type": result.type,
+            "play_count": result.play_count,
+        }
+        for result in results
+    ]
+
+    return {
+        "period": period,
+        "tracks": top_tracks_list,
+    }
+
+
+@bp.route("/dashboard/top-users")
+@require_auth
+def top_users() -> dict:
+    """Get top users by play count for a specific period."""
+    period = request.args.get("period", "all")
+    limit = int(request.args.get("limit", 10))
+
+    period_start = get_period_filter(period)
+
+    # Base query for users with play counts
+    query = (
+        db.session.query(
+            User.id,
+            User.name,
+            User.email,
+            User.picture,
+            func.count(SoundPlayed.id).label("play_count"),
+        )
+        .outerjoin(SoundPlayed, User.id == SoundPlayed.user_id)
+        .group_by(User.id, User.name, User.email, User.picture)
+    )
+
+    # Apply period filter if specified
+    if period_start:
+        query = query.filter(SoundPlayed.played_at >= period_start)
+
+    # Order by play count and limit results
+    results = query.order_by(desc("play_count")).limit(limit).all()
+
+    # Convert to list of dictionaries
+    top_users_list = [
+        {
+            "id": result.id,
+            "name": result.name,
+            "email": result.email,
+            "picture": result.picture,
+            "play_count": result.play_count,
+        }
+        for result in results
+    ]
+
+    return {
+        "period": period,
+        "users": top_users_list,
+    }

app/routes/referential.py

@@ -0,0 +1,17 @@
+"""Referential routes for reference data."""
+
+from flask import Blueprint
+
+bp = Blueprint("referential", __name__)
+
+
+@bp.route("/plans")
+def list_plans() -> dict:
+    """List all available plans."""
+    from app.models.plan import Plan
+    
+    plans = Plan.query.order_by(Plan.id).all()
+    return {
+        "plans": [plan.to_dict() for plan in plans],
+        "total": len(plans)
+    }

app/routes/soundboard.py

@@ -29,6 +29,9 @@ def get_sounds():
         # Get sounds from database
         sounds = Sound.find_by_type(sound_type)
 
+        # Order by name
+        sounds = sorted(sounds, key=lambda s: s.name.lower())
+
         # Convert to dict format
         sounds_data = [sound.to_dict() for sound in sounds]
 
@@ -59,6 +62,25 @@ def play_sound(sound_id: int):
         )
 
         if success:
+            # Get updated sound data to emit the new play count
+            sound = Sound.query.get(sound_id)
+            if sound:
+                # Emit sound_changed event to all connected clients
+                try:
+                    from app.services.socketio_service import SocketIOService
+
+                    SocketIOService.emit_sound_play_count_changed(
+                        sound_id, sound.play_count
+                    )
+                except Exception as e:
+                    # Don't fail the request if socket emission fails
+                    import logging
+
+                    logger = logging.getLogger(__name__)
+                    logger.warning(
+                        f"Failed to emit sound_play_count_changed event: {e}"
+                    )
+
             return jsonify({"message": "Sound playing", "sound_id": sound_id})
         return (
             jsonify({"error": "Sound not found or cannot be played"}),

app/services/credit_service.py

@@ -106,34 +106,3 @@ class CreditService:
                 "error": str(e),
                 "message": "Credit refill failed",
             }
-
-    @staticmethod
-    def get_user_credit_info(user_id: int) -> dict:
-        """Get detailed credit information for a specific user.
-
-        Args:
-            user_id: The user's ID
-
-        Returns:
-            dict: User's credit information
-
-        """
-        user = User.query.get(user_id)
-        if not user:
-            return {"error": "User not found"}
-
-        if not user.plan:
-            return {"error": "User has no plan assigned"}
-
-        return {
-            "user_id": user.id,
-            "email": user.email,
-            "current_credits": user.credits,
-            "plan": {
-                "code": user.plan.code,
-                "name": user.plan.name,
-                "daily_credits": user.plan.credits,
-                "max_credits": user.plan.max_credits,
-            },
-            "is_active": user.is_active,
-        }

app/services/decorators.py

@@ -172,6 +172,22 @@ def require_credits(credits_needed: int):
 
             # Check if user has enough credits
             if user.credits < credits_needed:
+                # Emit credits required event via SocketIO
+                try:
+                    from app.services.socketio_service import socketio_service
+
+                    socketio_service.emit_credits_required(
+                        user.id, credits_needed
+                    )
+                except Exception as e:
+                    # Don't fail the request if SocketIO emission fails
+                    import logging
+
+                    logger = logging.getLogger(__name__)
+                    logger.warning(
+                        f"Failed to emit credits_required event: {e}"
+                    )
+
                 return (
                     jsonify(
                         {

app/services/music_player_service.py

@@ -167,6 +167,20 @@ class MusicPlayerService:
             # Fallback if request context is not available
             return f"http://localhost:5000/api/sounds/{sound_type.lower()}/thumbnails/{thumbnail_filename}"
 
+    def _build_stream_url(self, sound_type: str, filename: str) -> str:
+        """Build absolute stream URL."""
+        try:
+            # Try to get base URL from current request context
+            if request:
+                base_url = request.url_root.rstrip("/")
+            else:
+                # Fallback to localhost if no request context
+                base_url = "http://localhost:5000"
+            return f"{base_url}/api/sounds/{sound_type.lower()}/audio/{filename}"
+        except Exception:
+            # Fallback if request context is not available
+            return f"http://localhost:5000/api/sounds/{sound_type.lower()}/audio/{filename}"
+
     def _load_playlist_with_context(
         self, playlist, reload: bool = False
     ) -> bool:
@@ -276,21 +290,13 @@ class MusicPlayerService:
     def _get_sound_file_path(self, sound: Sound) -> Optional[str]:
         """Get the file path for a sound, preferring normalized version."""
         try:
-            if sound.type == "STR":
-                # Stream sounds
             base_path = "sounds/stream"
-            elif sound.type == "SAY":
+            base_normalized_path = "sounds/normalized/stream"
-                # Say sounds
-                base_path = "sounds/say"
-            else:
-                # Soundboard sounds
-                base_path = "sounds/soundboard"
 
             # Check for normalized version first
             if sound.is_normalized and sound.normalized_filename:
                 normalized_path = os.path.join(
-                    "sounds/normalized",
+                    base_normalized_path,
-                    sound.type.lower(),
                     sound.normalized_filename,
                 )
                 if os.path.exists(normalized_path):
@@ -507,6 +513,12 @@ class MusicPlayerService:
                         sound = current_playlist_sound.sound
 
                         if sound:
+                            # Get the service URL from the associated stream
+                            service_url = None
+                            if sound.streams:
+                                # Get the first stream's URL if available
+                                service_url = sound.streams[0].url
+                            
                             return {
                                 "id": sound.id,
                                 "title": sound.name,
@@ -519,6 +531,8 @@ class MusicPlayerService:
                                     if sound.thumbnail
                                     else None
                                 ),
+                                "file_url": self._build_stream_url(sound.type, sound.filename),
+                                "service_url": service_url,
                                 "type": sound.type,
                             }
             return None
@@ -543,6 +557,12 @@ class MusicPlayerService:
                         ):
                             sound = playlist_sound.sound
                             if sound:
+                                # Get the service URL from the associated stream
+                                service_url = None
+                                if sound.streams:
+                                    # Get the first stream's URL if available
+                                    service_url = sound.streams[0].url
+                                
                                 tracks.append(
                                     {
                                         "id": sound.id,
@@ -556,6 +576,8 @@ class MusicPlayerService:
                                             if sound.thumbnail
                                             else None
                                         ),
+                                        "file_url": self._build_stream_url(sound.type, sound.filename),
+                                        "service_url": service_url,
                                         "type": sound.type,
                                     }
                                 )

app/services/socketio_service.py

@@ -6,6 +6,7 @@ from flask import request
 from flask_socketio import disconnect, emit, join_room, leave_room
 
 from app import socketio
+from app.services.decorators import require_credits
 
 logger = logging.getLogger(__name__)
 
@@ -30,7 +31,9 @@ class SocketIOService:
         """Emit an event to all connected clients."""
         try:
             socketio.emit(event, data)
-            logger.info(f"Successfully emitted {event} to all clients with data keys: {list(data.keys())}")
+            logger.info(
+                f"Successfully emitted {event} to all clients with data keys: {list(data.keys())}"
+            )
         except Exception as e:
             logger.error(f"Failed to emit {event}: {e}")
 
@@ -43,6 +46,23 @@ class SocketIOService:
             {"credits": new_credits},
         )
 
+    @staticmethod
+    def emit_sound_play_count_changed(sound_id: int, new_play_count: int) -> None:
+        """Emit sound_play_count_changed event to all connected clients."""
+        SocketIOService.emit_to_all(
+            "sound_play_count_changed",
+            {"sound_id": sound_id, "play_count": new_play_count},
+        )
+
+    @staticmethod
+    def emit_credits_required(user_id: int, credits_needed: int) -> None:
+        """Emit an event when credits are required."""
+        SocketIOService.emit_to_user(
+            user_id,
+            "credits_required",
+            {"credits_needed": credits_needed},
+        )
+
     @staticmethod
     def get_user_from_socketio() -> dict | None:
         """Get user from SocketIO connection using cookies."""
@@ -52,7 +72,9 @@ class SocketIOService:
 
             # Check if we have the access_token cookie
             access_token = request.cookies.get("access_token_cookie")
-            logger.debug(f"Access token from cookies: {access_token[:20] if access_token else None}...")
+            logger.debug(
+                f"Access token from cookies: {access_token[:20] if access_token else None}..."
+            )
 
             if not access_token:
                 logger.debug("No access token found in cookies")
@@ -77,7 +99,9 @@ class SocketIOService:
 
                 user = User.query.get(int(current_user_id))
                 if not user or not user.is_active:
-                    logger.debug(f"User not found or inactive: {current_user_id}")
+                    logger.debug(
+                        f"User not found or inactive: {current_user_id}"
+                    )
                     return None
 
                 logger.debug(f"Successfully found user: {user.email}")
@@ -97,7 +121,9 @@ class SocketIOService:
 def handle_connect(auth=None):
     """Handle client connection."""
     try:
-        logger.info(f"SocketIO connection established from {request.remote_addr}")
+        logger.info(
+            f"SocketIO connection established from {request.remote_addr}"
+        )
         logger.info(f"Session ID: {request.sid}")
 
     except Exception:
@@ -113,7 +139,7 @@ def handle_authenticate(data):
 
         if not user:
             logger.warning("SocketIO authentication failed - no user found")
-            emit("auth_error", {"error": "Authentication failed"})
+            # emit("auth_error", {"error": "Authentication failed"})
             disconnect()
             return
 
@@ -126,20 +152,56 @@ def handle_authenticate(data):
         logger.info(f"User {user_id} authenticated and joined room {user_room}")
 
         # Send current credits on authentication
-        emit("auth_success", {"user": user})
+        SocketIOService.emit_to_user(user_id, "auth_success", {"user": user})
-        emit("credits_changed", {"credits": user["credits"]})
+        SocketIOService.emit_to_user(
+            user_id, "credits_changed", {"credits": user["credits"]}
+        )
 
     except Exception:
         logger.exception("Error handling SocketIO authentication")
-        emit("auth_error", {"error": "Authentication failed"})
+        # emit("auth_error", {"error": "Authentication failed"})
         disconnect()
 
 
-@socketio.on("test_event")
+# @socketio.on("play_sound")
-def handle_test_event(data):
+# @require_credits(1)
-    """Test handler to verify SocketIO events are working."""
+# def handle_play_sound(data):
-    logger.debug(f"Test event received: {data}")
+#     """Handle play_sound event from client."""
-    emit("test_response", {"message": "Test event received successfully"})
+#     try:
+#         user = SocketIOService.get_user_from_socketio()
+
+#         if not user:
+#             logger.warning("SocketIO play_sound failed - no authenticated user")
+#             # emit("error", {"message": "Authentication required"})
+#             return
+
+#         user_id = int(user["id"])
+#         sound_id = data.get("soundId")
+#         if not sound_id:
+#             logger.warning("SocketIO play_sound failed - no soundId provided")
+#             SocketIOService.emit_to_user(
+#                 user_id, "error", {"message": "Sound ID required"}
+#             )
+#             return
+
+#         # Import and use the VLC service to play the sound
+#         from app.services.vlc_service import vlc_service
+
+#         logger.info(f"User {user_id} playing sound {sound_id} via SocketIO")
+
+#         # Play the sound using the VLC service
+#         success = vlc_service.play_sound(sound_id, user_id)
+
+#         if not success:
+#             SocketIOService.emit_to_user(
+#                 user_id,
+#                 "error",
+#                 {"message": f"Failed to play sound {sound_id}"},
+#             )
+
+#     except Exception as e:
+#         logger.exception(f"Error handling play_sound event: {e}")
+#         # emit("error", {"message": "Failed to play sound"})
 
 
 @socketio.on("disconnect")

app/services/sound_normalizer_service.py

@@ -7,7 +7,6 @@ import re
 from pathlib import Path
 
 import ffmpeg
-from pydub import AudioSegment
 
 from app.database import db
 from app.models.sound import Sound
@@ -632,9 +631,17 @@ class SoundNormalizerService:
             # Calculate file hash
             file_hash = SoundNormalizerService._calculate_file_hash(file_path)
 
-            # Get duration using pydub
+            # Get duration using ffmpeg
-            audio = AudioSegment.from_wav(file_path)
+            probe = ffmpeg.probe(file_path)
-            duration = len(audio)  # Duration in milliseconds
+            audio_stream = next(
+                (s for s in probe['streams'] if s['codec_type'] == 'audio'),
+                None
+            )
+            
+            if audio_stream and 'duration' in audio_stream:
+                duration = int(float(audio_stream['duration']) * 1000)  # Convert to milliseconds
+            else:
+                duration = 0
 
             return {
                 "duration": duration,

app/services/sound_scanner_service.py

@@ -4,8 +4,7 @@ import hashlib
 import logging
 from pathlib import Path
 
-from pydub import AudioSegment
+import ffmpeg
-from pydub.utils import mediainfo
 
 from app.database import db
 from app.models.sound import Sound
@@ -281,32 +280,31 @@ class SoundScannerService:
 
     @staticmethod
     def _extract_audio_metadata(file_path: str) -> dict:
-        """Extract metadata from audio file using pydub and mediainfo."""
+        """Extract metadata from audio file using ffmpeg-python."""
         try:
             # Get file size
             file_size = Path(file_path).stat().st_size
 
-            # Load audio file with pydub for basic info
+            # Use ffmpeg to probe audio metadata
-            audio = AudioSegment.from_file(file_path)
+            probe = ffmpeg.probe(file_path)
+            audio_stream = next(
+                (s for s in probe['streams'] if s['codec_type'] == 'audio'),
+                None
+            )
             
-            # Extract basic metadata from AudioSegment
+            if not audio_stream:
-            duration = len(audio)
+                raise ValueError("No audio stream found in file")
-            channels = audio.channels
-            sample_rate = audio.frame_rate
 
-            # Use mediainfo for more accurate bitrate information
+            # Extract metadata from ffmpeg probe
-            bitrate = None
+            duration = int(float(audio_stream.get('duration', 0)) * 1000)  # Convert to milliseconds
-            try:
+            channels = int(audio_stream.get('channels', 0))
-                info = mediainfo(file_path)
+            sample_rate = int(audio_stream.get('sample_rate', 0))
-                if info and "bit_rate" in info:
+            bitrate = int(audio_stream.get('bit_rate', 0)) if audio_stream.get('bit_rate') else None
-                    bitrate = int(info["bit_rate"])
+            
-                elif info and "bitrate" in info:
+            # Fallback bitrate calculation if not available
-                    bitrate = int(info["bitrate"])
+            if not bitrate and duration > 0:
-            except (ValueError, KeyError, TypeError):
-                # Fallback to calculated bitrate if mediainfo fails
-                if duration > 0:
                 file_size_bits = file_size * 8
-                    bitrate = int(file_size_bits / duration / 1000)
+                bitrate = int(file_size_bits / (duration / 1000))
 
             return {
                 "duration": duration,

pyproject.toml

@@ -15,7 +15,6 @@ dependencies = [
     "flask-migrate==4.1.0",
     "flask-socketio==5.5.1",
     "flask-sqlalchemy==3.1.1",
-    "pydub==0.25.1",
     "python-dotenv==1.1.1",
     "python-vlc>=3.0.21203",
     "requests==2.32.4",

uv.lock

@@ -505,15 +505,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/13/a3/a812df4e2dd5696d1f351d58b8fe16a405b234ad2886a0dab9183fb78109/pycparser-2.22-py3-none-any.whl", hash = "sha256:c3702b6d3dd8c7abc1afa565d7e63d53a1d0bd86cdc24edd75470f4de499cfcc", size = 117552 },
 ]
 
-[[package]]
-name = "pydub"
-version = "0.25.1"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/fe/9a/e6bca0eed82db26562c73b5076539a4a08d3cffd19c3cc5913a3e61145fd/pydub-0.25.1.tar.gz", hash = "sha256:980a33ce9949cab2a569606b65674d748ecbca4f0796887fd6f46173a7b0d30f", size = 38326 }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/a6/53/d78dc063216e62fc55f6b2eebb447f6a4b0a59f55c8406376f76bf959b08/pydub-0.25.1-py2.py3-none-any.whl", hash = "sha256:65617e33033874b59d87db603aa1ed450633288aefead953b30bded59cb599a6", size = 32327 },
-]
-
 [[package]]
 name = "pygments"
 version = "2.19.2"
@@ -645,7 +636,6 @@ dependencies = [
     { name = "flask-migrate" },
     { name = "flask-socketio" },
     { name = "flask-sqlalchemy" },
-    { name = "pydub" },
     { name = "python-dotenv" },
     { name = "python-vlc" },
     { name = "requests" },
@@ -671,7 +661,6 @@ requires-dist = [
     { name = "flask-migrate", specifier = "==4.1.0" },
     { name = "flask-socketio", specifier = "==5.5.1" },
     { name = "flask-sqlalchemy", specifier = "==3.1.1" },
-    { name = "pydub", specifier = "==0.25.1" },
     { name = "python-dotenv", specifier = "==1.1.1" },
     { name = "python-vlc", specifier = ">=3.0.21203" },
     { name = "requests", specifier = "==2.32.4" },