feat(media-api): notifications

fastanime/cli/commands/anilist/cmd.py

@@ -10,6 +10,7 @@ commands = {
     "download": "download.download",
     "auth": "auth.auth",
     "stats": "stats.stats",
+    "notifications": "notifications.notifications",
 }
 
 

fastanime/cli/commands/anilist/commands/notifications.py

@@ -0,0 +1,56 @@
+import click
+from fastanime.core.config import AppConfig
+from rich.console import Console
+from rich.table import Table
+
+
+@click.command(help="Check for new AniList notifications (e.g., for airing episodes).")
+@click.pass_obj
+def notifications(config: AppConfig):
+    """
+    Displays unread notifications from AniList.
+    Running this command will also mark the notifications as read on the AniList website.
+    """
+    from fastanime.cli.service.feedback import FeedbackService
+    from fastanime.libs.media_api.api import create_api_client
+
+    from ....service.auth import AuthService
+
+    feedback = FeedbackService(config.general.icons)
+    console = Console()
+    auth = AuthService(config.general.media_api)
+    api_client = create_api_client(config.general.media_api, config)
+    if profile := auth.get_auth():
+        api_client.authenticate(profile.token)
+
+    if not api_client.is_authenticated():
+        feedback.error(
+            "Authentication Required", "Please log in with 'fastanime anilist auth'."
+        )
+        return
+
+    with feedback.progress("Fetching notifications..."):
+        notifs = api_client.get_notifications()
+
+    if not notifs:
+        feedback.success("All caught up!", "You have no new notifications.")
+        return
+
+    table = Table(
+        title="🔔 AniList Notifications", show_header=True, header_style="bold magenta"
+    )
+    table.add_column("Date", style="dim", width=12)
+    table.add_column("Anime Title", style="cyan")
+    table.add_column("Details", style="green")
+
+    for notif in sorted(notifs, key=lambda n: n.created_at, reverse=True):
+        title = notif.media.title.english or notif.media.title.romaji or "Unknown"
+        date_str = notif.created_at.strftime("%Y-%m-%d")
+        details = f"Episode {notif.episode} has aired!"
+
+        table.add_row(date_str, title, details)
+
+    console.print(table)
+    feedback.info(
+        "Notifications have been marked as read on AniList.",
+    )

fastanime/libs/media_api/anilist/api.py

@@ -25,6 +25,7 @@ from ..types import (
     MediaItem,
     MediaReview,
     MediaSearchResult,
+    Notification,
     UserMediaListStatus,
     UserProfile,
 )
@@ -276,6 +277,20 @@ class AniListApi(BaseApiClient):
             return mapper.to_generic_reviews_list(response.json())
         return None
 
