wizarr/app/services/media/plex.py

import logging
import re
import threading
from typing import TYPE_CHECKING, Any

import structlog
from cachetools import TTLCache, cached
from plexapi.myplex import MyPlexAccount
from plexapi.server import PlexServer

from app.extensions import db
from app.models import Invitation, Library, MediaServer, User
from app.services.media.service import get_client_for_media_server
from app.services.notifications import notify

from .client_base import MediaClient, register_media_client
from .plex_custom import accept_invite_v2, update_shared_server

if TYPE_CHECKING:
    from app.services.media.user_details import MediaUserDetails


# Patch PlexAPI's acceptInvite method with our custom v2 implementation
MyPlexAccount.acceptInvite = accept_invite_v2  # type: ignore


def extract_plex_error_message(exception) -> str:
    """
    Extract human-readable error message from Plex API exceptions.

    Args:
        exception: Exception from plexapi

    Returns:
        Human-readable error message
    """
    error_message = str(exception)

    # Look for XML response with status attribute in the error message
    # Format: plexapi.exceptions.BadRequest: (400) bad_request; https://... <Response code="400" status="Error message"/>
    xml_pattern = r'<Response[^>]+status="([^"]*)"[^>]*/?>'
    xml_match = re.search(xml_pattern, error_message)
    if xml_match:
        return xml_match.group(1)

    # Look for JSON response patterns if XML doesn't work
    # This handles cases where Plex returns JSON errors
    json_pattern = r'"message":\s*"([^"]*)"'
    json_match = re.search(json_pattern, error_message)
    if json_match:
        return json_match.group(1)

    # Look for simple status messages in parentheses
    # Format: (400) some_error_message; ...
    status_pattern = r"\(\d+\)\s+([^;]+);"
    status_match = re.search(status_pattern, error_message)
    if status_match:
        error_text = status_match.group(1).strip()
        # Convert snake_case to readable text
        return error_text.replace("_", " ").title()

    # Fallback to the original exception message but clean it up
    if hasattr(exception, "message"):
        return str(exception.message)

    # Last resort: return a cleaned up version of the exception string
    clean_message = (
        error_message.split(";")[0] if ";" in error_message else error_message
    )
    clean_message = clean_message.replace("plexapi.exceptions.", "").replace(
        "BadRequest: ", ""
    )

    # Check if error is just a library/user ID (strip quotes first)
    stripped = clean_message.strip().strip("'\"")
    if stripped.isdigit():
        return f"Invalid library ID '{stripped}'. Check invitation library settings."

    return clean_message


class PlexInvitationError(Exception):
    """Custom exception for Plex invitation errors with user-friendly messages."""

    def __init__(self, message: str, original_exception=None):
        self.message = message
        self.original_exception = original_exception
        super().__init__(message)


