quaik8/sdb2-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Improve exception handling and logging in authentication and playlist services; enhance code readability and structure
All checks were successful
Backend CI / lint (push) Successful in 9m21s
Details
Backend CI / test (push) Successful in 4m18s
Details

Browse Source
This commit is contained in:
JSC
2025-08-13 00:04:55 +02:00
parent f094fbf140
commit bee1076239
14 changed files with 144 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
app/api/v1/auth.py
View File

@@ -181,7 +181,9 @@ async def logout(
user_id = int(user_id_str) user_id = int(user_id_str)
user = await auth_service.get_current_user(user_id) user = await auth_service.get_current_user(user_id)
logger.info("Found user from access token: %s", user.email) logger.info("Found user from access token: %s", user.email)
except (HTTPException, Exception) as e: except HTTPException as e:
logger.info("Access token validation failed: %s", str(e))
except Exception as e: # noqa: BLE001
logger.info("Access token validation failed: %s", str(e)) logger.info("Access token validation failed: %s", str(e))
# If no user found, try refresh token # If no user found, try refresh token
@@ -193,7 +195,9 @@ async def logout(
user_id = int(user_id_str) user_id = int(user_id_str)
user = await auth_service.get_current_user(user_id) user = await auth_service.get_current_user(user_id)
logger.info("Found user from refresh token: %s", user.email) logger.info("Found user from refresh token: %s", user.email)
except (HTTPException, Exception) as e: except HTTPException as e:
logger.info("Refresh token validation failed: %s", str(e))
except Exception as e: # noqa: BLE001
logger.info("Refresh token validation failed: %s", str(e)) logger.info("Refresh token validation failed: %s", str(e))
# If we found a user, revoke their refresh token # If we found a user, revoke their refresh token
@@ -484,7 +488,6 @@ async def change_password(
await auth_service.change_user_password( await auth_service.change_user_password(
current_user, request.current_password, request.new_password, current_user, request.current_password, request.new_password,
) )
return {"message": "Password changed successfully"}
except ValueError as e: except ValueError as e:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,
@@ -496,6 +499,8 @@ async def change_password(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to change password", detail="Failed to change password",
) from e ) from e
else:
return {"message": "Password changed successfully"}
@router.get("/user-providers") @router.get("/user-providers")

15
app/api/v1/dashboard.py
View File

@@ -33,9 +33,18 @@ async def get_track_statistics(
async def get_top_sounds( async def get_top_sounds(
_current_user: Annotated[User, Depends(get_current_user)], _current_user: Annotated[User, Depends(get_current_user)],
dashboard_service: Annotated[DashboardService, Depends(get_dashboard_service)], dashboard_service: Annotated[DashboardService, Depends(get_dashboard_service)],
sound_type: Annotated[str, Query(description="Sound type filter (SDB, TTS, EXT, or 'all')")], sound_type: Annotated[
period: Annotated[str, Query(description="Time period (today, 1_day, 1_week, 1_month, 1_year, all_time)")] = "all_time", str, Query(description="Sound type filter (SDB, TTS, EXT, or 'all')"),
limit: Annotated[int, Query(description="Number of top sounds to return", ge=1, le=100)] = 10, ],
period: Annotated[
str,
Query(
description="Time period (today, 1_day, 1_week, 1_month, 1_year, all_time)",
),
] = "all_time",
limit: Annotated[
int, Query(description="Number of top sounds to return", ge=1, le=100),
] = 10,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Get top sounds by play count for a specific period.""" """Get top sounds by play count for a specific period."""
return await dashboard_service.get_top_sounds( return await dashboard_service.get_top_sounds(

5
app/api/v1/main.py
View File

@@ -69,7 +69,10 @@ async def elements_docs() -> HTMLResponse:
<html lang="en"> <html lang="en">
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no"> <meta
name="viewport"
content="width=device-width, initial-scale=1, shrink-to-fit=no"
>
<title>API Documentation - elements</title> <title>API Documentation - elements</title>
<script src="https://unpkg.com/@stoplight/elements/web-components.min.js"></script> <script src="https://unpkg.com/@stoplight/elements/web-components.min.js"></script>
<link rel="stylesheet" href="https://unpkg.com/@stoplight/elements/styles.min.css"> <link rel="stylesheet" href="https://unpkg.com/@stoplight/elements/styles.min.css">

5
app/api/v1/playlists.py
View File

@@ -32,7 +32,7 @@ async def get_playlist_service(
@router.get("/") @router.get("/")
async def get_all_playlists( async def get_all_playlists( # noqa: PLR0913
current_user: Annotated[User, Depends(get_current_active_user_flexible)], # noqa: ARG001 current_user: Annotated[User, Depends(get_current_active_user_flexible)], # noqa: ARG001
playlist_service: Annotated[PlaylistService, Depends(get_playlist_service)], playlist_service: Annotated[PlaylistService, Depends(get_playlist_service)],
search: Annotated[ search: Annotated[
@@ -57,7 +57,7 @@ async def get_all_playlists(
] = 0, ] = 0,
) -> list[dict]: ) -> list[dict]:
"""Get all playlists from all users with search and sorting.""" """Get all playlists from all users with search and sorting."""
playlists = await playlist_service.search_and_sort_playlists( return await playlist_service.search_and_sort_playlists(
search_query=search, search_query=search,
sort_by=sort_by, sort_by=sort_by,
sort_order=sort_order, sort_order=sort_order,
@@ -66,7 +66,6 @@ async def get_all_playlists(
limit=limit, limit=limit,
offset=offset, offset=offset,
) )
return playlists
@router.get("/user") @router.get("/user")

2
app/api/v1/sounds.py
View File

@@ -35,7 +35,7 @@ async def get_sound_repository(
@router.get("/") @router.get("/")
async def get_sounds( async def get_sounds( # noqa: PLR0913
current_user: Annotated[User, Depends(get_current_active_user_flexible)], # noqa: ARG001 current_user: Annotated[User, Depends(get_current_active_user_flexible)], # noqa: ARG001
sound_repo: Annotated[SoundRepository, Depends(get_sound_repository)], sound_repo: Annotated[SoundRepository, Depends(get_sound_repository)],
types: Annotated[ types: Annotated[

5
app/main.py
View File

@@ -60,7 +60,10 @@ def create_app() -> FastAPI:
"""Create and configure the FastAPI application.""" """Create and configure the FastAPI application."""
app = FastAPI( app = FastAPI(
title="Soundboard API", title="Soundboard API",
description="API for the Soundboard application with authentication, sound management, and real-time features", description=(
"API for the Soundboard application with authentication, "
"sound management, and real-time features"
),
version="1.0.0", version="1.0.0",
lifespan=lifespan, lifespan=lifespan,
# Configure docs URLs for reverse proxy setup # Configure docs URLs for reverse proxy setup

74
app/repositories/playlist.py
View File

@@ -132,7 +132,9 @@ class PlaylistRepository(BaseRepository[Playlist]):
result = await self.session.exec(statement) result = await self.session.exec(statement)
return list(result.all()) return list(result.all())
except Exception: except Exception:
logger.exception("Failed to get playlist sound entries for playlist: %s", playlist_id) logger.exception(
"Failed to get playlist sound entries for playlist: %s", playlist_id,
)
raise raise
async def add_sound_to_playlist( async def add_sound_to_playlist(
@@ -302,19 +304,22 @@ class PlaylistRepository(BaseRepository[Playlist]):
) )
raise raise
async def search_and_sort( async def search_and_sort( # noqa: C901, PLR0913, PLR0912, PLR0915
self, self,
search_query: str | None = None, search_query: str | None = None,
sort_by: PlaylistSortField | None = None, sort_by: PlaylistSortField | None = None,
sort_order: SortOrder = SortOrder.ASC, sort_order: SortOrder = SortOrder.ASC,
user_id: int | None = None, user_id: int | None = None,
include_stats: bool = False, include_stats: bool = False, # noqa: FBT001, FBT002
limit: int | None = None, limit: int | None = None,
offset: int = 0, offset: int = 0,
) -> list[dict]: ) -> list[dict]:
"""Search and sort playlists with optional statistics.""" """Search and sort playlists with optional statistics."""
try: try:
if include_stats and sort_by in (PlaylistSortField.SOUND_COUNT, PlaylistSortField.TOTAL_DURATION): if include_stats and sort_by in (
PlaylistSortField.SOUND_COUNT,
PlaylistSortField.TOTAL_DURATION,
):
# Use subquery for sorting by stats # Use subquery for sorting by stats
subquery = ( subquery = (
select( select(
@@ -329,12 +334,18 @@ class PlaylistRepository(BaseRepository[Playlist]):
Playlist.created_at, Playlist.created_at,
Playlist.updated_at, Playlist.updated_at,
func.count(PlaylistSound.id).label("sound_count"), func.count(PlaylistSound.id).label("sound_count"),
func.coalesce(func.sum(Sound.duration), 0).label("total_duration"), func.coalesce(func.sum(Sound.duration), 0).label(
"total_duration",
),
User.name.label("user_name"), User.name.label("user_name"),
) )
.select_from(Playlist) .select_from(Playlist)
.join(User, Playlist.user_id == User.id, isouter=True) .join(User, Playlist.user_id == User.id, isouter=True)
.join(PlaylistSound, Playlist.id == PlaylistSound.playlist_id, isouter=True) .join(
PlaylistSound,
Playlist.id == PlaylistSound.playlist_id,
isouter=True,
)
.join(Sound, PlaylistSound.sound_id == Sound.id, isouter=True) .join(Sound, PlaylistSound.sound_id == Sound.id, isouter=True)
.group_by(Playlist.id, User.name) .group_by(Playlist.id, User.name)
) )
@@ -342,7 +353,9 @@ class PlaylistRepository(BaseRepository[Playlist]):
# Apply filters # Apply filters
if search_query and search_query.strip(): if search_query and search_query.strip():
search_pattern = f"%{search_query.strip().lower()}%" search_pattern = f"%{search_query.strip().lower()}%"
subquery = subquery.where(func.lower(Playlist.name).like(search_pattern)) subquery = subquery.where(
func.lower(Playlist.name).like(search_pattern),
)
if user_id is not None: if user_id is not None:
subquery = subquery.where(Playlist.user_id == user_id) subquery = subquery.where(Playlist.user_id == user_id)
@@ -350,14 +363,20 @@ class PlaylistRepository(BaseRepository[Playlist]):
# Apply sorting # Apply sorting
if sort_by == PlaylistSortField.SOUND_COUNT: if sort_by == PlaylistSortField.SOUND_COUNT:
if sort_order == SortOrder.DESC: if sort_order == SortOrder.DESC:
subquery = subquery.order_by(func.count(PlaylistSound.id).desc()) subquery = subquery.order_by(
func.count(PlaylistSound.id).desc(),
)
else: else:
subquery = subquery.order_by(func.count(PlaylistSound.id).asc()) subquery = subquery.order_by(func.count(PlaylistSound.id).asc())
elif sort_by == PlaylistSortField.TOTAL_DURATION: elif sort_by == PlaylistSortField.TOTAL_DURATION:
if sort_order == SortOrder.DESC: if sort_order == SortOrder.DESC:
subquery = subquery.order_by(func.coalesce(func.sum(Sound.duration), 0).desc()) subquery = subquery.order_by(
func.coalesce(func.sum(Sound.duration), 0).desc(),
)
else: else:
subquery = subquery.order_by(func.coalesce(func.sum(Sound.duration), 0).asc()) subquery = subquery.order_by(
func.coalesce(func.sum(Sound.duration), 0).asc(),
)
else: else:
# Default sorting by name # Default sorting by name
subquery = subquery.order_by(Playlist.name.asc()) subquery = subquery.order_by(Playlist.name.asc())
@@ -377,12 +396,18 @@ class PlaylistRepository(BaseRepository[Playlist]):
Playlist.created_at, Playlist.created_at,
Playlist.updated_at, Playlist.updated_at,
func.count(PlaylistSound.id).label("sound_count"), func.count(PlaylistSound.id).label("sound_count"),
func.coalesce(func.sum(Sound.duration), 0).label("total_duration"), func.coalesce(func.sum(Sound.duration), 0).label(
"total_duration",
),
User.name.label("user_name"), User.name.label("user_name"),
) )
.select_from(Playlist) .select_from(Playlist)
.join(User, Playlist.user_id == User.id, isouter=True) .join(User, Playlist.user_id == User.id, isouter=True)
.join(PlaylistSound, Playlist.id == PlaylistSound.playlist_id, isouter=True) .join(
PlaylistSound,
Playlist.id == PlaylistSound.playlist_id,
isouter=True,
)
.join(Sound, PlaylistSound.sound_id == Sound.id, isouter=True) .join(Sound, PlaylistSound.sound_id == Sound.id, isouter=True)
.group_by(Playlist.id, User.name) .group_by(Playlist.id, User.name)
) )
@@ -390,7 +415,9 @@ class PlaylistRepository(BaseRepository[Playlist]):
# Apply filters # Apply filters
if search_query and search_query.strip(): if search_query and search_query.strip():
search_pattern = f"%{search_query.strip().lower()}%" search_pattern = f"%{search_query.strip().lower()}%"
subquery = subquery.where(func.lower(Playlist.name).like(search_pattern)) subquery = subquery.where(
func.lower(Playlist.name).like(search_pattern),
)
if user_id is not None: if user_id is not None:
subquery = subquery.where(Playlist.user_id == user_id) subquery = subquery.where(Playlist.user_id == user_id)
@@ -426,9 +453,8 @@ class PlaylistRepository(BaseRepository[Playlist]):
rows = result.all() rows = result.all()
# Convert to dictionary format # Convert to dictionary format
playlists = [] playlists = [
for row in rows: {
playlists.append({
"id": row.id, "id": row.id,
"name": row.name, "name": row.name,
"description": row.description, "description": row.description,
@@ -442,12 +468,20 @@ class PlaylistRepository(BaseRepository[Playlist]):
"updated_at": row.updated_at, "updated_at": row.updated_at,
"sound_count": row.sound_count or 0, "sound_count": row.sound_count or 0,
"total_duration": row.total_duration or 0, "total_duration": row.total_duration or 0,
}) }
for row in rows
]
return playlists
except Exception: except Exception:
logger.exception( logger.exception(
"Failed to search and sort playlists: query=%s, sort_by=%s, sort_order=%s", (
search_query, sort_by, sort_order, "Failed to search and sort playlists: "
"query=%s, sort_by=%s, sort_order=%s"
),
search_query,
sort_by,
sort_order,
) )
raise raise
else:
return playlists

52
app/repositories/sound.py
View File

@@ -132,7 +132,7 @@ class SoundRepository(BaseRepository[Sound]):
logger.exception("Failed to get sounds by types: %s", sound_types) logger.exception("Failed to get sounds by types: %s", sound_types)
raise raise
async def search_and_sort( async def search_and_sort( # noqa: PLR0913
self, self,
search_query: str | None = None, search_query: str | None = None,
sound_types: list[str] | None = None, sound_types: list[str] | None = None,
@@ -177,8 +177,14 @@ class SoundRepository(BaseRepository[Sound]):
return list(result.all()) return list(result.all())
except Exception: except Exception:
logger.exception( logger.exception(
"Failed to search and sort sounds: query=%s, types=%s, sort_by=%s, sort_order=%s", (
search_query, sound_types, sort_by, sort_order, "Failed to search and sort sounds: "
"query=%s, types=%s, sort_by=%s, sort_order=%s"
),
search_query,
sound_types,
sort_by,
sort_order,
) )
raise raise
@@ -189,21 +195,26 @@ class SoundRepository(BaseRepository[Sound]):
func.count(Sound.id).label("count"), func.count(Sound.id).label("count"),
func.sum(Sound.play_count).label("total_plays"), func.sum(Sound.play_count).label("total_plays"),
func.sum(Sound.duration).label("total_duration"), func.sum(Sound.duration).label("total_duration"),
func.sum(Sound.size + func.coalesce(Sound.normalized_size, 0)).label("total_size"), func.sum(
Sound.size + func.coalesce(Sound.normalized_size, 0),
).label("total_size"),
).where(Sound.type == "SDB") ).where(Sound.type == "SDB")
result = await self.session.exec(statement) result = await self.session.exec(statement)
row = result.first() row = result.first()
return {
"count": row.count if row.count is not None else 0,
"total_plays": row.total_plays if row.total_plays is not None else 0,
"total_duration": row.total_duration if row.total_duration is not None else 0,
"total_size": row.total_size if row.total_size is not None else 0,
}
except Exception: except Exception:
logger.exception("Failed to get soundboard statistics") logger.exception("Failed to get soundboard statistics")
raise raise
else:
return {
"count": row.count if row.count is not None else 0,
"total_plays": row.total_plays if row.total_plays is not None else 0,
"total_duration": (
row.total_duration if row.total_duration is not None else 0
),
"total_size": row.total_size if row.total_size is not None else 0,
}
async def get_track_statistics(self) -> dict[str, int | float]: async def get_track_statistics(self) -> dict[str, int | float]:
"""Get statistics for EXT type sounds.""" """Get statistics for EXT type sounds."""
@@ -212,21 +223,26 @@ class SoundRepository(BaseRepository[Sound]):
func.count(Sound.id).label("count"), func.count(Sound.id).label("count"),
func.sum(Sound.play_count).label("total_plays"), func.sum(Sound.play_count).label("total_plays"),
func.sum(Sound.duration).label("total_duration"), func.sum(Sound.duration).label("total_duration"),
func.sum(Sound.size + func.coalesce(Sound.normalized_size, 0)).label("total_size"), func.sum(
Sound.size + func.coalesce(Sound.normalized_size, 0),
).label("total_size"),
).where(Sound.type == "EXT") ).where(Sound.type == "EXT")
result = await self.session.exec(statement) result = await self.session.exec(statement)
row = result.first() row = result.first()
return {
"count": row.count if row.count is not None else 0,
"total_plays": row.total_plays if row.total_plays is not None else 0,
"total_duration": row.total_duration if row.total_duration is not None else 0,
"total_size": row.total_size if row.total_size is not None else 0,
}
except Exception: except Exception:
logger.exception("Failed to get track statistics") logger.exception("Failed to get track statistics")
raise raise
else:
return {
"count": row.count if row.count is not None else 0,
"total_plays": row.total_plays if row.total_plays is not None else 0,
"total_duration": (
row.total_duration if row.total_duration is not None else 0
),
"total_size": row.total_size if row.total_size is not None else 0,
}
async def get_top_sounds( async def get_top_sounds(
self, self,
@@ -234,7 +250,7 @@ class SoundRepository(BaseRepository[Sound]):
date_filter: datetime | None = None, date_filter: datetime | None = None,
limit: int = 10, limit: int = 10,
) -> list[dict]: ) -> list[dict]:
"""Get top sounds by play count for a specific type and period using SoundPlayed records.""" """Get top sounds by play count for a specific type and period."""
try: try:
# Join SoundPlayed with Sound and count plays within the period # Join SoundPlayed with Sound and count plays within the period
statement = ( statement = (

4
app/schemas/auth.py
View File

@@ -84,7 +84,9 @@ class ApiTokenStatusResponse(BaseModel):
class ChangePasswordRequest(BaseModel): class ChangePasswordRequest(BaseModel):
"""Schema for password change request.""" """Schema for password change request."""
current_password: str | None = Field(None, description="Current password (required if user has existing password)") current_password: str | None = Field(
None, description="Current password (required if user has existing password)",
)
new_password: str = Field( new_password: str = Field(
..., ...,
min_length=8, min_length=8,

10
app/services/auth.py
View File

@@ -467,11 +467,13 @@ class AuthService:
# If user has existing password, verify it # If user has existing password, verify it
if had_existing_password: if had_existing_password:
if not current_password: if not current_password:
raise ValueError("Current password is required when changing existing password") msg = "Current password is required when changing existing password"
raise ValueError(msg)
if not PasswordUtils.verify_password(current_password, user.password_hash): if not PasswordUtils.verify_password(current_password, user.password_hash):
raise ValueError("Current password is incorrect") msg = "Current password is incorrect"
raise ValueError(msg)
else: else:
# User doesn't have a password (OAuth-only user), so we're setting their first password # User doesn't have a password (OAuth-only user), setting first password
logger.info("Setting first password for OAuth user: %s", user_email) logger.info("Setting first password for OAuth user: %s", user_email)
# Hash new password # Hash new password
@@ -509,6 +511,6 @@ class AuthService:
updated_at=user.updated_at, updated_at=user.updated_at,
) )
async def get_user_oauth_providers(self, user: User): async def get_user_oauth_providers(self, user: User) -> list:
"""Get OAuth providers connected to the user.""" """Get OAuth providers connected to the user."""
return await self.oauth_repo.get_by_user_id(user.id) return await self.oauth_repo.get_by_user_id(user.id)

12
app/services/credit.py
View File

@@ -223,11 +223,11 @@ class CreditService:
user_id, user_id,
) )
return transaction
except Exception: except Exception:
await session.rollback() await session.rollback()
raise raise
else:
return transaction
finally: finally:
await session.close() await session.close()
@@ -316,11 +316,11 @@ class CreditService:
user_id, user_id,
) )
return transaction
except Exception: except Exception:
await session.rollback() await session.rollback()
raise raise
else:
return transaction
finally: finally:
await session.close() await session.close()
@@ -503,11 +503,11 @@ class CreditService:
user_id, user_id,
) )
return transaction
except Exception: except Exception:
await session.rollback() await session.rollback()
raise raise
else:
return transaction
finally: finally:
await session.close() await session.close()

2
app/services/dashboard.py
View File

@@ -87,7 +87,7 @@ class DashboardService:
) )
raise raise
def _get_date_filter(self, period: str) -> datetime | None: def _get_date_filter(self, period: str) -> datetime | None: # noqa: PLR0911
"""Calculate the date filter based on the period.""" """Calculate the date filter based on the period."""
now = datetime.now(UTC) now = datetime.now(UTC)

11
app/services/playlist.py
View File

@@ -212,12 +212,13 @@ class PlaylistService:
"""Search all playlists by name.""" """Search all playlists by name."""
return await self.playlist_repo.search_by_name(query) return await self.playlist_repo.search_by_name(query)
async def search_and_sort_playlists( async def search_and_sort_playlists( # noqa: PLR0913
self, self,
search_query: str | None = None, search_query: str | None = None,
sort_by: PlaylistSortField | None = None, sort_by: PlaylistSortField | None = None,
sort_order: SortOrder = SortOrder.ASC, sort_order: SortOrder = SortOrder.ASC,
user_id: int | None = None, user_id: int | None = None,
*,
include_stats: bool = False, include_stats: bool = False,
limit: int | None = None, limit: int | None = None,
offset: int = 0, offset: int = 0,
@@ -269,7 +270,7 @@ class PlaylistService:
current_sounds = await self.playlist_repo.get_playlist_sounds(playlist_id) current_sounds = await self.playlist_repo.get_playlist_sounds(playlist_id)
position = len(current_sounds) position = len(current_sounds)
else: else:
# Ensure position doesn't create gaps - if position is too high, place at end # Ensure position doesn't create gaps - if too high, place at end
current_sounds = await self.playlist_repo.get_playlist_sounds(playlist_id) current_sounds = await self.playlist_repo.get_playlist_sounds(playlist_id)
max_position = len(current_sounds) max_position = len(current_sounds)
position = min(position, max_position) position = min(position, max_position)
@@ -329,7 +330,11 @@ class PlaylistService:
# Create sequential positions: 0, 1, 2, 3... # Create sequential positions: 0, 1, 2, 3...
sound_positions = [(sound.id, index) for index, sound in enumerate(sounds)] sound_positions = [(sound.id, index) for index, sound in enumerate(sounds)]
await self.playlist_repo.reorder_playlist_sounds(playlist_id, sound_positions) await self.playlist_repo.reorder_playlist_sounds(playlist_id, sound_positions)
logger.debug("Reordered %s sounds in playlist %s to eliminate gaps", len(sounds), playlist_id) logger.debug(
"Reordered %s sounds in playlist %s to eliminate gaps",
len(sounds),
playlist_id,
)
async def reorder_playlist_sounds( async def reorder_playlist_sounds(
self, self,

2
app/utils/audio.py
View File

@@ -34,7 +34,7 @@ def get_audio_duration(file_path: Path) -> int:
probe = ffmpeg.probe(str(file_path)) probe = ffmpeg.probe(str(file_path))
duration = float(probe["format"]["duration"]) duration = float(probe["format"]["duration"])
return int(duration * 1000) # Convert to milliseconds return int(duration * 1000) # Convert to milliseconds
except (ffmpeg.Error, KeyError, ValueError, TypeError, Exception) as e: except (ffmpeg.Error, KeyError, ValueError, TypeError, Exception) as e: # noqa: BLE001
logger.warning("Failed to get duration for %s: %s", file_path, e) logger.warning("Failed to get duration for %s: %s", file_path, e)
return 0 return 0