homelab-codex-ws/scripts/ha/lib/deploy_api.py

#!/usr/bin/env python3
"""Deploy path (repo -> instance) for the HA "api" adapter — see
kb/decisions/ha-configs-as-code.md, "Sync model" and "Validation gate".

Scope: automations/scripts/scenes only (the domains reachable via
`/api/config/<domain>/config/<id>`). Dashboards and helpers are WS-only —
ha_ws.py has no mutating command, deliberately — and are a separate task.

Hard sequence, no shortcuts, matching DESIGN.md:

  1. DRIFT-CHECK — re-import the instance's current state (reusing
     import_api.py's tested fetch/normalize logic) and diff it against the
     last-known-imported snapshot committed at repo HEAD. Any difference
     outside the objects this run is about to write aborts immediately —
     drift is never silently overwritten.
  2. VALIDATE — local sanity per object (parses, has the domain's required
     keys) plus a live `check_config` gate on the instance. Either failing
     aborts before a single write.
  3. WRITE — one POST per object, body = the repo file parsed back to JSON
     (the inverse of import_api's GET-json -> normalize -> yaml transform).
  4. VERIFY — GET the same object back, normalize, compare to what was
     just written. A mismatch is reported as an error; there is no
     auto-rollback.

Every function here is written to be called directly with an injected
duck-typed client (see tests/test_deploy_api_offline.sh) — no network, no
real HA instance, no live git repo required for the core logic.
"""
import difflib
import os
import subprocess
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import ha_api  # noqa: E402
import ha_write_api  # noqa: E402
import import_api  # noqa: E402
from normalize import dump_canonical_yaml, parse_yaml_text  # noqa: E402

DOMAIN_SUBDIRS = ("automations", "scripts", "scenes")
DOMAIN_API_NAME = {"automations": "automation", "scripts": "script", "scenes": "scene"}
REQUIRED_KEYS = {
    "automations": ("id",),
    "scenes": ("id", "entities"),
    "scripts": ("sequence",),
}


class DeployScopeError(RuntimeError):
    """A file argument can't be resolved to an in-scope object to deploy."""


# --- git-HEAD access (the "last-known-imported snapshot") ------------------


def _run_git(repo_root, args):
    return subprocess.run(
        ["git", "-C", repo_root] + args, capture_output=True, text=True, check=False
    )


def git_show(repo_root, relpath):
    """Return HEAD's content of relpath, or None if not tracked at HEAD."""
    proc = _run_git(repo_root, ["show", f"HEAD:{relpath}"])
    if proc.returncode != 0:
        return None
    return proc.stdout


def git_ls_tracked(repo_root, dir_relpath):
    """List tracked file paths (relative to repo_root) under dir_relpath at HEAD."""
    proc = _run_git(repo_root, ["ls-tree", "-r", "--name-only", "HEAD", "--", dir_relpath])
    if proc.returncode != 0:
        return []
    return [line for line in proc.stdout.splitlines() if line]


def load_repo_head_dir(repo_root, dir_relpath):
    """{key: canonical_yaml_text} for every *.yaml tracked at HEAD under dir_relpath."""
    result = {}
    for relpath in git_ls_tracked(repo_root, dir_relpath):
        if not relpath.endswith(".yaml"):
            continue
        key = os.path.basename(relpath)[: -len(".yaml")]
        text = git_show(repo_root, relpath)
        if text is None:
            continue
        result[key] = dump_canonical_yaml(parse_yaml_text(text))
    return result


def load_yaml_dir(dir_path):
    """{key: canonical_yaml_text} for every *.yaml file in dir_path (working tree)."""
    result = {}
    if not os.path.isdir(dir_path):
        return result
    for name in sorted(os.listdir(dir_path)):
        if not name.endswith(".yaml"):
            continue
        key = name[: -len(".yaml")]
        with open(os.path.join(dir_path, name), encoding="utf-8") as f:
            text = f.read()
        result[key] = dump_canonical_yaml(parse_yaml_text(text))
    return result


# --- target resolution ------------------------------------------------------


