KiCad - the KiCad Bridge

"""KiCad Plugin and Content Manager (PCM) automation: install packages from a
PCM repository WITHOUT the dialog, the way PCM itself does it on disk.

Three verbs — `pcm_install`, `pcm_list`, `pcm_uninstall`. Added 2026-09-01
after proving that a wiki page (adom/kicad-pcm) is a valid PCM repository:
KiCad read it fine, but PCM has no CLI and its dialog is an owned window the
bridge's input verbs cannot target, so "click Install" could not be automated
through the GUI. It can be automated on disk, because PCM is only files:

  1. extract  <zip>/<folder>/<contents>  ->  <3rdparty>/<folder>/<pkgid>/<contents>
     (kicad/pcm/pcm_task_manager.cpp extract(): folder must be one of
     plugins/footprints/symbols/3dmodels/resources/colors; identifier dots
     become underscores on disk so Python packages import flat)
  2. record   <config>/installed_packages.json  {"packages":[entry...]}
     (kicad/pcm/pcm_data.cpp to_json(PCM_INSTALLATION_ENTRY): package,
     current_version, repository_id, repository_name, install_timestamp, pinned)
  3. NOTHING ELSE. KiCad registers the libraries itself: symbol_lib_table.cpp
     PCM_SYM_LIB_TRAVERSER and fp_lib_table.cpp PCM_FP_LIB_TRAVERSER scan
     $KICADn_3RD_PARTY/symbols/<pkgid>/*.kicad_sym and footprints/<pkgid>/*.pretty
     when the GLOBAL TABLE LOADS and insert rows named <lib_prefix><name>
     (default prefix "PCM_"), as long as pcm.lib_auto_add is on (the default).
     That scan happens at editor load, which is why a running editor needs a
     restart to see a new package - reported here as restartNeeded, never done.

repository_id is sha256(repository url)[:16] (pcm.cpp SetRepositoryList), so
KiCad's own PCM dialog shows a bridge-installed package as Installed, offers
updates for it, and can uninstall it. The bridge is an automation of PCM, not
a parallel mechanism.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import tempfile
import time
import urllib.request
import zipfile
from pathlib import Path

_FOLDERS = {"plugins", "footprints", "symbols", "3dmodels", "resources", "colors"}
_MAX_ZIP = 512 * 1024 * 1024
_UA = "adom-kicad-bridge pcm"


from handlers.win_paths import shell_documents_dir as _shell_documents_dir


def _third_party_dir(kicad_info: dict) -> tuple[Path | None, str]:
    """Resolve KICAD<n>_3RD_PARTY exactly as KiCad does, in priority order:
    1. an explicit override in kicad_common.json environment.vars
    2. <shell Documents>/KiCad/<major.minor>/3rdparty (OneDrive-aware)
    3. the bridge's user_dir guess (last resort, and it was wrong once)
    Returns (path, how)."""
    ver = str(kicad_info.get("version") or "")
    major = ver.split(".")[0] if ver else ""
    config_dir = kicad_info.get("config_dir")
    if config_dir and major:
        try:
            common = json.loads((Path(config_dir) / "kicad_common.json").read_text(encoding="utf-8"))
            v = ((common.get("environment") or {}).get("vars") or {}).get(f"KICAD{major}_3RD_PARTY")
            if v:
                return Path(v), "kicad_common.json override"
        except (OSError, ValueError):
            pass
    docs = _shell_documents_dir()
    if docs and ver:
        return docs / "KiCad" / ver / "3rdparty", "shell Documents (OneDrive-aware)"
    if kicad_info.get("user_dir"):
        return Path(kicad_info["user_dir"]) / "3rdparty", "user_dir fallback"
    return None, "unresolved"


def _paths(kicad_info: dict) -> tuple[Path | None, Path | None, str | None]:
    """(3rdparty dir, installed_packages.json, KiCad version) or Nones."""
    config_dir = kicad_info.get("config_dir")
    ver = str(kicad_info.get("version") or "")
    third, _how = _third_party_dir(kicad_info)
    if not third or not config_dir:
        return None, None, None
    return third, Path(config_dir) / "installed_packages.json", ver


def _fetch(url: str, timeout: int = 120) -> bytes:
    req = urllib.request.Request(url, headers={"User-Agent": _UA})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read()


def _read_installed(p: Path) -> dict:
    try:
        d = json.loads(p.read_text(encoding="utf-8"))
        if isinstance(d, dict) and isinstance(d.get("packages"), list):
            return d
    except (OSError, ValueError):
        pass
    return {"packages": []}


def _write_installed(p: Path, d: dict) -> None:
    p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(d, indent=4) + "\n", encoding="utf-8")
    os.replace(tmp, p)


def _pkgdir(identifier: str) -> str:
    return identifier.replace(".", "_")


def _extract(zip_path: Path, third: Path, identifier: str) -> dict:
    """PCM's extract(): <folder>/<contents> -> <3rdparty>/<folder>/<pkgid>/<contents>.
    Zip-slip guarded; entries outside the known folders are ignored (as PCM does)."""
    pkgid = _pkgdir(identifier)
    written: list[str] = []
    libs = {"symbols": [], "footprints": [], "3dmodels": []}
    with zipfile.ZipFile(zip_path) as zf:
        total = sum(i.file_size for i in zf.infolist())
        if total > _MAX_ZIP:
            raise ValueError(f"package unpacks to {total} bytes, over the {_MAX_ZIP} cap")
        for info in zf.infolist():
            name = info.filename.replace("\\", "/")
            if not name or name.endswith("/"):
                continue
            parts = [p for p in name.split("/") if p not in ("", ".")]
            if any(p == ".." for p in parts) or re.match(r"^[A-Za-z]:", name) or name.startswith("/"):
                raise ValueError(f"unsafe zip entry: {info.filename}")
            if len(parts) < 2 or parts[0] not in _FOLDERS:
                continue
            target = (third / parts[0] / pkgid).joinpath(*parts[1:])
            if not str(target.resolve()).startswith(str((third / parts[0] / pkgid).resolve())):
                raise ValueError(f"unsafe zip entry: {info.filename}")
            target.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(info) as src, open(target, "wb") as dst:
                shutil.copyfileobj(src, dst)
            written.append(str(target))
            if parts[0] == "symbols" and name.endswith(".kicad_sym"):
                libs["symbols"].append(parts[-1][:-10])
            elif parts[0] == "footprints" and len(parts) >= 3 and parts[1].endswith(".pretty"):
                if parts[1][:-7] not in libs["footprints"]:
                    libs["footprints"].append(parts[1][:-7])
            elif parts[0] == "3dmodels" and len(parts) >= 3 and parts[1] not in libs["3dmodels"]:
                libs["3dmodels"].append(parts[1])
    return {"files": len(written), "libraries": libs, "packageDir": pkgid}


def handle_pcm_install(kicad_info: dict, args: dict) -> dict:
    """Install a package from a PCM repository, on disk, exactly as PCM would.

    Args:
      repositoryUrl  the repository.json URL (e.g. the adom/kicad-pcm one)
      identifier     package identifier (e.g. inc.adom.kicad.basic-parts)
      version        optional; default = newest listed
      libPrefix      optional; what KiCad will prefix nicknames with (default PCM_),
                     only used to REPORT the nicknames the scan will create
    """
    repo_url = (args.get("repositoryUrl") or "").strip()
    ident = (args.get("identifier") or "").strip()
    want = (args.get("version") or "").strip()
    prefix = args.get("libPrefix") or "PCM_"
    if not repo_url or not ident:
        return {"success": False, "error": "repositoryUrl and identifier are required",
                "_hint": "repositoryUrl is the repository.json URL; identifier like inc.adom.kicad.basic-parts."}
    third, installed_p, ver = _paths(kicad_info)
    if not third:
        return {"success": False, "error": "KiCad user/config directories unknown",
                "_hint": "kicad_readiness first; the bridge needs a detected KiCad to know 3rdparty and settings paths."}
    try:
        repo = json.loads(_fetch(repo_url))
        pkgs_url = repo["packages"]["url"]
        pkgs_raw = _fetch(pkgs_url)
        if hashlib.sha256(pkgs_raw).hexdigest() != repo["packages"].get("sha256"):
            return {"success": False, "error": "packages.json hash does not match repository.json",
                    "_hint": "The repository is mid-update or corrupted; retry in a minute."}
        packages = json.loads(pkgs_raw)["packages"]
    except Exception as e:  # pylint: disable=broad-except
        return {"success": False, "error": f"could not read repository: {type(e).__name__}: {e}"}
    pkg = next((p for p in packages if p.get("identifier") == ident), None)
    if not pkg:
        return {"success": False, "error": f"package not in repository: {ident}",
                "available": [p.get("identifier") for p in packages][:20]}
    versions = pkg.get("versions") or []
    if want:
        vsel = next((v for v in versions if v.get("version") == want), None)
        if not vsel:
            return {"success": False, "error": f"version {want} not offered",
                    "available": [v.get("version") for v in versions]}
    else:
        vsel = versions[-1] if versions else None
    if not vsel or not vsel.get("download_url"):
        return {"success": False, "error": "package has no downloadable version"}

    installed = _read_installed(installed_p)
    prior = next((e for e in installed["packages"] if e.get("package", {}).get("identifier") == ident), None)
    if prior and prior.get("current_version") == vsel["version"] and not args.get("force"):
        return {"success": True, "alreadyInstalled": True, "identifier": ident,
                "version": vsel["version"],
                "_hint": "Already installed at this version; pass force:true to reinstall."}

    fd, tmp = tempfile.mkstemp(suffix=".zip")
    os.close(fd)
    try:
        blob = _fetch(vsel["download_url"], timeout=600)
        got = hashlib.sha256(blob).hexdigest()
        if vsel.get("download_sha256") and got != vsel["download_sha256"]:
            return {"success": False, "error": "download sha256 mismatch",
                    "expected": vsel["download_sha256"], "got": got,
                    "_hint": "Refusing to install a package whose bytes do not match the repository's digest."}
        Path(tmp).write_bytes(blob)
        # replace-on-update: remove the previous extraction first, as PCM does
        for folder in _FOLDERS:
            old = third / folder / _pkgdir(ident)
            if old.exists():
                shutil.rmtree(old, ignore_errors=True)
        ext = _extract(Path(tmp), third, ident)
    except Exception as e:  # pylint: disable=broad-except
        return {"success": False, "error": f"install failed: {type(e).__name__}: {e}"}
    finally:
        Path(tmp).unlink(missing_ok=True)

    entry = {"package": {k: v for k, v in pkg.items() if k != "versions"} | {"versions": [vsel]},
             "current_version": vsel["version"],
             "repository_id": hashlib.sha256(repo_url.encode()).hexdigest()[:16],
             "repository_name": repo.get("name", ""),
             "install_timestamp": int(time.time()),
             "pinned": False}
    installed["packages"] = [e for e in installed["packages"]
                             if e.get("package", {}).get("identifier") != ident] + [entry]
    _write_installed(installed_p, installed)

    running = bool(kicad_info.get("running") or kicad_info.get("runningInstances"))
    nick = lambda names: [f"{prefix}{n}" for n in names]
    return {"success": True, "identifier": ident, "version": vsel["version"],
            "filesInstalled": ext["files"], "installedTo": str(third / "*" / ext["packageDir"]),
            "thirdPartyDir": str(third), "thirdPartyResolvedBy": _third_party_dir(kicad_info)[1],
            "symbolLibraries": nick(ext["libraries"]["symbols"]),
            "footprintLibraries": nick(ext["libraries"]["footprints"]),
            "modelSets": ext["libraries"]["3dmodels"],
            "kicadRunning": running, "restartNeeded": running,
            "_hint": ("Installed exactly as PCM would; KiCad's PCM dialog will list it as Installed. "
                      "Library nicknames above are created by KiCad's own PCM scan when the global "
                      "library table next LOADS - so a KiCad that is already running must be restarted "
                      "(never done for you) before the libraries appear.")}


def handle_pcm_list(kicad_info: dict, args: dict) -> dict:
    """Packages KiCad's PCM considers installed on this machine (read-only)."""
    third, installed_p, ver = _paths(kicad_info)
    if not installed_p:
        return {"success": False, "error": "KiCad config directory unknown"}
    d = _read_installed(installed_p)
    out = []
    for e in d["packages"]:
        p = e.get("package", {})
        out.append({"identifier": p.get("identifier"), "name": p.get("name"),
                    "type": p.get("type"), "version": e.get("current_version"),
                    "repository": e.get("repository_name"),
                    "onDisk": any((third / f / _pkgdir(p.get("identifier", ""))).exists() for f in _FOLDERS)})
    return {"success": True, "count": len(out), "packages": out,
            "thirdPartyDir": str(third), "thirdPartyResolvedBy": _third_party_dir(kicad_info)[1],
            "kicadVersion": ver}


def handle_pcm_uninstall(kicad_info: dict, args: dict) -> dict:
    """Remove a PCM-installed package: its 3rdparty directories + its record."""
    ident = (args.get("identifier") or "").strip()
    if not ident:
        return {"success": False, "error": "identifier is required"}
    third, installed_p, _ = _paths(kicad_info)
    if not third:
        return {"success": False, "error": "KiCad user/config directories unknown"}
    removed = []
    for folder in _FOLDERS:
        d = third / folder / _pkgdir(ident)
        if d.exists():
            shutil.rmtree(d, ignore_errors=True)
            removed.append(str(d))
    installed = _read_installed(installed_p)
    before = len(installed["packages"])
    installed["packages"] = [e for e in installed["packages"]
                             if e.get("package", {}).get("identifier") != ident]
    _write_installed(installed_p, installed)
    running = bool(kicad_info.get("running") or kicad_info.get("runningInstances"))
    return {"success": True, "identifier": ident, "removedDirs": removed,
            "recordRemoved": before != len(installed["packages"]),
            "kicadRunning": running, "restartNeeded": running,
            "_hint": "KiCad drops the PCM_ library rows itself on next load when pcm.lib_auto_remove is on."}