Gabriel Vidal
c1206dc45f
Follow-up to the move-projects-out migration: skill docs (SKILL.md descriptions and bodies), CLAUDE.md commit guidance, service READMEs/ CLAUDE.mds, the kanban Dockerfile comment and two backend docstrings all still pointed at the old in-repo projects/ location. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
518 lines
18 KiB
Python
518 lines
18 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"\'`)]*)?')
|
||
|
||
# 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]:
|
||
if not PROJECTS_DIR.is_dir():
|
||
return []
|
||
items: list[dict] = []
|
||
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))
|
||
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 project_detail(slug: str) -> dict | None:
|
||
entry = _safe_entry(slug)
|
||
if not entry:
|
||
return 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 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:
|
||
return 0
|
||
hit = _LOC_CACHE.get(slug)
|
||
if hit and time.time() - hit[0] < _LOC_TTL_SECS:
|
||
return hit[1]
|
||
total = 0
|
||
for dirpath, dirnames, filenames in os.walk(entry):
|
||
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
|
||
_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
|