| 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"<li><code>{html.escape(warning)}</code></li>" for warning in report.warnings[:50] |
| ) |
| if len(report.warnings) > 50: |
| items += f"\n<li>{len(report.warnings) - 50} more</li>" |
| warnings_section = _html_section("Data Warnings", f"<ul>{items}</ul>") |
|
|
| filter_line = "" |
| if report.days_filter is not None: |
| filter_line = f"<p>Filter: last {report.days_filter} local calendar days</p>" |
|
|
| return f"""<!doctype html> |
| <html lang="en"> |
| <head> |
| <meta charset="utf-8"> |
| <meta name="viewport" content="width=device-width, initial-scale=1"> |
| <title>Codex Usage Report</title> |
| <style> |
| :root {{ |
| color-scheme: light dark; |
| --bg: #f7f7f4; |
| --panel: #ffffff; |
| --text: #1f2528; |
| --muted: #667078; |
| --border: #d9ddd8; |
| --accent: #176b5d; |
| --row: #f1f4f2; |
| }} |
| @media (prefers-color-scheme: dark) {{ |
| :root {{ |
| --bg: #151716; |
| --panel: #1d211f; |
| --text: #eef1ee; |
| --muted: #a6aea9; |
| --border: #373d39; |
| --accent: #74c7b8; |
| --row: #242a27; |
| }} |
| }} |
| body {{ |
| margin: 0; |
| background: var(--bg); |
| color: var(--text); |
| font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; |
| line-height: 1.45; |
| }} |
| main {{ |
| max-width: 1180px; |
| margin: 0 auto; |
| padding: 32px 20px 56px; |
| }} |
| h1 {{ |
| margin: 0 0 10px; |
| font-size: 2rem; |
| letter-spacing: 0; |
| }} |
| h2 {{ |
| margin: 32px 0 12px; |
| font-size: 1.1rem; |
| letter-spacing: 0; |
| }} |
| .meta {{ |
| color: var(--muted); |
| display: grid; |
| gap: 4px; |
| margin-bottom: 24px; |
| }} |
| .summary {{ |
| display: grid; |
| grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); |
| gap: 12px; |
| }} |
| .metric {{ |
| background: var(--panel); |
| border: 1px solid var(--border); |
| border-radius: 8px; |
| padding: 14px 16px; |
| }} |
| .metric span {{ |
| display: block; |
| color: var(--muted); |
| font-size: .85rem; |
| }} |
| .metric strong {{ |
| display: block; |
| margin-top: 4px; |
| font-size: 1.25rem; |
| }} |
| .table-wrap {{ |
| overflow-x: auto; |
| border: 1px solid var(--border); |
| background: var(--panel); |
| border-radius: 8px; |
| }} |
| table {{ |
| border-collapse: collapse; |
| width: 100%; |
| min-width: 640px; |
| }} |
| th, td {{ |
| padding: 9px 11px; |
| border-bottom: 1px solid var(--border); |
| text-align: left; |
| vertical-align: top; |
| white-space: nowrap; |
| }} |
| th {{ |
| color: var(--muted); |
| font-weight: 650; |
| font-size: .82rem; |
| background: var(--row); |
| }} |
| tr:last-child td {{ |
| border-bottom: 0; |
| }} |
| td.num {{ |
| text-align: right; |
| font-variant-numeric: tabular-nums; |
| }} |
| code {{ |
| font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; |
| font-size: .9em; |
| }} |
| .empty {{ |
| color: var(--muted); |
| margin: 0; |
| }} |
| </style> |
| </head> |
| <body> |
| <main> |
| <h1>Codex Usage Report</h1> |
| <div class="meta"> |
| <div>Generated: {html.escape(report.generated_at.isoformat(timespec="seconds"))}</div> |
| <div>Timezone: <code>{html.escape(report.timezone_name)}</code></div> |
| <div>State DB: <code>{html.escape(str(report.state_path))}</code></div> |
| <div>Sessions dir: <code>{html.escape(str(report.sessions_dir))}</code></div> |
| <div>Date range: {html.escape(report.date_range_label)}</div> |
| {filter_line} |
| </div> |
| <section class="summary"> |
| {_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))} |
| </section> |
| {rollout_section} |
| {sections} |
| {warnings_section} |
| </main> |
| </body> |
| </html> |
| """ |
|
|
|
|
| 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 '<p class="empty">No rows.</p>' |
| numeric_columns = numeric_columns or set() |
| head = "".join(f"<th>{html.escape(header)}</th>" 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"<td{klass}>{html.escape(str(cell))}</td>") |
| body_rows.append("<tr>" + "".join(cells) + "</tr>") |
| return ( |
| '<div class="table-wrap"><table><thead><tr>' |
| + head |
| + "</tr></thead><tbody>" |
| + "".join(body_rows) |
| + "</tbody></table></div>" |
| ) |
|
|
|
|
| def _html_section(title: str, content: str) -> str: |
| return f"<section><h2>{html.escape(title)}</h2>{content}</section>" |
|
|
|
|
| def _metric(label: str, value: str) -> str: |
| return f'<div class="metric"><span>{html.escape(label)}</span><strong>{html.escape(value)}</strong></div>' |
|
|
|
|
| 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 |
|
|
