admin
181bc0dc3c
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
231 lines
9.7 KiB
Python
231 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""check-currency.py — how far behind upstream is each pin, INCLUDING an upstream that changed its tag shape (R-731).
|
|
|
|
WHAT IT IS. The standing form of the 2026-09-30 catalog-currency audit
|
|
(`felhom.eu/documentation/audits/catalog-currency-2026-09-30/00-currency.py`, which needed `requests`; this needs only
|
|
the standard library, through `image_digest.py`'s registry client). It reads every pin from the templates
|
|
(`ladder.images_in` — the same reading the gates make), lists the repository's tags anonymously, and reports, per pin:
|
|
|
|
* SAME-SHAPE newest — within the pin's major and at all. Two tags compare only when the text between their numbers
|
|
is identical and they carry as many numbers (`16-alpine` only against `<N>-alpine`). This is what the badge and
|
|
every shape-based tool see.
|
|
* THE SHAPE-SWITCH CONTROL — the reason this file exists. The same-shape rule read three apps as up to date that were
|
|
not: gramps-web (`v25.6.0`; upstream dropped the `v` at `26.9.1`), jellyfin (`10.11.11`; 12.x publishes two-part
|
|
`12.1`), kimai (`apache-2.57.0`; plain `2.67.0`). In the audit the control was a one-off pass whose output survived
|
|
(`09-shape-switch-check.txt`) and whose code did not. Here it is standing: a RELEASE-LIKE tag under ANY shape whose
|
|
version core (the first three numbers of its leading version run) is higher than the pin's, while no same-shape tag
|
|
is, is reported `SHAPE-SWITCH?` — a question for a person, never an accusation. Release-like excludes what fooled
|
|
the first pass: unstable markers (`rc`, `beta`, `dev`, …), architecture-prefixed and date-rebuild tags (sonarr's
|
|
`amd64-5.14-…`), and branch tags (tandoor's `dependabot-pip-…`).
|
|
|
|
It accuses nothing and gates nothing (network, throttled registries): exit 0 when every pin was read, 2 when any could
|
|
not be (a throttle or an error is INCONCLUSIVE, never "up to date").
|
|
|
|
USAGE
|
|
python3 scripts/check-currency.py # every template
|
|
python3 scripts/check-currency.py kimai jellyfin # only these
|
|
python3 scripts/check-currency.py --json=<file> # also write the rows
|
|
Fixture tests (no network): scripts/test_check_currency.py.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(HERE)
|
|
sys.path.insert(0, HERE)
|
|
import image_digest # noqa: E402
|
|
import ladder # noqa: E402
|
|
|
|
UNSTABLE = ("rc", "beta", "alpha", "dev", "nightly", "canary", "edge", "preview", "snapshot", "-pr", "test",
|
|
"unstable")
|
|
HASH_RE = re.compile(r"(?<![0-9a-z])(?=[0-9a-f]*[a-f])(?=[0-9a-f]*[0-9])[0-9a-f]{7,40}(?![0-9a-z])")
|
|
# A release-like tag for the shape-switch control: an optional `v`, a dotted version run, and at most ONE short word
|
|
# flavour (`-alpine`, `-apache`, `-rootless`). Anything else (arch prefixes, branch names, build stamps) is not a
|
|
# release a household could be moved to without a person reading it.
|
|
RELEASE_RE = re.compile(r"^(?:[a-z]+-)?v?(\d+(?:\.\d+){0,3})(?:-[a-z]+)?$")
|
|
ARCH_WORDS = ("amd64", "arm64", "arm64v8", "armhf", "arm32v7", "armv7", "aarch64", "x86_64", "i386", "s390x",
|
|
"ppc64le")
|
|
|
|
|
|
def tokenize(tag):
|
|
"""(shape, numbers) or None when the tag carries no number."""
|
|
t = HASH_RE.sub("<hash>", tag.lower())
|
|
parts = re.split(r"(\d+)", t)
|
|
nums = tuple(int(p) for p in parts[1::2])
|
|
if not nums:
|
|
return None
|
|
return tuple(parts[0::2]), nums
|
|
|
|
|
|
def unstable(tag):
|
|
low = tag.lower()
|
|
return any(mk in low for mk in UNSTABLE)
|
|
|
|
|
|
def date_rebuild(nums):
|
|
return any(n >= 20000 for n in nums)
|
|
|
|
|
|
def newest_same_shape(tags, cur):
|
|
"""{'cur': nums, 'all': (tag, nums) | None, 'maj': (tag, nums) | None} or None when the pin has no number."""
|
|
c = tokenize(cur)
|
|
if not c:
|
|
return None
|
|
shape, nums = c
|
|
best_all = best_maj = None
|
|
for t in tags:
|
|
if unstable(t) and not unstable(cur):
|
|
continue
|
|
p = tokenize(t)
|
|
if not p or p[0] != shape or len(p[1]) != len(nums):
|
|
continue
|
|
if date_rebuild(p[1]) and not date_rebuild(nums):
|
|
continue
|
|
if best_all is None or p[1] > best_all[1]:
|
|
best_all = (t, p[1])
|
|
if p[1][0] == nums[0] and (best_maj is None or p[1] > best_maj[1]):
|
|
best_maj = (t, p[1])
|
|
return {"cur": nums, "all": best_all, "maj": best_maj}
|
|
|
|
|
|
def core(tag):
|
|
"""The version core of a RELEASE-LIKE tag (first three numbers of its leading version run, zero-padded), or None."""
|
|
low = tag.lower()
|
|
if unstable(low):
|
|
return None
|
|
m = RELEASE_RE.match(low)
|
|
if not m:
|
|
return None
|
|
prefix = low.split("-", 1)[0] if re.match(r"^[a-z]+-", low) else ""
|
|
if prefix in ARCH_WORDS or prefix == "v":
|
|
return None
|
|
nums = tuple(int(x) for x in m.group(1).split("."))
|
|
if date_rebuild(nums):
|
|
return None
|
|
return (nums + (0, 0, 0))[:3]
|
|
|
|
|
|
def pin_core(tag):
|
|
"""The pin's own core: the first dotted run in the tag, zero-padded (the pin need not be release-like itself)."""
|
|
m = re.search(r"\d+(?:\.\d+)*", tag)
|
|
if not m:
|
|
return None
|
|
nums = tuple(int(x) for x in m.group(0).split("."))
|
|
return (nums + (0, 0, 0))[:3]
|
|
|
|
|
|
def shape_switch(tags, cur):
|
|
"""(tag, core) of the highest release-like tag under ANY shape that is above the pin's core while NO same-shape tag
|
|
is above the pin — else None. That is the case the same-shape rule reads as up to date."""
|
|
same = newest_same_shape(tags, cur)
|
|
pc = pin_core(cur)
|
|
if same is None or pc is None:
|
|
return None
|
|
if same["all"] and same["all"][1] > same["cur"]:
|
|
return None # the same-shape rule already sees a newer release
|
|
best = None
|
|
for t in tags:
|
|
c = core(t)
|
|
if c and c > pc and (best is None or c > best[1]):
|
|
best = (t, c)
|
|
return best
|
|
|
|
|
|
# ── the registry (network; never reached by the tests) ───────────────────────────────────────────────────────────
|
|
|
|
def _get(url, token=None, timeout=30):
|
|
h = {"User-Agent": image_digest.UA}
|
|
if token:
|
|
h["Authorization"] = "Bearer " + token
|
|
req = urllib.request.Request(url, headers=h)
|
|
return urllib.request.urlopen(req, timeout=timeout)
|
|
|
|
|
|
def tags_all(host, repo, max_pages=200):
|
|
url = "https://%s/v2/%s/tags/list?n=1000" % (host, repo)
|
|
token = None
|
|
try:
|
|
r = _get(url)
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 401 or "WWW-Authenticate" not in e.headers:
|
|
raise
|
|
token = image_digest._bearer(e.headers["WWW-Authenticate"])
|
|
r = _get(url, token)
|
|
tags, pages = [], 0
|
|
while True:
|
|
with r:
|
|
body = json.load(r)
|
|
link = r.headers.get("Link")
|
|
tags.extend(body.get("tags") or [])
|
|
pages += 1
|
|
m = re.search(r'<([^>]+)>;\s*rel="next"', link or "")
|
|
if not m or pages >= max_pages:
|
|
break
|
|
nxt = m.group(1)
|
|
if nxt.startswith("/"):
|
|
nxt = "https://%s%s" % (host, nxt)
|
|
r = _get(nxt, token)
|
|
return tags
|
|
|
|
|
|
def main(argv):
|
|
out_json = None
|
|
apps = []
|
|
for a in argv:
|
|
if a.startswith("--json="):
|
|
out_json = a.split("=", 1)[1]
|
|
elif a.startswith("-"):
|
|
print("unknown option: %s" % a)
|
|
return 2
|
|
else:
|
|
apps.append(a)
|
|
tdir = os.path.join(ROOT, "templates")
|
|
every = sorted(d for d in os.listdir(tdir) if os.path.isfile(os.path.join(tdir, d, "docker-compose.yml")))
|
|
rows, cache, errors = [], {}, 0
|
|
for app in (apps or every):
|
|
imgs = ladder.images_in(open(os.path.join(tdir, app, "docker-compose.yml"), encoding="utf-8").read())
|
|
for svc, ref in imgs.items():
|
|
host, repo, tag = image_digest.split_ref(ref)
|
|
row = {"app": app, "service": svc, "ref": ref}
|
|
if host.startswith("gitea.dooplex.hu"):
|
|
row["status"] = "internal"
|
|
rows.append(row)
|
|
print("%-18s %-22s internal image, not upstream" % (app, svc))
|
|
continue
|
|
try:
|
|
if (host, repo) not in cache:
|
|
cache[(host, repo)] = tags_all(host, repo)
|
|
time.sleep(0.15)
|
|
tags = cache[(host, repo)]
|
|
same = newest_same_shape(tags, tag)
|
|
sw = shape_switch(tags, tag)
|
|
row.update(status="ok", n_tags=len(tags),
|
|
newest_major=same["maj"][0] if same and same["maj"] else None,
|
|
newest_all=same["all"][0] if same and same["all"] else None,
|
|
shape_switch=sw[0] if sw else None)
|
|
verdict = "SHAPE-SWITCH? %s" % sw[0] if sw else (
|
|
"behind" if same and same["all"] and same["all"][1] > same["cur"] else "current")
|
|
print("%-18s %-22s %-28s same-shape: major %s, all %s — %s" % (
|
|
app, svc, tag, row["newest_major"], row["newest_all"], verdict))
|
|
except Exception as e: # recorded, never guessed
|
|
errors += 1
|
|
row.update(status="error", error=("%s: %s" % (type(e).__name__, e))[:300])
|
|
print("%-18s %-22s %-28s INCONCLUSIVE: %s" % (app, svc, tag, row["error"]))
|
|
rows.append(row)
|
|
if out_json:
|
|
json.dump(rows, open(out_json, "w"), indent=1)
|
|
switches = [r for r in rows if r.get("shape_switch")]
|
|
print("\ncheck-currency: %d pin(s); %d shape switch(es) to read by a person; %d INCONCLUSIVE"
|
|
% (len(rows), len(switches), errors))
|
|
return 2 if errors else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|