Refactor reporting functions to cut cognitive complexity (S3776, batch 1/4)

CHANGELOG.md

@@ -25,6 +25,7 @@ This project follows the structure from Keep a Changelog and intends to use Sema
 
 ### Changed
 
+- **Refactored reporting functions** (export CSV, rate limits, OS breakdown, report data fetching, base-cost and summary rendering) into smaller helpers to cut cognitive complexity and clear SonarCloud S3776 high-severity issues (batch 1/4).
 - **Refactored repeated string literals** (GUI selectors, HTML fragments, the `reports.` TOML prefix, the Git LFS label, and the workflows pathspec) into module-level constants to clear SonarCloud high-severity S1192/S7688 issues.
 - **GitHub Actions workflow now defaults to `--email-format html`:** The committed `email-report.yml` and its template were updated from `--email-format text` to `--email-format html`. Existing users who re-run setup (option 5 / wizard) will have their workflow re-rendered with this default. To keep plain-text output, set `email_format = "text"` in `.github-usage/config.toml` before re-running setup, or select **Text** in the wizard.
 - **Terminology:** the interactive CLI/TUI usage report is now called the **local full report** in user-facing copy (README, `start.sh`, CLI help, TUI). Internal `legacy_*` module names are unchanged for now (tracked in `TO_DO.md`).

src/github_usage/export_csv.py