def build_targets(config_dir, repo_root, file_args):
    """Resolve CLI file arguments to a deploy target list.

    No args -> every automations/scripts/scenes *.yaml file currently in
    config_dir (working tree) — "wszystkie w scope".

    Explicit args -> each is tried as-is, then repo-root-relative, then
    config-dir-relative; whichever resolves to an existing file wins. Any
    file outside config_dir/{automations,scripts,scenes}/ is a hard error
    (dashboards/helpers are out of scope for this adapter).

    Returns a list of (subdir, key, abs_path), in sorted order for the
    no-args case and CLI-argument order otherwise.
    """
    if not file_args:
        targets = []
        for subdir in DOMAIN_SUBDIRS:
            dir_path = os.path.join(config_dir, subdir)
            if not os.path.isdir(dir_path):
                continue
            for name in sorted(os.listdir(dir_path)):
                if name.endswith(".yaml"):
                    targets.append((subdir, name[: -len(".yaml")], os.path.join(dir_path, name)))
        return targets

    config_dir_abs = os.path.abspath(config_dir)
    targets = []
    for arg in file_args:
        candidates = [arg, os.path.join(repo_root, arg), os.path.join(config_dir, arg)]
        resolved = next((c for c in candidates if os.path.isfile(c)), None)
        if resolved is None:
            raise DeployScopeError(
                f"file not found: {arg!r} (tried as given, repo-root-relative, "
                "and config-dir-relative)"
            )
        resolved = os.path.abspath(resolved)
        if not resolved.startswith(config_dir_abs + os.sep):
            raise DeployScopeError(
                f"{arg!r} resolves outside {config_dir} — deploy.sh only handles "
                "this instance's automations/scripts/scenes"
            )
        if not resolved.endswith(".yaml"):
            raise DeployScopeError(f"{arg!r} is not a .yaml file")
        subdir = os.path.basename(os.path.dirname(resolved))
        if subdir not in DOMAIN_SUBDIRS:
            raise DeployScopeError(
                f"{arg!r} is outside deploy.sh's scope (only automations/scripts/scenes are "
                "handled here — dashboards/helpers are WS-only and a separate task)"
            )
        key = os.path.basename(resolved)[: -len(".yaml")]
        targets.append((subdir, key, resolved))
    return targets


# --- 1. drift-check ----------------------------------------------------------


def drift_check(live_snapshot_dir, repo_root, config_relpath, target_keys):
    """Compare a fresh live snapshot (already on disk, canonical) against
    repo HEAD, for every object NOT in target_keys.

    Returns (aborts, warnings):
      aborts   -- [(subdir, key, reason, live_text_or_None, repo_text_or_None)]
                  content differs, or repo HEAD has an object missing live
                  (deleted out-of-band) -- never silently overwritten.
      warnings -- [(subdir, key, reason)]
                  object lives on the instance but has no repo file -- DELETE
                  is out of scope for this task, so this is informational only.
    """
    aborts = []
    warnings = []
    for subdir in DOMAIN_SUBDIRS:
        live = load_yaml_dir(os.path.join(live_snapshot_dir, subdir))
        repo = load_repo_head_dir(repo_root, f"{config_relpath}/{subdir}")
        for key in sorted(set(live) | set(repo)):
            if (subdir, key) in target_keys:
                continue
            live_val = live.get(key)
            repo_val = repo.get(key)
            if live_val is None and repo_val is not None:
                aborts.append(
                    (
                        subdir,
                        key,
                        "present in repo (HEAD), missing on the live instance — "
                        "deleted out-of-band since the last import",
                        None,
                        repo_val,
                    )
                )
            elif live_val is not None and repo_val is None:
                warnings.append(
                    (
                        subdir,
                        key,
                        "present on the live instance, no repo file — object was removed "
                        "from the repo; DELETE is out of scope here, left untouched on the "
                        "instance",
                    )
                )
            elif live_val != repo_val:
                aborts.append(
                    (
                        subdir,
                        key,
                        "content differs between the live instance and repo HEAD — "
                        "uncaptured drift",
                        live_val,
                        repo_val,
                    )
                )
    return aborts, warnings


# --- 2. validation -----------------------------------------------------------


def local_validate(target_objects):
    """target_objects: {(subdir, key): parsed_obj}. Returns [(subdir, key, reason)]."""
    errors = []
    for (subdir, key), obj in target_objects.items():
        if not isinstance(obj, dict):
            errors.append((subdir, key, "not a mapping at the top level"))
            continue
        for required in REQUIRED_KEYS[subdir]:
            if required not in obj:
                errors.append((subdir, key, f"missing required key '{required}'"))
        if subdir in ("automations", "scenes") and "id" in obj and str(obj["id"]) != key:
            errors.append(
                (subdir, key, f"'id' field {obj['id']!r} does not match filename '{key}.yaml'")
            )
    return errors


def remote_validate(client):
    """Live check_config gate. Returns (valid, raw_response)."""
    return client.check_config()


