libportable.py

libportable.py

"""
portablelib.py — Python port of the portapps Go library.
Windows-specific features (registry, msgbox, mutex, console, locale, env)
use ctypes/winreg and are no-ops / raise NotImplementedError on non-Windows.
"""

import ctypes
import json
import logging
import os
import platform
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import requests
import yaml

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------

logging.basicConfig(format="%(asctime)s [%(levelname)s] %(message)s", level=logging.DEBUG)
log = logging.getLogger("portapps")

_IS_WINDOWS = platform.system() == "Windows"

# ---------------------------------------------------------------------------
# pkg/win
# ---------------------------------------------------------------------------

@dataclass
class WinVersion:
    major: int = 0
    minor: int = 0
    build: int = 0


def get_win_version() -> WinVersion:
    if not _IS_WINDOWS:
        return WinVersion()
    ver = sys.getwindowsversion()
    return WinVersion(major=ver.major, minor=ver.minor, build=ver.build)


# --- msgbox constants (mirrors win/msgbox.go) ---
MB_OK                = 0x000000
MB_OKCANCEL          = 0x000001
MB_ABORTRETRYIGNORE  = 0x000002
MB_YESNOCANCEL       = 0x000003
MB_YESNO             = 0x000004
MB_RETRYCANCEL       = 0x000005
MB_CANCELTRYCONTINUE = 0x000006
MB_ICONERROR         = 0x000010
MB_ICONQUESTION      = 0x000020
MB_ICONWARNING       = 0x000030
MB_ICONINFORMATION   = 0x000040
MB_TOPMOST           = 0x041000

IDOK       = 1
IDCANCEL   = 2
IDABORT    = 3
IDRETRY    = 4
IDIGNORE   = 5
IDYES      = 6
IDNO       = 7
IDTRYAGAIN = 10
IDCONTINUE = 11


def msg_box(title: str, message: str, flags: int = MB_OK) -> int:
    if _IS_WINDOWS:
        return ctypes.windll.user32.MessageBoxW(0, message, title, flags)
    print(f"[{title}] {message}")
    return IDOK


def get_console_title() -> str:
    if _IS_WINDOWS:
        buf = ctypes.create_unicode_buffer(256)
        ctypes.windll.kernel32.GetConsoleTitleW(buf, len(buf))
        return buf.value
    return ""


def set_console_title(title: str) -> None:
    if _IS_WINDOWS:
        ctypes.windll.kernel32.SetConsoleTitleW(title)


def locale() -> str:
    if _IS_WINDOWS:
        buf = ctypes.create_unicode_buffer(128)
        ret = ctypes.windll.kernel32.GetUserDefaultLocaleName(buf, len(buf))
        return buf.value if ret else "en-US"
    return "en-US"


def refresh_env() -> None:
    """Broadcast WM_SETTINGCHANGE so new env vars are picked up system-wide."""
    if not _IS_WINDOWS:
        return
    HWND_BROADCAST = 0xFFFF
    WM_SETTINGCHANGE = 0x001A
    SMTO_ABORTIFHUNG = 0x0002
    result = ctypes.c_long()
    ctypes.windll.user32.SendMessageTimeoutW(
        HWND_BROADCAST, WM_SETTINGCHANGE, 0,
        "Environment", SMTO_ABORTIFHUNG, 5000,
        ctypes.byref(result),
    )


def _open_env_key(hive_flag: int):
    import winreg
    return winreg.OpenKey(hive_flag, "Environment", 0, winreg.KEY_ALL_ACCESS)


def set_perm_env(hive_flag: int, name: str, value: str) -> None:
    if not _IS_WINDOWS:
        raise NotImplementedError
    import winreg
    with _open_env_key(hive_flag) as key:
        winreg.SetValueEx(key, name, 0, winreg.REG_SZ, value)


def delete_perm_env(hive_flag: int, name: str) -> None:
    if not _IS_WINDOWS:
        raise NotImplementedError
    import winreg
    with _open_env_key(hive_flag) as key:
        winreg.DeleteValue(key, name)


