diff --git a/addon.xml b/addon.xml
index ab2bc75..de46ccf 100644
--- a/addon.xml
+++ b/addon.xml
@@ -7,6 +7,7 @@
+
diff --git a/lib/streaming_providers/base/ui/console_notification_adapter.py b/lib/streaming_providers/base/ui/console_notification_adapter.py
index 9ff0cd4..97bd47f 100644
--- a/lib/streaming_providers/base/ui/console_notification_adapter.py
+++ b/lib/streaming_providers/base/ui/console_notification_adapter.py
@@ -48,7 +48,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
def show_remote_login(
self,
login_code: str,
- qr_url: str,
+ qr_target_url: str,
expires_in: int,
interval: int = 10
) -> NotificationResult:
@@ -57,7 +57,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
Args:
login_code: Short login code
- qr_url: URL to QR code
+ qr_target_url: The URL to encode in QR (displayed to user)
expires_in: Expiration time in seconds
interval: Update interval
@@ -71,7 +71,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
# Print header
print("\n" + "=" * 70)
- print(" MAGENTATV REMOTE LOGIN REQUIRED")
+ print(" REMOTE LOGIN REQUIRED")
print("=" * 70)
print()
@@ -80,7 +80,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
print()
print(f" Option 1: Scan QR Code")
print(f" Visit this URL on your mobile device:")
- print(f" {qr_url}")
+ print(f" {qr_target_url}")
print()
print(f" Option 2: Manual Entry")
print(f" Login Code: {login_code}")
@@ -91,7 +91,7 @@ class ConsoleNotificationAdapter(NotificationInterface):
print()
logger.info(f"Remote login started: code={login_code}, expires_in={expires_in}s")
- logger.info(f"QR code URL: {qr_url}")
+ logger.info(f"QR target URL: {qr_target_url}")
return NotificationResult.CONTINUE
@@ -164,4 +164,4 @@ class ConsoleNotificationAdapter(NotificationInterface):
Returns:
bool: Always False (console can't be cancelled interactively)
"""
- return False
+ return False
\ No newline at end of file
diff --git a/lib/streaming_providers/base/ui/kodi_notification_adapter.py b/lib/streaming_providers/base/ui/kodi_notification_adapter.py
index 602b98e..9aed284 100644
--- a/lib/streaming_providers/base/ui/kodi_notification_adapter.py
+++ b/lib/streaming_providers/base/ui/kodi_notification_adapter.py
@@ -1,6 +1,10 @@
"""
Kodi notification adapter with threaded polling support
-Displays QR code in WindowDialog while polling happens in background thread
+Architecture:
+1. Starts polling in background thread
+2. Shows QR code dialog (blocking)
+3. Dialog monitors thread status and auto-closes on success
+4. User can cancel by closing dialog
"""
import time
import os
@@ -10,44 +14,104 @@ from typing import Optional, Callable
from .notification_interface import NotificationInterface, NotificationResult
from ..utils.logger import logger
-# Import the lightweight SVG converter
+# Import QR generator
try:
- from .svg_to_png import convert_svg_to_png
+ from .qr_generator import generate_qr_code_png
+ QR_GENERATOR_AVAILABLE = True
except ImportError:
- logger.warning("svg_to_png not available, will try to use SVG directly")
- convert_svg_to_png = None
+ logger.warning("qr_generator not available")
+ QR_GENERATOR_AVAILABLE = False
+
+
+class PollingThread(threading.Thread):
+ """
+ Background thread for polling authentication status
+ """
+
+ def __init__(self, poll_callback: Callable, expires_in: int, interval: int):
+ """
+ Initialize polling thread
+
+ Args:
+ poll_callback: Function to call for polling (returns token_data or None)
+ expires_in: Total time before expiration
+ interval: Polling interval in seconds
+ """
+ super().__init__(daemon=True)
+ self.poll_callback = poll_callback
+ self.expires_in = expires_in
+ self.interval = interval
+
+ self.auth_completed = False
+ self.token_data = None
+ self.error = None
+ self.stop_event = threading.Event()
+ self.start_time = None
+
+ def run(self):
+ """Run polling loop"""
+ self.start_time = time.time()
+ logger.info("Polling thread started")
+
+ try:
+ # Call the polling callback (blocking)
+ self.token_data = self.poll_callback()
+
+ if self.token_data:
+ self.auth_completed = True
+ logger.info("Polling thread: Authentication successful")
+ else:
+ logger.warning("Polling thread: Authentication failed/timed out")
+
+ except Exception as e:
+ logger.error(f"Polling thread error: {e}", exc_info=True)
+ self.error = str(e)
+
+ def stop(self):
+ """Signal thread to stop"""
+ self.stop_event.set()
+
+ def get_remaining_time(self) -> int:
+ """Get remaining time in seconds"""
+ if not self.start_time:
+ return self.expires_in
+ elapsed = time.time() - self.start_time
+ return max(0, int(self.expires_in - elapsed))
class QRCodeDialog:
"""
- WindowDialog for displaying QR code
- Non-blocking when used with threading
+ WindowDialog for displaying QR code with status monitoring
"""
- def __init__(self, xbmcgui, qr_image_path: str, login_code: str, qr_url: str,
- expires_in: int, poll_callback: Optional[Callable] = None):
+ def __init__(self, xbmcgui, xbmc, qr_image_path: str, login_code: str,
+ expires_in: int, polling_thread: Optional[PollingThread] = None):
"""
Initialize QR code dialog
Args:
xbmcgui: xbmcgui module
+ xbmc: xbmc module
qr_image_path: Path to QR code PNG file
login_code: Login code for manual entry
- qr_url: Full QR code URL
expires_in: Expiration time in seconds
- poll_callback: Optional callback function that returns True if auth completed
+ polling_thread: Optional polling thread to monitor
"""
self.xbmcgui = xbmcgui
+ self.xbmc = xbmc
self.qr_image_path = qr_image_path
self.login_code = login_code
- self.qr_url = qr_url
self.expires_in = expires_in
- self.poll_callback = poll_callback
+ self.polling_thread = polling_thread
self.dialog = None
- self.start_time = time.time()
self.user_closed = False
- self.auth_completed = False
+ self.time_label = None
+ self.status_label = None
+
+ # Background monitoring thread
+ self.monitor_thread = None
+ self.monitor_stop = threading.Event()
def show(self):
"""Show the QR code dialog"""
@@ -58,79 +122,75 @@ class QRCodeDialog:
screen_width = self.dialog.getWidth()
screen_height = self.dialog.getHeight()
- # Calculate positions and sizes
- qr_size = min(screen_width // 3, screen_height // 3, 600)
- qr_x = (screen_width - qr_size) // 2
- qr_y = 80
+ # Calculate layout
+ dialog_width = int(screen_width * 0.8)
+ dialog_height = int(screen_height * 0.8)
+ dialog_x = (screen_width - dialog_width) // 2
+ dialog_y = (screen_height - dialog_height) // 2
- # Background panel
- bg_width = screen_width - 200
- bg_height = screen_height - 160
- bg_x = 100
- bg_y = 80
-
- # Semi-transparent background
+ # Background
bg = self.xbmcgui.ControlImage(
- bg_x, bg_y, bg_width, bg_height,
- '' # No image, just uses aspect for background
+ dialog_x, dialog_y, dialog_width, dialog_height,
+ aspectRatio=0 # Scale to fit
)
- bg.setColorDiffuse('0xDD000000')
+ bg.setColorDiffuse('0xE0000000') # Semi-transparent black
self.dialog.addControl(bg)
# Title
- title_y = bg_y + 20
+ title_y = dialog_y + 30
title = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=title_y,
- width=bg_width - 100, height=40,
+ x=dialog_x + 50, y=title_y,
+ width=dialog_width - 100, height=50,
label='[B]MagentaTV Remote Login[/B]',
- font='font13_title',
+ font='font30',
textColor='0xFFFFFFFF',
- alignment=0x00000002 # Center aligned
+ alignment=0x00000002 # Center
)
self.dialog.addControl(title)
- # QR Code image
- qr_y_pos = title_y + 60
+ # QR Code
+ qr_size = min(dialog_width // 2, dialog_height // 2, 400)
+ qr_x = (screen_width - qr_size) // 2
+ qr_y = title_y + 70
+
if os.path.exists(self.qr_image_path):
qr_image = self.xbmcgui.ControlImage(
- qr_x, qr_y_pos, qr_size, qr_size,
+ qr_x, qr_y, qr_size, qr_size,
self.qr_image_path
)
self.dialog.addControl(qr_image)
else:
- logger.error(f"QR code image not found: {self.qr_image_path}")
+ logger.error(f"QR image not found: {self.qr_image_path}")
# Instructions
- instructions_y = qr_y_pos + qr_size + 30
+ instructions_y = qr_y + qr_size + 40
- # Line 1: Scan QR code
- line1 = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=instructions_y,
- width=bg_width - 100, height=30,
+ 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',
textColor='0xFFFFFFFF',
alignment=0x00000002
)
- self.dialog.addControl(line1)
+ self.dialog.addControl(inst1)
- # Line 2: Or enter code manually
- line2_y = instructions_y + 35
- line2 = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=line2_y,
- width=bg_width - 100, height=30,
- label=f'Or enter code manually: [B][COLOR yellow]{self.login_code}[/COLOR][/B]',
+ inst2_y = instructions_y + 35
+ inst2 = self.xbmcgui.ControlLabel(
+ x=dialog_x + 50, y=inst2_y,
+ width=dialog_width - 100, height=30,
+ label=f'Or enter code: [COLOR yellow]{self.login_code}[/COLOR]',
font='font13',
textColor='0xFFCCCCCC',
alignment=0x00000002
)
- self.dialog.addControl(line2)
+ self.dialog.addControl(inst2)
- # Line 3: Time remaining (will be updated)
- line3_y = line2_y + 40
+ # Time remaining label
+ time_y = inst2_y + 45
self.time_label = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=line3_y,
- width=bg_width - 100, height=30,
+ x=dialog_x + 50, y=time_y,
+ width=dialog_width - 100, height=30,
label=f'Time remaining: {self._format_time(self.expires_in)}',
font='font12',
textColor='0xFFFF8800',
@@ -138,46 +198,99 @@ class QRCodeDialog:
)
self.dialog.addControl(self.time_label)
- # Line 4: Waiting message
- line4_y = line3_y + 35
- line4 = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=line4_y,
- width=bg_width - 100, height=30,
+ # Status label
+ status_y = time_y + 35
+ self.status_label = self.xbmcgui.ControlLabel(
+ x=dialog_x + 50, y=status_y,
+ width=dialog_width - 100, height=30,
label='Waiting for authentication...',
font='font12',
textColor='0xFFAAAAAA',
alignment=0x00000002
)
- self.dialog.addControl(line4)
+ self.dialog.addControl(self.status_label)
- # Line 5: Close instruction
- line5_y = line4_y + 30
- line5 = self.xbmcgui.ControlLabel(
- x=bg_x + 50, y=line5_y,
- width=bg_width - 100, height=25,
+ # Cancel hint
+ cancel_y = status_y + 35
+ cancel = self.xbmcgui.ControlLabel(
+ x=dialog_x + 50, y=cancel_y,
+ width=dialog_width - 100, height=25,
label='(Press any key to cancel)',
font='font10',
textColor='0xFF888888',
alignment=0x00000002
)
- self.dialog.addControl(line5)
+ self.dialog.addControl(cancel)
- # Show dialog (blocking call)
+ # Start background monitor if we have polling thread
+ if self.polling_thread:
+ self._start_monitor()
+
+ # Show dialog (blocking)
logger.info("Showing QR code dialog")
self.dialog.doModal()
- # Dialog closed by user
+ # Dialog closed
self.user_closed = True
- logger.info("User closed QR code dialog")
+ self._stop_monitor()
+ logger.info("QR code dialog closed by user")
except Exception as e:
logger.error(f"Failed to show QR code dialog: {e}", exc_info=True)
finally:
self._cleanup()
- def close(self, success: bool = False):
- """Close the dialog programmatically"""
- self.auth_completed = success
+ def _start_monitor(self):
+ """Start background monitor thread"""
+ self.monitor_stop.clear()
+ self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
+ self.monitor_thread.start()
+ logger.debug("Started dialog monitor thread")
+
+ def _stop_monitor(self):
+ """Stop background monitor thread"""
+ self.monitor_stop.set()
+ if self.monitor_thread:
+ self.monitor_thread.join(timeout=1.0)
+ logger.debug("Stopped dialog monitor thread")
+
+ def _monitor_loop(self):
+ """Monitor polling thread status and update UI"""
+ while not self.monitor_stop.is_set():
+ try:
+ # Update countdown
+ if self.polling_thread:
+ remaining = self.polling_thread.get_remaining_time()
+ if self.time_label:
+ self.time_label.setLabel(f'Time remaining: {self._format_time(remaining)}')
+
+ # Check if auth completed
+ if self.polling_thread.auth_completed:
+ logger.info("Monitor: Authentication completed, closing dialog")
+ if self.status_label:
+ self.status_label.setLabel('[COLOR green]Authentication successful![/COLOR]')
+ time.sleep(1) # Show success message briefly
+ self.close_dialog()
+ break
+
+ # Check for errors
+ if self.polling_thread.error:
+ logger.error(f"Monitor: Polling error: {self.polling_thread.error}")
+ if self.status_label:
+ self.status_label.setLabel('[COLOR red]Authentication failed[/COLOR]')
+ time.sleep(2)
+ self.close_dialog()
+ break
+
+ # Sleep briefly
+ time.sleep(0.5)
+
+ except Exception as e:
+ logger.error(f"Monitor loop error: {e}")
+ break
+
+ def close_dialog(self):
+ """Close dialog programmatically"""
if self.dialog:
try:
self.dialog.close()
@@ -193,7 +306,8 @@ class QRCodeDialog:
pass
self.dialog = None
- def _format_time(self, seconds: int) -> str:
+ @staticmethod
+ def _format_time(seconds: int) -> str:
"""Format seconds as MM:SS"""
if seconds <= 0:
return "Expired"
@@ -204,13 +318,12 @@ class QRCodeDialog:
class KodiNotificationAdapter(NotificationInterface):
"""
- Kodi-based notification adapter with threading support
+ Kodi notification adapter with fast QR generation and threading
- Features:
- - Shows QR code in WindowDialog
- - Polls in background thread
- - Auto-closes on auth success
- - User can cancel anytime
+ Architecture:
+ 1. Generate QR code directly from target URL (fast)
+ 2. Start polling in background thread
+ 3. Show dialog that monitors thread and auto-closes on success
"""
def __init__(self, http_manager=None):
@@ -218,7 +331,7 @@ class KodiNotificationAdapter(NotificationInterface):
Initialize Kodi notification adapter
Args:
- http_manager: Optional HTTPManager instance for QR code download
+ http_manager: Optional HTTPManager instance
"""
super().__init__()
@@ -239,132 +352,177 @@ class KodiNotificationAdapter(NotificationInterface):
self._qr_dialog = None
self._qr_image_path = None
self._http_manager = http_manager
-
- # Threading
- self._poll_thread = None
- self._poll_callback = None
- self._auth_completed = False
- self._stop_polling = threading.Event()
+ self._polling_thread = None
@property
def supports_qr_display(self) -> bool:
- """Kodi can display QR code images"""
+ """Kodi can display QR codes"""
return True
@property
def supports_countdown(self) -> bool:
- """Kodi supports live countdown"""
+ """Kodi supports countdown via monitor thread"""
return True
@property
def is_blocking(self) -> bool:
- """With threading, operation is non-blocking from caller's perspective"""
- return False
+ """Dialog is blocking but polling happens in thread"""
+ return True # From caller's perspective, it blocks
- def show_remote_login(
+ def show_remote_login_with_polling(
self,
login_code: str,
- qr_url: str,
+ qr_target_url: str,
expires_in: int,
- interval: int = 10
+ interval: int,
+ poll_callback: Callable
) -> NotificationResult:
"""
- Show remote login dialog with QR code
+ Show remote login with integrated polling
- This starts a background thread and shows the QR dialog.
- The dialog is blocking but polling happens in background.
+ This is the main method that coordinates everything:
+ 1. Generate QR code from target URL (fast!)
+ 2. Start polling thread
+ 3. Show dialog (blocking, but thread runs)
+ 4. Return result based on outcome
Args:
login_code: Short login code
- qr_url: URL to QR code SVG
+ qr_target_url: The URL to encode in QR code (NOT the SVG URL!)
expires_in: Expiration time in seconds
- interval: Update interval (not used in threaded mode)
+ interval: Polling interval
+ poll_callback: Function to call for polling
Returns:
- NotificationResult indicating outcome
+ NotificationResult
"""
if not self._kodi_available:
return NotificationResult.ERROR
self._is_active = True
self._is_cancelled = False
- self._auth_completed = False
- self._stop_polling.clear()
try:
- # Download and convert QR code to PNG
- qr_image_path = self._download_and_convert_qr(qr_url)
+ # Step 1: Generate QR code (fast!)
+ qr_image_path = self._generate_qr_code(qr_target_url)
if not qr_image_path:
- logger.error("Failed to download/convert QR code")
+ logger.error("Failed to generate QR code")
return NotificationResult.ERROR
self._qr_image_path = qr_image_path
- # Create QR dialog
+ # Step 2: Start polling thread
+ self._polling_thread = PollingThread(poll_callback, expires_in, interval)
+ self._polling_thread.start()
+ logger.info("Started polling thread")
+
+ # Step 3: Show dialog (blocking, but thread runs)
self._qr_dialog = QRCodeDialog(
self.xbmcgui,
+ self.xbmc,
qr_image_path,
login_code,
- qr_url,
- expires_in
+ expires_in,
+ self._polling_thread
)
- # Show dialog (this is blocking, but that's okay)
- # The polling will happen in the remote_login_handler
self._qr_dialog.show()
- # Check if user closed dialog (cancelled)
- if self._qr_dialog.user_closed and not self._qr_dialog.auth_completed:
- logger.info("User cancelled login via dialog close")
+ # Step 4: Determine outcome
+ if self._polling_thread.auth_completed:
+ logger.info("Authentication successful")
+ return NotificationResult.CONTINUE
+ elif self._qr_dialog.user_closed:
+ logger.info("User cancelled")
+ self._is_cancelled = True
+ return NotificationResult.CANCELLED
+ else:
+ logger.warning("Authentication timed out")
+ return NotificationResult.TIMEOUT
+
+ except Exception as e:
+ logger.error(f"Failed to show remote login: {e}", exc_info=True)
+ return NotificationResult.ERROR
+ finally:
+ self._is_active = False
+ self._cleanup_qr_image()
+
+ def show_remote_login(
+ self,
+ login_code: str,
+ qr_target_url: str,
+ expires_in: int,
+ interval: int = 10
+ ) -> NotificationResult:
+ """
+ Simplified version without polling (for backward compatibility)
+ Just shows the dialog, no polling
+
+ Args:
+ login_code: Short login code
+ qr_target_url: The URL to encode in QR code
+ expires_in: Expiration time
+ interval: Update interval (unused in this version)
+ """
+ if not self._kodi_available:
+ return NotificationResult.ERROR
+
+ self._is_active = True
+ self._is_cancelled = False
+
+ try:
+ qr_image_path = self._generate_qr_code(qr_target_url)
+
+ if not qr_image_path:
+ return NotificationResult.ERROR
+
+ self._qr_image_path = qr_image_path
+
+ self._qr_dialog = QRCodeDialog(
+ self.xbmcgui,
+ self.xbmc,
+ qr_image_path,
+ login_code,
+ expires_in,
+ None # No polling thread
+ )
+
+ self._qr_dialog.show()
+
+ if self._qr_dialog.user_closed:
self._is_cancelled = True
return NotificationResult.CANCELLED
return NotificationResult.CONTINUE
except Exception as e:
- logger.error(f"Failed to show Kodi QR dialog: {e}", exc_info=True)
- self._is_active = False
+ logger.error(f"Failed to show QR dialog: {e}", exc_info=True)
return NotificationResult.ERROR
+ finally:
+ self._is_active = False
+ self._cleanup_qr_image()
def update_countdown(self, remaining_seconds: int) -> bool:
- """
- Update countdown (called by polling loop)
-
- Args:
- remaining_seconds: Seconds remaining
-
- Returns:
- bool: True to continue, False if user cancelled
- """
- # Check if user closed dialog
+ """Check if user cancelled"""
if self._qr_dialog and self._qr_dialog.user_closed:
self._is_cancelled = True
return False
-
return not self._is_cancelled
def close(self, success: bool = False, message: Optional[str] = None):
- """
- Close QR dialog and show result notification
-
- Args:
- success: Whether authentication succeeded
- message: Optional message
- """
+ """Close and show notification"""
if not self._is_active:
return
self._is_active = False
- self._stop_polling.set()
try:
- # Close QR dialog if open
if self._qr_dialog:
- self._qr_dialog.close(success)
+ self._qr_dialog.close_dialog()
self._qr_dialog = None
- # Show result notification
+ # Show result
if success:
self.xbmcgui.Dialog().notification(
"MagentaTV",
@@ -380,66 +538,54 @@ class KodiNotificationAdapter(NotificationInterface):
5000
)
- # Cleanup QR image
self._cleanup_qr_image()
except Exception as e:
- logger.error(f"Failed to close Kodi dialog: {e}")
+ logger.error(f"Failed to close dialog: {e}")
def is_cancelled(self) -> bool:
- """
- Check if user cancelled
-
- Returns:
- bool: True if user closed dialog
- """
- if self._qr_dialog and self._qr_dialog.user_closed and not self._qr_dialog.auth_completed:
- self._is_cancelled = True
+ """Check if cancelled"""
return self._is_cancelled
- def _download_and_convert_qr(self, qr_url: str) -> Optional[str]:
+ def get_token_data(self) -> Optional[dict]:
+ """Get token data from polling thread"""
+ if self._polling_thread:
+ return self._polling_thread.token_data
+ return None
+
+ @staticmethod
+ def _generate_qr_code(target_url: str) -> Optional[str]:
"""
- Download QR code SVG and convert to PNG
+ Generate QR code PNG file from target URL
+
+ This is GENERIC - works for any provider!
+ No provider-specific logic here.
Args:
- qr_url: URL to SVG QR code
+ target_url: The URL to encode in the QR code
Returns:
- Path to PNG file or None if failed
+ Path to PNG file
"""
try:
- logger.info(f"Downloading QR code from: {qr_url}")
-
- # Download SVG
- if self._http_manager:
- response = self._http_manager.get(
- qr_url,
- operation='qr_download',
- timeout=10
- )
- else:
- try:
- import requests
- response = requests.get(qr_url, timeout=10)
- except ImportError:
- logger.error("Neither http_manager nor requests available")
- return None
-
- response.raise_for_status()
- svg_data = response.content
-
- logger.info(f"Downloaded SVG: {len(svg_data)} bytes")
-
- # Convert SVG to PNG
- if convert_svg_to_png:
- logger.info("Converting SVG to PNG...")
- png_data = convert_svg_to_png(svg_data, output_size=512)
- logger.info(f"Converted to PNG: {len(png_data)} bytes")
- else:
- logger.error("SVG to PNG converter not available")
+ if not QR_GENERATOR_AVAILABLE:
+ logger.error("QR generator not available")
return None
- # Save PNG to temp file
+ logger.info(f"Generating QR code for: {target_url}")
+ start_time = time.time()
+
+ # Generate QR code directly from target URL
+ png_data = generate_qr_code_png(target_url, size=512)
+
+ if not png_data:
+ logger.error("Failed to generate QR code PNG")
+ return None
+
+ elapsed = time.time() - start_time
+ logger.info(f"Generated QR code in {elapsed:.2f}s: {len(png_data)} bytes")
+
+ # Save to temp file
temp_dir = tempfile.gettempdir()
png_filename = f"magentatv_qr_{int(time.time())}.png"
png_path = os.path.join(temp_dir, png_filename)
@@ -451,11 +597,11 @@ class KodiNotificationAdapter(NotificationInterface):
return png_path
except Exception as e:
- logger.error(f"Failed to download/convert QR code: {e}", exc_info=True)
+ logger.error(f"Failed to generate QR code: {e}", exc_info=True)
return None
def _cleanup_qr_image(self):
- """Cleanup temporary QR image file"""
+ """Cleanup temporary QR image"""
if self._qr_image_path and os.path.exists(self._qr_image_path):
try:
os.remove(self._qr_image_path)
diff --git a/lib/streaming_providers/base/ui/notification_interface.py b/lib/streaming_providers/base/ui/notification_interface.py
index 02bdd64..56f3bcf 100644
--- a/lib/streaming_providers/base/ui/notification_interface.py
+++ b/lib/streaming_providers/base/ui/notification_interface.py
@@ -38,7 +38,7 @@ class NotificationInterface(ABC):
def show_remote_login(
self,
login_code: str,
- qr_url: str,
+ qr_target_url: str,
expires_in: int,
interval: int = 10
) -> NotificationResult:
@@ -46,14 +46,15 @@ class NotificationInterface(ABC):
Show remote login notification to user
This method should:
- 1. Display the login code and QR code/URL
+ 1. Display the login code and QR target URL
2. Show countdown timer
3. Allow user cancellation
4. Return when done or cancelled
Args:
login_code: Short code user can type (e.g., "PY48E62Q")
- qr_url: Full URL to QR code SVG (e.g., "https://wcps.t-online.de/caas/default/v1/remoteLogin/PY48E62Q")
+ qr_target_url: The actual URL to encode in QR code / display to user
+ (e.g., "https://telekom.de/tv-login?login_code=PY48E62Q")
expires_in: Total seconds until expiration
interval: Polling interval in seconds (for countdown updates)
@@ -148,4 +149,4 @@ class NotificationInterface(ABC):
'supports_countdown': self.supports_countdown,
'is_blocking': self.is_blocking,
'is_active': self.is_active
- }
+ }
\ No newline at end of file
diff --git a/lib/streaming_providers/base/ui/qr_generator.py b/lib/streaming_providers/base/ui/qr_generator.py
new file mode 100644
index 0000000..c3b8b39
--- /dev/null
+++ b/lib/streaming_providers/base/ui/qr_generator.py
@@ -0,0 +1,98 @@
+"""
+Fast QR Code Generator for Kodi
+Pure Python implementation using qrcode library
+Generates PNG QR codes directly from URLs
+
+This is a GENERIC module - no provider-specific logic!
+Size: qrcode library is ~50KB, pure Python
+"""
+import io
+from typing import Optional
+
+
+def generate_qr_code_png(data: str, size: int = 512) -> Optional[bytes]:
+ """
+ Generate QR code PNG from data string
+
+ This is a generic function that works for ANY provider.
+ No provider-specific logic here!
+
+ Args:
+ data: String to encode (URL, code, etc.)
+ size: Output size in pixels (square)
+
+ Returns:
+ PNG file content as bytes, or None if failed
+ """
+ try:
+ import qrcode # type: ignore
+ from qrcode.image.pure import PyPNGImage # type: ignore
+
+ # Create QR code
+ qr = qrcode.QRCode(
+ version=1, # Auto-size
+ error_correction=qrcode.constants.ERROR_CORRECT_L,
+ box_size=10,
+ border=4,
+ )
+
+ qr.add_data(data)
+ qr.make(fit=True)
+
+ # Generate image using pure Python PNG backend
+ img = qr.make_image(image_factory=PyPNGImage, fill_color="black", back_color="white")
+
+ # Convert to PNG bytes
+ buffer = io.BytesIO()
+ img.save(buffer, format='PNG')
+ png_data = buffer.getvalue()
+
+ return png_data
+
+ except ImportError:
+ # Fallback: Try with PIL if available
+ try:
+ import qrcode
+
+ qr = qrcode.QRCode(
+ version=1,
+ error_correction=qrcode.constants.ERROR_CORRECT_L,
+ box_size=10,
+ border=4,
+ )
+
+ qr.add_data(data)
+ qr.make(fit=True)
+
+ img = qr.make_image(fill_color="black", back_color="white")
+
+ buffer = io.BytesIO()
+ img.save(buffer, format='PNG')
+ png_data = buffer.getvalue()
+
+ return png_data
+
+ except Exception as e:
+ print(f"Failed to generate QR code: {e}")
+ return None
+
+ except Exception as e:
+ print(f"Failed to generate QR code: {e}")
+ return None
+
+
+# Test function
+if __name__ == '__main__':
+ # Test QR generation
+ test_url = "https://example.com/login?code=ABC123"
+
+ png_data = generate_qr_code_png(test_url, size=512)
+ if png_data:
+ print(f"Generated QR code PNG: {len(png_data)} bytes")
+
+ # Save to file for testing
+ with open('test_qr.png', 'wb') as f:
+ f.write(png_data)
+ print("Saved to test_qr.png")
+ else:
+ print("Failed to generate QR code")
\ No newline at end of file
diff --git a/lib/streaming_providers/providers/magenta2/remote_login_handler.py b/lib/streaming_providers/providers/magenta2/remote_login_handler.py
index ef49782..397bee9 100644
--- a/lib/streaming_providers/providers/magenta2/remote_login_handler.py
+++ b/lib/streaming_providers/providers/magenta2/remote_login_handler.py
@@ -2,10 +2,13 @@
"""
Remote Login Handler for Magenta2 Backchannel Authentication
Implements QR code-based authentication flow as fallback when line auth fails
+
+This handler contains provider-specific logic for MagentaTV.
"""
import time
from typing import Dict, Optional, Any
from dataclasses import dataclass
+from urllib.parse import urlparse, parse_qs, unquote
from ...base.network import HTTPManager
from ...base.utils.logger import logger
@@ -26,18 +29,19 @@ class RemoteLoginSession:
interval: int
expires_in: int
qr_code_url: str
+ qr_target_url: Optional[str] # The actual URL to encode in QR
started_at: float
class RemoteLoginHandler:
"""
- Handles the complete backchannel authentication / remote login flow
+ Handles backchannel authentication / remote login flow for MagentaTV
- Flow:
- 1. Start backchannel auth -> get login code and QR URL
- 2. Display QR code to user (via notification adapter)
- 3. Poll token endpoint until user completes mobile authentication
- 4. Handle countdown internally with notification updates
+ Responsibilities:
+ - Start backchannel auth session
+ - Extract QR target URL (provider-specific)
+ - Poll for token
+ - Coordinate with notification system (generic)
"""
def __init__(self, http_manager: HTTPManager, sam3_client_id: str,
@@ -72,12 +76,7 @@ class RemoteLoginHandler:
logger.debug(f"RemoteLoginHandler initialized with {self._notifier.__class__.__name__}")
def set_notifier(self, notifier: NotificationInterface) -> None:
- """
- Set custom notification interface
-
- Args:
- notifier: Notification interface to use
- """
+ """Set custom notification interface"""
self._notifier = notifier
logger.debug(f"Notifier set to: {notifier.__class__.__name__}")
@@ -97,7 +96,6 @@ class RemoteLoginHandler:
try:
logger.info("Starting remote login (backchannel auth)")
- # Build request payload
payload = {
'client_id': self.sam3_client_id,
'scope': scope
@@ -108,13 +106,10 @@ class RemoteLoginHandler:
'User-Agent': SSO_USER_AGENT
}
- # DEBUG: Log exact request details
logger.debug(f"Backchannel auth request:")
logger.debug(f" URL: {self.backchannel_start_url}")
- logger.debug(f" Headers: {headers}")
logger.debug(f" Payload: {payload}")
- # Start backchannel auth
response = self.http_manager.post(
self.backchannel_start_url,
operation='backchannel_start',
@@ -136,9 +131,12 @@ class RemoteLoginHandler:
if not all([initial_login_code, auth_req_id, auth_req_sec]):
raise Exception("Incomplete backchannel auth response")
- # Build QR code URL
+ # Build QR code SVG URL
qr_code_url = self.qr_code_url_template.format(code=initial_login_code)
+ # Extract the actual target URL for the QR code (provider-specific)
+ qr_target_url = self._extract_qr_target_url(qr_code_url, initial_login_code)
+
# Create session
session = RemoteLoginSession(
initial_login_code=initial_login_code,
@@ -147,6 +145,7 @@ class RemoteLoginHandler:
interval=interval,
expires_in=expires_in,
qr_code_url=qr_code_url,
+ qr_target_url=qr_target_url,
started_at=time.time()
)
@@ -156,6 +155,8 @@ class RemoteLoginHandler:
f"Remote login started: code={initial_login_code}, "
f"interval={interval}s, expires_in={expires_in}s"
)
+ if qr_target_url:
+ logger.info(f"QR target URL: {qr_target_url}")
return session
@@ -163,22 +164,80 @@ class RemoteLoginHandler:
logger.error(f"Failed to start remote login: {e}")
raise Exception(f"Remote login start failed: {e}")
+ def _extract_qr_target_url(self, qr_code_url: str, login_code: str) -> str:
+ """
+ Extract the actual target URL from QR code redirect
+
+ PROVIDER-SPECIFIC LOGIC for MagentaTV:
+ The qr_code_url redirects (302) to:
+ https://wcps.t-online.de/usqrg/v1/default/QrCode?target=
+
+ We extract the 'target' parameter which is the actual URL to encode in QR.
+
+ Args:
+ qr_code_url: QR code SVG URL
+ login_code: Login code (for fallback)
+
+ Returns:
+ Target URL to encode in QR code
+ """
+ try:
+ logger.debug(f"Extracting QR target URL from: {qr_code_url}")
+
+ # Make GET request without following redirects
+ response = self.http_manager.get(
+ qr_code_url,
+ operation='qr_redirect',
+ allow_redirects=False, # Don't follow redirects automatically
+ timeout=5
+ )
+
+ # Check for redirect
+ if response.status_code in (301, 302, 303, 307, 308):
+ redirect_url = response.headers.get('Location')
+
+ if redirect_url:
+ logger.debug(f"Got redirect to: {redirect_url}")
+
+ # Parse the redirect URL
+ parsed = urlparse(redirect_url)
+ query_params = parse_qs(parsed.query)
+
+ # Extract 'target' parameter
+ if 'target' in query_params:
+ target_url = query_params['target'][0]
+ # URL decode it
+ decoded_url = unquote(target_url)
+ logger.info(f"Extracted QR target URL: {decoded_url}")
+ return decoded_url
+ else:
+ logger.warning("No 'target' parameter in redirect URL")
+ else:
+ logger.warning(f"No redirect found (status: {response.status_code})")
+
+ except Exception as e:
+ logger.warning(f"Failed to extract QR target URL: {e}")
+
+ # Fallback: construct URL from login code
+ fallback_url = f"https://telekom.de/tv-login?login_code={login_code}"
+ logger.info(f"Using fallback QR target URL: {fallback_url}")
+ return fallback_url
+
def poll_for_token(self, session: RemoteLoginSession) -> Optional[Dict[str, Any]]:
"""
- Poll token endpoint until user completes authentication or timeout
- NOW HANDLES COUNTDOWN UPDATES INTERNALLY
+ Poll token endpoint until user completes authentication
+
+ This method is called by the polling thread in the notification adapter.
+ It checks for cancellation periodically.
Args:
session: Active remote login session
Returns:
Token data dict if successful, None if timed out/cancelled
-
- Raises:
- Exception: If polling fails with unexpected error
"""
try:
- logger.info("Starting token polling for remote login")
+ logger.info("Starting token polling")
start_time = session.started_at
next_poll_time = start_time
@@ -206,41 +265,31 @@ class RemoteLoginHandler:
# Check if session expired
if elapsed >= session.expires_in:
- logger.warning(
- f"Remote login session expired after {elapsed:.1f}s "
- f"(limit: {session.expires_in}s)"
- )
- self._notifier.close(success=False, message="Session expired")
+ logger.warning(f"Session expired after {elapsed:.1f}s")
return None
- # Update countdown display (every second or as needed)
+ # Check if user cancelled (every second)
if current_time - last_countdown_update >= 1.0:
- # Check if user cancelled
if self._notifier.is_cancelled():
logger.info("User cancelled remote login")
- self._notifier.close(success=False, message="Cancelled by user")
return None
- # Update countdown
+ # Update countdown in notifier
if not self._notifier.update_countdown(int(remaining)):
- logger.info("Countdown update returned False - user cancelled")
- self._notifier.close(success=False, message="Cancelled by user")
+ logger.info("Countdown update returned False - cancelled")
return None
last_countdown_update = current_time
# Wait until next poll time
if current_time < next_poll_time:
- sleep_time = min(1.0, next_poll_time - current_time) # Sleep max 1 second for countdown updates
+ sleep_time = min(1.0, next_poll_time - current_time)
time.sleep(sleep_time)
continue
# Perform poll
poll_count += 1
- logger.debug(
- f"Polling attempt {poll_count}/{max_polls} "
- f"(remaining: {remaining:.0f}s)"
- )
+ logger.debug(f"Poll {poll_count}/{max_polls} (remaining: {remaining:.0f}s)")
try:
response = self.http_manager.post(
@@ -251,78 +300,117 @@ class RemoteLoginHandler:
timeout=DEFAULT_REQUEST_TIMEOUT
)
- # 202 = User hasn't completed authentication yet
+ # 202 = Not yet completed
if response.status_code == 202:
logger.debug("Authentication not yet completed (202)")
next_poll_time = time.time() + session.interval
continue
- # Success - user completed authentication
+ # 200 = Success
if response.status_code == 200:
token_data = response.json()
- logger.info(
- f"✓ Remote login successful after {elapsed:.1f}s "
- f"({poll_count} polls)"
- )
- self._notifier.close(success=True)
+ logger.info(f"✓ Authentication successful after {elapsed:.1f}s ({poll_count} polls)")
return token_data
- # Other status codes = error
+ # Other = error
response.raise_for_status()
except Exception as e:
- # If we get an error during polling, check if we should retry
if elapsed < session.expires_in:
logger.debug(f"Poll error (will retry): {e}")
next_poll_time = time.time() + session.interval
continue
else:
- # Session expired, give up
raise
except Exception as e:
- logger.error(f"Remote login polling failed: {e}")
- self._notifier.close(success=False, message=str(e))
- raise Exception(f"Remote login polling failed: {e}")
+ logger.error(f"Polling failed: {e}")
+ return None
def perform_complete_flow(self, scope: str = "tvhubs offline_access") -> Optional[Dict[str, Any]]:
"""
- Perform complete remote login flow:
- 1. Start session
- 2. Display QR code via notifier
- 3. Poll for completion with automatic countdown updates
+ Perform complete remote login flow with integrated polling
+
+ Flow:
+ 1. Start session (gets login code + extracts QR target URL)
+ 2. Check if notifier supports integrated polling
+ 3. If yes: Pass polling callback to notifier (threaded)
+ 4. If no: Show notification, then poll manually (console)
Args:
scope: OAuth scopes to request
Returns:
- Token data dict if successful, None if failed/timeout/cancelled
+ Token data dict if successful, None if failed
"""
try:
- # Step 1: Start session
+ # Step 1: Start session and extract QR target URL
session = self.start_remote_login(scope)
- # Step 2: Display QR code to user via notifier
- result = self._notifier.show_remote_login(
- login_code=session.initial_login_code,
- qr_url=session.qr_code_url,
- expires_in=session.expires_in,
- interval=session.interval
- )
+ # Step 2: Check if notifier supports integrated polling (Kodi adapter)
+ if hasattr(self._notifier, 'show_remote_login_with_polling'):
+ # Kodi adapter with threading support
+ logger.info("Using integrated polling (threaded)")
- if result != NotificationResult.CONTINUE:
- logger.warning(f"Failed to show notification: {result}")
- return None
+ # Create polling callback
+ def poll_callback():
+ return self.poll_for_token(session)
- # Step 3: Poll for completion (handles countdown internally)
- token_data = self.poll_for_token(session)
+ # Show with integrated polling
+ # Pass the EXTRACTED TARGET URL, not the SVG URL!
+ result = self._notifier.show_remote_login_with_polling(
+ login_code=session.initial_login_code,
+ qr_target_url=session.qr_target_url,
+ expires_in=session.expires_in,
+ interval=session.interval,
+ poll_callback=poll_callback
+ )
+
+ if result == NotificationResult.CONTINUE:
+ # Get token data from adapter (if supported)
+ if hasattr(self._notifier, 'get_token_data'):
+ token_data = self._notifier.get_token_data()
+ if token_data:
+ logger.info("✓ Remote login completed successfully")
+ self._notifier.close(success=True)
+ return token_data
+ else:
+ logger.warning("No token data available")
+ return None
+ else:
+ logger.error("Notifier doesn't support get_token_data()")
+ return None
+ else:
+ logger.warning(f"Remote login result: {result}")
+ return None
- if token_data:
- logger.info("✓ Remote login flow completed successfully")
else:
- logger.warning("Remote login flow timed out or was cancelled")
+ # Console adapter or other - manual polling
+ logger.info("Using manual polling")
- return token_data
+ # Show notification with extracted target URL
+ result = self._notifier.show_remote_login(
+ login_code=session.initial_login_code,
+ qr_target_url=session.qr_target_url,
+ expires_in=session.expires_in,
+ interval=session.interval
+ )
+
+ if result != NotificationResult.CONTINUE:
+ logger.warning(f"Failed to show notification: {result}")
+ return None
+
+ # Poll manually
+ token_data = self.poll_for_token(session)
+
+ if token_data:
+ logger.info("✓ Remote login completed successfully")
+ self._notifier.close(success=True)
+ else:
+ logger.warning("Remote login timed out or cancelled")
+ self._notifier.close(success=False, message="Timeout or cancelled")
+
+ return token_data
except Exception as e:
logger.error(f"Remote login flow failed: {e}")
@@ -350,6 +438,7 @@ class RemoteLoginHandler:
return {
'login_code': session.initial_login_code,
'qr_code_url': session.qr_code_url,
+ 'qr_target_url': session.qr_target_url,
'elapsed_seconds': elapsed,
'remaining_seconds': remaining,
'is_expired': remaining <= 0,