/
/
/
1"""Authentication service for nicovideo."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7
8from niconico.exceptions import LoginFailureError
9
10from music_assistant.providers.nicovideo.helpers import log_verbose
11from music_assistant.providers.nicovideo.services.base import NicovideoBaseService
12
13if TYPE_CHECKING:
14 from asyncio import TimerHandle
15
16 from music_assistant.providers.nicovideo.services.manager import NicovideoServiceManager
17
18
19class NicovideoAuthService(NicovideoBaseService):
20 """Handles authentication and session management for nicovideo."""
21
22 def __init__(self, service_manager: NicovideoServiceManager) -> None:
23 """Initialize the NicovideoAuthService with a reference to the parent service manager."""
24 super().__init__(service_manager)
25 self._periodic_relogin_task: TimerHandle | None = None
26
27 @property
28 def is_logged_in(self) -> bool:
29 """Check if the user is logged in to niconico."""
30 return bool(self.niconico_py_client.logined)
31
32 async def try_login(self) -> bool:
33 """Attempt to login to niconico with the configured credentials."""
34 if self.is_logged_in:
35 return True
36
37 config = self.nicovideo_config
38 username = config.auth.mail
39 password = config.auth.password
40 mfa = config.auth.mfa
41 user_session = config.auth.user_session
42 max_retries = 3
43 retry_delay_seconds = 1
44 async with self.service_manager.niconico_api_throttler.bypass():
45 for attempt in range(max_retries):
46 try:
47 self.logger.debug(
48 "Trying to log in... (Number of attempts: %d/%d)",
49 attempt + 1,
50 max_retries,
51 )
52 if user_session:
53 self.logger.debug("Using user_session for login.")
54 await asyncio.to_thread(
55 self.niconico_py_client.login_with_session,
56 str(user_session),
57 )
58 else:
59 self.logger.debug("Using mail and password for login.")
60 if not username or not password:
61 self.logger.debug(
62 "Username and password are not set in the configuration.",
63 )
64 return False
65 await asyncio.to_thread(
66 self.niconico_py_client.login_with_mail,
67 str(username),
68 str(password),
69 str(mfa) if mfa else None,
70 )
71 self.logger.info("Successfully authenticated with Nicovideo!")
72 # Clear MFA code after successful use (one-time password should not be reused)
73 if mfa:
74 config.auth.clear_mfa_code()
75 session = self.niconico_py_client.get_user_session()
76 if session:
77 config.auth.save_user_session(session)
78 log_verbose(
79 self.logger,
80 "Saved user session for future logins (length: %d chars)",
81 len(session),
82 )
83 return True
84 except LoginFailureError as err:
85 if user_session:
86 user_session = None # Clear session on failure
87 self.logger.warning("Login with user_session failed: %s", err)
88 else:
89 self.logger.error("Login with mail and password failed: %s", err)
90 return False
91 except Exception as e:
92 if (
93 "Name or service not known" in str(e)
94 or "Max retries exceeded" in str(e)
95 or "ConnectionError" in str(e)
96 ):
97 self.logger.warning(
98 "Network or DNS error occurred: %s. Retrying in %d seconds...",
99 e,
100 retry_delay_seconds,
101 )
102 await asyncio.sleep(retry_delay_seconds)
103 else:
104 self.logger.error("An unexpected error has occurred.: %s", e)
105 return False
106 self.logger.error(
107 "Could not login after exceeding the maximum number of retries (%d).",
108 max_retries,
109 )
110 return False
111
112 async def try_logout(self) -> None:
113 """Log out from the niconico service."""
114 if self.niconico_py_client:
115 if self.is_logged_in:
116 await asyncio.to_thread(self.niconico_py_client.logout)
117 self.service_manager.reset_niconico_py_client()
118
119 def start_periodic_relogin_task(self) -> None:
120 """Start the periodic re-login task."""
121 # Cancel existing task if any
122 self.stop_periodic_relogin_task()
123 self._periodic_relogin_task = self.service_manager.mass.call_later(
124 30 * 24 * 60 * 60, self._schedule_periodic_relogin
125 )
126
127 def stop_periodic_relogin_task(self) -> None:
128 """Stop the periodic re-login task."""
129 if self._periodic_relogin_task and not self._periodic_relogin_task.cancelled():
130 self._periodic_relogin_task.cancel()
131 self._periodic_relogin_task = None
132
133 async def _schedule_periodic_relogin(self) -> None:
134 """Periodic re-login every 30 days."""
135 try:
136 self.logger.debug("Performing periodic re-login to refresh the session.")
137
138 config = self.nicovideo_config
139 if not (config.auth.mail or config.auth.password):
140 self.logger.debug("No login credentials provided, skipping periodic re-login.")
141 self.start_periodic_relogin_task()
142 return
143
144 await self.try_logout()
145 await asyncio.sleep(3) # Short delay to ensure logout completes
146 await self.try_login()
147 self.start_periodic_relogin_task()
148 except asyncio.CancelledError:
149 self.logger.debug("Periodic relogin task was cancelled.")
150 raise
151