def get_perm_env(hive_flag: int, name: str) -> str:
    if not _IS_WINDOWS:
        raise NotImplementedError
    import winreg
    with _open_env_key(hive_flag) as key:
        value, _ = winreg.QueryValueEx(key, name)
        return value


# ---------------------------------------------------------------------------
# pkg/mutex
# ---------------------------------------------------------------------------

def create_mutex(name: str):
    """Returns a mutex handle (Windows) or None. Raises RuntimeError if already running."""
    if not _IS_WINDOWS:
        return None
    full = f"Portapps{name}"
    handle = ctypes.windll.kernel32.OpenMutexW(0x1F0001, False, full)
    if handle:
        ctypes.windll.kernel32.CloseHandle(handle)
        raise RuntimeError("already running")
    return ctypes.windll.kernel32.CreateMutexW(None, False, full)


def release_mutex(handle) -> None:
    if _IS_WINDOWS and handle:
        ctypes.windll.kernel32.CloseHandle(handle)


# ---------------------------------------------------------------------------
# pkg/proc
# ---------------------------------------------------------------------------

@dataclass
class CmdOptions:
    command: str
    args: list[str] = field(default_factory=list)
    working_dir: str = ""
    hide_window: bool = False


@dataclass
class CmdResult:
    options: CmdOptions
    exit_code: int = 0
    stdout: str = ""
    stderr: str = ""


def cmd(options: CmdOptions) -> CmdResult:
    kwargs: dict[str, Any] = {
        "capture_output": True,
        "text": True,
        "cwd": options.working_dir or None,
    }
    if _IS_WINDOWS and options.hide_window:
        si = subprocess.STARTUPINFO()
        si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
        si.wShowWindow = 0
        kwargs["startupinfo"] = si

    proc = subprocess.run([options.command] + options.args, **kwargs)
    return CmdResult(
        options=options,
        exit_code=proc.returncode,
        stdout=proc.stdout.strip(),
        stderr=proc.stderr.strip(),
    )


def quick_cmd(command: str, args: list[str]) -> None:
    result = cmd(CmdOptions(command=command, args=args, hide_window=True))
    if result.exit_code != 0:
        msg = f"exit code {result.exit_code}"
        if result.stderr:
            msg += f"\n{result.stderr}"
        raise RuntimeError(msg)


# ---------------------------------------------------------------------------
# pkg/registry
# ---------------------------------------------------------------------------

MAX_BACKUP = 19


class Key:
    """Windows registry key wrapper (mirrors registry/key.go)."""

    def __init__(self, key: str, arch: str = "64", default: str = ""):
        self.key = key
        self.arch = arch
        self.default = default

    def _reg_args(self, *extra) -> list[str]:
        return [*extra, self.key, f"/reg:{self.arch}"]

    def add(self, force: bool = False) -> None:
        args = self._reg_args("add")
        if self.default:
            args += ["/d", self.default]
        if force:
            args.append("/f")
        quick_cmd("reg", args)

    def delete(self, force: bool = False) -> None:
        args = self._reg_args("delete")
        if force:
            args.append("/f")
        quick_cmd("reg", args)

    def exists(self) -> bool:
        result = cmd(CmdOptions("reg", self._reg_args("query"), hide_window=True))
        return result.exit_code == 0

    def export(self, file: str) -> None:
        if not self.exists():
            return
        quick_cmd("reg", ["export", self.key, file, "/y", f"/reg:{self.arch}"])
        reg_dir = Path(file).parent
        reg_files = sorted(p for p in reg_dir.iterdir() if p.suffix != ".reg" and p.is_file())
        while len(reg_files) > MAX_BACKUP:
            reg_files[0].unlink()
            reg_files = reg_files[1:]

    def import_reg(self, file: str) -> None:
        import time
        self.export(f"{file}.{time.strftime('%Y%m%d%H%M%S')}")
        if not Path(file).exists():
            raise FileNotFoundError(f"reg file {file} not found")
        quick_cmd("reg", ["import", file, f"/reg:{self.arch}"])

    def open(self):
        if not _IS_WINDOWS:
            raise NotImplementedError
        import winreg
        hives = {
            "HKCR": winreg.HKEY_CLASSES_ROOT,
            "HKCU": winreg.HKEY_CURRENT_USER,
            "HKLM": winreg.HKEY_LOCAL_MACHINE,
            "HKU":  winreg.HKEY_USERS,
            "HKCC": winreg.HKEY_CURRENT_CONFIG,
        }
        hive, subkey = self.key.split("\\", 1)
        if hive not in hives:
            raise ValueError(f"unknown hive {hive}")
        return winreg.OpenKey(hives[hive], subkey, 0, winreg.KEY_ALL_ACCESS)