# --- 3+4. write + verify -----------------------------------------------------


def write_and_verify(client, target_objects):
    """target_objects: {(subdir, key): parsed_obj}, already locally validated.

    Returns {"written": [...], "write_failed": [...], "verify_failed": [...]}.
    No auto-rollback on a verify mismatch — writes to other objects continue,
    and the mismatch is reported so an operator can act on it directly.
    """
    report = {"written": [], "write_failed": [], "verify_failed": []}
    for (subdir, key), obj in sorted(target_objects.items()):
        domain = DOMAIN_API_NAME[subdir]
        try:
            client.post_config(domain, key, obj)
        except Exception as exc:  # noqa: BLE001 - surface any transport/HTTP error verbatim
            report["write_failed"].append((subdir, key, str(exc)))
            continue

        try:
            live_obj = client.get(f"/api/config/{domain}/config/{key}")
        except Exception as exc:  # noqa: BLE001 - surface any transport error verbatim
            report["verify_failed"].append((subdir, key, f"verify GET failed: {exc}"))
            continue

        if live_obj is None:
            report["verify_failed"].append(
                (subdir, key, "verify GET returned 404 immediately after a successful write")
            )
            continue

        expected_text = dump_canonical_yaml(obj)
        actual_text = dump_canonical_yaml(live_obj)
        if expected_text != actual_text:
            report["verify_failed"].append(
                (subdir, key, "post-write GET does not match what was written", expected_text, actual_text)
            )
            continue

        report["written"].append((subdir, key))
    return report


# --- orchestration ------------------------------------------------------------


def new_result(instance, dry_run):
    return {
        "instance": instance,
        "dry_run": dry_run,
        "targets": [],
        "drift_aborts": [],
        "drift_warnings": [],
        "validation_errors": [],
        "check_config_valid": None,
        "check_config_raw": None,
        "written": [],
        "write_failed": [],
        "verify_failed": [],
        "aborted": False,
        "abort_reason": None,
    }


def run(repo_root, config_dir, config_relpath, instance, file_args, dry_run, client):
    """The full DRIFT-CHECK -> VALIDATE -> WRITE -> VERIFY sequence.

    `client` is duck-typed: .get(path), .post_config(domain, id, body),
    .check_config() -- see ha_write_api.WriteClient for the real
    implementation and tests/test_deploy_api_offline.sh for the fake used
    in offline tests.
    """
    result = new_result(instance, dry_run)

    try:
        targets = build_targets(config_dir, repo_root, file_args)
    except DeployScopeError as exc:
        result["aborted"] = True
        result["abort_reason"] = str(exc)
        return result

    if not targets:
        result["aborted"] = True
        result["abort_reason"] = "no target objects resolved — nothing to deploy"
        return result

    result["targets"] = [(subdir, key) for subdir, key, _ in targets]
    target_keys = {(subdir, key) for subdir, key, _ in targets}

    states = client.get("/api/states")

    with tempfile.TemporaryDirectory(prefix="ha-deploy-drift-") as snapshot_dir:
        live_report = import_api.new_report()
        import_api.import_automations(client, states, snapshot_dir, live_report)
        import_api.import_scripts(client, states, snapshot_dir, live_report)
        import_api.import_scenes(client, states, snapshot_dir, live_report)
        aborts, warnings = drift_check(snapshot_dir, repo_root, config_relpath, target_keys)

    result["drift_aborts"] = aborts
    result["drift_warnings"] = warnings
    if aborts:
        result["aborted"] = True
        result["abort_reason"] = "drift detected outside deploy scope — never overwritten"
        return result

    target_objects = {}
    parse_errors = []
    for subdir, key, path in targets:
        with open(path, encoding="utf-8") as f:
            text = f.read()
        try:
            target_objects[(subdir, key)] = parse_yaml_text(text)
        except Exception as exc:  # noqa: BLE001 - any parse failure is a validation rejection
            parse_errors.append((subdir, key, f"does not parse as YAML: {exc}"))

    result["validation_errors"] = parse_errors + local_validate(target_objects)
    if result["validation_errors"]:
        result["aborted"] = True
        result["abort_reason"] = "local validation failed — nothing was written"
        return result

    valid, raw = remote_validate(client)
    result["check_config_valid"] = valid
    result["check_config_raw"] = raw
    if not valid:
        result["aborted"] = True
        result["abort_reason"] = "instance check_config reported invalid — nothing was written"
        return result

    if dry_run:
        return result

    write_report = write_and_verify(client, target_objects)
    result["written"] = write_report["written"]
    result["write_failed"] = write_report["write_failed"]
    result["verify_failed"] = write_report["verify_failed"]
    return result


