#!/usr/bin/python # -*- coding: utf-8 -*- from __future__ import absolute_import, print_function import base64 import glob import io try: import six except ImportError: # six is a hard runtime dependency (python3-six, declared in # CONTROL/control) used throughout this module for Py2/3 text # helpers - there is no functional fallback without it, but a clear # message here beats a bare traceback on images missing the package. print("[VUTILS] FATAL: 'six' package not found - install python3-six") raise import socket import ssl import select import threading import types try: import urllib3 except ImportError: urllib3 = None import xml.etree.ElementTree as ET import zlib from datetime import datetime as _datetime from collections import OrderedDict from difflib import SequenceMatcher from json import dump, load, loads from Components.config import config from Components.NimManager import nimmanager from os import listdir, makedirs, remove, unlink, rename from os.path import basename, exists, getmtime, getsize, isfile, join, splitext from enigma import eTimer from random import choice from re import DOTALL, IGNORECASE, compile, findall, search, sub from shutil import copy2 from sys import maxsize from six import iteritems, unichr from six.moves import html_entities, html_parser from time import sleep, time, strftime, localtime from unicodedata import normalize from . import ( __version__, country_codes, PY2, PY3, PORT, PLUGIN_ROOT, PROXY_HOST, PROXY_BASE_URL, PROXY_STATUS_URL, FLAG_CACHE_DIR, LOG_FILE, CACHE_FILE, UNMATCHED_FILE, HOST_MAIN, ALIAS_FILE, INSTALLER_URL ) from .epg_name_utils import ( clean_name_for_similarity, tokenize_for_compat, token_pair_compatible, tokens_compatible, ) """ ######################################################### # # # Vavoo Stream Live Plugin # # Created by Lululla (https://github.com/Belfagor2005) # # License: CC BY-NC-SA 4.0 # # https://creativecommons.org/licenses/by-nc-sa/4.0 # # Last Modified: 202600503 # # # # Credits: # # - Original concept by Lululla # # - Background images by @oktus # # - Additional contributions by Qu4k3 # # - Linuxsat-support.com & Corvoboys communities # # # # Usage of this code without proper attribution # # is strictly prohibited. # # For modifications and redistribution, # # please maintain this credit header. # ######################################################### """ # Disable SSL warnings if urllib3 is not None: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) _original_getaddrinfo = socket.getaddrinfo try: from . import channel_alias alias_available = True except Exception: alias_available = False print("[VavooEPGMatcher] channel_alias module not found, using default matching") pass try: from urllib.parse import quote # , unquote except ImportError: from urllib import quote # , unquote try: import requests except Exception: requests = None try: unicode except NameError: unicode = str try: from Components.AVSwitch import AVSwitch except ImportError: from Components.AVSwitch import eAVControl as AVSwitch _epg_lock = threading.Lock() _starting_lock = threading.Lock() _unmatched_lock = threading.Lock() LOG_MAX_BYTES = 1024 * 1024 DEBUG_ENABLED = str( __import__("os").environ.get( "VAVOO_DEBUG", "0")).lower() in ( "1", "true", "yes", "on") def set_debug_enabled(enabled): """Toggle DEBUG-level logging at runtime (config menu "Debug logging"). VAVOO_DEBUG only takes effect at process start and, since the proxy runs as a thread inside the same Enigma2 process, an env var set in an SSH session never reaches it anyway (Enigma2 itself is started by the box's init supervisor, not that shell). This is checked on every debug() call instead, so a config toggle takes effect immediately with no restart needed. """ global DEBUG_ENABLED DEBUG_ENABLED = bool(enabled) def _rotate_log_if_needed(): try: if isfile(LOG_FILE) and getsize(LOG_FILE) >= LOG_MAX_BYTES: backup = LOG_FILE + ".1" try: if isfile(backup): remove(backup) except Exception: pass try: __import__("os").rename(LOG_FILE, backup) except Exception: pass except Exception: pass def _safe_console_write(line): try: import sys sys.stdout.write(line + "\n") sys.stdout.flush() except Exception: pass def _append_to_log(line): try: _rotate_log_if_needed() with open(LOG_FILE, "a") as log_file: log_file.write(line + "\n") except Exception: pass def log(msg, level="INFO", area="VUTILS"): try: msg = ensure_str(msg, errors='ignore') except Exception: try: msg = str(msg) except Exception: msg = '' line = "[{0}] [{1}] [{2}] {3}".format( _datetime.now().strftime("%Y-%m-%d %H:%M:%S"), level, area, msg ) _safe_console_write(line) _append_to_log(line) return line def debug(msg, area="VUTILS"): if DEBUG_ENABLED: return log(msg, level="DEBUG", area=area) return None def warning(msg, area="VUTILS"): return log(msg, level="WARNING", area=area) def error(msg, area="VUTILS"): return log(msg, level="ERROR", area=area) def log_exception(msg="", area="VUTILS"): import traceback if msg: error(msg, area=area) try: tb = traceback.format_exc() if not tb or tb.strip() == "NoneType: None": tb = "".join(traceback.format_stack()[:-1]) for line in tb.rstrip().splitlines(): error(line, area=area) except Exception as e: error("Failed to capture traceback: {0}".format(e), area=area) def trace_error(prefix="", area="VUTILS"): log_exception(prefix, area=area) def plugin_print(*args, **kwargs): sep = kwargs.get('sep', ' ') end = kwargs.get('end', '\n') level = kwargs.get('level', 'INFO') area = kwargs.get('area', 'VUTILS') try: msg = sep.join([ensure_str(x, errors='ignore') for x in args]) except Exception: try: msg = sep.join([str(x) for x in args]) except Exception: msg = '' if end and msg.endswith('\n'): msg = msg.rstrip('\n') return log(msg, level=level, area=area) def make_print(area, level="INFO"): """Return a drop-in print() replacement that routes through log().""" def _module_print(*args, **kwargs): kwargs.setdefault("area", area) kwargs.setdefault("level", level) return plugin_print(*args, **kwargs) return _module_print PLUGIN_PATH = PLUGIN_ROOT if PY3: from urllib.request import urlopen, Request from urllib.error import URLError, HTTPError ssl_context = ssl.create_default_context() for _ssl_opt in ( "OP_NO_SSLv2", "OP_NO_SSLv3", "OP_NO_TLSv1", "OP_NO_TLSv1_1"): ssl_context.options |= getattr(ssl, _ssl_opt, 0) unichr_func = unichr else: from urllib2 import urlopen, Request, URLError, HTTPError ssl = None ssl_context = None unichr_func = chr print = make_print("VUTILS") log("===== Vavoo session start =====", area="VUTILS") def getDNSinfo(): """Return (local_dns, external_dns). Never raises – returns 'n/a' on failure.""" dns_box = "n/a" dns_external = "n/a" try: with open("/etc/resolv.conf", "r") as f: for line in f: if line.startswith("nameserver"): dns_box = line.split()[1] break except BaseException: dns_box = "n/a" try: data = urlopen("https://1.1.1.1/cdn-cgi/trace", timeout=5).read() data = data.decode("utf-8") for line in data.split("\n"): if line.startswith("h="): dns_external = line.split("=")[1].strip() break except Exception as e: debug("External DNS check failed: {}".format(e)) return dns_box, dns_external def _log_dns_info_async(): """Runs getDNSinfo() (a real network call) off the import path so plugin/menu load never blocks on a slow/unreachable network.""" try: dns_box, dns_ext = getDNSinfo() except Exception: dns_box, dns_ext = "n/a", "n/a" print("DNS box:", dns_box) print("DNS out:", dns_ext) print("Vavoo Version: ", __version__) _dns_info_thread = threading.Thread(target=_log_dns_info_async) _dns_info_thread.setDaemon(True) _dns_info_thread.start() def get_screen_width(): """Get current screen width""" try: from enigma import getDesktop desktop = getDesktop(0) width = desktop.size().width() print("Screen width detected: %d" % width) return width except Exception as e: print("Error getting screen width: %s" % str(e)) return 1920 class AspectManager(object): """Manages aspect ratio settings for the plugin""" def __init__(self): try: self.init_aspect = self.get_current_aspect() print("[INFO] Initial aspect ratio:", self.init_aspect) except Exception as e: print("[ERROR] Failed to initialize aspect manager:", str(e)) self.init_aspect = 0 def get_current_aspect(self): """Get current aspect ratio setting""" try: aspect = AVSwitch().getAspectRatioSetting() return int(aspect) if aspect is not None else 0 except (ValueError, TypeError, Exception) as e: print("[ERROR] Failed to get aspect ratio:", str(e)) return 0 def restore_aspect(self): """Restore original aspect ratio""" try: if hasattr(self, 'init_aspect') and self.init_aspect is not None: print("[INFO] Restoring aspect ratio to:", self.init_aspect) AVSwitch().setAspectRatio(self.init_aspect) else: print("[WARNING] No initial aspect ratio to restore") except Exception as e: print("[ERROR] Failed to restore aspect ratio:", str(e)) aspect_manager = AspectManager() class_types = (type,) if PY3 else (type, types.ClassType) text_type = six.text_type binary_type = six.binary_type MAXSIZE = maxsize _UNICODE_MAP = { k: unichr(v) for k, v in iteritems( html_entities.name2codepoint)} _ESCAPE_RE = compile(r"[&<>\"']") _UNESCAPE_RE = compile(r"&\s*(#?)(\w+?)\s*;") _ESCAPE_DICT = { "&": "&", "<": "<", ">": ">", '"': """, "'": "'", } std_headers = { 'User-Agent': 'Mozilla/5.0 (X11; U; Linux x86_64; en-US; rv:1.9.2.6) Gecko/20100627 Firefox/3.6.6', 'Accept-Charset': 'ISO-8859-1,utf-8;q=0.7,*;q=0.7', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-us,en;q=0.5'} USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.6312.88 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1" ] def RequestAgent(): """Get random user agent from list""" return choice(USER_AGENTS) def ensure_str(s, encoding="utf-8", errors="strict"): if s is None: return "" if isinstance(s, text_type): return s if isinstance(s, binary_type): return s.decode(encoding, errors) return text_type(s) def html_escape(value): """Escape HTML special characters""" value = ensure_str(value, errors='ignore').strip() return _ESCAPE_RE.sub(lambda m: _ESCAPE_DICT[m.group(0)], value) def html_unescape(value): """Unescape HTML entities""" return _UNESCAPE_RE.sub(_convert_entity, ensure_str(value).strip()) def _convert_entity(m): """Helper for HTML entity conversion, compatible with Python 2 and 3""" if m.group(1) == "#": try: return unichr(int(m.group(2)[1:], 16)) if m.group( 2)[:1].lower() == "x" else unichr(int(m.group(2))) except ValueError: return "&#%s;" % m.group(2) return _UNICODE_MAP.get(m.group(2), "&%s;" % m.group(2)) def b64decoder(data): """Robust base64 decoding with padding correction""" if not data: return "" try: data = ensure_str(data, errors='ignore').strip() pad = len(data) % 4 if pad == 1: return "" if pad: data += "=" * (4 - pad) decoded = base64.b64decode(data.encode('ascii')) try: return decoded.decode('utf-8') except UnicodeDecodeError: return decoded except Exception as e: print("Base64 decoding error: %s" % e) return "" def getUrl(url, timeout=30, retries=3, backoff=2): """Fetch URL with exponential backoff retry logic""" # detect 451 HTTP_451_SENTINEL = "__HTTP451__" headers = {'User-Agent': RequestAgent()} if not url: raise ValueError("Empty URL passed to getUrl") url = ensure_str(url, errors='ignore').strip() if not url.startswith(("http://", "https://")): raise ValueError("Invalid URL (missing scheme): %s" % url) for i in range(retries): try: # No socket.setdefaulttimeout() here - every urlopen() call # below already gets an explicit timeout=timeout, and this # runs in the same process as the proxy, whose own comments # elsewhere warn that mutating the global default poisons # unrelated sockets (e.g. streaming) that don't set their own. request = Request(url, headers=headers) if PY3: import ssl unverified_context = ssl.create_default_context() unverified_context.check_hostname = False unverified_context.verify_mode = ssl.CERT_NONE response = urlopen( request, timeout=timeout, context=unverified_context) else: # Python 2: prova a usare ssl se disponibile, altrimenti # fallback try: import ssl # Python 2.7.9+ supporta ssl._create_unverified_context unverified_context = ssl._create_unverified_context() if hasattr( ssl, '_create_unverified_context') else None if unverified_context: response = urlopen( request, timeout=timeout, context=unverified_context) else: response = urlopen(request, timeout=timeout) except ImportError: response = urlopen(request, timeout=timeout) data = response.read() return data except HTTPError as e: # detect 451 code = getattr(e, 'code', None) if code == 451: print("HTTP 451 for URL: {0}".format(url)) return HTTP_451_SENTINEL if i < retries - 1: wait_time = backoff ** i print( "HTTP error {0} on attempt {1}, retrying in {2} seconds...".format( code, i + 1, wait_time)) select.select([], [], [], wait_time) continue print( "Failed after {0} attempts for URL: {1}".format( retries, url)) print("HTTPError: {0}".format(e)) return "" except (URLError, socket.timeout, socket.error) as e: err_no = getattr(e, 'errno', None) if err_no is None and getattr(e, 'args', None): err_no = e.args[0] retryable_socket_errors = (104, 110, 111) is_retryable_socket = isinstance(e, socket.error) and ( err_no in retryable_socket_errors or err_no is None ) is_retryable = not isinstance( e, socket.error) or is_retryable_socket if is_retryable and i < retries - 1: wait_time = backoff ** i # Exponential backoff print( "Attempt {0} failed, retrying in {1} seconds...".format( i + 1, wait_time)) select.select([], [], [], wait_time) else: print( "Failed after {0} attempts for URL: {1}".format( retries, url)) print("Error: {0}".format(e)) return "" except Exception as e: if i < retries - 1: wait_time = backoff ** i print( "Unexpected error on attempt {0}, retrying in {1} seconds...".format( i + 1, wait_time)) print("Error: {0}".format(e)) select.select([], [], [], wait_time) continue print( "Failed after {0} attempts for URL: {1}".format( retries, url)) print("Unexpected error: {0}".format(e)) try: trace_error() except BaseException: pass return "" def get_external_ip(): """Get external IP using multiple fallback services""" from subprocess import Popen, PIPE def _decode_cmd_output(value): if value is None: return "" if isinstance(value, binary_type): return value.decode('utf-8', 'ignore').strip() return ensure_str(value, errors='ignore').strip() services = [ # --max-time bounds curl itself (Python 2 has no # Popen.communicate(timeout=...), so this is the only reliable # way to bound both versions) - this call runs while holding # addon_sig_lock (see refresh_addon_sig_if_needed()), which # every channel resolve retry also needs, so an unbounded hang # here previously meant an unbounded freeze for all playback. lambda: Popen( [ 'curl', '-s', '--max-time', '5', 'ifconfig.me'], stdout=PIPE).communicate()[0], ] if requests is not None: services.extend([ lambda: requests.get( 'https://v4.ident.me', timeout=5).text.strip(), lambda: requests.get( 'https://api.ipify.org', timeout=5).text.strip(), lambda: requests.get( 'https://api.myip.com', timeout=5).json().get( "ip", "").strip(), lambda: requests.get( 'https://checkip.amazonaws.com', timeout=5).text.strip(), ]) for service in services: try: ip = service() ip = _decode_cmd_output(ip) if ip: return ip except Exception: continue return None def set_cache(key, data, timeout): file_path = join(PLUGIN_PATH, key + '.json') try: if not isinstance(data, dict): data = {"value": data} # _is_cache_valid() requires both of these to consider a cache # entry valid - previously never written here, so timeout had no # effect and get_cache() would always treat this entry as expired. data['sigValidUntil'] = int(time()) + timeout data['ip'] = get_external_ip() temp_path = file_path + ".tmp" if PY2: converted_data = convert_to_unicode(data) with io.open(temp_path, 'w', encoding='utf-8') as cache_file: dump( converted_data, cache_file, indent=4, ensure_ascii=False) else: with io.open(temp_path, 'w', encoding='utf-8') as cache_file: dump(data, cache_file, indent=4, ensure_ascii=False) rename(temp_path, file_path) except Exception as e: print("Error saving cache:", e) trace_error() def convert_to_unicode(data): if isinstance(data, dict): return {convert_to_unicode(key): convert_to_unicode(value) for key, value in data.items()} elif isinstance(data, list): return [convert_to_unicode(element) for element in data] elif PY2 and isinstance(data, str): # Decode strings to Unicode for Python 2 return data.decode('utf-8', 'ignore') elif PY2 and isinstance(data, unicode): return data else: return data def get_cache(key): file_path = join(PLUGIN_PATH, key + '.json') if not (exists(file_path) and getsize(file_path) > 0): return None try: data = _read_json_file(file_path) if isinstance(data, str): data = {"value": data} _write_json_file(file_path, data) if not isinstance(data, dict): print( "Unexpected data format in {}: Expected a dict, got {}".format( file_path, type(data))) remove(file_path) return None if _is_cache_valid(data): return data.get('value') except ValueError as e: print("Error decoding JSON from", file_path, ":", e) trace_error() except Exception as e: print("Unexpected error reading cache file {}:".format(file_path), e) remove(file_path) trace_error() return None def _read_json_file(file_path): with io.open(file_path, 'r', encoding='utf-8') as f: return load(f) def _write_json_file(file_path, data): temp_path = file_path + ".tmp" with io.open(temp_path, 'w', encoding='utf-8') as f: dump(data, f, indent=4, ensure_ascii=False) rename(temp_path, file_path) def _is_cache_valid(data): return ( data.get('sigValidUntil', 0) > int(time()) and data.get('ip', "") == get_external_ip() ) # ============================================================================ # FUNCTIONS FOR VAVOO PROXY # ============================================================================ def getAuthSignature(): """Get authentication - ALWAYS use proxy""" print("Using proxy authentication system") return "PROXY_ACTIVE" def get_new_auth_signature(): """ New Vavoo authentication system via local proxy Returns a valid token for the proxy """ try: print("Using new proxy authentication system...") try: req = Request(PROXY_STATUS_URL) response = urlopen(req, timeout=5) if response.getcode() == 200: data = loads(response.read().decode('utf-8')) if data.get("initialized", False): print("Proxy active and running") return "PROXY_ACTIVE" except BaseException as e: debug("Proxy status check failed, will try starting it: {}".format(e)) try: from .vavoo_proxy import run_proxy_in_background print("Starting proxy in background...") run_proxy_in_background() select.select([], [], [], 5) return "PROXY_STARTED" except Exception as e: trace_error() print("Proxy start error: " + str(e)) except Exception as e: trace_error() print("New auth error: " + str(e)) print("Falling back to old authentication system") return getAuthSignature() def get_proxy_channels(country_name): """Get channels for a country from proxy - with retry""" country_name = ensure_str(country_name, errors='ignore').strip() max_retries = 3 for attempt in range(max_retries): try: print("Getting channels for '" + str(country_name) + "' (attempt " + str(attempt + 1) + "/" + str(max_retries) + ")") # URL-encode encoded_country = quote(country_name.encode( 'utf-8')) if PY2 else quote(country_name) # Build URL proxy_url = PROXY_BASE_URL + \ "/channels?country={}".format(encoded_country) # Fetch with timeout response = getUrl(proxy_url, timeout=15) print("Request URL: " + proxy_url) if not response: print( "Empty response for '" + str(country_name) + "'") continue # Parse JSON channels = loads(response) if not isinstance(channels, list): print("Invalid response format: " + str(type(channels))) continue print("Successfully got " + str(len(channels)) + " channels for '" + str(country_name) + "'") # Process channels processed_channels = [] for channel in channels: if isinstance(channel, dict): channel_id = channel.get('id', '') if not channel_id: continue # Build proxy URL proxy_stream_url = PROXY_BASE_URL + \ "/vavoo?channel={}".format(channel_id) processed_channels.append({ 'id': channel_id, 'name': channel.get('name', 'Unknown'), 'url': proxy_stream_url, 'logo': channel.get('logo', ''), 'country': channel.get('country', country_name) }) return processed_channels except Exception as e: print("Attempt " + str(attempt + 1) + " failed for '" + str(country_name) + "': " + str(e)) if attempt < max_retries - 1: sleep(2) # Wait before retry print("All attempts failed for '" + str(country_name) + "'") return [] def get_proxy_catalog_url(): """ Get the proxy catalog URL """ return PROXY_BASE_URL + "/catalog" def get_proxy_playlist_url(): """ Get the proxy playlist URL """ return PROXY_BASE_URL + "/playlist.m3u" def get_proxy_status(): """Get detailed proxy status""" try: if requests is not None: response = requests.get(PROXY_STATUS_URL, timeout=3) if response.status_code == 200: return response.json() else: req = Request(PROXY_STATUS_URL) response = urlopen(req, timeout=3) if response.getcode() == 200: return loads(response.read().decode('utf-8', 'ignore')) except BaseException: return None return None def is_proxy_running(): """Check if the proxy is running""" try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: return s.connect_ex((PROXY_HOST, PORT)) == 0 finally: s.close() except BaseException: return False def is_proxy_ready(timeout=2): """Check if the proxy is ready to receive requests. The proxy's HTTP server only binds its port after the full channel catalog has loaded, so while that is in progress every connection is refused. getUrl() retries refused connections with exponential backoff (seconds of blocking per call) which is fine for a one-off fetch but far too slow for a readiness poll called every few hundred ms from the UI/reactor thread. Check the port cheaply first (a plain connect_ex, effectively instant) and only pay for the HTTP round trip - with a single attempt, since we already know the port is open - once it is actually listening. """ try: if not is_proxy_running(): return False response = getUrl(PROXY_STATUS_URL, timeout=timeout, retries=1) if response: data = loads(response) return data.get("initialized", False) return False except BaseException: return False _original_getAuthSignature = getAuthSignature def getAuthSignature(): """ Wrapper that uses the proxy first, then falls back to the old system """ print("getAuthSignature called...") try: if is_proxy_running(): print("Proxy active, using new system") return "PROXY_AUTH" except BaseException: trace_error() pass print("Falling back to old authentication system") return _original_getAuthSignature() # =================================== def fetch_vec_list(): """Fetch vector list from GitHub""" try: url = "{}/data.json".format(HOST_MAIN) if requests is not None: # Use requests if available response = requests.get(url, timeout=10) vec_list = response.json() else: # Fallback to urllib req = Request(url) response = urlopen(req, timeout=10) data = response.read() if isinstance(data, bytes): data = data.decode('utf-8', 'ignore') vec_list = loads(data) set_cache("vec_list", vec_list, 3600) print( "[Fetch] Vector list loaded: {} entries".format( len(vec_list) if vec_list else 0)) return vec_list except Exception as e: print("[Fetch] Vector list error: {}".format(str(e))) return None def _version_tuple(version_str): """Turn '1.9' / '1.10' into comparable (1, 9) / (1, 10) tuples - plain string comparison would wrongly say '1.9' > '1.10'.""" parts = [] for chunk in sub(r'[^0-9.]', '', version_str or '').split('.'): try: parts.append(int(chunk)) except ValueError: parts.append(0) return tuple(parts) if parts else (0,) def is_remote_version_newer(local_version, remote_version): """True if remote_version is numerically greater than local_version.""" return _version_tuple(remote_version) > _version_tuple(local_version) def check_remote_installer_version(): """Fetch installer.sh from GitHub and pull out its version and changelog. Returns (version, changelog, raw_content), any/all of which are None on failure. installer.sh is expected to contain lines shaped like: version='1.76' changelog="- line one - line two" raw_content is returned too so callers that want to actually run the installer don't need a second fetch. """ try: content = getUrl(INSTALLER_URL, timeout=10, retries=2) if not content: print("[Update] Could not fetch installer.sh") return None, None, None content = ensure_str(content, errors='ignore') version_match = search(r"version\s*=\s*['\"]([^'\"]+)['\"]", content) if not version_match: print("[Update] Could not find version= in installer.sh") return None, None, None remote_version = version_match.group(1).strip() changelog_match = search( r'changelog\s*=\s*"(.*?)"', content, IGNORECASE | DOTALL) changelog = changelog_match.group( 1).strip() if changelog_match else "" return remote_version, changelog, content except Exception as e: print("[Update] Error checking installer.sh version: {}".format(e)) return None, None, None def remove_parentheses(text): """Remove parentheses and their content from text""" return sub( r'\s*\([^()]*\)\s*', ' ', ensure_str( text, errors='ignore')).strip() def purge(directory, pattern): """Delete files matching pattern in directory""" for f in listdir(directory): file_path = join(directory, f) if isfile(file_path) and search(pattern, f): remove(file_path) """ # def ReloadBouquets(delay=2000): # from enigma import eDVBDB, eTimer # try: # def do_reload(): # try: # db = eDVBDB.getInstance() # db.reloadBouquets() # db.reloadServicelist() # print("Bouquets reloaded successfully") # except Exception as e: # print("Error during service reload: " + str(e)) # reload_timer = eTimer() # try: # reload_timer.callback.append(do_reload) # except BaseException: # reload_timer.timeout.connect(do_reload) # reload_timer.start(delay, True) # except Exception as e: # print("Error setting up service reload: " + str(e)) # do_reload() """ def ReloadBouquets(delay=500): """Reload bouquets after delay (non‑blocking in main thread).""" from enigma import eDVBDB def do_reload(): try: db = eDVBDB.getInstance() db.reloadBouquets() db.reloadServicelist() print("Bouquets reloaded successfully") except Exception as e: print("Error during service reload: " + str(e)) if delay <= 0: do_reload() return # If in main thread, use non‑blocking timer if threading.current_thread() is threading.main_thread(): try: from twisted.internet import reactor if reactor.running: reactor.callLater(delay / 1000.0, do_reload) return except Exception as e: debug( "reactor.callLater unavailable, falling back to eTimer: {}".format(e)) # Fallback: eTimer timer = eTimer() try: timer.callback.append(do_reload) except AttributeError: timer.timeout.connect(do_reload) timer.start(delay, True) else: # Background thread: safe to sleep import time time.sleep(delay / 1000.0) do_reload() def ensure_sref_trailing_colon(sref): if sref and not sref.endswith(':'): return sref + ':' return sref def unique_fallback_sref(servicetype, channel_id): """Build a fallback (no Rytec/EPG match) service reference whose sid:tsid pair is unique per Vavoo channel_id, instead of the old literal "servicetype:0:0:0:0:0:0:0:0:0:" used for every unmatched channel. Enigma2's eEPGCache indexes events by the sid:tsid:onid:namespace portion of a service reference, not by the stream URL appended after it - so every unmatched channel sharing that identical all-zero tuple was, as far as EPG lookup is concerned, literally the same service: whatever programme data ever ended up cached under that one shared null key got shown for all of them at once (confirmed via a user's box: a cluster of unrelated, genuinely-unmatched channels - "13EME RUE", "A LA CARTE 1-11", "20 MINUTES TV" - all displaying one other channel's guide data). onid/namespace are deliberately left at 0: every real Rytec-sourced sref uses a non-zero namespace (the known satellite/terrestrial/ cable ranges), so a synthetic entry here can never collide with an actual matched channel - only sid/tsid need to vary to keep unmatched channels apart from each other. """ h = zlib.crc32( ensure_str( channel_id, errors='ignore').encode('utf-8')) & 0xffffffff sid = (h & 0xffff) or 1 tsid = ((h >> 16) & 0xffff) or 1 return "%s:0:1:%x:%x:0:0:0:0:0:" % (servicetype, sid, tsid) def sanitizeFilename(filename): """Sanitize filename for safe filesystem use""" filename = ensure_str(filename, errors='ignore') # Remove unsafe characters filename = sub(r'[\\/:*?"<>|\0]', '', filename) filename = ''.join(c for c in filename if ord(c) > 31) normalized = normalize('NFKD', filename).encode('ascii', 'ignore') if isinstance(normalized, binary_type): filename = normalized.decode('ascii', 'ignore') else: filename = normalized filename = filename.rstrip('. ').strip() # Handle reserved names reserved = ( ["CON", "PRN", "AUX", "NUL"] + ["COM" + str(i) for i in range(1, 10)] + ["LPT" + str(i) for i in range(1, 10)] ) if filename.upper() in reserved or not filename: if filename: filename = "__" + filename else: filename = "__" # Truncate if necessary if len(filename) > 255: base, ext = splitext(filename) ext = ext[:254] filename = base[:255 - len(ext)] + ext return filename or "__" def decodeHtml(text): text = ensure_str(text, errors='ignore') if PY3: import html text = html.unescape(text) else: h = html_parser.HTMLParser() text = h.unescape(text) replacements = { '&': '&', ''': "'", '<': '<', '>': '>', '–': '-', '"': '"', 'ñ': 'ñ', '’': "'", ' ': ' ', '=': '=', '?': '?', ',': ',', '.': '.', ':': ':', '(': '(', ')': ')', '!': '!', '$': '$', '#': '#', '*': '*', '_': '_', '[': '[', ']': ']', '½': '1/2', '˜': '~', '“': '"', '”': '"' } for entity, char in replacements.items(): text = text.replace(entity, char) return text.strip() def remove_line(filename, pattern): """Remove lines containing pattern from file""" if not isfile(filename): return with open(filename, 'r') as f: lines = [line for line in f if pattern not in line] with open(filename, 'w') as f: f.writelines(lines) def getserviceinfo(service_ref): """Get service name and URL from service reference""" try: from ServiceReference import ServiceReference ref = ServiceReference(service_ref) return ref.getServiceName(), ref.getPath() except Exception: return None, None # ============================================================================ # FLAG DOWNLOAD FUNCTIONS # ============================================================================ def initialize_cache_with_local_flags(): """Copy all local flags from skin/cowntry/ to cache directory. Skips the copy if a previous run in this boot session already populated the (tmpfs) cache dir, since FLAG_CACHE_DIR contents only disappear on reboot and the source files never change without one. """ local_dir = join(PLUGIN_PATH, 'skin/cowntry') cache_dir = FLAG_CACHE_DIR if not exists(local_dir): print("Local flags directory not found: %s" % local_dir) return 0 marker = join(cache_dir, '.flags_initialized') if exists(marker): print("Flag cache already initialized, skipping copy") return 0 # Python 2 compatible directory creation if not exists(cache_dir): try: makedirs(cache_dir) except Exception as e: debug("Could not create flag cache dir {}: {}".format(cache_dir, e)) copied = 0 for filename in listdir(local_dir): if filename.lower().endswith('.png'): src = join(local_dir, filename) dst = join(cache_dir, filename.lower()) try: with open(src, 'rb') as f: # Check PNG header header = f.read(8) if header == b'\x89PNG\r\n\x1a\n': copy2(src, dst) copied += 1 print("Copied local flag: %s" % filename) else: print("Skipping invalid PNG: %s" % filename) except Exception as e: print("Error copying %s: %s" % (filename, e)) print("Initialized cache with %d local flags" % copied) try: with open(marker, 'wb') as f: f.write(b'1') except Exception as e: debug("Could not write flag cache marker {}: {} (flags will " "be re-copied on next start)".format(marker, e)) return copied def download_flag_online( country_name, cache_dir=FLAG_CACHE_DIR, screen_width=None): """ Download country flag from online service (TV Garden style) Returns: (success, flag_path_or_error_message) """ try: # 1. Determine screen width if not provided if screen_width is None: screen_width = get_screen_width() # must return int print( "Processing %s with screen_width=%d" % (country_name, screen_width)) # 2. Get country code country_code = get_country_code(country_name) if not country_code: return False, "No country code found for: %s" % country_name country_code_lower = country_code.lower() special_flags = ['bk', 'internat'] if country_code_lower in special_flags: local_path = join( PLUGIN_PATH, 'skin/cowntry', '%s.png' % country_code_lower) if exists(local_path): print( "Using special flag: %s -> %s" % (country_name, local_path)) return True, local_path # 3. Create cache directory (Python 2 safe) if not exists(cache_dir): try: makedirs(cache_dir) except Exception as e: debug( "Could not create flag cache dir {}: {}".format( cache_dir, e)) # 4. Cache file path cache_file = join(cache_dir, "%s.png" % country_code_lower) # 5. Check fresh cache (<7 days) if exists(cache_file): try: file_age = time() - getmtime(cache_file) if file_age < 604800: print("Cache HIT: %s" % country_name) return True, cache_file except Exception as e: debug("Flag cache freshness check failed for {}: {}".format( cache_file, e)) # 6. Set fixed flag dimensions if screen_width >= 2560: # WQHD width, height = 80, 60 elif screen_width >= 1920: # FHD width, height = 60, 45 else: # HD width, height = 40, 30 # 7. Build URL url = "https://flagcdn.com/%dx%d/%s.png" % ( width, height, country_code_lower) print("Downloading %s (%dx%d) from: %s" % (country_name, width, height, url)) # 8. Download req = Request(url, headers={'User-Agent': 'Vavoo-Stream/1.0'}) try: if PY3: response = urlopen(req, timeout=5, context=ssl_context) else: response = urlopen(req, timeout=5) except Exception as e: print("Network error for %s: %s" % (country_name, e)) return False, "Network error: %s" % e # 9. Read data if response.getcode() != 200: return False, "Download failed (HTTP %d)" % response.getcode() flag_data = response.read() try: response.close() except Exception: pass # 10. Validate small file if len(flag_data) < 100: print( "Warning: Flag file too small (%d bytes)" % len(flag_data)) # 11. Validate PNG header in memory BEFORE writing (no double file # open) if flag_data[:8] != b'\x89PNG\r\n\x1a\n': print("ERROR: Downloaded data is not a valid PNG file!") return False, "Invalid PNG file downloaded" # 12. Save to cache try: with open(cache_file, 'wb') as f: f.write(flag_data) print("Flag %dx%d saved: %s (%d bytes)" % (width, height, cache_file, len(flag_data))) return True, cache_file except Exception as e: print("Error saving to cache: %s" % e) return False, "Save error: %s" % e except Exception as e: print("Flag download error: %s" % e) return False, "Flag download error: %s" % e def get_country_code_from_bouquet_name(name): """Extract country code from a bouquet display name (e.g., 'Italy', 'Italy ➾ Sports').""" separators = ["➾", "⟾", "->", "→"] base_name = name for sep in separators: if sep in name: base_name = name.split(sep)[0].strip() break # Case-insensitive lookup: .capitalize() breaks multi-word names # like "United Kingdom" -> "United kingdom", which then can't match # the Title-Case keys in country_codes. base_name_lower = base_name.lower() for key, code in country_codes.items(): if key.lower() == base_name_lower: return code return None # Language/demonym adjectives sometimes used as IPTV group names instead # of a country name (e.g. "Italian", "German") - not real country names, # so they don't belong in __init__.py's country_codes table, but # get_country_code() has always accepted them too. _EXTRA_COUNTRY_ALIASES = { 'italia': 'it', 'italiana': 'it', 'italian': 'it', 'german': 'de', 'french': 'fr', 'spanish': 'es', 'english': 'gb', 'british': 'gb', 'default': 'us', } def get_country_code(country_name): """ Extract country code from country name. Handles formats like 'France', 'France ➾ Sports', etc. Returns ISO 2-letter country code or empty string if not found. """ country_name = ensure_str(country_name, errors='ignore').strip() if not country_name: return "" if any(char in country_name for char in '0123456789.'): return "" separators = ["➾", "⟾", "->", "→", "»", "›"] for sep in separators: if sep in country_name: country_name = country_name.split(sep)[0].strip() break country_name = country_name.strip() if len(country_name) < 2: return "" # country_codes (__init__.py) is the single source of truth for real # country names - checked case-insensitively. name_lower = country_name.lower() for key, code in country_codes.items(): if key.lower() == name_lower: return code if name_lower in _EXTRA_COUNTRY_ALIASES: return _EXTRA_COUNTRY_ALIASES[name_lower] # Partial match against the canonical table, for group names that # embed a country name inside a longer string (e.g. "France Sport" # contains "France"). Deliberately one-directional: only "a known # country name is a substring of the input" - never the reverse # ("input is a substring of a country name"), which used to let a # short/abbreviated input spuriously match any country name that # happens to contain those letters (e.g. "Ira" matching "United # Arab Emirates" via "em-IRA-tes") with no real use case needing # it - any legitimate abbreviation is already handled by the # exact-match loop above via its own country_codes entry. Prefer # the longest/most specific matching name when more than one embeds # in the input, alphabetically tie-broken so the result never # depends on Python's dict iteration order (not guaranteed on # Python 2). candidates = [key for key in country_codes if key.lower() in name_lower] if candidates: best = sorted(candidates, key=lambda k: (-len(k), k))[0] return country_codes[best] return "" def cleanup_flag_cache(max_age_days=7): """ Remove old cached flag files from cache directory. Only files older than max_age_days are deleted. """ cache_dir = FLAG_CACHE_DIR if not exists(cache_dir): return now = time() max_age = max_age_days * 86400 try: for filename in listdir(cache_dir): filepath = join(cache_dir, filename) if isfile(filepath): try: file_age = now - getmtime(filepath) if file_age > max_age: unlink(filepath) print("Removed old flag: %s" % filename) except Exception as e: print( "Error removing %s: %s" % (filename, str(e))) except Exception as e: print("Error cleaning flag cache: %s" % str(e)) def cleanup_old_temp_files(max_age_hours=1): """ Remove old temporary files in /tmp matching specific patterns. Files older than max_age_hours are deleted. """ try: now = time() max_age = max_age_hours * 3600 # seconds patterns = [ "/tmp/*vavoo*", "/tmp/*flag*", "/tmp/tmp*.png" ] # "/tmp/*vavoo*" also matches the proxy's own bookkeeping files # (vavoo_proxy.py's PID_FILE/BOOTING_FILE - hardcoded here rather # than imported to avoid a circular import, since vavoo_proxy.py # itself imports from this module). Deleting either out from # under a running/booting proxy would confuse its own PID/boot # tracking. excluded_basenames = ("vavoo_proxy.pid", "vavoo_proxy_booting") cleaned = 0 for pattern in patterns: for filepath in glob.glob(pattern): try: if basename(filepath) in excluded_basenames: continue if isfile(filepath): file_age = now - getmtime(filepath) if file_age > max_age: unlink(filepath) cleaned += 1 print( "Cleaned old temp file: %s" % filepath) except Exception as e: print( "Error removing %s: %s" % (filepath, str(e))) if cleaned > 0: print("Total cleaned old temp files: %d" % cleaned) return cleaned except Exception as e: print("Error cleaning temp files: %s" % str(e)) return 0 def preload_country_flags(country_list, cache_dir=FLAG_CACHE_DIR): """ Preload flags for a list of countries. Each chunk of countries is downloaded in a separate daemon thread. Compatible with Python 2 and 3. """ def download_flags_worker(countries): for country in countries: try: success, _ = download_flag_online(country, cache_dir) if success: print("Preloaded flag for: %s" % country) except Exception as e: print( "Error preloading flag for %s: %s" % (country, str(e))) # Split list into chunks to avoid overloading chunk_size = 10 threads = [] if not country_list: return threads total = len(country_list) for i in range(0, total, chunk_size): chunk = country_list[i:i + chunk_size] t = threading.Thread( target=download_flags_worker, args=(chunk,) ) t.setDaemon(True) t.start() threads.append(t) return threads # ==================== START EPG ==================== _epg_matcher = None _epg_matcher_lock = threading.Lock() def get_epg_matcher(similarity_threshold=None): global _epg_matcher # Se non viene passato un valore, usa quello dalla configurazione if similarity_threshold is None: # ConfigSelectionNumber.value comes back as a str (not int) on # at least some images/Python versions - confirmed via a # user-submitted log where this crashed every single EPG lookup # and every bouquet auto-update with "TypeError: unsupported # operand type(s) for /: 'str' and 'float'". float() handles # both str and numeric value types safely. similarity_threshold = float( config.plugins.vavoo.similarity_threshold.value) / 100.0 if _epg_matcher is None: with _epg_matcher_lock: if _epg_matcher is None: _epg_matcher = VavooEPGMatcher(similarity_threshold) return _epg_matcher # Aggiorna la soglia nel matcher esistente (per modifiche dinamiche) _epg_matcher.similarity_threshold = similarity_threshold return _epg_matcher def calculate_similarity(a, b): """ Calculate similarity ratio between two strings using SequenceMatcher. Returns a float between 0.0 and 1.0. """ return SequenceMatcher(None, a, b).ratio() def get_orbital_position(service_ref): """ Extract orbital position from service reference namespace. Returns orbital position in tenths of a degree (e.g., 130 = 13.0°E, -50 = 5.0°W) """ parts = service_ref.split(':') if len(parts) < 4: return 0 try: namespace_str = parts[3] if parts[3] else '0' namespace = int(namespace_str, 16) # Case 1: Default - Position * 65536 # The namespace is a multiple of 65536 (0x10000) if namespace % 0x10000 == 0: pos = namespace // 0x10000 # Determine East/West from sign (Enigma convention) if pos < 1800: # East return pos else: # West (pos > 1800, e.g., 3600-50=3550 for 5°W) return -(3600 - pos) # Case 2: Exception - also contains frequency and polarization # Extract the base part (Position * 65536) base = namespace & 0xFFFF0000 pos = base // 0x10000 if pos < 1800: return pos else: return -(3600 - pos) except (ValueError, TypeError): return 0 def get_configured_satellites(): """ Get list of satellites configured by user in Enigma2. Returns list of orbital positions in tenths of degree (e.g., [130, 192, 282]) """ try: configured_sats = [] # Method 1: Direct from NimManager if hasattr(nimmanager, 'getConfiguredSats'): sats = nimmanager.getConfiguredSats() if sats: configured_sats = list(sats) print( "[SatConfig] Found {} configured satellites via NimManager".format( len(configured_sats))) return configured_sats # Method 2: Parse from config for slot in nimmanager.nim_slots: if slot.isCompatible( "DVB-S") and slot.config.dvbs.configMode.value != "nothing": # Simple mode - check diseqc settings if slot.config.dvbs.configMode.value == "simple": for port in ['diseqcA', 'diseqcB', 'diseqcC', 'diseqcD']: orbpos = getattr(slot.config.dvbs, port).value if orbpos and orbpos not in configured_sats and orbpos < 3600: configured_sats.append(orbpos) print( "[SatConfig] Found configured sat: {} (port {})".format( orbpos, port)) # Advanced mode - check each configured satellite elif hasattr(slot.config.dvbs, 'advanced') and slot.config.dvbs.advanced: for sat_config in slot.config.dvbs.advanced.sat.values(): if sat_config.enabled.value: orbpos = sat_config.sat.value.orbital_position if orbpos and orbpos not in configured_sats: configured_sats.append(orbpos) print( "[SatConfig] Found configured sat: {}".format(orbpos)) print("[SatConfig] Total configured satellites: {}".format(configured_sats)) return configured_sats except Exception as e: print("[SatConfig] Error getting configured satellites: {}".format(e)) return [] def get_satellite_priority(orbpos, configured_sats): """ Returns priority boost for a satellite based on user configuration. 1.0 = same as configured 0.5 = other satellite """ if orbpos in configured_sats: return 1.0 return 0.5 # get_country_code() returns real ISO 3166-1 codes (needed for flag # downloads, e.g. "gb" for United Kingdom), but the Rytec database groups # channels by its own historical suffix convention, which doesn't always # match ISO - e.g. UK channels are "BBCOne.uk", not "BBCOne.gb". Without # this translation, every channel from a mismatched country would find # zero candidates in rytec_by_country and always report "ID not found". RYTEC_COUNTRY_CODE_OVERRIDES = { 'gb': 'uk', } # channel_alias.ALIAS_MAP is curated for Italian channels specifically # (see its own module docstring) - used by find_match()'s country-gate # below to always trust an alias hit when matching for Italy itself, # even for the handful of entries pinned to a non-".it" Rytec id. _ALIAS_MAP_HOME_COUNTRY = "it" class VavooEPGMatcher(object): def __init__(self, similarity_threshold=0.70): self.similarity_threshold = similarity_threshold # Guards the lazy-init blocks below (_configured_sats, # _checked_temp_cache) against a TOCTOU race if find_match() is # ever called concurrently on this same matcher instance from # two threads. self._lazy_init_lock = threading.Lock() # Guards self.cache/self.normalized_index/self.new_matches - # see find_match()'s docstring. RLock: _find_match_internal() # (called from inside find_match()'s locked section) does its # own lazy-init re-entrancy on this same thread. self._cache_lock = threading.RLock() # (clean_name, original_name, service_ref) self.rytec_entries = [] # Same entries grouped by country code (from the id's suffix, e.g. # "Rai1.it" -> "it"), so matching against a specific country # doesn't have to linearly scan every other country's entries too. self.rytec_by_country = {} self.rytec_by_id = {} # (original_id, service_ref) self.rytec_names = {} self.cache = load_cache() # persistent cache self.new_matches = {} # matches found in this session self.normalized_index = {} # map normalized_key -> original_key self._build_normalized_index() # build index at startup self.alias_map = {} self._load_alias_map() if not self.rytec_entries: self._load_rytec_database() @staticmethod def is_valid_rytec_id(id_val): """Returns True if id_val appears to be a valid Rytec ID (e.g. 'Rai1.it').""" # isinstance(id_val, str) alone rejects every value on Python 2, # where json.load()/XML parsing produce `unicode` (not a `str` # subclass there) - use text_type (six.text_type) too so this # doesn't silently reject every cached/parsed id on Python 2. if not id_val or not isinstance(id_val, (str, text_type)): return False if '.' not in id_val: return False parts = id_val.split('.') if len(parts) < 2: return False suffix = parts[-1] return len(suffix) in (2, 3) and suffix.isalpha() @staticmethod def _normalize_rytec_sref(raw_sref): """Apply the same "1:" -> "4097:" type conversion _find_match_internal() applies to a freshly-matched Rytec sref, plus trailing-colon normalization, so a raw self.rytec_by_id[...] value can be compared against a cached sref on equal terms.""" if not raw_sref: return None parts = raw_sref.split(':') if parts and parts[0] == '1': parts[0] = '4097' return ensure_sref_trailing_colon(':'.join(parts)) def _cleanup_stale_unmatched( self, channel_name, country_code, servicetype): """Remove this channel's entry from the unmatched cache if one exists there from before it started matching successfully. Only the live-rematch success path in find_match() used to do this (via save_unmatched(matched=True)); the alias/local-cache/ temp-cache fast-path hits never did, leaving "ghost" unmatched entries around indefinitely for channels that actually match fine now. Checks the same in-memory unmatched cache save_unmatched() itself stages against (lazily loaded once, kept for the process lifetime) - cheap on the common case (nothing to clean up), and always current with anything staged earlier in this same run, unlike a separate per-matcher-instance snapshot would be. """ key = "%s_%s" % (channel_name.strip(), country_code or '') with _unmatched_lock: _load_unmatched_cache_locked() found = key in _unmatched_cache_data if found: save_unmatched( channel_name, country_code, servicetype, matched=True) def _load_alias_map(self): if exists(ALIAS_FILE): try: with open(ALIAS_FILE, 'r') as f: self.alias_map = load(f) return except Exception: self.alias_map = {} pass def _load_rytec_database(self): rytec_paths = [ "/etc/epgimport/rytec.channels.xml", "/usr/lib/enigma2/python/Plugins/Extensions/EPGImport/rytec.channels.xml"] rytec_file = None for path in rytec_paths: if exists(path): rytec_file = path break if not rytec_file: print("[VavooEPGMatcher] Rytec database not found.") return try: with io.open(rytec_file, 'r', encoding='utf-8') as f: content = f.read() pattern = r'([^<]+)\s*(?:)?' matches = findall(pattern, content, IGNORECASE) for match in matches: original_id = match[0].strip() service_ref = match[1].strip() comment = match[2].strip() if len( match) > 2 and match[2] else None if comment: channel_name = comment else: channel_name = original_id.replace( '.it', '').replace( '.de', '').replace( '.fr', '') channel_name = channel_name.replace( '-', ' ').replace('_', ' ') # clean_name = self._clean_name(channel_name) clean_name = self._clean_name_for_similarity(channel_name) # Precomputed once here instead of on every _find_match_internal() # call (which re-tokenized every candidate's clean_name on # every single channel match, against what can be # thousands of entries per country - this list never # changes after load). entry_tokens = _tokenize_for_compat(clean_name) entry = ( clean_name, channel_name, original_id, service_ref, entry_tokens) self.rytec_entries.append(entry) entry_country = original_id.split( '.')[-1] if '.' in original_id else "" self.rytec_by_country.setdefault( entry_country, []).append(entry) self.rytec_names[original_id] = clean_name self.rytec_by_id[original_id] = service_ref print("[VavooEPGMatcher] Loaded {} Rytec channels".format( len(self.rytec_entries))) except Exception as e: print("[VavooEPGMatcher] Error loading database: {}".format(e)) def _clean_name_for_key(self, name): """Pulisce il nome per generare la chiave: mantiene suffissi .c, .s, .b.""" if not name: return "" cleaned = name.lower() # Rimuovi solo parentesi e indicatori di qualità (NON i suffissi) cleaned = sub(r'\s*\([^)]*\)\s*', '', cleaned) cleaned = sub(r'\b(4k|hd|sd|fhd|uhd|hq|hevc|h265|h264)\b', '', cleaned) # Mantieni i punti, sostituisci altri non alfanumerici con spazio cleaned = sub(r'[^\w\s\.]', ' ', cleaned) cleaned = sub(r'\s+', ' ', cleaned).strip() return cleaned def _clean_name_for_similarity(self, name): """Normalize a channel name for similarity comparison - see epg_name_utils.clean_name_for_similarity() (single source of truth, shared with generate_epg_channel_db.py's off-box mirror).""" return clean_name_for_similarity(name) def _normalize_key(self, channel_name, country_code): clean_name = self._clean_name_for_key(channel_name) return "{}_{}".format(clean_name, country_code) def _build_normalized_index(self): self.normalized_index = {} for key in self.cache: self._index_cache_key(key) def _index_cache_key(self, key): """Add/update a single cache key in normalized_index without rescanning the whole cache - see find_match(), which calls this once per matched channel and would otherwise turn a bulk export into an O(n^2) scan as the cache grows over time.""" if '_' in key: name_part, country_part = key.rsplit('_', 1) else: name_part, country_part = key, '' norm_key = self._normalize_key(name_part, country_part) self.normalized_index[norm_key] = key def _get_signal_priority(self, service_ref): """ Determine signal priority based on service reference type. Priority levels: 1 = Satellite (best) - Italian-satellite bonus is applied by the caller separately, via orbital position + a boost multiplier 2 = Terrestrial DVB-T 3 = Cable 4 = Other / IPTV or unknown """ parts = service_ref.split(':') if len(parts) < 7: return 4 # Unknown / other try: # sref format: type:flags:servicetype:sid:tsid:onid:namespace:... # - namespace (what encodes orbital position / terrestrial / # cable, checked below) is field 6, not field 3 (that's the # sid). namespace_str = parts[6] if parts[6] else '0' namespace = int(namespace_str, 16) # Known satellite namespaces. Enigma2 encodes orbital position # as (tenths_of_a_degree << 16) for east, or # ((3600 - tenths_of_a_degree) << 16) for west - verified # against the terrestrial/cable checks below, which use the # same top-16-bits convention (0xFFFF0000). satellite_namespaces = [ 0x820000, # 13.0°E HotBird (Italy) 0xC00000, # 19.2°E Astra 0xEB0000, # 23.5°E Astra 3 0x11A0000, # 28.2°E Astra 2 0xA00000, # 16.0°E Eutelsat 16A 0x5A0000, # 9.0°E Eutelsat 9B 0x460000, # 7.0°E Eutelsat 7E 0xDDE0000, # 5.0°W Eutelsat 5WA (Italy) 0xCE40000, # 30.0°W Hispasat 0x1A40000, # 42.0°E Türksat 0x300000, # 4.8°E Astra 4A / Sirius 0x13B0000, # 31.5°E Astra 5B 0x14A0000, # 33.0°E Eutelsat 33E 0xFF0000, # 25.5°E Es'hail / Arabsat 0x1C20000, # 45.0°E AzerSpace 0x1860000, # 39.0°E Hellas Sat 0x1680000, # 36.0°E Eutelsat 36B 0x1040000, # 26.0°E Badr 0x130000, # 1.9°E BulgariaSat ] # Check for satellite match. Mask must be 0xFFFF0000 (top 16 # bits, matching how the value above was built) - the # previous 0xFFF00000 zeroed out part of the orbital-position # field itself, so it could only ever match a namespace by # coincidence, not because it was actually that satellite. for sat_ns in satellite_namespaces: if namespace & 0xFFFF0000 == sat_ns: # Italian satellite bonus is applied by the caller # via priority + a boost multiplier, not by this # priority level - satellite is always priority 1. return 1 # Terrestrial DVB-T if namespace & 0xFFFF0000 == 0xEEEE0000: return 2 # Terrestrial # Cable if namespace & 0xFFFF0000 == 0xFFFF0000: return 3 # Cable # Default fallback return 4 # Other / IPTV / unknown except (ValueError, TypeError): return 4 def _find_match_internal( self, channel_name, country_code, channel_id=None, servicetype="4097"): """ Search for a match in ALL Rytec channels, then apply boost based on user configuration. """ if not channel_name: return None, None # Clean the input channel name clean_input = self._clean_name_for_similarity(channel_name) # Load user-configured satellites (e.g., [130] for 13°E) if not hasattr(self, '_configured_sats'): with self._lazy_init_lock: if not hasattr(self, '_configured_sats'): self._configured_sats = get_configured_satellites() print("[Match] User has {} configured satellites: {}".format( len(self._configured_sats), self._configured_sats)) candidates = [] # Restrict the scan to the requested country's entries up front # (pre-grouped in rytec_by_country at load time) instead of # walking every entry for every country and discarding most of # them - this is the hot loop of EPG matching, run once per # channel against what can be thousands of Rytec entries. if not country_code: entries_to_scan = self.rytec_entries elif country_code == "bk": balkan_codes = [ "ba", "hr", "rs", "si", "me", "mk", "al", "bg", "ro"] entries_to_scan = [ entry for code in balkan_codes for entry in self.rytec_by_country.get(code, []) ] elif RYTEC_COUNTRY_CODE_OVERRIDES.get(country_code): entries_to_scan = self.rytec_by_country.get( RYTEC_COUNTRY_CODE_OVERRIDES[country_code], []) else: entries_to_scan = self.rytec_by_country.get(country_code, []) # Pass 1: search all matches by similarity (ignore priority for now) # Reuse one SequenceMatcher with seq1 fixed to clean_input instead # of building a new one per candidate (set_seq2() lets it skip # re-analyzing seq1 every time), and check the cheap upper-bound # estimates (real_quick_ratio(), then quick_ratio()) before paying # for the full ratio() computation - both are always >= the real # ratio, so this only skips candidates that could never reach the # threshold anyway, never a false negative. This loop runs against # every Rytec entry for the country on every uncached match, so it # matters most for countries with large Rytec databases (e.g. Italy). source_tokens = _tokenize_for_compat(clean_input) sm = SequenceMatcher(None, clean_input) for clean_entry, orig_name, rytec_id, service_ref, entry_tokens in entries_to_scan: # A shared textual prefix with a *different* word at a # position both share is a strong signal these are # different channels (e.g. "canal motogp" vs "canal plus # foot") - raw character similarity alone can clear even a # fairly high threshold for such pairs since most of the # string still overlaps. A longer entry may still add # trailing descriptive words - only a genuine word-for-word # prefix mismatch is rejected. if source_tokens: if entry_tokens and not _tokens_compatible( source_tokens, entry_tokens): continue sm.set_seq2(clean_entry) if sm.real_quick_ratio() < self.similarity_threshold: continue if sm.quick_ratio() < self.similarity_threshold: continue score = sm.ratio() if score < self.similarity_threshold: continue # Extract additional info signal_priority = self._get_signal_priority(service_ref) orbpos = 0 if signal_priority == 1: orbpos = get_orbital_position(service_ref) # Calculate boost boost = 1.0 # 1. User-configured satellite → max boost if orbpos and self._configured_sats and orbpos in self._configured_sats: boost = 1.5 print( "[Match] FOUND! Satellite {} is user-configured!".format(orbpos)) # 2. Italian satellite (important) but not configured elif country_code == 'it' and orbpos in [130, -50]: # 13°E or 5°W boost = 1.3 # 3. Other satellites elif signal_priority == 1: boost = 1.2 # 4. Terrestrial elif signal_priority == 2: boost = 1.1 # 5. Cable/IPTV else: boost = 1.0 adjusted_score = score * boost candidates.append(( adjusted_score, score, signal_priority, orbpos, clean_entry, orig_name, rytec_id, service_ref )) # Sort primarily by raw textual score, using the satellite/signal # boost only as a tie-breaker among near-equal matches. Every # candidate here already cleared similarity_threshold, so using # `adjusted_score = score * boost` as the sort key let a marginal # boosted candidate (e.g. score 0.71 * 1.5 = 1.065) always beat a # near-perfect unboosted one (max possible 1.0 * 1.0) regardless of # how much better the textual match actually was. Bucketing raw # score into 0.05-wide bands means boost can only decide between # candidates that are already comparably good matches. _PRIORITY_TIE_BAND = 0.05 candidates.sort( key=lambda x: (round(x[1] / _PRIORITY_TIE_BAND), x[0]), reverse=True) for adj_score, orig_score, priority, orbpos, clean_entry, orig_name, rytec_id, service_ref in candidates: # Pick the first candidate above base similarity threshold if orig_score >= self.similarity_threshold: parts = service_ref.split(':') if parts and parts[0] == '1': # Conversion for Enigma2 service reference parts[0] = '4097' converted = ':'.join(parts) sat_info = " (sat {})".format(orbpos) if orbpos else "" conf_info = " [CONFIGURED]" if orbpos in self._configured_sats else "" print("[Match] CHOSEN: '{}' -> {}{}{} (score:{}→{}, priority:{})".format( channel_name, rytec_id, sat_info, conf_info, orig_score, adj_score, priority )) # Many Rytec entries (pan-European channels like # Eurosport, Nickelodeon, Discovery, DMAX...) genuinely # broadcast via one shared satellite transponder # serving multiple countries, so Rytec correctly has # separate per-country ids that all point at the exact # same real DVB tuple. Every country whose Vavoo # catalog matches to one of those would otherwise # collide on that identical tuple in Enigma2's EPG # cache (which keys purely on the tuple, not the # embedded stream URL), showing one country's # programme data for all of them. Give this Vavoo # channel its own unique reference instead - the # correct id above still goes into channels.xml, and # EPGImport's own parser already supports one id # fanning out to multiple different service refs. if channel_id: return rytec_id, unique_fallback_sref( servicetype, channel_id) return rytec_id, converted # No Rytec candidate matched. Before giving up, try this # country's own EPG programme feed directly - some countries # (confirmed for es/pl/tr and others) have far richer channel # coverage in their own epg_.xml than Rytec does; Rytec may # simply never have catalogued this channel at all. Reuses the # same feed-index/name-matching machinery write_epg_mapping_file() # already uses as a same-purpose fallback for the opposite case # (Rytec matched, but under an id the feed itself doesn't use). # The feed only gives an id, not a real DVB service ref (it's # programme-guide metadata, not satellite broadcast data), so a # synthesized, collision-free sref is generated the same way an # otherwise-unmatched channel already gets one. if channel_id and country_code: feed_ids, name_index = _get_epg_feed_index(country_code) if name_index: feed_match_id = _find_feed_id_by_name( channel_name, self, name_index) if feed_match_id: fallback_sref = unique_fallback_sref( servicetype, channel_id) print( "[Match] FEED-DIRECT MATCH: '{}' -> {} " "(own EPG feed, no Rytec entry)".format( channel_name, feed_match_id)) return feed_match_id, fallback_sref print("[Match] No match found for '{}'".format(channel_name)) return None, "4097:0:0:0:0:0:0:0:0:0:" def find_match( self, channel_name, country_code=None, servicetype="4097", channel_id=None): """Public entry point - serializes the whole lookup (including every self.cache/self.normalized_index/self.new_matches read and write below) behind self._cache_lock. This matcher is a per-process singleton (get_epg_matcher()) called concurrently from real, sanctioned usage - a bouquet export's background thread looping this per channel, while watching an already exported channel spawns a separate per-channel EPG-overlay daemon thread that also calls this - and those dicts were previously mutated with no locking at all: reassigning self.normalized_index mid-iteration elsewhere (_build_normalized_index(), also called from update_complete_cache() - see its matching lock use there) could raise "dictionary changed size during iteration" and abort EPG matching for a whole country's remaining channels, or silently drop a concurrent thread's cache write. An RLock (not a plain Lock) since _find_match_internal() below can itself re-enter lazy-init code paths on this same instance/thread.""" with self._cache_lock: return self._find_match_locked( channel_name, country_code, servicetype, channel_id) def _find_match_locked( self, channel_name, country_code=None, servicetype="4097", channel_id=None): if not channel_name: return None, None # 0. Apply alias normalization (if available) if alias_available: # Clean the channel name using the same rules as playlist_generator norm_name = channel_alias.normalize_channel_name(channel_name) if norm_name: # If we have an EPG ID for this canonical name, use it # directly. self.alias_map (loaded from the on-disk # channel_alias.json, if present) can override/add to the # curated channel_alias.ALIAS_MAP, which is the actual # source of the ~300 hand-mapped Italian channel aliases. alias_id = self.alias_map.get( norm_name) or channel_alias.ALIAS_MAP.get(norm_name) if alias_id: # ALIAS_MAP is curated for Italian channels # specifically (~300 hand-mapped entries, see # channel_alias.py's own header), but this lookup # had no country check - it fired identically for # every country whose channel list happened to # share a name with one of those entries (e.g. # generic pan-European brands like "Eurosport 1" or # "Nickelodeon"), so Poland's and Spain's own # exports both got Italy's Rytec sref and collided # in eEPGCache, since they became the same physical # DVB tuple. Only trust the hit when the alias id's # own country suffix agrees with the country # actually being matched (same convention # _load_rytec_database() uses to bucket entries) - # or no country was specified at all (e.g. the # in-player "now playing" overlay). alias_country = ( alias_id.rsplit('.', 1)[-1].lower() if '.' in alias_id else "") effective_country = RYTEC_COUNTRY_CODE_OVERRIDES.get( country_code, country_code) if country_code else None # ALIAS_MAP's home country (see its own docstring) - # always trust a hit when matching for Italy itself, # even if the curator pinned a specific entry to a # non-".it" Rytec id (e.g. "DISNEY CHANNEL"/"DISNEY # JUNIOR" -> a ".ch" id, because Italy's own Rytec # catalog has no native entry for them). The suffix # check below still guards every OTHER country # against reusing one of these curated Italian # picks for an unrelated channel of the same name. if (not effective_country or not alias_country or effective_country.lower() == _ALIAS_MAP_HOME_COUNTRY or alias_country == effective_country.lower()): alias_sref = self.rytec_by_id.get(alias_id) if alias_sref: if alias_sref.startswith('1:'): alias_sref = '4097' + alias_sref[1:] print( "[Match] ALIAS HIT: {} -> {}".format(norm_name, alias_id)) self._cleanup_stale_unmatched( channel_name, country_code, servicetype) # Rytec's raw tuple can be (correctly) # shared by several countries' entries for # the same pan-European channel brand - see # the synthetic-sref note on the Rytec # candidate match below. Same fix here: # give this Vavoo channel its own unique # reference instead of reusing the shared # one directly, when we have a channel_id # to key it off. if channel_id: return alias_id, unique_fallback_sref( servicetype, channel_id) return alias_id, alias_sref # Normalize the original name for key generation search_key = self._normalize_key(channel_name, country_code or "") # 1. Local cache via normalized index if search_key in self.normalized_index: cached_key = self.normalized_index[search_key] cached = self.cache[cached_key] id_val = cached.get('id') if self.is_valid_rytec_id(id_val): # Check if the channel name matches the Rytec name associated with that ID # rytec_clean = self.rytec_names.get(id_val, '') channel_clean = self._clean_name_for_similarity(channel_name) rytec_clean = self.rytec_names.get(id_val, '') if rytec_clean: channel_clean = self._clean_name_for_similarity( channel_name) current_rytec_sref = self._normalize_rytec_sref( self.rytec_by_id.get(id_val)) cached_sref = ensure_sref_trailing_colon( cached.get('sref')) if channel_clean == rytec_clean and ( not current_rytec_sref or cached_sref == current_rytec_sref): print( "[Match] Local cache HIT (valid ID & name match): {}".format(cached_key)) self._cleanup_stale_unmatched( channel_name, country_code, servicetype) return cached.get('id'), cached.get('sref') elif channel_clean != rytec_clean: print( "[Match] Cache ID '{}' name mismatch (channel='{}', rytec='{}'), will re-match".format( id_val, channel_clean, rytec_clean)) # Do not return, proceed with live matching else: # Name still matches, but the cached sref no # longer matches what Rytec has for this id - # leftover from a stale/mismatched match (see # find_match()'s live-rematch branch below). # Re-match instead of trusting a sref that # actually belongs to a different real channel. print( "[Match] Cache ID '{}' sref stale for '{}' (cached={}, rytec={}), will re-match".format( id_val, channel_name, cached_sref, current_rytec_sref)) else: # We don't have the Rytec name, accept the cache print( "[Match] Local cache HIT (valid ID, no Rytec name): {}".format(cached_key)) self._cleanup_stale_unmatched( channel_name, country_code, servicetype) return cached.get('id'), cached.get('sref') else: # Invalid ID, proceed with live matching print("[Match] Local cache has invalid ID, will try to re-match.") # 2. Online cache if not hasattr(self, '_checked_temp_cache'): with self._lazy_init_lock: if not hasattr(self, '_checked_temp_cache'): self._checked_temp_cache = False self._temp_cache = None if not self._checked_temp_cache: with self._lazy_init_lock: # Re-check inside the lock - another thread may have # already done this download/load while we were waiting. if not self._checked_temp_cache: print("[Match] Checking temp cache once...") self._temp_cache = load_temp_cache() if not self._temp_cache: print("[Match] Temp cache not found, downloading once...") if download_epg_cache_if_needed(): self._temp_cache = load_temp_cache() self._checked_temp_cache = True if self._temp_cache and search_key in self._temp_cache: cached = self._temp_cache[search_key] # Unlike the local cache above, this comes from a pre-built # file downloaded from GitHub - validate the ID before # trusting it, same as the local cache path does, instead of # blindly returning whatever is there. if self.is_valid_rytec_id(cached.get('id')): print("[Match] Temp cache HIT: {}".format(search_key)) new_entry = cached.copy() new_entry['name'] = channel_name # original name self.cache[search_key] = new_entry self._index_cache_key(search_key) # Deferred to save_cache() (called once per batch by callers, # e.g. after a whole bouquet export) instead of writing the # full cache file here - this branch fires per matched # channel, and a full rewrite per channel turns a bulk export # into many redundant whole-file writes as the cache grows. self.new_matches[search_key] = new_entry self._cleanup_stale_unmatched( channel_name, country_code, servicetype) return cached.get('id'), cached.get('sref') else: print( "[Match] Temp cache has invalid ID for {}, ignoring".format(search_key)) # 2.5 Community-curated channel database (see # generate_epg_channel_db.py): a pre-solved name -> this # country's own EPG feed id mapping, built and reviewed offline # instead of guessed live by fuzzy-matching against Rytec. # Country-scoped by construction (one file per country), so no # collision risk from unrelated countries sharing a name. if country_code: curated_db = _load_curated_channel_db(country_code) if curated_db: curated_key = self._clean_name_for_similarity(channel_name) curated_id = curated_db.get(curated_key) if curated_id and channel_id: print( "[Match] CURATED DB HIT: {} -> {}".format( channel_name, curated_id)) self._cleanup_stale_unmatched( channel_name, country_code, servicetype) return curated_id, unique_fallback_sref( servicetype, channel_id) # 3. Live matching print("[Match] Doing local matching for: {}".format(channel_name)) if search_key in self.new_matches: m = self.new_matches[search_key] return m['id'], m['sref'] result_id, result_sref = self._find_match_internal( channel_name, country_code, channel_id=channel_id, servicetype=servicetype) # Is there already an entry in the cache (even with invalid ID)? # Resolve via normalized_index first: the raw key an entry is # stored under can differ from search_key (e.g. name-cleaning # rules changed since it was written, same drift as the # _clean_name_for_key '&' case). Step 1 above already proves such # an entry is reachable via normalized_index - looking it up with # self.cache.get(search_key) directly would miss it and silently # discard a valid existing match every time this branch runs. existing_key = self.normalized_index.get(search_key, search_key) existing_entry = self.cache.get(existing_key) if result_id and result_sref: # Decide which sref to keep final_sref = result_sref if existing_entry and existing_entry.get('id') == result_id: # Only reuse the previously cached sref when it's for the # SAME matched id - this keeps re-matches of an unchanged # channel stable. When the freshly matched id differs # from what's cached, the old sref belongs to whatever # channel the cache used to (possibly wrongly) point at; # blindly keeping it here is exactly how an id gets # updated to the right channel while its sref silently # keeps pointing at a different one. existing_sref = existing_entry.get('sref') current_rytec_sref = self._normalize_rytec_sref( self.rytec_by_id.get(result_id)) # If the existing one has a valid sref (not fallback) that # still agrees with what Rytec currently has for this id, # preserve it - this is a no-op in the common case # (result_sref should already match) and only matters for # candidate-selection ties. If it disagrees, it's stale # (e.g. left over from before this id was matched # correctly) - fall through to the freshly matched # result_sref instead of re-entrenching it. if (existing_sref and existing_sref != "4097:0:0:0:0:0:0:0:0:0:" and (not current_rytec_sref or ensure_sref_trailing_colon(existing_sref) == current_rytec_sref)): final_sref = existing_sref print( "[Match] Preserving existing sref: {}".format(existing_sref)) # self.new_matches[search_key] = {'id': result_id, 'sref': final_sref} self.new_matches[search_key] = { 'id': result_id, 'sref': final_sref, 'name': channel_name } save_unmatched( channel_name, country_code, servicetype, matched=True) return result_id, final_sref else: # If no live match, but there is an existing entry with a valid # sref, use it (with matched=False) if existing_entry: # Compare against the normalized form: some legacy entries # were written without the trailing colon, which made this # equality check silently treat an actual fallback sref as # "valid" and skip the downgrade below. existing_sref = existing_entry.get('sref') normalized_sref = ensure_sref_trailing_colon(existing_sref) if normalized_sref and normalized_sref != "4097:0:0:0:0:0:0:0:0:0:": print("[Match] No live match, using existing sref (invalid ID)") # Update cache with matched=False if needed if existing_entry.get('matched', True) is not False: existing_entry['matched'] = False self.cache[existing_key] = existing_entry self._index_cache_key(existing_key) # Not deferred to new_matches/update_complete_cache(): # that path always writes matched=True, which would # be wrong here. This branch is rare (only when a # previously-matched channel stops live-matching), # so an eager write is an acceptable tradeoff. save_cache(self.cache) return existing_entry.get('id'), existing_sref elif existing_entry.get('matched') is not False: # Stale entry with an invalid id AND no usable sref - # a leftover from before this channel ever properly # matched. Live re-matching just failed too, so # there's nothing worth preserving; without this, the # cache keeps claiming matched=True with no real EPG # data behind it, forever, since nothing else ever # revisits an entry once it's "matched". existing_entry['matched'] = False self.cache[existing_key] = existing_entry self._index_cache_key(existing_key) save_cache(self.cache) # Otherwise, no match and no valid sref save_unmatched( channel_name, country_code, servicetype, matched=False) return None, None # ==================== EPG CACHE FUNCTIONS ==================== # Safety-net cap: the matched-channel cache is meant to persist # indefinitely (unlike vavoo_proxy.py's short-TTL resolve_cache), so # this is deliberately generous - real usage (every unique channel # name across every country a user has ever exported) tops out well # below this. It only guards against unbounded/pathological growth, # trimming the oldest entries by their own 'timestamp' field. MAX_CACHE_ENTRIES = 15000 def _prune_cache_if_needed(cache, max_entries=MAX_CACHE_ENTRIES): """Drop the oldest entries (by 'timestamp') if cache exceeds max_entries.""" if len(cache) <= max_entries: return cache try: ordered = sorted( cache.items(), key=lambda kv: kv[1].get('timestamp', ''), ) to_drop = len(cache) - max_entries for key, _ in ordered[:to_drop]: del cache[key] print( "[Cache] Pruned {} oldest entries (cap: {})".format( to_drop, max_entries)) except Exception as e: print("[Cache] Error pruning cache: {}".format(e)) return cache def load_temp_cache(): """Load EPG cache from /tmp/vavoo_epg_cache.json""" temp_file = "/tmp/vavoo_epg_cache.json" try: if exists(temp_file): with open(temp_file, 'r') as f: return load(f) except Exception as e: print("[Cache] Error loading {}: {}".format(temp_file, e)) return None def load_cache(): try: with open(CACHE_FILE, 'r') as f: data = load(f, object_pairs_hook=OrderedDict) # Converti tutte le chiavi in minuscolo per compatibilità return OrderedDict((k.lower(), v) for k, v in data.items()) except BaseException: return OrderedDict() def save_cache(cache): """Save cache to file with complete format validation""" try: required_fields = [ 'id', 'name', 'country', 'sref', 'timestamp', 'matched'] for key, value in cache.items(): missing = [f for f in required_fields if f not in value] if missing: print( "[Cache] ERROR: Entry {} missing fields: {}".format( key, missing)) return False cache = _prune_cache_if_needed(cache) temp_file = CACHE_FILE + ".tmp" with open(temp_file, 'w') as f: dump(cache, f, indent=2, sort_keys=True) rename(temp_file, CACHE_FILE) print("[Cache] Saved {} entries".format(len(cache))) return True except Exception as e: print("[Cache] Error saving cache: {}".format(e)) return False def clean_cache_and_unmatched(): """ Move to the unmatched cache all entries from the main cache that have invalid IDs or that do not match the channel name. Also fixes the formatting of sref. """ # Load main cache if not exists(CACHE_FILE): return with open(CACHE_FILE, 'r') as f: main_cache = load(f) # Load existing unmatched cache unmatched = {} if exists(UNMATCHED_FILE): with open(UNMATCHED_FILE, 'r') as f: unmatched = load(f) # Get matcher to access Rytec names matcher = get_epg_matcher() new_main = {} moved = 0 for key, value in main_cache.items(): # Fix sref sref = value.get('sref', '') if sref and not sref.endswith(':'): value['sref'] = sref + ':' id_val = value.get('id') if not matcher.is_valid_rytec_id(id_val): # Move to unmatched unmatched[key] = value moved += 1 continue # Check if the ID matches the channel name # Extract name from Rytec comment if possible, otherwise use the ID rytec_name = matcher.rytec_names.get(id_val, '') if not rytec_name: # Try to derive from ID: remove country suffix and replace dots # with spaces rytec_name = id_val.split('.')[0].replace('.', ' ') # rsplit on the last '_' (not split()[0] on the first) to match # how these "name_country" keys are actually built elsewhere # (_index_cache_key(), save_cache()) - a name containing a # literal underscore would otherwise get truncated here. clean_rytec_name = matcher._clean_name_for_similarity(rytec_name) clean_channel_name = matcher._clean_name_for_similarity( key.rsplit('_', 1)[0]) if clean_rytec_name != clean_channel_name: # Move to unmatched to be re-matched unmatched[key] = value moved += 1 else: new_main[key] = value # Save caches cache_temp = CACHE_FILE + ".tmp" with open(cache_temp, 'w') as f: dump(new_main, f, indent=2) rename(cache_temp, CACHE_FILE) unmatched_temp = UNMATCHED_FILE + ".tmp" with open(unmatched_temp, 'w') as f: dump(unmatched, f, indent=2) rename(unmatched_temp, UNMATCHED_FILE) with matcher._cache_lock: matcher.cache = new_main matcher._build_normalized_index() print("[Cache] Cleanup completed: moved {} entries to unmatched".format(moved)) return moved def cleanup_cache_matched_flag(): """Fix the matched flag for entries with invalid id.""" if not exists(CACHE_FILE): return with open(CACHE_FILE, 'r') as f: cache = load(f) changed = False for key, value in cache.items(): if not VavooEPGMatcher.is_valid_rytec_id(value.get('id')): if value.get('matched', False): value['matched'] = False changed = True if changed: temp_file = CACHE_FILE + ".tmp" with open(temp_file, 'w') as f: dump(cache, f, indent=2) rename(temp_file, CACHE_FILE) print("[Cache] Cleaned matched flags for invalid IDs.") def download_epg_cache_if_needed(): """Download vavoo_epg_cache.json to /tmp/ if not exists""" temp_file = "/tmp/vavoo_epg_cache.json" # If already exists, don't download if exists(temp_file): return True try: import requests url = "{}/vavoo_epg_cache.json".format(HOST_MAIN) print("[Cache] Downloading to /tmp...") response = requests.get(url, timeout=10) if response.status_code == 200: download_tmp = temp_file + ".tmp" with open(download_tmp, 'wb') as f: f.write(response.content) rename(download_tmp, temp_file) print("[Cache] Downloaded to: {}".format(temp_file)) return True except Exception as e: print("[Cache] Download error: {}".format(e)) return False _curated_channel_db_cache = {} _CURATED_CHANNEL_DB_TTL = 86400 def _load_curated_channel_db(country_code): """Fetch/cache this country's community-curated Vavoo channel name -> EPG feed channel id mapping (see generate_epg_channel_db.py), used as a fast, pre-solved alternative to live Rytec fuzzy-matching for names Rytec doesn't cover well. Keys are normalized the same way _clean_name_for_similarity() normalizes names for comparison elsewhere in this file - the generator script uses the identical normalization when building the file. Never raises - returns {} on any failure so callers just fall through to the existing matching chain, same convention as _get_epg_feed_index(). """ if not country_code: return {} cached = _curated_channel_db_cache.get(country_code) if cached and (time() - cached[0] < _CURATED_CHANNEL_DB_TTL): return cached[1] db = {} try: url = "{}/epg-channel-db/vavoo_channels_{}.json".format( HOST_MAIN, country_code.lower()) data = getUrl(url, timeout=15, retries=1) if data: parsed = loads(ensure_str(data)) if isinstance(parsed, dict): db = parsed except Exception as e: debug( "Could not fetch curated channel db for {}: {}".format( country_code, e)) if db: # Only cache successful, non-empty results. An empty result # (missing file, transient error) isn't cached at all, so the # next export retries instead of being stuck on a stale miss # for up to _CURATED_CHANNEL_DB_TTL - unlike # _get_epg_feed_index(), this is called once per country per # export, not per channel, so the retry cost is negligible. _curated_channel_db_cache[country_code] = (time(), db) return db def update_complete_cache( matched_channels, unmatched_channels, country_code, servicetype="4097"): """Update the complete cache with matched channels only; unmatched go to unmatched.json. This is the sole writer of CACHE_FILE for a bouquet export - matched_channels (built by create_bouquet_file()/ process_epg_matching_background()) is already a full superset of VavooEPGMatcher.new_matches (every entry added there came from a find_match() call that also produced a matched_channels entry), so a separate matcher.save_cache() call right before this one would just re-read, re-write, and re-index the exact same file a second time for no additional coverage. Clearing matcher.new_matches here takes over save_cache()'s other job - without it, that dict (a per-process singleton's, so never otherwise reset) would grow unbounded across every export for the life of the process. """ try: matcher = get_epg_matcher() complete_cache = {} # Load existing cache if exists(CACHE_FILE): try: with open(CACHE_FILE, 'r') as f: complete_cache = load(f) print( "[Cache] Loaded %d existing entries" % len(complete_cache)) except Exception as e: print("[Cache] Error loading cache: %s" % e) complete_cache = {} # Add matched channels (only these go to main cache) for m in matched_channels: key = matcher._normalize_key(m['name'], country_code) complete_cache[key] = { 'id': m['rytec_id'], 'sref': ensure_sref_trailing_colon(m['dvb_ref']), 'name': m['name'], 'country': country_code, 'matched': True, 'timestamp': strftime('%Y-%m-%d %H:%M:%S', localtime()) } print("[Cache] Added matched: %s -> %s" % (key, m['rytec_id'])) # Save main cache complete_cache = _prune_cache_if_needed(complete_cache) temp_file = CACHE_FILE + ".tmp" with open(temp_file, 'w') as f: dump(complete_cache, f, indent=4, sort_keys=True) rename(temp_file, CACHE_FILE) # Update matcher with new cache - under the same lock find_match() # holds around its own cache/normalized_index/new_matches use, # since this wholesale swap+clear can otherwise race with a # concurrent find_match() call on the same singleton (e.g. the # Now Playing EPG overlay) reading/writing those same attributes # mid-export. with matcher._cache_lock: matcher.cache = complete_cache matcher._build_normalized_index() matcher.new_matches.clear() # Process unmatched channels: save them to unmatched.json with their # original sref for u in unmatched_channels: # u should contain 'name' and optionally 'original_sref' sref = u.get('original_sref') # if available save_unmatched( u['name'], country_code, servicetype, matched=False, sref=sref) print( "[Cache] Updated main cache with %d entries" % len(complete_cache)) except Exception as e: print("[Cache] Error updating complete cache: %s" % e) trace_error() def save_unmatched( channel_name, country_code, servicetype="4097", matched=False, sref=None): """Stage an unmatched-channel update in memory. Call flush_unmatched_cache() once per batch (e.g. once per bouquet export) to actually write it - see that function's docstring for why. If sref is provided, it will be used as the service reference; otherwise a fallback is built from servicetype. """ global _unmatched_cache_dirty with _unmatched_lock: try: _load_unmatched_cache_locked() key = "%s_%s" % (channel_name.strip(), country_code or '') if matched and key in _unmatched_cache_data: # Remove if now matched del _unmatched_cache_data[key] _unmatched_cache_dirty = True print("[Unmatched] Removed matched channel: %s" % key) elif not matched: # Add or update unmatched timestamp = strftime('%Y-%m-%d %H:%M:%S', localtime()) # Use provided sref or build fallback if sref is not None: fallback_sref = ensure_sref_trailing_colon(sref) else: fallback_sref = "%s:0:0:0:0:0:0:0:0:0:" % servicetype old_data = _unmatched_cache_data.get(key, {}) attempts = old_data.get('attempts', 0) + 1 _unmatched_cache_data[key] = { 'id': key, 'name': channel_name.strip(), 'country': country_code or '', 'sref': fallback_sref, 'timestamp': timestamp, 'matched': False, 'attempts': attempts } _unmatched_cache_dirty = True print( "[Unmatched] Staged: %s (attempt #%d)" % (key, attempts)) except Exception as e: print("[Unmatched] Error: %s" % e) # In-memory mirror of UNMATCHED_FILE, lazily loaded once and kept for the # life of the process instead of every save_unmatched() call doing its # own full read+parse+write+rename - that was previously called once per # channel processed (both matched and unmatched) during every bouquet # export, from find_match() itself and again from callers looping over # their own matched/unmatched lists, making a multi-hundred-channel # export do a multi-hundred-entry full file rewrite that many times. # All access must go through _unmatched_lock. _unmatched_cache_data = None _unmatched_cache_dirty = False def _load_unmatched_cache_locked(): """Populate _unmatched_cache_data from disk if not already loaded. Caller must already hold _unmatched_lock.""" global _unmatched_cache_data if _unmatched_cache_data is not None: return data = {} if exists(UNMATCHED_FILE): try: with open(UNMATCHED_FILE, 'r') as f: content = f.read().strip() if content: data = loads(content) # Convert old format if needed for key, value in list(data.items()): if 'matched' not in value: data[key] = { 'id': value.get('id', key), 'name': value.get( 'name', key.split('_')[0] if '_' in key else key), 'country': value.get('country', ''), 'sref': value.get( 'sref', "4097:0:0:0:0:0:0:0:0:0:"), 'timestamp': value.get( 'timestamp', strftime( '%Y-%m-%d %H:%M:%S', localtime())), 'matched': False, 'attempts': 1} print( "[Unmatched] Converted old format: %s" % key) except Exception as read_error: print( "[Unmatched] Corrupted file, starting fresh: %s" % read_error) data = {} _unmatched_cache_data = data def invalidate_unmatched_cache(): """Force the next save_unmatched()/flush_unmatched_cache() call to reload from disk. Call this after anything writes UNMATCHED_FILE directly instead of through save_unmatched() (e.g. a cache-repair routine), so a later flush doesn't overwrite that write with a stale in-memory copy.""" global _unmatched_cache_data, _unmatched_cache_dirty with _unmatched_lock: _unmatched_cache_data = None _unmatched_cache_dirty = False def flush_unmatched_cache(): """Write pending save_unmatched() updates to UNMATCHED_FILE, if any. Call once per batch (e.g. at the end of a bouquet export, alongside update_complete_cache()) - not once per channel.""" global _unmatched_cache_dirty with _unmatched_lock: if not _unmatched_cache_dirty or _unmatched_cache_data is None: return try: temp_file = UNMATCHED_FILE + ".tmp" with open(temp_file, 'w') as f: dump(_unmatched_cache_data, f, indent=4, sort_keys=True) rename(temp_file, UNMATCHED_FILE) _unmatched_cache_dirty = False print( "[Unmatched] Cache flushed - total entries: %d" % len(_unmatched_cache_data)) except Exception as e: print("[Unmatched] Error flushing cache: %s" % e) # country_code -> (timestamp, set(ids), {clean_name: id}) _epg_feed_index_cache = {} _EPG_FEED_INDEX_TTL = 300 # getUrl() swallows the HTTP status code on any failure (always returns # "" whether it was a genuine 404 or a transient timeout/connection # error), so a permanent "this country has no feed" can't be reliably # told apart from a one-off blip here. Cache a failure/empty result # much more briefly than a real success - still long enough to avoid # re-fetching per unmatched channel within one export (which can be # hundreds of calls in a few seconds), but short enough that a # transient failure doesn't keep the feed-fallback disabled for the # rest of an in-progress export or an immediate retry. _EPG_FEED_INDEX_NEGATIVE_TTL = 60 def _get_epg_feed_index(country_code): """ Fetch and index this country's actual EPG feed: the set of values it defines, plus a cleaned-display-name -> id lookup. Used by write_epg_mapping_file() as a fallback for channels whose Rytec-matched id isn't one the feed itself uses - the same class of mismatch plugin.py's get_current_epg() already works around for the in-plugin overlay, but that fix doesn't cover this separate, epgimport-facing pipeline (exported bouquets), which is what this is for. Never raises - returns (None, None) on any failure so callers just skip the fallback and keep existing (correct-for-most-channels) behavior. """ if not country_code: return None, None cached = _epg_feed_index_cache.get(country_code) if cached: cached_time, cached_ids, cached_index = cached ttl = (_EPG_FEED_INDEX_TTL if cached_ids is not None else _EPG_FEED_INDEX_NEGATIVE_TTL) if time() - cached_time < ttl: return cached_ids, cached_index try: url = "http://{}:{}/epg/{}.xml".format( PROXY_HOST, PORT, country_code.lower()) # These feeds run several MB to 10+ MB (full multi-day guide for # every channel in the country) - the default 10s/3-retries # getUrl() timeout budget (tuned for small JSON/HTML responses) # routinely isn't enough to pull one down over a set-top box's # real-world connection, which silently disabled this whole # id-correction path and left most channels stuck on their # (usually wrong, see below) Rytec-matched id. xml_data = getUrl(url, timeout=30, retries=2) if not xml_data: warning( "EPG feed fetch returned empty for {} - id-correction " "fallback disabled for this export, channels will keep " "their Rytec-matched id even if the feed uses a " "different one".format(country_code)) # Cache the failure too (empty result, same TTL) - without # this, a country with no feed at all (most of them - only # a minority of country codes have one) would re-attempt this # same failing network fetch for every single unmatched # channel once _find_match_internal() starts consulting this # index as a fallback, instead of once per TTL window. _epg_feed_index_cache[country_code] = (time(), None, None) return None, None xml_source = io.BytesIO( xml_data if isinstance(xml_data, binary_type) else xml_data.encode('utf-8')) matcher = get_epg_matcher() feed_ids = set() name_index = {} # Stream-parse instead of ET.fromstring(): these feeds are almost # entirely entries (tens of thousands of them) with # the actual definitions front-loaded before any of # them, per the XMLTV convention. Building a full DOM of the # whole guide just to read the channel list is both slow and # memory-heavy on box-class hardware - stop as soon as the # section starts instead. for event, elem in ET.iterparse(xml_source, events=('start', 'end')): if event == 'end' and elem.tag == 'channel': chan_id = elem.get('id') if chan_id: feed_ids.add(chan_id) for dn in elem.findall('display-name'): dn_text = (dn.text or '').strip() if not dn_text: continue clean_dn = matcher._clean_name_for_similarity(dn_text) if clean_dn: # Tokens precomputed once here rather than by # every _find_feed_id_by_name() call that # scans this same index (once per channel # whose Rytec id isn't one this feed uses, # which - see write_epg_mapping_file() - can # be a large fraction of a country's channels). name_index[clean_dn] = ( chan_id, _tokenize_for_compat(clean_dn)) elem.clear() elif event == 'start' and elem.tag == 'programme': break except Exception as e: warning("Could not fetch/parse EPG feed for {}: {}".format(country_code, e)) # Same reasoning as the empty-response branch above: cache the # failure so repeated calls for this country within the TTL # window short-circuit instead of retrying the network fetch. _epg_feed_index_cache[country_code] = (time(), None, None) return None, None _epg_feed_index_cache[country_code] = (time(), feed_ids, name_index) return feed_ids, name_index # _tokenize_for_compat/_token_pair_compatible/_tokens_compatible are # thin aliases onto epg_name_utils - single source of truth, shared # with generate_epg_channel_db.py's off-box mirror. Kept as # module-level names here since they're called that way throughout # this file. _tokenize_for_compat = tokenize_for_compat _token_pair_compatible = token_pair_compatible _tokens_compatible = tokens_compatible def _find_feed_id_by_name(channel_name, matcher, name_index): """Fuzzy-match channel_name against an EPG feed's display-name index.""" if not name_index: return None clean = matcher._clean_name_for_similarity(channel_name) if not clean: return None source_tokens = _tokenize_for_compat(clean) best_score = 0.0 best_id = None threshold = matcher.similarity_threshold sm = SequenceMatcher(None, clean) # Only the threshold-or-above outcome is ever used by callers (a # below-threshold best_score is never returned), so - same as the # equivalent Rytec-scanning loop in _find_match_internal() - a # candidate whose cheap upper-bound estimate already can't reach the # threshold can be skipped before paying for the full ratio() # computation. Feed indexes run into the hundreds/thousands of # entries (see this function's callers), scanned once per exported # channel whose Rytec id isn't in this feed's own id set, so this # matters for country-wide export time. for dn_key, (dn_id, candidate_tokens) in name_index.items(): if (source_tokens and candidate_tokens and not _tokens_compatible(source_tokens, candidate_tokens)): continue sm.set_seq2(dn_key) if sm.real_quick_ratio() < threshold: continue if sm.quick_ratio() < threshold: continue score = sm.ratio() if score > best_score: best_score = score best_id = dn_id if best_id and best_score >= threshold: return best_id return None def write_epg_mapping_file(epg_entries, country_code): """ Write the EPG mapping file for a specific country. epg_entries: list of tuples (rytec_id, full_service_ref, channel_name), where full_service_ref is the DVB-style tuple *with the stream URL appended* (e.g. "4097:0:1:...:0:0:0:http%3a//host/stream") - not the bare tuple. EPGImport's own channelFilter() only fast-accepts a channel ref if it contains an embedded URL; a bare tuple falls through to a fake-recording probe that fails silently for it, so passing the bare tuple here means the channel gets dropped entirely during EPGImport's own channels.xml parse pass. """ epg_dir = "/etc/epgimport" if not exists(epg_dir): makedirs(epg_dir) if country_code: filename = "vavoo_{}.channels.xml".format(country_code.lower()) else: filename = "vavoo.channels.xml" channels_file = join(epg_dir, filename) # Some channels' Rytec-matched id isn't one this country's actual # feed uses for (it has its own convention for a # handful of channels, e.g. "RTP.1.HD.pt" vs the Rytec # "rtp1.pt") - epgimport can't find programme data filed under # an id the feed never defines. Fall back to the feed's own id, # found by display-name match, but only when the Rytec id # genuinely isn't one the feed already has - this never touches # a channel that's already matching correctly. # # Deliberately outside _epg_lock below: this can fetch and parse a # whole country's multi-MB XMLTV feed (up to ~60s across getUrl()'s # own retries), only refreshed every 300s per _get_epg_feed_index()'s # own cache. Holding a single global lock across that would make # every other country's (unrelated) channels.xml write queue up # behind whichever one's feed happens to be slow/uncached - the lock # only needs to protect the actual file write below. feed_ids, feed_name_index = _get_epg_feed_index(country_code) matcher = get_epg_matcher() if feed_ids else None # Use the full ref as key to avoid duplicates, but also store the # channel name. Unlike a bare dvb_ref, a full_service_ref # (tuple + URL) never needs a trailing-colon fixup - it already # ends with the stream URL, and blindly appending ':' here would # corrupt that URL instead of "completing" a bare tuple. unique = {} for epg_id, dvb_ref, ch_name in epg_entries: # isinstance(dvb_ref, (str, text_type)): plain `str` alone # would drop every `unicode` dvb_ref on Python 2, silently # writing an empty/near-empty EPG mapping file. if dvb_ref and isinstance( dvb_ref, (str, text_type)) and dvb_ref.strip(): if feed_ids and epg_id not in feed_ids: fallback_id = _find_feed_id_by_name( ch_name, matcher, feed_name_index) if fallback_id: debug( "EPG mapping fallback: '{}' rytec id '{}' not " "in feed, using '{}' instead".format( ch_name, epg_id, fallback_id)) epg_id = fallback_id unique[dvb_ref] = (epg_id, ch_name) if not unique: print("[EPG] No entries to write, skipping.") return None xml_lines = ['', ''] for dvb_ref, (epg_id, ch_name) in unique.items(): # Add comment with channel name (optional but useful for # readability) xml_lines.append( ' {}'.format(epg_id, dvb_ref, ch_name)) xml_lines.append('') with _epg_lock: try: temp_path = channels_file + ".tmp" with open(temp_path, 'w') as f: # 'encoding' arg removed for Py2 compatibility f.write('\n'.join(xml_lines)) rename(temp_path, channels_file) print( "[EPG] Written {} entries to {}".format( len(unique), filename)) return filename except Exception as e: print("[EPG] Error writing {}: {}".format(filename, e)) return None def update_epg_sources(): """ Scan /etc/epgimport for vavoo_*.channels.xml files and generate a master vavoo.sources.xml containing a source for each. """ epg_dir = "/etc/epgimport" sources_file = join(epg_dir, "vavoo.sources.xml") pattern = join(epg_dir, "vavoo_*.channels.xml") files = glob.glob(pattern) if not files: with _epg_lock: if exists(sources_file): remove(sources_file) print("[EPG] Removed sources file (no channels).") return sources_list = [] for f in sorted(files): basename_file = basename(f) # Extract country code from filename (example: vavoo_it.channels.xml -> # it) parts = basename_file.replace(".channels.xml", "").split("_") if len(parts) > 1: country_code = parts[1].lower() else: country_code = "unknown" # Build the source entry WITHOUT XML header source_entry = ''' Vavoo {} http://{}:{}/epg/{}.xml '''.format( basename_file, country_code.upper(), PROXY_HOST, PORT, country_code ) sources_list.append(source_entry) # Build the complete XML with a SINGLE header sources_xml = ''' {} '''.format("\n".join(sources_list)) try: with _epg_lock: temp_path = sources_file + ".tmp" with open(temp_path, "w") as f: f.write(sources_xml) rename(temp_path, sources_file) print( "[EPG] Sources file updated with %d entries." % len(sources_list)) except Exception as e: print("[EPG] Error writing sources file: %s" % e) ensure_vavoo_epg_sources_enabled() def ensure_vavoo_epg_sources_enabled(): """Auto-enable every currently-exported Vavoo EPG source ("Vavoo ", one per exported country - see the built above) in EPGImport's own settings, using its proper save API (EPGConfig.storeUserSettings()) instead of hand-editing epgimport.conf - that file is a pickle blob, not plain text, and an earlier attempt at this by appending a raw line to it corrupted EPGImport's own config. Merges into whatever the user already has enabled there; never removes anything. Only runs when the user has opted in via the "EPG Auto Update" config toggle - otherwise EPGImport's own source list is left exactly as the user configured it, same as the existing manual "Start EPG update now?" flow (which still requires the source to already be enabled). Never raises - this is a best-effort convenience a failure here shouldn't be allowed to break bouquet export over. """ try: if not (config.plugins.vavoo.epg_enabled.value and config.plugins.vavoo.epg_auto_update.value): return epg_dir = "/etc/epgimport" pattern = join(epg_dir, "vavoo_*.channels.xml") files = glob.glob(pattern) if not files: return wanted = set() for f in files: basename_file = basename(f) parts = basename_file.replace(".channels.xml", "").split("_") country_code = parts[1].upper() if len(parts) > 1 else "UNKNOWN" wanted.add("Vavoo {}".format(country_code)) from Plugins.Extensions.EPGImport import EPGConfig settings = EPGConfig.loadUserSettings() current = list(settings.get("sources") or []) missing = sorted(wanted - set(current)) if not missing: return EPGConfig.storeUserSettings(sources=current + missing) print( "[EPG] Auto-enabled EPGImport source(s): %s" % ", ".join(missing)) except ImportError: # EPGImport not installed - nothing to enable. pass except Exception as e: print("[EPG] Error auto-enabling EPGImport sources: %s" % e) def fix_cache_format( remove_duplicates=True, remove_unmatched=False, remove_invalid=False): """ Fix cache file. - Lowercase keys only (not name or id) - Remove extra fields - Mark duplicates as matched=False (case-insensitive comparison for names) - Remove entries based on flags Returns (fixed_count, removed_count) """ try: if not exists(CACHE_FILE): print("[Cache] No cache file found") return 0, 0 with open(CACHE_FILE, 'r') as f: cache = load(f) required = {'id', 'name', 'country', 'sref', 'timestamp', 'matched'} new_cache = {} modified = 0 duplicates = 0 keys_changed = False # Lista degli sref di fallback da considerare invalidi fallback_srefs = [ "4097:0:0:0:0:0:0:0:0:0:", "4097:0:1:1:1:40:0:0:0:0" ] for key, value in cache.items(): new_key = key.lower().strip() if new_key != key: keys_changed = True changed = False # Remove extra fields extra = set(value.keys()) - required if extra: for k in extra: del value[k] changed = True # Name: keep original case if 'name' not in value: value['name'] = key changed = True # Country: lowercase if 'country' not in value or not value['country']: parts = new_key.rsplit('_', 1) value['country'] = parts[-1] if len(parts) > 1 else '' changed = True else: new_country = str(value['country']).lower().strip() if new_country != value['country']: value['country'] = new_country changed = True # ID: keep original case if 'id' not in value or not value['id']: value['id'] = key changed = True # sref: se è vuoto o mancante, assegna il primo fallback (ma poi # verrà eventualmente rimosso) if 'sref' not in value or not value['sref']: value['sref'] = fallback_srefs[0] changed = True # timestamp if 'timestamp' not in value: value['timestamp'] = strftime('%Y-%m-%d %H:%M:%S', localtime()) changed = True # matched if 'matched' not in value: value['matched'] = False changed = True if changed: modified += 1 new_cache[new_key] = value # Mark duplicates (case-insensitive comparison on name) if remove_duplicates: groups = {} for k, v in new_cache.items(): name_key = v['name'].lower().strip() if v['name'] else '' country_key = v['country'] group = (name_key, country_key) groups.setdefault(group, []).append(k) for group, keys in groups.items(): if len(keys) > 1: for dup in keys[1:]: if new_cache[dup].get('matched', False): new_cache[dup]['matched'] = False duplicates += 1 modified += 1 # Remove entries based on flags removed = 0 to_delete = [] for k, v in new_cache.items(): if remove_unmatched and v.get('matched') is False: to_delete.append(k) elif remove_invalid and v.get('sref') in fallback_srefs: to_delete.append(k) to_delete = list(set(to_delete)) for k in to_delete: del new_cache[k] removed += 1 if removed: print( "[Cache] Removed %d entries (unmatched=%s, invalid=%s)" % (removed, remove_unmatched, remove_invalid)) # Save if any changes if modified > 0 or duplicates > 0 or keys_changed or removed > 0: temp_file = CACHE_FILE + ".tmp" with open(temp_file, 'w') as f: dump( new_cache, f, indent=4, sort_keys=True, ensure_ascii=False) rename(temp_file, CACHE_FILE) print( "[Cache] Fixed %d entries, marked %d duplicates, removed %d entries, keys lowercased" % (modified, duplicates, removed)) return modified, removed else: print("[Cache] No changes needed") return 0, 0 except Exception as e: print("[Cache] Error: %s" % e) trace_error() return 0, 0 def returnIMDB(text_clear, session): from Tools.Directories import SCOPE_PLUGINS, resolveFilename TMDB = resolveFilename(SCOPE_PLUGINS, "Extensions/{}".format('TMDB')) tmdbx = resolveFilename(SCOPE_PLUGINS, "Extensions/{}".format('tmdb')) IMDb = resolveFilename(SCOPE_PLUGINS, "Extensions/{}".format('IMDb')) text = html_unescape(text_clear) if exists(TMDB): try: # from Plugins.Extensions.TMBD.plugin import TMBD # print("[XCF] Opening TMDB for: %s" % text) # session.open(TMBD.tmdbScreen, text, 0) from Plugins.Extensions.TMBD.plugin import tmdbScreen print("[XCF] Opening TMDB for: %s" % text) session.open(tmdbScreen, text, 2) return True except Exception as e: print("[XCF] TMDB error: ", str(e)) if exists(tmdbx): try: # from Plugins.Extensions.tmdb.plugin import tmdb # print("[XCF] Opening tmdb for: %s" % text) # session.open(tmdbScreen, text, 0) from Plugins.Extensions.tmdb.tmdb import tmdbScreen session.open(tmdbScreen, text, 2) return True except Exception as e: print("[XCF] tmdb error: ", str(e)) if exists(IMDb): try: from Plugins.Extensions.IMDb.plugin import IMDB # main as imdb print("[XCF] Opening IMDb for: %s" % text) session.open(IMDB, text, False) # imdb(session, text) return True except Exception as e: print("[XCF] IMDb error: ", str(e)) return False