# ---------------------------------------------------------------------------
# pkg/shortcut
# ---------------------------------------------------------------------------

@dataclass
class ShortcutProperty:
    value: str = ""
    clear: bool = False


@dataclass
class Shortcut:
    shortcut_path: str = ""
    target_path: str = ""
    arguments: ShortcutProperty = field(default_factory=ShortcutProperty)
    description: ShortcutProperty = field(default_factory=ShortcutProperty)
    icon_location: ShortcutProperty = field(default_factory=ShortcutProperty)
    working_directory: ShortcutProperty = field(default_factory=ShortcutProperty)


def create_shortcut(shortcut: Shortcut) -> None:
    if not _IS_WINDOWS:
        raise NotImplementedError
    import pythoncom
    import win32com.client
    pythoncom.CoInitialize()
    try:
        shell = win32com.client.Dispatch("WScript.Shell")
        lnk = shell.CreateShortcut(shortcut.shortcut_path)
        lnk.TargetPath = shortcut.target_path
        if shortcut.arguments.value or shortcut.arguments.clear:
            lnk.Arguments = shortcut.arguments.value
        if shortcut.description.value or shortcut.description.clear:
            lnk.Description = shortcut.description.value
        if shortcut.icon_location.value or shortcut.icon_location.clear:
            lnk.IconLocation = shortcut.icon_location.value
        if shortcut.working_directory.value or shortcut.working_directory.clear:
            lnk.WorkingDirectory = shortcut.working_directory.value
        lnk.Save()
    finally:
        pythoncom.CoUninitialize()


# ---------------------------------------------------------------------------
# pkg/utl
# ---------------------------------------------------------------------------

def set_file_attributes(path: str, attrs: int) -> None:
    """Set Windows file attributes."""
    if not _IS_WINDOWS:
        raise NotImplementedError
    ctypes.windll.kernel32.SetFileAttributesW(path, attrs)


def copy_file(src: str, dest: str) -> None:
    shutil.copy2(src, dest)


def copy_folder(source: str, dest: str) -> None:
    shutil.copytree(source, dest, dirs_exist_ok=True)


def remove_contents(directory: str) -> None:
    for item in Path(directory).iterdir():
        if item.is_dir():
            shutil.rmtree(item)
        else:
            item.unlink()


def create_folder(*parts: str) -> str:
    folder = Path(*parts)
    folder.mkdir(parents=True, exist_ok=True)
    return str(folder)


def create_file(path: str, content: str) -> None:
    Path(path).write_text(content)


def format_unix_path(path: str) -> str:
    return path.replace("\\", "/")


def format_windows_path(path: str) -> str:
    return path.replace("/", "\\")


def exists(name: str) -> bool:
    return Path(name).exists()


def write_to_file(name: str, content: str) -> None:
    Path(name).write_text(content)


def append_to_file(name: str, content: str) -> None:
    with open(name, "a") as f:
        f.write(content)


def file_contains(name: str, text: str) -> bool:
    return text in Path(name).read_text()


def replace_by_prefix(filename: str, prefix: str, replace: str) -> None:
    lines = Path(filename).read_text().splitlines()
    lines = [replace if line.startswith(prefix) else line for line in lines]
    Path(filename).write_text("\n".join(lines))