@register_media_client("plex")
class PlexClient(MediaClient):
    """Wrapper that connects to Plex using admin credentials."""

    def __init__(self, *args, **kwargs):
        if "url_key" not in kwargs:
            kwargs["url_key"] = "server_url"
        if "token_key" not in kwargs:
            kwargs["token_key"] = "api_key"  # noqa: S105  # Parameter name, not actual password

        super().__init__(*args, **kwargs)
        self._server = None
        self._admin = None

    @property
    def server(self) -> PlexServer:
        if self._server is None:
            self._server = PlexServer(self.url, self.token)
        return self._server

    @property
    def admin(self) -> MyPlexAccount:
        if self._admin is None:
            try:
                self._admin = MyPlexAccount(token=self.token)
            except Exception as e:
                logging.error(f"Failed to connect to Plex MyPlexAccount: {e}")
                # Instead of raising, return None and let callers handle it
                raise ConnectionError(f"Unable to connect to Plex servers: {e}") from e
        return self._admin

    def libraries(self) -> dict[str, str]:
        """Get all libraries with their global IDs.

        Returns:
            dict: Mapping of {global_id: library_name} for database storage
                  where global_id is used as external_id in the Library model
        """
        # Get global library IDs from Plex API
        library_map = self._get_all_library_global_ids()

        # Return {global_id: name} so external_id stores the global ID
        return {str(global_id): name for name, global_id in library_map.items()}

    # ─── Helper Methods ────────────────────────────────────────────────────────

    def _get_server_users(self) -> list[User]:
        """Get all users for this server from database."""
        return db.session.query(User).filter(User.server_id == self.server_id).all()

    def _extract_plex_permissions(self, plex_user) -> dict[str, bool]:
        """Extract all permissions from a Plex user object."""
        return {
            "is_admin": getattr(plex_user, "admin", False),
            "allow_downloads": getattr(plex_user, "allowSync", False),
            "allow_live_tv": getattr(plex_user, "allowChannels", False),
            "allow_camera_upload": getattr(plex_user, "allowCameraUpload", False),
        }

    def _filter_users_for_server(self, admin_users, server_id: str) -> dict[str, Any]:
        """Filter Plex users who have access to this specific server."""
        users_by_email = {}
        for plex_user in admin_users:
            email = getattr(plex_user, "email", None)
            servers = getattr(plex_user, "servers", []) or []

            if email and any(s.machineIdentifier == server_id for s in servers):
                users_by_email[email] = plex_user

        return users_by_email

    def _sync_user_permissions(self, user: User, plex_user) -> None:
        """Sync permissions and library access from Plex to database user."""
        # Update basic info
        user.username = getattr(plex_user, "title", user.username)
        user.photo = getattr(plex_user, "thumb", None)

        # Update permissions
        permissions = self._extract_plex_permissions(plex_user)
        user.is_admin = permissions["is_admin"]
        user.allow_downloads = permissions["allow_downloads"]
        user.allow_live_tv = permissions["allow_live_tv"]
        user.allow_camera_upload = permissions["allow_camera_upload"]

        # Update library access
        library_names, has_full_access = self._get_user_library_access(plex_user)
        if has_full_access:
            user.set_accessible_libraries(None)
        else:
            user.set_accessible_libraries(library_names or [])

    # ─── Library Access Methods ────────────────────────────────────────────────

    def _get_user_library_access(self, plex_user) -> tuple[list[str] | None, bool]:
        """Extract library access: (library_names | None, has_full_access)."""
        # Find this server's share
        matching_share = next(
            (
                s
                for s in getattr(plex_user, "servers", []) or []
                if getattr(s, "machineIdentifier", None)
                == self.server.machineIdentifier
            ),
            None,
        )

        if not matching_share:
            logging.warning(
                f"PLEX: No server share found for user {getattr(plex_user, 'email', 'unknown')}"
            )
            return [], False

        # Check if user has full access
        if getattr(matching_share, "allLibraries", False):
            return None, True

        return self._extract_library_names_from_share(matching_share), False

    def _extract_library_names_from_share(self, server_share) -> list[str]:
        """Extract library names from Plex server share object."""
        if not server_share:
            return []

        # Get sections (API varies by plexapi version)
        try:
            sections_data = server_share.sections()
        except (TypeError, AttributeError):
            sections_data = getattr(server_share, "sections", None)

        if not sections_data:
            return []

        # Handle callable sections
        if callable(sections_data):
            try:
                sections_data = sections_data()
            except Exception as exc:
                logging.warning(f"PLEX: Failed to call sections(): {exc}")
                return []

        # Extract library names (skip non-shared sections)
        return [
            section.title
            for section in sections_data or []
            if getattr(section, "title", None)
            and not (hasattr(section, "shared") and section.shared is False)
        ]

    def get_movie_posters(self, limit: int = 10) -> list[str]:
        """Get movie poster URLs for background display."""
        if not self.url:
            return []

        poster_urls = []
        try:
            # Get movie libraries
            for library in self.server.library.sections():
                if library.type == "movie":
                    # Get recent movies from this library
                    movies = library.recentlyAdded(maxresults=limit)
                    for movie in movies[:limit]:
                        poster_url = None

                        if hasattr(movie, "posterUrl") and movie.posterUrl:
                            poster_url = movie.posterUrl
                        elif hasattr(movie, "thumb") and movie.thumb:
                            # Fallback to thumb
                            poster_url = movie.thumb

                        if poster_url:
                            # Convert to full URL if needed
                            if poster_url.startswith("/"):
                                poster_url = f"{self.url.rstrip('/')}{poster_url}"

                            # Generate secure proxy URL with opaque token
                            proxied_url = self.generate_image_proxy_url(poster_url)
                            poster_urls.append(proxied_url)

                        if len(poster_urls) >= limit:
                            break

                    if len(poster_urls) >= limit:
                        break
        except Exception as e:
            # Log error but don't break the login process
            import logging

            logging.warning(f"Failed to fetch movie posters: {e}")

        return poster_urls[:limit]

    def get_recent_items(
        self, library_id: str | None = None, limit: int = 10
    ) -> list[dict]:
        """Get recently added items from Plex server."""
        if not self.url:
            return []

        try:
            items = []

            # Get all library sections or specific library if provided
            if library_id:
                try:
                    library = self.server.library.sectionByID(library_id)
                    libraries = [library] if library else []
                except Exception:
                    libraries = []
            else:
                libraries = list(self.server.library.sections())

            for library in libraries:
                if len(items) >= limit:
                    break

                try:
                    # Get recently added items from this library
                    recent_items = library.recentlyAdded(maxresults=limit - len(items))

                    for item in recent_items:
                        if len(items) >= limit:
                            break

                        # Only use posterUrl - skip items without proper posters
                        thumb_url = None
                        if hasattr(item, "posterUrl") and item.posterUrl:
                            thumb_url = item.posterUrl

                            # Convert relative URLs to full URLs
                            if thumb_url.startswith("/"):
                                thumb_url = f"{self.url.rstrip('/')}{thumb_url}"

                            # Generate secure proxy URL with opaque token
                            thumb_url = self.generate_image_proxy_url(thumb_url)

                            # Extract year from releaseDate
                            year = None
                            if hasattr(item, "year") and item.year:
                                year = item.year
                            elif (
                                hasattr(item, "originallyAvailableAt")
                                and item.originallyAvailableAt
                            ):
                                from contextlib import suppress

                                with suppress(Exception):
                                    year = item.originallyAvailableAt.year

                            # Get item type
                            item_type = getattr(item, "type", "unknown").lower()

                            # Get added date
                            added_at = None
                            if hasattr(item, "addedAt") and item.addedAt:
                                from contextlib import suppress

                                with suppress(Exception):
                                    # Convert to ISO format string like Jellyfin
                                    added_at = item.addedAt.isoformat() + "Z"

                            # Only add items that have poster images
                            items.append(
                                {
                                    "title": getattr(item, "title", "Unknown"),
                                    "year": year,
                                    "thumb": thumb_url,
                                    "type": item_type,
                                    "added_at": added_at,
                                }
                            )

                except Exception as exc:
                    logging.debug(f"Failed to process Plex media item: {exc}")
                    continue

            return items

        except Exception:
            return []

    def scan_libraries(
        self, url: str | None = None, token: str | None = None
    ) -> dict[str, str]:
        if url and token:
            try:
                from plexapi.server import PlexServer

                temp_server = PlexServer(url, token)
                return {lib.title: lib.title for lib in temp_server.library.sections()}
            except Exception as e:
                logging.error(f"Failed to scan Plex libraries: {e}")
                return {}
        else:
            return self.libraries()

    def create_user(self, *args, **kwargs):
        raise NotImplementedError(
            "PlexClient does not support create_user; use invite_friend or invite_home"
        )

    def _do_join(
        self,
        username: str,  # noqa: ARG002
        password: str,  # noqa: ARG002
        confirm: str,  # noqa: ARG002
        email: str,  # noqa: ARG002
        code: str,  # noqa: ARG002
    ) -> tuple[bool, str]:
        """Interface method - not implemented for Plex (uses OAuth instead)."""
        return (
            False,
            "Plex does not support direct user creation. Users must be invited via email.",
        )

    def invite_friend(
        self,
        email: str,
        sections: list[str],
        allow_sync: bool,
        allow_channels: bool,
        allow_camera_upload: bool = False,
    ):
        try:
            self.admin.inviteFriend(
                user=email,
                server=self.server,
                sections=sections,
                allowSync=allow_sync,
                allowChannels=allow_channels,
                allowCameraUpload=allow_camera_upload,
            )
        except Exception as e:
            # Extract human-readable error message and raise custom exception
            error_message = extract_plex_error_message(e)
            logging.error(f"Failed to invite friend {email}: {error_message}")
            raise PlexInvitationError(error_message, e) from e

    def invite_home(
        self,
        email: str,
        sections: list[str],
        allow_sync: bool,
        allow_channels: bool,
        allow_camera_upload: bool = False,
    ):
        try:
            self.admin.createExistingUser(
                user=email,
                server=self.server,
                sections=sections,
                allowSync=allow_sync,
                allowChannels=allow_channels,
                allowCameraUpload=allow_camera_upload,
            )
        except Exception as e:
            # Extract human-readable error message and raise custom exception
            error_message = extract_plex_error_message(e)
            logging.error(f"Failed to invite home user {email}: {error_message}")
            raise PlexInvitationError(error_message, e) from e

    def get_user(self, db_id: int) -> dict:
        """Get user info in legacy format for backward compatibility."""
        details = self.get_user_details(db_id)

        # Convert to legacy Plex format
        return {
            "Name": details.username,
            "Id": details.user_id,
            "Configuration": {
                "admin": details.is_admin,
                "allowSync": details.allow_downloads,
                "allowChannels": details.allow_live_tv,
                "allowCameraUpload": details.allow_camera_upload,
            },
            "Policy": {},
        }

    def get_user_details(self, user_identifier: str | int) -> "MediaUserDetails":
        """Get detailed user information from database (no API calls)."""
        from app.services.media.user_details import MediaUserDetails, UserLibraryAccess

        user = db.session.get(User, user_identifier)
        if not user:
            raise ValueError(f"No user found with id {user_identifier}")

        # Build library access from stored names
        library_names = user.get_accessible_libraries()
        library_access = None

        if library_names:
            libs_by_name = {
                lib.name: lib
                for lib in Library.query.filter(
                    Library.server_id == self.server_id,
                    Library.name.in_(library_names),
                ).all()
            }
            library_access = [
                UserLibraryAccess(
                    library_id=lib.external_id
                    if (lib := libs_by_name.get(name))
                    else f"plex_{name}",
                    library_name=name,
                    has_access=True,
                )
                for name in library_names
            ]

        return MediaUserDetails(
            user_id=str(user.id),
            username=user.username,
            email=user.email,
            is_admin=user.is_admin or False,
            is_enabled=True,
            created_at=None,
            last_active=None,
            allow_downloads=user.allow_downloads or False,
            allow_live_tv=user.allow_live_tv or False,
            allow_camera_upload=user.allow_camera_upload or False,
            library_access=library_access,
        )

    def update_user(self, info: dict, form: dict) -> None:
        self.admin.updateFriend(
            info["Name"],
            self.server,
            allowSync=bool(form.get("allowSync")),
            allowChannels=bool(form.get("allowChannels")),
            allowCameraUpload=bool(form.get("allowCameraUpload")),
        )

    def _get_all_library_global_ids(self) -> dict[str, int]:
        """Get mapping of all server libraries (name -> global ID).

        This fetches library information from the Plex server's own libraries endpoint,
        which includes the global library section IDs needed for the sharing API.

        Returns:
            dict: Mapping of library title to global ID {title: id}
        """
        try:
            base = "https://plex.tv"
            url = f"{base}/api/v2/servers/{self.server.machineIdentifier}"

            params = {
                "X-Plex-Product": "Wizarr",
                "X-Plex-Version": "1.0",
                "X-Plex-Client-Identifier": self.admin.uuid,
                "X-Plex-Token": self.admin.authToken,
                "X-Plex-Platform": "Web",
                "X-Plex-Features": "external-media,indirect-media,hub-style-list",
                "X-Plex-Language": "en",
            }

            headers = {"Accept": "application/json"}

            resp = self.admin._session.get(url, params=params, headers=headers)
            resp.raise_for_status()
            server_data = resp.json()

            # Extract libraries from server data (Plex uses 'librarySections' key)
            libraries = server_data.get("librarySections", [])

            library_map = {}

            for lib in libraries:
                title = lib.get("title")
                lib_id = lib.get("id")
                if title and lib_id:
                    library_map[title] = lib_id

            logging.debug(f"Found {len(library_map)} libraries with global IDs")
            return library_map

        except Exception as e:
            logging.error(f"Failed to get library global IDs: {e}", exc_info=True)
            return {}

    def _get_share_data(self, email: str) -> dict | None:
        """Get the complete share data for a user.

        Returns the full share object which includes the shared_server ID
        and the library mappings with their global IDs.

        Args:
            email: User's email address

        Returns:
            dict: The share data, or None if not found
        """
        try:
            # First, try to get the Plex user object to find their ID
            plex_user_id = None
            try:
                plex_user = self.admin.user(email)
                if plex_user:
                    plex_user_id = getattr(plex_user, "id", None)
            except Exception as exc:
                # If getting by email fails, might not exist or different identifier
                logging.debug(f"Could not get Plex user by email {email}: {exc}")

            # GET the list of owned/accepted shares
            base = "https://clients.plex.tv"
            url = f"{base}/api/v2/shared_servers/owned/accepted"

            params = {
                "X-Plex-Product": "Wizarr",
                "X-Plex-Version": "1.0",
                "X-Plex-Client-Identifier": self.admin.uuid,
                "X-Plex-Token": self.admin.authToken,
                "X-Plex-Platform": "Web",
                "X-Plex-Platform-Version": "1.0",
                "X-Plex-Features": "external-media,indirect-media,hub-style-list",
                "X-Plex-Language": "en",
            }

            headers = {"Accept": "application/json"}

            resp = self.admin._session.get(url, params=params, headers=headers)
            resp.raise_for_status()
            shared_servers = resp.json()

            # Find the share matching this server and user
            for share in shared_servers:
                # Check if this share is for our server
                if share.get("machineIdentifier") != self.server.machineIdentifier:
                    continue

                # Try matching by Plex user ID first (most reliable)
                if plex_user_id and share.get("invitedId") == plex_user_id:
                    return share

                # Fallback: Try matching by email/username
                invited = share.get("invited", {})
                invited_email = (
                    invited.get("email")
                    or invited.get("username")
                    or share.get("invitedEmail")
                )

                if invited_email and invited_email.lower() == email.lower():
                    return share

            logging.warning(
                f"No shared_server found for {email} on server {self.server.friendlyName}"
            )
            return None
        except Exception as e:
            logging.error(f"Failed to get share data for {email}: {e}")
            return None

    def _get_shared_server_id(self, email: str) -> int | None:
        """Get the shared_server ID from the admin's perspective."""
        share = self._get_share_data(email)
        return share.get("id") if share else None

    def _get_current_plex_state(self, email: str) -> tuple[dict, list | None]:
        """Get current permissions and library access for a Plex user.

        Args:
            email: User's email address

        Returns:
            Tuple of (permissions_dict, sections_list)
            permissions_dict has keys: allow_downloads, allow_live_tv, allow_camera_upload
            sections_list is either None or a list of LibrarySection objects
        """
        plex_user = self.admin.user(email)
        if not plex_user:
            raise ValueError(f"Plex user not found: {email}")

        # Extract current permissions
        permissions = self._extract_plex_permissions(plex_user)

        # Find this server's share for the user
        matching_share = next(
            (
                s
                for s in getattr(plex_user, "servers", []) or []
                if getattr(s, "machineIdentifier", None)
                == self.server.machineIdentifier
            ),
            None,
        )

        # Get current sections (libraries) for this user
        sections = None
        if matching_share:
            library_names = self._extract_library_names_from_share(matching_share)
            # Convert library names to section objects
            if library_names:
                all_sections = self.server.library.sections()
                sections = [s for s in all_sections if s.title in library_names]

        return permissions, sections

    def update_user_permissions(
        self, _user_identifier: str, _permissions: dict[str, bool]
    ) -> bool:
        """Update user permissions on Plex using the shared_servers API.

        Args:
            _user_identifier: User's email address
            _permissions: Dict with keys: allow_downloads, allow_live_tv, allow_camera_upload

        Returns:
            bool: True if successful, False otherwise
        """
        try:
            # Get the shared_server ID
            shared_server_id = self._get_shared_server_id(_user_identifier)
            if not shared_server_id:
                logging.error(f"Could not find shared_server ID for {_user_identifier}")
                return False

            # Get current library section IDs to preserve them
            # Use share data to get the global library IDs
            share = self._get_share_data(_user_identifier)
            if not share:
                logging.error(f"Could not get share data for {_user_identifier}")
                return False

            section_ids = [lib["id"] for lib in share.get("libraries", [])]

            # Build settings with new permissions
            settings = {
                "allowSync": _permissions.get("allow_downloads", False),
                "allowChannels": _permissions.get("allow_live_tv", False),
                "allowCameraUpload": _permissions.get("allow_camera_upload", False),
                "filterMovies": "",
                "filterMusic": "",
                "filterPhotos": None,
                "filterTelevision": "",
                "filterAll": None,
                "allowSubtitleAdmin": False,
                "allowTuners": 0,
            }

            # Call the custom API method
            success = update_shared_server(
                self.admin, shared_server_id, settings, section_ids
            )

            if success:
                logging.info(
                    f"Successfully updated permissions for {_user_identifier} via shared_servers API"
                )

            return success
        except Exception as e:
            logging.error(f"Failed to update permissions for {_user_identifier}: {e}")
            return False

    def update_user_libraries(
        self, _user_identifier: str, _library_names: list[str] | None
    ) -> bool:
        """Update user's library access on Plex using the shared_servers API.

        Args:
            _user_identifier: User's email address
            _library_names: List of library names to grant access to, or None for all libraries

        Returns:
            bool: True if successful, False otherwise
        """
        try:
            # Get the shared_server ID
            shared_server_id = self._get_shared_server_id(_user_identifier)
            if not shared_server_id:
                logging.error(f"Could not find shared_server ID for {_user_identifier}")
                return False

            # Get current permissions to preserve them
            current_perms, _ = self._get_current_plex_state(_user_identifier)

            # Get the share data to access library ID mappings
            share = self._get_share_data(_user_identifier)
            if not share:
                logging.error(f"Could not get share data for {_user_identifier}")
                return False

            # Log current share state
            current_libs = share.get("libraries", [])
            logging.info(
                f"Current libraries in share: {[lib['title'] for lib in current_libs]}"
            )
            logging.info(
                f"Current library IDs in share: {[lib['id'] for lib in current_libs]}"
            )

            # Get library global IDs from database (external_id stores the global ID)
            # This assumes libraries have been scanned and stored correctly
            from app.models import Library

            section_ids = []
            if _library_names is not None:
                logging.info(f"Requested libraries: {_library_names}")
                libraries = (
                    Library.query.filter_by(server_id=self.server_id)
                    .filter(Library.name.in_(_library_names))
                    .all()
                )

                for lib in libraries:
                    section_ids.append(int(lib.external_id))
                    logging.info(f"  ✓ {lib.name} -> {lib.external_id}")

                # Check for missing libraries
                found_names = {lib.name for lib in libraries}
                missing = set(_library_names) - found_names
                for name in missing:
                    logging.warning(
                        f"  ✗ Library '{name}' not found in database (scan libraries to fix)"
                    )

                logging.info(f"Converted to section IDs: {section_ids}")
            else:
                # None means all libraries - get all enabled libraries for this server
                libraries = Library.query.filter_by(
                    server_id=self.server_id, enabled=True
                ).all()
                section_ids = [int(lib.external_id) for lib in libraries]
                logging.info(f"Using all library IDs: {section_ids}")

            # Build settings with preserved permissions
            settings = {
                "allowSync": current_perms["allow_downloads"],
                "allowChannels": current_perms["allow_live_tv"],
                "allowCameraUpload": current_perms["allow_camera_upload"],
                "filterMovies": "",
                "filterMusic": "",
                "filterPhotos": None,
                "filterTelevision": "",
                "filterAll": None,
                "allowSubtitleAdmin": False,
                "allowTuners": 0,
            }

            # Call the custom API method
            success = update_shared_server(
                self.admin, shared_server_id, settings, section_ids
            )

            if success:
                logging.info(
                    f"Successfully updated library access for {_user_identifier} via shared_servers API"
                )

            return success
        except Exception as e:
            logging.error(
                f"Failed to update library access for {_user_identifier}: {e}"
            )
            return False

    def enable_user(self, user_id: str) -> bool:  # noqa: ARG002
        """Enable a user account on Plex.

        Args:
            user_id: The user's Plex ID (unused - Plex doesn't support enable/disable)

        Returns:
            bool: True if the user was successfully enabled, False otherwise
        """
        try:
            # Plex doesn't have a direct disable/enable feature
            # Return False to indicate this operation is not supported
            structlog.get_logger().warning(
                "Plex does not support disabling/enabling users"
            )
            return False
        except Exception as e:
            structlog.get_logger().error(f"Failed to enable Plex user: {e}")
            return False

    def disable_user(self, user_id: str) -> bool:
        """Disable a user account on Plex.

        Note: Plex doesn't have a direct disable feature for managed users.
        This implementation removes all library access which effectively disables the account.

        Args:
            user_id: The user's Plex ID

        Returns:
            bool: True if the user was successfully disabled, False otherwise
        """
        try:
            # For Plex, we remove all library access to effectively disable the user
            user = self.admin.user(user_id)
            if user:
                user.removeFriend()
                return True
            return False
        except Exception as e:
            structlog.get_logger().error(f"Failed to disable Plex user: {e}")
            return False

    def delete_user(self, email: str) -> None:
        """Remove a user from the Plex server."""
        try:
            self.admin.removeHomeUser(email)
        except Exception:
            try:
                self.admin.removeFriend(email)
            except Exception as e:
                logging.error("Error removing friend: %s", e)

    @cached(cache=TTLCache(maxsize=1024, ttl=600))
    def list_users(self) -> list[User]:
        """Sync users from Plex to database with all permissions and library access."""
        try:
            admin_users = self.admin.users()
        except (ConnectionError, Exception) as e:
            logging.error(f"Failed to connect to Plex admin API: {e}")
            return self._get_server_users()

        plex_users_by_email = self._filter_users_for_server(
            admin_users, self.server.machineIdentifier
        )

        # Remove users no longer in Plex, add new users
        known_emails = set(plex_users_by_email.keys())
        for db_user in self._get_server_users():
            if db_user.email not in known_emails:
                db.session.delete(db_user)

        for email, plex_user in plex_users_by_email.items():
            if not User.query.filter_by(email=email, server_id=self.server_id).first():
                db.session.add(
                    User(
                        email=email,
                        username=getattr(plex_user, "title", "Unknown"),
                        token="None",  # noqa: S106  # Placeholder string, not actual password
                        code="None",
                        server_id=self.server_id,
                    )
                )

        # Sync all permissions and library access
        for user in self._get_server_users():
            if plex_user := plex_users_by_email.get(user.email):
                self._sync_user_permissions(user, plex_user)

        try:
            db.session.commit()
            logging.info(f"Synced {len(plex_users_by_email)} Plex users to database")
        except Exception as e:
            logging.error(f"Failed to sync Plex user metadata: {e}")
            db.session.rollback()

        return self._get_server_users()

    def _get_user_identifier_for_details(self, user: User) -> str | int | None:
        """Plex uses database ID for get_user_details."""
        return user.id

    def now_playing(self) -> list[dict]:
        try:
            sessions = self.server.sessions()
            now_playing_sessions = []

            for session in sessions:
                view_offset = getattr(session, "viewOffset", None)
                if view_offset is None:
                    continue

                progress = 0.0
                duration = getattr(session, "duration", None)
                if duration and view_offset:
                    progress = max(0.0, min(1.0, view_offset / duration))

                media_type = getattr(session, "type", "unknown").lower()

                media_title = getattr(session, "title", "Unknown")
                if media_type == "episode":
                    grandparent_title = getattr(session, "grandparentTitle", "")
                    season_num = getattr(session, "parentIndex", None)
                    episode_num = getattr(session, "index", None)
                    if grandparent_title:
                        media_title = f"{grandparent_title}"
                        if season_num and episode_num:
                            media_title += f" S{season_num:02d}E{episode_num:02d}"
                        media_title += f" - {getattr(session, 'title', '')}"

                players = getattr(session, "players", [])
                state = "stopped"
                if players:
                    player_state = getattr(players[0], "state", "stopped")
                    state = {
                        "paused": "paused",
                        "playing": "playing",
                        "buffering": "buffering",
                    }.get(player_state, "stopped")

                user_info = "Unknown User"
                usernames = getattr(session, "usernames", None)
                users = getattr(session, "users", None)
                if usernames:
                    user_info = usernames[0]
                elif users:
                    user_info = users[0].title

                client_name = device_name = ""
                if players:
                    client_name = getattr(players[0], "product", "")
                    device_name = getattr(players[0], "title", "")

                artwork_url = None
                images_attr = getattr(session, "image", None)
                if images_attr:
                    images_list = (
                        images_attr
                        if isinstance(images_attr, list | tuple | set)
                        else [images_attr]
                    )
                    for img in images_list:
                        if getattr(img, "type", None) == "coverPoster":
                            img_key = getattr(img, "key", None) or getattr(
                                img, "thumb", None
                            )
                            if img_key:
                                artwork_url = (
                                    img_key
                                    if str(img_key).startswith("http")
                                    else self.server.url(img_key, includeToken=True)
                                )
                            elif getattr(img, "thumbUrl", None):
                                artwork_url = img.thumbUrl
                            elif getattr(img, "url", None):
                                artwork_url = img.url
                            if artwork_url:
                                break

                for attr in ("grandparentThumb", "parentThumb", "art"):
                    if artwork_url is not None:
                        break
                    val = getattr(session, attr, None)
                    if val:
                        artwork_url = (
                            val
                            if str(val).startswith("http")
                            else self.server.url(val, includeToken=True)
                        )

                thumb_url = getattr(session, "thumbUrl", None)
                if artwork_url is None and thumb_url:
                    artwork_url = thumb_url

                # Check for transcoding using Plex API structure
                is_transcoding = False
                transcode_speed = None
                media_list = getattr(session, "media", [])

                # Method 1: Check for TranscodeSession using python-plexapi properties
                # This is the most reliable method as it directly reflects the actual transcoding state
                transcode_session = getattr(session, "transcodeSession", None)
                if transcode_session:
                    # TranscodeSession object has videoDecision and audioDecision attributes
                    video_decision = getattr(transcode_session, "videoDecision", None)
                    audio_decision = getattr(transcode_session, "audioDecision", None)

                    # Only consider it transcoding if either video or audio is actually being transcoded
                    # "copy" and "direct" mean no transcoding is happening
                    if video_decision == "transcode" or audio_decision == "transcode":
                        is_transcoding = True
                        transcode_speed = getattr(transcode_session, "speed", None)

                # Method 2: Check transcodeSessions list property (fallback)
                if not is_transcoding and not transcode_session:
                    transcode_sessions = getattr(session, "transcodeSessions", [])
                    if transcode_sessions:
                        # Check each transcode session for actual transcoding
                        for ts in transcode_sessions:
                            ts_video_decision = getattr(ts, "videoDecision", None)
                            ts_audio_decision = getattr(ts, "audioDecision", None)
                            if (
                                ts_video_decision == "transcode"
                                or ts_audio_decision == "transcode"
                            ):
                                is_transcoding = True
                                transcode_speed = getattr(ts, "speed", None)
                                break

                video_codec = audio_codec = container = video_resolution = None
                if media_list:
                    media_obj = media_list[0]
                    video_codec = getattr(media_obj, "videoCodec", None)
                    audio_codec = getattr(media_obj, "audioCodec", None)
                    container = getattr(media_obj, "container", None)
                    video_resolution = getattr(media_obj, "videoResolution", None)

                transcoding_info = {
                    "is_transcoding": is_transcoding,
                    "video_codec": video_codec,
                    "audio_codec": audio_codec,
                    "container": container,
                    "video_resolution": video_resolution,
                    "transcoding_speed": transcode_speed,
                    "direct_play": not is_transcoding,
                }

                session_info = {
                    "user_name": user_info,
                    "media_title": media_title,
                    "media_type": media_type,
                    "progress": progress,
                    "state": state,
                    "session_id": str(getattr(session, "sessionKey", "")),
                    "client": client_name,
                    "device_name": device_name,
                    "position_ms": getattr(session, "viewOffset", 0),
                    "duration_ms": getattr(session, "duration", 0),
                    "artwork_url": artwork_url,
                    "transcoding_info": transcoding_info,
                    "thumbnail_url": getattr(session, "thumbUrl", None),
                }

                now_playing_sessions.append(session_info)

            return now_playing_sessions

        except Exception as e:
            logging.error(f"Failed to get now playing from Plex: {e}")
            return []

    def statistics(self):
        try:
            stats = {
                "library_stats": {},
                "user_stats": {},
                "server_stats": {},
                "content_stats": {},
            }

            # Try to get session information from the local server
            try:
                sessions = self.server.sessions()
                transcode_sessions = self.server.transcodeSessions()
            except Exception as e:
                logging.warning(f"Failed to get Plex session info: {e}")
                sessions = []
                transcode_sessions = []

            # Try to get user stats - this is where the SSL error occurs
            try:
                users = self.list_users()
                stats["user_stats"] = {
                    "total_users": len(users),
                    "active_sessions": len(sessions),
                }
            except (ConnectionError, Exception) as e:
                logging.error(f"Failed to get Plex user stats: {e}")
                stats["user_stats"] = {
                    "total_users": 0,
                    "active_sessions": len(sessions),
                    "connection_error": "Unable to connect to Plex.tv servers",
                }

            # Try to get server stats
            try:
                stats["server_stats"] = {
                    "version": getattr(self.server, "version", "Unknown"),
                    "transcoding_sessions": len(transcode_sessions),
                }
            except Exception as e:
                logging.error(f"Failed to get Plex server stats: {e}")
                stats["server_stats"] = {
                    "version": "Unknown",
                    "transcoding_sessions": 0,
                    "error": str(e),
                }

            return stats

        except Exception as e:
            logging.error(f"Failed to get Plex statistics: {e}")
            return {
                "library_stats": {},
                "user_stats": {"total_users": 0, "active_sessions": 0},
                "server_stats": {"version": "Unknown", "transcoding_sessions": 0},
                "content_stats": {},
                "error": str(e),
            }

    def get_user_count(self) -> int:
        """Get lightweight user count from database without triggering Plex.tv sync."""
        try:
            # Count existing users in database for this server instead of calling list_users()
            from app.models import MediaServer, User

            if hasattr(self, "server_id") and self.server_id:
                count = User.query.filter_by(server_id=self.server_id).count()
            else:
                # Fallback for legacy settings: find MediaServer for this server type
                servers = MediaServer.query.filter_by(server_type="plex").all()
                if servers:
                    server_ids = [s.id for s in servers]
                    count = User.query.filter(User.server_id.in_(server_ids)).count()
                else:
                    # Ultimate fallback: try Plex API (expensive)
                    try:
                        users = self.list_users()
                        count = len(users) if users else 0
                    except Exception as api_error:
                        logging.warning(f"Plex API fallback failed: {api_error}")
                        count = 0
            return count
        except Exception as e:
            logging.error(f"Failed to get Plex user count from database: {e}")
            return 0

    def get_server_info(self) -> dict:
        """Get lightweight server information without triggering user sync."""
        try:
            # Get basic server info and session counts without calling list_users()
            sessions = []
            transcode_sessions = []

            try:
                sessions = self.server.sessions()
                transcode_sessions = self.server.transcodeSessions()
            except Exception as e:
                logging.warning(f"Failed to get Plex session info: {e}")

            return {
                "version": getattr(self.server, "version", "Unknown"),
                "transcoding_sessions": len(transcode_sessions),
                "active_sessions": len(sessions),
            }
        except Exception as e:
            logging.error(f"Failed to get Plex server info: {e}")
            return {
                "version": "Unknown",
                "transcoding_sessions": 0,
                "active_sessions": 0,
            }

    def get_readonly_statistics(self) -> dict:
        """Get lightweight statistics without triggering user synchronization."""
        try:
            user_count = self.get_user_count()
            server_info = self.get_server_info()

            return {
                "user_stats": {
                    "total_users": user_count,
                    "active_sessions": server_info.get("active_sessions", 0),
                },
                "server_stats": {
                    "version": server_info.get("version", "Unknown"),
                    "transcoding_sessions": server_info.get("transcoding_sessions", 0),
                },
                "library_stats": {},  # Minimal for health cards
                "content_stats": {},  # Minimal for health cards
            }
        except Exception as e:
            logging.error(f"Failed to get Plex readonly statistics: {e}")
            return {
                "user_stats": {"total_users": 0, "active_sessions": 0},
                "server_stats": {"version": "Unknown", "transcoding_sessions": 0},
                "library_stats": {},
                "content_stats": {},
                "error": str(e),
            }


