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

    Generated: {html.escape(report.generated_at.isoformat(timespec="seconds"))}
    Timezone: {html.escape(report.timezone_name)}
    State DB: {html.escape(str(report.state_path))}
    Sessions dir: {html.escape(str(report.sessions_dir))}
    Date range: {html.escape(report.date_range_label)}
    {filter_line}
    {_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