def replace_in_file(filename: str, old: str, new: str) -> None:
    Path(filename).write_text(Path(filename).read_text().replace(old, new))


def is_dir_empty(name: str) -> bool:
    return not any(Path(name).iterdir())


def roaming_path() -> str:
    return os.environ.get("APPDATA", "")


def start_menu_path() -> str:
    return str(Path(roaming_path()) / "Microsoft" / "Windows" / "Start Menu" / "Programs")


def cleanup(folders: list[str]) -> None:
    for folder in folders:
        try:
            shutil.rmtree(folder, ignore_errors=True)
        except Exception as e:
            log.error("Cannot cleanup %s: %s", folder, e)


def download_file(filepath: str, url: str) -> None:
    with requests.get(url, timeout=30, stream=True) as r:
        r.raise_for_status()
        with open(filepath, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)


def find_electron_app_folder(prefix: str, source: str) -> str:
    for entry in Path(source).iterdir():
        if entry.is_dir() and entry.name.startswith(prefix):
            return entry.name
    raise FileNotFoundError(f"Electron main path does not exist with prefix '{prefix}' in {source}")


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

@dataclass
class CommonConfig:
    disable_log: bool = False
    args: list[str] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    app_path: str = ""


@dataclass
class Config:
    common: CommonConfig = field(default_factory=CommonConfig)
    app: Any = None


# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------

@dataclass
class AppInfo:
    id: str = ""
    guid: str = ""
    name: str = ""
    version: str = ""
    release: str = ""
    date: str = ""
    publisher: str = ""
    url: str = ""
    portapps_version: str = ""


@dataclass
class AppPrev:
    info: AppInfo = field(default_factory=AppInfo)
    win_version: WinVersion = field(default_factory=WinVersion)
    root_path: str = ""
    app_path: str = ""
    data_path: str = ""