# ─── Invite & onboarding ────────────────────────────────────────────────


def handle_oauth_token(app, token: str, code: str) -> None:
    with app.app_context():
        account = MyPlexAccount(token=token)
        email = account.email

        inv = Invitation.query.filter_by(code=code).first()
        servers: list[MediaServer] = []

        if inv and inv.servers:
            # A single Plex OAuth token can authorize shares on every Plex server
            # attached to the invite. Non-Plex servers continue through the
            # password-based step after this route returns.
            plex_servers = [s for s in inv.servers if s.server_type == "plex"]
            servers = plex_servers or [inv.servers[0]]
        elif inv and inv.server:
            # Fallback to legacy single server relationship
            servers = [inv.server]
        else:
            # Last resort fallback
            fallback_server = MediaServer.query.first()
            if fallback_server:
                servers = [fallback_server]

        if not servers:
            raise ValueError("No media server found")

        from flask import current_app

        from app.services.expiry import calculate_user_expiry

        for server in servers:
            server_id = server.id

            db.session.query(User).filter(
                User.email == email, User.server_id == server_id
            ).delete(synchronize_session=False)
            db.session.commit()

            expires = calculate_user_expiry(inv, server_id) if inv else None

            client = PlexClient(media_server=server)
            new_user = client._create_user_with_identity_linking(
                {
                    "token": token,
                    "email": email,
                    "username": account.username,
                    "code": code,
                    "expires": expires,
                    "server_id": server_id,
                }
            )
            db.session.commit()

            _invite_user(email, code, new_user.id, server)

            # Mark invitation as used for this server
            if inv:
                from app.services.invites import mark_server_used

                mark_server_used(inv, server_id, new_user)

            # Pass only what we need: server credentials and Flask app instance
            # Use the specific server being processed for this invitation
            post_setup_client = PlexClient(media_server=server)
            server_url = post_setup_client.url
            api_token = post_setup_client.token
            threading.Thread(
                target=_post_join_setup,
                args=(current_app._get_current_object(), server_url, api_token, token),  # type: ignore
                daemon=True,
            ).start()

        notify(
            "User Joined",
            f"User {account.username} has joined your server!",
            "tada",
            event_type="user_joined",
        )


