extendedMediaForwards/tools/bundle.py

#!/usr/bin/env python3
"""Bundle a multi-file exteraGram plugin (plugin.json + src/) into a single .plugin file.

Usage: python3 tools/bundle.py [--dev|--release] [--out DIR] [--check]

Also enables the repo's git hooks (scripts/pre-commit, scripts/post-commit) on
first run; see ensure_git_hooks().
"""
from __future__ import annotations

import argparse
import ast
import base64
import json
import os
import re
import stat
import subprocess
import sys
import zlib
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
SRC_DIR = REPO_ROOT / "src"
HOOKS_DIR = "scripts"
HOOK_NAMES = ("pre-commit", "post-commit")
PLUGIN_JSON = REPO_ROOT / "plugin.json"

RECOGNIZED_KEYS = {
    "id",
    "name",
    "description",
    "author",
    "version",
    "icon",
    "app_version",
    "sdk_version",
    "requirements",
}
REQUIRED_KEYS = {"id", "name"}
ID_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{1,31}$")

KEY_TO_DUNDER = {
    "id": "__id__",
    "name": "__name__",
    "description": "__description__",
    "author": "__author__",
    "version": "__version__",
    "icon": "__icon__",
    "app_version": "__app_version__",
    "sdk_version": "__sdk_version__",
    "requirements": "__requirements__",
}
METADATA_DUNDERS = set(KEY_TO_DUNDER.values()) | {"__min_version__"}

RUNTIME_PREAMBLE = '''\
import base64 as _b64, builtins as _bi, json as _j, linecache as _lc, types as _t, zlib as _z

_BUNDLE = _j.loads(_z.decompress(_b64.b64decode(_BUNDLE_BLOB)).decode("utf-8"))
_MODS = _BUNDLE["modules"]
_LOADED = {}
_ROOTS = {n.partition(".")[0] for n in _MODS}
_REAL_IMPORT = _bi.__import__


def _bundle_register_source(info):
    lines = info["src"].splitlines(True)
    _lc.cache[info["file"]] = (len(info["src"]), None, lines, info["file"])


def _bundle_load(name):
    mod = _LOADED.get(name)
    if mod is not None:
        return mod
    info = _MODS.get(name)
    if info is None:
        raise ImportError("no bundled module named %r" % name)
    parent = name.rpartition(".")[0]
    if parent:
        _bundle_load(parent)
    mod = _t.ModuleType(name)
    mod.__file__ = info["file"]
    mod.__package__ = name if info["pkg"] else parent
    if info["pkg"]:
        mod.__path__ = []
    builtins_ns = dict(vars(_bi))
    builtins_ns["__import__"] = _bundle_import
    mod.__dict__["__builtins__"] = builtins_ns
    _LOADED[name] = mod
    _bundle_register_source(info)
    try:
        exec(compile(info["src"], info["file"], "exec"), mod.__dict__)
    except BaseException:
        _LOADED.pop(name, None)
        raise
    if parent:
        setattr(_LOADED[parent], name.rpartition(".")[2], mod)
    return mod


def _bundle_import(name, globals=None, locals=None, fromlist=(), level=0):
    if level:
        pkg = (globals or {}).get("__package__") or ""
        parts = pkg.split(".") if pkg else []
        if level > 1:
            parts = parts[: 1 - level]
        if not parts:
            return _REAL_IMPORT(name, globals, locals, fromlist, level)
        full = ".".join(parts) + ("." + name if name else "")
    else:
        full = name
    if full.partition(".")[0] not in _ROOTS or full not in _MODS:
        return _REAL_IMPORT(name, globals, locals, fromlist, level)
    mod = _bundle_load(full)
    if not fromlist:
        return mod if level else _bundle_load(full.partition(".")[0])
    for item in fromlist:
        if item != "*" and not hasattr(mod, item) and full + "." + item in _MODS:
            _bundle_load(full + "." + item)
    return mod
'''

ENTRY_EXECUTION = '''\
_ENTRY = _MODS[_BUNDLE["entry"]]
_bundle_register_source(_ENTRY)
_bundle_builtins = dict(vars(_bi))
_bundle_builtins["__import__"] = _bundle_import
globals()["__builtins__"] = _bundle_builtins
exec(compile(_ENTRY["src"], _ENTRY["file"], "exec"), globals())
'''


class ValidationError(Exception):
    pass