class App:
    def __init__(self, id: str, name: str, app_cfg: Any = None):
        self.id = id
        self.name = name
        self.info = AppInfo()
        self.prev = AppPrev()
        self.args: list[str] = []
        self.process: str = ""
        self._logfile = None
        self.config = None
        self.app_path = ""
        self.data_path = ""
        self.working_dir = ""

        # WinVersion
        try:
            self.win_version = get_win_version()
        except Exception as e:
            self.fatal_box(f"Cannot get Windows version: {e}")

        # Root path
        try:
            ex = Path(sys.argv[0]).resolve()
            self.root_path = str(ex.parent)
        except Exception as e:
            self.fatal_box(f"Cannot get root absolute path: {e}")

        # Load info
        info_file = Path(self.root_path) / "portapp.json"
        try:
            with open(info_file) as f:
                data = json.load(f)
            self.info = AppInfo(**{k: data.get(k, "") for k in AppInfo.__dataclass_fields__})
        except FileNotFoundError as e:
            self.fatal_box(f"Cannot load portapps.json: {e}")
        except json.JSONDecodeError as e:
            self.fatal_box(f"Cannot unmarshal portapps.json: {e}")

        # Load config
        try:
            self.config = self._load_config(app_cfg)
        except Exception as e:
            self.fatal_box(f"Cannot load configuration: {e}")

        if app_cfg is not None and isinstance(app_cfg, dict):
            try:
                app_cfg.update(self.config.app or {})
            except Exception as e:
                self.fatal_box(f"Cannot decode {self.name} configuration: {e}")

        # Init logger
        try:
            self._init_logger()
        except Exception as e:
            self.fatal_box(f"Cannot configure logger: {e}")

        # Startup
        log.info("--------")
        log.info("Operating System: Windows %d.%d.%d",
                 self.win_version.major, self.win_version.minor, self.win_version.build)
        log.info("Starting %s %s-%s (portapps %s)...",
                 self.name, self.info.version, self.info.release, self.info.portapps_version)
        log.info("Release date: %s", self.info.date)
        log.info("Publisher: %s (%s)", self.info.publisher, self.info.url)
        log.info("Root path: %s", self.root_path)

        # Display config
        b = yaml.dump(self._config_dict())
        log.info("Configuration:\n%s", b)

        # Set paths
        self.app_path = str(Path(self.root_path) / "app")
        if self.config.common.app_path:
            self.app_path = self.config.common.app_path
        self.data_path = str(Path(self.root_path) / "data")
        self.working_dir = self.app_path

        # Load previous
        prev_file = Path(self.root_path) / "portapp-prev.json"
        if prev_file.exists():
            try:
                with open(prev_file) as f:
                    p = json.load(f)
            except (OSError, IOError) as e:
                self.fatal_box(f"Cannot load portapp-prev: {e}")
            except json.JSONDecodeError as e:
                log.error("Cannot unmarshal portapp-prev")
                prev_file.unlink(missing_ok=True)
            else:
                wv = p.get("win_version", {})
                self.prev = AppPrev(
                    info=AppInfo(**{k: p.get("info", {}).get(k, "") for k in AppInfo.__dataclass_fields__}),
                    win_version=WinVersion(**wv) if wv else WinVersion(),
                    root_path=p.get("root_path", ""),
                    app_path=p.get("app_path", ""),
                    data_path=p.get("data_path", ""),
                )

        # Load env vars from config
        if len(self.config.common.env) > 0:
            log.info("Setting environment variables from config...")
            for key, value in self.config.common.env.items():
                os.environ[key] = self._expand(value)

    # --- config loading ---

    def _load_config(self, app_cfg: Any) -> Config:
        cfg_path = Path(self.root_path) / f"{self.id}.yml"
        
        # Initialize with defaults
        config = Config(
            common=CommonConfig(
                disable_log=False,
                args=[],
                env={},
                app_path="",
            ),
            app=app_cfg,
        )

        # Write sample config
        sample_path = Path(self.root_path) / f"{self.id}.sample.yml"
        raw = yaml.dump(self._config_to_dict(config))
        sample_path.write_text(raw)

        # Skip if config file not found
        if not cfg_path.exists():
            return config

        # Read config
        raw_data = yaml.safe_load(cfg_path.read_text())
        if raw_data is None:
            return config

        # Unmarshal into config
        if "common" in raw_data:
            cr = raw_data["common"]
            config.common = CommonConfig(
                disable_log=cr.get("disable_log", False),
                args=cr.get("args", []),
                env=cr.get("env", {}),
                app_path=cr.get("app_path", ""),
            )
        if "app" in raw_data:
            config.app = raw_data["app"]

        return config

    def _config_to_dict(self, config: Config) -> dict:
        return {
            "common": {
                "disable_log": config.common.disable_log,
                "args": config.common.args,
                "env": config.common.env,
                "app_path": config.common.app_path,
            },
            "app": config.app,
        }

    def _config_dict(self) -> dict:
        return self._config_to_dict(self.config)

    # --- logger ---

    def _init_logger(self) -> None:
        if self.config.common.disable_log:
            logging.disable(logging.CRITICAL)
            return
        
        log_folder = Path(self.root_path) / "log"
        log_folder.mkdir(exist_ok=True)
        log_path = log_folder / f"{self.id}.log"
        
        self._logfile = open(log_path, "a")
        fh = logging.FileHandler(log_path)
        fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
        logging.getLogger().addHandler(fh)
        
        # Add fatal level (mirrors zerolog fatal hook)
        logging.addLevelName(60, "FATAL")
        def fatal(msg, *args, **kwargs):
            log.log(60, msg, *args, **kwargs)
            self.error_box(msg % args if args else msg)
        log.fatal = fatal

    # --- placeholder expansion ---

    def _expand(self, value: str) -> str:
        for placeholder, replacement in {
            "@ROOT_PATH@":    self.root_path,
            "@APP_PATH@":     self.app_path,
            "@DATA_PATH@":    self.data_path,
            "@DRIVE_LETTER@": self.root_path[0],
        }.items():
            value = value.replace(placeholder, replacement)
        return value

    # --- dialogs ---

    def error_box(self, msg: Any) -> None:
        msg_box(f"{self.name} portable", str(msg), MB_OK | MB_ICONERROR)

    def error_box_log(self, msg: Any) -> None:
        log.error("%s", msg)
        self.error_box(msg)

    def fatal_box(self, msg: Any) -> None:
        self.error_box(msg)
        sys.exit(1)

    def fatal_box_log(self, msg: Any) -> None:
        log.error("%s", msg)
        self.fatal_box(msg)

    # --- electron ---

    def electron_app_path(self) -> str:
        folder = find_electron_app_folder("app-", self.app_path)
        return str(Path(self.app_path) / folder)

    # --- launch ---

    def launch(self, extra_args: list[str] = []) -> None:
        log.info("Process: %s", self.process)
        log.info("Args (config file): %s", " ".join(self.config.common.args))
        log.info("Args (cmd line): %s", " ".join(extra_args))
        log.info("Args (hardcoded): %s", " ".join(self.args))
        log.info("Working dir: %s", self.working_dir)
        log.info("App path: %s", self.app_path)
        log.info("Data path: %s", self.data_path)
        log.info("Previous path: %s", self.prev.root_path)

        if not Path(self.process).exists():
            log.fatal("Application not found in %s", self.process)
            sys.exit(1)

        log.info("Launching %s", self.name)
        all_args = self.config.common.args + extra_args + self.args
        
        out = self._logfile if not self.config.common.disable_log else None
        
        log.info("Exec %s %s", self.process, " ".join(all_args))
        try:
            result = subprocess.run([self.process] + all_args, cwd=self.working_dir, stdout=out, stderr=out)
            if result.returncode != 0:
                log.fatal("Command failed")
                sys.exit(result.returncode)
        except Exception as e:
            log.fatal("Command failed: %s", e)
            sys.exit(1)

    # --- close ---

    def close(self) -> None:
        log.info("Closing %s", self.name)
        prev_file = Path(self.root_path) / "portapp-prev.json"
        
        prev_data = {
            "info": self.info.__dict__,
            "win_version": self.win_version.__dict__,
            "root_path": self.root_path,
            "app_path": self.app_path,
            "data_path": self.data_path,
        }
        
        try:
            json_prev = json.dumps(prev_data, indent=2)
        except Exception as e:
            log.error("Cannot marshal portapp-prev")
            return
        
        try:
            prev_file.write_text(json_prev)
        except Exception as e:
            log.error("Cannot write portapp-prev")
        
        if self._logfile:
            self._logfile.close()

