| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| """ | |
| smarttube_config.py | |
| ================================================================ | |
| PlayBox Titanium -- SmartTube Config Manager | |
| Eksport / Optymalizacja / Import konfiguracji SmartTube | |
| Uzycie: | |
| python3 smarttube_config.py --device 192.168.1.3:5555 | |
| python3 smarttube_config.py --dry-run | |
| python3 smarttube_config.py --export-only | |
| python3 smarttube_config.py --validate | |
| python3 smarttube_config.py --restore ./SmartTubeBackup_original.zip | |
| python3 smarttube_config.py --show ./SmartTubeBackup.zip | |
| ================================================================ | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import io | |
| import json | |
| import shutil | |
| import subprocess | |
| import sys | |
| import time | |
| import zipfile | |
| import xml.etree.ElementTree as ET | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Dict, List, Optional, Tuple | |
| # ================================================================ | |
| # Kolory (ASCII only -- Termux safe) | |
| # ================================================================ | |
| _R = "\033[0m"; _G = "\033[92m"; _Y = "\033[93m" | |
| _B = "\033[94m"; _E = "\033[91m"; _H = "\033[95m" | |
| _C = "\033[96m"; _D = "\033[2m"; _BO = "\033[1m" | |
| def ok(m): print(_G + " [OK] " + str(m) + _R) | |
| def warn(m): print(_Y + " [!!] " + str(m) + _R) | |
| def err(m): print(_E + " [XX] " + str(m) + _R) | |
| def info(m): print(_B + " [ ] " + str(m) + _R) | |
| def fix(m): print(_Y + " [>>] " + str(m) + _R) | |
| def hdr(m): print("\n" + _H + _BO + "=" * 62 + "\n " + str(m) + "\n" + "=" * 62 + _R) | |
| def sub(m): print("\n" + _C + " -- " + str(m) + " --" + _R) | |
| def step(n, t, m): print(_BO + _B + "\n [" + str(n) + "/" + str(t) + "] " + m + _R) | |
| # ================================================================ | |
| # Stale | |
| # ================================================================ | |
| PKG_STABLE = "org.smarttube.stable" | |
| PKG_BETA = "org.smarttube.beta" | |
| PKG_LEGACY = "com.liskovsoft.smarttubetv" | |
| DEVICE_BACKUP_ZIP = "/sdcard/SmartTubeBackup.zip" | |
| RESTORE_BROADCAST = "com.liskovsoft.smarttube.RESTORE" | |
| MIN_BACKUP_SIZE_BYTES = 10 * 1024 # 10 KB | |
| LOCAL_DIR = Path.home() / ".playbox_cache" / "smarttube_configs" | |
| LAUNCH_TIMEOUT = 20 | |
| # ================================================================ | |
| # ADB wrapper | |
| # ================================================================ | |
| class _ADB: | |
| dev: Optional[str] = None | |
| _TO = 30 | |
| @classmethod | |
| def connect(cls, target: str) -> bool: | |
| try: | |
| r = subprocess.run( | |
| ["adb", "connect", target], | |
| capture_output=True, text=True, timeout=10 | |
| ) | |
| if "connected" in r.stdout.lower(): | |
| cls.dev = target | |
| ok("ADB: " + target) | |
| return True | |
| err("ADB connect failed: " + r.stdout.strip()) | |
| return False | |
| except FileNotFoundError: | |
| err("'adb' nie znalezione -- zainstaluj Android Platform Tools") | |
| sys.exit(1) | |
| except subprocess.TimeoutExpired: | |
| err("ADB timeout: " + target) | |
| return False | |
| @classmethod | |
| def autodetect(cls) -> bool: | |
| try: | |
| out = subprocess.check_output( | |
| ["adb", "devices"], text=True, timeout=5 | |
| ) | |
| for line in out.splitlines(): | |
| if "\tdevice" in line: | |
| cls.dev = line.split("\t")[0].strip() | |
| ok("ADB auto-wykryty: " + cls.dev) | |
| return True | |
| except Exception: | |
| pass | |
| return False | |
| @classmethod | |
| def sh(cls, cmd: str, silent: bool = False) -> str: | |
| if not cls.dev: | |
| return "" | |
| try: | |
| return subprocess.check_output( | |
| ["adb", "-s", cls.dev, "shell", cmd], | |
| stderr=subprocess.STDOUT, text=True, timeout=cls._TO | |
| ).strip() | |
| except subprocess.TimeoutExpired: | |
| if not silent: | |
| warn("Timeout: " + cmd[:60]) | |
| return "" | |
| except subprocess.CalledProcessError as e: | |
| return (e.output or "").strip() | |
| except Exception as e: | |
| if not silent: | |
| err(str(e)) | |
| return "" | |
| @classmethod | |
| def pull(cls, remote: str, local: str) -> bool: | |
| try: | |
| subprocess.check_call( | |
| ["adb", "-s", cls.dev, "pull", remote, local], | |
| stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, | |
| timeout=120 | |
| ) | |
| return Path(local).exists() | |
| except Exception: | |
| return False | |
| @classmethod | |
| def push(cls, local: str, remote: str) -> bool: | |
| try: | |
| subprocess.check_call( | |
| ["adb", "-s", cls.dev, "push", local, remote], | |
| stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, | |
| timeout=120 | |
| ) | |
| return True | |
| except Exception: | |
| return False | |
| @classmethod | |
| def pkg_exists(cls, pkg: str) -> bool: | |
| return pkg in cls.sh("pm list packages " + pkg, silent=True) | |
| @classmethod | |
| def pkg_running(cls, pkg: str) -> bool: | |
| out = cls.sh("pidof " + pkg + " 2>/dev/null", silent=True) | |
| return bool(out and out.strip().split()[0].isdigit()) | |
| @classmethod | |
| def file_exists(cls, path: str) -> bool: | |
| return "yes" in cls.sh("test -f " + path + " && echo yes", silent=True) | |
| @classmethod | |
| def file_size(cls, path: str) -> int: | |
| out = cls.sh("stat -c %s " + path + " 2>/dev/null", silent=True) | |
| try: | |
| return int(out.strip()) | |
| except (ValueError, AttributeError): | |
| return 0 | |
| @classmethod | |
| def file_mtime(cls, path: str) -> str: | |
| return cls.sh("stat -c '%y' " + path + " 2>/dev/null", silent=True) | |
| @classmethod | |
| def pkg_version_code(cls, pkg: str) -> str: | |
| out = cls.sh( | |
| "dumpsys package " + pkg + " | grep versionCode", | |
| silent=True | |
| ) | |
| for part in (out or "").split(): | |
| if part.startswith("versionCode="): | |
| return part.split("=")[1] | |
| return "" | |
| def detect_pkg() -> Optional[str]: | |
| for pkg in (PKG_STABLE, PKG_BETA, PKG_LEGACY): | |
| if _ADB.pkg_exists(pkg): | |
| return pkg | |
| return None | |
| # ================================================================ | |
| # WALIDACJA backupu | |
| # ================================================================ | |
| class BackupValidator: | |
| @dataclass | |
| class Result: | |
| ok: bool | |
| size_bytes: int | |
| mtime: str | |
| zip_valid: bool | |
| has_prefs: bool | |
| prefs_files: List[str] | |
| has_databases: bool | |
| db_files: List[str] | |
| all_entries: List[str] | |
| version_in_zip: str | |
| warnings: List[str] | |
| errors: List[str] | |
| @classmethod | |
| def validate_local(cls, zip_path: Path): | |
| warns = [] | |
| errors = [] | |
| prefs = [] | |
| dbs = [] | |
| all_e = [] | |
| ver = "" | |
| size = zip_path.stat().st_size if zip_path.exists() else 0 | |
| if not zip_path.exists(): | |
| errors.append("Plik nie istnieje: " + str(zip_path)) | |
| return cls._make(False, 0, "", False, False, [], False, [], [], "", warns, errors) | |
| if size < MIN_BACKUP_SIZE_BYTES: | |
| errors.append("Plik za maly: " + str(size) + " B (min " + str(MIN_BACKUP_SIZE_BYTES) + " B)") | |
| if not zipfile.is_zipfile(str(zip_path)): | |
| errors.append("Plik nie jest poprawnym ZIP (uszkodzony?)") | |
| return cls._make(False, size, "", False, False, [], False, [], [], "", warns, errors) | |
| try: | |
| with zipfile.ZipFile(str(zip_path), "r") as zf: | |
| bad = zf.testzip() | |
| if bad: | |
| errors.append("Uszkodzony plik w ZIP: " + bad) | |
| all_e = zf.namelist() | |
| # POPRAWKA: szukaj gdziekolwiek w sciezce wewnatrz zipa | |
| prefs = [n for n in all_e if "shared_prefs/" in n and n.endswith(".xml")] | |
| dbs = [n for n in all_e if "databases/" in n and n.endswith(".db")] | |
| for candidate in ("META-INF/MANIFEST.MF", "version.txt", "manifest.xml"): | |
| if candidate in all_e: | |
| try: | |
| content = zf.read(candidate).decode("utf-8", errors="replace") | |
| for line in content.splitlines(): | |
| if "version" in line.lower() or "versionCode" in line: | |
| ver = line.strip()[:40] | |
| break | |
| except Exception: | |
| pass | |
| except zipfile.BadZipFile as e: | |
| errors.append("Blad odczytu ZIP: " + str(e)) | |
| return cls._make(False, size, "", True, False, [], False, [], [], "", warns, errors) | |
| has_prefs = len(prefs) > 0 | |
| has_dbs = len(dbs) > 0 | |
| if not has_prefs: | |
| errors.append("Brak shared_prefs/*.xml -- backup niekompletny") | |
| oauth_suspects = [n for n in all_e if any(kw in n.lower() for kw in ("token", "oauth", "auth", "account", "credential"))] | |
| if oauth_suspects: | |
| warns.append("Znaleziono potencjalne pliki auth: " + ", ".join(oauth_suspects[:3]) + " -- sprawdz recznie przed udostepnieniem backupu") | |
| mtime = datetime.fromtimestamp(zip_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S") | |
| age_days = (datetime.now() - datetime.fromtimestamp(zip_path.stat().st_mtime)).days | |
| if age_days > 30: | |
| warns.append("Backup ma " + str(age_days) + " dni -- rozważ odswiezenie") | |
| is_ok = len(errors) == 0 | |
| return cls._make(is_ok, size, mtime, True, has_prefs, prefs, has_dbs, dbs, all_e, ver, warns, errors) | |
| @classmethod | |
| def validate_device(cls, pkg: str) -> dict: | |
| result = { | |
| "exists": False, | |
| "size": 0, | |
| "mtime": "", | |
| "zip_ok": False, | |
| "contents": [], | |
| "warnings": [], | |
| "errors": [], | |
| } | |
| if not _ADB.file_exists(DEVICE_BACKUP_ZIP): | |
| result["errors"].append("Brak pliku: " + DEVICE_BACKUP_ZIP) | |
| return result | |
| result["exists"] = True | |
| result["size"] = _ADB.file_size(DEVICE_BACKUP_ZIP) | |
| result["mtime"] = _ADB.file_mtime(DEVICE_BACKUP_ZIP) | |
| if result["size"] < MIN_BACKUP_SIZE_BYTES: | |
| result["errors"].append("Plik za maly: " + str(result["size"]) + " B") | |
| return result | |
| contents_raw = _ADB.sh("unzip -l " + DEVICE_BACKUP_ZIP + " 2>/dev/null", silent=True) | |
| if contents_raw: | |
| result["zip_ok"] = True | |
| for line in contents_raw.splitlines(): | |
| parts = line.strip().split() | |
| if len(parts) >= 4 and parts[-1] not in ("Name", "----"): | |
| result["contents"].append(parts[-1]) | |
| app_ver = _ADB.pkg_version_code(pkg) | |
| if app_ver: | |
| result["app_version_code"] = app_ver | |
| return result | |
| @classmethod | |
| def _make(cls, ok, size, mtime, zip_valid, has_prefs, prefs, has_dbs, dbs, all_e, ver, warns, errors): | |
| return { | |
| "ok": ok, | |
| "size_bytes": size, | |
| "mtime": mtime, | |
| "zip_valid": zip_valid, | |
| "has_prefs": has_prefs, | |
| "prefs_files": prefs, | |
| "has_databases": has_dbs, | |
| "db_files": dbs, | |
| "all_entries": all_e, | |
| "version_in_zip": ver, | |
| "warnings": warns, | |
| "errors": errors, | |
| } | |
| @classmethod | |
| def print_report(cls, r: dict, label: str = "") -> None: | |
| sub("Walidacja backupu" + (" -- " + label if label else "")) | |
| stat_ok = r.get("ok", False) | |
| size = r.get("size_bytes", 0) | |
| mtime = r.get("mtime", "?") | |
| _s = _G if stat_ok else _E | |
| print(" Status : " + _s + ("OK" if stat_ok else "BLAD") + _R) | |
| print(" Rozmiar : " + str(size) + " B" + | |
| (" (" + _G + "OK" + _R + ")" if size >= MIN_BACKUP_SIZE_BYTES | |
| else " (" + _E + "ZA MALY" + _R + ")")) | |
| print(" Data : " + str(mtime)) | |
| print(" ZIP valid : " + (_G + "TAK" + _R if r.get("zip_valid") else _E + "NIE" + _R)) | |
| prefs = r.get("prefs_files", []) | |
| print(" shared_prefs: " + (_G if prefs else _E) + str(len(prefs)) + " plikow" + _R) | |
| for p in prefs: | |
| print(" " + _C + p + _R) | |
| dbs = r.get("db_files", []) | |
| if dbs: | |
| print(" databases : " + str(len(dbs)) + " plikow") | |
| for d in dbs: | |
| print(" " + _C + d + _R) | |
| ver = r.get("version_in_zip", "") | |
| if ver: | |
| print(" Wersja : " + _D + ver + _R) | |
| app_ver = r.get("app_version_code", "") | |
| if app_ver: | |
| print(" App ver : " + _D + "versionCode=" + app_ver + _R) | |
| warns = r.get("warnings", []) | |
| errors = r.get("errors", []) | |
| for w in warns: warn(w) | |
| for e in errors: err(e) | |
| if not warns and not errors: | |
| ok("Backup kompletny i poprawny") | |
| print() | |
| info("UWAGA: Tokeny OAuth/konto Google NIE sa w backupie") | |
| info(" Po przywroceniu wymagane jest ponowne logowanie do konta") | |
| # ================================================================ | |
| # KROK 1 -- Eksport i walidacja | |
| # ================================================================ | |
| class ConfigExporter: | |
| @classmethod | |
| def export(cls, pkg: str, local_dir: Path) -> Optional[Path]: | |
| hdr("KROK 1 -- Eksport SmartTubeBackup.zip") | |
| local_dir.mkdir(parents=True, exist_ok=True) | |
| if _ADB.file_exists(DEVICE_BACKUP_ZIP): | |
| size = _ADB.file_size(DEVICE_BACKUP_ZIP) | |
| mtime = _ADB.file_mtime(DEVICE_BACKUP_ZIP) | |
| ok("Znaleziono: " + DEVICE_BACKUP_ZIP) | |
| info(" Rozmiar: " + str(size) + " B") | |
| info(" Data : " + mtime) | |
| if size < MIN_BACKUP_SIZE_BYTES: | |
| warn("Plik wydaje sie za maly (" + str(size) + " B)") | |
| warn("Moz byc uszkodzony -- generuje nowy") | |
| if not cls._trigger_backup(pkg): | |
| warn("Nie mozna wygenerowac nowego -- uzywam istniejacego") | |
| else: | |
| warn("Brak pliku: " + DEVICE_BACKUP_ZIP) | |
| info("Uruchamiam SmartTube aby wygenerowac backup...") | |
| if not cls._trigger_backup(pkg): | |
| err("Nie mozna wygenerowac backupu") | |
| cls._print_manual_hint() | |
| return None | |
| local_zip = local_dir / "SmartTubeBackup_original.zip" | |
| info("Pobieranie: " + DEVICE_BACKUP_ZIP + " -> " + str(local_zip)) | |
| if not _ADB.pull(DEVICE_BACKUP_ZIP, str(local_zip)): | |
| err("Pull nieudany: " + DEVICE_BACKUP_ZIP) | |
| return None | |
| ok("Pobrano: " + str(local_zip) + " (" + str(local_zip.stat().st_size) + " B)") | |
| v = BackupValidator.validate_local(local_zip) | |
| BackupValidator.print_report(v, "oryginál") | |
| if not v["ok"]: | |
| err("Backup niepoprawny -- nie mozna kontynuowac") | |
| return None | |
| return local_zip | |
| @classmethod | |
| def _trigger_backup(cls, pkg: str) -> bool: | |
| _ADB.sh("am force-stop " + pkg, silent=True) | |
| time.sleep(1) | |
| export_actions = [ | |
| "com.liskovsoft.smarttube.BACKUP", | |
| "com.liskovsoft.smartyoutubetv2.BACKUP_EXPORT", | |
| "com.liskovsoft.smarttubetv.EXPORT", | |
| ] | |
| for action in export_actions: | |
| out = _ADB.sh("am broadcast -a " + action + " -p " + pkg + " 2>/dev/null", silent=True) | |
| if "result=0" in out or "Broadcast completed" in out: | |
| ok(" Broadcast: " + action) | |
| time.sleep(3) | |
| if _ADB.file_exists(DEVICE_BACKUP_ZIP): | |
| if _ADB.file_size(DEVICE_BACKUP_ZIP) > MIN_BACKUP_SIZE_BYTES: | |
| ok(" Backup wygenerowany") | |
| return True | |
| info(" Uruchamiam SmartTube (zapisuje backup przy starcie)...") | |
| _ADB.sh("monkey -p " + pkg + " -c android.intent.category.LAUNCHER 1 2>/dev/null", silent=True) | |
| for i in range(LAUNCH_TIMEOUT): | |
| time.sleep(1) | |
| print(" Czekam na SmartTubeBackup.zip... " + str(i + 1) + "/" + str(LAUNCH_TIMEOUT) + "s", end="\r") | |
| if _ADB.file_exists(DEVICE_BACKUP_ZIP): | |
| if _ADB.file_size(DEVICE_BACKUP_ZIP) > MIN_BACKUP_SIZE_BYTES: | |
| print() | |
| ok(" Backup wygenerowany po " + str(i + 1) + "s") | |
| time.sleep(2) | |
| _ADB.sh("am force-stop " + pkg, silent=True) | |
| return True | |
| print() | |
| _ADB.sh("am force-stop " + pkg, silent=True) | |
| time.sleep(2) | |
| if _ADB.file_exists(DEVICE_BACKUP_ZIP): | |
| if _ADB.file_size(DEVICE_BACKUP_ZIP) > MIN_BACKUP_SIZE_BYTES: | |
| ok(" Backup wygenerowany przy zamknieciu") | |
| return True | |
| return False | |
| @classmethod | |
| def _print_manual_hint(cls) -> None: | |
| print() | |
| warn("Recznie wygeneruj backup w SmartTube:") | |
| print(" " + _Y + " 1. Otworz SmartTube" + _R) | |
| print(" " + _Y + " 2. Ustawienia -> Kopia zapasowa -> Utworz kopie zapasowa" + _R) | |
| print(" " + _Y + " 3. Uruchom ponownie: python3 smarttube_config.py" + _R) | |
| # ================================================================ | |
| # Baza optymalnych ustawien dla BCM72604 | |
| # ================================================================ | |
| BCM72604_OPTIMAL: Dict[str, Tuple[str, str, str]] = { | |
| "video_codec": ("int", "2", "VP9 -- jedyny HW codec BCM72604"), | |
| "av1_enabled": ("boolean", "false", "AV1 off -- brak HW, 100% CPU A15"), | |
| "vp9_enabled": ("boolean", "true", "VP9 on -- VDec32 HW path"), | |
| "preferred_codec": ("string", "vp9", "Preferowany VP9"), | |
| "codec_black_list": ("string", "av1", "AV1 na czarnej liscie"), | |
| "use_tunnel_mode": ("boolean", "true", "Tunnel -- hardware clock sync BCM"), | |
| "tunnel_mode_enabled": ("boolean", "true", "Tunnel mode enabled"), | |
| "tunneled_playback": ("boolean", "true", "Tunneled playback"), | |
| "video_resolution": ("string", "1080", "Max 1080p"), | |
| "preferred_resolution": ("int", "4", "1080p index"), | |
| "max_video_resolution": ("string", "1080", "Max 1080p"), | |
| "default_quality": ("string", "hd1080","HD 1080p domyslnie"), | |
| "hdr_mode": ("boolean", "true", "HDR10 wlaczony"), | |
| "hdr_enabled": ("boolean", "true", "HDR enabled"), | |
| "hdr10_enabled": ("boolean", "true", "HDR10 enabled"), | |
| "video_frame_rate_switch": ("boolean", "true", "Auto fps switch"), | |
| "auto_frame_rate": ("boolean", "true", "Automatyczna czestotliwosc"), | |
| "frame_rate_switching": ("boolean", "true", "Przelaczanie fps"), | |
| "switch_frame_rate": ("boolean", "true", "Przelacz fps do tresci"), | |
| "player_data_source": ("int", "0", "ExoPlayer (nie MediaPlayer)"), | |
| "buffer_type": ("int", "2", "Duzy bufor"), | |
| "min_buffer_duration": ("string", "15000", "Min bufor 15s"), | |
| "max_buffer_duration": ("string", "50000", "Max bufor 50s"), | |
| "playback_buffer_ms": ("string", "2500", "Bufor do startu 2.5s"), | |
| "back_buffer_ms": ("string", "10000", "Bufor cofania 10s"), | |
| "hardware_acceleration": ("boolean", "true", "HW acceleration"), | |
| "hw_acceleration": ("boolean", "true", "HW acceleration"), | |
| "use_hardware_decoder": ("boolean", "true", "HW dekoder"), | |
| "background_play_mode": ("int", "0", "Brak odtwarzania w tle"), | |
| "background_playback": ("boolean", "false", "Background off"), | |
| "player_stats": ("boolean", "false", "Stats overlay off"), | |
| "show_stats": ("boolean", "false", "Statystyki off"), | |
| "debug_view": ("boolean", "false", "Debug view off"), | |
| "audio_delay_ms": ("int", "0", "Brak opoznienia audio"), | |
| "frame_interpolation": ("boolean", "false", "Interpolacja off"), | |
| "smooth_playback": ("boolean", "false", "Smooth off"), | |
| "abr_enabled": ("boolean", "true", "ABR wlaczony"), | |
| "remember_quality": ("boolean", "true", "Pamietaj jakosc"), | |
| "force_default_quality": ("boolean", "false", "Nie wymuszaj domyslnej"), | |
| "speed_buttons": ("boolean", "true", "Przyciski szybkosci"), | |
| "default_speed": ("float", "1.0", "Szybkosc 1.0x"), | |
| "suggestions_enabled": ("boolean", "false", "Sugestie off (CPU)"), | |
| "feed_enabled": ("boolean", "false", "Feed off"), | |
| "autoplay_enabled": ("boolean", "true", "Autoplay on"), | |
| "subtitles_font_size": ("int", "16", "Rozmiar napisow TV"), | |
| "live_stream_quality": ("string", "1080", "Live 1080p"), | |
| } | |
| # ================================================================ | |
| # KROK 2 -- Optymalizacja ZIP | |
| # ================================================================ | |
| class ZipOptimizer: | |
| def __init__(self, source_zip: Path, dry_run: bool = False): | |
| self.source = source_zip | |
| self.dry_run = dry_run | |
| self.changes_by_file: Dict[str, List[Tuple[str, str, str]]] = {} | |
| def _optimize_xml_bytes(self, data: bytes, entry_name: str) -> Tuple[bytes, int]: | |
| changes = [] | |
| try: | |
| root = ET.fromstring(data.decode("utf-8", errors="replace")) | |
| except ET.ParseError as e: | |
| warn(" XML parse error w " + entry_name + ": " + str(e)) | |
| return data, 0 | |
| def _set(key, typ, value): | |
| for el in root: | |
| if el.get("name") == key: | |
| old = el.text if el.tag == "string" else el.get("value", "") | |
| if str(old) == str(value) and el.tag == typ: | |
| return | |
| if typ == "string": | |
| el.tag = "string" | |
| el.text = value | |
| if "value" in el.attrib: | |
| del el.attrib["value"] | |
| else: | |
| el.tag = typ | |
| el.set("value", value) | |
| el.text = None | |
| changes.append((key, str(old), value)) | |
| return | |
| if typ == "string": | |
| ne = ET.SubElement(root, "string", {"name": key}) | |
| ne.text = value | |
| else: | |
| ET.SubElement(root, typ, {"name": key, "value": value}) | |
| changes.append((key, "<brak>", value)) | |
| for key, (typ, value, _desc) in BCM72604_OPTIMAL.items(): | |
| _set(key, typ, value) | |
| for el in root: | |
| name = el.get("name", "") | |
| val = el.text if el.tag == "string" else el.get("value", "") | |
| if val and val in ["2160", "4k", "4K", "UHD", "2160p", "5", "6", "7"]: | |
| if any(kw in name.lower() for kw in ("resolution", "quality")): | |
| old = val | |
| if el.tag == "string": | |
| el.text = "1080" | |
| else: | |
| el.set("value", "4") | |
| changes.append((name, old, "1080(4K->1080p)")) | |
| if "av1" in name.lower() and val not in ("false", "0", ""): | |
| old = val | |
| if el.tag == "boolean": | |
| el.set("value", "false") | |
| elif el.tag == "string": | |
| el.text = "false" | |
| elif el.tag == "int": | |
| el.set("value", "0") | |
| changes.append((name, old, "disabled(av1)")) | |
| if not changes: | |
| return data, 0 | |
| _indent_xml(root) | |
| buf = io.BytesIO() | |
| ET.ElementTree(root).write(buf, encoding="utf-8", xml_declaration=True) | |
| return buf.getvalue(), len(changes) | |
| def optimize(self, output_zip: Path) -> bool: | |
| hdr("KROK 2 -- Optymalizacja shared_prefs w ZIP") | |
| if not zipfile.is_zipfile(str(self.source)): | |
| err("Zrodlowy ZIP niepoprawny") | |
| return False | |
| total_changes = 0 | |
| total_xml = 0 | |
| try: | |
| with zipfile.ZipFile(str(self.source), "r") as zin: | |
| entries = zin.infolist() | |
| info("Plikow w ZIP: " + str(len(entries))) | |
| if self.dry_run: | |
| for entry in entries: | |
| if entry.is_dir(): | |
| continue | |
| fname = entry.filename | |
| raw = zin.read(fname) | |
| # POPRAWKA: szukaj gdziekolwiek w sciezce | |
| if "shared_prefs/" in fname and fname.endswith(".xml"): | |
| _, n = self._optimize_xml_bytes(raw, fname) | |
| if n > 0: | |
| fix(" [DRY] " + fname + ": " + str(n) + " zmian") | |
| total_changes += n | |
| else: | |
| ok(" [DRY] " + fname + ": brak zmian") | |
| total_xml += 1 | |
| else: | |
| info(" [DRY] " + _D + fname + " (kopia bez zmian)" + _R) | |
| info("Lacznie: " + str(total_changes) + " potencjalnych zmian w " + str(total_xml) + " plikach XML") | |
| return True | |
| with zipfile.ZipFile(str(output_zip), "w", compression=zipfile.ZIP_DEFLATED) as zout: | |
| for entry in entries: | |
| raw = zin.read(entry.filename) | |
| fname = entry.filename | |
| if entry.is_dir(): | |
| zout.mkdir(fname) | |
| continue | |
| # POPRAWKA: szukaj gdziekolwiek w sciezce | |
| if "shared_prefs/" in fname and fname.endswith(".xml"): | |
| info(" XML: " + fname) | |
| new_raw, n = self._optimize_xml_bytes(raw, fname) | |
| if n > 0: | |
| fix(" " + str(n) + " zmian") | |
| self.changes_by_file[fname] = [] | |
| total_changes += n | |
| total_xml += 1 | |
| else: | |
| ok(" brak zmian -- juz optymalny") | |
| zout.writestr(entry, new_raw) | |
| elif "databases/" in fname and fname.endswith(".db"): | |
| info(" DB: " + fname + _D + " (binarna -- kopia bez zmian)" + _R) | |
| zout.writestr(entry, raw) | |
| else: | |
| info(" " + _D + fname + " (kopia bez zmian)" + _R) | |
| zout.writestr(entry, raw) | |
| except zipfile.BadZipFile as e: | |
| err("Blad ZIP: " + str(e)) | |
| return False | |
| except Exception as e: | |
| err("Blad przetwarzania: " + str(e)) | |
| import traceback | |
| traceback.print_exc() | |
| return False | |
| print() | |
| ok("Optymalizacja ZIP zakonczona:") | |
| ok(" Zmian lacznic : " + str(total_changes)) | |
| ok(" Plikow XML : " + str(total_xml)) | |
| ok(" Wynikowy ZIP : " + str(output_zip) + " (" + str(output_zip.stat().st_size) + " B)") | |
| return True | |
| def print_changes(self) -> None: | |
| if not self.changes_by_file: | |
| return | |
| sub("Zmienione pliki") | |
| for fname, changes in self.changes_by_file.items(): | |
| print(" " + _BO + fname + _R) | |
| for key, old, new in changes: | |
| ks = (key[:32] + "..") if len(key) > 34 else key.ljust(34) | |
| print(" " + _C + ks + _R + " " + _D + str(old)[:10].ljust(10) + _R + " -> " + _G + str(new)[:14] + _R) | |
| # ================================================================ | |
| # KROK 3 -- Import do SmartTube (Zaktualizowany system uderzeniowy A+B) | |
| # ================================================================ | |
| class ConfigImporter: | |
| """ | |
| Importuje zoptymalizowany plik XML i ZIP do SmartTube. | |
| Wdrożono dwukanałowy proces wypychania dla maksymalnej stabilności. | |
| """ | |
| @classmethod | |
| def import_zip(cls, pkg: str, optimized_zip: Path, dry_run: bool = False) -> bool: | |
| hdr("KROK 3 -- Wgrywanie konfiguracji do SmartTube") | |
| if dry_run: | |
| info("[DRY RUN] Import pominieto") | |
| return True | |
| local_size = optimized_zip.stat().st_size | |
| remote_backup_dir = "/sdcard/data/" + pkg + "/backup" | |
| info("Cel A (glowny) : " + remote_backup_dir) | |
| info("Cel B (zapasowy): " + DEVICE_BACKUP_ZIP) | |
| print() | |
| v = BackupValidator.validate_local(optimized_zip) | |
| if not v["ok"]: | |
| err("Zoptymalizowany ZIP nie przeszedl walidacji") | |
| for e in v["errors"]: | |
| err(" " + e) | |
| return False | |
| _ADB.sh("am force-stop " + pkg, silent=True) | |
| time.sleep(1.5) | |
| ok("SmartTube zatrzymany") | |
| print("\n " + _C + "-- Kanal A -- wypakowanie do katalogu importu --" + _R) | |
| unpack_dir = optimized_zip.parent / "_unpack_tmp" | |
| unpack_dir.mkdir(parents=True, exist_ok=True) | |
| with zipfile.ZipFile(str(optimized_zip), "r") as zf: | |
| zf.extractall(str(unpack_dir)) | |
| _ADB.sh("mkdir -p " + remote_backup_dir, silent=True) | |
| ok("Katalog importu: " + remote_backup_dir) | |
| pushed_count = 0 | |
| for item in unpack_dir.rglob("*"): | |
| if item.is_file(): | |
| size = item.stat().st_size | |
| remote_file = remote_backup_dir + "/" + item.name | |
| print(" " + _B + "[ ]" + _R + " Push: " + item.name + " (" + str(size) + " B) -> " + remote_file) | |
| _ADB.push(str(item), remote_file) | |
| r_size = _ADB.file_size(remote_file) | |
| if r_size == size: | |
| print(" " + _G + "[OK]" + _R + " OK -- " + str(size) + " B") | |
| pushed_count += 1 | |
| else: | |
| print(" " + _E + "[XX]" + _R + " BLAD -- " + str(r_size) + " B != " + str(size) + " B") | |
| ok("Wgrano : " + str(pushed_count) + " plikow") | |
| ok("Sciezka : " + remote_backup_dir) | |
| ls_out = _ADB.sh("ls -la " + remote_backup_dir, silent=True) | |
| if ls_out: | |
| for line in ls_out.splitlines(): | |
| print(" " + _D + line + _R) | |
| shutil.rmtree(str(unpack_dir), ignore_errors=True) | |
| print("\n " + _C + "-- Kanal B -- push ZIP jako SmartTubeBackup.zip --" + _R) | |
| _ADB.push(str(optimized_zip), DEVICE_BACKUP_ZIP) | |
| zip_r_size = _ADB.file_size(DEVICE_BACKUP_ZIP) | |
| if zip_r_size == local_size: | |
| ok("ZIP na urzadzeniu: " + str(zip_r_size) + " B") | |
| else: | |
| err("Blad rozmiaru ZIP na urzadzeniu: " + str(zip_r_size) + " B") | |
| print("\n " + _C + "-- Restarty systemowe --" + _R) | |
| info("Wysylanie intentu przywrocenia (Broadcast RESTORE)...") | |
| broadcast_cmd = "am broadcast -a " + RESTORE_BROADCAST + " --es path " + DEVICE_BACKUP_ZIP + " " + pkg + " 2>/dev/null" | |
| _ADB.sh(broadcast_cmd, silent=True) | |
| info("Testowy rozruch (wczytanie wstrzyknietych wlasciwosci)...") | |
| _ADB.sh("monkey -p " + pkg + " -c android.intent.category.LAUNCHER 1 2>/dev/null", silent=True) | |
| time.sleep(4) | |
| info("Twardy reset procesu (dla finalizacji)...") | |
| _ADB.sh("am force-stop " + pkg, silent=True) | |
| time.sleep(2) | |
| _ADB.sh("monkey -p " + pkg + " -c android.intent.category.LAUNCHER 1 2>/dev/null", silent=True) | |
| time.sleep(3) | |
| if _ADB.pkg_running(pkg): | |
| ok("Proces SmartTube stabilny po imporcie.") | |
| else: | |
| warn("Uruchom SmartTube recznie.") | |
| print() | |
| sub("Instrukcje post-wdrożeniowe") | |
| print(" " + _G + _BO + "Aplikacja wczyta nowy zestaw zasad. Zweryfikuj:" + _R) | |
| print(" " + _C + " - Kodek: VP9 | HDR: Włączony" + _R) | |
| print(" " + _Y + "UWAGA: Konieczne ponowne zalogowanie do konta Google!" + _R) | |
| return True | |
| # ================================================================ | |
| # Pomocnicze | |
| # ================================================================ | |
| def _indent_xml(elem: ET.Element, level: int = 0) -> None: | |
| indent = "\n" + " " * level | |
| if len(elem): | |
| if not elem.text or not elem.text.strip(): | |
| elem.text = indent + " " | |
| if not elem.tail or not elem.tail.strip(): | |
| elem.tail = indent | |
| last = None | |
| for child in elem: | |
| _indent_xml(child, level + 1) | |
| last = child | |
| if last is not None and (not last.tail or not last.tail.strip()): | |
| last.tail = indent | |
| else: | |
| if level and (not elem.tail or not elem.tail.strip()): | |
| elem.tail = indent | |
| def print_zip_summary(zip_path: Path) -> None: | |
| KEYS = [ | |
| "video_codec", "use_tunnel_mode", "hdr_mode", | |
| "video_resolution", "auto_frame_rate", | |
| "hardware_acceleration", "av1_enabled", "vp9_enabled", | |
| "buffer_type", "background_play_mode", "player_stats", | |
| ] | |
| if not zip_path.exists() or not zipfile.is_zipfile(str(zip_path)): | |
| return | |
| print(" " + _BO + zip_path.name + ":" + _R) | |
| xml_shown = 0 | |
| try: | |
| with zipfile.ZipFile(str(zip_path), "r") as zf: | |
| for entry in zf.infolist(): | |
| if entry.is_dir(): | |
| continue | |
| fname = entry.filename | |
| # POPRAWKA: szukaj gdziekolwiek w sciezce | |
| if not ("shared_prefs/" in fname and fname.endswith(".xml")): | |
| continue | |
| raw = zf.read(fname) | |
| found = {} | |
| try: | |
| root = ET.fromstring(raw.decode("utf-8", errors="replace")) | |
| for el in root: | |
| name = el.get("name", "") | |
| if name in KEYS: | |
| val = el.text if el.tag == "string" else el.get("value", "?") | |
| found[name] = (el.tag, val) | |
| except Exception: | |
| pass | |
| if found: | |
| print(" " + _D + " [" + fname + "]" + _R) | |
| for key in KEYS: | |
| if key in found: | |
| typ, val = found[key] | |
| opt = BCM72604_OPTIMAL.get(key, (None, None, None))[1] | |
| col = _G if (opt and str(val) == str(opt)) else (_E if opt else _B) | |
| print(" " + _C + key[:34].ljust(34) + _R + " = " + col + str(val) + _R + _D + " [" + typ + "]" + _R) | |
| xml_shown += 1 | |
| except Exception as e: | |
| warn("Nie mozna odczytac ZIP: " + str(e)) | |
| if xml_shown == 0: | |
| warn(" Brak plikow shared_prefs/*.xml w " + zip_path.name) | |
| print() | |
| # ================================================================ | |
| # Glowny menedzer | |
| # ================================================================ | |
| class SmartTubeConfigManager: | |
| @classmethod | |
| def run_full(cls, | |
| device: Optional[str] = None, | |
| dry_run: bool = False, | |
| export_only: bool = False, | |
| validate_only:bool = False, | |
| restore_zip: Optional[str] = None) -> bool: | |
| hdr("SmartTube Config Manager -- BCM72604 Optimizer") | |
| info("Backup path: " + DEVICE_BACKUP_ZIP) | |
| info("Tryb: " + ("DRY RUN" if dry_run | |
| else "VALIDATE" if validate_only | |
| else "RESTORE" if restore_zip | |
| else "EXPORT ONLY" if export_only | |
| else "PELNY")) | |
| if device: | |
| if not _ADB.connect(device): | |
| return False | |
| else: | |
| if not _ADB.autodetect(): | |
| err("Brak polaczenia ADB -- podaj --device IP:PORT") | |
| return False | |
| pkg = detect_pkg() | |
| if not pkg: | |
| err("SmartTube nie zainstalowane") | |
| return False | |
| ok("Pakiet: " + pkg) | |
| if validate_only: | |
| sub("Walidacja na urzadzeniu") | |
| rv = BackupValidator.validate_device(pkg) | |
| BackupValidator.print_report(rv, "urzadzenie") | |
| return rv.get("ok", False) | |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| work_dir = LOCAL_DIR / ts | |
| work_dir.mkdir(parents=True, exist_ok=True) | |
| info("Katalog roboczy: " + str(work_dir)) | |
| if restore_zip: | |
| return cls._do_restore(pkg, Path(restore_zip), dry_run) | |
| step(1, 3, "Eksport SmartTubeBackup.zip z urzadzenia") | |
| original_zip = ConfigExporter.export(pkg, work_dir / "original") | |
| if original_zip is None: | |
| return False | |
| sub("Stan konfiguracji PRZED optymalizacja") | |
| print_zip_summary(original_zip) | |
| if export_only: | |
| ok("Eksport zakonczone") | |
| info("Plik: " + str(original_zip)) | |
| return True | |
| step(2, 3, "Optymalizacja shared_prefs dla BCM72604") | |
| optimized_zip = work_dir / "SmartTubeBackup.zip" | |
| optimizer = ZipOptimizer(original_zip, dry_run=dry_run) | |
| if not optimizer.optimize(optimized_zip): | |
| err("Optymalizacja nieudana") | |
| return False | |
| if not dry_run: | |
| sub("Stan konfiguracji PO optymalizacji") | |
| print_zip_summary(optimized_zip) | |
| v2 = BackupValidator.validate_local(optimized_zip) | |
| BackupValidator.print_report(v2, "zoptymalizowany") | |
| if not v2["ok"]: | |
| err("Zoptymalizowany ZIP nie przeszedl walidacji") | |
| return False | |
| step(3, 3, "Import do SmartTube") | |
| if not dry_run: | |
| try: | |
| ans = input("\n " + _Y + "Wyslac zoptymalizowany ZIP i wykonac import? [T/n] > " + _R).strip().lower() | |
| proceed = ans in ("", "t", "y", "tak", "yes") | |
| except (EOFError, KeyboardInterrupt): | |
| proceed = False | |
| if proceed: | |
| if ConfigImporter.import_zip(pkg, optimized_zip, dry_run): | |
| hdr("Gotowe!") | |
| ok("Konfiguracja SmartTube zoptymalizowana dla BCM72604 ✓") | |
| info("Oryginalna kopia: " + str(original_zip)) | |
| info("Rollback: --restore " + str(original_zip)) | |
| return True | |
| return False | |
| else: | |
| info("Import anulowany") | |
| info("Plik gotowy: " + str(optimized_zip)) | |
| info("Recznie: --restore " + str(optimized_zip)) | |
| else: | |
| info("[DRY RUN] Wszystkie kroki przeanalizowane bez zapisu") | |
| return True | |
| @classmethod | |
| def _do_restore(cls, pkg: str, zip_path: Path, dry_run: bool) -> bool: | |
| hdr("RESTORE -- Przywracanie konfiguracji") | |
| if not zip_path.exists(): | |
| err("Plik nie istnieje: " + str(zip_path)) | |
| return False | |
| if not zipfile.is_zipfile(str(zip_path)): | |
| err("Nie jest poprawnym ZIP: " + str(zip_path)) | |
| return False | |
| v = BackupValidator.validate_local(zip_path) | |
| BackupValidator.print_report(v, "restore") | |
| if not v["ok"]: | |
| err("Plik nie przeszedl walidacji -- przerywam") | |
| return False | |
| ok("Przywracam: " + str(zip_path)) | |
| return ConfigImporter.import_zip(pkg, zip_path, dry_run) | |
| # ================================================================ | |
| # CLI | |
| # ================================================================ | |
| def main() -> None: | |
| parser = argparse.ArgumentParser( | |
| description="SmartTube Config Manager -- BCM72604 Optimizer", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog=( | |
| "PRZYKLADY:\n" | |
| " python3 smarttube_config.py --device 192.168.1.3:5555\n" | |
| " python3 smarttube_config.py --dry-run\n" | |
| " python3 smarttube_config.py --validate\n" | |
| " python3 smarttube_config.py --export-only\n" | |
| " python3 smarttube_config.py --show ./SmartTubeBackup.zip\n" | |
| " python3 smarttube_config.py --restore ./SmartTubeBackup_original.zip\n" | |
| ) | |
| ) | |
| parser.add_argument("--device", default=None) | |
| parser.add_argument("--dry-run", action="store_true", help="Pokaz zmiany bez zapisu i importu") | |
| parser.add_argument("--export-only", action="store_true", help="Tylko pobierz ZIP") | |
| parser.add_argument("--validate", action="store_true", help="Tylko sprawdz backup na urzadzeniu") | |
| parser.add_argument("--restore", default=None, metavar="ZIP", help="Przywroc z lokalnego pliku ZIP") | |
| parser.add_argument("--show", default=None, metavar="ZIP", help="Pokaz kluczowe ustawienia z pliku ZIP") | |
| args = parser.parse_args() | |
| if args.show: | |
| p = Path(args.show) | |
| if not p.exists(): | |
| err("Plik nie istnieje: " + args.show) | |
| sys.exit(1) | |
| hdr("Zawartosc: " + p.name) | |
| v = BackupValidator.validate_local(p) | |
| BackupValidator.print_report(v, p.name) | |
| print_zip_summary(p) | |
| sys.exit(0) | |
| success = SmartTubeConfigManager.run_full( | |
| device = args.device, | |
| dry_run = args.dry_run, | |
| export_only = args.export_only, | |
| validate_only = args.validate, | |
| restore_zip = args.restore, | |
| ) | |
| sys.exit(0 if success else 1) | |
| if __name__ == "__main__": | |
| main() |
Użycie
python3 smarttube_config.py --device 192.168.1.3:5555
python3 smarttube_config.py --dry-run --device 192.168.1.3:5555
python3 smarttube_config.py --show /ścieżka/SmartTubeBackup.zip
python3 smarttube_config.py
--restore ~/.playbox_cache/smarttube_configs/20240101_120000/original/SmartTubeBackup_original.zip