mirror of
https://github.com/Benexl/FastAnime.git
synced 2026-01-01 23:39:54 -08:00
95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, Literal, Optional
|
|
|
|
from pydantic import BaseModel, Field, computed_field
|
|
|
|
from ....libs.api.types import MediaItem, UserListStatusType
|
|
from ...utils import converters
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Type aliases
|
|
DownloadStatus = Literal[
|
|
"not_downloaded", "queued", "downloading", "completed", "failed", "paused"
|
|
]
|
|
REGISTRY_VERSION = "1.0"
|
|
|
|
|
|
class MediaEpisode(BaseModel):
|
|
episode_number: str
|
|
|
|
# Download tracking
|
|
download_status: DownloadStatus = "not_downloaded"
|
|
file_path: Path
|
|
download_date: datetime = Field(default_factory=datetime.now)
|
|
|
|
|
|
class MediaRecord(BaseModel):
|
|
media_item: MediaItem
|
|
media_episodes: list[MediaEpisode] = Field(default_factory=list)
|
|
|
|
|
|
class MediaRegistryIndexEntry(BaseModel):
|
|
media_id: int
|
|
media_api: Literal["anilist", "NONE", "jikan"] = "NONE"
|
|
|
|
status: UserListStatusType = "watching"
|
|
progress: str = "0"
|
|
last_watch_position: Optional[str] = None
|
|
last_watched: datetime = Field(default_factory=datetime.now)
|
|
total_duration: Optional[str] = None
|
|
total_episodes: int = 0
|
|
|
|
score: float = 0
|
|
repeat: int = 0
|
|
notes: str = ""
|
|
|
|
last_notified_episode: Optional[str] = None
|
|
|
|
# for first watch only
|
|
start_date: datetime = Field(default_factory=datetime.now)
|
|
completed_at: datetime = Field(default_factory=datetime.now)
|
|
|
|
@computed_field
|
|
@property
|
|
def watch_completion_percentage(self) -> float:
|
|
"""Watch completion percentage."""
|
|
if self.total_duration and self.last_watch_position:
|
|
return (
|
|
converters.time_to_seconds(self.last_watch_position)
|
|
/ converters.time_to_seconds(self.total_duration)
|
|
) * 100
|
|
return 0.0
|
|
|
|
|
|
class MediaRegistryIndex(BaseModel):
|
|
version: str = Field(default=REGISTRY_VERSION)
|
|
last_updated: datetime = Field(default_factory=datetime.now)
|
|
|
|
media_index: Dict[str, MediaRegistryIndexEntry] = Field(default_factory=dict)
|
|
|
|
@computed_field
|
|
@property
|
|
def status_breakdown(self) -> Dict[str, int]:
|
|
"""Get breakdown by user status."""
|
|
breakdown = {}
|
|
for entry in self.media_index.values():
|
|
breakdown[entry.status] = breakdown.get(entry.status, 0) + 1
|
|
return breakdown
|
|
|
|
@computed_field
|
|
@property
|
|
def media_count_breakdown(self) -> Dict[str, int]:
|
|
breakdown = {}
|
|
for entry in self.media_index.values():
|
|
breakdown[entry.media_api] = breakdown.get(entry.media_api, 0) + 1
|
|
return breakdown
|
|
|
|
@computed_field
|
|
@property
|
|
def media_count(self) -> int:
|
|
"""Get the number of media."""
|
|
return len(self.media_index)
|