Gabriel Vidal
6641c588a1
Attribute a conversation to a project/service only when it actually created or edited a file there — the target path of a file-mutating tool (Write/Edit/MultiEdit/NotebookEdit) — instead of any cwd, read, search, or bash command that merely mentions the path. This focuses the auto-tag list on projects the conversation genuinely changed. Bumps PARSER_VERSION to 12 to force a full re-parse of all archived transcripts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""Aggregate Claude conversation spend per homelab project.
|
||
|
||
A conversation is attributed to a project when the project's directory name
|
||
appears in its inferred ``projectsAuto`` list (projects it created/edited files
|
||
in) *or* the manual ``projects`` list in the action-metadata sidecar — the exact
|
||
same union the viewer uses to decide which conversations to list on a project
|
||
page (``meta.projects``). Every
|
||
attributed conversation's real per-turn usage is rolled up into a per-project
|
||
total: dollar/token totals, mean/median per conversation, and a per-token-type
|
||
($ + tokens) breakdown. The project *detail* view additionally divides the total
|
||
by the project's lines of code (see ``projects.count_loc``).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from statistics import median
|
||
|
||
from conversations import price_for, usage_cost, usage_tokens, zero_usage
|
||
|
||
# Token types we break spend down by, in display order. Each carries its own
|
||
# token count and dollar cost (priced with the conversation's model).
|
||
TYPES = ("input", "output", "cacheRead", "cacheWrite")
|
||
|
||
|
||
def _project_slugs(summary: dict, meta_all: dict) -> set[str]:
|
||
"""The set of project dir-names a conversation is attributed to."""
|
||
slugs = set(summary.get("projectsAuto") or [])
|
||
sid = summary.get("sessionId") or ""
|
||
m = meta_all.get(sid) or {}
|
||
for p in (m.get("projects") or []):
|
||
if p:
|
||
slugs.add(p)
|
||
return slugs
|
||
|
||
|
||
def _type_costs(u: dict, model: str) -> dict[str, dict]:
|
||
"""Split one usage block into per-token-type {tokens, cost}.
|
||
|
||
Mirrors ``conversations.usage_cost``: input at the input rate, output at the
|
||
output rate, cache reads at 0.1× input, cache writes priced off their
|
||
weighted *units* (5m ≈ 1.25×, 1h ≈ 2×) but counted in raw tokens.
|
||
"""
|
||
pi, po = price_for(model)
|
||
pi /= 1e6
|
||
po /= 1e6
|
||
return {
|
||
"input": {"tokens": u["input"], "cost": u["input"] * pi},
|
||
"output": {"tokens": u["output"], "cost": u["output"] * po},
|
||
"cacheRead": {"tokens": u["cacheRead"], "cost": u["cacheRead"] * 0.1 * pi},
|
||
"cacheWrite": {"tokens": u["cacheWriteTokens"], "cost": u["cacheWriteUnits"] * pi},
|
||
}
|
||
|
||
|
||
def _aggregate(convs: list[dict]) -> dict:
|
||
"""Roll a project's attributed conversations into a spend summary."""
|
||
costs: list[float] = []
|
||
tokens: list[int] = []
|
||
total_cost = 0.0
|
||
total_tokens = 0
|
||
by_type = {t: {"tokens": 0, "cost": 0.0} for t in TYPES}
|
||
|
||
for s in convs:
|
||
model = s.get("model") or ""
|
||
u = {**zero_usage(), **(s.get("usage") or {})}
|
||
c = s.get("cost")
|
||
if c is None:
|
||
c = usage_cost(u, model)
|
||
tk = s.get("tokens")
|
||
if tk is None:
|
||
tk = usage_tokens(u)
|
||
costs.append(c)
|
||
tokens.append(tk)
|
||
total_cost += c
|
||
total_tokens += tk
|
||
for t, v in _type_costs(u, model).items():
|
||
by_type[t]["tokens"] += int(v["tokens"])
|
||
by_type[t]["cost"] += v["cost"]
|
||
|
||
n = len(convs)
|
||
return {
|
||
"conversations": n,
|
||
"cost": total_cost,
|
||
"tokens": total_tokens,
|
||
"meanCost": total_cost / n if n else 0.0,
|
||
"medianCost": float(median(costs)) if costs else 0.0,
|
||
"meanTokens": total_tokens / n if n else 0.0,
|
||
"medianTokens": float(median(tokens)) if tokens else 0.0,
|
||
"byType": by_type,
|
||
}
|
||
|
||
|
||
def costs_by_project(summaries: list[tuple[str, dict]], meta_all: dict) -> dict[str, dict]:
|
||
"""{project dir-name -> spend aggregate} across all conversations.
|
||
|
||
``summaries`` is ``store.all_summaries()`` (``[(path, summary), …]``);
|
||
``meta_all`` is ``meta_store.all()``.
|
||
"""
|
||
buckets: dict[str, list[dict]] = {}
|
||
for _path, s in summaries:
|
||
if not s.get("messages"):
|
||
continue
|
||
for slug in _project_slugs(s, meta_all):
|
||
buckets.setdefault(slug, []).append(s)
|
||
return {slug: _aggregate(convs) for slug, convs in buckets.items()}
|