import sys import subprocess import importlib.util import ctypes DEPENDENCIES = { "requests": "requests", "webview": "pywebview", "socks": "PySocks", } def console_process_count() -> int: if sys.platform != "win32": return 0 process_ids = (ctypes.c_ulong * 16)() return ctypes.windll.kernel32.GetConsoleProcessList(process_ids, len(process_ids)) def bootstrap_console() -> bool: """Hide an owned empty console immediately when dependencies already exist.""" if sys.platform != "win32": return False missing = any(importlib.util.find_spec(module) is None for module in DEPENDENCIES) kernel32 = ctypes.windll.kernel32 user32 = ctypes.windll.user32 console_window = kernel32.GetConsoleWindow() if missing: if not console_window and kernel32.AllocConsole(): kernel32.SetConsoleTitleW("BAS Loader - Installing dependencies") sys.stdout = open("CONOUT$", "w", encoding="utf-8", buffering=1) sys.stderr = open("CONOUT$", "w", encoding="utf-8", buffering=1) sys.stdin = open("CONIN$", "r", encoding="utf-8") return False # Do not hide a PowerShell/cmd console shared with the parent process. if console_window and console_process_count() <= 1: user32.ShowWindow(console_window, 0) return True return False BOOTSTRAP_CONSOLE_HIDDEN = bootstrap_console() def ensure_dependencies(): for module, package in DEPENDENCIES.items(): if importlib.util.find_spec(module) is not None: continue print(f"[INSTALL] {package}", flush=True) result = subprocess.run([ sys.executable, "-m", "pip", "install", "--disable-pip-version-check", package ]) if result.returncode != 0: raise RuntimeError( f"Failed to install dependency: {package}" ) print(f"[OK] {package} installed", flush=True) ensure_dependencies() import os import json import math import time import hashlib import threading import shutil import re import winreg import argparse import atexit from urllib.parse import quote, urljoin, urlparse from pathlib import Path from configparser import ConfigParser from xml.etree import ElementTree as ET from concurrent.futures import ThreadPoolExecutor, as_completed import requests import webview from zipfile import ZipFile LOG_CALLBACK = None PROGRESS_CALLBACK = None REQUEST_PROXIES = None REGISTRY_PATH = r"Software\BASLoader" CONSOLE_WAS_HIDDEN = BOOTSTRAP_CONSOLE_HIDDEN # ----------------- logging ----------------- def _log(level: str, msg: str): line = f"[{level}] {msg}" if sys.stdout is not None: print(line) if LOG_CALLBACK: LOG_CALLBACK(line) def log_info(msg: str): _log("INFO", msg) def log_warn(msg: str): _log("WARN", msg) def log_error(msg: str): _log("ERROR", msg) def log_fatal(msg: str): _log("FATAL", msg) # ----------------- progress bar ----------------- class ProgressBar: """Simple thread-safe text progress bar (Windows-friendly).""" def __init__(self, total_bytes: int, length: int = 40): self.total = total_bytes self.length = length self.downloaded = 0 self.lock = threading.Lock() self.start_time = time.time() self.last_line_length = 0 def _format_eta(self, seconds: float) -> str: seconds = int(max(0, seconds)) if seconds < 60: return f"{seconds}s" minutes, sec = divmod(seconds, 60) if minutes < 60: return f"{minutes}m {sec}s" hours, minutes = divmod(minutes, 60) return f"{hours}h {minutes}m" def update(self, delta: int): if self.total <= 0: return with self.lock: self.downloaded += delta now = time.time() elapsed = max(now - self.start_time, 1e-6) fraction = min(1.0, self.downloaded / self.total) filled = int(self.length * fraction) bar = "#" * filled + "-" * (self.length - filled) percent = int(fraction * 100) # speed speed_bps = self.downloaded / elapsed speed_mbps = speed_bps / (1024 * 1024) # ETA remaining = self.total - self.downloaded eta_seconds = remaining / speed_bps if speed_bps > 0 else 0 eta_str = self._format_eta(eta_seconds) line = ( f"[DOWNLOAD] |{bar}| {percent:3d}% " f"({self.downloaded}/{self.total} bytes) " f"{speed_mbps:5.2f} MB/s ETA {eta_str}" ) output = line.ljust(self.last_line_length) self.last_line_length = len(line) if sys.stdout is not None: sys.stdout.write("\r" + output) sys.stdout.flush() if PROGRESS_CALLBACK: PROGRESS_CALLBACK(percent, speed_mbps, eta_str) def finish(self): self.update(0) if sys.stdout is not None: sys.stdout.write("\n") sys.stdout.flush() def wait_for_exit(): try: input("\nPress Enter to exit...") except EOFError: pass def ensure_console(): """Show the existing console or allocate one after RUN.""" global CONSOLE_WAS_HIDDEN if os.name != "nt": return kernel32 = ctypes.windll.kernel32 user32 = ctypes.windll.user32 console_window = kernel32.GetConsoleWindow() if console_window: user32.ShowWindow(console_window, 5) user32.SetForegroundWindow(console_window) CONSOLE_WAS_HIDDEN = False return if not kernel32.AllocConsole(): return kernel32.SetConsoleTitleW("BAS Loader") sys.stdout = open("CONOUT$", "w", encoding="utf-8", buffering=1) sys.stderr = open("CONOUT$", "w", encoding="utf-8", buffering=1) sys.stdin = open("CONIN$", "r", encoding="utf-8") def restore_hidden_console(): if os.name != "nt" or not CONSOLE_WAS_HIDDEN: return console_window = ctypes.windll.kernel32.GetConsoleWindow() if console_window: ctypes.windll.user32.ShowWindow(console_window, 5) def hide_console_for_webview(): """Hide the empty console while the webview form is displayed.""" global CONSOLE_WAS_HIDDEN if os.name != "nt": return console_window = ctypes.windll.kernel32.GetConsoleWindow() if ( not console_window or console_process_count() > 1 or not ctypes.windll.user32.IsWindowVisible(console_window) ): return ctypes.windll.user32.ShowWindow(console_window, 0) CONSOLE_WAS_HIDDEN = True atexit.register(restore_hidden_console) # ----------------- helpers ----------------- def parse_proxy(proxy_string: str) -> dict: server = "" port = 0 name = "" password = "" detected_type = "" original = proxy_string if not proxy_string: raise ValueError("Proxy is empty") normalized = re.sub(r"[@\\/\s]", ":", proxy_string) parts = normalized.split(":") for raw_part in parts: lowered = raw_part.lower() if lowered in ("socks5a", "socks5", "socks"): detected_type = "socks5" continue if lowered in ("socks4a", "socks4"): detected_type = "socks4" continue if lowered in ("http", "https", "ftp"): detected_type = "http" continue if 0 < len(raw_part) <= 5 and raw_part.isdigit(): port = int(raw_part) continue if not server and ("." in raw_part or lowered == "localhost"): server = raw_part continue if raw_part: if not name: name = raw_part elif not password: password = raw_part if not server or not 1 <= port <= 65535: raise ValueError(f"Failed to parse proxy: '{original}'") return { "server": server, "port": port, "type": detected_type, "name": name, "password": password, } def configure_requests_proxy(proxy_string: str, proxy_type: str): global REQUEST_PROXIES parsed = parse_proxy(proxy_string) scheme = "socks5h" if proxy_type == "SOCKS 5" else "http" auth = "" if parsed["name"]: auth = quote(parsed["name"], safe="") if parsed["password"]: auth += ":" + quote(parsed["password"], safe="") auth += "@" proxy_url = f"{scheme}://{auth}{parsed['server']}:{parsed['port']}" REQUEST_PROXIES = {"http": proxy_url, "https": proxy_url} def load_gui_settings() -> tuple[bool, str, str]: try: with winreg.OpenKey(winreg.HKEY_CURRENT_USER, REGISTRY_PATH) as key: use_proxy = bool(winreg.QueryValueEx(key, "UseProxy")[0]) proxy_value = str(winreg.QueryValueEx(key, "ProxyValue")[0]) proxy_type = str(winreg.QueryValueEx(key, "ProxyType")[0]) return use_proxy, proxy_value, proxy_type except OSError: return False, "", "HTTP" def save_gui_settings(use_proxy: bool, proxy_value: str, proxy_type: str): with winreg.CreateKey(winreg.HKEY_CURRENT_USER, REGISTRY_PATH) as key: winreg.SetValueEx(key, "UseProxy", 0, winreg.REG_DWORD, int(use_proxy)) winreg.SetValueEx(key, "ProxyValue", 0, winreg.REG_SZ, proxy_value) winreg.SetValueEx(key, "ProxyType", 0, winreg.REG_SZ, proxy_type) def read_project_info(project_xml: Path) -> dict: log_info(f"Reading project info from: {project_xml}") tree = ET.parse(project_xml) root = tree.getroot() is_remote = False script_name = "" engine_version = "" # Read EngineVersion element if present for elem in root.iter(): if elem.tag == "EngineVersion" and elem.text: engine_version = elem.text.strip() log_info(f"EngineVersion: {engine_version}") break # Remote project: for elem in root.iter(): if elem.tag == "Remote": is_remote = True script_name = elem.attrib.get("ScriptName", "") server = elem.attrib.get("Server", "") username = elem.attrib.get("Username", "") password = elem.attrib.get("Password", "") log_info( f"Remote project.xml ScriptName={script_name} " f"Server={server} Username={username} Password={password}" ) return { "IsRemote": True, "ScriptName": script_name, "Server": server, "Username": username, "Password": password, "EngineVersion": engine_version, } # Local project: ScriptName on root or default script_name = root.attrib.get("ScriptName", "") if not script_name: log_info("ScriptName is empty in local project, will use hash of default name") script_name = "data/project.xml" log_info(f"Local project ScriptName: {script_name}") return { "IsRemote": False, "ScriptName": script_name, "Server": "", "Username": "", "Password": "", "EngineVersion": engine_version, } def read_remote_settings(ini_path: Path) -> dict: log_info(f"Reading settings from: {ini_path}") config = ConfigParser() text = ini_path.read_text(encoding="utf-8") if "[" not in text: text = "[DEFAULT]\n" + text config.read_string(text) is_engines_in_appdata = config.getboolean("DEFAULT", "IsEnginesInAppData", fallback=False) keep_version_number = config.getint("DEFAULT", "KeepVersionNumber", fallback=0) log_info(f"IsEnginesInAppData = {is_engines_in_appdata}") log_info(f"KeepVersionNumber = {keep_version_number}") return { "IsEnginesInAppData": is_engines_in_appdata, "KeepVersionNumber": keep_version_number, } def read_bas_api_endpoint(ini_path: Path) -> str: log_info(f"Reading BAS settings from: {ini_path}") config = ConfigParser() text = ini_path.read_text(encoding="utf-8-sig") if "[" not in text: text = "[DEFAULT]\n" + text config.read_string(text) api_endpoint = config.get("DEFAULT", "ApiEndpoint", fallback="").strip() if not api_endpoint: raise RuntimeError(f"ApiEndpoint not found in {ini_path}") return api_endpoint def http_get_json(url: str) -> dict: log_info(f"GET JSON: {url}") resp = requests.get(url, timeout=30, verify=False, proxies=REQUEST_PROXIES) resp.raise_for_status() return resp.json() def http_get_range( url: str, start: int, end: int, session: requests.Session, allow_full_response: bool = False, ) -> bytes: headers = {"Range": f"bytes={start}-{end}"} resp = session.get( url, headers=headers, timeout=120, verify=False, proxies=REQUEST_PROXIES, ) allowed_statuses = (200, 206) if allow_full_response else (206,) if resp.status_code not in allowed_statuses: raise RuntimeError(f"Bad status code {resp.status_code} for range {start}-{end}") data = resp.content expected_size = end - start + 1 if len(data) != expected_size: raise RuntimeError( f"Bad response size for range {start}-{end}: " f"expected {expected_size}, got {len(data)}" ) return data def sha1_bytes(data: bytes) -> str: h = hashlib.sha1() h.update(data) return h.hexdigest() def sha1_file(path: Path, chunk_size: int = 1024 * 1024) -> str: h = hashlib.sha1() with path.open("rb") as f: while True: chunk = f.read(chunk_size) if not chunk: break h.update(chunk) return h.hexdigest() def verify_existing_zip(path: Path, expected_size: int, expected_sha1: str) -> bool: """ Check if ZIP at 'path' exists, matches expected size and SHA1. Returns True if valid, False otherwise. """ if not path.is_file(): return False if path.stat().st_size != expected_size: log_warn(f"{path} size mismatch, ignoring it") return False log_info(f"Verifying SHA1 of {path}...") current_sha1 = sha1_file(path) log_info(f"{path} SHA1: {current_sha1}") if current_sha1.lower() != expected_sha1.lower(): log_warn(f"{path} SHA1 mismatch, ignoring it") return False log_info(f"{path} matches expected SHA1") return True def build_script_hash(engversion: str) -> str: sha = hashlib.sha256(engversion.encode("utf-8")).hexdigest() return "SID" + sha[:8] def build_local_script_name_hash(script_name: str) -> str: sha = hashlib.sha256(script_name.encode("utf-8")).hexdigest() return sha[:8] def get_filename_from_url(url: str) -> str: parsed = urlparse(url) return Path(parsed.path).name def version_key(version: str) -> tuple: """Build a sortable key for dotted version directory names.""" parts = re.findall(r"\d+|[^\d]+", version) return tuple((0, int(part)) if part.isdigit() else (1, part.lower()) for part in parts) def find_previous_version(apps_dir: Path, current_version: str) -> Path | None: candidates = [ path for path in apps_dir.iterdir() if path.is_dir() and path.name != current_version and not path.name.startswith(".") and version_key(path.name) < version_key(current_version) ] if apps_dir.is_dir() else [] return max(candidates, key=lambda path: version_key(path.name), default=None) def _read_text_lines(path: Path) -> list[str]: try: return path.read_text(encoding="utf-8-sig").splitlines(keepends=True) except UnicodeDecodeError: return path.read_text(encoding="cp1251").splitlines(keepends=True) def _scan_ini(lines: list[str]) -> tuple[dict[tuple[str, str], int], dict[str, tuple[int, int]]]: entries: dict[tuple[str, str], int] = {} section_headers: dict[str, int] = {} section = "" for index, line in enumerate(lines): stripped = line.strip() section_match = re.match(r"^\[([^]]+)]", stripped) if section_match: section = section_match.group(1).strip().casefold() section_headers[section] = index continue if not stripped or stripped.startswith(("#", ";")): continue key_match = re.match(r"^\s*([^#;][^=:]*?)\s*[=:]", line) if not key_match: continue key = key_match.group(1).strip().casefold() if key: entries[(section, key)] = index sections: dict[str, tuple[int, int]] = {} ordered = sorted(section_headers.items(), key=lambda item: item[1]) first_header = ordered[0][1] if ordered else len(lines) sections[""] = (-1, first_header) for position, (name, header_index) in enumerate(ordered): end = ordered[position + 1][1] if position + 1 < len(ordered) else len(lines) sections[name] = (header_index, end) return entries, sections def _original_section_name(lines: list[str], entry_index: int) -> str: for index in range(entry_index - 1, -1, -1): stripped = lines[index].strip() section_match = re.match(r"^\[([^]]+)]", stripped) if section_match: return section_match.group(1).strip() return "" def merge_ini_values(old_path: Path, new_path: Path): """Keep new-version keys, overlay old values, and add old-only keys.""" if not new_path.is_file(): shutil.copy2(old_path, new_path) log_info(f"Copied INI absent in new version: {old_path.name}") return old_lines = _read_text_lines(old_path) new_lines = _read_text_lines(new_path) old_entries, _ = _scan_ini(old_lines) replaced = 0 added = 0 for (section, key), old_index in old_entries.items(): new_entries, new_sections = _scan_ini(new_lines) existing_index = new_entries.get((section, key)) old_line = old_lines[old_index] if existing_index is not None: newline = "\r\n" if new_lines[existing_index].endswith("\r\n") else "\n" new_lines[existing_index] = old_line.rstrip("\r\n") + newline replaced += 1 continue if section in new_sections: insert_at = new_sections[section][1] new_lines.insert(insert_at, old_line if old_line.endswith(("\n", "\r")) else old_line + "\n") else: if new_lines and new_lines[-1].strip(): new_lines.append("\n") original_section = _original_section_name(old_lines, old_index) new_lines.extend([ f"[{original_section}]\n", old_line if old_line.endswith(("\n", "\r")) else old_line + "\n", ]) added += 1 new_path.write_text("".join(new_lines), encoding="utf-8", newline="") log_info(f"Merged INI {old_path.name}: values={replaced}, added={added}") def migrate_previous_version(previous_dir: Path | None, version_dir: Path): if previous_dir is None: log_info("No previous BAS version found, migration skipped") return log_info(f"Migrating user data from BAS {previous_dir.name}") old_custom = previous_dir / "custom" if old_custom.is_dir(): shutil.copytree(old_custom, version_dir / "custom", dirs_exist_ok=True) log_info("Custom folder copied") for old_ini in sorted(previous_dir.glob("*.ini")): merge_ini_values(old_ini, version_dir / old_ini.name) def launch_bas(base_dir: Path, bas_root: Path): candidates = [ base_dir / "BrowserAutomationStudio.exe", bas_root / "BrowserAutomationStudio.exe", ] executable = next((path for path in candidates if path.is_file()), None) if executable is None: checked = ", ".join(str(path) for path in candidates) raise RuntimeError(f"BrowserAutomationStudio.exe not found. Checked: {checked}") log_info(f"Starting BrowserAutomationStudio: {executable}") subprocess.Popen([str(executable)], cwd=str(executable.parent)) def launch_project_script(target_dir: Path): executables = [ exe for exe in target_dir.glob("*.exe") if exe.name.lower() != "remoteexecutescriptsilent.exe" ] if not executables: raise RuntimeError(f"Project executable not found in: {target_dir}") executable = executables[0] log_info(f"Starting project script: {executable}") subprocess.Popen([str(executable)], cwd=str(target_dir)) def application_base_dir() -> Path: if getattr(sys, "frozen", False): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parent # ----------------- chunk downloader ----------------- class ChunkDownloader: def __init__( self, url: str, total_size: int, chunk_size: int, expected_chunks_sha1: list[str] | None = None, max_workers: int = 10, max_retries: int = 5, ): self.url = url self.total_size = total_size self.chunk_size = chunk_size self.expected_chunks_sha1 = expected_chunks_sha1 or [] self.max_workers = max_workers self.max_retries = max_retries self.num_chunks = math.ceil(total_size / chunk_size) if self.expected_chunks_sha1 and self.num_chunks != len(self.expected_chunks_sha1): log_warn( f"Number of chunks ({self.num_chunks}) " f"does not match sha1 list ({len(self.expected_chunks_sha1)})." ) self.chunks_data: list[bytes | None] = [None] * self.num_chunks self.lock = threading.Lock() self.progress = ProgressBar(total_bytes=total_size) def _download_single_chunk(self, index: int): with requests.Session() as session: start = index * self.chunk_size end = min(self.total_size - 1, (index + 1) * self.chunk_size - 1) expected_sha1 = ( self.expected_chunks_sha1[index] if index < len(self.expected_chunks_sha1) else None ) last_error: Exception | None = None for attempt in range(1, self.max_retries + 1): try: data = http_get_range( self.url, start, end, session, allow_full_response=self.num_chunks == 1, ) if expected_sha1: actual_sha1 = sha1_bytes(data) if actual_sha1.lower() != expected_sha1.lower(): last_error = RuntimeError( f"Chunk {index} SHA1 mismatch " f"(expected {expected_sha1}, got {actual_sha1})" ) continue with self.lock: self.chunks_data[index] = data self.progress.update(len(data)) return except Exception as e: last_error = e log_error( f"Chunk {index} failed after {self.max_retries} attempts: {last_error}" ) raise RuntimeError(f"Chunk {index} failed after {self.max_retries} attempts") def download_all(self): log_info(f"Starting multi-thread download, chunks: {self.num_chunks}, workers: {self.max_workers}") try: with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = [ executor.submit(self._download_single_chunk, idx) for idx in range(self.num_chunks) ] for f in as_completed(futures): f.result() self.progress.finish() log_info("All chunks downloaded successfully") except Exception: self.progress.finish() raise def save_to_file(self, out_path: Path): log_info(f"Writing combined ZIP to: {out_path}") with out_path.open("wb") as f: for i, data in enumerate(self.chunks_data): if data is None: raise RuntimeError(f"Missing data for chunk {i}") f.write(data) log_info("Combined ZIP file written") def download_missing_bas_version(base_dir: Path) -> bool: """ Download the current BAS application archive when its version directory is absent. Returns True when this BAS-specific branch handled the run. """ roaming_appdata = os.environ.get("APPDATA", "") if not roaming_appdata: log_warn("APPDATA is not set, skipping BAS application check") return False settings_path = Path(roaming_appdata) / "BrowserAutomationStudio" / "settings.ini" if not settings_path.is_file(): log_info(f"BAS settings not found: {settings_path}") return False api_endpoint = read_bas_api_endpoint(settings_path) latest_json = http_get_json(api_endpoint) if not latest_json.get("success", True): raise RuntimeError(latest_json.get("message") or "BAS latest endpoint returned an error") version = str(latest_json.get("version", "")).strip() script = str(latest_json.get("script", "")).strip() if not version or not script: raise RuntimeError("version or script not found in BAS latest response") apps_dir = settings_path.parent / "apps" version_dir = apps_dir / version log_info(f"Checking BAS version directory: {version_dir}") if version_dir.is_dir(): log_info(f"BAS version {version} is already installed") launch_bas(base_dir, settings_path.parent) return True previous_dir = find_previous_version(apps_dir, version) if previous_dir: log_info(f"Previous BAS version: {previous_dir.name}") download_url = urljoin(api_endpoint, "/" + script.lstrip("/")) log_info(f"BAS version {version} is missing") log_info(f"HEAD: {download_url}") head = requests.head( download_url, allow_redirects=True, timeout=30, verify=False, proxies=REQUEST_PROXIES, ) head.raise_for_status() content_length = head.headers.get("Content-Length") if not content_length: raise RuntimeError("Content-Length not found in HEAD response") total_size = int(content_length) if total_size <= 0: raise RuntimeError(f"Invalid Content-Length: {content_length}") archive_name = get_filename_from_url(head.url) or get_filename_from_url(download_url) if not archive_name: archive_name = f"BrowserAutomationStudio-{version}.zip" temp_value = os.environ.get("TEMP", "").strip() if temp_value: temp_root = Path(temp_value) else: local_appdata = os.environ.get("LOCALAPPDATA", "").strip() if not local_appdata: raise RuntimeError("TEMP and LOCALAPPDATA environment variables are not set") temp_root = Path(local_appdata) / "Temp" temp_root.mkdir(parents=True, exist_ok=True) archive_path = temp_root / archive_name log_info(f"BAS archive size: {total_size} bytes") log_info(f"BAS archive path: {archive_path}") if archive_path.is_file() and archive_path.stat().st_size == total_size: log_info("Archive with matching size already exists, skipping download") else: downloader = ChunkDownloader( url=head.url, total_size=total_size, chunk_size=10 * 1024 * 1024, max_workers=10, max_retries=10, ) downloader.download_all() downloader.save_to_file(archive_path) log_info(f"BAS archive downloaded successfully: {archive_path}") log_info(f"Creating BAS version directory: {version_dir}") version_dir.mkdir(parents=True, exist_ok=False) try: log_info(f"Extracting BAS archive to: {version_dir}") with ZipFile(archive_path, "r") as zf: zf.extractall(version_dir) migrate_previous_version(previous_dir, version_dir) except Exception: shutil.rmtree(version_dir, ignore_errors=True) raise log_info(f"BAS version {version} installed successfully") try: archive_path.unlink() log_info(f"Temporary BAS archive removed: {archive_path}") except OSError as cleanup_error: log_warn(f"Could not remove temporary BAS archive: {cleanup_error}") launch_bas(base_dir, settings_path.parent) return True # ----------------- main logic ----------------- def main(): base_dir = application_base_dir() data_dir = base_dir / "data" project_xml = data_dir / "project.xml" remote_ini = data_dir / "remote_settings.ini" # MODE 1 has priority. MODE 2 is used only when this is not a project loader # directory and project.xml cannot be read. try: info = read_project_info(project_xml) log_info("MODE 1: project engine loader") except Exception as project_error: log_warn(f"MODE 1 unavailable: {project_error}") log_info("MODE 2: BAS application updater") if download_missing_bas_version(base_dir): return raise RuntimeError( "Neither mode is available: project.xml could not be read and " "BAS settings.ini was not found" ) from project_error script_name = info["ScriptName"] is_remote = info["IsRemote"] engine_version_from_xml = info["EngineVersion"] # 1.1 Remote settings settings = read_remote_settings(remote_ini) is_engines_in_appdata = settings["IsEnginesInAppData"] # 2. Engine version if is_remote: props_url = f"https://bablosoft.com/scripts/{script_name}/properties" props_json = http_get_json(props_url) engversion = props_json.get("engversion") if not engversion: raise RuntimeError("engversion not found in properties response") log_info(f"Engine version (remote): {engversion}") else: engversion = engine_version_from_xml if not engversion: raise RuntimeError("EngineVersion not found in local project.xml") log_info(f"Engine version (local): {engversion}") # 2.1 Hashes for paths script_hash = build_script_hash(engversion) if not is_remote: local_script_hash = build_local_script_name_hash("") else: local_script_hash = None # 3. Meta URL and JSON if is_remote: meta_url = ( f"https://bablosoft.com/distr/FastExecuteScriptProtected64/" f"{engversion}/FastExecuteScriptProtected.x64.zip.meta.json" ) else: meta_url = ( f"https://bablosoft.com/distr/FastExecuteScript64/" f"{engversion}/FastExecuteScript.x64.zip.meta.json" ) log_info(f"Meta URL: {meta_url}") meta_json = http_get_json(meta_url) total_size = int(meta_json["TotalSize"]) expected_zip_sha1 = meta_json["Checksum"] chunks_sha1 = meta_json["Chunks"] file_url = meta_json["Url"] log_info(f"Total size: {total_size} bytes") log_info(f"Chunks count: {len(chunks_sha1)}") # 4. Temp directory in LOCALAPPDATA\Temp local_appdata = Path(os.environ.get("LOCALAPPDATA", "")) if not local_appdata: raise RuntimeError("LOCALAPPDATA environment variable is not set") temp_root = local_appdata / "Temp" temp_root.mkdir(parents=True, exist_ok=True) zip_filename = get_filename_from_url(file_url) zip_path = temp_root / f"{zip_filename}.{engversion}" log_info(f"Temp ZIP path: {zip_path}") # Engine storage location in AppData (for reuse) appdata = Path(os.environ.get("LOCALAPPDATA", "")) if not appdata: raise RuntimeError("LOCALAPPDATA environment variable is not set") engines_root = appdata / "BasEngines" if is_remote: engines_dir = engines_root / "enginesprotected" / engversion else: engines_dir = engines_root / "engines" / engversion engine_zip_path = engines_dir / "engine.zip" # 4.1 Reuse ZIP from BasEngines or Temp if valid # 4.1.1 Try BasEngines\...\engine.zip (only when IsEnginesInAppData = true) if is_engines_in_appdata: if verify_existing_zip(engine_zip_path, total_size, expected_zip_sha1): log_info("Reusing engine.zip from BasEngines (copying to temp)") temp_root.mkdir(parents=True, exist_ok=True) zip_path.write_bytes(engine_zip_path.read_bytes()) # 4.1.2 Try Temp ZIP if zip_path.is_file() and not verify_existing_zip(zip_path, total_size, expected_zip_sha1): # invalid temp ZIP zip_path.unlink() # 4.2 Download if no valid ZIP if not zip_path.is_file(): log_info("No valid ZIP found, starting multi-threaded download") downloader = ChunkDownloader( url=file_url, total_size=total_size, chunk_size=10 * 1024 * 1024, expected_chunks_sha1=chunks_sha1, max_workers=10, max_retries=10, ) downloader.download_all() downloader.save_to_file(zip_path) log_info("Calculating SHA1 of combined ZIP...") zip_sha1 = sha1_file(zip_path) log_info(f"ZIP SHA1: {zip_sha1}") if zip_sha1.lower() != expected_zip_sha1.lower(): raise RuntimeError( f"ZIP SHA1 mismatch (expected {expected_zip_sha1}, got {zip_sha1})" ) log_info("ZIP SHA1 verified successfully") else: log_info("Using existing ZIP (BasEngines or Temp)") # 5. Store engine.zip in AppData if needed if is_engines_in_appdata: if is_remote: engines_dir = local_appdata / "BasEngines" / "enginesprotected" / engversion else: engines_dir = local_appdata / "BasEngines" / "engines" / engversion engines_dir.mkdir(parents=True, exist_ok=True) engine_zip_path = engines_dir / "engine.zip" log_info(f"Copying ZIP to: {engine_zip_path}") engine_zip_path.write_bytes(zip_path.read_bytes()) else: log_info("IsEnginesInAppData = false, skipping engine.zip copy to AppData") # 6. Extract engine into application folder if is_remote: target_dir = base_dir / "appsremote" / script_name / script_hash / "engine" else: target_dir = base_dir / "appslocal" / local_script_hash / script_hash / "engine" target_dir.mkdir(parents=True, exist_ok=True) # Copy local project.xml into target_dir for non-remote projects if not is_remote: source_project_xml = data_dir / "project.xml" dest_project_xml = target_dir / "project.xml" log_info(f"Copying local project.xml to: {dest_project_xml}") dest_project_xml.write_bytes(source_project_xml.read_bytes()) log_info(f"Extracting ZIP to: {target_dir}") with ZipFile(zip_path, "r") as zf: zf.extractall(target_dir) log_info("Extraction completed successfully") launch_project_script(target_dir) # ----------------- pywebview GUI ----------------- WEBVIEW_WINDOW = None WEBVIEW_HTML = r"""

