feat: Add favorites management system with full API endpoints
Implement a comprehensive favorites system for anime tracking with the following features:
- Add/remove anime from favorites with unique anime_id
- Toggle favorite status (add if not exists, remove if exists)
- List favorites with sorting (title, rating, year, created_at, updated_at)
- Filter favorites by provider and genre
- Get detailed statistics (total count, provider breakdown, genre distribution, top-rated)
- Persistent storage using JSON file (favorites.json)
- Full REST API with 6 endpoints
API Endpoints:
- GET /api/favorites - List all favorites with sorting/filtering
- POST /api/favorites - Add anime to favorites
- DELETE /api/favorites/{anime_id} - Remove from favorites
- GET /api/favorites/{anime_id} - Get specific favorite details
- GET /api/favorites/stats - Get favorites statistics
- POST /api/favorites/toggle - Toggle favorite status
Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>
This commit is contained in:
root
@@ -0,0 +1,195 @@
|
||||
"""
|
||||
Favorites management system for Ohm Stream Downloader
|
||||
Stores user's favorite anime with metadata in a local JSON file
|
||||
"""
|
||||
import json
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Optional
|
||||
from datetime import datetime
|
||||
import aiofiles
|
||||
|
||||
|
||||
class FavoritesManager:
|
||||
"""Manages user's favorite anime list"""
|
||||
|
||||
def __init__(self, storage_path: str = "data/favorites.json"):
|
||||
self.storage_path = Path(storage_path)
|
||||
self.storage_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._favorites: Dict[str, Dict] = {}
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _load(self):
|
||||
"""Load favorites from disk"""
|
||||
async with self._lock:
|
||||
if self.storage_path.exists():
|
||||
try:
|
||||
async with aiofiles.open(self.storage_path, 'r', encoding='utf-8') as f:
|
||||
content = await f.read()
|
||||
self._favorites = json.loads(content) if content.strip() else {}
|
||||
except Exception as e:
|
||||
print(f"Error loading favorites: {e}")
|
||||
self._favorites = {}
|
||||
else:
|
||||
self._favorites = {}
|
||||
|
||||
async def _save(self):
|
||||
"""Save favorites to disk"""
|
||||
async with self._lock:
|
||||
try:
|
||||
async with aiofiles.open(self.storage_path, 'w', encoding='utf-8') as f:
|
||||
await f.write(json.dumps(self._favorites, indent=2, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
print(f"Error saving favorites: {e}")
|
||||
|
||||
async def add_favorite(
|
||||
self,
|
||||
anime_id: str,
|
||||
title: str,
|
||||
url: str,
|
||||
provider: str,
|
||||
metadata: Optional[Dict] = None,
|
||||
poster_url: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Add an anime to favorites"""
|
||||
await self._load()
|
||||
|
||||
if anime_id in self._favorites:
|
||||
# Update existing favorite
|
||||
self._favorites[anime_id]["updated_at"] = datetime.now().isoformat()
|
||||
if metadata:
|
||||
self._favorites[anime_id]["metadata"] = metadata
|
||||
if poster_url:
|
||||
self._favorites[anime_id]["poster_url"] = poster_url
|
||||
else:
|
||||
# Add new favorite
|
||||
self._favorites[anime_id] = {
|
||||
"id": anime_id,
|
||||
"title": title,
|
||||
"url": url,
|
||||
"provider": provider,
|
||||
"metadata": metadata or {},
|
||||
"poster_url": poster_url,
|
||||
"created_at": datetime.now().isoformat(),
|
||||
"updated_at": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
await self._save()
|
||||
return self._favorites[anime_id]
|
||||
|
||||
async def remove_favorite(self, anime_id: str) -> bool:
|
||||
"""Remove an anime from favorites"""
|
||||
await self._load()
|
||||
|
||||
if anime_id in self._favorites:
|
||||
del self._favorites[anime_id]
|
||||
await self._save()
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
async def get_favorite(self, anime_id: str) -> Optional[Dict]:
|
||||
"""Get a specific favorite by ID"""
|
||||
await self._load()
|
||||
return self._favorites.get(anime_id)
|
||||
|
||||
async def list_favorites(
|
||||
self,
|
||||
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"""
|
||||
await self._load()
|
||||
|
||||
favorites = list(self._favorites.values())
|
||||
|
||||
# Apply filters
|
||||
if filter_provider:
|
||||
favorites = [f for f in favorites if f["provider"] == filter_provider]
|
||||
|
||||
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) -> bool:
|
||||
"""Check if an anime is in favorites"""
|
||||
await self._load()
|
||||
return anime_id in self._favorites
|
||||
|
||||
async def toggle_favorite(
|
||||
self,
|
||||
anime_id: str,
|
||||
title: str,
|
||||
url: str,
|
||||
provider: str,
|
||||
metadata: Optional[Dict] = None,
|
||||
poster_url: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Toggle an anime in favorites (add if not exists, remove if exists)"""
|
||||
is_fav = await self.is_favorite(anime_id)
|
||||
|
||||
if is_fav:
|
||||
await self.remove_favorite(anime_id)
|
||||
return {"action": "removed", "anime_id": anime_id}
|
||||
else:
|
||||
fav = await self.add_favorite(anime_id, title, url, provider, metadata, poster_url)
|
||||
return {"action": "added", "anime_id": anime_id, "favorite": fav}
|
||||
|
||||
async def get_stats(self) -> Dict:
|
||||
"""Get statistics about favorites"""
|
||||
await self._load()
|
||||
|
||||
total = len(self._favorites)
|
||||
|
||||
# Count by provider
|
||||
by_provider = {}
|
||||
for fav in self._favorites.values():
|
||||
provider = fav["provider"]
|
||||
by_provider[provider] = by_provider.get(provider, 0) + 1
|
||||
|
||||
# Count by genre
|
||||
by_genre = {}
|
||||
for fav in self._favorites.values():
|
||||
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
|
||||
}
|
||||
|
||||
|
||||
# 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
|
||||
Reference in New Issue
Block a user