UmbraAudio/scripts/commit_check_notify.py

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Weekly commit reminder for the UmbraAudio working tree.

Invoked by ``~/.umbraaudio/umbra_commit_check.sh`` from the launchd job
``com.umbraaudio.commitcheck``. Under macOS's external-volume TCC policy only
the project's venv python may read ``/Volumes/1TB_APL_Storage``, so this file
(and ``commit_check.sh``, which it pipes to bash) is opened by python, never by
a shell.

The actual test lives in ``scripts/commit_check.sh`` — this module only
schedules it, records the outcome in ``logs/commit_check.log`` and raises a
macOS notification when the tree is dirty.
"""

from __future__ import annotations

import argparse
import logging
import os
import subprocess
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
CHECK_SCRIPT = REPO_ROOT / "scripts" / "commit_check.sh"
LOG_PATH = REPO_ROOT / "logs" / "commit_check.log"
SAFE_PATH = "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"

LOGGER = logging.getLogger("umbraaudio.commit_check")


def _configure_logging() -> None:
    """Log to stderr and to logs/commit_check.log (best effort)."""
    LOGGER.setLevel(logging.INFO)
    formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")

    stream = logging.StreamHandler(sys.stdout)
    stream.setFormatter(formatter)
    LOGGER.addHandler(stream)

    try:
        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        file_handler = logging.FileHandler(LOG_PATH)
        file_handler.setFormatter(formatter)
        LOGGER.addHandler(file_handler)
    except OSError:  # pragma: no cover - logging must never break the reminder
        LOGGER.warning("cannot write %s", LOG_PATH, exc_info=True)


def _applescript_string(value: str) -> str:
    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'


def send_notification(title: str, subtitle: str, message: str) -> bool:
    """Post a macOS notification; returns False when osascript is unavailable."""
    parts = [
        f"display notification {_applescript_string(message)}",
        f"with title {_applescript_string(title)}",
    ]
    if subtitle:
        parts.append(f"subtitle {_applescript_string(subtitle)}")
    try:
        result = subprocess.run(
            ["/usr/bin/osascript", "-e", " ".join(parts)],
            capture_output=True,
            timeout=20,
        )
    except (OSError, subprocess.TimeoutExpired):
        LOGGER.warning("could not run osascript", exc_info=True)
        return False
    if result.returncode != 0:
        LOGGER.warning(
            "osascript exited %s: %s",
            result.returncode,
            result.stderr.decode(errors="replace").strip(),
        )
        return False
    return True


def run_check(days: int, any_dirty: bool) -> subprocess.CompletedProcess[bytes]:
    """Run scripts/commit_check.sh via ``bash -s`` (stdin avoids the volume)."""
    args = ["/bin/bash", "-s", "--"]
    if any_dirty:
        args.append("--any")
    args += ["--days", str(days)]
    env = dict(os.environ)
    env["PATH"] = SAFE_PATH
    return subprocess.run(
        args,
        input=CHECK_SCRIPT.read_bytes(),
        cwd=str(REPO_ROOT),
        capture_output=True,
        env=env,
    )


def main(argv: list[str] | None = None) -> int:
    """Run the reminder; returns the check's exit code (0 clean, 1 dirty, 2 error)."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--days", type=int, default=7, help="age threshold in days")
    parser.add_argument(
        "--age-only",
        action="store_true",
        help="only remind when the tree has been dirty longer than --days",
    )
    parser.add_argument(
        "--no-notify",
        action="store_true",
        help="log the result but do not post a notification",
    )
    args = parser.parse_args(argv)

    any_dirty = not args.age_only

    _configure_logging()
    LOGGER.info(
        "commit check starting (repo=%s, days=%d, any=%s)",
        REPO_ROOT,
        args.days,
        any_dirty,
    )

    try:
        result = run_check(args.days, any_dirty)
    except OSError:
        LOGGER.exception("could not execute scripts/commit_check.sh")
        return 2

    output = result.stderr.decode(errors="replace").strip()
    if result.returncode == 0:
        LOGGER.info("clean / within threshold")
        return 0
    if result.returncode == 2:
        LOGGER.error("commit_check.sh failed (rc=2): %s", output)
        return 2

    LOGGER.warning("uncommitted work detected:\n%s", output)
    if not args.no_notify:
        dirty_line = next(
            (line.strip() for line in output.splitlines()
             if line.strip().lower().startswith("dirty for")),
            "",
        )
        sent = send_notification(
            title="UmbraAudio: commit reminder",
            subtitle="Uncommitted work",
            message=dirty_line or "The working tree is dirty - commit your work.",
        )
        LOGGER.info("notification %s", "sent" if sent else "not sent")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())