@@ -70,10 +70,28 @@ def _write_sections(
     if include_forecast:
         _write_forecast_section(writer, data, premium_requests_limit=premium_requests_limit)
 
+    _write_warnings_section(writer, data)
+    _write_actions_section(writer, data)
+    _write_storage_summary_section(writer, data)
+    _write_copilot_section(writer, data)
+    _write_git_lfs_section(writer, data)
+    _write_monthly_costs_section(writer, data)
+    _write_repo_consumers_sections(writer, data)
+    _write_artifact_storage_section(writer, data)
+    _write_storage_analysis_section(writer, data)
+    _write_release_assets_section(writer, data)
+    _write_key_insights_section(writer, data)
+    _write_unavailable_data_section(writer, data)
+    _write_sources_section(writer, data)
+
+
+def _write_warnings_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Warnings")
     for warning in data.get("warnings") or []:
         writer.writerow([warning])
 
+
+def _write_actions_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Actions Usage")
     actions = _coerce_section(data.get("actions"), {})
     for key, value in actions.items():
@@ -97,12 +115,16 @@ def _write_sections(
         for row in vis_rows:
             writer.writerow(row)
 
+
+def _write_storage_summary_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     storage_summary = _coerce_section(data.get("storage_summary"), {})
     if storage_summary:
         _write_section_header(writer, "Storage Summary")
         for key, value in storage_summary.items():
             writer.writerow([key, value])
 
+
+def _write_copilot_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Copilot Usage")
     copilot = _coerce_section(data.get("copilot"), {})
     for key, value in copilot.items():
@@ -111,11 +133,15 @@ def _write_sections(
         else:
             writer.writerow([key, value])
 
+
+def _write_git_lfs_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Git LFS")
     git_lfs = _coerce_section(data.get("git_lfs"), {})
     for key, value in git_lfs.items():
         writer.writerow([key, value])
 
+
+def _write_monthly_costs_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Monthly Costs")
     costs = _coerce_section(data.get("monthly_costs"), {})
     for category, amounts in costs.items():
@@ -126,44 +152,49 @@ def _write_sections(
         else:
             writer.writerow([category, amounts])
 
-    _write_section_header(writer, "Top Repos by Minutes")
+
+def _write_consumer_row(writer, entry: dict) -> None:  # type: ignore[type-arg]
+    """Write a single repository consumer row (name, visibility, minutes, cost, storage) to the CSV."""
+    writer.writerow(
+        [
+            entry.get("repo", ""),
+            repo_visibility(entry),
+            entry.get("minutes", ""),
+            entry.get("gross", ""),
+            entry.get("storage_avg_mb", ""),
+        ]
+    )
+
+
+def _write_repo_consumers_sections(writer, data: dict) -> None:  # type: ignore[type-arg]
     consumers = _coerce_section(data.get("repo_consumers"), {})
+    _write_section_header(writer, "Top Repos by Minutes")
     for entry in consumers.get("by_minutes") or []:
-        writer.writerow(
-            [
-                entry.get("repo", ""),
-                repo_visibility(entry),
-                entry.get("minutes", ""),
-                entry.get("gross", ""),
-                entry.get("storage_avg_mb", ""),
-            ]
-        )
+        _write_consumer_row(writer, entry)
 
     _write_section_header(writer, "Top Repos by Cost")
     for entry in consumers.get("by_cost") or []:
-        writer.writerow(
-            [
-                entry.get("repo", ""),
-                repo_visibility(entry),
-                entry.get("minutes", ""),
-                entry.get("gross", ""),
-                entry.get("storage_avg_mb", ""),
-            ]
-        )
+        _write_consumer_row(writer, entry)
+
 
+def _write_artifact_storage_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Artifact Storage")
     artifacts = _coerce_section(data.get("artifact_storage"), {})
     for entry in artifacts.get("top_repos") or []:
         writer.writerow(
             [entry.get("repo", ""), repo_visibility(entry), entry.get("artifact_bytes", "")]
         )
 
+
+def _write_storage_analysis_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     analysis_rows = storage_analysis_export_rows(data.get("storage_analysis"))
     if analysis_rows:
         _write_section_header(writer, "Storage Analysis")
         for row in analysis_rows:
             writer.writerow(row)
 
+
+def _write_release_assets_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Release Assets")
     releases = _coerce_section(data.get("release_assets"), {})
     for entry in releases.get("top_repos") or []:
@@ -175,14 +206,20 @@ def _write_sections(
             ]
         )
 
+
+def _write_key_insights_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Key Insights")
     for insight in data.get("insights") or []:
         writer.writerow([insight])
 
+
+def _write_unavailable_data_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     _write_section_header(writer, "Unavailable Data")
     for error_key, error_msg in (data.get("errors") or {}).items():
         writer.writerow([error_key, error_msg])
 
+
+def _write_sources_section(writer, data: dict) -> None:  # type: ignore[type-arg]
     sources = data.get("sources")
     source_rows = sources_rows(sources if isinstance(sources, dict) else None)
     if source_rows:

src/github_usage/export_visibility.py

@@ -140,7 +140,7 @@ def storage_analysis_export_rows(storage_analysis: dict | None) -> list[list]:
 
 
 def per_visibility_sku_rows(actions: dict | None) -> list[list]:
-    """SKU × visibility table from ``actions['skus']`` when present."""
+    """SKU x visibility table from ``actions['skus']`` when present."""
     if not actions:
         return []
     skus = actions.get("skus") or {}

src/github_usage/report_account.py

@@ -13,14 +13,8 @@ def fetch_rate_limits(api) -> dict:
     return data if isinstance(data, dict) else {}
 
 
-def render_rate_limits(data: dict) -> None:
-    """Print rate limits from a pre-fetched ``/rate_limit`` response."""
-    print_sep("API Rate Limit")
-    resources = data.get("resources", {})
-    if not isinstance(resources, dict):
-        resources = {}
-
-    # Standard limits
+def _print_standard_rate_limits(resources: dict) -> None:
+    """Print the core / GraphQL / search / code-scanning rate-limit rows."""
     print()
     for name, key in [
         ("Core API", "core"),
@@ -37,7 +31,6 @@ def render_rate_limits(data: dict) -> None:
         lim = r.get("limit")
         if lim is None:
             lim = "?"
-        used = r.get("used", 0)
         reset_ts = r.get("reset", 0)
         reset_str = ""
         if reset_ts:
@@ -46,7 +39,9 @@ def render_rate_limits(data: dict) -> None:
             )
         print(f"  {name:<25} {rem:>6} / {lim:<6} remaining{reset_str}")
 
-    # Premium / high-tier
+
+def _print_premium_rate_limits(resources: dict) -> None:
+    """Print the high-tier (limit > 5000) rate-limit rows."""
     print()
     print("  Premium API tiers:")
     for name, res in resources.items():
@@ -64,6 +59,16 @@ def render_rate_limits(data: dict) -> None:
     print()
 
 
+def render_rate_limits(data: dict) -> None:
+    """Print rate limits from a pre-fetched ``/rate_limit`` response."""
+    print_sep("API Rate Limit")
+    resources = data.get("resources", {})
+    if not isinstance(resources, dict):
+        resources = {}
+    _print_standard_rate_limits(resources)
+    _print_premium_rate_limits(resources)
+
+
 def fetch_account_info(api) -> dict:
     """Return account metadata from ``GET /user``."""
     user = api.request("GET", "/user")

src/github_usage/report_actions.py

@@ -90,6 +90,26 @@ def show_actions_top_consumers(repo_data, visibility_by_repo=None):
     print()
 
 
+def _print_repo_os_breakdown(owner, name, os_millis, total_os):
+    """Print one repo's per-OS minutes and accumulate its millis into ``total_os``."""
+    print(f"  {owner}/{name}:")
+    for os_name in ["UBUNTU", "WINDOWS", "MACOS"]:
+        mins = os_millis[os_name] / 60000
+        total_os[os_name] += os_millis[os_name]
+        if mins > 0:
+            print(f"    {os_name:<10} {mins:>8.1f} min")
+    print()
+
+
+def _print_os_totals(total_os):
+    """Print the aggregated per-OS minute totals across all repos."""
+    print("  TOTAL:")
+    for os_name in ["UBUNTU", "WINDOWS", "MACOS"]:
+        mins = total_os[os_name] / 60000
+        if mins > 0:
+            print(f"    {os_name:<10} {mins:>8.1f} min")
+
+
 def show_actions_os_breakdown(api, repos):
     """Show Ubuntu/Windows/macOS breakdown for top repos."""
     print_sep("Actions Compute by OS (from workflow runs)")
@@ -104,19 +124,9 @@ def show_actions_os_breakdown(api, repos):
         minutes, os_millis, _ = get_actions_from_runs(api, owner, name)
         if minutes > 0:
             found = True
-            print(f"  {owner}/{name}:")
-            for os_name in ["UBUNTU", "WINDOWS", "MACOS"]:
-                mins = os_millis[os_name] / 60000
-                total_os[os_name] += os_millis[os_name]
-                if mins > 0:
-                    print(f"    {os_name:<10} {mins:>8.1f} min")
-            print()
+            _print_repo_os_breakdown(owner, name, os_millis, total_os)
     if found:
-        print("  TOTAL:")
-        for os_name in ["UBUNTU", "WINDOWS", "MACOS"]:
-            mins = total_os[os_name] / 60000
-            if mins > 0:
-                print(f"    {os_name:<10} {mins:>8.1f} min")
+        _print_os_totals(total_os)
     else:
         print("  No detailed OS breakdown available from workflow runs API.")
         print("  (Use the Actions Summary above for total minutes by OS type)")

src/github_usage/report_data.py

@@ -237,6 +237,14 @@ def _rate_limit(api: GitHubAPIClient) -> tuple[int | None, int | None]:
     return core.get("limit"), core.get("remaining")
 
 
+def _try_section(report: dict, errors: dict, key: str, getter) -> None:
+    """Store ``getter()`` under ``report[key]``, or record its RuntimeError in ``errors``."""
+    try:
+        report[key] = getter()
+    except RuntimeError as exc:
+        errors[key] = str(exc)
+
+
 def _fetch_sections(
     api: GitHubAPIClient,
     username: str,
@@ -260,10 +268,7 @@ def _fetch_sections(
         ("git_lfs", include_lfs, lambda: get_gitlfs_usage(api, username)),
     ]:
         if enabled:
-            try:
-                report[key] = getter()
-            except RuntimeError as exc:
-                errors[key] = str(exc)
+            _try_section(report, errors, key, getter)
 
     try:
         report["monthly_costs"] = get_monthly_costs(api, username)
@@ -277,28 +282,36 @@ def _fetch_sections(
         }
 
     if include_consumers:
-        try:
-            report["repo_consumers"] = get_repo_consumers(api, repos, max_repos=max_repos)
-        except RuntimeError as exc:
-            errors["repo_consumers"] = str(exc)
+        _try_section(
+            report,
+            errors,
+            "repo_consumers",
+            lambda: get_repo_consumers(api, repos, max_repos=max_repos),
+        )
     if include_artifact_storage:
-        try:
-            report["artifact_storage"] = get_artifact_storage_details(api, repos, max_repos)
-        except RuntimeError as exc:
-            errors["artifact_storage"] = str(exc)
+        _try_section(
+            report,
+            errors,
+            "artifact_storage",
+            lambda: get_artifact_storage_details(api, repos, max_repos),
+        )
     if include_release_assets:
-        try:
-            report["release_assets"] = get_release_asset_details(api, repos, max_repos)
-        except RuntimeError as exc:
-            errors["release_assets"] = str(exc)
+        _try_section(
+            report,
+            errors,
+            "release_assets",
+            lambda: get_release_asset_details(api, repos, max_repos),
+        )
 
     if include_consumers and report.get("repo_consumers"):
-        try:
-            report["workflow_breakdown"] = workflow_breakdown_for_top_private(
+        _try_section(
+            report,
+            errors,
+            "workflow_breakdown",
+            lambda: workflow_breakdown_for_top_private(
                 api, report["repo_consumers"], runs_cache=runs_cache
-            )
-        except RuntimeError as exc:
-            errors["workflow_breakdown"] = str(exc)
+            ),
+        )
 
 
 def build_report_data(