+    def get_notifications(self) -> Optional[List[Notification]]:
+        """Fetches the user's unread notifications from AniList."""
+        if not self.is_authenticated():
+            logger.warning("Cannot fetch notifications: user is not authenticated.")
+            return None
+
+        response = execute_graphql(
+            ANILIST_ENDPOINT, self.http_client, gql.GET_NOTIFICATIONS, {}
+        )
+        if response and "errors" not in response.json():
+            return mapper.to_generic_notifications(response.json())
+        logger.error(f"Failed to fetch notifications: {response.text}")
+        return None
+
     def transform_raw_search_data(self, raw_data: Any) -> Optional[MediaSearchResult]:
         """
         Transform raw AniList API response data into a MediaSearchResult.

fastanime/libs/media_api/anilist/mapper.py

@@ -25,6 +25,8 @@ from ..types import (
     MediaTagItem,
     MediaTitle,
     MediaTrailer,
+    Notification,
+    NotificationType,
     PageInfo,
     Reviewer,
     StreamingEpisode,
@@ -45,6 +47,8 @@ from .types import (
     AnilistMediaTag,
     AnilistMediaTitle,
     AnilistMediaTrailer,
+    AnilistNotification,
+    AnilistNotifications,
     AnilistPageInfo,
     AnilistReview,
     AnilistReviews,
@@ -520,3 +524,70 @@ def to_generic_airing_schedule_result(data: Dict) -> Optional[AiringScheduleResu
     except (KeyError, IndexError, TypeError) as e:
         logger.error(f"Error parsing airing schedule data: {e}")
         return None
+
+
+def _to_generic_media_item_from_notification_partial(
+    data: AnilistBaseMediaDataSchema,
+) -> MediaItem:
+    """
+    A specialized mapper for the partial MediaItem object received in notifications.
+    It provides default values for fields not present in the notification's media payload.
+    """
+    return MediaItem(
+        id=data["id"],
+        id_mal=data.get("idMal"),
+        title=_to_generic_media_title(data["title"]),
+        cover_image=_to_generic_media_image(data["coverImage"]),
+        # Provide default/empty values for fields not in notification payload
+        type="ANIME",
+        status=MediaStatus.RELEASING,  # Assume releasing for airing notifications
+        format=None,
+        description=None,
+        episodes=None,
+        duration=None,
+        genres=[],
+        tags=[],
+        studios=[],
+        synonymns=[],
+        average_score=None,
+        popularity=None,
+        favourites=None,
+        streaming_episodes={},
+        user_status=None,
+    )
+
+
+def _to_generic_notification(anilist_notification: AnilistNotification) -> Notification:
+    """Maps a single AniList notification to a generic Notification object."""
+    return Notification(
+        id=anilist_notification["id"],
+        type=NotificationType(anilist_notification["type"]),
+        episode=anilist_notification.get("episode"),
+        contexts=anilist_notification.get("contexts", []),
+        created_at=datetime.fromtimestamp(anilist_notification["createdAt"]),
+        media=_to_generic_media_item_from_notification_partial(
+            anilist_notification["media"]
+        ),
+    )
+
+
+def to_generic_notifications(
+    data: AnilistNotifications,
+) -> Optional[List[Notification]]:
+    """Top-level mapper for a list of notifications."""
+    if not data or "data" not in data:
+        return None
+
+    page_data = data["data"].get("Page", {})
+    if not page_data:
+        return None
+
+    raw_notifications = page_data.get("notifications", [])
+    if not raw_notifications:
+        return []
+
+    return [
+        _to_generic_notification(notification)
+        for notification in raw_notifications
+        if notification
+    ]

fastanime/libs/media_api/anilist/types.py

@@ -1,4 +1,4 @@
-from typing import Literal, Optional, TypedDict
+from typing import List, Literal, Optional, TypedDict
 
 
 class AnilistPageInfo(TypedDict):
@@ -226,28 +226,6 @@ class AnilistDataSchema(TypedDict):
     data: AnilistPages
 
 
-class AnilistNotification(TypedDict):
-    id: int
-    type: str
-    episode: int
-    context: str
-    createdAt: str
-    media: AnilistBaseMediaDataSchema
-
-
-class AnilistNotificationPage(TypedDict):
-    pageInfo: AnilistPageInfo
-    notifications: list[AnilistNotification]
-
-
-class AnilistNotificationPages(TypedDict):
-    Page: AnilistNotificationPage
-
-
-class AnilistNotifications(TypedDict):
-    data: AnilistNotificationPages
-
-
 class AnilistMediaList(TypedDict):
     media: AnilistBaseMediaDataSchema
     status: AnilistMediaListStatus
@@ -271,3 +249,25 @@ class AnilistMediaListPages(TypedDict):
 
 class AnilistMediaLists(TypedDict):
     data: AnilistMediaListPages
+
+
+class AnilistNotification(TypedDict):
+    id: int
+    type: str
+    episode: int
+    contexts: List[str]
+    createdAt: int  # This is a Unix timestamp
+    media: AnilistBaseMediaDataSchema  # This will be a partial response
+
+
+class AnilistNotificationPage(TypedDict):
+    pageInfo: AnilistPageInfo
+    notifications: list[AnilistNotification]
+
+
+class AnilistNotificationPages(TypedDict):
+    Page: AnilistNotificationPage
+
+
+class AnilistNotifications(TypedDict):
+    data: AnilistNotificationPages

fastanime/libs/media_api/base.py

@@ -18,6 +18,7 @@ from .types import (
     MediaItem,
     MediaReview,
     MediaSearchResult,
+    Notification,
     UserProfile,
 )
 
@@ -95,6 +96,11 @@ class BaseApiClient(abc.ABC):
     ) -> Optional[List[MediaReview]]:
         pass
 
+    @abc.abstractmethod
+    def get_notifications(self) -> Optional[List[Notification]]:
+        """Fetches the user's unread notifications."""
+        pass
+
     @abc.abstractmethod
     def transform_raw_search_data(self, raw_data: Dict) -> Optional[MediaSearchResult]:
         """

fastanime/libs/media_api/jikan/api.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import logging
 from typing import TYPE_CHECKING, List, Optional
 
@@ -20,6 +18,7 @@ from ..types import (
     MediaItem,
     MediaSearchResult,
     MediaTitle,
+    Notification,
     UserProfile,
 )
 from . import mapper
@@ -183,6 +182,11 @@ class JikanApi(BaseApiClient):
             logger.error(f"Failed to fetch related anime for media {params.id}: {e}")
             return None
 
+    def get_notifications(self) -> Optional[List[Notification]]:
+        """Jikan is a public API and does not support user notifications."""
+        logger.warning("Jikan API does not support fetching user notifications.")
+        return None
+
     def get_airing_schedule_for(
         self, params: MediaAiringScheduleParams
     ) -> Optional[AiringScheduleResult]:

fastanime/libs/media_api/types.py

@@ -65,6 +65,13 @@ class MediaFormat(Enum):
     ONE_SHOT = "ONE_SHOT"
 
 
+class NotificationType(Enum):
+    AIRING = "AIRING"
+    RELATED_MEDIA_ADDITION = "RELATED_MEDIA_ADDITION"
+    MEDIA_DATA_CHANGE = "MEDIA_DATA_CHANGE"
+    # ... add other types as needed
+
+
 # MODELS
 class BaseMediaApiModel(BaseModel):
     model_config = ConfigDict(frozen=True)
@@ -227,6 +234,17 @@ class MediaItem(BaseMediaApiModel):
     user_status: Optional[UserListItem] = None
 
 
+class Notification(BaseMediaApiModel):
+    """A generic representation of a user notification."""
+
+    id: int
+    type: NotificationType
+    episode: Optional[int] = None
+    contexts: List[str] = Field(default_factory=list)
+    created_at: datetime
+    media: MediaItem
+
+
 class PageInfo(BaseMediaApiModel):
     """Generic pagination information."""