nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-19 07:32:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix temp dir on Kodi for qr-code

Browse Source
This commit is contained in:
Nirvana
2026-05-22 10:44:31 +02:00
parent 4f7a8cefdc
commit 6d37464484
3 changed files with 164 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/ui/kodi_notification_adapter.py
+107 -258
View File
@@ -8,12 +8,12 @@ Architecture:
"""
import os
import tempfile
import threading
import time
from typing import Callable, Optional
from ..utils.logger import logger
from ..utils.vfs import get_vfs # Import VFS
from .notification_interface import NotificationInterface, NotificationResult
# Import QR generator
@@ -27,19 +27,7 @@ except ImportError:
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
@@ -52,30 +40,23 @@ class PollingThread(threading.Thread):
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
@@ -83,30 +64,7 @@ class PollingThread(threading.Thread):
class QRCodeDialog:
"""
WindowDialog for displaying QR code with status monitoring
"""
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
expires_in: Expiration time in seconds
polling_thread: Optional polling thread to monitor
"""
def __init__(self, xbmcgui, xbmc, qr_image_path, login_code, expires_in, polling_thread=None):
self.xbmcgui = xbmcgui
self.xbmc = xbmc
self.qr_image_path = qr_image_path
@@ -116,190 +74,152 @@ class QRCodeDialog:
self.dialog = None
self.user_closed = False
self.closed_by_auth = False # FIX #2: separate flag for auth-triggered close
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"""
try:
# Create WindowDialog
self.dialog = self.xbmcgui.WindowDialog()
screen_width = self.dialog.getWidth()
screen_height = self.dialog.getHeight()
# 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 - Use a solid color label instead of ControlImage
bg = self.xbmcgui.ControlLabel(
dialog_x, dialog_y, dialog_width, dialog_height, label=""
)
# Set semi-transparent black background
bg = self.xbmcgui.ControlLabel(dialog_x, dialog_y, dialog_width, dialog_height, label="")
try:
bg.setColorDiffuse("0xE0000000")
except:
pass # Ignore if method not available
pass
self.dialog.addControl(bg)
# Title
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",
textColor="0xFFFFFFFF",
alignment=0x00000002, # Center
x=dialog_x + 50, y=title_y, width=dialog_width - 100, height=50,
label="[B]MagentaTV Remote Login[/B]", font="font30",
textColor="0xFFFFFFFF", alignment=0x00000002,
)
self.dialog.addControl(title)
# 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, qr_size, qr_size, self.qr_image_path
)
if self._file_exists(self.qr_image_path):
qr_image = self.xbmcgui.ControlImage(qr_x, qr_y, qr_size, qr_size, self.qr_image_path)
self.dialog.addControl(qr_image)
else:
logger.error(f"QR image not found: {self.qr_image_path}")
# Instructions
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",
textColor="0xFFFFFFFF",
alignment=0x00000002,
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(inst1)
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,
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(inst2)
# Time remaining label
time_y = inst2_y + 45
self.time_label = self.xbmcgui.ControlLabel(
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",
alignment=0x00000002,
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", alignment=0x00000002,
)
self.dialog.addControl(self.time_label)
# 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,
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(self.status_label)
# 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,
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(cancel)
# 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
self.user_closed = True
# FIX #2: only mark as user_closed if auth didn't trigger the close
self.user_closed = not self.closed_by_auth
self._stop_monitor()
logger.info("QR code dialog closed by user")
logger.info(f"QR code dialog closed (user_closed={self.user_closed})")
except Exception as e:
logger.error(f"Failed to show QR code dialog: {e}", exc_info=True)
finally:
self._cleanup()
@staticmethod
def _file_exists(filepath: str) -> bool:
try:
from ..utils.vfs import exists as vfs_exists
return vfs_exists(filepath)
except:
return os.path.exists(filepath)
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")
# FIX (new): handle expiry explicitly
if remaining <= 0:
logger.info("Monitor: Time expired, closing dialog")
if self.status_label:
self.status_label.setLabel(
"[COLOR green]Authentication successful![/COLOR]"
)
time.sleep(1) # Show success message briefly
self.status_label.setLabel("[COLOR red]Session expired[/COLOR]")
time.sleep(1)
self.close_dialog()
break
# Check for errors
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)
self.close_dialog(by_auth=True) # FIX #2: pass flag
break
if self.polling_thread.error:
logger.error(f"Monitor: Polling error: {self.polling_thread.error}")
if self.status_label:
@@ -308,15 +228,17 @@ class QRCodeDialog:
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):
def close_dialog(self, by_auth: bool = False):
"""Close dialog programmatically"""
# FIX #2: set closed_by_auth BEFORE calling close() so show() sees it
if by_auth:
self.closed_by_auth = True
if self.dialog:
try:
self.dialog.close()
@@ -324,7 +246,6 @@ class QRCodeDialog:
pass
def _cleanup(self):
"""Cleanup dialog resources"""
if self.dialog:
try:
del self.dialog
@@ -334,7 +255,6 @@ class QRCodeDialog:
@staticmethod
def _format_time(seconds: int) -> str:
"""Format seconds as MM:SS"""
if seconds <= 0:
return "Expired"
minutes = seconds // 60
@@ -343,30 +263,12 @@ class QRCodeDialog:
class KodiNotificationAdapter(NotificationInterface):
"""
Kodi notification adapter with fast QR generation and threading
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):
"""
Initialize Kodi notification adapter
Args:
http_manager: Optional HTTPManager instance
"""
super().__init__()
# Import Kodi modules
try:
import xbmc
import xbmcgui
import xbmcvfs
self.xbmcgui = xbmcgui
self.xbmc = xbmc
self.xbmcvfs = xbmcvfs
@@ -381,48 +283,23 @@ class KodiNotificationAdapter(NotificationInterface):
self._http_manager = http_manager
self._polling_thread = None
self._vfs = get_vfs(addon_subdir="temp")
@property
def supports_qr_display(self) -> bool:
"""Kodi can display QR codes"""
return True
@property
def supports_countdown(self) -> bool:
"""Kodi supports countdown via monitor thread"""
return True
@property
def is_blocking(self) -> bool:
"""Dialog is blocking but polling happens in thread"""
return True # From caller's perspective, it blocks
return True
def show_remote_login_with_polling(
self,
login_code: str,
qr_target_url: str,
expires_in: int,
interval: int,
poll_callback: Callable,
self, login_code, qr_target_url, expires_in, interval, poll_callback
) -> NotificationResult:
"""
Show remote login with integrated polling
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_target_url: The URL to encode in QR code (NOT the SVG URL!)
expires_in: Expiration time in seconds
interval: Polling interval
poll_callback: Function to call for polling
Returns:
NotificationResult
"""
if not self._kodi_available:
return NotificationResult.ERROR
@@ -430,33 +307,30 @@ class KodiNotificationAdapter(NotificationInterface):
self._is_cancelled = False
try:
# 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 generate QR code")
return NotificationResult.ERROR
self._qr_image_path = qr_image_path
# 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,
expires_in,
self.xbmcgui, self.xbmc,
qr_image_path, login_code, expires_in,
self._polling_thread,
)
self._qr_dialog.show()
# Step 4: Determine outcome
# FIX #4: stop the polling thread if the user cancelled
if self._qr_dialog.user_closed and self._polling_thread.is_alive():
logger.info("User cancelled — stopping polling thread")
self._polling_thread.stop()
self._polling_thread.join(timeout=2.0)
if self._polling_thread.auth_completed:
logger.info("Authentication successful")
return NotificationResult.CONTINUE
@@ -465,7 +339,7 @@ class KodiNotificationAdapter(NotificationInterface):
self._is_cancelled = True
return NotificationResult.CANCELLED
else:
logger.warning("Authentication timed out")
logger.warning("Authentication timed out or failed")
return NotificationResult.TIMEOUT
except Exception as e:
@@ -476,18 +350,8 @@ class KodiNotificationAdapter(NotificationInterface):
self._cleanup_qr_image()
def show_remote_login(
self, login_code: str, qr_target_url: str, expires_in: int, interval: int = 10
self, login_code, qr_target_url, expires_in, interval=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
@@ -496,28 +360,24 @@ class KodiNotificationAdapter(NotificationInterface):
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.xbmcgui, self.xbmc,
qr_image_path, login_code, expires_in,
None,
)
self._qr_dialog.show()
if self._qr_dialog.user_closed:
self._is_cancelled = True
return NotificationResult.CANCELLED
return NotificationResult.CONTINUE
# FIX #5: no polling thread means no auth — dialog expired naturally
return NotificationResult.TIMEOUT
except Exception as e:
logger.error(f"Failed to show QR dialog: {e}", exc_info=True)
@@ -527,14 +387,12 @@ class KodiNotificationAdapter(NotificationInterface):
self._cleanup_qr_image()
def update_countdown(self, remaining_seconds: int) -> bool:
"""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 and show notification"""
if not self._is_active:
return
@@ -545,20 +403,15 @@ class KodiNotificationAdapter(NotificationInterface):
self._qr_dialog.close_dialog()
self._qr_dialog = None
# Show result
if success:
self.xbmcgui.Dialog().notification(
"MagentaTV",
"Remote login successful!",
self.xbmcgui.NOTIFICATION_INFO,
3000,
"MagentaTV", "Remote login successful!",
self.xbmcgui.NOTIFICATION_INFO, 3000,
)
elif message and not self._is_cancelled:
self.xbmcgui.Dialog().notification(
"MagentaTV",
f"Remote login failed: {message}",
self.xbmcgui.NOTIFICATION_ERROR,
5000,
"MagentaTV", f"Remote login failed: {message}",
self.xbmcgui.NOTIFICATION_ERROR, 5000,
)
self._cleanup_qr_image()
@@ -567,28 +420,17 @@ class KodiNotificationAdapter(NotificationInterface):
logger.error(f"Failed to close dialog: {e}")
def is_cancelled(self) -> bool:
"""Check if cancelled"""
return self._is_cancelled
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]:
def _generate_qr_code(self, target_url: str) -> Optional[str]:
"""
Generate QR code PNG file from target URL
This is GENERIC - works for any provider!
No provider-specific logic here.
Args:
target_url: The URL to encode in the QR code
Returns:
Path to PNG file
Generate QR code PNG file using VFS for cross-platform compatibility.
Uses write_binary to avoid data corruption from text encoding roundtrips.
"""
try:
if not QR_GENERATOR_AVAILABLE:
@@ -598,9 +440,7 @@ class KodiNotificationAdapter(NotificationInterface):
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
@@ -608,28 +448,37 @@ class KodiNotificationAdapter(NotificationInterface):
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)
filename = f"magentatv_qr_{int(time.time())}.png"
filepath = self._vfs.join_path(filename)
with open(png_path, "wb") as f:
f.write(png_data)
logger.info(f"QR code saved to: {png_path}")
return png_path
# FIX #1: write raw bytes via write_binary, not text
if self._vfs.write_binary(filepath, png_data):
if self._vfs.exists(filepath):
logger.info(f"QR code saved using VFS: {filepath}")
return filepath
else:
logger.error(f"QR code file not found after write: {filepath}")
return None
else:
logger.error(f"Failed to write QR code using VFS: {filepath}")
return None
except Exception as e:
logger.error(f"Failed to generate QR code: {e}", exc_info=True)
return None
def _cleanup_qr_image(self):
"""Cleanup temporary QR image"""
if self._qr_image_path and os.path.exists(self._qr_image_path):
try:
os.remove(self._qr_image_path)
logger.debug(f"Cleaned up QR image: {self._qr_image_path}")
except Exception as e:
logger.warning(f"Failed to cleanup QR image: {e}")
finally:
self._qr_image_path = None
if not self._qr_image_path:
return
try:
if self._vfs.exists(self._qr_image_path):
if self._vfs.delete(self._qr_image_path):
logger.debug(f"Cleaned up QR image using VFS: {self._qr_image_path}")
else:
logger.warning(f"Failed to delete QR image: {self._qr_image_path}")
else:
logger.debug(f"QR image already cleaned up: {self._qr_image_path}")
except Exception as e:
logger.warning(f"Error during QR image cleanup: {e}")
finally:
self._qr_image_path = None
lib/streaming_providers/base/utils/vfs.py
+51 -1
View File
@@ -225,6 +225,46 @@ class VFS:
logger.error(f"Error writing file {filepath}: {e}")
return False
def write_binary(self, filepath: str, data: bytes) -> bool:
"""
Write binary content to file
Args:
filepath: File path to write
data: Raw bytes to write
Returns:
True if successful, False otherwise
"""
try:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if is_kodi_environment():
import xbmcvfs
dir_path = "/".join(filepath.split("/")[:-1])
if dir_path and not xbmcvfs.exists(dir_path):
xbmcvfs.mkdirs(dir_path)
with xbmcvfs.File(filepath, "w") as f:
bytes_written = f.write(bytearray(data))
logger.debug(
f"Kodi binary file write: {bytes_written} bytes to {filepath}"
)
return bytes_written > 0
else:
import pathlib
path = pathlib.Path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "wb") as f:
f.write(data)
return True
except Exception as e:
logger.error(f"Error writing binary file {filepath}: {e}")
return False
def delete(self, filepath: str) -> bool:
"""
Delete file
@@ -510,6 +550,16 @@ def write_text(
return get_vfs(config_dir, addon_subdir).write_text(filepath, content, encoding)
def write_binary(
filepath: str,
data: bytes,
config_dir: Optional[str] = None,
addon_subdir: str = "",
) -> bool:
"""Write binary file"""
return get_vfs(config_dir, addon_subdir).write_binary(filepath, data)
def read_json(
filepath: str, config_dir: Optional[str] = None, addon_subdir: str = ""
) -> Optional[dict]:
@@ -538,4 +588,4 @@ def delete(
def join_path(*parts, config_dir: Optional[str] = None, addon_subdir: str = "") -> str:
"""Join path components"""
vfs = get_vfs(config_dir, addon_subdir)
return vfs.join_path(*parts)
return vfs.join_path(*parts)
lib/streaming_providers/providers/joyn/auth.py
+6 -4
View File
@@ -679,6 +679,12 @@ class JoynAuthenticator(BaseOAuth2Authenticator):
"Accept-Encoding": "gzip, deflate",
}
# Pop session.send()-level kwargs BEFORE passing **kwargs to RawRequest.
# RawRequest.__init__() does not accept allow_redirects — that is a
# session.send() / session.request() concern only. Passing it to
# RawRequest() directly causes: "unexpected keyword argument 'allow_redirects'"
allow_redirects = kwargs.pop("allow_redirects", True)
# Honour explicit content_type kwarg
content_type = kwargs.pop("content_type", None)
if content_type:
@@ -697,10 +703,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator):
prepared = req.prepare()
try:
allow_redirects = kwargs.pop("allow_redirects", True)
req = RawRequest(method=method.upper(), url=url, headers=request_headers,
cookies=session.cookies, **kwargs)
prepared = req.prepare()
return session.send(prepared, timeout=self._config.timeout, allow_redirects=allow_redirects)
except Exception as e:
logger.debug(f"_make_7pass_request error: {e}")