def _invite_user(email: str, code: str, _user_id: int, server: MediaServer) -> None:
    inv = Invitation.query.filter_by(code=code).first()
    if not inv:
        raise ValueError(f"No invitation found with code {code}")

    client = get_client_for_media_server(server)

    # Use library names, not IDs (PlexAPI expects names or LibrarySection objects)
    libs = (
        [lib.name for lib in inv.libraries if lib.server_id == server.id]
        if inv.libraries
        else []
    )

    if not libs:
        libs = [
            lib.name
            for lib in Library.query.filter_by(enabled=True, server_id=server.id).all()
        ]

    allow_sync = bool(inv.allow_downloads)
    allow_tv = bool(inv.allow_live_tv)
    allow_camera_upload = bool(inv.allow_mobile_uploads)

    try:
        if inv.plex_home:
            client.invite_home(email, libs, allow_sync, allow_tv, allow_camera_upload)
        else:
            client.invite_friend(email, libs, allow_sync, allow_tv, allow_camera_upload)
    except PlexInvitationError as e:
        if "already sharing" in e.message.lower():
            # User already has a share — update libraries and permissions instead
            structlog.get_logger().info(
                "User already has share, updating libraries and permissions",
                email=email,
            )
            permissions = {
                "allow_downloads": allow_sync,
                "allow_live_tv": allow_tv,
                "allow_camera_upload": allow_camera_upload,
            }
            client.update_user_libraries(email, libs if libs else None)
            client.update_user_permissions(email, permissions)
        else:
            # Re-raise PlexInvitationError to preserve the user-friendly message
            raise
    except Exception as e:
        # Handle any other unexpected errors
        error_message = extract_plex_error_message(e)
        logging.error(f"Unexpected error inviting {email} to Plex: {error_message}")
        raise PlexInvitationError(error_message, e) from e

    logging.info("Invited %s to Plex", email)

    PlexClient.list_users.cache_clear()
    db.session.commit()


