nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-08 08:52:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

make qr code multi provider

Browse Source
This commit is contained in:
Nirvana
2026-06-02 11:23:55 +02:00
parent fada57e1e6
commit 8298adaad8
3 changed files with 123 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/ui/console_notification_adapter.py
+27 -12
View File
@@ -25,13 +25,27 @@ class ConsoleNotificationAdapter(NotificationInterface):
- CI/CD pipelines
"""
def __init__(self):
"""Initialize console notification adapter"""
def __init__(self, provider_name: str = "Service",
success_message: Optional[str] = None,
failure_template: Optional[str] = None):
"""
Initialize console notification adapter
Args:
provider_name: Name of the provider (e.g., "MagentaTV")
success_message: Custom success message (optional)
failure_template: Template for failure messages (optional)
"""
super().__init__()
self._start_time = None
self._expires_in = 0
self._last_update = 0
# Provider-specific configuration
self.provider_name = provider_name
self.success_message = success_message or f"{provider_name} login successful!"
self.failure_template = failure_template or f"{provider_name} login failed: {{}}"
@property
def supports_qr_display(self) -> bool:
"""Console cannot display QR codes"""
@@ -48,7 +62,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
return False
def show_remote_login(
self, login_code: str, qr_target_url: str, expires_in: int, interval: int = 10
self, login_code: str, qr_target_url: str, expires_in: int, interval: int = 10
) -> NotificationResult:
"""
Show remote login information in console
@@ -69,12 +83,12 @@ class ConsoleNotificationAdapter(NotificationInterface):
# Print header
print("\n" + "=" * 70)
print(" REMOTE LOGIN REQUIRED")
print(f" {self.provider_name} - REMOTE LOGIN REQUIRED")
print("=" * 70)
print()
# Print instructions
print("Please authenticate using your mobile device:")
print(f"Please authenticate using your mobile device for {self.provider_name}:")
print()
print(f" Option 1: Scan QR Code")
print(f" Visit this URL on your mobile device:")
@@ -88,7 +102,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
print("=" * 70)
print()
logger.info(f"Remote login started: code={login_code}, expires_in={expires_in}s")
logger.info(f"Remote login started for {self.provider_name}: code={login_code}, expires_in={expires_in}s")
logger.info(f"QR target URL: {qr_target_url}")
return NotificationResult.CONTINUE
@@ -110,7 +124,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
# Print milestone updates (every 30 seconds, or at key intervals)
if remaining_seconds <= 0:
print(f"⏰ Remote login expired")
print(f"⏰ {self.provider_name} remote login expired")
return True
# Print at: 240s, 180s, 120s, 60s, 30s, 10s
@@ -125,7 +139,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
else:
time_str = f"{seconds}s"
print(f"⏳ Waiting for authentication... {time_str} remaining")
print(f"⏳ {self.provider_name}: Waiting for authentication... {time_str} remaining")
return True
@@ -146,11 +160,12 @@ class ConsoleNotificationAdapter(NotificationInterface):
print("=" * 70)
if success:
print("✓ Remote login successful!")
print(f"✓ {self.success_message}")
elif message:
print(f"✗ Remote login failed: {message}")
failure_msg = self.failure_template.format(message)
print(f"✗ {failure_msg}")
else:
print("✗ Remote login failed")
print(f"✗ {self.failure_template.format('unknown error')}")
print("=" * 70)
print()
@@ -162,4 +177,4 @@ class ConsoleNotificationAdapter(NotificationInterface):
Returns:
bool: Always False (console can't be cancelled interactively)
"""
return False
return False
lib/streaming_providers/base/ui/kodi_notification_adapter.py
+21 -8
View File
@@ -64,13 +64,15 @@ class PollingThread(threading.Thread):
class QRCodeDialog:
def __init__(self, xbmcgui, xbmc, qr_image_path, login_code, expires_in, polling_thread=None):
def __init__(self, xbmcgui, xbmc, qr_image_path, login_code, expires_in,
polling_thread=None, provider_name: str = "Service"):
self.xbmcgui = xbmcgui
self.xbmc = xbmc
self.qr_image_path = qr_image_path
self.login_code = login_code
self.expires_in = expires_in
self.polling_thread = polling_thread
self.provider_name = provider_name
self.dialog = None
self.user_closed = False
@@ -103,7 +105,7 @@ class QRCodeDialog:
title_y = dialog_y + 30
title = self.xbmcgui.ControlLabel(
x=dialog_x + 50, y=title_y, width=dialog_width - 100, height=50,
label="[B]MagentaTV Remote Login[/B]", font="font30",
label=f"[B]{self.provider_name} Remote Login[/B]", font="font30",
textColor="0xFFFFFFFF", alignment=0x00000002,
)
self.dialog.addControl(title)
@@ -121,7 +123,7 @@ class QRCodeDialog:
instructions_y = qr_y + qr_size + 40
inst1 = self.xbmcgui.ControlLabel(
x=dialog_x + 50, y=instructions_y, width=dialog_width - 100, height=30,
label="[B]Scan QR code with your MagentaTV app[/B]", font="font13",
label="[B]Scan QR code with your app[/B]", font="font13",
textColor="0xFFFFFFFF", alignment=0x00000002,
)
self.dialog.addControl(inst1)
@@ -161,7 +163,7 @@ class QRCodeDialog:
if self.polling_thread:
self._start_monitor()
logger.info("Showing QR code dialog")
logger.info(f"Showing QR code dialog for {self.provider_name}")
self.dialog.doModal()
# FIX #2: only mark as user_closed if auth didn't trigger the close
@@ -263,7 +265,8 @@ class QRCodeDialog:
class KodiNotificationAdapter(NotificationInterface):
def __init__(self, http_manager=None):
def __init__(self, http_manager=None, provider_name: str = "Service",
success_message: str = None, failure_template: str = None):
super().__init__()
try:
import xbmc
@@ -285,6 +288,11 @@ class KodiNotificationAdapter(NotificationInterface):
self._vfs = get_vfs(addon_subdir="temp")
# Provider-specific configuration
self.provider_name = provider_name
self.success_message = success_message or f"{provider_name} login successful!"
self.failure_template = failure_template or f"{provider_name} login failed: {{}}"
@property
def supports_qr_display(self) -> bool:
return True
@@ -322,6 +330,7 @@ class KodiNotificationAdapter(NotificationInterface):
self.xbmcgui, self.xbmc,
qr_image_path, login_code, expires_in,
self._polling_thread,
provider_name=self.provider_name,
)
self._qr_dialog.show()
@@ -369,6 +378,7 @@ class KodiNotificationAdapter(NotificationInterface):
self.xbmcgui, self.xbmc,
qr_image_path, login_code, expires_in,
None,
provider_name=self.provider_name,
)
self._qr_dialog.show()
@@ -405,12 +415,13 @@ class KodiNotificationAdapter(NotificationInterface):
if success:
self.xbmcgui.Dialog().notification(
"MagentaTV", "Remote login successful!",
self.provider_name, self.success_message,
self.xbmcgui.NOTIFICATION_INFO, 3000,
)
elif message and not self._is_cancelled:
failure_msg = self.failure_template.format(message)
self.xbmcgui.Dialog().notification(
"MagentaTV", f"Remote login failed: {message}",
self.provider_name, failure_msg,
self.xbmcgui.NOTIFICATION_ERROR, 5000,
)
@@ -448,7 +459,9 @@ class KodiNotificationAdapter(NotificationInterface):
elapsed = time.time() - start_time
logger.info(f"Generated QR code in {elapsed:.2f}s: {len(png_data)} bytes")
filename = f"magentatv_qr_{int(time.time())}.png"
filename = f"qr_{self.provider_name.lower()}_{int(time.time())}.png"
# Sanitize filename - remove any problematic characters
filename = "".join(c for c in filename if c.isalnum() or c in '._-')
filepath = self._vfs.join_path(filename)
# FIX #1: write raw bytes via write_binary, not text
lib/streaming_providers/base/ui/notification_factory.py
+75 -22
View File
@@ -6,7 +6,7 @@ Factory for creating appropriate notification adapters
Auto-detects environment (Kodi vs standalone) and creates correct adapter
"""
from typing import Optional
from typing import Optional, Dict
from ..utils.logger import logger
from .notification_interface import NotificationInterface
@@ -20,14 +20,21 @@ class NotificationFactory:
the appropriate adapter:
- KodiNotificationAdapter if running in Kodi
- ConsoleNotificationAdapter if running standalone
Supports provider-specific configuration for UI text.
"""
_cached_adapter: Optional[NotificationInterface] = None
_cached_adapters: Dict[str, NotificationInterface] = {} # Cache by provider name
_environment_detected: Optional[str] = None
@classmethod
def create(
cls, force_environment: Optional[str] = None, http_manager=None
cls,
force_environment: Optional[str] = None,
http_manager=None,
provider_name: str = "Service",
success_message: Optional[str] = None,
failure_template: Optional[str] = None
) -> NotificationInterface:
"""
Create appropriate notification adapter
@@ -36,14 +43,20 @@ class NotificationFactory:
force_environment: Force specific environment ('kodi' or 'console')
If None, auto-detects
http_manager: Optional HTTPManager instance for network requests
provider_name: Name of the provider (e.g., "MagentaTV", "Netflix")
success_message: Custom success message (optional)
failure_template: Template for failure messages (optional)
Returns:
NotificationInterface: Appropriate adapter for current environment
"""
# Return cached adapter if available and no http_manager change
if cls._cached_adapter is not None and force_environment is None and http_manager is None:
logger.debug(f"Using cached notification adapter: {cls._environment_detected}")
return cls._cached_adapter
# Create cache key based on provider configuration
cache_key = f"{provider_name}:{success_message}:{failure_template}"
# Return cached adapter if available and no environment/http_manager change
if cache_key in cls._cached_adapters and force_environment is None and http_manager is None:
logger.debug(f"Using cached notification adapter: {cls._environment_detected} for {provider_name}")
return cls._cached_adapters[cache_key]
# Detect environment
if force_environment:
@@ -53,14 +66,23 @@ class NotificationFactory:
environment = cls._detect_environment()
logger.info(f"Detected notification environment: {environment}")
# Create appropriate adapter
# Create appropriate adapter with provider configuration
if environment == "kodi":
adapter = cls._create_kodi_adapter(http_manager=http_manager)
adapter = cls._create_kodi_adapter(
http_manager=http_manager,
provider_name=provider_name,
success_message=success_message,
failure_template=failure_template
)
else:
adapter = cls._create_console_adapter()
adapter = cls._create_console_adapter(
provider_name=provider_name,
success_message=success_message,
failure_template=failure_template
)
# Cache the adapter
cls._cached_adapter = adapter
cls._cached_adapters[cache_key] = adapter
cls._environment_detected = environment
return adapter
@@ -87,12 +109,21 @@ class NotificationFactory:
return "console"
@classmethod
def _create_kodi_adapter(cls, http_manager=None) -> NotificationInterface:
def _create_kodi_adapter(
cls,
http_manager=None,
provider_name: str = "Service",
success_message: Optional[str] = None,
failure_template: Optional[str] = None
) -> NotificationInterface:
"""
Create Kodi notification adapter
Args:
http_manager: Optional HTTPManager for QR code download
provider_name: Name of the provider
success_message: Custom success message
failure_template: Template for failure messages
Returns:
KodiNotificationAdapter
@@ -100,33 +131,56 @@ class NotificationFactory:
try:
from .kodi_notification_adapter import KodiNotificationAdapter
adapter = KodiNotificationAdapter(http_manager=http_manager)
logger.info("✓ Kodi notification adapter created")
adapter = KodiNotificationAdapter(
http_manager=http_manager,
provider_name=provider_name,
success_message=success_message,
failure_template=failure_template
)
logger.info(f"✓ Kodi notification adapter created for {provider_name}")
return adapter
except Exception as e:
logger.error(f"Failed to create Kodi adapter: {e}")
logger.warning("Falling back to console adapter")
return cls._create_console_adapter()
return cls._create_console_adapter(
provider_name=provider_name,
success_message=success_message,
failure_template=failure_template
)
@classmethod
def _create_console_adapter(cls) -> NotificationInterface:
def _create_console_adapter(
cls,
provider_name: str = "Service",
success_message: Optional[str] = None,
failure_template: Optional[str] = None
) -> NotificationInterface:
"""
Create console notification adapter
Args:
provider_name: Name of the provider
success_message: Custom success message
failure_template: Template for failure messages
Returns:
ConsoleNotificationAdapter
"""
from .console_notification_adapter import ConsoleNotificationAdapter
adapter = ConsoleNotificationAdapter()
logger.info("✓ Console notification adapter created")
adapter = ConsoleNotificationAdapter(
provider_name=provider_name,
success_message=success_message,
failure_template=failure_template
)
logger.info(f"✓ Console notification adapter created for {provider_name}")
return adapter
@classmethod
def reset_cache(cls):
"""Reset cached adapter (useful for testing)"""
cls._cached_adapter = None
"""Reset cached adapters (useful for testing)"""
cls._cached_adapters.clear()
cls._environment_detected = None
logger.debug("Notification adapter cache reset")
@@ -150,7 +204,6 @@ class NotificationFactory:
"""
try:
import xbmcgui
return True
except ImportError:
return False
return False