my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	#!/usr/bin/env python3
#2	"""
#3	Summarize CodexBar local cost usage by model.
#4
#5	Defaults to current model (most recent daily entry), or list all models.
#6	"""
#7
#8	from __future__ import annotations
#9
#10	import argparse
#11	import json
#12	import os
#13	import subprocess
#14	import sys
#15	from dataclasses import dataclass
#16	from datetime import date, datetime, timedelta
#17	from typing import Any, Dict, Iterable, List, Optional, Tuple
#18
#19
#20	def eprint(msg: str) -> None:
#21	print(msg, file=sys.stderr)
#22
#23
#24	def run_codexbar_cost(provider: str) -> List[Dict[str, Any]]:
#25	cmd = ["codexbar", "cost", "--format", "json", "--provider", provider]
#26	try:
#27	output = subprocess.check_output(cmd, text=True)
#28	except FileNotFoundError:
#29	raise RuntimeError("codexbar not found on PATH. Install CodexBar CLI first.")
#30	except subprocess.CalledProcessError as exc:
#31	raise RuntimeError(f"codexbar cost failed (exit {exc.returncode}).")
#32	try:
#33	payload = json.loads(output)
#34	except json.JSONDecodeError as exc:
#35	raise RuntimeError(f"Failed to parse codexbar JSON output: {exc}")
#36	if not isinstance(payload, list):
#37	raise RuntimeError("Expected codexbar cost JSON array.")
#38	return payload
#39
#40
#41	def load_payload(input_path: Optional[str], provider: str) -> Dict[str, Any]:
#42	if input_path:
#43	if input_path == "-":
#44	raw = sys.stdin.read()
#45	else:
#46	with open(input_path, "r", encoding="utf-8") as handle:
#47	raw = handle.read()
#48	data = json.loads(raw)
#49	else:
#50	data = run_codexbar_cost(provider)
#51
#52	if isinstance(data, dict):
#53	return data
#54
#55	if isinstance(data, list):
#56	for entry in data:
#57	if isinstance(entry, dict) and entry.get("provider") == provider:
#58	return entry
#59	raise RuntimeError(f"Provider '{provider}' not found in codexbar payload.")
#60
#61	raise RuntimeError("Unsupported JSON input format.")
#62
#63
#64	@dataclass
#65	class ModelCost:
#66	model: str
#67	cost: float
#68
#69
#70	def parse_daily_entries(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
#71	daily = payload.get("daily")
#72	if not daily:
#73	return []
#74	if not isinstance(daily, list):
#75	return []
#76	return [entry for entry in daily if isinstance(entry, dict)]
#77
#78
#79	def parse_date(value: str) -> Optional[date]:
#80	try:
#81	return datetime.strptime(value, "%Y-%m-%d").date()
#82	except Exception:
#83	return None
#84
#85
#86	def filter_by_days(entries: List[Dict[str, Any]], days: Optional[int]) -> List[Dict[str, Any]]:
#87	if not days:
#88	return entries
#89	cutoff = date.today() - timedelta(days=days - 1)
#90	filtered: List[Dict[str, Any]] = []
#91	for entry in entries:
#92	day = entry.get("date")
#93	if not isinstance(day, str):
#94	continue
#95	parsed = parse_date(day)
#96	if parsed and parsed >= cutoff:
#97	filtered.append(entry)
#98	return filtered
#99
#100
#101	def aggregate_costs(entries: Iterable[Dict[str, Any]]) -> Dict[str, float]:
#102	totals: Dict[str, float] = {}
#103	for entry in entries:
#104	breakdowns = entry.get("modelBreakdowns")
#105	if not breakdowns:
#106	continue
#107	if not isinstance(breakdowns, list):
#108	continue
#109	for item in breakdowns:
#110	if not isinstance(item, dict):
#111	continue
#112	model = item.get("modelName")
#113	cost = item.get("cost")
#114	if not isinstance(model, str):
#115	continue
#116	if not isinstance(cost, (int, float)):
#117	continue
#118	totals[model] = totals.get(model, 0.0) + float(cost)
#119	return totals
#120
#121
#122	def pick_current_model(entries: List[Dict[str, Any]]) -> Tuple[Optional[str], Optional[str]]:
#123	if not entries:
#124	return None, None
#125	sorted_entries = sorted(
#126	entries,
#127	key=lambda entry: entry.get("date") or "",
#128	)
#129	for entry in reversed(sorted_entries):
#130	breakdowns = entry.get("modelBreakdowns")
#131	if isinstance(breakdowns, list) and breakdowns:
#132	scored: List[ModelCost] = []
#133	for item in breakdowns:
#134	if not isinstance(item, dict):
#135	continue
#136	model = item.get("modelName")
#137	cost = item.get("cost")
#138	if isinstance(model, str) and isinstance(cost, (int, float)):
#139	scored.append(ModelCost(model=model, cost=float(cost)))
#140	if scored:
#141	scored.sort(key=lambda item: item.cost, reverse=True)
#142	return scored[0].model, entry.get("date") if isinstance(entry.get("date"), str) else None
#143	models_used = entry.get("modelsUsed")
#144	if isinstance(models_used, list) and models_used:
#145	last = models_used[-1]
#146	if isinstance(last, str):
#147	return last, entry.get("date") if isinstance(entry.get("date"), str) else None
#148	return None, None
#149
#150
#151	def usd(value: Optional[float]) -> str:
#152	if value is None:
#153	return "—"
#154	return f"${value:,.2f}"
#155
#156
#157	def latest_day_cost(entries: List[Dict[str, Any]], model: str) -> Tuple[Optional[str], Optional[float]]:
#158	if not entries:
#159	return None, None
#160	sorted_entries = sorted(
#161	entries,
#162	key=lambda entry: entry.get("date") or "",
#163	)
#164	for entry in reversed(sorted_entries):
#165	breakdowns = entry.get("modelBreakdowns")
#166	if not isinstance(breakdowns, list):
#167	continue
#168	for item in breakdowns:
#169	if not isinstance(item, dict):
#170	continue
#171	if item.get("modelName") == model:
#172	cost = item.get("cost") if isinstance(item.get("cost"), (int, float)) else None
#173	day = entry.get("date") if isinstance(entry.get("date"), str) else None
#174	return day, float(cost) if cost is not None else None
#175	return None, None
#176
#177
#178	def render_text_current(
#179	provider: str,
#180	model: str,
#181	latest_date: Optional[str],
#182	total_cost: Optional[float],
#183	latest_cost: Optional[float],
#184	latest_cost_date: Optional[str],
#185	entry_count: int,
#186	) -> str:
#187	lines = [f"Provider: {provider}", f"Current model: {model}"]
#188	if latest_date:
#189	lines.append(f"Latest model date: {latest_date}")
#190	lines.append(f"Total cost (rows): {usd(total_cost)}")
#191	if latest_cost_date:
#192	lines.append(f"Latest day cost: {usd(latest_cost)} ({latest_cost_date})")
#193	lines.append(f"Daily rows: {entry_count}")
#194	return "\n".join(lines)
#195
#196
#197	def render_text_all(provider: str, totals: Dict[str, float]) -> str:
#198	lines = [f"Provider: {provider}", "Models:"]
#199	for model, cost in sorted(totals.items(), key=lambda item: item[1], reverse=True):
#200	lines.append(f"- {model}: {usd(cost)}")
#201	return "\n".join(lines)
#202
#203
#204	def build_json_current(
#205	provider: str,
#206	model: str,
#207	latest_date: Optional[str],
#208	total_cost: Optional[float],
#209	latest_cost: Optional[float],
#210	latest_cost_date: Optional[str],
#211	entry_count: int,
#212	) -> Dict[str, Any]:
#213	return {
#214	"provider": provider,
#215	"mode": "current",
#216	"model": model,
#217	"latestModelDate": latest_date,
#218	"totalCostUSD": total_cost,
#219	"latestDayCostUSD": latest_cost,
#220	"latestDayCostDate": latest_cost_date,
#221	"dailyRowCount": entry_count,
#222	}
#223
#224
#225	def build_json_all(provider: str, totals: Dict[str, float]) -> Dict[str, Any]:
#226	return {
#227	"provider": provider,
#228	"mode": "all",
#229	"models": [
#230	{"model": model, "totalCostUSD": cost}
#231	for model, cost in sorted(totals.items(), key=lambda item: item[1], reverse=True)
#232	],
#233	}
#234
#235
#236	def main() -> int:
#237	parser = argparse.ArgumentParser(description="Summarize CodexBar model usage from local cost logs.")
#238	parser.add_argument("--provider", choices=["codex", "claude"], default="codex")
#239	parser.add_argument("--mode", choices=["current", "all"], default="current")
#240	parser.add_argument("--model", help="Explicit model name to report instead of auto-current.")
#241	parser.add_argument("--input", help="Path to codexbar cost JSON (or '-' for stdin).")
#242	parser.add_argument("--days", type=int, help="Limit to last N days (based on daily rows).")
#243	parser.add_argument("--format", choices=["text", "json"], default="text")
#244	parser.add_argument("--pretty", action="store_true", help="Pretty-print JSON output.")
#245
#246	args = parser.parse_args()
#247
#248	try:
#249	payload = load_payload(args.input, args.provider)
#250	except Exception as exc:
#251	eprint(str(exc))
#252	return 1
#253
#254	entries = parse_daily_entries(payload)
#255	entries = filter_by_days(entries, args.days)
#256
#257	if args.mode == "current":
#258	model = args.model
#259	latest_date = None
#260	if not model:
#261	model, latest_date = pick_current_model(entries)
#262	if not model:
#263	eprint("No model data found in codexbar cost payload.")
#264	return 2
#265	totals = aggregate_costs(entries)
#266	total_cost = totals.get(model)
#267	latest_cost_date, latest_cost = latest_day_cost(entries, model)
#268
#269	if args.format == "json":
#270	payload_out = build_json_current(
#271	provider=args.provider,
#272	model=model,
#273	latest_date=latest_date,
#274	total_cost=total_cost,
#275	latest_cost=latest_cost,
#276	latest_cost_date=latest_cost_date,
#277	entry_count=len(entries),
#278	)
#279	indent = 2 if args.pretty else None
#280	print(json.dumps(payload_out, indent=indent, sort_keys=args.pretty))
#281	else:
#282	print(
#283	render_text_current(
#284	provider=args.provider,
#285	model=model,
#286	latest_date=latest_date,
#287	total_cost=total_cost,
#288	latest_cost=latest_cost,
#289	latest_cost_date=latest_cost_date,
#290	entry_count=len(entries),
#291	)
#292	)
#293	return 0
#294
#295	totals = aggregate_costs(entries)
#296	if not totals:
#297	eprint("No model breakdowns found in codexbar cost payload.")
#298	return 2
#299
#300	if args.format == "json":
#301	payload_out = build_json_all(provider=args.provider, totals=totals)
#302	indent = 2 if args.pretty else None
#303	print(json.dumps(payload_out, indent=indent, sort_keys=args.pretty))
#304	else:
#305	print(render_text_all(provider=args.provider, totals=totals))
#306	return 0
#307
#308
#309	if __name__ == "__main__":
#310	raise SystemExit(main())
#311

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public