4f5b7609c6
The repo-root project entry — the one every conversation with no `projects/<slug>` attribution lands in — is now called `homelab` instead of `lab`, which is what it actually is and what the composer's auto-tag will match when you type it. Its description doubles as the tag's scope: the spawn composer writes it verbatim into the `Work in the following homelab location(s):` block, so it now spells out the edit surface (infra: services/Traefik/compose; Claude config: .claude skills and agents; root scripts and docs) and says to stay outside `projects/*` — anything that changes a project belongs under that project's own tag. Nothing to migrate: the attribution is purely derived (no conversation sidecar records the slug) and no tag overrides referenced the old id.
675 lines
24 KiB
Python
675 lines
24 KiB
Python
"""Project gallery + per-project detail, mined from the mounted ``~/projects`` dir.
|
||
|
||
Ported and extended from ``services/projects-gallery``. Scans each directory
|
||
under ``PROJECTS_DIR`` and extracts metadata from ``package.json`` (with README
|
||
/ deploy-script / ``index.html`` fallbacks), the project's own git history, and
|
||
— for the detail view — the contents of key files (README.md, CLAUDE.md,
|
||
``index.html`` head meta, an og-image). The ai-agent frontend renders this as a
|
||
browsable gallery whose project pages cross-link to the conversations that
|
||
worked on them.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import githist
|
||
import goal as goalmd
|
||
|
||
PROJECTS_DIR = Path(os.environ.get("PROJECTS_DIR", "/workspace/projects")).resolve()
|
||
# Superproject root (its .git is mounted read-only) so we can read each
|
||
# project's last commit date when the project itself has no own .git.
|
||
REPO_DIR = Path(os.environ.get("REPO_DIR", "/workspace")).resolve()
|
||
|
||
# Directories that are not real projects / are pure noise.
|
||
IGNORE_DIRS = {
|
||
"node_modules", ".git", "dist", ".turbo", "__pycache__",
|
||
".venv", "venv", ".next", "build", ".cache", "pb_data", "templates",
|
||
}
|
||
|
||
# Domains we recognise as Gabriel's own deploy targets when sniffing READMEs.
|
||
OWN_DOMAINS = (
|
||
"gabvdl.xyz", "vidal--ayrinhac.xyz", "forgecode.dev", "hemicycle.dev",
|
||
)
|
||
|
||
URL_RE = re.compile(r'https?://[a-zA-Z0-9._~%-]+\.[a-zA-Z]{2,}(?:/[^\s"\'`)]*)?')
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# the root project — the homelab repo itself
|
||
# --------------------------------------------------------------------------- #
|
||
# Plenty of conversations never touch a `projects/<slug>` directory: they wire a
|
||
# service, edit a skill, fix a root script, write a doc, or just answer a
|
||
# question. Those used to fall out of every project-shaped view (gallery, tags,
|
||
# cost rollups, graph, goals) as "(unattributed)".
|
||
#
|
||
# `homelab` is where they land instead — and it is not a placeholder bucket: it
|
||
# is a real project entry backed by the **homelab repo root** (`REPO_DIR`,
|
||
# `~/homelab` on the host). Its git history is the repo's history, its CLAUDE.md
|
||
# is the repo's instructions, its GOAL.md — when the repo has one — puts it on
|
||
# the goals board like any other unit. The one difference from a `projects/*`
|
||
# entry is where it lives on disk, which `project_dir()` / `repo_rel()` below
|
||
# resolve for callers.
|
||
#
|
||
# Its description is also a **scope**, not just a label: the spawn composer
|
||
# writes it verbatim into the `Work in the following homelab location(s):` block,
|
||
# so it is the sentence an agent actually reads before it starts editing. It says
|
||
# to stay outside `projects/*` — a change to a project belongs under that
|
||
# project's own tag, where its gallery card, costs, goals and conversations are.
|
||
#
|
||
# If a real `~/projects/homelab` directory ever appears it wins: the directory
|
||
# scan emits it and this record is skipped, so the tag keeps exactly one meaning.
|
||
ROOT_SLUG = os.environ.get("ROOT_PROJECT_SLUG", "homelab")
|
||
ROOT_DESCRIPTION = (
|
||
"The homelab repo itself — infrastructure (services, Traefik, compose), "
|
||
"Claude config (.claude skills and agents), root scripts and docs. "
|
||
"Edit only outside `projects/*`: anything that changes a project belongs to "
|
||
"that project's own tag, not this one."
|
||
)
|
||
ROOT_STACK = ["Homelab", "Docker", "Traefik"]
|
||
|
||
|
||
def attributed_projects(projects: list[str]) -> list[str]:
|
||
"""A conversation's project attribution, with the root fallback applied.
|
||
|
||
A conversation that edited no `projects/<slug>` file is attributed to
|
||
:data:`ROOT_SLUG` — including one that only touched `services/<slug>`, since
|
||
the services *are* the lab (it keeps its service tags too). Purely derived:
|
||
nothing is written to the conversation's metadata sidecar, so the fallback is
|
||
retroactive and drops away by itself the moment a real project tag appears.
|
||
"""
|
||
return list(projects) if projects else [ROOT_SLUG]
|
||
|
||
|
||
def project_dir(slug: str) -> Path | None:
|
||
"""On-disk root of a project slug — the repo root for :data:`ROOT_SLUG`."""
|
||
entry = _safe_entry(slug)
|
||
if entry is not None:
|
||
return entry
|
||
return REPO_DIR if slug == ROOT_SLUG else None
|
||
|
||
|
||
def repo_rel(slug: str) -> str:
|
||
"""How to name a project slug as a path, for prompts and labels."""
|
||
return "." if project_dir(slug) == REPO_DIR else f"projects/{slug}"
|
||
|
||
|
||
def _root_record() -> dict:
|
||
"""The gallery card for the homelab repo itself."""
|
||
ts, iso = _last_commit(REPO_DIR)
|
||
return {
|
||
"dir": ROOT_SLUG,
|
||
"name": ROOT_SLUG,
|
||
"description": ROOT_DESCRIPTION,
|
||
"version": None,
|
||
"url": None,
|
||
"stack": list(ROOT_STACK),
|
||
"keywords": [],
|
||
"hasPackageJson": False,
|
||
"goal": goalmd.goal_summary(REPO_DIR),
|
||
"updatedAt": iso,
|
||
"isRoot": True,
|
||
"_ts": ts,
|
||
}
|
||
|
||
|
||
# og-image candidates, in priority order, relative to the project root.
|
||
OG_IMAGE_CANDIDATES = (
|
||
"public/og-image.png", "public/og-image.jpg", "public/og-image.jpeg",
|
||
"public/og-image.webp", "public/og-thumbnail.webp", "public/og.png",
|
||
"public/og.jpg", "public/thumbnail.png", "public/thumbnail.webp",
|
||
"og-image.png", "og-image.webp", "og.png", "thumbnail.png",
|
||
)
|
||
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg", ".avif"}
|
||
MAX_TEXT = 40_000 # cap key-file payloads so detail responses stay bounded
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# small readers / cleaners
|
||
# --------------------------------------------------------------------------- #
|
||
def _norm_url(url: str | None) -> str | None:
|
||
"""Drop query/fragment and trailing slash from a resolved deploy URL."""
|
||
if not url:
|
||
return None
|
||
url = url.split("?", 1)[0].split("#", 1)[0]
|
||
return url.rstrip("/")
|
||
|
||
|
||
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 _read_package_json(root: Path) -> dict | None:
|
||
p = root / "package.json"
|
||
if not p.is_file():
|
||
return None
|
||
try:
|
||
return json.loads(p.read_text(encoding="utf-8", errors="replace"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return 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:
|
||
"""Drop unfilled scaffold placeholders and light markdown for display."""
|
||
if "@@" in text:
|
||
return ""
|
||
text = re.sub(r'\[([^\]]+)\]\([^)]*\)', r'\1', text) # [label](url) -> label
|
||
text = re.sub(r'[*`]+', '', text) # **bold** / `code`
|
||
return text.strip()
|
||
|
||
|
||
def _description(pkg: dict | None, readme: str | None) -> str:
|
||
if pkg and isinstance(pkg.get("description"), str):
|
||
desc = _clean(pkg["description"])
|
||
if desc:
|
||
return desc
|
||
if readme:
|
||
for line in readme.splitlines():
|
||
cleaned = _clean(line.strip())
|
||
if not cleaned or cleaned.startswith(("#", "!", "<", "---", "```", "http")):
|
||
continue
|
||
if re.match(r'(?i)(project\s+)?url\s*[:=]', cleaned):
|
||
continue
|
||
return cleaned
|
||
return ""
|
||
|
||
|
||
# Subdirs a fullstack project might keep its frontend under.
|
||
_FRONTEND_DIRS = ("", "frontend", "web", "ui", "client", "app")
|
||
|
||
|
||
def _stack(root: Path, pkg: dict | None) -> list[str]:
|
||
"""Best-effort detection of the tech in a project, as short tags."""
|
||
tags: list[str] = []
|
||
deps: dict = {}
|
||
if pkg:
|
||
deps.update(pkg.get("dependencies", {}))
|
||
deps.update(pkg.get("devDependencies", {}))
|
||
for sub in _FRONTEND_DIRS:
|
||
if sub:
|
||
nested = _read_package_json(root / sub)
|
||
if nested:
|
||
deps.update(nested.get("dependencies", {}))
|
||
deps.update(nested.get("devDependencies", {}))
|
||
has = lambda *names: any(n in deps for n in names) # noqa: E731
|
||
|
||
if has("react", "react-dom"):
|
||
tags.append("React")
|
||
if has("vue"):
|
||
tags.append("Vue")
|
||
if has("svelte"):
|
||
tags.append("Svelte")
|
||
if has("astro"):
|
||
tags.append("Astro")
|
||
if has("three", "@react-three/fiber"):
|
||
tags.append("Three.js")
|
||
if has("vite", "@vitejs/plugin-react"):
|
||
tags.append("Vite")
|
||
if has("typescript") or (root / "tsconfig.json").exists():
|
||
tags.append("TypeScript")
|
||
if has("tailwindcss", "@tailwindcss/vite"):
|
||
tags.append("Tailwind")
|
||
if has("pocketbase"):
|
||
tags.append("PocketBase")
|
||
if has("rxdb"):
|
||
tags.append("RxDB")
|
||
if has("gsap"):
|
||
tags.append("GSAP")
|
||
if has("leaflet"):
|
||
tags.append("Leaflet")
|
||
if (root / "Cargo.toml").exists():
|
||
tags.append("Rust")
|
||
if (root / "go.mod").exists():
|
||
tags.append("Go")
|
||
if (root / "pyproject.toml").exists() or (root / "requirements.txt").exists():
|
||
tags.append("Python")
|
||
if (root / "Dockerfile").exists():
|
||
tags.append("Docker")
|
||
seen: set[str] = set()
|
||
return [t for t in tags if not (t in seen or seen.add(t))]
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# git — standalone repos hit a per-repo TTL cache; superproject-tracked dirs
|
||
# share one bucketed history walk cached on the repo HEAD (see githist.py).
|
||
# Per-request `git log` pairs × 50 projects made /api/projects a 3.4 s endpoint.
|
||
# --------------------------------------------------------------------------- #
|
||
_HISTORY = githist.BucketedHistory(REPO_DIR, "projects")
|
||
_OWN_REPO_LOG = githist.RepoLogCache()
|
||
|
||
|
||
def _history_for(entry: Path) -> dict | None:
|
||
"""{ts, iso, commits} for a project: its own repo, else the superproject."""
|
||
if (entry / ".git").exists():
|
||
return _OWN_REPO_LOG.get(entry)
|
||
return _HISTORY.get(entry.name)
|
||
|
||
|
||
def _last_commit(entry: Path) -> tuple[int | None, str | None]:
|
||
b = _history_for(entry)
|
||
return (b["ts"], b["iso"]) if b else (None, None)
|
||
|
||
|
||
def _recent_commits(entry: Path) -> list[dict]:
|
||
b = _history_for(entry)
|
||
return list(b["commits"]) if b else []
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# deploy URL + index.html head meta + og image
|
||
# --------------------------------------------------------------------------- #
|
||
def _find_url(root: Path, pkg: dict | None, name: str) -> str | None:
|
||
"""Find the project's public deploy URL, best-effort."""
|
||
hp = pkg.get("homepage") if pkg else None
|
||
if isinstance(hp, str) and hp.startswith("http"):
|
||
return _norm_url(hp.strip())
|
||
|
||
deploy_files: list[Path] = []
|
||
scripts_dir = root / "scripts"
|
||
if scripts_dir.is_dir():
|
||
deploy_files += sorted(scripts_dir.glob("deploy*.sh"))
|
||
deploy_files += sorted(root.glob("deploy*.sh"))
|
||
deploy_files += [root / ".zipgo.json"]
|
||
|
||
url_candidates: list[str] = []
|
||
host_candidates: list[str] = []
|
||
for f in deploy_files:
|
||
if not f.is_file():
|
||
continue
|
||
try:
|
||
txt = f.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
continue
|
||
for line in txt.splitlines():
|
||
if re.search(r'\bURL\s*[=:]', line):
|
||
m = URL_RE.search(line)
|
||
if m:
|
||
url_candidates.append(m.group(0))
|
||
hm = re.search(
|
||
r'\b(?:HOST|domain|d)\s*[=:]\s*["\']?'
|
||
r'([a-zA-Z0-9.-]+\.(?:xyz|dev|com|app|net))',
|
||
line,
|
||
)
|
||
if hm:
|
||
host_candidates.append("https://" + hm.group(1))
|
||
|
||
def best(cands: list[str]) -> str | None:
|
||
cleaned = [c.rstrip("/") for c in cands]
|
||
nondemo = [c for c in cleaned if "demo" not in c]
|
||
pool = nondemo or cleaned
|
||
return min(pool, key=len) if pool else None
|
||
|
||
chosen = best(url_candidates) or best(host_candidates)
|
||
if chosen:
|
||
return _norm_url(chosen)
|
||
|
||
for idx in (root / "index.html", root / "public" / "index.html"):
|
||
if idx.is_file():
|
||
try:
|
||
t = idx.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
continue
|
||
m = re.search(r'og:url["\']\s+content=["\']([^"\']+)', t)
|
||
if m:
|
||
return _norm_url(m.group(1))
|
||
|
||
tokens = {t for t in re.split(r'[-_]', f"{root.name} {name}".lower()) if len(t) > 2}
|
||
rp = _readme_path(root)
|
||
readme = _read_text(rp) if rp else None
|
||
if readme:
|
||
for mm in URL_RE.finditer(readme):
|
||
url = mm.group(0)
|
||
if not any(d in url for d in OWN_DOMAINS):
|
||
continue
|
||
host = (_norm_url(url) or "").split("//", 1)[-1].split("/", 1)[0]
|
||
is_apex = host.count(".") == 1
|
||
if is_apex or any(tok in host for tok in tokens):
|
||
return _norm_url(url)
|
||
return None
|
||
|
||
|
||
def _index_html(root: Path) -> Path | None:
|
||
for cand in (root / "index.html", root / "public" / "index.html"):
|
||
if cand.is_file():
|
||
return cand
|
||
return None
|
||
|
||
|
||
_META_RE = (
|
||
lambda key: re.compile(
|
||
r'<meta[^>]+(?:name|property)=["\']' + key + r'["\'][^>]*content=["\']([^"\']*)["\']',
|
||
re.IGNORECASE,
|
||
)
|
||
)
|
||
|
||
|
||
def _index_meta(root: Path) -> dict | None:
|
||
"""Pull title / description / og:image / og:url / theme-color from index.html."""
|
||
idx = _index_html(root)
|
||
if not idx:
|
||
return None
|
||
try:
|
||
html = idx.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
return None
|
||
|
||
def grab(pat: re.Pattern) -> str | None:
|
||
m = pat.search(html)
|
||
return m.group(1).strip() if m else None
|
||
|
||
title_m = re.search(r'<title[^>]*>(.*?)</title>', html, re.IGNORECASE | re.DOTALL)
|
||
meta = {
|
||
"title": (title_m.group(1).strip() if title_m else None),
|
||
"description": grab(_META_RE("description")) or grab(_META_RE("og:description")),
|
||
"ogImage": grab(_META_RE("og:image")) or grab(_META_RE("twitter:image")),
|
||
"ogUrl": grab(_META_RE("og:url")),
|
||
"themeColor": grab(_META_RE("theme-color")),
|
||
}
|
||
return meta if any(meta.values()) else None
|
||
|
||
|
||
def _resolve_local_image(root: Path, ref: str) -> str | None:
|
||
"""If an og:image content is a LOCAL path, return its project-relative path."""
|
||
ref = ref.split("?", 1)[0].split("#", 1)[0]
|
||
if not ref:
|
||
return None
|
||
cand = ref.lstrip("/")
|
||
for base in (root, root / "public"):
|
||
p = (base / cand)
|
||
try:
|
||
p = p.resolve()
|
||
p.relative_to(root.resolve())
|
||
except (OSError, ValueError):
|
||
continue
|
||
if p.is_file() and p.suffix.lower() in IMAGE_EXTS:
|
||
return str(p.relative_to(root))
|
||
return None
|
||
|
||
|
||
def _og_image(root: Path, meta: dict | None) -> dict | None:
|
||
"""Resolve an og-image to either an absolute URL or a local asset path."""
|
||
if meta and meta.get("ogImage"):
|
||
ref = meta["ogImage"]
|
||
if ref.startswith("http"):
|
||
return {"kind": "url", "src": ref}
|
||
local = _resolve_local_image(root, ref)
|
||
if local:
|
||
return {"kind": "asset", "src": local}
|
||
for cand in OG_IMAGE_CANDIDATES:
|
||
p = root / cand
|
||
if p.is_file():
|
||
return {"kind": "asset", "src": cand}
|
||
return None
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# project name + base record
|
||
# --------------------------------------------------------------------------- #
|
||
def _project_name(entry: Path, pkg: dict | None) -> str:
|
||
name = entry.name
|
||
if pkg and isinstance(pkg.get("name"), str):
|
||
n = pkg["name"].split("/")[-1]
|
||
if n and "_" not in n and not n.startswith("@"):
|
||
name = n
|
||
return name
|
||
|
||
|
||
def _base_record(entry: Path) -> dict:
|
||
pkg = _read_package_json(entry)
|
||
rp = _readme_path(entry)
|
||
readme = _read_text(rp) if rp else None
|
||
name = _project_name(entry, pkg)
|
||
version = pkg.get("version") if pkg else None
|
||
keywords = pkg.get("keywords", []) if pkg else []
|
||
if not isinstance(keywords, list):
|
||
keywords = []
|
||
ts, iso = _last_commit(entry)
|
||
return {
|
||
"dir": entry.name,
|
||
"name": name,
|
||
"description": _description(pkg, readme),
|
||
"version": version if isinstance(version, str) else None,
|
||
"url": _find_url(entry, pkg, name),
|
||
"stack": _stack(entry, pkg),
|
||
"keywords": [k for k in keywords if isinstance(k, str)][:6],
|
||
"hasPackageJson": pkg is not None,
|
||
"goal": goalmd.goal_summary(entry),
|
||
"updatedAt": iso,
|
||
"_ts": ts,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# public API
|
||
# --------------------------------------------------------------------------- #
|
||
def list_projects() -> list[dict]:
|
||
items: list[dict] = []
|
||
if PROJECTS_DIR.is_dir():
|
||
for entry in sorted(PROJECTS_DIR.iterdir(), key=lambda p: p.name.lower()):
|
||
if not entry.is_dir() or entry.name in IGNORE_DIRS or entry.name.startswith("."):
|
||
continue
|
||
items.append(_base_record(entry))
|
||
# The repo-root card, unless a real `projects/<ROOT_SLUG>` dir already claims
|
||
# the slug (then that directory *is* the root project).
|
||
if not any(i["dir"] == ROOT_SLUG for i in items):
|
||
items.append(_root_record())
|
||
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:
|
||
"""Resolve a slug to a project dir, refusing traversal."""
|
||
name = Path(slug).name
|
||
if not name or name in IGNORE_DIRS or name.startswith("."):
|
||
return None
|
||
entry = (PROJECTS_DIR / name).resolve()
|
||
try:
|
||
entry.relative_to(PROJECTS_DIR)
|
||
except ValueError:
|
||
return None
|
||
return entry if entry.is_dir() else None
|
||
|
||
|
||
# Notable files we surface as "present" in the detail view.
|
||
KEY_FILES = (
|
||
"README.md", "CLAUDE.md", "GOAL.md", "package.json", "index.html",
|
||
"Dockerfile", "docker-compose.yml", "vite.config.ts", "tsconfig.json",
|
||
".zipgo.json", "LICENSE",
|
||
)
|
||
|
||
|
||
def _root_detail() -> dict:
|
||
"""Detail view for the root project (the repo itself).
|
||
|
||
Only the repo-level context the container actually mounts is readable here
|
||
(REPO_DIR holds the root `CLAUDE.md`, `.claude/`, `data/` and the read-only
|
||
`.git`), so the page shows the homelab's own instructions, goal and commit
|
||
history rather than a project's README/build scripts.
|
||
"""
|
||
rec = _root_record()
|
||
rec.pop("_ts", None)
|
||
present = []
|
||
for f in KEY_FILES:
|
||
p = REPO_DIR / f
|
||
if p.is_file():
|
||
try:
|
||
present.append({"name": f, "bytes": p.stat().st_size})
|
||
except OSError:
|
||
present.append({"name": f, "bytes": 0})
|
||
rp = _readme_path(REPO_DIR)
|
||
claude_md = REPO_DIR / "CLAUDE.md"
|
||
rec.update({
|
||
"readme": _read_text(rp) if rp else None,
|
||
"claudeMd": _read_text(claude_md) if claude_md.is_file() else None,
|
||
"goal": goalmd.goal_detail(REPO_DIR),
|
||
"indexMeta": None,
|
||
"ogImage": None,
|
||
"scripts": [],
|
||
"keyFiles": present,
|
||
"commits": _recent_commits(REPO_DIR),
|
||
})
|
||
return rec
|
||
|
||
|
||
def project_detail(slug: str) -> dict | None:
|
||
entry = _safe_entry(slug)
|
||
if entry is None:
|
||
return _root_detail() if slug == ROOT_SLUG else None
|
||
rec = _base_record(entry)
|
||
rec.pop("_ts", None)
|
||
|
||
pkg = _read_package_json(entry)
|
||
rp = _readme_path(entry)
|
||
meta = _index_meta(entry)
|
||
|
||
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})
|
||
|
||
claude_md = _read_text(entry / "CLAUDE.md") if (entry / "CLAUDE.md").is_file() else None
|
||
scripts = list(pkg.get("scripts", {}).keys()) if pkg else []
|
||
|
||
rec.update({
|
||
"readme": _read_text(rp) if rp else None,
|
||
"claudeMd": claude_md,
|
||
"goal": goalmd.goal_detail(entry),
|
||
"indexMeta": meta,
|
||
"ogImage": _og_image(entry, meta),
|
||
"scripts": [s for s in scripts if isinstance(s, str)][:24],
|
||
"keyFiles": present,
|
||
"commits": _recent_commits(entry),
|
||
})
|
||
return rec
|
||
|
||
|
||
# Source extensions counted toward a project's lines of code. Data/lock/vendor
|
||
# files are excluded so the LoC (and the cost-per-LoC it feeds) reflects code.
|
||
_LOC_EXTS = {
|
||
".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".py", ".rs", ".go",
|
||
".css", ".scss", ".sass", ".less", ".html", ".vue", ".svelte",
|
||
".md", ".sh", ".bash", ".yml", ".yaml", ".toml", ".sql", ".json",
|
||
".c", ".cpp", ".h", ".hpp", ".java", ".rb", ".php", ".astro",
|
||
}
|
||
_LOC_SKIP_NAMES = {
|
||
"package-lock.json", "pnpm-lock.yaml", "yarn.lock", "composer.lock",
|
||
"bun.lockb", "cargo.lock", "poetry.lock",
|
||
}
|
||
|
||
|
||
_LOC_CACHE: dict[str, tuple[float, int]] = {}
|
||
_LOC_TTL_SECS = 300.0
|
||
|
||
|
||
def _walk_loc(root: Path) -> int:
|
||
total = 0
|
||
for dirpath, dirnames, filenames in os.walk(root):
|
||
dirnames[:] = [d for d in dirnames
|
||
if d not in IGNORE_DIRS and not d.startswith(".")]
|
||
for fn in filenames:
|
||
low = fn.lower()
|
||
if low in _LOC_SKIP_NAMES or ".min." in low:
|
||
continue
|
||
if Path(fn).suffix.lower() not in _LOC_EXTS:
|
||
continue
|
||
try:
|
||
with (Path(dirpath) / fn).open("rb") as fh:
|
||
total += sum(1 for _ in fh)
|
||
except OSError:
|
||
continue
|
||
return total
|
||
|
||
|
||
def _root_loc() -> int:
|
||
"""Lines of source in the lab itself — the repo's **git-tracked** files.
|
||
|
||
A directory walk is the wrong tool here: unlike a project dir, what the
|
||
container mounts as the repo is a mix of source and runtime state
|
||
(`services/media/`, `services/home-assistant/`'s config tree, the untracked
|
||
`data/` logs), and counting that state would put the lab an order of
|
||
magnitude above every real project. `git ls-files` is the honest boundary:
|
||
what's actually committed. Paths are resolved back to whatever the container
|
||
mounts — `services/*` under SERVICES_DIR, everything else under REPO_DIR —
|
||
and anything unmounted is simply not counted.
|
||
"""
|
||
out = githist._git(["ls-files", "-z"], REPO_DIR, timeout=60)
|
||
if out is None:
|
||
return 0
|
||
services_dir = Path(os.environ.get("SERVICES_DIR", "/services-src"))
|
||
services_rel = os.environ.get("REPO_SERVICES_REL", "services")
|
||
total = 0
|
||
for rel in out.split("\0"):
|
||
if not rel:
|
||
continue
|
||
p = Path(rel)
|
||
if p.suffix.lower() not in _LOC_EXTS or p.name.lower() in _LOC_SKIP_NAMES:
|
||
continue
|
||
if ".min." in p.name.lower() or any(part in IGNORE_DIRS for part in p.parts):
|
||
continue
|
||
if p.parts and p.parts[0] == services_rel:
|
||
full = services_dir.joinpath(*p.parts[1:])
|
||
else:
|
||
full = REPO_DIR / p
|
||
try:
|
||
with full.open("rb") as fh:
|
||
total += sum(1 for _ in fh)
|
||
except OSError:
|
||
continue # not mounted in this container — skip it
|
||
return total
|
||
|
||
|
||
def count_loc(slug: str) -> int:
|
||
"""Total lines of source in a project (excludes deps/build/lock files).
|
||
|
||
TTL-cached: the walk reads every source file, and the figure only feeds a
|
||
cost-per-LoC stat — 5-minute staleness is invisible."""
|
||
entry = _safe_entry(slug)
|
||
if not entry and slug != ROOT_SLUG:
|
||
return 0
|
||
hit = _LOC_CACHE.get(slug)
|
||
if hit and time.time() - hit[0] < _LOC_TTL_SECS:
|
||
return hit[1]
|
||
total = _walk_loc(entry) if entry else _root_loc()
|
||
_LOC_CACHE[slug] = (time.time(), total)
|
||
return total
|
||
|
||
|
||
def project_asset_path(slug: str, rel: str) -> Path | None:
|
||
"""Resolve a whitelisted image asset within a project, for serving."""
|
||
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 p.is_file() and p.suffix.lower() in IMAGE_EXTS:
|
||
return p
|
||
return None
|