wizarr/app/services/media/utils.py

"""Shared utilities for media server clients."""

import datetime
import logging

from app.models import Library
from app.services.media.user_details import MediaUserDetails, UserLibraryAccess


class StandardizedPermissions:
    """Helper class for standardized permission mapping."""

    def __init__(self, server_type: str):
        self.server_type = server_type
        self.is_admin = False
        self.allow_downloads = False
        self.allow_live_tv = False
        self.allow_camera_upload = False

    @classmethod
    def for_plex(cls, plex_user) -> "StandardizedPermissions":
        """Extract standardized permissions from Plex user."""
        perms = cls("plex")
        perms.is_admin = getattr(plex_user, "admin", False)
        perms.allow_downloads = getattr(plex_user, "allowSync", True)
        perms.allow_live_tv = getattr(plex_user, "allowChannels", True)
        perms.allow_camera_upload = getattr(plex_user, "allowCameraUpload", False)
        return perms

    @classmethod
    def for_jellyfin(cls, policy: dict) -> "StandardizedPermissions":
        """Extract standardized permissions from Jellyfin/Emby policy."""
        perms = cls("jellyfin")
        perms.is_admin = policy.get("IsAdministrator", False)
        perms.allow_downloads = policy.get("EnableContentDownloading", False)
        perms.allow_live_tv = policy.get("EnableLiveTvAccess", False)
        perms.allow_camera_upload = False  # Not supported
        return perms

    @classmethod
    def for_audiobookshelf(
        cls, permissions: dict, user_type: str = "user"
    ) -> "StandardizedPermissions":
        """Extract standardized permissions from AudiobookShelf.

        Args:
            permissions: The permissions dict from the user object
            user_type: The user type ("root", "admin", or "user")
        """
        perms = cls("audiobookshelf")
        # AudiobookShelf admin status comes from user.type, not permissions.admin
        perms.is_admin = user_type in ("root", "admin")
        perms.allow_downloads = permissions.get("download", False)
        perms.allow_live_tv = False  # Not supported
        perms.allow_camera_upload = False  # Not supported
        return perms

    @classmethod
    def for_navidrome(cls, raw_user: dict) -> "StandardizedPermissions":
        """Extract standardized permissions from Navidrome user."""
        perms = cls("navidrome")
        perms.is_admin = raw_user.get("adminRole", False)
        perms.allow_downloads = raw_user.get("downloadRole", True)
        perms.allow_live_tv = False  # Not supported
        perms.allow_camera_upload = False  # Not supported
        return perms

    @classmethod
    def for_basic_server(
        cls, server_type: str, is_admin: bool = False, allow_downloads: bool = True
    ) -> "StandardizedPermissions":
        """Create basic permissions for servers without complex permission systems."""
        perms = cls(server_type)
        perms.is_admin = is_admin
        perms.allow_downloads = allow_downloads
        perms.allow_live_tv = False
        perms.allow_camera_upload = False
        return perms


class LibraryAccessHelper:
    """Helper for handling library access patterns."""

    @staticmethod
    def create_restricted_access(
        enabled_libraries: list[str], server_id: int
    ) -> list[UserLibraryAccess]:
        """Create library access list for restricted users."""
        if not enabled_libraries:
            return []

        libs_q = (
            Library.query.filter(
                Library.external_id.in_(enabled_libraries),
                Library.server_id == server_id,
            )
            .order_by(Library.name)
            .all()
        )

        return [
            UserLibraryAccess(
                library_id=lib.external_id, library_name=lib.name, has_access=True
            )
            for lib in libs_q
        ]

    @staticmethod
    def create_full_access() -> None:
        """Return None to indicate full access to all libraries."""
        return  # Full access indicator for standardized system

    @staticmethod
    def create_from_sections(
        sections: list[str], server_id: int
    ) -> list[UserLibraryAccess] | None:
        """Create library access from section names (Plex-style)."""
        if not sections:
            return None  # Full access

        # Try to match sections with database libraries
        libs_q = (
            Library.query.filter(
                Library.name.in_(sections),
                Library.server_id == server_id,
            )
            .order_by(Library.name)
            .all()
        )

        library_access = [
            UserLibraryAccess(
                library_id=lib.external_id, library_name=lib.name, has_access=True
            )
            for lib in libs_q
        ]

        # If no DB matches, create from section names directly
        if not library_access and sections:
            library_access = [
                UserLibraryAccess(
                    library_id=f"section_{i}", library_name=section, has_access=True
                )
                for i, section in enumerate(sections)
            ]

        return library_access or None


class DateHelper:
    """Helper for parsing common date formats."""

    @staticmethod
    def parse_iso_date(date_str: str | None) -> datetime.datetime | None:
        """Parse ISO 8601 date string, handling common variations."""
        if not date_str:
            return None

        try:
            # Handle different ISO formats
            if date_str.endswith("Z"):
                date_str = date_str.rstrip("Z")

            # Handle microseconds
            if "." in date_str:
                # Remove microseconds if present
                date_str = date_str.split(".")[0]

            return datetime.datetime.fromisoformat(date_str)
        except (ValueError, AttributeError) as e:
            logging.warning(f"Failed to parse date '{date_str}': {e}")
            return None

    @staticmethod
    def parse_timestamp(timestamp: int | float | None) -> datetime.datetime | None:
        """Parse Unix timestamp (seconds or milliseconds)."""
        if not timestamp:
            return None

        try:
            # Handle milliseconds timestamp
            if timestamp > 1e10:  # Likely milliseconds
                timestamp = timestamp / 1000

            return datetime.datetime.fromtimestamp(timestamp, tz=datetime.UTC)
        except (ValueError, OSError) as e:
            logging.warning(f"Failed to parse timestamp '{timestamp}': {e}")
            return None


def create_standardized_user_details(
    user_id: str,
    username: str,
    email: str | None,
    permissions: StandardizedPermissions,
    library_access: list[UserLibraryAccess] | None,
    created_at: datetime.datetime | None = None,
    last_active: datetime.datetime | None = None,
    is_enabled: bool = True,
) -> MediaUserDetails:
    """Create a standardized MediaUserDetails object."""
    return MediaUserDetails(
        user_id=user_id,
        username=username,
        email=email,
        is_admin=permissions.is_admin,
        is_enabled=is_enabled,
        created_at=created_at,
        last_active=last_active,
        allow_downloads=permissions.allow_downloads,
        allow_live_tv=permissions.allow_live_tv,
        allow_camera_upload=permissions.allow_camera_upload,
        library_access=library_access,
    )