def load_metadata() -> dict:
    if not PLUGIN_JSON.is_file():
        raise ValidationError(f"plugin.json not found at {PLUGIN_JSON}")
    try:
        text = PLUGIN_JSON.read_text(encoding="utf-8")
    except OSError as exc:
        raise ValidationError(f"cannot read plugin.json: {exc}")
    try:
        data = json.loads(text)
    except json.JSONDecodeError as exc:
        raise ValidationError(f"plugin.json is not valid JSON: {exc}")
    if not isinstance(data, dict):
        raise ValidationError("plugin.json must be a JSON object")
    unknown = set(data) - RECOGNIZED_KEYS
    if unknown:
        raise ValidationError(
            "plugin.json has unrecognized keys: " + ", ".join(sorted(unknown))
        )
    missing = REQUIRED_KEYS - set(data)
    if missing:
        raise ValidationError(
            "plugin.json is missing required keys: " + ", ".join(sorted(missing))
        )
    plugin_id = data["id"]
    if not isinstance(plugin_id, str) or not ID_RE.match(plugin_id):
        raise ValidationError(
            f"plugin.json id {plugin_id!r} is invalid: must be 2-32 chars, start with a "
            "letter, and contain only A-Za-z0-9_-"
        )
    if not isinstance(data["name"], str) or not data["name"]:
        raise ValidationError("plugin.json name must be a non-empty string")
    return data


def collect_modules() -> dict:
    if not SRC_DIR.is_dir():
        raise ValidationError(f"src/ directory not found at {SRC_DIR}")
    modules: dict = {}
    for path in sorted(SRC_DIR.rglob("*")):
        if path.is_dir():
            continue
        rel = path.relative_to(SRC_DIR)
        if any(part.startswith(".") for part in rel.parts):
            continue
        if "__pycache__" in rel.parts:
            continue
        if path.suffix == ".pyc":
            continue
        if path.suffix != ".py":
            print(
                f"warning: skipping non-Python file {path.relative_to(REPO_ROOT)} "
                "(single-file plugins cannot carry assets)",
                file=sys.stderr,
            )
            continue
        is_pkg = rel.name == "__init__.py"
        dotted_parts = rel.parts[:-1] if is_pkg else rel.parts[:-1] + (rel.stem,)
        name = ".".join(dotted_parts)
        if not name:
            raise ValidationError(f"cannot derive a module name for {path}")
        try:
            src = path.read_text(encoding="utf-8")
        except OSError as exc:
            raise ValidationError(f"cannot read {path}: {exc}")
        modules[name] = {"src": src, "path": path, "pkg": is_pkg}
    return modules


def validate_entry(modules: dict) -> None:
    main_path = SRC_DIR / "main.py"
    if not main_path.is_file():
        raise ValidationError("src/main.py does not exist")
    entry = modules.get("main")
    if entry is None:
        raise ValidationError("src/main.py could not be collected as the 'main' module")
    try:
        tree = ast.parse(entry["src"], filename=str(main_path))
    except SyntaxError as exc:
        raise ValidationError(f"src/main.py has a syntax error: {exc}")

    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and node.level > 0:
            raise ValidationError(
                "src/main.py must use absolute imports only (found a relative import); "
                "it is executed at plugin-module top level"
            )
        if isinstance(node, (ast.Assign, ast.AnnAssign)):
            targets = node.targets if isinstance(node, ast.Assign) else [node.target]
            for target in targets:
                if isinstance(target, ast.Name) and target.id in METADATA_DUNDERS:
                    raise ValidationError(
                        f"src/main.py must not assign metadata dunder {target.id!r} "
                        "(metadata belongs to plugin.json only)"
                    )

    def _base_names(class_node: ast.ClassDef) -> set:
        names = set()
        for base in class_node.bases:
            if isinstance(base, ast.Name):
                names.add(base.id)
            elif isinstance(base, ast.Attribute):
                names.add(base.attr)
        return names

    plugin_classes = [
        node.name
        for node in tree.body
        if isinstance(node, ast.ClassDef) and "BasePlugin" in _base_names(node)
    ]
    if len(plugin_classes) != 1:
        raise ValidationError(
            "src/main.py must contain exactly one class inheriting BasePlugin, found "
            f"{len(plugin_classes)}"
        )


def validate_modules_syntax(modules: dict) -> None:
    for name, info in modules.items():
        try:
            ast.parse(info["src"], filename=str(info["path"]))
        except SyntaxError as exc:
            raise ValidationError(f"{info['path']} has a syntax error: {exc}")