# --- reporting ----------------------------------------------------------------


def _print_diff(live_text, repo_text, label_live, label_repo):
    diff = difflib.unified_diff(
        (repo_text or "").splitlines(keepends=True),
        (live_text or "").splitlines(keepends=True),
        fromfile=label_repo,
        tofile=label_live,
    )
    sys.stderr.writelines(diff)


def print_report(result):
    mode = "DRY-RUN (plan only, no mutating calls)" if result["dry_run"] else "LIVE"
    print(f"== ha deploy: instance={result['instance']} mode={mode} ==", file=sys.stderr)

    print(f"-> targets ({len(result['targets'])}):", file=sys.stderr)
    for subdir, key in result["targets"]:
        print(f"     {subdir}/{key}.yaml", file=sys.stderr)

    for subdir, key, reason in result["drift_warnings"]:
        print(f"-> warning: {subdir}/{key}: {reason}", file=sys.stderr)

    if result["drift_aborts"]:
        print("-> DRIFT DETECTED — aborting, nothing was written:", file=sys.stderr)
        for subdir, key, reason, live_text, repo_text in result["drift_aborts"]:
            print(f"     {subdir}/{key}: {reason}", file=sys.stderr)
            _print_diff(live_text, repo_text, f"live:{subdir}/{key}", f"repo-HEAD:{subdir}/{key}")

    if result["validation_errors"]:
        print("-> LOCAL VALIDATION FAILED — aborting, nothing was written:", file=sys.stderr)
        for subdir, key, reason in result["validation_errors"]:
            print(f"     {subdir}/{key}: {reason}", file=sys.stderr)

    if result["check_config_valid"] is not None:
        status = "valid" if result["check_config_valid"] else "INVALID"
        print(f"-> check_config: {status}", file=sys.stderr)
        if not result["check_config_valid"]:
            print(f"     {result['check_config_raw']}", file=sys.stderr)

    if result["aborted"]:
        print(f"== ABORTED: {result['abort_reason']} ==", file=sys.stderr)
        return

    if result["dry_run"]:
        print("== dry-run complete: plan above would be applied on a live run ==", file=sys.stderr)
        return

    print(f"-> written ({len(result['written'])}):", file=sys.stderr)
    for subdir, key in result["written"]:
        print(f"     {subdir}/{key}.yaml", file=sys.stderr)

    if result["write_failed"]:
        print(f"-> write failed ({len(result['write_failed'])}):", file=sys.stderr)
        for subdir, key, reason in result["write_failed"]:
            print(f"     {subdir}/{key}: {reason}", file=sys.stderr)

    if result["verify_failed"]:
        print(f"-> verify failed ({len(result['verify_failed'])}):", file=sys.stderr)
        for entry in result["verify_failed"]:
            subdir, key, reason = entry[0], entry[1], entry[2]
            print(f"     {subdir}/{key}: {reason}", file=sys.stderr)
            if len(entry) == 5:
                _print_diff(entry[3], entry[4], f"written:{subdir}/{key}", f"live-after-write:{subdir}/{key}")

    total_errors = len(result["write_failed"]) + len(result["verify_failed"])
    print(
        f"== deploy complete: {len(result['written'])} written, "
        f"{len(result['write_failed'])} write errors, {len(result['verify_failed'])} verify errors ==",
        file=sys.stderr,
    )
    if total_errors:
        print("== see errors above — no auto-rollback was performed ==", file=sys.stderr)


def main(argv):
    if len(argv) < 8:
        print(
            "usage: deploy_api.py <base_url> <token_path> <repo_root> <config_dir> "
            "<config_relpath> <instance> <0|1:dry_run> [file...]",
            file=sys.stderr,
        )
        return 2
    _, base_url, token_path, repo_root, config_dir, config_relpath, instance, dry_run_flag = argv[:8]
    file_args = argv[8:]
    dry_run = dry_run_flag == "1"

    try:
        token = ha_api.read_token(token_path)
    except ha_api.HaApiError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    client = ha_write_api.WriteClient(base_url, token)

    try:
        result = run(repo_root, config_dir, config_relpath, instance, file_args, dry_run, client)
    except Exception as exc:  # noqa: BLE001 - surface any transport/HTTP error verbatim
        print(f"error: deploy failed: {exc}", file=sys.stderr)
        return 1

    print_report(result)

    if result["aborted"] or result["write_failed"] or result["verify_failed"]:
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))