diff --git a/fastanime/cli/utils/watch_history_types.py b/fastanime/cli/utils/watch_history_types.py index 2123efd..ac251b6 100644 --- a/fastanime/cli/utils/watch_history_types.py +++ b/fastanime/cli/utils/watch_history_types.py @@ -6,17 +6,17 @@ Provides comprehensive data structures for tracking and managing local watch his from __future__ import annotations import logging -from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional +from pydantic import BaseModel, Field + from ...libs.api.types import MediaItem logger = logging.getLogger(__name__) -@dataclass -class WatchHistoryEntry: +class WatchHistoryEntry(BaseModel): """ Represents a single entry in the watch history. Contains media information and viewing progress. @@ -26,104 +26,16 @@ class WatchHistoryEntry: last_watched_episode: int = 0 watch_progress: float = 0.0 # Progress within the episode (0.0-1.0) times_watched: int = 1 - first_watched: datetime = field(default_factory=datetime.now) - last_watched: datetime = field(default_factory=datetime.now) + first_watched: datetime = Field(default_factory=datetime.now) + last_watched: datetime = Field(default_factory=datetime.now) status: str = "watching" # watching, completed, dropped, paused notes: str = "" - def to_dict(self) -> dict: - """Convert entry to dictionary for JSON serialization.""" - return { - "media_item": { - "id": self.media_item.id, - "id_mal": self.media_item.id_mal, - "type": self.media_item.type, - "title": { - "romaji": self.media_item.title.romaji, - "english": self.media_item.title.english, - "native": self.media_item.title.native, - }, - "status": self.media_item.status, - "format": self.media_item.format, - "cover_image": { - "large": self.media_item.cover_image.large if self.media_item.cover_image else None, - "medium": self.media_item.cover_image.medium if self.media_item.cover_image else None, - } if self.media_item.cover_image else None, - "banner_image": self.media_item.banner_image, - "description": self.media_item.description, - "episodes": self.media_item.episodes, - "duration": self.media_item.duration, - "genres": self.media_item.genres, - "synonyms": self.media_item.synonyms, - "average_score": self.media_item.average_score, - "popularity": self.media_item.popularity, - "favourites": self.media_item.favourites, - }, - "last_watched_episode": self.last_watched_episode, - "watch_progress": self.watch_progress, - "times_watched": self.times_watched, - "first_watched": self.first_watched.isoformat(), - "last_watched": self.last_watched.isoformat(), - "status": self.status, - "notes": self.notes, - } + # With Pydantic, serialization is automatic! + # No need for manual to_dict() and from_dict() methods + # Use: entry.model_dump() and WatchHistoryEntry.model_validate(data) - @classmethod - def from_dict(cls, data: dict) -> "WatchHistoryEntry": - """Create entry from dictionary.""" - from ...libs.api.types import MediaImage, MediaTitle - - media_data = data["media_item"] - - # Reconstruct MediaTitle - title_data = media_data.get("title", {}) - title = MediaTitle( - romaji=title_data.get("romaji"), - english=title_data.get("english"), - native=title_data.get("native"), - ) - - # Reconstruct MediaImage if present - cover_data = media_data.get("cover_image") - cover_image = None - if cover_data: - cover_image = MediaImage( - large=cover_data.get("large", ""), - medium=cover_data.get("medium"), - ) - - # Reconstruct MediaItem - media_item = MediaItem( - id=media_data["id"], - id_mal=media_data.get("id_mal"), - type=media_data.get("type", "ANIME"), - title=title, - status=media_data.get("status"), - format=media_data.get("format"), - cover_image=cover_image, - banner_image=media_data.get("banner_image"), - description=media_data.get("description"), - episodes=media_data.get("episodes"), - duration=media_data.get("duration"), - genres=media_data.get("genres", []), - synonyms=media_data.get("synonyms", []), - average_score=media_data.get("average_score"), - popularity=media_data.get("popularity"), - favourites=media_data.get("favourites"), - ) - - return cls( - media_item=media_item, - last_watched_episode=data.get("last_watched_episode", 0), - watch_progress=data.get("watch_progress", 0.0), - times_watched=data.get("times_watched", 1), - first_watched=datetime.fromisoformat(data.get("first_watched", datetime.now().isoformat())), - last_watched=datetime.fromisoformat(data.get("last_watched", datetime.now().isoformat())), - status=data.get("status", "watching"), - notes=data.get("notes", ""), - ) - - def update_progress(self, episode: int, progress: float = 0.0, status: str = None): + def update_progress(self, episode: int, progress: float = 0.0, status: Optional[str] = None): """Update watch progress for this entry.""" self.last_watched_episode = max(self.last_watched_episode, episode) self.watch_progress = progress @@ -169,41 +81,16 @@ class WatchHistoryEntry: return status_emojis.get(self.status, "❓") -@dataclass -class WatchHistoryData: +class WatchHistoryData(BaseModel): """Complete watch history data container.""" - entries: Dict[int, WatchHistoryEntry] = field(default_factory=dict) - last_updated: datetime = field(default_factory=datetime.now) + entries: Dict[int, WatchHistoryEntry] = Field(default_factory=dict) + last_updated: datetime = Field(default_factory=datetime.now) format_version: str = "1.0" - def to_dict(self) -> dict: - """Convert to dictionary for JSON serialization.""" - return { - "entries": {str(k): v.to_dict() for k, v in self.entries.items()}, - "last_updated": self.last_updated.isoformat(), - "format_version": self.format_version, - } - - @classmethod - def from_dict(cls, data: dict) -> "WatchHistoryData": - """Create from dictionary.""" - entries = {} - entries_data = data.get("entries", {}) - - for media_id_str, entry_data in entries_data.items(): - try: - media_id = int(media_id_str) - entry = WatchHistoryEntry.from_dict(entry_data) - entries[media_id] = entry - except (ValueError, KeyError) as e: - logger.warning(f"Skipping invalid watch history entry {media_id_str}: {e}") - - return cls( - entries=entries, - last_updated=datetime.fromisoformat(data.get("last_updated", datetime.now().isoformat())), - format_version=data.get("format_version", "1.0"), - ) + # With Pydantic, serialization is automatic! + # No need for manual to_dict() and from_dict() methods + # Use: data.model_dump() and WatchHistoryData.model_validate(data) def add_or_update_entry(self, media_item: MediaItem, episode: int = 0, progress: float = 0.0, status: str = "watching") -> WatchHistoryEntry: """Add or update a watch history entry.""" diff --git a/fastanime/libs/api/jikan/mapper.py b/fastanime/libs/api/jikan/mapper.py index 8c6ddf8..75e345b 100644 --- a/fastanime/libs/api/jikan/mapper.py +++ b/fastanime/libs/api/jikan/mapper.py @@ -32,30 +32,39 @@ JIKAN_STATUS_MAP = { def _to_generic_title(jikan_titles: list[dict]) -> MediaTitle: """Extracts titles from Jikan's list of title objects.""" - title_obj = MediaTitle() + # Initialize with default values + romaji = None + english = None + native = None + # Jikan's default title is often the romaji one. # We prioritize specific types if available. for t in jikan_titles: type_ = t.get("type") title_ = t.get("title") if type_ == "Default": - title_obj.romaji = title_ + romaji = title_ elif type_ == "English": - title_obj.english = title_ + english = title_ elif type_ == "Japanese": - title_obj.native = title_ - return title_obj + native = title_ + + return MediaTitle( + romaji=romaji, + english=english, + native=native + ) def _to_generic_image(jikan_images: dict) -> MediaImage: """Maps Jikan's image structure.""" if not jikan_images: - return MediaImage() + return MediaImage(large="") # Provide empty string as fallback # Jikan provides different image formats under a 'jpg' key. jpg_images = jikan_images.get("jpg", {}) return MediaImage( + large=jpg_images.get("large_image_url", ""), # Fallback to empty string medium=jpg_images.get("image_url"), - large=jpg_images.get("large_image_url"), ) @@ -71,7 +80,7 @@ def _to_generic_media_item(data: dict) -> MediaItem: id_mal=data["mal_id"], title=_to_generic_title(data.get("titles", [])), cover_image=_to_generic_image(data.get("images", {})), - status=JIKAN_STATUS_MAP.get(data.get("status")), + status=JIKAN_STATUS_MAP.get(data.get("status", ""), None), episodes=data.get("episodes"), duration=data.get("duration"), average_score=score, diff --git a/fastanime/libs/api/types.py b/fastanime/libs/api/types.py index 25d069d..a9bf246 100644 --- a/fastanime/libs/api/types.py +++ b/fastanime/libs/api/types.py @@ -1,9 +1,10 @@ from __future__ import annotations -from dataclasses import dataclass, field from datetime import datetime from typing import List, Literal, Optional +from pydantic import BaseModel, Field + # --- Generic Enums and Type Aliases --- MediaType = Literal["ANIME", "MANGA"] @@ -17,8 +18,12 @@ UserListStatusType = Literal[ # --- Generic Data Models --- -@dataclass(frozen=True) -class MediaImage: +class BaseApiModel(BaseModel): + """Base model for all API types.""" + pass + + +class MediaImage(BaseApiModel): """A generic representation of media imagery URLs.""" large: str @@ -26,8 +31,7 @@ class MediaImage: extra_large: Optional[str] = None -@dataclass(frozen=True) -class MediaTitle: +class MediaTitle(BaseApiModel): """A generic representation of media titles.""" romaji: Optional[str] = None @@ -35,8 +39,7 @@ class MediaTitle: native: Optional[str] = None -@dataclass(frozen=True) -class MediaTrailer: +class MediaTrailer(BaseApiModel): """A generic representation of a media trailer.""" id: str @@ -44,16 +47,14 @@ class MediaTrailer: thumbnail_url: Optional[str] = None -@dataclass(frozen=True) -class AiringSchedule: +class AiringSchedule(BaseApiModel): """A generic representation of the next airing episode.""" episode: int airing_at: datetime | None = None -@dataclass(frozen=True) -class Studio: +class Studio(BaseApiModel): """A generic representation of an animation studio.""" id: int | None = None @@ -62,24 +63,21 @@ class Studio: is_animation_studio: bool | None = None -@dataclass(frozen=True) -class MediaTag: +class MediaTag(BaseApiModel): """A generic representation of a descriptive tag.""" name: str rank: Optional[int] = None # Percentage relevance from 0-100 -@dataclass(frozen=True) -class StreamingEpisode: +class StreamingEpisode(BaseApiModel): """A generic representation of a streaming episode.""" title: str thumbnail: Optional[str] = None -@dataclass(frozen=True) -class UserListStatus: +class UserListStatus(BaseApiModel): """Generic representation of a user's list status for a media item.""" id: int | None = None @@ -94,8 +92,7 @@ class UserListStatus: created_at: Optional[str] = None -@dataclass(frozen=True) -class MediaItem: +class MediaItem(BaseApiModel): """ The definitive, backend-agnostic representation of a single media item. This is the primary data model the application will interact with. @@ -104,7 +101,7 @@ class MediaItem: id: int id_mal: Optional[int] = None type: MediaType = "ANIME" - title: MediaTitle = field(default_factory=MediaTitle) + title: MediaTitle = Field(default_factory=MediaTitle) status: Optional[str] = None format: Optional[str] = None # e.g., TV, MOVIE, OVA @@ -115,10 +112,10 @@ class MediaItem: description: Optional[str] = None episodes: Optional[int] = None duration: Optional[int] = None # In minutes - genres: List[str] = field(default_factory=list) - tags: List[MediaTag] = field(default_factory=list) - studios: List[Studio] = field(default_factory=list) - synonyms: List[str] = field(default_factory=list) + genres: List[str] = Field(default_factory=list) + tags: List[MediaTag] = Field(default_factory=list) + studios: List[Studio] = Field(default_factory=list) + synonyms: List[str] = Field(default_factory=list) average_score: Optional[float] = None popularity: Optional[int] = None @@ -127,14 +124,13 @@ class MediaItem: next_airing: Optional[AiringSchedule] = None # streaming episodes - streaming_episodes: List[StreamingEpisode] = field(default_factory=list) + streaming_episodes: List[StreamingEpisode] = Field(default_factory=list) # user related user_status: Optional[UserListStatus] = None -@dataclass(frozen=True) -class PageInfo: +class PageInfo(BaseApiModel): """Generic pagination information.""" total: int @@ -143,16 +139,14 @@ class PageInfo: per_page: int -@dataclass(frozen=True) -class MediaSearchResult: +class MediaSearchResult(BaseApiModel): """A generic representation of a page of media search results.""" page_info: PageInfo - media: List[MediaItem] = field(default_factory=list) + media: List[MediaItem] = Field(default_factory=list) -@dataclass(frozen=True) -class UserProfile: +class UserProfile(BaseApiModel): """A generic representation of a user's profile.""" id: int