BAS UPDATER & LAUNCHER

""" def webview_call(function_name: str, *args): if WEBVIEW_WINDOW is None: return encoded = ",".join(json.dumps(arg, ensure_ascii=False) for arg in args) try: WEBVIEW_WINDOW.evaluate_js(f"window.{function_name}({encoded})") except Exception: pass class WebviewApi: def __init__(self): self.running = False self.worker_thread = None self.error = None def get_settings(self): use_proxy, proxy, proxy_type = load_gui_settings() if proxy_type not in ("HTTP", "SOCKS 5"): proxy_type = "HTTP" return {"use_proxy": use_proxy, "proxy": proxy, "proxy_type": proxy_type} def start_run(self, options): global LOG_CALLBACK, PROGRESS_CALLBACK, REQUEST_PROXIES if self.running: return {"ok": False, "error": "Loader is already running"} try: use_proxy = bool(options.get("use_proxy")) proxy = str(options.get("proxy", "")).strip() proxy_type = str(options.get("proxy_type", "HTTP")) if proxy_type not in ("HTTP", "SOCKS 5"): raise ValueError("Unsupported proxy type") if use_proxy: configure_requests_proxy(proxy, proxy_type) else: REQUEST_PROXIES = None save_gui_settings(use_proxy, proxy, proxy_type) except Exception as error: return {"ok": False, "error": str(error)} # RUN switches from the web UI to the CLI. All following logs and the # existing progress bar are written directly to the console. ensure_console() LOG_CALLBACK = None PROGRESS_CALLBACK = None self.running = True self.worker_thread = threading.Thread(target=self._worker, daemon=False) self.worker_thread.start() threading.Timer(0.15, self._close_window).start() return {"ok": True} def _close_window(self): try: if WEBVIEW_WINDOW is not None: WEBVIEW_WINDOW.destroy() except Exception: pass def _worker(self): try: main() log_info("Done") except Exception as error: log_fatal(str(error)) self.error = error self.running = False return self.running = False def run_webview(): global WEBVIEW_WINDOW api = WebviewApi() width, height = 560, 300 screen = webview.screens[0] WEBVIEW_WINDOW = webview.create_window( "BAS Updater & Launcher", html=WEBVIEW_HTML, js_api=api, width=width, height=height, x=(screen.width - width) // 2, y=(screen.height - height) // 2, resizable=False, background_color="#0d0f14", ) webview.start(debug=False) if api.worker_thread is not None: api.worker_thread.join() return api.error def run_from_arguments(): """Run in console mode with an optional positional proxy.""" parser = argparse.ArgumentParser(description="BAS loader") parser.add_argument( "proxy", nargs="?", help="Proxy in ip:port, login:password@ip:port or similar format", ) parser.add_argument( "--proxy-type", choices=("http", "socks5"), default="http", help="Proxy protocol (default: http)", ) parser.add_argument( "--run", action="store_true", help="Run without GUI and without a proxy", ) args = parser.parse_args() if args.proxy: selected_type = "SOCKS 5" if args.proxy_type == "socks5" else "HTTP" configure_requests_proxy(args.proxy, selected_type) save_gui_settings(True, args.proxy, selected_type) log_info(f"Command-line proxy enabled: {selected_type}") else: global REQUEST_PROXIES REQUEST_PROXIES = None log_info("Command-line mode without proxy") main() log_info("Done") if __name__ == "__main__": try: if len(sys.argv) > 1: run_from_arguments() else: hide_console_for_webview() gui_error = run_webview() if gui_error is not None: wait_for_exit() sys.exit(1) except Exception as error: ensure_console() log_fatal(str(error)) wait_for_exit() sys.exit(1)