Gabriel Vidal
604ba94cf3
Projects and services can keep a GOAL.md (the north star the goal-keeper cron
agent pushes forward every 5h), but nothing in the viewer showed it existed.
- backend/goal.py: shared parser for both catalogs. goal_summary() -> the
{done,total} checklist rollup that tags a card; goal_detail() adds the
markdown for the detail page. Counting skips fenced code blocks so a "- [ ]"
inside a snippet isn't mistaken for a task. No GOAL.md => goal: null.
- projects.py / svc.py: goal on the summary + detail records, GOAL.md added to
KEY_FILES.
- Cards get an x/n tag (emerald once complete; a GOAL.md with no checklist
still tags, since the point is that the file exists).
- Detail pages get a Goal button in the header that jumps to a new GOAL.md
section (progress bar + the rendered markdown, task lists as checkboxes).
- Mock seed carries the goal states (partial / complete / no-checklist / absent)
so every visual state is reachable with ?mock=1.
- CLAUDE.md: document goal.py, and fix the openapi re-dump command — it needs
--entrypoint python (the image entrypoint otherwise boots the server) and
os._exit(0) (importing main starts threads that block a normal exit).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
625 lines
21 KiB
Python
625 lines
21 KiB
Python
"""Homelab services catalog — mined from the repo's ``services/`` dir.
|
|
|
|
A sibling of :mod:`projects`, but for the *infrastructure* side of the repo:
|
|
every directory under ``SERVICES_DIR`` is a self-hosted service made of a Docker
|
|
Compose stack (``docker-compose.yml`` / ``compose*.yml``) and/or a Traefik file
|
|
provider config (``traefik.yml``). This module reads those files (never Docker
|
|
itself — pure static parsing) and surfaces, per service:
|
|
|
|
* the **containers** it defines (image, profiles, ports, volumes, env *keys*
|
|
only — never values — depends-on, networks),
|
|
* the **Traefik routers/services** and the public **URLs** they expose,
|
|
* the **middleware / auth chain** guarding it,
|
|
* a read-only **file tree** (secrets elided) with text-file previews,
|
|
* README / CLAUDE.md, notable key files and recent git commits.
|
|
|
|
Descriptions and icons are enriched from the homepage dashboard config
|
|
(``homepage/config/services.yaml``) keyed by container name / host.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import githist
|
|
import goal as goalmd
|
|
|
|
try: # PyYAML ships with uvicorn[standard]; degrade gracefully if absent.
|
|
import yaml
|
|
except Exception: # pragma: no cover
|
|
yaml = None
|
|
|
|
SERVICES_DIR = Path(os.environ.get("SERVICES_DIR", "/services-src")).resolve()
|
|
# Superproject root (its .git is mounted read-only) — used to read each
|
|
# service dir's git history via a pathspec, since services have no own .git.
|
|
REPO_DIR = Path(os.environ.get("REPO_DIR", "/workspace")).resolve()
|
|
# Where the services dir lives *inside the repo* for git pathspecs.
|
|
REPO_SERVICES_REL = os.environ.get("REPO_SERVICES_REL", "services")
|
|
|
|
MAX_TEXT = 40_000 # cap README / CLAUDE.md payloads
|
|
MAX_FILE_BYTES = 512 * 1024 # cap a single file preview
|
|
|
|
# Compose filenames we recognise (globs), primary first.
|
|
COMPOSE_GLOBS = ("docker-compose.yml", "docker-compose.yaml",
|
|
"compose.yml", "compose.yaml",
|
|
"docker-compose.*.yml", "compose.*.yml")
|
|
|
|
# Dirs we never descend into for the file tree (runtime data / vendored / secret).
|
|
IGNORE_DIRS = {
|
|
"node_modules", ".git", "dist", ".turbo", "__pycache__", ".venv", "venv",
|
|
".next", "build", ".cache", "pb_data", "data", "acme", "logs", "uploads",
|
|
"images", "cache", "coverage", ".pytest_cache", "meili_data_v1.12",
|
|
"data-node", "gitea", "grafana-data", "loki-data", "prometheus-data",
|
|
}
|
|
IGNORE_FILES = {".DS_Store"}
|
|
|
|
# Files whose *contents* we refuse to serve even though they're listed (secrets).
|
|
SECRET_RE = re.compile(
|
|
r"(^\.env)|(\.env$)|(secret)|(\.key$)|(\.pem$)|(\.crt$)|(\.pfx$)"
|
|
r"|(^id_rsa)|(acme\.json)|(\.db$)|(\.sqlite\d?$)|(users_database)"
|
|
r"|(password)|(credentials)|(token)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
TEXT_EXT = {
|
|
".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".json", ".md", ".mdx",
|
|
".css", ".scss", ".html", ".htm", ".svg", ".txt", ".yml", ".yaml",
|
|
".toml", ".sh", ".bash", ".py", ".example", ".gitignore", ".dockerignore",
|
|
".cfg", ".ini", ".conf", ".xml", ".rs", ".go", ".sql", ".env", ".gitkeep",
|
|
}
|
|
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg", ".avif", ".ico"}
|
|
|
|
# Host / PathPrefix live inside backtick-quoted tokens; a host looks domain-ish.
|
|
_BACKTICK_RE = re.compile(r"`([^`]+)`")
|
|
_HOST_RE = re.compile(r"^[A-Za-z0-9.*-]+\.[A-Za-z]{2,}$")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# small readers
|
|
# --------------------------------------------------------------------------- #
|
|
def _read_text(p: Path, limit: int = MAX_TEXT) -> str | None:
|
|
try:
|
|
t = p.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return None
|
|
if len(t) > limit:
|
|
t = t[:limit] + f"\n\n… [{len(t) - limit} more chars truncated]"
|
|
return t
|
|
|
|
|
|
def _load_yaml(p: Path) -> dict | None:
|
|
if yaml is None or not p.is_file():
|
|
return None
|
|
try:
|
|
doc = yaml.safe_load(p.read_text(encoding="utf-8", errors="replace"))
|
|
except Exception:
|
|
return None
|
|
return doc if isinstance(doc, dict) else None
|
|
|
|
|
|
def _readme_path(root: Path) -> Path | None:
|
|
for cand in ("README.md", "README", "readme.md", "Readme.md"):
|
|
p = root / cand
|
|
if p.is_file():
|
|
return p
|
|
return None
|
|
|
|
|
|
def _clean(text: str) -> str:
|
|
if "@@" in text:
|
|
return ""
|
|
text = re.sub(r"\[([^\]]+)\]\([^)]*\)", r"\1", text)
|
|
text = re.sub(r"[*`]+", "", text)
|
|
return text.strip()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# git — one bucketed history walk for all services, cached on the repo HEAD
|
|
# (see githist.py); a per-request `git log` per service made this 11 s.
|
|
# --------------------------------------------------------------------------- #
|
|
_HISTORY = githist.BucketedHistory(REPO_DIR, REPO_SERVICES_REL)
|
|
|
|
|
|
def _last_commit(name: str) -> tuple[int | None, str | None]:
|
|
b = _HISTORY.get(name)
|
|
return (b["ts"], b["iso"]) if b else (None, None)
|
|
|
|
|
|
def _recent_commits(name: str) -> list[dict]:
|
|
b = _HISTORY.get(name)
|
|
return list(b["commits"]) if b else []
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# compose parsing
|
|
# --------------------------------------------------------------------------- #
|
|
def _compose_files(root: Path) -> list[Path]:
|
|
seen: dict[str, Path] = {}
|
|
for pat in COMPOSE_GLOBS:
|
|
for p in sorted(root.glob(pat)):
|
|
if p.is_file():
|
|
seen.setdefault(p.name, p)
|
|
return list(seen.values())
|
|
|
|
|
|
def _env_keys(env) -> list[str]:
|
|
"""Environment keys only — values are deliberately dropped."""
|
|
keys: list[str] = []
|
|
if isinstance(env, dict):
|
|
keys = [str(k) for k in env.keys()]
|
|
elif isinstance(env, list):
|
|
for item in env:
|
|
if isinstance(item, str):
|
|
keys.append(item.split("=", 1)[0].strip())
|
|
# de-dup, keep order
|
|
seen: set[str] = set()
|
|
return [k for k in keys if k and not (k in seen or seen.add(k))]
|
|
|
|
|
|
def _as_list(v) -> list[str]:
|
|
if v is None:
|
|
return []
|
|
if isinstance(v, str):
|
|
return [v]
|
|
if isinstance(v, list):
|
|
return [str(x) for x in v]
|
|
if isinstance(v, dict): # depends_on: {svc: {condition: ...}}
|
|
return [str(k) for k in v.keys()]
|
|
return []
|
|
|
|
|
|
def _parse_volume(v) -> dict | None:
|
|
if isinstance(v, str):
|
|
# source:target[:mode] — split from the right so ${PWD}/... survives.
|
|
parts = v.split(":")
|
|
if len(parts) == 1:
|
|
return {"source": None, "target": parts[0], "mode": None, "readOnly": False}
|
|
mode = None
|
|
if len(parts) >= 3:
|
|
mode = parts[-1]
|
|
target = parts[-2]
|
|
source = ":".join(parts[:-2])
|
|
else:
|
|
target = parts[-1]
|
|
source = ":".join(parts[:-1])
|
|
return {
|
|
"source": source or None, "target": target, "mode": mode,
|
|
"readOnly": bool(mode) and "ro" in mode,
|
|
}
|
|
if isinstance(v, dict):
|
|
return {
|
|
"source": v.get("source"),
|
|
"target": v.get("target"),
|
|
"mode": v.get("type"),
|
|
"readOnly": bool(v.get("read_only")),
|
|
}
|
|
return None
|
|
|
|
|
|
def _parse_ports(ports) -> list[str]:
|
|
out: list[str] = []
|
|
for p in ports or []:
|
|
if isinstance(p, (str, int)):
|
|
out.append(str(p))
|
|
elif isinstance(p, dict):
|
|
pub = p.get("published")
|
|
tgt = p.get("target")
|
|
host = p.get("host_ip")
|
|
s = ":".join(str(x) for x in (host, pub, tgt) if x is not None)
|
|
if p.get("protocol"):
|
|
s += f"/{p['protocol']}"
|
|
out.append(s or str(p))
|
|
return out
|
|
|
|
|
|
def _container(name: str, spec: dict, compose_file: str) -> dict:
|
|
if not isinstance(spec, dict):
|
|
spec = {}
|
|
build = spec.get("build")
|
|
if isinstance(build, dict):
|
|
build = build.get("context") or build.get("dockerfile") or "build"
|
|
nets = spec.get("networks")
|
|
if isinstance(nets, dict):
|
|
networks = list(nets.keys())
|
|
else:
|
|
networks = _as_list(nets)
|
|
volumes = [pv for v in (spec.get("volumes") or []) if (pv := _parse_volume(v))]
|
|
return {
|
|
"key": name,
|
|
"containerName": spec.get("container_name") or name,
|
|
"image": spec.get("image"),
|
|
"build": build if isinstance(build, str) else None,
|
|
"profiles": _as_list(spec.get("profiles")),
|
|
"restart": spec.get("restart"),
|
|
"networkMode": spec.get("network_mode"),
|
|
"networks": networks,
|
|
"ports": _parse_ports(spec.get("ports")),
|
|
"dependsOn": _as_list(spec.get("depends_on")),
|
|
"envKeys": _env_keys(spec.get("environment")),
|
|
"envFiles": _as_list(spec.get("env_file")),
|
|
"volumes": volumes,
|
|
"hasHealthcheck": bool(spec.get("healthcheck")),
|
|
"composeFile": compose_file,
|
|
}
|
|
|
|
|
|
def _parse_compose(root: Path) -> dict:
|
|
files = _compose_files(root)
|
|
containers: list[dict] = []
|
|
networks: set[str] = set()
|
|
for f in files:
|
|
doc = _load_yaml(f)
|
|
if not doc:
|
|
continue
|
|
for cname, spec in (doc.get("services") or {}).items():
|
|
containers.append(_container(cname, spec, f.name))
|
|
for n in (doc.get("networks") or {}):
|
|
networks.add(str(n))
|
|
return {
|
|
"composeFiles": [f.name for f in files],
|
|
"containers": containers,
|
|
"topNetworks": sorted(networks),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# traefik parsing
|
|
# --------------------------------------------------------------------------- #
|
|
def _hosts_in_rule(rule: str) -> list[str]:
|
|
return [tok for tok in _BACKTICK_RE.findall(rule or "") if _HOST_RE.match(tok)]
|
|
|
|
|
|
def _router_url(hosts: list[str], entrypoints: list[str], tls: bool) -> list[str]:
|
|
scheme = "https" if (tls or "websecure" in entrypoints or "web" in entrypoints) else "http"
|
|
urls = []
|
|
for h in hosts:
|
|
if "*" in h:
|
|
continue
|
|
urls.append(f"{scheme}://{h}")
|
|
return urls
|
|
|
|
|
|
def _parse_traefik(root: Path) -> dict:
|
|
doc = _load_yaml(root / "traefik.yml")
|
|
routers: list[dict] = []
|
|
services: list[dict] = []
|
|
middlewares_used: set[str] = set()
|
|
urls: list[str] = []
|
|
if not doc:
|
|
return {"routers": [], "traefikServices": [], "middlewares": [],
|
|
"urls": [], "hasTraefik": False}
|
|
|
|
for kind in ("http", "tcp"):
|
|
block = doc.get(kind) or {}
|
|
for rname, r in (block.get("routers") or {}).items():
|
|
if not isinstance(r, dict):
|
|
continue
|
|
rule = r.get("rule") or ""
|
|
hosts = _hosts_in_rule(rule)
|
|
eps = _as_list(r.get("entryPoints"))
|
|
mws = _as_list(r.get("middlewares"))
|
|
tls = bool(r.get("tls"))
|
|
ru = _router_url(hosts, eps, tls) if kind == "http" else \
|
|
[f"tcp://{h}" for h in hosts]
|
|
for u in ru:
|
|
if u not in urls:
|
|
urls.append(u)
|
|
middlewares_used.update(mws)
|
|
routers.append({
|
|
"name": rname,
|
|
"kind": kind,
|
|
"rule": rule,
|
|
"hosts": hosts,
|
|
"url": ru[0] if ru else None,
|
|
"entryPoints": eps,
|
|
"service": r.get("service"),
|
|
"middlewares": mws,
|
|
"tls": tls,
|
|
"priority": r.get("priority"),
|
|
})
|
|
for sname, s in (block.get("services") or {}).items():
|
|
if not isinstance(s, dict):
|
|
continue
|
|
lb = s.get("loadBalancer") or {}
|
|
servers = [srv.get("url") or srv.get("address")
|
|
for srv in (lb.get("servers") or []) if isinstance(srv, dict)]
|
|
services.append({"name": sname,
|
|
"servers": [x for x in servers if x]})
|
|
|
|
return {
|
|
"routers": routers,
|
|
"traefikServices": services,
|
|
"middlewares": sorted(middlewares_used),
|
|
"urls": urls,
|
|
"hasTraefik": True,
|
|
}
|
|
|
|
|
|
def _auth_kind(middlewares: list[str]) -> str | None:
|
|
joined = " ".join(middlewares).lower()
|
|
has_auth = "auth-chain" in joined or "authelia" in joined
|
|
has_public = "public-chain" in joined
|
|
has_api = "api-chain" in joined
|
|
if has_auth and (has_public or has_api):
|
|
return "mixed"
|
|
if has_auth:
|
|
return "auth"
|
|
if has_public:
|
|
return "public"
|
|
if has_api:
|
|
return "api"
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# homepage dashboard enrichment (services.yaml → per-container description/icon)
|
|
# --------------------------------------------------------------------------- #
|
|
_HOMEPAGE_CACHE: dict | None = None
|
|
|
|
|
|
def _homepage_index() -> dict:
|
|
"""Map container name and host → {description, icon, href, group}."""
|
|
global _HOMEPAGE_CACHE
|
|
if _HOMEPAGE_CACHE is not None:
|
|
return _HOMEPAGE_CACHE
|
|
idx: dict[str, dict] = {}
|
|
doc_path = SERVICES_DIR / "homepage" / "config" / "services.yaml"
|
|
if yaml is not None and doc_path.is_file():
|
|
try:
|
|
groups = yaml.safe_load(doc_path.read_text(encoding="utf-8", errors="replace"))
|
|
except Exception:
|
|
groups = None
|
|
for group in groups or []:
|
|
if not isinstance(group, dict):
|
|
continue
|
|
for gname, entries in group.items():
|
|
for entry in entries or []:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
for label, meta in entry.items():
|
|
if not isinstance(meta, dict):
|
|
continue
|
|
rec = {
|
|
"label": label,
|
|
"description": meta.get("description"),
|
|
"icon": meta.get("icon"),
|
|
"href": meta.get("href"),
|
|
"group": gname,
|
|
}
|
|
cont = meta.get("container")
|
|
if isinstance(cont, str):
|
|
idx.setdefault(cont, rec)
|
|
href = meta.get("href") or ""
|
|
m = re.search(r"https?://([^/]+)", href)
|
|
if m:
|
|
idx.setdefault("host:" + m.group(1), rec)
|
|
_HOMEPAGE_CACHE = idx
|
|
return idx
|
|
|
|
|
|
def _homepage_for(name: str, containers: list[dict], urls: list[str]) -> dict | None:
|
|
idx = _homepage_index()
|
|
for c in containers:
|
|
rec = idx.get(c["containerName"]) or idx.get(c["key"])
|
|
if rec:
|
|
return rec
|
|
for u in urls:
|
|
m = re.search(r"https?://([^/]+)", u)
|
|
if m and ("host:" + m.group(1)) in idx:
|
|
return idx["host:" + m.group(1)]
|
|
return idx.get(name)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# file tree (read-only)
|
|
# --------------------------------------------------------------------------- #
|
|
def _is_text(path: Path) -> bool:
|
|
if path.name in {"Dockerfile", "Makefile", "LICENSE", "README", "Corefile"}:
|
|
return True
|
|
return path.suffix.lower() in TEXT_EXT
|
|
|
|
|
|
def _tree(root: Path, base: Path, depth: int = 0) -> list[dict]:
|
|
if depth > 5:
|
|
return []
|
|
out: list[dict] = []
|
|
try:
|
|
entries = sorted(root.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))
|
|
except OSError:
|
|
return out
|
|
for entry in entries:
|
|
if entry.name in IGNORE_FILES or entry.name.startswith(".git"):
|
|
continue
|
|
if entry.is_dir():
|
|
if entry.name in IGNORE_DIRS:
|
|
continue
|
|
out.append({
|
|
"name": entry.name,
|
|
"path": str(entry.relative_to(base)),
|
|
"type": "dir",
|
|
"children": _tree(entry, base, depth + 1),
|
|
})
|
|
else:
|
|
try:
|
|
size = entry.stat().st_size
|
|
except OSError:
|
|
size = 0
|
|
secret = bool(SECRET_RE.search(entry.name))
|
|
out.append({
|
|
"name": entry.name,
|
|
"path": str(entry.relative_to(base)),
|
|
"type": "file",
|
|
"size": size,
|
|
"text": _is_text(entry) and not secret,
|
|
"secret": secret,
|
|
})
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# record assembly
|
|
# --------------------------------------------------------------------------- #
|
|
KEY_FILES = (
|
|
"README.md", "CLAUDE.md", "GOAL.md", "docker-compose.yml", "traefik.yml",
|
|
"Dockerfile", ".env.example", "Corefile", "pyproject.toml",
|
|
)
|
|
|
|
|
|
def _description(root: Path, homepage: dict | None) -> str:
|
|
rp = _readme_path(root)
|
|
readme = _read_text(rp) if rp else None
|
|
if readme:
|
|
for line in readme.splitlines():
|
|
cleaned = _clean(line.strip())
|
|
if not cleaned or cleaned.startswith(("#", "!", "<", "---", "```", "http")):
|
|
continue
|
|
return cleaned
|
|
if homepage and homepage.get("description"):
|
|
return str(homepage["description"])
|
|
return ""
|
|
|
|
|
|
def _base_record(entry: Path) -> dict:
|
|
name = entry.name
|
|
compose = _parse_compose(entry)
|
|
traefik = _parse_traefik(entry)
|
|
homepage = _homepage_for(name, compose["containers"], traefik["urls"])
|
|
ts, iso = _last_commit(name)
|
|
|
|
profiles: set[str] = set()
|
|
images: list[str] = []
|
|
for c in compose["containers"]:
|
|
profiles.update(c["profiles"])
|
|
if c["image"]:
|
|
images.append(c["image"])
|
|
|
|
return {
|
|
"dir": name,
|
|
"name": name,
|
|
"description": _description(entry, homepage),
|
|
"icon": (homepage or {}).get("icon"),
|
|
"group": (homepage or {}).get("group"),
|
|
"urls": traefik["urls"],
|
|
"containers": [c["containerName"] for c in compose["containers"]],
|
|
"containerCount": len(compose["containers"]),
|
|
"images": sorted(set(images)),
|
|
"profiles": sorted(profiles),
|
|
"auth": _auth_kind(traefik["middlewares"]),
|
|
"hasCompose": bool(compose["composeFiles"]),
|
|
"hasTraefik": traefik["hasTraefik"],
|
|
"goal": goalmd.goal_summary(entry),
|
|
"updatedAt": iso,
|
|
"_ts": ts,
|
|
"_compose": compose,
|
|
"_traefik": traefik,
|
|
"_homepage": homepage,
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# public API
|
|
# --------------------------------------------------------------------------- #
|
|
def list_services() -> list[dict]:
|
|
if not SERVICES_DIR.is_dir():
|
|
return []
|
|
items: list[dict] = []
|
|
for entry in sorted(SERVICES_DIR.iterdir(), key=lambda p: p.name.lower()):
|
|
if not entry.is_dir() or entry.name.startswith("."):
|
|
continue
|
|
rec = _base_record(entry)
|
|
for k in ("_compose", "_traefik", "_homepage"):
|
|
rec.pop(k, None)
|
|
items.append(rec)
|
|
# newest commit first
|
|
items.sort(key=lambda i: i["_ts"] if i["_ts"] is not None else -1, reverse=True)
|
|
for i in items:
|
|
i.pop("_ts", None)
|
|
return items
|
|
|
|
|
|
def _safe_entry(slug: str) -> Path | None:
|
|
name = Path(slug).name
|
|
if not name or name.startswith("."):
|
|
return None
|
|
entry = (SERVICES_DIR / name).resolve()
|
|
try:
|
|
entry.relative_to(SERVICES_DIR)
|
|
except ValueError:
|
|
return None
|
|
return entry if entry.is_dir() else None
|
|
|
|
|
|
def service_detail(slug: str) -> dict | None:
|
|
entry = _safe_entry(slug)
|
|
if not entry:
|
|
return None
|
|
rec = _base_record(entry)
|
|
compose = rec.pop("_compose")
|
|
traefik = rec.pop("_traefik")
|
|
homepage = rec.pop("_homepage")
|
|
rec.pop("_ts", None)
|
|
|
|
present = []
|
|
for f in KEY_FILES:
|
|
p = entry / f
|
|
if p.is_file():
|
|
try:
|
|
present.append({"name": f, "bytes": p.stat().st_size})
|
|
except OSError:
|
|
present.append({"name": f, "bytes": 0})
|
|
|
|
# union of env keys + volumes across containers
|
|
env_keys: list[str] = []
|
|
for c in compose["containers"]:
|
|
for k in c["envKeys"]:
|
|
if k not in env_keys:
|
|
env_keys.append(k)
|
|
|
|
rp = _readme_path(entry)
|
|
claude = entry / "CLAUDE.md"
|
|
|
|
rec.update({
|
|
"containersDetail": compose["containers"],
|
|
"composeFiles": compose["composeFiles"],
|
|
"topNetworks": compose["topNetworks"],
|
|
"routers": traefik["routers"],
|
|
"traefikServices": traefik["traefikServices"],
|
|
"middlewares": traefik["middlewares"],
|
|
"envKeys": env_keys,
|
|
"homepage": homepage,
|
|
"readme": _read_text(rp) if rp else None,
|
|
"claudeMd": _read_text(claude) if claude.is_file() else None,
|
|
"goal": goalmd.goal_detail(entry),
|
|
"keyFiles": present,
|
|
"commits": _recent_commits(entry.name),
|
|
"tree": _tree(entry, entry),
|
|
})
|
|
return rec
|
|
|
|
|
|
def service_file(slug: str, rel: str) -> str | None:
|
|
"""Return a whitelisted text file's contents from within a service dir."""
|
|
entry = _safe_entry(slug)
|
|
if not entry:
|
|
return None
|
|
rel_p = Path(rel)
|
|
if rel_p.is_absolute() or ".." in rel_p.parts:
|
|
return None
|
|
p = (entry / rel_p).resolve()
|
|
try:
|
|
p.relative_to(entry.resolve())
|
|
except ValueError:
|
|
return None
|
|
if not p.is_file() or not _is_text(p) or SECRET_RE.search(p.name):
|
|
return None
|
|
# block any path that dips through an ignored/secret dir
|
|
if any(part in IGNORE_DIRS for part in rel_p.parts):
|
|
return None
|
|
try:
|
|
if p.stat().st_size > MAX_FILE_BYTES:
|
|
return None
|
|
return p.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return None
|