ai-agent/backend/projects.py

"""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