a291f3cb6e
Conversations that never edit a `projects/<slug>` file — service work, skills, scripts, docs, plain questions — used to fall out of every project-shaped view as "(unattributed)". They now land on a root project card backed by the repo root itself. - projects.py: ROOT_SLUG (`lab`), attributed_projects() (derived at read time, never written to the sidecar), project_dir()/repo_rel() so callers resolve the slug to REPO_DIR instead of assuming projects/<dir>, plus the gallery record and detail view (repo CLAUDE.md, GOAL.md, git history). A real ~/projects/lab dir would win over the synthesised record. - main.py: _conv_meta applies the fallback (so tags, project pages, graph and dashboards all see it), and the goals board + goal-work resolve the unit's dir through project_dir()/repo_rel(). - project_costs.py: same fallback, so the lab's spend rolls up on its card. - gitdiff.py: `project:lab` resolves to the superproject. - count_loc: the lab's LoC comes from `git ls-files` (tracked source only) — a directory walk would count services/media and the untracked data/ tree. - Frontend: house icon, dashed tinted frame and a `root` badge on the card and detail page; no .env editor; the tag's repo path reads `.`.
666 lines
24 KiB
Python
666 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)".
|
||
#
|
||
# `lab` 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.
|
||
#
|
||
# If a real `~/projects/lab` 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", "lab")
|
||
ROOT_DESCRIPTION = (
|
||
"The homelab repo itself — services, Traefik, skills, scripts and docs. "
|
||
"Home for every conversation that didn't work on a specific project."
|
||
)
|
||
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
|