run.ag.py

#!/usr/bin/env python3
import json
import os
import sys
from pathlib import Path

# Create portapp.json before App init
root_path = Path(sys.argv[0]).resolve().parent
portapp_file = root_path / "portapp.json"
if not portapp_file.exists():
    portapp_file.write_text(json.dumps({
        "id": "antigravity",
        "guid": "",
        "name": "Antigravity",
        "version": "1.0.0",
        "release": "1",
        "date": "2026-05-06",
        "publisher": "Antigravity",
        "url": "",
        "portapps_version": "3.0.0",
        "antigravity_folder": "vscodium"
    }, indent=2))

from libportable import App

cfg = {"cleanup": False}
app = App("antigravity", "Antigravity", cfg)

with open(portapp_file) as f:
    portapp = json.load(f)

Path(app.data_path).mkdir(parents=True, exist_ok=True)
app.process = str(Path(app.root_path) / portapp["antigravity_folder"] / "antigravity.exe")
app.working_dir = str(Path(app.root_path) / portapp["antigravity_folder"])

os.environ["VSCODE_APPDATA"] = str(Path(app.data_path) / "appdata")
if not app.config.common.disable_log:
    os.environ["VSCODE_LOGS"] = str(Path(app.data_path) / "logs")
os.environ["VSCODE_EXTENSIONS"] = str(Path(app.data_path) / "extensions")

try:
    app.launch(sys.argv[1:])
finally:
    app.close()