def format_metadata_constants(data: dict) -> str:
    lines = []
    for key, dunder in KEY_TO_DUNDER.items():
        if key not in data:
            continue
        value = data[key]
        if key in ("description", "icon") and value == "":
            continue
        if key == "requirements" and value == []:
            continue
        lines.append(f"{dunder} = {value!r}")
    return "\n".join(lines)


def build_payload(modules: dict, release: bool) -> dict:
    mod_dict = {}
    for name, info in modules.items():
        if release:
            file_value = str(info["path"].relative_to(REPO_ROOT))
        else:
            file_value = str(info["path"].resolve())
        mod_dict[name] = {"src": info["src"], "file": file_value, "pkg": info["pkg"]}
    return {"modules": mod_dict, "entry": "main"}


def encode_payload(payload: dict) -> str:
    data = json.dumps(payload, sort_keys=True).encode("utf-8")
    compressed = zlib.compress(data, 9)
    return base64.b64encode(compressed).decode("ascii")


def format_blob_literal(blob: str, width: int = 96) -> str:
    if not blob:
        return '_BUNDLE_BLOB = ""'
    lines = [blob[i : i + width] for i in range(0, len(blob), width)]
    body = "\n".join(f'    "{line}"' for line in lines)
    return "_BUNDLE_BLOB = (\n" + body + "\n)"


def assemble_bundle(data: dict, modules: dict, release: bool) -> str:
    header = "# Generated by tools/bundle.py from plugin.json + src/ — do not edit; edit the sources.\n"
    metadata = format_metadata_constants(data)
    payload = build_payload(modules, release)
    blob_literal = format_blob_literal(encode_payload(payload))
    text = "\n".join([header, metadata, "", blob_literal, "", RUNTIME_PREAMBLE, ENTRY_EXECUTION])
    if not text.endswith("\n"):
        text += "\n"
    return text


def ensure_git_hooks() -> None:
    """Point git at scripts/ for hooks, once per clone.

    Git never enables hooks from repository content, so a fresh clone (or a
    repo created from the template) starts without them. Every workflow runs
    this bundler first, directly or via scripts/deploy.sh and scripts/watch.sh,
    which makes it the one place that can switch them on without an extra
    setup step. Skipped in CI, outside a git work tree, and when
    core.hooksPath is already set (including to something else on purpose).
    Never fails the build.
    """
    if os.environ.get("CI"):
        return
    hooks = [REPO_ROOT / HOOKS_DIR / name for name in HOOK_NAMES]
    if not any(h.is_file() for h in hooks):
        return

    def git(*args):
        return subprocess.run(["git", *args], cwd=REPO_ROOT, capture_output=True, text=True)

    try:
        top = git("rev-parse", "--show-toplevel")
        if top.returncode != 0 or Path(top.stdout.strip()).resolve() != REPO_ROOT:
            return
        if git("config", "--get", "core.hooksPath").stdout.strip():
            return
        # Git silently skips hooks that are not executable.
        for hook in hooks:
            if hook.is_file():
                mode = hook.stat().st_mode
                hook.chmod(mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
        if git("config", "core.hooksPath", HOOKS_DIR).returncode == 0:
            print(f"enabled git hooks from {HOOKS_DIR}/ (git config core.hooksPath {HOOKS_DIR})", file=sys.stderr)
    except (OSError, subprocess.SubprocessError):
        pass


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="Bundle a multi-file exteraGram plugin into a single .plugin file."
    )
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument(
        "--dev", action="store_true", help="embed absolute local paths for debugging (default)"
    )
    mode.add_argument(
        "--release", action="store_true", help="embed repo-relative paths for distribution"
    )
    parser.add_argument("--out", default="dist", help="output directory (default: dist)")
    parser.add_argument("--check", action="store_true", help="validate only; do not write output")
    return parser.parse_args(argv)


def main(argv=None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)
    ensure_git_hooks()
    release = bool(args.release)
    try:
        data = load_metadata()
        modules = collect_modules()
        validate_entry(modules)
        validate_modules_syntax(modules)
    except ValidationError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    if args.check:
        print("ok")
        return 0

    text = assemble_bundle(data, modules, release)
    out_dir = Path(args.out)
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{data['id']}.plugin"
    out_path.write_text(text, encoding="utf-8")
    print(f"wrote {out_path} ({out_path.stat().st_size} bytes)")
    return 0


if __name__ == "__main__":
    sys.exit(main())