from __future__ import annotations
import html
from pathlib import Path
from .core import Bucket, SessionUsage, UsageReport
def render_markdown(report: UsageReport) -> str:
lines: list[str] = [
"# Codex Usage Report",
"",
f"Generated: {report.generated_at.isoformat(timespec='seconds')}",
f"Timezone: `{report.timezone_name}`",
f"State DB: `{report.state_path}`",
f"Sessions dir: `{report.sessions_dir}`",
f"Date range: {report.date_range_label}",
]
if report.days_filter is not None:
lines.append(f"Filter: last {report.days_filter} local calendar days")
lines.extend(
[
"",
"## Summary",
"",
_markdown_table(
["Metric", "Value"],
[
["Sessions", _fmt(report.sessions)],
["Total tokens", _fmt(report.total_tokens)],
["Rollouts with token breakdown", _fmt(report.parsed_rollout_count)],
["Missing rollouts", _fmt(report.missing_rollout_count)],
["Warnings", _fmt(report.warnings)],
],
),
"",
]
)
if report.parsed_rollout_count:
lines.extend(
[
"## Rollout Token Breakdown",
"",
_markdown_table(
["Input", "Cached input", "Output", "Reasoning output", "Rollout total"],
[
[
_fmt(report.rollout_totals.input_tokens),
_fmt(report.rollout_totals.cached_input_tokens),
_fmt(report.rollout_totals.output_tokens),
_fmt(report.rollout_totals.reasoning_output_tokens),
_fmt(report.rollout_totals.total_tokens),
]
],
),
"",
]
)
sections: list[tuple[str, list[Bucket]]] = [
("Daily Tokens", report.daily),
("Weekly Tokens", report.weekly),
("Tokens By Cwd", report.by_cwd),
("Tokens By Model", report.by_model),
("Tokens By Role", report.by_role),
("Main Vs Subagent", report.by_kind),
]
for title, buckets in sections:
lines.extend(["## " + title, "", _render_bucket_markdown(buckets), ""])
lines.extend(["## Top Sessions", "", _render_top_sessions_markdown(report.top_sessions), ""])
if report.warnings:
lines.extend(
[
"## Data Warnings",
"",
*[f"- `{_escape_md(warning)}`" for warning in report.warnings[:50]],
]
)
if len(report.warnings) > 50:
lines.append(f"- ... {len(report.warnings) - 50} more")
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def render_html(report: UsageReport) -> str:
sections = "\n".join(
[
_html_section("Daily Tokens", _render_bucket_html(report.daily)),
_html_section("Weekly Tokens", _render_bucket_html(report.weekly)),
_html_section("Tokens By Cwd", _render_bucket_html(report.by_cwd)),
_html_section("Tokens By Model", _render_bucket_html(report.by_model)),
_html_section("Tokens By Role", _render_bucket_html(report.by_role)),
_html_section("Main Vs Subagent", _render_bucket_html(report.by_kind)),
_html_section("Top Sessions", _render_top_sessions_html(report.top_sessions)),
]
)
rollout_section = ""
if report.parsed_rollout_count:
rollout_section = _html_section(
"Rollout Token Breakdown",
_html_table(
["Input", "Cached input", "Output", "Reasoning output", "Rollout total"],
[
[
_fmt(report.rollout_totals.input_tokens),
_fmt(report.rollout_totals.cached_input_tokens),
_fmt(report.rollout_totals.output_tokens),
_fmt(report.rollout_totals.reasoning_output_tokens),
_fmt(report.rollout_totals.total_tokens),
]
],
),
)
warnings_section = ""
if report.warnings:
items = "\n".join(
f"
{html.escape(warning)}" for warning in report.warnings[:50]
)
if len(report.warnings) > 50:
items += f"\n{len(report.warnings) - 50} more"
warnings_section = _html_section("Data Warnings", f"")
filter_line = ""
if report.days_filter is not None:
filter_line = f"Filter: last {report.days_filter} local calendar days
"
return f"""
Codex Usage Report
Codex Usage Report
{_metric("Sessions", _fmt(report.sessions))}
{_metric("Total tokens", _fmt(report.total_tokens))}
{_metric("Rollout breakdowns", _fmt(report.parsed_rollout_count))}
{_metric("Missing rollouts", _fmt(report.missing_rollout_count))}
{_metric("Warnings", _fmt(report.warnings))}
{rollout_section}
{sections}
{warnings_section}
"""
def write_reports(report: UsageReport, out_dir: Path, output_format: str) -> list[Path]:
out_dir.mkdir(parents=True, exist_ok=True)
written: list[Path] = []
if output_format in {"md", "both"}:
path = out_dir / "codex-usage.md"
path.write_text(render_markdown(report), encoding="utf-8")
written.append(path)
if output_format in {"html", "both"}:
path = out_dir / "codex-usage.html"
path.write_text(render_html(report), encoding="utf-8")
written.append(path)
return written
def _render_bucket_markdown(buckets: list[Bucket]) -> str:
return _markdown_table(
["Label", "Sessions", "Tokens", "Rollout total", "Input", "Cached input", "Output", "Reasoning"],
[
[
bucket.label,
_fmt(bucket.sessions),
_fmt(bucket.tokens),
_fmt(bucket.rollout.total_tokens),
_fmt(bucket.rollout.input_tokens),
_fmt(bucket.rollout.cached_input_tokens),
_fmt(bucket.rollout.output_tokens),
_fmt(bucket.rollout.reasoning_output_tokens),
]
for bucket in buckets
],
)
def _render_bucket_html(buckets: list[Bucket]) -> str:
return _html_table(
["Label", "Sessions", "Tokens", "Rollout total", "Input", "Cached input", "Output", "Reasoning"],
[
[
bucket.label,
_fmt(bucket.sessions),
_fmt(bucket.tokens),
_fmt(bucket.rollout.total_tokens),
_fmt(bucket.rollout.input_tokens),
_fmt(bucket.rollout.cached_input_tokens),
_fmt(bucket.rollout.output_tokens),
_fmt(bucket.rollout.reasoning_output_tokens),
]
for bucket in buckets
],
numeric_columns={1, 2, 3, 4, 5, 6, 7},
)
def _render_top_sessions_markdown(sessions: list[SessionUsage]) -> str:
return _markdown_table(
["Created", "Tokens", "Kind", "Model", "Role", "Cwd", "Session", "Rollout"],
[
[
session.created_at.strftime("%Y-%m-%d %H:%M"),
_fmt(session.tokens),
session.kind,
session.model,
session.role,
_compact_for_report(session.cwd),
session.thread_id,
str(session.rollout_path or ""),
]
for session in sessions
],
)
def _render_top_sessions_html(sessions: list[SessionUsage]) -> str:
return _html_table(
["Created", "Tokens", "Kind", "Model", "Role", "Cwd", "Session", "Rollout"],
[
[
session.created_at.strftime("%Y-%m-%d %H:%M"),
_fmt(session.tokens),
session.kind,
session.model,
session.role,
_compact_for_report(session.cwd),
session.thread_id,
str(session.rollout_path or ""),
]
for session in sessions
],
numeric_columns={1},
)
def _markdown_table(headers: list[str], rows: list[list[str]]) -> str:
if not rows:
return "_No rows._"
escaped_headers = [_escape_md(header) for header in headers]
lines = [
"| " + " | ".join(escaped_headers) + " |",
"| " + " | ".join("---" for _ in headers) + " |",
]
for row in rows:
lines.append("| " + " | ".join(_escape_md(str(cell)) for cell in row) + " |")
return "\n".join(lines)
def _html_table(
headers: list[str], rows: list[list[str]], *, numeric_columns: set[int] | None = None
) -> str:
if not rows:
return 'No rows.
'
numeric_columns = numeric_columns or set()
head = "".join(f"{html.escape(header)} | " for header in headers)
body_rows = []
for row in rows:
cells = []
for index, cell in enumerate(row):
klass = ' class="num"' if index in numeric_columns else ""
cells.append(f"{html.escape(str(cell))} | ")
body_rows.append("" + "".join(cells) + "
")
return (
''
+ head
+ "
"
+ "".join(body_rows)
+ "
"
)
def _html_section(title: str, content: str) -> str:
return f"{html.escape(title)}
{content}"
def _metric(label: str, value: str) -> str:
return f'{html.escape(label)}{html.escape(value)}
'
def _fmt(value: int | list[object]) -> str:
if isinstance(value, list):
value = len(value)
return f"{value:,}"
def _escape_md(value: str) -> str:
return value.replace("\\", "\\\\").replace("|", "\\|").replace("\n", " ")
def _compact_for_report(value: str) -> str:
if not value:
return "unknown"
try:
path = Path(value).expanduser()
home = Path.home()
if path.is_absolute():
relative = path.relative_to(home)
return "~" if str(relative) == "." else "~/" + str(relative)
return str(path)
except ValueError:
return value