206 lines
8.1 KiB
Python
206 lines
8.1 KiB
Python
"""The **skills catalog** — every ``.claude/skills/<name>/SKILL.md`` on disk,
|
||
joined with the usage mined from the Claude Code transcripts.
|
||
|
||
Two halves, deliberately kept apart:
|
||
|
||
* :func:`catalog` is the *disk* half — a static scan of the skill dirs the
|
||
container can see (the repo's own ``.claude/skills`` plus each project's),
|
||
reading each ``SKILL.md``'s frontmatter for its name/description and the dir's
|
||
file count + newest mtime. No DB, no transcripts.
|
||
* :func:`usage_by_skill` is the *transcript* half — it turns the per-transcript
|
||
``skills_json`` blobs (``{skill: {count, last}}``, written by the indexer) plus
|
||
the matching conversation summaries into a per-skill rollup: how many calls,
|
||
in how many conversations, when it last ran, and what those conversations cost
|
||
in total. Attribution is deliberately coarse — a conversation's whole spend is
|
||
credited to *each* skill it invoked, so ``sessionCost`` answers "how much money
|
||
flowed through work that used this skill", not "what did this skill cost".
|
||
The narrower per-invocation figure is ``estSpend`` (context load × calls),
|
||
computed by the caller from the file-token table.
|
||
|
||
``main.py`` merges the two: the union of disk skills and used-skill names, so a
|
||
built-in/plugin skill (``code-review``, ``artifact-design``, …) that never had a
|
||
``SKILL.md`` in this repo still shows up with its usage, marked ``builtin``.
|
||
|
||
**Name collisions merge.** The ``skills`` table is keyed by bare skill name, so a
|
||
project-local skill sharing a name with a repo one shares its counts. That's the
|
||
same identity Claude Code itself uses when a session invokes ``Skill(name)``.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import json
|
||
import os
|
||
import pathlib
|
||
|
||
from conversations import usage_cost, usage_tokens, zero_usage
|
||
|
||
WORKSPACE = pathlib.Path(os.environ.get("WORKSPACE", "/workspace")).resolve()
|
||
PROJECTS_DIR = pathlib.Path(
|
||
os.environ.get("PROJECTS_DIR", "/workspace/projects")).resolve()
|
||
|
||
# Dirs a skill scan never descends into (mirrors main.IGNORE_DIRS' intent, but
|
||
# a skill dir is small and flat — this only guards against a stray vendor tree).
|
||
IGNORE_DIRS = {"node_modules", ".git", "__pycache__", "dist", "build", ".venv"}
|
||
|
||
# A skill dir is only a skill if it holds this file.
|
||
SKILL_FILE = "SKILL.md"
|
||
|
||
|
||
def _iso(epoch: float) -> str:
|
||
return (datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc)
|
||
.isoformat().replace("+00:00", "Z"))
|
||
|
||
|
||
def _parse_frontmatter(text: str) -> dict:
|
||
"""Top-level ``key: value`` pairs from a ``---`` frontmatter block.
|
||
|
||
Hand-rolled rather than PyYAML on purpose: a skill ``description`` is one
|
||
long unquoted sentence that routinely contains ``: `` (``"via the gh CLI —
|
||
PRs, issues: …"``), which is a YAML syntax error. The shape here is fixed
|
||
(a flat map of single-line values), so splitting on the first colon is both
|
||
correct and total.
|
||
"""
|
||
if not text.startswith("---"):
|
||
return {}
|
||
end = text.find("\n---", 3)
|
||
if end == -1:
|
||
return {}
|
||
out: dict[str, str] = {}
|
||
for line in text[3:end].split("\n"):
|
||
if not line.strip() or line.startswith((" ", "\t", "#")):
|
||
continue # nested map entries / comments: not used here
|
||
key, sep, val = line.partition(":")
|
||
if not sep:
|
||
continue
|
||
out[key.strip()] = val.strip().strip("'\"")
|
||
return out
|
||
|
||
|
||
def _dir_stats(d: pathlib.Path) -> tuple[int, float]:
|
||
"""(file count, newest mtime) for a skill dir, walked once."""
|
||
files = 0
|
||
newest = 0.0
|
||
for dirpath, dirnames, filenames in os.walk(d):
|
||
dirnames[:] = [x for x in dirnames if x not in IGNORE_DIRS]
|
||
for fn in filenames:
|
||
files += 1
|
||
try:
|
||
mt = (pathlib.Path(dirpath) / fn).stat().st_mtime
|
||
except OSError:
|
||
continue
|
||
if mt > newest:
|
||
newest = mt
|
||
return files, newest
|
||
|
||
|
||
def _skill_roots() -> list[tuple[str, str, pathlib.Path]]:
|
||
"""The ``.claude/skills`` dirs to scan, as (source, sourceKind, path).
|
||
|
||
``source`` labels where a skill comes from in the UI: ``"repo"`` for the
|
||
homelab's own skills, the project's dir-name for a project-local one.
|
||
"""
|
||
roots: list[tuple[str, str, pathlib.Path]] = [
|
||
("repo", "repo", WORKSPACE / ".claude" / "skills"),
|
||
]
|
||
try:
|
||
entries = sorted(PROJECTS_DIR.iterdir())
|
||
except OSError:
|
||
entries = []
|
||
for entry in entries:
|
||
if entry.name.startswith(".") or not entry.is_dir():
|
||
continue
|
||
d = entry / ".claude" / "skills"
|
||
if d.is_dir():
|
||
roots.append((entry.name, "project", d))
|
||
return roots
|
||
|
||
|
||
def catalog() -> list[dict]:
|
||
"""Every skill found on disk, newest-edited first.
|
||
|
||
Each record is the *static* half of a skill card — ``path`` is repo-relative
|
||
so the frontend can deep-link it into the file editor (``/_/<path>``).
|
||
"""
|
||
out: list[dict] = []
|
||
for source, source_kind, root in _skill_roots():
|
||
try:
|
||
entries = sorted(root.iterdir())
|
||
except OSError:
|
||
continue
|
||
for entry in entries:
|
||
if entry.name.startswith(".") or not entry.is_dir():
|
||
continue
|
||
md = entry / SKILL_FILE
|
||
if not md.is_file():
|
||
continue
|
||
try:
|
||
fm = _parse_frontmatter(md.read_text(encoding="utf-8", errors="replace"))
|
||
except OSError:
|
||
fm = {}
|
||
files, newest = _dir_stats(entry)
|
||
try:
|
||
rel = entry.relative_to(WORKSPACE).as_posix()
|
||
except ValueError:
|
||
rel = entry.as_posix()
|
||
out.append({
|
||
# The invocation name is the dir name — that's what `Skill(...)`
|
||
# takes, and what the transcript counts are keyed by. Frontmatter
|
||
# `name:` is only a fallback for a dir/frontmatter mismatch.
|
||
"name": entry.name,
|
||
"title": fm.get("name") or entry.name,
|
||
"description": fm.get("description", ""),
|
||
"dir": rel,
|
||
"path": f"{rel}/{SKILL_FILE}",
|
||
"source": source,
|
||
"sourceKind": source_kind,
|
||
"files": files,
|
||
"updatedAt": _iso(newest) if newest else None,
|
||
})
|
||
out.sort(key=lambda s: s["updatedAt"] or "", reverse=True)
|
||
return out
|
||
|
||
|
||
def usage_by_skill(summaries: list[tuple[str, dict]],
|
||
skill_rows: list[tuple[str, str]]) -> dict[str, dict]:
|
||
"""``{skill name -> {count, lastUsed, conversations, sessionCost, sessionTokens}}``.
|
||
|
||
``summaries`` is ``store.all_summaries()`` (``[(path, summary), …]``) and
|
||
``skill_rows`` is ``store.all_transcript_skill_rows()`` — the same transcript
|
||
path keys both, which is what lets a skill's calls be joined to the spend of
|
||
the conversations they ran in.
|
||
"""
|
||
by_path = dict(summaries)
|
||
out: dict[str, dict] = {}
|
||
for path, raw in skill_rows:
|
||
try:
|
||
contrib = json.loads(raw or "{}")
|
||
except ValueError:
|
||
continue
|
||
if not contrib:
|
||
continue
|
||
s = by_path.get(path) or {}
|
||
# Same fallback as project_costs: older summaries predate the stored
|
||
# cost/tokens rollup and only carry the raw `usage` block.
|
||
cost = s.get("cost")
|
||
tokens = s.get("tokens")
|
||
if cost is None or tokens is None:
|
||
u = {**zero_usage(), **(s.get("usage") or {})}
|
||
model = s.get("model") or ""
|
||
if cost is None:
|
||
cost = usage_cost(u, model)
|
||
if tokens is None:
|
||
tokens = usage_tokens(u)
|
||
for name, d in contrib.items():
|
||
e = out.setdefault(name, {
|
||
"count": 0, "lastUsed": None, "conversations": 0,
|
||
"sessionCost": 0.0, "sessionTokens": 0,
|
||
})
|
||
e["count"] += d.get("count", 0)
|
||
last = d.get("last")
|
||
if last and (e["lastUsed"] is None or last > e["lastUsed"]):
|
||
e["lastUsed"] = last
|
||
e["conversations"] += 1
|
||
e["sessionCost"] += cost
|
||
e["sessionTokens"] += tokens
|
||
return out
|