root
5d264d8f3b
- router_favorites.py: toutes les routes requièrent maintenant l'auth - GET utilise get_optional_user + login_prompt.html pour HTMX - POST/DELETE/toggle requièrent get_current_user_from_token - Filtrage par user_id dans toutes les requêtes favorites - router_downloads.py: GET list et GET status protégés (401 sans token) - router_recommendations.py: GET protégé (login_prompt HTMX, 401 JSON) - router_sonarr.py: tous les endpoints de gestion protégés - Webhooks restent publics (reçus de Sonarr) - app/favorites.py: ajout du paramètre user_id à toutes les méthodes Closes #15
214 lines
7.4 KiB
Python
214 lines
7.4 KiB
Python
"""
|
|
Favorites management system for Ohm Stream Downloader
|
|
Stores user's favorite anime with metadata using SQLModel
|
|
"""
|
|
import logging
|
|
from typing import List, Dict, Optional
|
|
from datetime import datetime
|
|
|
|
from sqlmodel import Session, select
|
|
from app.database import engine
|
|
from app.models.favorites import FavoriteTable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FavoritesManager:
|
|
"""Manages user's favorite anime list using SQL database"""
|
|
|
|
def __init__(self, storage_path: str = None):
|
|
# Database connection is managed via engine and sessions
|
|
pass
|
|
|
|
async def add_favorite(
|
|
self,
|
|
anime_id: str,
|
|
title: str,
|
|
url: str,
|
|
provider: str,
|
|
metadata: Optional[Dict] = None,
|
|
poster_url: Optional[str] = None,
|
|
user_id: str = "default"
|
|
) -> Dict:
|
|
"""Add an anime to favorites"""
|
|
with Session(engine) as session:
|
|
statement = select(FavoriteTable).where(
|
|
FavoriteTable.anime_id == anime_id,
|
|
FavoriteTable.user_id == user_id
|
|
)
|
|
existing = session.exec(statement).first()
|
|
|
|
if existing:
|
|
# Update existing favorite
|
|
existing.updated_at = datetime.now()
|
|
if metadata:
|
|
existing.anime_metadata = metadata
|
|
if poster_url:
|
|
existing.poster_url = poster_url
|
|
session.add(existing)
|
|
session.commit()
|
|
session.refresh(existing)
|
|
return self._to_dict(existing)
|
|
else:
|
|
# Add new favorite
|
|
fav = FavoriteTable(
|
|
anime_id=anime_id,
|
|
title=title,
|
|
url=url,
|
|
provider=provider,
|
|
anime_metadata=metadata or {},
|
|
poster_url=poster_url,
|
|
user_id=user_id
|
|
)
|
|
session.add(fav)
|
|
session.commit()
|
|
session.refresh(fav)
|
|
return self._to_dict(fav)
|
|
|
|
async def remove_favorite(self, anime_id: str, user_id: str = "default") -> bool:
|
|
"""Remove an anime from favorites"""
|
|
with Session(engine) as session:
|
|
statement = select(FavoriteTable).where(
|
|
FavoriteTable.anime_id == anime_id,
|
|
FavoriteTable.user_id == user_id
|
|
)
|
|
existing = session.exec(statement).first()
|
|
if existing:
|
|
session.delete(existing)
|
|
session.commit()
|
|
return True
|
|
return False
|
|
|
|
async def get_favorite(self, anime_id: str, user_id: str = "default") -> Optional[Dict]:
|
|
"""Get a specific favorite by ID"""
|
|
with Session(engine) as session:
|
|
statement = select(FavoriteTable).where(
|
|
FavoriteTable.anime_id == anime_id,
|
|
FavoriteTable.user_id == user_id
|
|
)
|
|
existing = session.exec(statement).first()
|
|
if existing:
|
|
return self._to_dict(existing)
|
|
return None
|
|
|
|
async def list_favorites(
|
|
self,
|
|
user_id: str = "default",
|
|
sort_by: str = "created_at",
|
|
order: str = "desc",
|
|
filter_provider: Optional[str] = None,
|
|
filter_genre: Optional[str] = None
|
|
) -> List[Dict]:
|
|
"""List all favorites with optional sorting and filtering"""
|
|
with Session(engine) as session:
|
|
statement = select(FavoriteTable).where(FavoriteTable.user_id == user_id)
|
|
|
|
if filter_provider:
|
|
statement = statement.where(FavoriteTable.provider == filter_provider)
|
|
|
|
# SQLite JSON filtering for genres is complex, handle it in Python
|
|
results = session.exec(statement).all()
|
|
favorites = [self._to_dict(fav) for fav in results]
|
|
|
|
if filter_genre:
|
|
favorites = [
|
|
f for f in favorites
|
|
if filter_genre in f.get("metadata", {}).get("genres", [])
|
|
]
|
|
|
|
# Sort favorites
|
|
reverse = order == "desc"
|
|
if sort_by == "title":
|
|
favorites.sort(key=lambda x: x["title"].lower(), reverse=reverse)
|
|
elif sort_by == "rating":
|
|
favorites.sort(
|
|
key=lambda x: float(x.get("metadata", {}).get("rating", "0").split("/")[0]),
|
|
reverse=reverse
|
|
)
|
|
elif sort_by == "year":
|
|
favorites.sort(
|
|
key=lambda x: x.get("metadata", {}).get("release_year", 0),
|
|
reverse=reverse
|
|
)
|
|
else: # created_at, updated_at
|
|
favorites.sort(key=lambda x: x.get(sort_by, ""), reverse=reverse)
|
|
|
|
return favorites
|
|
|
|
async def is_favorite(self, anime_id: str, user_id: str = "default") -> bool:
|
|
"""Check if an anime is in favorites"""
|
|
with Session(engine) as session:
|
|
statement = select(FavoriteTable).where(
|
|
FavoriteTable.anime_id == anime_id,
|
|
FavoriteTable.user_id == user_id
|
|
)
|
|
return session.exec(statement).first() is not None
|
|
|
|
async def toggle_favorite(
|
|
self,
|
|
anime_id: str,
|
|
title: str,
|
|
url: str,
|
|
provider: str,
|
|
metadata: Optional[Dict] = None,
|
|
poster_url: Optional[str] = None,
|
|
user_id: str = "default"
|
|
) -> Dict:
|
|
"""Toggle an anime in favorites (add if not exists, remove if exists)"""
|
|
is_fav = await self.is_favorite(anime_id, user_id=user_id)
|
|
|
|
if is_fav:
|
|
await self.remove_favorite(anime_id, user_id=user_id)
|
|
return {"action": "removed", "anime_id": anime_id}
|
|
else:
|
|
fav = await self.add_favorite(anime_id, title, url, provider, metadata, poster_url, user_id=user_id)
|
|
return {"action": "added", "anime_id": anime_id, "favorite": fav}
|
|
|
|
async def get_stats(self, user_id: str = "default") -> Dict:
|
|
"""Get statistics about favorites"""
|
|
favorites = await self.list_favorites(user_id=user_id)
|
|
total = len(favorites)
|
|
|
|
# Count by provider
|
|
by_provider = {}
|
|
for fav in favorites:
|
|
provider = fav["provider"]
|
|
by_provider[provider] = by_provider.get(provider, 0) + 1
|
|
|
|
# Count by genre
|
|
by_genre = {}
|
|
for fav in favorites:
|
|
for genre in fav.get("metadata", {}).get("genres", []):
|
|
by_genre[genre] = by_genre.get(genre, 0) + 1
|
|
|
|
return {
|
|
"total": total,
|
|
"by_provider": by_provider,
|
|
"by_genre": by_genre
|
|
}
|
|
|
|
def _to_dict(self, fav: FavoriteTable) -> Dict:
|
|
"""Convert a FavoriteTable instance to a dictionary for API compatibility"""
|
|
return {
|
|
"id": fav.anime_id,
|
|
"title": fav.title,
|
|
"url": fav.url,
|
|
"provider": fav.provider,
|
|
"metadata": fav.anime_metadata,
|
|
"poster_url": fav.poster_url,
|
|
"created_at": fav.created_at.isoformat() if fav.created_at else None,
|
|
"updated_at": fav.updated_at.isoformat() if fav.updated_at else None
|
|
}
|
|
|
|
|
|
# Global favorites manager instance
|
|
_favorites_manager: Optional[FavoritesManager] = None
|
|
|
|
|
|
def get_favorites_manager() -> FavoritesManager:
|
|
"""Get the global favorites manager instance"""
|
|
global _favorites_manager
|
|
if _favorites_manager is None:
|
|
_favorites_manager = FavoritesManager()
|
|
return _favorites_manager
|