8d4aa4094c
- usage_tracker.py: daily token/call buckets per user (home/{user}/usage.json)
- Hook into local backend (OpenAI usage field) and Gemini API (usage_metadata)
- Claude/Gemini CLI backends produce no structured token data and are not tracked
- Fix CLAUDE.md stale tool count (27 → 39) and refresh tool list
- scripts/import_knowledge.py: walk markdown dirs, chunk by H2, call local LLM
for summaries, create AE journal entries with path-derived tags; resumable via
state file; --dry-run and --limit flags for safe testing
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
"""
|
|
API usage and token tracking.
|
|
|
|
Writes daily buckets to home/{username}/usage.json:
|
|
|
|
{
|
|
"2026-05-01": {
|
|
"gemini_api/gemini-2.0-flash": {"calls": 3, "prompt_tokens": 8400, "completion_tokens": 520},
|
|
"local/llama3.2:latest": {"calls": 2, "prompt_tokens": 1200, "completion_tokens": 310}
|
|
}
|
|
}
|
|
|
|
Claude CLI and Gemini CLI backends produce no structured token data and are not tracked.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_LOCK = asyncio.Lock()
|
|
|
|
|
|
def _usage_path(username: str) -> Path:
|
|
return settings.home_root() / username / "usage.json"
|
|
|
|
|
|
async def record(
|
|
username: str,
|
|
backend: str,
|
|
model_name: str,
|
|
prompt_tokens: int,
|
|
completion_tokens: int,
|
|
) -> None:
|
|
"""Append one call's token counts to the daily usage log for this user.
|
|
|
|
backend — "gemini_api" | "local"
|
|
model_name — the exact model string (e.g. "gemini-2.0-flash", "llama3.2:latest")
|
|
"""
|
|
path = _usage_path(username)
|
|
today = date.today().isoformat()
|
|
key = f"{backend}/{model_name}"
|
|
|
|
async with _LOCK:
|
|
try:
|
|
data: dict = json.loads(path.read_text()) if path.exists() else {}
|
|
except Exception:
|
|
data = {}
|
|
|
|
entry = data.setdefault(today, {}).setdefault(
|
|
key, {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0}
|
|
)
|
|
entry["calls"] += 1
|
|
entry["prompt_tokens"] += prompt_tokens
|
|
entry["completion_tokens"] += completion_tokens
|
|
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, indent=2))
|
|
except Exception as e:
|
|
logger.warning("Failed to write usage data to %s: %s", path, e)
|
|
|
|
|
|
def read_usage(username: str) -> dict:
|
|
"""Return the full usage dict for this user. Empty dict if no file yet."""
|
|
path = _usage_path(username)
|
|
try:
|
|
return json.loads(path.read_text()) if path.exists() else {}
|
|
except Exception:
|
|
return {}
|