/
/
/
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