Coverage for scripts/status.py: 100%
120 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-31 06:18 +0000
1#!/usr/bin/env python3
2"""Show the current rhiza sync status from `.rhiza/template.lock`.
4A stdlib-only port of the `rhiza status` command, bundled with this plugin so
5`/rhiza:status` works without the `rhiza` CLI (or PyYAML) installed. It reads
6the authoritative lock written by the last sync and reports the template
7repository, ref, SHA, sync timestamp, strategy, and included templates/paths.
8With `--files` it also renders the managed files as a directory tree — the view
9that used to live in the separate `/rhiza:tree` command.
11Usage:
12 uv run --python 3.12 --no-project python scripts/status.py [TARGET] [--json] [--files] [--check]
14 TARGET repository root to inspect (default: current directory)
15 --json emit a single JSON object on stdout instead of human-readable lines
16 --files append the managed files as a `tree`-style listing (human output)
17 --check compare the pinned ref to the latest upstream release (needs network)
19When no lock is present the repo has never been synced; this prints a hint to
20stderr and exits 0 (nothing to report is not an error). The `--json` payload
21mirrors `rhiza status --json` field-for-field (including `files`), so tools like
22stats.py can read either interchangeably. Where the CLI uses `rich` for the file
23tree, this renders plain ASCII connectors so it stays dependency-free.
25Everything is offline and deterministic except `--check`, which shells out to
26`git ls-remote --tags` (no `gh`, no auth for public repos) to see whether a newer
27release exists; a network or git failure there is reported, never fatal.
28"""
30from __future__ import annotations
32import argparse
33import json
34import re
35import subprocess
36import sys
37from pathlib import Path
38from typing import Any
40_SEMVER_RE = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)$")
42sys.path.insert(0, str(Path(__file__).resolve().parent))
43from _rhiza_yaml import load_yaml # noqa: E402
45LOCK_REL = Path(".rhiza") / "template.lock"
48def _as_list(value: Any) -> list[str]:
49 """Normalise a scalar/None/list lock field to a list of strings."""
50 if value is None:
51 return []
52 if isinstance(value, list):
53 return [str(x) for x in value]
54 return [str(value)]
57def _status_dict(lock: dict[str, Any]) -> dict[str, Any]:
58 """Build the machine-readable status payload from a parsed lock.
60 Mirrors the field set emitted by `rhiza status --json`.
61 """
62 host = str(lock.get("host", "github"))
63 repo = str(lock.get("repo", ""))
64 return {
65 "repository": f"{host}/{repo}" if repo else host,
66 "host": host,
67 "repo": repo,
68 "ref": str(lock.get("ref", "main")),
69 "sha": str(lock.get("sha", "")),
70 "synced_at": str(lock.get("synced_at", "")),
71 "strategy": str(lock.get("strategy", "")),
72 "templates": _as_list(lock.get("templates")),
73 "include": _as_list(lock.get("include")),
74 "files": _as_list(lock.get("files")),
75 }
78def _build_tree(paths: list[str]) -> dict[str, Any]:
79 """Fold a flat path list into a nested {name: subtree} dict."""
80 root: dict[str, Any] = {}
81 for path in sorted(paths):
82 node = root
83 for part in Path(path).parts:
84 node = node.setdefault(part, {})
85 return root
88def _render(node: dict[str, Any], prefix: str = "") -> list[str]:
89 """Render a nested tree dict into Unix-`tree`-style lines."""
90 lines: list[str] = []
91 items = sorted(node.items())
92 for i, (name, child) in enumerate(items):
93 last = i == len(items) - 1
94 lines.append(f"{prefix}{'└── ' if last else '├── '}{name}")
95 if child:
96 lines.extend(_render(child, prefix + (" " if last else "│ ")))
97 return lines
100def _print_file_tree(files: list[str]) -> None:
101 """Print the managed files as a `tree`-style listing plus a total count."""
102 if not files:
103 print("No files are tracked in template.lock", file=sys.stderr)
104 return
105 print("\nFiles managed by Rhiza:")
106 print(".")
107 for line in _render(_build_tree(files)):
108 print(line)
109 print(f"\n{len(files)} file{'s' if len(files) != 1 else ''} managed by Rhiza")
112def _parse_semver(tag: str) -> tuple[int, int, int] | None:
113 """Parse a `vX.Y.Z` tag into a (major, minor, patch) tuple, or None."""
114 m = _SEMVER_RE.match(tag)
115 return (int(m.group(1)), int(m.group(2)), int(m.group(3))) if m else None
118def _remote_url(host: str, repo: str) -> str:
119 """Build an HTTPS clone URL for the given host alias and owner/repo slug."""
120 base = "gitlab.com" if "gitlab" in host else "github.com"
121 return f"https://{base}/{repo}"
124def _remote_tags(host: str, repo: str) -> list[str]:
125 """Return the remote's tag names via `git ls-remote --tags`, or [] on failure."""
126 try:
127 proc = subprocess.run( # nosec B603 B607
128 ["git", "ls-remote", "--tags", _remote_url(host, repo)],
129 capture_output=True,
130 text=True,
131 check=True,
132 timeout=20,
133 )
134 except (OSError, subprocess.SubprocessError):
135 return []
136 tags: set[str] = set()
137 for line in proc.stdout.splitlines():
138 _, _, name = line.partition("refs/tags/")
139 if name:
140 tags.add(name.removesuffix("^{}"))
141 return sorted(tags)
144def _outdated_message(ref: str, tags: list[str]) -> str:
145 """Compose a one-line 'update available?' summary for *ref* against *tags*."""
146 releases = [(t, v) for t in tags if (v := _parse_semver(t)) is not None]
147 if not releases:
148 return "Update : could not determine the latest release"
149 latest_tag, latest_ver = max(releases, key=lambda tv: tv[1])
150 current = _parse_semver(ref)
151 if current is None:
152 return (
153 f"Update : latest release is {latest_tag} "
154 f"(current ref '{ref}' is not a release tag) — run /update"
155 )
156 behind = sum(1 for _, v in releases if v > current)
157 if behind == 0:
158 return f"Update : up to date ({latest_tag} is the latest release)"
159 plural = "s" if behind != 1 else ""
160 return f"Update : {ref} → {latest_tag} ({behind} release{plural} behind) — run /update"
163def _print_outdated(payload: dict[str, Any]) -> None:
164 """Print the outdated-check line for a status payload (network via git)."""
165 if not payload["repo"]:
166 print("Update : no template repository recorded in the lock")
167 return
168 tags = _remote_tags(payload["host"], payload["repo"])
169 print(_outdated_message(payload["ref"], tags))
172def status(
173 target: Path,
174 *,
175 json_output: bool = False,
176 show_files: bool = False,
177 check: bool = False,
178) -> int:
179 """Print the sync status; return a process exit code."""
180 lock_path = (target / LOCK_REL).resolve()
181 if not lock_path.exists():
182 print(
183 "No template.lock found — run /rhiza:update to perform the first sync", file=sys.stderr
184 )
185 return 0
187 try:
188 lock = load_yaml(lock_path)
189 except (OSError, ValueError) as exc:
190 print(f"Could not read {lock_path}: {exc}", file=sys.stderr)
191 return 1
193 payload = _status_dict(lock)
194 if json_output:
195 print(json.dumps(payload, indent=2))
196 return 0
198 print(f"Repository : {payload['repository']}")
199 print(f"Ref : {payload['ref']}")
200 sha = payload["sha"]
201 print(f"SHA : {sha[:12]}" if sha else "SHA : (unknown)")
202 print(f"Synced at : {payload['synced_at'] or '(unknown)'}")
203 print(f"Strategy : {payload['strategy'] or '(unknown)'}")
204 if payload["templates"]:
205 print(f"Templates : {', '.join(payload['templates'])}")
206 elif payload["include"]:
207 print(f"Include : {', '.join(payload['include'])}")
208 if check:
209 _print_outdated(payload)
210 if show_files:
211 _print_file_tree(payload["files"])
212 return 0
215def main(argv: list[str] | None = None) -> int:
216 """Entry point: print status; return an exit code."""
217 parser = argparse.ArgumentParser(
218 description="Show the current rhiza sync status from .rhiza/template.lock.",
219 )
220 parser.add_argument(
221 "target",
222 nargs="?",
223 default=".",
224 help="Repository root to inspect (default: current directory).",
225 )
226 parser.add_argument(
227 "--json",
228 dest="json_output",
229 action="store_true",
230 help="Emit the status as a single JSON object on stdout.",
231 )
232 parser.add_argument(
233 "--files",
234 "--tree",
235 dest="show_files",
236 action="store_true",
237 help="Append the managed files as a tree-style listing (human output).",
238 )
239 parser.add_argument(
240 "--check",
241 dest="check",
242 action="store_true",
243 help="Compare the pinned ref to the latest upstream release (needs network).",
244 )
245 args = parser.parse_args(argv)
246 return status(
247 Path(args.target),
248 json_output=args.json_output,
249 show_files=args.show_files,
250 check=args.check,
251 )
254if __name__ == "__main__":
255 raise SystemExit(main())