1054 lines
47 KiB
Python
Executable File
1054 lines
47 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Interactive TUI for browsing and updating unified version.json files.
|
|
|
|
Features:
|
|
- Scans packages/**/version.json and lists all packages
|
|
- Per-package view:
|
|
- Choose base or any variant
|
|
- List all sources/components with current ref (tag/rev/url/version) and hash
|
|
- For GitHub sources: fetch candidates (latest release tag, latest tag, latest commit)
|
|
- For Git sources: fetch latest commit (HEAD)
|
|
- For URL sources: recompute hash (url/urlTemplate with rendered variables)
|
|
- Actions on a component:
|
|
- Update to one of the candidates (sets tag or rev) and optionally re-hash
|
|
- Recompute hash (prefetch)
|
|
- Edit any field via path=value (e.g., variables.version=2025.07)
|
|
- Writes changes back to version.json
|
|
|
|
Dependencies:
|
|
- Standard library + external CLI tools:
|
|
- nix-prefetch-url (or `nix prefetch-url`) and `nix hash to-sri`
|
|
- nix-prefetch-git
|
|
- git
|
|
- Optional: GITHUB_TOKEN env var to increase GitHub API rate limits
|
|
|
|
Usage:
|
|
scripts/version_tui.py
|
|
Controls:
|
|
- Up/Down to navigate lists
|
|
- Enter to select
|
|
- Backspace to go back
|
|
- q to quit
|
|
- On component screen:
|
|
r = refresh candidates
|
|
h = recompute hash (prefetch)
|
|
e = edit arbitrary field (path=value)
|
|
s = save to disk
|
|
"""
|
|
|
|
import curses
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import traceback
|
|
import urllib.request
|
|
import urllib.error
|
|
from urllib.parse import urlparse
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PKGS_DIR = ROOT / "packages"
|
|
|
|
Json = Dict[str, Any]
|
|
|
|
# ------------------------------ Utilities ------------------------------
|
|
|
|
def eprintln(*args, **kwargs):
|
|
print(*args, file=sys.stderr, **kwargs)
|
|
|
|
def load_json(path: Path) -> Json:
|
|
with path.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
def save_json(path: Path, data: Json):
|
|
tmp = path.with_suffix(".tmp")
|
|
with tmp.open("w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
f.write("\n")
|
|
tmp.replace(path)
|
|
|
|
def render_templates(value: Any, variables: Dict[str, Any]) -> Any:
|
|
if isinstance(value, str):
|
|
def repl(m):
|
|
name = m.group(1)
|
|
return str(variables.get(name, m.group(0)))
|
|
return re.sub(r"\$\{([^}]+)\}", repl, value)
|
|
elif isinstance(value, dict):
|
|
return {k: render_templates(v, variables) for k, v in value.items()}
|
|
elif isinstance(value, list):
|
|
return [render_templates(v, variables) for v in value]
|
|
return value
|
|
|
|
def deep_set(o: Json, path: List[str], value: Any):
|
|
cur = o
|
|
for p in path[:-1]:
|
|
if p not in cur or not isinstance(cur[p], dict):
|
|
cur[p] = {}
|
|
cur = cur[p]
|
|
cur[path[-1]] = value
|
|
|
|
# ------------------------------ Merge helpers (match lib/versioning.nix) ------------------------------
|
|
|
|
def deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
|
|
out = dict(a)
|
|
for k, v in b.items():
|
|
if k in out and isinstance(out[k], dict) and isinstance(v, dict):
|
|
out[k] = deep_merge(out[k], v)
|
|
else:
|
|
out[k] = v
|
|
return out
|
|
|
|
def merge_sources(base_sources: Dict[str, Any], overrides: Dict[str, Any]) -> Dict[str, Any]:
|
|
names = set(base_sources.keys()) | set(overrides.keys())
|
|
result: Dict[str, Any] = {}
|
|
for n in names:
|
|
if n in base_sources and n in overrides:
|
|
if isinstance(base_sources[n], dict) and isinstance(overrides[n], dict):
|
|
result[n] = deep_merge(base_sources[n], overrides[n])
|
|
else:
|
|
result[n] = overrides[n]
|
|
elif n in overrides:
|
|
result[n] = overrides[n]
|
|
else:
|
|
result[n] = base_sources[n]
|
|
return result
|
|
|
|
def merged_view(spec: Json, variant_name: Optional[str]) -> Tuple[Dict[str, Any], Dict[str, Any], Json]:
|
|
"""
|
|
Returns (merged_variables, merged_sources, target_dict_to_write)
|
|
merged_* are used for display/prefetch; target_dict_to_write is where updates must be written (base or selected variant).
|
|
"""
|
|
base_vars = spec.get("variables", {}) or {}
|
|
base_sources = spec.get("sources", {}) or {}
|
|
if variant_name:
|
|
vdict = spec.get("variants", {}).get(variant_name)
|
|
if not isinstance(vdict, dict):
|
|
raise ValueError(f"Variant '{variant_name}' not found")
|
|
v_vars = vdict.get("variables", {}) or {}
|
|
v_sources = vdict.get("sources", {}) or {}
|
|
merged_vars = dict(base_vars); merged_vars.update(v_vars)
|
|
merged_srcs = merge_sources(base_sources, v_sources)
|
|
return merged_vars, merged_srcs, vdict
|
|
else:
|
|
return dict(base_vars), dict(base_sources), spec
|
|
|
|
def run_cmd(args: List[str]) -> Tuple[int, str, str]:
|
|
try:
|
|
p = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
|
|
return p.returncode, p.stdout.strip(), p.stderr.strip()
|
|
except Exception as e:
|
|
return 1, "", str(e)
|
|
|
|
def run_get_stdout(args: List[str]) -> Optional[str]:
|
|
code, out, err = run_cmd(args)
|
|
if code != 0:
|
|
eprintln(f"Command failed: {' '.join(args)}\n{err}")
|
|
return None
|
|
return out
|
|
|
|
def nix_prefetch_url(url: str) -> Optional[str]:
|
|
# returns SRI
|
|
out = run_get_stdout(["nix-prefetch-url", "--type", "sha256", url])
|
|
if out is None:
|
|
out = run_get_stdout(["nix", "prefetch-url", url])
|
|
if out is None:
|
|
return None
|
|
sri = run_get_stdout(["nix", "hash", "to-sri", "--type", "sha256", out.strip()])
|
|
return sri
|
|
|
|
def nix_prefetch_git(url: str, rev: str) -> Optional[str]:
|
|
out = run_get_stdout(["nix-prefetch-git", "--no-deepClone", "--rev", rev, url])
|
|
if out is None:
|
|
return None
|
|
base32 = None
|
|
try:
|
|
data = json.loads(out)
|
|
base32 = data.get("sha256") or data.get("hash")
|
|
except Exception:
|
|
lines = [l for l in out.splitlines() if l.strip()]
|
|
if lines:
|
|
base32 = lines[-1].strip()
|
|
if not base32:
|
|
return None
|
|
sri = run_get_stdout(["nix", "hash", "to-sri", "--type", "sha256", base32])
|
|
return sri
|
|
|
|
def http_get_json(url: str, token: Optional[str] = None) -> Any:
|
|
req = urllib.request.Request(url, headers={"Accept": "application/vnd.github+json"})
|
|
if token:
|
|
req.add_header("Authorization", f"Bearer {token}")
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
def http_get_text(url: str) -> Optional[str]:
|
|
try:
|
|
# Provide a basic User-Agent to avoid some hosts rejecting the request
|
|
req = urllib.request.Request(url, headers={"User-Agent": "version-tui/1.0"})
|
|
with urllib.request.urlopen(req) as resp:
|
|
return resp.read().decode("utf-8")
|
|
except Exception as e:
|
|
eprintln(f"http_get_text failed for {url}: {e}")
|
|
return None
|
|
|
|
def gh_latest_release(owner: str, repo: str, token: Optional[str]) -> Optional[str]:
|
|
try:
|
|
data = http_get_json(f"https://api.github.com/repos/{owner}/{repo}/releases/latest", token)
|
|
return data.get("tag_name")
|
|
except Exception as e:
|
|
eprintln(f"latest_release failed for {owner}/{repo}: {e}")
|
|
return None
|
|
|
|
def gh_latest_tag(owner: str, repo: str, token: Optional[str]) -> Optional[str]:
|
|
try:
|
|
data = http_get_json(f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=100", token)
|
|
tags = [t.get("name") for t in data if "name" in t]
|
|
return tags[0] if tags else None
|
|
except Exception as e:
|
|
eprintln(f"latest_tag failed for {owner}/{repo}: {e}")
|
|
return None
|
|
|
|
def gh_list_tags(owner: str, repo: str, token: Optional[str]) -> List[str]:
|
|
try:
|
|
data = http_get_json(f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=100", token)
|
|
return [t.get("name") for t in data if isinstance(t, dict) and "name" in t]
|
|
except Exception as e:
|
|
eprintln(f"list_tags failed for {owner}/{repo}: {e}")
|
|
return []
|
|
|
|
def gh_head_commit(owner: str, repo: str) -> Optional[str]:
|
|
out = run_get_stdout(["git", "ls-remote", f"https://github.com/{owner}/{repo}.git", "HEAD"])
|
|
if not out:
|
|
return None
|
|
return out.split()[0]
|
|
|
|
def gh_tarball_url(owner: str, repo: str, ref: str) -> str:
|
|
return f"https://codeload.github.com/{owner}/{repo}/tar.gz/{ref}"
|
|
|
|
def gh_release_tags_api(owner: str, repo: str, token: Optional[str]) -> List[str]:
|
|
"""
|
|
Return recent release tag names for a repo using GitHub API.
|
|
"""
|
|
try:
|
|
data = http_get_json(f"https://api.github.com/repos/{owner}/{repo}/releases?per_page=50", token)
|
|
return [r.get("tag_name") for r in data if isinstance(r, dict) and "tag_name" in r]
|
|
except Exception as e:
|
|
eprintln(f"releases list failed for {owner}/{repo}: {e}")
|
|
return []
|
|
|
|
# ------------------------------ Data scanning ------------------------------
|
|
|
|
def find_packages() -> List[Tuple[str, Path]]:
|
|
results = []
|
|
for p in PKGS_DIR.rglob("version.json"):
|
|
# name is directory name under packages (e.g., raspberrypi/linux-rpi => raspberrypi/linux-rpi)
|
|
rel = p.relative_to(PKGS_DIR).parent
|
|
results.append((str(rel), p))
|
|
results.sort()
|
|
return results
|
|
|
|
# ------------------------------ TUI helpers ------------------------------
|
|
|
|
class ScreenBase:
|
|
def __init__(self, stdscr):
|
|
self.stdscr = stdscr
|
|
self.status = ""
|
|
|
|
def draw_status(self, height, width):
|
|
if self.status:
|
|
self.stdscr.addstr(height-1, 0, self.status[:max(0, width-1)])
|
|
else:
|
|
self.stdscr.addstr(height-1, 0, "q: quit, Backspace: back, Enter: select")
|
|
|
|
def set_status(self, text: str):
|
|
self.status = text
|
|
|
|
def run(self):
|
|
raise NotImplementedError
|
|
|
|
def prompt_input(stdscr, prompt: str) -> Optional[str]:
|
|
curses.echo()
|
|
stdscr.addstr(prompt)
|
|
stdscr.clrtoeol()
|
|
s = stdscr.getstr().decode("utf-8")
|
|
curses.noecho()
|
|
return s
|
|
|
|
def show_popup(stdscr, lines: List[str]):
|
|
h, w = stdscr.getmaxyx()
|
|
box_h = min(len(lines)+4, h-2)
|
|
box_w = min(max(len(l) for l in lines)+4, w-2)
|
|
top = (h - box_h)//2
|
|
left = (w - box_w)//2
|
|
win = curses.newwin(box_h, box_w, top, left)
|
|
win.box()
|
|
for i, line in enumerate(lines, start=1):
|
|
if i >= box_h-1:
|
|
break
|
|
win.addstr(i, 2, line[:box_w-3])
|
|
win.addstr(box_h-2, 2, "Press any key")
|
|
win.refresh()
|
|
win.getch()
|
|
|
|
# ------------------------------ Screens ------------------------------
|
|
|
|
class PackagesScreen(ScreenBase):
|
|
def __init__(self, stdscr):
|
|
super().__init__(stdscr)
|
|
self.packages = find_packages()
|
|
self.idx = 0
|
|
|
|
def run(self):
|
|
while True:
|
|
self.stdscr.clear()
|
|
h, w = self.stdscr.getmaxyx()
|
|
|
|
# Determine split layout
|
|
left_w = max(30, min(60, w // 3))
|
|
right_x = left_w + 1
|
|
right_w = max(0, w - right_x)
|
|
|
|
# Left pane: package list
|
|
self.stdscr.addstr(0, 0, "Packages (version.json)"[:max(0, left_w-1)])
|
|
max_rows = h - 3
|
|
for i, (name, _path) in enumerate(self.packages[:max_rows], start=0):
|
|
sel = ">" if i == self.idx else " "
|
|
self.stdscr.addstr(1 + i, 0, f"{sel} {name}"[:max(0, left_w-1)])
|
|
|
|
# Right pane: preview of selected package (non-interactive summary)
|
|
if right_w >= 20 and self.packages:
|
|
try:
|
|
name, path = self.packages[self.idx]
|
|
self.stdscr.addstr(0, right_x, f"{name}"[:max(0, right_w-1)])
|
|
self.stdscr.addstr(1, right_x, f"{path}"[:max(0, right_w-1)])
|
|
self.stdscr.addstr(2, right_x, "Sources:"[:max(0, right_w-1)])
|
|
|
|
spec = load_json(path)
|
|
merged_vars, merged_srcs, _ = merged_view(spec, None)
|
|
snames = sorted(list(merged_srcs.keys()))
|
|
max_src_rows = max(0, h - 6)
|
|
for i2, sname in enumerate(snames[:max_src_rows], start=0):
|
|
comp = merged_srcs[sname]
|
|
fetcher = comp.get("fetcher", "none")
|
|
# Construct concise reference similar to detail view
|
|
display_ref = comp.get("tag") or comp.get("rev") or comp.get("version") or ""
|
|
if fetcher == "github":
|
|
rendered = render_templates(comp, merged_vars)
|
|
tag = rendered.get("tag")
|
|
rev = rendered.get("rev")
|
|
owner = (rendered.get("owner") or merged_vars.get("owner") or "")
|
|
repo = (rendered.get("repo") or merged_vars.get("repo") or "")
|
|
if tag and owner and repo:
|
|
display_ref = f"{owner}/{repo}@{tag}"
|
|
elif tag:
|
|
display_ref = tag
|
|
elif rev and owner and repo:
|
|
display_ref = f"{owner}/{repo}@{rev[:7]}"
|
|
elif rev:
|
|
display_ref = rev[:12]
|
|
elif fetcher == "url":
|
|
rendered = render_templates(comp, merged_vars)
|
|
url = rendered.get("url") or rendered.get("urlTemplate") or ""
|
|
if url:
|
|
owner = str(merged_vars.get("owner", "") or "")
|
|
repo = str(merged_vars.get("repo", "") or "")
|
|
rp = str(merged_vars.get("releasePrefix", "") or "")
|
|
rs = str(merged_vars.get("releaseSuffix", "") or "")
|
|
base = str(merged_vars.get("base", "") or "")
|
|
rel = str(merged_vars.get("release", "") or "")
|
|
tag = f"{rp}{base}-{rel}{rs}" if (base and rel) else ""
|
|
parsed = urlparse(url)
|
|
filename = os.path.basename(parsed.path) if parsed and parsed.path else ""
|
|
if owner and repo and tag and filename:
|
|
display_ref = f"{owner}/{repo}@{tag} · {filename}"
|
|
elif filename:
|
|
display_ref = filename
|
|
else:
|
|
display_ref = url
|
|
else:
|
|
display_ref = ""
|
|
# Truncate reference to fit right pane
|
|
if isinstance(display_ref, str):
|
|
max_ref = max(0, right_w - 30)
|
|
ref_short = (display_ref[:max_ref] + ("..." if len(display_ref) > max_ref else ""))
|
|
else:
|
|
ref_short = display_ref
|
|
self.stdscr.addstr(3 + i2, right_x, f"{sname:<20} {fetcher:<7} {ref_short}"[:max(0, right_w-1)])
|
|
|
|
# Hint line for workflow
|
|
hint = "Enter: open details | k/j: move | q: quit"
|
|
if h >= 5:
|
|
self.stdscr.addstr(h - 5, right_x, hint[:max(0, right_w-1)])
|
|
except Exception as e:
|
|
self.stdscr.addstr(2, right_x, f"Failed to load: {e}"[:max(0, right_w-1)])
|
|
|
|
self.draw_status(h, w)
|
|
self.stdscr.refresh()
|
|
ch = self.stdscr.getch()
|
|
if ch in (ord('q'), 27): # q or ESC
|
|
return None
|
|
elif ch in (curses.KEY_UP, ord('k')):
|
|
self.idx = max(0, self.idx-1)
|
|
elif ch in (curses.KEY_DOWN, ord('j')):
|
|
self.idx = min(len(self.packages)-1, self.idx+1)
|
|
elif ch in (curses.KEY_ENTER, 10, 13):
|
|
if not self.packages:
|
|
continue
|
|
name, path = self.packages[self.idx]
|
|
try:
|
|
spec = load_json(path)
|
|
except Exception as e:
|
|
self.set_status(f"Failed to load {path}: {e}")
|
|
continue
|
|
screen = PackageDetailScreen(self.stdscr, name, path, spec)
|
|
ret = screen.run()
|
|
if ret == "reload":
|
|
# re-scan on save
|
|
self.packages = find_packages()
|
|
self.idx = min(self.idx, len(self.packages)-1)
|
|
else:
|
|
pass
|
|
|
|
class PackageDetailScreen(ScreenBase):
|
|
def __init__(self, stdscr, pkg_name: str, path: Path, spec: Json):
|
|
super().__init__(stdscr)
|
|
self.pkg_name = pkg_name
|
|
self.path = path
|
|
self.spec = spec
|
|
self.variants = ["<base>"] + sorted(list(self.spec.get("variants", {}).keys()))
|
|
self.vidx = 0
|
|
self.gh_token = os.environ.get("GITHUB_TOKEN")
|
|
self.candidates: Dict[str, Dict[str, str]] = {} # name -> {release, tag, commit}
|
|
self.url_candidates: Dict[str, Dict[str, str]] = {} # name -> {base, release, tag}
|
|
# initialize view
|
|
self.recompute_view()
|
|
|
|
def select_variant(self):
|
|
# Recompute merged and target views when variant changes
|
|
self.recompute_view()
|
|
|
|
def recompute_view(self):
|
|
# Set cursor to base or selected variant dict for manual edits
|
|
if self.vidx == 0:
|
|
self.cursor = self.spec
|
|
variant_name = None
|
|
else:
|
|
variant_name = self.variants[self.vidx]
|
|
self.cursor = self.spec["variants"][variant_name]
|
|
# Compute merged view and target dict for writing
|
|
self.merged_vars, self.merged_srcs, self.target_dict = merged_view(self.spec, variant_name)
|
|
self.snames = sorted(list(self.merged_srcs.keys()))
|
|
self.sidx = 0
|
|
|
|
def fetch_candidates_for(self, name: str):
|
|
comp = self.merged_srcs[name]
|
|
fetcher = comp.get("fetcher", "none")
|
|
c = {"release": "", "tag": "", "commit": ""}
|
|
if fetcher == "github":
|
|
owner = comp.get("owner")
|
|
repo = comp.get("repo")
|
|
if owner and repo:
|
|
r = gh_latest_release(owner, repo, self.gh_token)
|
|
if r:
|
|
c["release"] = r
|
|
t = gh_latest_tag(owner, repo, self.gh_token)
|
|
if t:
|
|
c["tag"] = t
|
|
m = gh_head_commit(owner, repo)
|
|
if m:
|
|
c["commit"] = m
|
|
|
|
# Special-case raspberrypi/linux: prefer latest stable_* tag or series-specific tags
|
|
try:
|
|
if owner == "raspberrypi" and repo == "linux":
|
|
tags_all = gh_list_tags(owner, repo, self.gh_token)
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
cur_tag = str(rendered.get("tag") or "")
|
|
# If current tag uses stable_YYYYMMDD scheme, pick latest stable_* tag
|
|
if cur_tag.startswith("stable_"):
|
|
stable_tags = sorted(
|
|
[x for x in tags_all if re.match(r"^stable_\d{8}$", x)],
|
|
reverse=True,
|
|
)
|
|
if stable_tags:
|
|
c["tag"] = stable_tags[0]
|
|
else:
|
|
# Try to pick a tag matching the current major.minor series if available
|
|
mm = str(self.merged_vars.get("modDirVersion") or "")
|
|
m2 = re.match(r"^(\d+)\.(\d+)", mm)
|
|
if m2:
|
|
base = f"rpi-{m2.group(1)}.{m2.group(2)}"
|
|
series_tags = [x for x in tags_all if (
|
|
x == f"{base}.y"
|
|
or x.startswith(f"{base}.y")
|
|
or x.startswith(f"{base}.")
|
|
)]
|
|
series_tags.sort(reverse=True)
|
|
if series_tags:
|
|
c["tag"] = series_tags[0]
|
|
except Exception as _e:
|
|
# Fallback to previously computed values
|
|
pass
|
|
elif fetcher == "git":
|
|
url = comp.get("url")
|
|
if url:
|
|
out = run_get_stdout(["git", "ls-remote", url, "HEAD"])
|
|
if out:
|
|
c["commit"] = out.split()[0]
|
|
elif fetcher == "url":
|
|
# Heuristic for GitHub release assets with variables in version.json (e.g., proton-cachyos)
|
|
owner = self.merged_vars.get("owner")
|
|
repo = self.merged_vars.get("repo")
|
|
if owner and repo:
|
|
tags = gh_release_tags_api(str(owner), str(repo), self.gh_token)
|
|
prefix = str(self.merged_vars.get("releasePrefix", ""))
|
|
suffix = str(self.merged_vars.get("releaseSuffix", ""))
|
|
latest = next((t for t in tags if (t and t.startswith(prefix) and t.endswith(suffix))), None)
|
|
if latest:
|
|
c["release"] = latest
|
|
mid = latest
|
|
if prefix and mid.startswith(prefix):
|
|
mid = mid[len(prefix):]
|
|
if suffix and mid.endswith(suffix):
|
|
mid = mid[:-len(suffix)]
|
|
parts = mid.split("-")
|
|
if len(parts) >= 2:
|
|
base, rel = parts[0], parts[-1]
|
|
self.url_candidates[name] = {"base": base, "release": rel, "tag": latest}
|
|
self.candidates[name] = c
|
|
|
|
def prefetch_hash_for(self, name: str) -> Optional[str]:
|
|
comp = self.merged_srcs[name]
|
|
fetcher = comp.get("fetcher", "none")
|
|
if fetcher == "github":
|
|
owner = comp.get("owner")
|
|
repo = comp.get("repo")
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
ref = rendered.get("tag") or rendered.get("rev")
|
|
if owner and repo and ref:
|
|
url = gh_tarball_url(owner, repo, ref)
|
|
return nix_prefetch_url(url)
|
|
elif fetcher == "git":
|
|
url = comp.get("url")
|
|
rev = comp.get("rev")
|
|
if url and rev:
|
|
return nix_prefetch_git(url, rev)
|
|
elif fetcher == "url":
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
url = rendered.get("url") or rendered.get("urlTemplate")
|
|
if url:
|
|
return nix_prefetch_url(url)
|
|
return None
|
|
|
|
def cachyos_suffix(self) -> str:
|
|
if self.vidx == 0:
|
|
return ""
|
|
v = self.variants[self.vidx]
|
|
mapping = {"rc": "-rc", "hardened": "-hardened", "lts": "-lts"}
|
|
return mapping.get(v, "")
|
|
|
|
def fetch_cachyos_linux_latest(self, suffix: str) -> Optional[str]:
|
|
"""
|
|
Try to determine latest linux version from upstream:
|
|
- Prefer .SRCINFO (preprocessed)
|
|
- Fallback to PKGBUILD (parse pkgver= line)
|
|
Tries both 'CachyOS' and 'cachyos' org casing just in case.
|
|
"""
|
|
bases = [
|
|
"https://raw.githubusercontent.com/CachyOS/linux-cachyos/master",
|
|
"https://raw.githubusercontent.com/cachyos/linux-cachyos/master",
|
|
]
|
|
paths = [
|
|
f"linux-cachyos{suffix}/.SRCINFO",
|
|
f"linux-cachyos{suffix}/PKGBUILD",
|
|
]
|
|
|
|
def parse_srcinfo(text: str) -> Optional[str]:
|
|
m = re.search(r"^\s*pkgver\s*=\s*([^\s#]+)\s*$", text, re.MULTILINE)
|
|
if not m:
|
|
return None
|
|
v = m.group(1).strip()
|
|
return v
|
|
|
|
def parse_pkgbuild(text: str) -> Optional[str]:
|
|
# Parse assignments and expand variables in pkgver
|
|
# Build a simple env map from VAR=value lines
|
|
env: Dict[str, str] = {}
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
m_assign = re.match(r'^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.+)$', line)
|
|
if m_assign:
|
|
key = m_assign.group(1)
|
|
val = m_assign.group(2).strip()
|
|
# Remove trailing comments
|
|
val = re.sub(r'\s+#.*$', '', val).strip()
|
|
# Strip surrounding quotes
|
|
if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
|
|
val = val[1:-1]
|
|
env[key] = val
|
|
|
|
m = re.search(r"^\s*pkgver\s*=\s*(.+)$", text, re.MULTILINE)
|
|
if not m:
|
|
return None
|
|
raw = m.group(1).strip()
|
|
# Strip quotes
|
|
if (raw.startswith('"') and raw.endswith('"')) or (raw.startswith("'") and raw.endswith("'")):
|
|
raw = raw[1:-1]
|
|
|
|
def expand_vars(s: str) -> str:
|
|
def repl_braced(mb):
|
|
key = mb.group(1)
|
|
return env.get(key, mb.group(0))
|
|
def repl_unbraced(mu):
|
|
key = mu.group(1)
|
|
return env.get(key, mu.group(0))
|
|
# Expand ${var} then $var
|
|
s = re.sub(r"\$\{([^}]+)\}", repl_braced, s)
|
|
s = re.sub(r"\$([A-Za-z_][A-Za-z0-9_]*)", repl_unbraced, s)
|
|
return s
|
|
|
|
v = expand_vars(raw).strip()
|
|
# normalize rc form like 6.19.rc6 -> 6.19-rc6
|
|
v = v.replace(".rc", "-rc")
|
|
return v
|
|
|
|
# Try .SRCINFO first, then PKGBUILD
|
|
for base in bases:
|
|
# .SRCINFO
|
|
url = f"{base}/{paths[0]}"
|
|
text = http_get_text(url)
|
|
if text:
|
|
ver = parse_srcinfo(text)
|
|
if ver:
|
|
return ver.replace(".rc", "-rc")
|
|
# PKGBUILD fallback
|
|
url = f"{base}/{paths[1]}"
|
|
text = http_get_text(url)
|
|
if text:
|
|
ver = parse_pkgbuild(text)
|
|
if ver:
|
|
return ver.replace(".rc", "-rc")
|
|
|
|
return None
|
|
|
|
def linux_tarball_url_for_version(self, version: str) -> str:
|
|
# Use torvalds snapshot for -rc, stable releases from CDN
|
|
if "-rc" in version:
|
|
return f"https://git.kernel.org/torvalds/t/linux-{version}.tar.gz"
|
|
parts = version.split(".")
|
|
major = parts[0] if parts else "6"
|
|
major_minor = ".".join(parts[:2]) if len(parts) >= 2 else version
|
|
ver_for_tar = major_minor if version.endswith(".0") else version
|
|
return f"https://cdn.kernel.org/pub/linux/kernel/v{major}.x/linux-{ver_for_tar}.tar.xz"
|
|
|
|
def update_linux_from_pkgbuild(self, name: str):
|
|
suffix = self.cachyos_suffix()
|
|
latest = self.fetch_cachyos_linux_latest(suffix)
|
|
if not latest:
|
|
self.set_status("linux: failed to get version from PKGBUILD")
|
|
return
|
|
url = self.linux_tarball_url_for_version(latest)
|
|
sri = nix_prefetch_url(url)
|
|
if not sri:
|
|
self.set_status("linux: prefetch failed")
|
|
return
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["version"] = latest
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: updated version to {latest} and refreshed hash")
|
|
|
|
def set_ref(self, name: str, kind: str, value: str):
|
|
# Write to selected target dict (base or variant override)
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
comp = ts.setdefault(name, {})
|
|
if kind in ("release", "tag"):
|
|
comp["tag"] = value
|
|
if "rev" in comp:
|
|
del comp["rev"]
|
|
elif kind == "commit":
|
|
comp["rev"] = value
|
|
if "tag" in comp:
|
|
del comp["tag"]
|
|
|
|
def save(self):
|
|
save_json(self.path, self.spec)
|
|
|
|
def run(self):
|
|
while True:
|
|
self.stdscr.clear()
|
|
h, w = self.stdscr.getmaxyx()
|
|
title = f"{self.pkg_name} [{self.path}]"
|
|
self.stdscr.addstr(0, 0, title[:w-1])
|
|
# Variant line
|
|
vline = "Variants: " + " | ".join(
|
|
[f"[{v}]" if i == self.vidx else v for i, v in enumerate(self.variants)]
|
|
)
|
|
self.stdscr.addstr(1, 0, vline[:w-1])
|
|
# Sources header
|
|
self.stdscr.addstr(2, 0, "Sources:")
|
|
# List sources
|
|
for i, name in enumerate(self.snames[:h-8], start=0):
|
|
comp = self.merged_srcs[name]
|
|
fetcher = comp.get("fetcher", "none")
|
|
# Render refs so variables resolve; compress long forms for display
|
|
display_ref = comp.get("tag") or comp.get("rev") or comp.get("version") or ""
|
|
if fetcher == "github":
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
tag = rendered.get("tag")
|
|
rev = rendered.get("rev")
|
|
owner = (rendered.get("owner") or self.merged_vars.get("owner") or "")
|
|
repo = (rendered.get("repo") or self.merged_vars.get("repo") or "")
|
|
if tag and owner and repo:
|
|
display_ref = f"{owner}/{repo}@{tag}"
|
|
elif tag:
|
|
display_ref = tag
|
|
elif rev and owner and repo:
|
|
display_ref = f"{owner}/{repo}@{rev[:7]}"
|
|
elif rev:
|
|
display_ref = rev[:12]
|
|
elif fetcher == "url":
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
url = rendered.get("url") or rendered.get("urlTemplate") or ""
|
|
if url:
|
|
# Prefer a concise label like owner/repo@tag · filename
|
|
owner = str(self.merged_vars.get("owner", "") or "")
|
|
repo = str(self.merged_vars.get("repo", "") or "")
|
|
rp = str(self.merged_vars.get("releasePrefix", "") or "")
|
|
rs = str(self.merged_vars.get("releaseSuffix", "") or "")
|
|
base = str(self.merged_vars.get("base", "") or "")
|
|
rel = str(self.merged_vars.get("release", "") or "")
|
|
tag = f"{rp}{base}-{rel}{rs}" if (base and rel) else ""
|
|
parsed = urlparse(url)
|
|
filename = os.path.basename(parsed.path) if parsed and parsed.path else ""
|
|
if owner and repo and tag and filename:
|
|
display_ref = f"{owner}/{repo}@{tag} · {filename}"
|
|
elif filename:
|
|
display_ref = filename
|
|
else:
|
|
display_ref = url
|
|
else:
|
|
display_ref = ""
|
|
ref_short = display_ref if not isinstance(display_ref, str) else (display_ref[:60] + ("..." if len(display_ref) > 60 else ""))
|
|
sel = ">" if i == self.sidx else " "
|
|
self.stdscr.addstr(3+i, 0, f"{sel} {name:<20} {fetcher:<7} ref={ref_short}"[:w-1])
|
|
|
|
# Latest candidates section for selected component (auto-fetched)
|
|
if self.snames:
|
|
_sel_name = self.snames[self.sidx]
|
|
_comp = self.merged_srcs[_sel_name]
|
|
_fetcher = _comp.get("fetcher", "none")
|
|
# Preload candidates lazily for selected item
|
|
if _fetcher in ("github", "git", "url") and _sel_name not in self.candidates:
|
|
self.fetch_candidates_for(_sel_name)
|
|
y_latest = h - 6
|
|
if y_latest >= 3:
|
|
self.stdscr.addstr(y_latest, 0, "Latest:"[:w-1])
|
|
if _fetcher in ("github", "git"):
|
|
_cand = self.candidates.get(_sel_name, {})
|
|
_line = f" release={_cand.get('release') or '-'} tag={_cand.get('tag') or '-'} commit={( (_cand.get('commit') or '')[:12] ) or '-'}"
|
|
self.stdscr.addstr(y_latest+1, 0, _line[:w-1])
|
|
elif _fetcher == "url":
|
|
_cand_u = self.url_candidates.get(_sel_name, {}) or {}
|
|
_tag = _cand_u.get("tag") or (self.candidates.get(_sel_name, {}).get("release") or "-")
|
|
_line = f" tag={_tag} base={_cand_u.get('base') or '-'} release={_cand_u.get('release') or '-'}"
|
|
self.stdscr.addstr(y_latest+1, 0, _line[:w-1])
|
|
else:
|
|
if self.pkg_name == "linux-cachyos" and _sel_name == "linux":
|
|
_suffix = self.cachyos_suffix()
|
|
_latest = self.fetch_cachyos_linux_latest(_suffix)
|
|
self.stdscr.addstr(y_latest+1, 0, f" linux from PKGBUILD: {_latest or '-'}"[:w-1])
|
|
else:
|
|
self.stdscr.addstr(y_latest+1, 0, " -"[:w-1])
|
|
|
|
# Footer instructions
|
|
self.stdscr.addstr(h-4, 0, "Enter: component actions | r: refresh candidates | h: prefetch hash | e: edit field | s: save | Backspace: back | q: quit")
|
|
self.draw_status(h, w)
|
|
self.stdscr.refresh()
|
|
|
|
ch = self.stdscr.getch()
|
|
if ch in (ord('q'), 27):
|
|
return None
|
|
elif ch == curses.KEY_BACKSPACE or ch == 127:
|
|
return "reload"
|
|
elif ch in (curses.KEY_LEFT, ord('h')):
|
|
self.vidx = max(0, self.vidx-1)
|
|
self.select_variant()
|
|
elif ch in (curses.KEY_RIGHT, ord('l')):
|
|
self.vidx = min(len(self.variants)-1, self.vidx+1)
|
|
self.select_variant()
|
|
elif ch in (curses.KEY_UP, ord('k')):
|
|
self.sidx = max(0, self.sidx-1)
|
|
elif ch in (curses.KEY_DOWN, ord('j')):
|
|
self.sidx = min(len(self.snames)-1, self.sidx+1)
|
|
elif ch in (ord('r'),):
|
|
if self.snames:
|
|
name = self.snames[self.sidx]
|
|
comp = self.merged_srcs[name]
|
|
fetcher = comp.get("fetcher", "none")
|
|
if self.pkg_name == "linux-cachyos" and name == "linux":
|
|
# Show available linux version from upstream PKGBUILD (.SRCINFO)
|
|
suffix = self.cachyos_suffix()
|
|
latest = self.fetch_cachyos_linux_latest(suffix)
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
cur_version = str(rendered.get("version") or "")
|
|
url_hint = self.linux_tarball_url_for_version(latest) if latest else "-"
|
|
lines = [
|
|
f"linux-cachyos ({'base' if self.vidx == 0 else self.variants[self.vidx]}):",
|
|
f" current : {cur_version or '-'}",
|
|
f" available: {latest or '-'}",
|
|
f" tarball : {url_hint}",
|
|
]
|
|
show_popup(self.stdscr, lines)
|
|
else:
|
|
self.fetch_candidates_for(name)
|
|
cand = self.candidates.get(name, {})
|
|
lines = [
|
|
f"Candidates for {name}:",
|
|
f" latest release: {cand.get('release') or '-'}",
|
|
f" latest tag : {cand.get('tag') or '-'}",
|
|
f" latest commit : {cand.get('commit') or '-'}",
|
|
]
|
|
show_popup(self.stdscr, lines)
|
|
elif ch in (ord('i'),):
|
|
# Show full rendered URL for URL-based sources
|
|
if self.snames:
|
|
name = self.snames[self.sidx]
|
|
comp = self.merged_srcs[name]
|
|
if comp.get("fetcher", "none") == "url":
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
url = rendered.get("url") or rendered.get("urlTemplate") or ""
|
|
if url:
|
|
show_popup(self.stdscr, ["Full URL:", url])
|
|
else:
|
|
self.set_status("No URL available")
|
|
elif ch in (ord('h'),):
|
|
if self.snames:
|
|
name = self.snames[self.sidx]
|
|
sri = self.prefetch_hash_for(name)
|
|
if sri:
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: updated hash")
|
|
else:
|
|
self.set_status(f"{name}: hash prefetch failed")
|
|
elif ch in (ord('e'),):
|
|
s = prompt_input(self.stdscr, "Edit path=value (relative to selected base/variant): ")
|
|
if s:
|
|
if "=" not in s:
|
|
self.set_status("Invalid input, expected key.path=value")
|
|
else:
|
|
k, v = s.split("=", 1)
|
|
path = [p for p in k.split(".") if p]
|
|
deep_set(self.cursor, path, v)
|
|
self.set_status(f"Set {k}={v}")
|
|
elif ch in (ord('s'),):
|
|
try:
|
|
self.save()
|
|
self.set_status("Saved.")
|
|
except Exception as e:
|
|
self.set_status(f"Save failed: {e}")
|
|
elif ch in (curses.KEY_ENTER, 10, 13):
|
|
if not self.snames:
|
|
continue
|
|
name = self.snames[self.sidx]
|
|
comp = self.merged_srcs[name]
|
|
fetcher = comp.get("fetcher", "none")
|
|
if fetcher in ("github", "git"):
|
|
# Ensure candidates loaded
|
|
if name not in self.candidates:
|
|
self.fetch_candidates_for(name)
|
|
cand = self.candidates.get(name, {})
|
|
# Present small menu
|
|
items = []
|
|
if fetcher == "github":
|
|
items = [
|
|
("Use latest release (tag)", ("release", cand.get("release"))),
|
|
("Use latest tag", ("tag", cand.get("tag"))),
|
|
("Use latest commit (rev)", ("commit", cand.get("commit"))),
|
|
("Recompute hash", ("hash", None)),
|
|
("Cancel", ("cancel", None)),
|
|
]
|
|
else:
|
|
items = [
|
|
("Use latest commit (rev)", ("commit", cand.get("commit"))),
|
|
("Recompute hash", ("hash", None)),
|
|
("Cancel", ("cancel", None)),
|
|
]
|
|
# Build header with current and available refs
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
cur_tag = rendered.get("tag") or ""
|
|
cur_rev = rendered.get("rev") or ""
|
|
cur_version = rendered.get("version") or ""
|
|
if cur_tag:
|
|
current_str = f"current: tag={cur_tag}"
|
|
elif cur_rev:
|
|
current_str = f"current: rev={cur_rev[:12]}"
|
|
elif cur_version:
|
|
current_str = f"current: version={cur_version}"
|
|
else:
|
|
current_str = "current: -"
|
|
header_lines = [
|
|
current_str,
|
|
f"available: release={cand.get('release') or '-'} tag={cand.get('tag') or '-'} commit={(cand.get('commit') or '')[:12] or '-'}",
|
|
]
|
|
choice = select_menu(self.stdscr, f"Actions for {name}", [label for label, _ in items], header=header_lines)
|
|
if choice is not None:
|
|
kind, val = items[choice][1]
|
|
if kind in ("release", "tag", "commit"):
|
|
if val:
|
|
self.set_ref(name, kind, val)
|
|
# update hash
|
|
sri = self.prefetch_hash_for(name)
|
|
if sri:
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: set {kind} and updated hash")
|
|
else:
|
|
self.set_status(f"No candidate {kind}")
|
|
elif kind == "hash":
|
|
sri = self.prefetch_hash_for(name)
|
|
if sri:
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: updated hash")
|
|
else:
|
|
self.set_status("hash prefetch failed")
|
|
else:
|
|
pass
|
|
elif fetcher == "url":
|
|
# Offer latest release update (for proton-cachyos-like schemas) and/or hash recompute
|
|
cand = self.url_candidates.get(name)
|
|
menu_items: List[Tuple[str, Tuple[str, Optional[Dict[str, str]]]]] = []
|
|
if cand and cand.get("base") and cand.get("release"):
|
|
menu_items.append(("Use latest release (update variables.base/release)", ("update_vars", cand)))
|
|
menu_items.append(("Recompute hash (prefetch)", ("hash", None)))
|
|
menu_items.append(("Cancel", ("cancel", None)))
|
|
|
|
# Build header with current and available release info
|
|
base = str(self.merged_vars.get("base") or "")
|
|
rel = str(self.merged_vars.get("release") or "")
|
|
rp = str(self.merged_vars.get("releasePrefix") or "")
|
|
rs = str(self.merged_vars.get("releaseSuffix") or "")
|
|
current_tag = f"{rp}{base}-{rel}{rs}" if (base and rel) else ""
|
|
if current_tag:
|
|
current_str = f"current: {current_tag}"
|
|
elif base or rel:
|
|
current_str = f"current: base={base or '-'} release={rel or '-'}"
|
|
else:
|
|
current_str = "current: -"
|
|
header_lines = [
|
|
current_str,
|
|
f"available: tag={(cand.get('tag') or '-') if cand else '-'} base={(cand.get('base') or '-') if cand else '-'} release={(cand.get('release') or '-') if cand else '-'}",
|
|
]
|
|
choice = select_menu(self.stdscr, f"Actions for {name}", [label for label, _ in menu_items], header=header_lines)
|
|
if choice is not None:
|
|
kind, payload = menu_items[choice][1]
|
|
if kind == "update_vars" and isinstance(payload, dict):
|
|
# Write variables into selected base/variant dict
|
|
vars_dict = self.target_dict.setdefault("variables", {})
|
|
vars_dict["base"] = payload["base"]
|
|
vars_dict["release"] = payload["release"]
|
|
# Recompute merged view to reflect new variables
|
|
self.recompute_view()
|
|
# Prefetch and update hash
|
|
sri = self.prefetch_hash_for(name)
|
|
if sri:
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: updated to {payload['base']}.{payload['release']} and refreshed hash")
|
|
else:
|
|
self.set_status("hash prefetch failed after variable update")
|
|
elif kind == "hash":
|
|
sri = self.prefetch_hash_for(name)
|
|
if sri:
|
|
ts = self.target_dict.setdefault("sources", {})
|
|
compw = ts.setdefault(name, {})
|
|
compw["hash"] = sri
|
|
self.set_status(f"{name}: updated hash")
|
|
else:
|
|
self.set_status("hash prefetch failed")
|
|
else:
|
|
pass
|
|
else:
|
|
if self.pkg_name == "linux-cachyos" and name == "linux":
|
|
# Offer update of linux version from upstream PKGBUILD (.SRCINFO)
|
|
suffix = self.cachyos_suffix()
|
|
latest = self.fetch_cachyos_linux_latest(suffix)
|
|
rendered = render_templates(comp, self.merged_vars)
|
|
cur_version = str(rendered.get("version") or "")
|
|
header_lines = [
|
|
f"current: version={cur_version or '-'}",
|
|
f"available: version={latest or '-'}",
|
|
]
|
|
opts = []
|
|
if latest:
|
|
opts.append(f"Update linux version to {latest} from PKGBUILD (.SRCINFO)")
|
|
else:
|
|
opts.append("Update linux version from PKGBUILD (.SRCINFO)")
|
|
opts.append("Cancel")
|
|
choice = select_menu(self.stdscr, f"Actions for {name}", opts, header=header_lines)
|
|
if choice == 0 and latest:
|
|
self.update_linux_from_pkgbuild(name)
|
|
else:
|
|
pass
|
|
else:
|
|
show_popup(self.stdscr, [f"{name}: fetcher={fetcher}", "Use 'e' to edit fields manually."])
|
|
else:
|
|
pass
|
|
|
|
def select_menu(stdscr, title: str, options: List[str], header: Optional[List[str]] = None) -> Optional[int]:
|
|
idx = 0
|
|
while True:
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
stdscr.addstr(0, 0, title[:w-1])
|
|
y = 1
|
|
if header:
|
|
for line in header:
|
|
if y >= h-2:
|
|
break
|
|
stdscr.addstr(y, 0, str(line)[:w-1])
|
|
y += 1
|
|
start_y = y + 1
|
|
max_opts = max(0, h - start_y - 1)
|
|
for i, opt in enumerate(options[:max_opts], start=0):
|
|
sel = ">" if i == idx else " "
|
|
stdscr.addstr(start_y + i, 0, f"{sel} {opt}"[:w-1])
|
|
stdscr.addstr(h-1, 0, "Enter: select | Backspace: cancel")
|
|
stdscr.refresh()
|
|
ch = stdscr.getch()
|
|
if ch in (curses.KEY_UP, ord('k')):
|
|
idx = max(0, idx-1)
|
|
elif ch in (curses.KEY_DOWN, ord('j')):
|
|
idx = min(len(options)-1, idx+1)
|
|
elif ch in (curses.KEY_ENTER, 10, 13):
|
|
return idx
|
|
elif ch == curses.KEY_BACKSPACE or ch == 127 or ch == 27:
|
|
return None
|
|
|
|
# ------------------------------ main ------------------------------
|
|
|
|
def main(stdscr):
|
|
curses.curs_set(0)
|
|
stdscr.nodelay(False)
|
|
try:
|
|
screen = PackagesScreen(stdscr)
|
|
screen.run()
|
|
except Exception:
|
|
curses.endwin()
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
curses.wrapper(main)
|