d1d4bd1514
Persists per-day token usage (usage_store, ~/.config/wsai/usage.json, survives
restarts) and surfaces it in status as claude_usage.{today,week}. The navbar now
shows two cards — 오늘 토큰 / 주간 토큰 (input+output, with per-card tooltips
breaking down input/output tokens and requests) — replacing the session-only
token count.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
"""Persisted per-day Claude token usage, for the dashboard's 하루/일주일 사용량.
|
|
|
|
Each brain call records its token usage into a date bucket
|
|
(``{ "2026-08-22": {"req": n, "in": n, "out": n}, ... }``) persisted to disk so
|
|
the totals survive a restart. ``summary()`` returns today's and the last-7-days
|
|
aggregates. Kept in memory; the file is read once and rewritten on each record.
|
|
|
|
Path: ``$WSAI_USAGE_PATH`` or ``~/.config/wsai/usage.json`` (disk, never tmpfs).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
|
|
_LOCK = threading.Lock()
|
|
_DATA: dict[str, dict[str, int]] | None = None
|
|
_KEEP_DAYS = 90
|
|
|
|
|
|
def _path() -> Path:
|
|
override = os.environ.get("WSAI_USAGE_PATH")
|
|
return Path(override) if override else (Path.home() / ".config" / "wsai" / "usage.json")
|
|
|
|
|
|
def _load() -> dict[str, dict[str, int]]:
|
|
global _DATA
|
|
if _DATA is not None:
|
|
return _DATA
|
|
try:
|
|
_DATA = json.loads(_path().read_text(encoding="utf-8"))
|
|
assert isinstance(_DATA, dict)
|
|
except (OSError, ValueError, AssertionError):
|
|
_DATA = {}
|
|
return _DATA
|
|
|
|
|
|
def _write() -> None:
|
|
p = _path()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(json.dumps(_DATA, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def record(input_tokens: int, output_tokens: int) -> None:
|
|
"""Add one call's usage to today's bucket and persist."""
|
|
today = date.today().isoformat()
|
|
with _LOCK:
|
|
data = _load()
|
|
b = data.setdefault(today, {"req": 0, "in": 0, "out": 0})
|
|
b["req"] += 1
|
|
b["in"] += int(input_tokens or 0)
|
|
b["out"] += int(output_tokens or 0)
|
|
# Prune buckets older than _KEEP_DAYS so the file can't grow forever.
|
|
cutoff = (date.today() - timedelta(days=_KEEP_DAYS)).isoformat()
|
|
for d in [d for d in data if d < cutoff]:
|
|
del data[d]
|
|
_write()
|
|
|
|
|
|
def summary() -> dict:
|
|
"""Today's and the last-7-days token totals."""
|
|
with _LOCK:
|
|
data = _load()
|
|
today = date.today().isoformat()
|
|
week_start = (date.today() - timedelta(days=6)).isoformat()
|
|
t = data.get(today, {"req": 0, "in": 0, "out": 0})
|
|
wk = {"req": 0, "in": 0, "out": 0}
|
|
for d, b in data.items():
|
|
if d >= week_start:
|
|
wk["req"] += b.get("req", 0)
|
|
wk["in"] += b.get("in", 0)
|
|
wk["out"] += b.get("out", 0)
|
|
return {
|
|
"today": {"requests": t.get("req", 0), "input": t.get("in", 0), "output": t.get("out", 0)},
|
|
"week": {"requests": wk["req"], "input": wk["in"], "output": wk["out"]},
|
|
}
|