def _post_join_setup(app, server_url: str, api_token: str, token: str):
    # Create a PlexServer instance with Flask app context
    from plexapi.server import PlexServer

    with app.app_context():
        try:
            server = PlexServer(server_url, api_token)
            user = MyPlexAccount(token=token)
            # use username as the v2 API returns only username, not e-mail
            admin_account = server.myPlexAccount()
            user.acceptInvite(admin_account.username)
            user.enableViewStateSync()
            _opt_out_online_sources(user)
        except ValueError as exc:
            if "No pending invite" in str(exc):
                # This is expected - the invite was already accepted in the main flow
                logging.info(
                    "Invite already accepted, proceeding with remaining setup: %s", exc
                )
                try:
                    # Still try to enable view state sync and opt out of online sources
                    user = MyPlexAccount(token=token)
                    user.enableViewStateSync()
                    _opt_out_online_sources(user)
                except Exception as setup_exc:
                    logging.warning("Partial post-join setup failed: %s", setup_exc)
            else:
                logging.error("Post-join setup failed with ValueError: %s", exc)
        except Exception as exc:
            logging.error("Post-join setup failed: %s", exc)


def _opt_out_online_sources(user: MyPlexAccount):
    online_sources = user.onlineMediaSources()
    for src in online_sources:
        if src and hasattr(src, "optOut"):
            src.optOut()


# ─── User queries / mutate ────────────────────────────────────────────────