music-assistant-server

1"""
2Manager service for niconico API integration with MusicAssistant.
3
4Services Layer: API integration and data transformation coordination
5- Coordinates API calls through niconico.py adapter
6- Manages authentication and session management
7- Handles API rate limiting and throttling
8- Delegates data transformation to converters
9"""
10
11from __future__ import annotations
12
13import asyncio
14import inspect
15from collections.abc import Callable
16from typing import TYPE_CHECKING
17
18from niconico import NicoNico
19from niconico.exceptions import LoginFailureError, LoginRequiredError, PremiumRequiredError
20from pydantic import ValidationError
21
22from music_assistant.helpers.throttle_retry import ThrottlerManager
23from music_assistant.models.music_provider import MusicProvider
24from music_assistant.providers.nicovideo.converters.manager import (
25    NicovideoConverterManager,
26)
27from music_assistant.providers.nicovideo.services.auth import NicovideoAuthService
28from music_assistant.providers.nicovideo.services.mylist import NicovideoMylistService
29from music_assistant.providers.nicovideo.services.search import NicovideoSearchService
30from music_assistant.providers.nicovideo.services.series import NicovideoSeriesService
31from music_assistant.providers.nicovideo.services.user import NicovideoUserService
32from music_assistant.providers.nicovideo.services.video import NicovideoVideoService
33
34if TYPE_CHECKING:
35    from music_assistant.providers.nicovideo.config import NicovideoConfig
36
37
38class NicovideoServiceManager:
39    """Central manager for all niconico services and MusicAssistant integration."""
40
41    def __init__(self, provider: MusicProvider, nicovideo_config: NicovideoConfig) -> None:
42        """Initialize service manager with provider and config."""
43        self.provider = provider
44        self.nicovideo_config = nicovideo_config
45        self.mass = provider.mass
46        self.reset_niconico_py_client()
47
48        self.niconico_api_throttler = ThrottlerManager(rate_limit=5, period=1)
49
50        self.logger = provider.logger
51
52        # Initialize services for different functionality
53        self.auth = NicovideoAuthService(self)
54        self.video = NicovideoVideoService(self)
55        self.series = NicovideoSeriesService(self)
56        self.mylist = NicovideoMylistService(self)
57        self.search = NicovideoSearchService(self)
58        self.user = NicovideoUserService(self)
59
60        # Initialize converter
61        self.converter_manager = NicovideoConverterManager(provider, self.logger)
62
63    def reset_niconico_py_client(self) -> None:
64        """Reset the niconico.py client instance."""
65        self.niconico_py_client = NicoNico()
66
67    def _extract_caller_info(self) -> str:
68        """Extract best-effort caller info file:function:line for diagnostics."""
69        frame = inspect.currentframe()
70        caller_info = "unknown"
71        try:
72            caller_frame = None
73            if frame and frame.f_back and frame.f_back.f_back:
74                caller_frame = frame.f_back.f_back  # Skip this method and acquire context
75            if caller_frame:
76                caller_filename = caller_frame.f_code.co_filename
77                caller_function = caller_frame.f_code.co_name
78                caller_line = caller_frame.f_lineno
79                filename = caller_filename.rsplit("/", 1)[-1]
80                caller_info = f"{filename}:{caller_function}:{caller_line}"
81        except Exception:
82            caller_info = "stack_inspection_failed"
83        finally:
84            del frame  # Prevent reference cycles
85        return caller_info
86
87    def _log_call_exception(self, operation: str, err: Exception) -> None:
88        """Log exceptions with classification and caller info."""
89        caller_info = self._extract_caller_info()
90        if isinstance(err, LoginRequiredError):
91            self.logger.debug(
92                "Authentication required for %s called from %s: %s", operation, caller_info, err
93            )
94        elif isinstance(err, PremiumRequiredError):
95            self.logger.warning(
96                "Premium account required for %s called from %s: %s", operation, caller_info, err
97            )
98        elif isinstance(err, LoginFailureError):
99            self.logger.warning(
100                "Login failed for %s called from %s: %s", operation, caller_info, err
101            )
102        elif isinstance(err, (ConnectionError, TimeoutError)):
103            self.logger.warning("Network error %s called from %s: %s", operation, caller_info, err)
104        elif isinstance(err, ValidationError):
105            try:
106                detailed_errors = err.errors()
107                self.logger.warning(
108                    "Validation error %s called from %s: %s\nDetailed errors: %s",
109                    operation,
110                    caller_info,
111                    err,
112                    detailed_errors,
113                )
114            except Exception:
115                self.logger.warning("Error %s called from %s: %s", operation, caller_info, err)
116        else:
117            self.logger.warning("Error %s called from %s: %s", operation, caller_info, err)
118
119    async def _call_with_throttler[T, **P](
120        self,
121        func: Callable[P, T],
122        *args: P.args,
123        **kwargs: P.kwargs,
124    ) -> T | None:
125        """Call function with API throttling."""
126        try:
127            async with self.niconico_api_throttler.acquire():
128                return await asyncio.to_thread(func, *args, **kwargs)
129        except Exception as err:
130            operation = func.__name__ if hasattr(func, "__name__") else "unknown_function"
131            self._log_call_exception(operation, err)
132            return None
133