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

refactor: Remove unused playlist routes and related logic; clean up sound and stream models

Browse Source
This commit is contained in:
JSC
2025-07-12 22:00:04 +02:00
parent 627b95c961
commit 688b95b6af
12 changed files with 34 additions and 1139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
app/models/playlist.py
View File

@@ -81,19 +81,6 @@ class Playlist(db.Model):
),
}
def to_detailed_dict(self) -> dict:
"""Convert playlist to detailed dictionary with sounds."""
playlist_dict = self.to_dict()
playlist_dict["sounds"] = [
{
"sound": ps.sound.to_dict() if ps.sound else None,
"order": ps.order,
"added_at": ps.added_at.isoformat() if ps.added_at else None,
}
for ps in sorted(self.playlist_sounds, key=lambda x: x.order)
]
return playlist_dict
@classmethod
def create_playlist(
cls,
@@ -123,26 +110,6 @@ class Playlist(db.Model):
return playlist
@classmethod
def find_by_name(
cls, name: str, user_id: Optional[int] = None
) -> Optional["Playlist"]:
"""Find playlist by name, optionally filtered by user."""
query = cls.query.filter_by(name=name)
if user_id is not None:
query = query.filter_by(user_id=user_id)
return query.first()
@classmethod
def find_by_user(cls, user_id: int) -> list["Playlist"]:
"""Find all playlists for a user."""
return cls.query.filter_by(user_id=user_id).order_by(cls.name).all()
@classmethod
def find_system_playlists(cls) -> list["Playlist"]:
"""Find all system playlists (user_id is None)."""
return cls.query.filter_by(user_id=None).order_by(cls.name).all()
@classmethod
def find_current_playlist(
cls, user_id: Optional[int] = None
@@ -163,22 +130,6 @@ class Playlist(db.Model):
query = query.filter_by(user_id=user_id)
return query.first()
def set_as_current(self, commit: bool = True) -> None:
"""Set this playlist as the current one and unset others."""
# Unset other current playlists for the same user/system
if self.user_id is not None:
Playlist.query.filter_by(
user_id=self.user_id, is_current=True
).update({"is_current": False})
else:
Playlist.query.filter_by(user_id=None, is_current=True).update(
{"is_current": False}
)
self.is_current = True
if commit:
db.session.commit()
def add_sound(
self, sound_id: int, order: Optional[int] = None, commit: bool = True
) -> "PlaylistSound":
@@ -203,87 +154,3 @@ class Playlist(db.Model):
db.session.commit()
return playlist_sound
def remove_sound(self, sound_id: int, commit: bool = True) -> bool:
"""Remove a sound from the playlist."""
from app.models.playlist_sound import PlaylistSound
playlist_sound = PlaylistSound.query.filter_by(
playlist_id=self.id, sound_id=sound_id
).first()
if playlist_sound:
db.session.delete(playlist_sound)
if commit:
db.session.commit()
return True
return False
def reorder_sounds(
self, sound_orders: list[dict], commit: bool = True
) -> None:
"""Reorder sounds in the playlist.
Args:
sound_orders: List of dicts with 'sound_id' and 'order' keys
"""
from app.models.playlist_sound import PlaylistSound
for item in sound_orders:
playlist_sound = PlaylistSound.query.filter_by(
playlist_id=self.id, sound_id=item["sound_id"]
).first()
if playlist_sound:
playlist_sound.order = item["order"]
if commit:
db.session.commit()
def get_total_duration(self) -> int:
"""Get total duration of all sounds in the playlist in milliseconds."""
from app.models.sound import Sound
total = (
db.session.query(db.func.sum(Sound.duration))
.join(self.playlist_sounds)
.filter(Sound.id.in_([ps.sound_id for ps in self.playlist_sounds]))
.scalar()
)
return total or 0
def duplicate(
self, new_name: str, user_id: Optional[int] = None, commit: bool = True
) -> "Playlist":
"""Create a duplicate of this playlist."""
new_playlist = Playlist.create_playlist(
name=new_name,
description=self.description,
genre=self.genre,
user_id=user_id,
is_main=False,
is_deletable=True,
is_current=False,
commit=commit,
)
# Copy all sounds with their order
for ps in self.playlist_sounds:
new_playlist.add_sound(ps.sound_id, ps.order, commit=False)
if commit:
db.session.commit()
return new_playlist
def save(self, commit: bool = True) -> None:
"""Save changes to the playlist."""
if commit:
db.session.commit()
def delete(self, commit: bool = True) -> None:
"""Delete the playlist."""
db.session.delete(self)
if commit:
db.session.commit()

123
app/models/playlist_sound.py
View File

@@ -63,126 +63,3 @@ class PlaylistSound(db.Model):
"added_at": self.added_at.isoformat() if self.added_at else None,
"sound": self.sound.to_dict() if self.sound else None,
}
@classmethod
def create_playlist_sound(
cls,
playlist_id: int,
sound_id: int,
order: int,
commit: bool = True,
) -> "PlaylistSound":
"""Create a new playlist-sound relationship."""
playlist_sound = cls(
playlist_id=playlist_id,
sound_id=sound_id,
order=order,
)
db.session.add(playlist_sound)
if commit:
db.session.commit()
return playlist_sound
@classmethod
def find_by_playlist(cls, playlist_id: int) -> list["PlaylistSound"]:
"""Find all sounds in a playlist ordered by their position."""
return (
cls.query.filter_by(playlist_id=playlist_id)
.order_by(cls.order)
.all()
)
@classmethod
def find_by_sound(cls, sound_id: int) -> list["PlaylistSound"]:
"""Find all playlists containing a specific sound."""
return cls.query.filter_by(sound_id=sound_id).all()
@classmethod
def find_by_playlist_and_sound(
cls, playlist_id: int, sound_id: int
) -> Optional["PlaylistSound"]:
"""Find a specific playlist-sound relationship."""
return cls.query.filter_by(
playlist_id=playlist_id, sound_id=sound_id
).first()
@classmethod
def get_next_order(cls, playlist_id: int) -> int:
"""Get the next order number for a playlist."""
max_order = (
db.session.query(db.func.max(cls.order))
.filter_by(playlist_id=playlist_id)
.scalar()
)
return (max_order or 0) + 1
@classmethod
def reorder_playlist(
cls, playlist_id: int, sound_orders: list[dict], commit: bool = True
) -> None:
"""Reorder all sounds in a playlist.
Args:
playlist_id: ID of the playlist
sound_orders: List of dicts with 'sound_id' and 'order' keys
"""
for item in sound_orders:
playlist_sound = cls.query.filter_by(
playlist_id=playlist_id, sound_id=item["sound_id"]
).first()
if playlist_sound:
playlist_sound.order = item["order"]
if commit:
db.session.commit()
def move_to_position(self, new_order: int, commit: bool = True) -> None:
"""Move this sound to a new position in the playlist."""
old_order = self.order
if new_order == old_order:
return
# Get all other sounds in the playlist
other_sounds = (
PlaylistSound.query.filter_by(playlist_id=self.playlist_id)
.filter(PlaylistSound.id != self.id)
.order_by(PlaylistSound.order)
.all()
)
# Remove this sound from its current position
remaining_sounds = [ps for ps in other_sounds if ps.order != old_order]
# Insert at new position
if new_order <= len(remaining_sounds):
remaining_sounds.insert(new_order - 1, self)
else:
remaining_sounds.append(self)
# Update all order values
for i, ps in enumerate(remaining_sounds, 1):
ps.order = i
if commit:
db.session.commit()
def get_previous_sound(self) -> Optional["PlaylistSound"]:
"""Get the previous sound in the playlist."""
return (
PlaylistSound.query.filter_by(playlist_id=self.playlist_id)
.filter(PlaylistSound.order < self.order)
.order_by(PlaylistSound.order.desc())
.first()
)
def get_next_sound(self) -> Optional["PlaylistSound"]:
"""Get the next sound in the playlist."""
return (
PlaylistSound.query.filter_by(playlist_id=self.playlist_id)
.filter(PlaylistSound.order > self.order)
.order_by(PlaylistSound.order.asc())
.first()
)

15
app/models/sound.py
View File

@@ -197,21 +197,6 @@ class Sound(db.Model):
"""Find all sounds by type."""
return cls.query.filter_by(type=sound_type).all()
@classmethod
def get_most_played(cls, limit: int = 10) -> list["Sound"]:
"""Get the most played sounds."""
return cls.query.order_by(cls.play_count.desc()).limit(limit).all()
@classmethod
def get_music_sounds(cls) -> list["Sound"]:
"""Get all music sounds."""
return cls.query.filter_by(is_music=True).all()
@classmethod
def get_deletable_sounds(cls) -> list["Sound"]:
"""Get all deletable sounds."""
return cls.query.filter_by(is_deletable=True).all()
@classmethod
def create_sound(
cls,

170
app/models/sound_played.py
View File

@@ -92,173 +92,3 @@ class SoundPlayed(db.Model):
if commit:
db.session.commit()
return play_record
@classmethod
def get_user_plays(
cls,
user_id: int,
limit: int = 50,
offset: int = 0,
) -> list["SoundPlayed"]:
"""Get recent plays for a specific user."""
return (
cls.query.filter_by(user_id=user_id)
.order_by(cls.played_at.desc())
.offset(offset)
.limit(limit)
.all()
)
@classmethod
def get_sound_plays(
cls,
sound_id: int,
limit: int = 50,
offset: int = 0,
) -> list["SoundPlayed"]:
"""Get recent plays for a specific sound."""
return (
cls.query.filter_by(sound_id=sound_id)
.order_by(cls.played_at.desc())
.offset(offset)
.limit(limit)
.all()
)
@classmethod
def get_recent_plays(
cls,
limit: int = 100,
offset: int = 0,
) -> list["SoundPlayed"]:
"""Get recent plays across all users and sounds."""
return (
cls.query.order_by(cls.played_at.desc())
.offset(offset)
.limit(limit)
.all()
)
@classmethod
def get_user_play_count(cls, user_id: int) -> int:
"""Get total play count for a user."""
return cls.query.filter_by(user_id=user_id).count()
@classmethod
def get_sound_play_count(cls, sound_id: int) -> int:
"""Get total play count for a sound."""
return cls.query.filter_by(sound_id=sound_id).count()
@classmethod
def get_popular_sounds(
cls,
limit: int = 10,
days: int | None = None,
) -> list[dict]:
"""Get most popular sounds with play counts."""
from app.models.sound import Sound
query = (
db.session.query(
cls.sound_id,
func.count(cls.id).label("play_count"),
func.max(cls.played_at).label("last_played"),
)
.group_by(cls.sound_id)
.order_by(func.count(cls.id).desc())
)
if days:
query = query.filter(
cls.played_at >= text(f"datetime('now', '-{days} days')"),
)
results = query.limit(limit).all()
# Get sound details
popular_sounds = []
for result in results:
sound = Sound.query.get(result.sound_id)
if sound:
popular_sounds.append(
{
"sound": sound.to_dict(),
"play_count": result.play_count,
"last_played": (
result.last_played.isoformat()
if result.last_played
else None
),
},
)
return popular_sounds
@classmethod
def get_user_stats(cls, user_id: int) -> dict:
"""Get comprehensive stats for a user."""
from app.models.sound import Sound
total_plays = cls.query.filter_by(user_id=user_id).count()
if total_plays == 0:
return {
"total_plays": 0,
"unique_sounds": 0,
"favorite_sound": None,
"first_play": None,
"last_play": None,
}
# Get unique sounds count
unique_sounds = (
db.session.query(cls.sound_id)
.filter_by(user_id=user_id)
.distinct()
.count()
)
# Get favorite sound
favorite_query = (
db.session.query(
cls.sound_id,
func.count(cls.id).label("play_count"),
)
.filter_by(user_id=user_id)
.group_by(cls.sound_id)
.order_by(func.count(cls.id).desc())
.first()
)
favorite_sound = None
if favorite_query:
sound = Sound.query.get(favorite_query.sound_id)
if sound:
favorite_sound = {
"sound": sound.to_dict(),
"play_count": favorite_query.play_count,
}
# Get first and last play dates
first_play = (
cls.query.filter_by(user_id=user_id)
.order_by(cls.played_at.asc())
.first()
)
last_play = (
cls.query.filter_by(user_id=user_id)
.order_by(cls.played_at.desc())
.first()
)
return {
"total_plays": total_plays,
"unique_sounds": unique_sounds,
"favorite_sound": favorite_sound,
"first_play": (
first_play.played_at.isoformat() if first_play else None
),
"last_play": (
last_play.played_at.isoformat() if last_play else None
),
}

80
app/models/stream.py
View File

@@ -4,7 +4,14 @@ from datetime import datetime
from typing import TYPE_CHECKING, Optional
from zoneinfo import ZoneInfo
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
from sqlalchemy import (
DateTime,
ForeignKey,
Integer,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import db
@@ -51,9 +58,7 @@ class Stream(db.Model):
# Constraints
__table_args__ = (
UniqueConstraint(
"service", "service_id", name="unique_service_stream"
),
UniqueConstraint("service", "service_id", name="unique_service_stream"),
)
def __repr__(self) -> str:
@@ -117,70 +122,3 @@ class Stream(db.Model):
db.session.commit()
return stream
@classmethod
def find_by_service_and_id(
cls, service: str, service_id: str
) -> Optional["Stream"]:
"""Find stream by service and service_id."""
return cls.query.filter_by(
service=service, service_id=service_id
).first()
@classmethod
def find_by_sound(cls, sound_id: int) -> list["Stream"]:
"""Find all streams for a specific sound."""
return cls.query.filter_by(sound_id=sound_id).all()
@classmethod
def find_by_service(cls, service: str) -> list["Stream"]:
"""Find all streams for a specific service."""
return cls.query.filter_by(service=service).all()
@classmethod
def find_by_status(cls, status: str) -> list["Stream"]:
"""Find all streams with a specific status."""
return cls.query.filter_by(status=status).all()
@classmethod
def find_active_streams(cls) -> list["Stream"]:
"""Find all active streams."""
return cls.query.filter_by(status="active").all()
def update_metadata(
self,
title: Optional[str] = None,
track: Optional[str] = None,
artist: Optional[str] = None,
album: Optional[str] = None,
genre: Optional[str] = None,
commit: bool = True,
) -> None:
"""Update stream metadata."""
if title is not None:
self.title = title
if track is not None:
self.track = track
if artist is not None:
self.artist = artist
if album is not None:
self.album = album
if genre is not None:
self.genre = genre
if commit:
db.session.commit()
def set_status(self, status: str, commit: bool = True) -> None:
"""Update stream status."""
self.status = status
if commit:
db.session.commit()
def is_active(self) -> bool:
"""Check if stream is active."""
return self.status == "active"
def get_display_name(self) -> str:
"""Get a display name for the stream (title or track or service_id)."""
return self.title or self.track or self.service_id