#!/usr/bin/env python3 """Backfill numeric Codex token history through Baxian's OTLP endpoint. Only token counters, timestamps, and an opaque session id are read from local Codex JSONL files. Prompt, response, code, tool output, and file content are never included in the upload. """ from __future__ import annotations import argparse import datetime as dt import json import os import re import ssl import time import urllib.request from pathlib import Path from typing import Any, Iterable TOKEN_FIELDS = ( "input_tokens", "cached_input_tokens", "output_tokens", "reasoning_output_tokens", "total_tokens", ) SESSION_ID = re.compile( r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})" ) def parse_timestamp(value: str) -> int | None: try: return int(dt.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()) except (TypeError, ValueError): return None def positive_int(value: Any) -> int: if isinstance(value, bool): return 0 if isinstance(value, (int, float)): return max(0, int(value)) return 0 def token_files(codex_home: Path) -> Iterable[Path]: for directory in (codex_home / "sessions", codex_home / "archived_sessions"): if directory.is_dir(): yield from sorted(directory.rglob("*.jsonl")) def collect_records(codex_home: Path, cutoff: int) -> tuple[list[dict[str, Any]], dict[str, int]]: records: list[dict[str, Any]] = [] totals = {field: 0 for field in TOKEN_FIELDS} totals.update({"files": 0, "active_minutes": 0}) minute_buckets: set[int] = set() for path in token_files(codex_home): match = SESSION_ID.search(path.name) source_id = match.group(1) if match else path.stem previous = {field: 0 for field in TOKEN_FIELDS} sequence = 0 found = False with path.open("r", encoding="utf-8") as stream: for raw_line in stream: try: item = json.loads(raw_line) except json.JSONDecodeError: continue payload = item.get("payload") if not isinstance(payload, dict) or payload.get("type") != "token_count": continue info = payload.get("info") usage = info.get("total_token_usage") if isinstance(info, dict) else None if not isinstance(usage, dict): continue event_time = parse_timestamp(item.get("timestamp", "")) if event_time is None or event_time >= cutoff: continue current = {field: positive_int(usage.get(field)) for field in TOKEN_FIELDS} delta = { field: ( current[field] - previous[field] if current[field] >= previous[field] else current[field] ) for field in TOKEN_FIELDS } previous = current if delta["total_tokens"] <= 0: continue found = True sequence += 1 minute_buckets.add(event_time // 60) for field in TOKEN_FIELDS: totals[field] += delta[field] records.append( history_record( source_id=source_id, sequence=sequence, event_time=event_time, usage=delta, ) ) if found: totals["files"] += 1 totals["active_minutes"] = len(minute_buckets) return records, totals def string_attribute(key: str, value: str) -> dict[str, Any]: return {"key": key, "value": {"stringValue": value}} def int_attribute(key: str, value: int) -> dict[str, Any]: return {"key": key, "value": {"intValue": str(value)}} def history_record( source_id: str, sequence: int, event_time: int, usage: dict[str, int] ) -> dict[str, Any]: return { "timeUnixNano": str(event_time * 1_000_000_000), "attributes": [ string_attribute("event.name", "codex.sse_event"), string_attribute("event.kind", "response.completed"), string_attribute("model", "Codex history"), string_attribute("baxian.source", "history"), string_attribute("baxian.session_id", source_id), int_attribute("baxian.sequence", sequence), int_attribute("input_token_count", usage["input_tokens"]), int_attribute("cached_token_count", usage["cached_input_tokens"]), int_attribute("output_token_count", usage["output_tokens"]), int_attribute("reasoning_token_count", usage["reasoning_output_tokens"]), int_attribute("tool_token_count", usage["total_tokens"]), ], "body": {"stringValue": "baxian.history.backfill"}, } def batches(items: list[dict[str, Any]], size: int = 500) -> Iterable[list[dict[str, Any]]]: for start in range(0, len(items), size): yield items[start : start + size] def verified_tls_context() -> ssl.SSLContext: """Use the OS CA bundle when framework Python does not discover it.""" system_bundle = Path("/etc/ssl/cert.pem") if system_bundle.is_file(): return ssl.create_default_context(cafile=str(system_bundle)) return ssl.create_default_context() def upload(endpoint: str, token: str, records: list[dict[str, Any]]) -> tuple[int, int]: accepted = 0 tokens = 0 context = verified_tls_context() for batch in batches(records): document = { "resourceLogs": [ { "resource": { "attributes": [ string_attribute("service.name", "baxian-history"), string_attribute("deployment.environment", "baxian"), ] }, "scopeLogs": [ { "scope": {"name": "baxian-history", "version": "1.0"}, "logRecords": batch, } ], } ] } request = urllib.request.Request( endpoint, data=json.dumps(document, separators=(",", ":")).encode(), method="POST", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": "Baxian-History/1.0", }, ) with urllib.request.urlopen(request, timeout=30, context=context) as response: accepted += int(response.headers.get("X-Baxian-Accepted", "0")) tokens += int(response.headers.get("X-Baxian-Tokens", "0")) return accepted, tokens def main() -> None: parser = argparse.ArgumentParser(description="回填 Codex 历史 Token 数值到八仙") parser.add_argument( "--endpoint", default="https://baxian.wanhe.cn/v1/logs", help="八仙 OTLP 地址" ) parser.add_argument("--token", default=os.environ.get("BAXIAN_TOKEN"), help="个人八仙令牌") parser.add_argument( "--codex-home", type=Path, default=Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")), ) parser.add_argument( "--cutoff", type=int, default=int(time.time()), help="只回填此 Unix 时间之前的记录", ) parser.add_argument("--dry-run", action="store_true", help="只计算，不上传") args = parser.parse_args() records, totals = collect_records(args.codex_home, args.cutoff) summary = { "sessions": totals["files"], "records": len(records), "activeMinutes": totals["active_minutes"], "inputTokens": totals["input_tokens"], "cachedTokens": totals["cached_input_tokens"], "outputTokens": totals["output_tokens"], "reasoningTokens": totals["reasoning_output_tokens"], "totalTokens": totals["total_tokens"], "cutoff": args.cutoff, } print(json.dumps(summary, ensure_ascii=False, indent=2)) if args.dry_run: return if not args.token: parser.error("请通过 --token 或 BAXIAN_TOKEN 提供个人令牌") accepted, uploaded_tokens = upload(args.endpoint, args.token, records) print( json.dumps( {"acceptedRecords": accepted, "uploadedTokens": uploaded_tokens}, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()