Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 34 additions & 41 deletions src/excel_dashboard_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,9 +239,7 @@ def build_dashboard_operator_rows(
),
(
"Freshness Detail",
operator_context[
"follow_through_recovery_freshness_hotspot_summary"
],
operator_context["follow_through_recovery_freshness_hotspot_summary"],
),
(
"Rebuild Hotspot",
Expand Down Expand Up @@ -281,51 +279,39 @@ def build_dashboard_operator_rows(
),
(
"Fragile Reacquisition Confidence Hotspot",
operator_context[
"follow_through_fragile_reacquisition_confidence_hotspot"
],
operator_context["follow_through_fragile_reacquisition_confidence_hotspot"],
),
(
"Reacquisition Softening Hotspot",
operator_context["follow_through_reacquisition_softening_hotspot"],
),
(
"Revalidation Needed Hotspot",
operator_context[
"follow_through_reacquisition_revalidation_hotspot"
],
operator_context["follow_through_reacquisition_revalidation_hotspot"],
),
(
"Retired Confidence Hotspot",
operator_context[
"follow_through_reacquisition_retired_confidence_hotspot"
],
operator_context["follow_through_reacquisition_retired_confidence_hotspot"],
),
(
"Under Revalidation Hotspot",
operator_context["follow_through_under_revalidation_hotspot"],
),
(
"Rebuilding Restored Confidence Hotspot",
operator_context[
"follow_through_rebuilding_restored_confidence_hotspot"
],
operator_context["follow_through_rebuilding_restored_confidence_hotspot"],
),
(
"Re-Earning Confidence Hotspot",
operator_context["follow_through_reearning_confidence_hotspot"],
),
(
"Just Re-Earned Confidence Hotspot",
operator_context[
"follow_through_just_reearned_confidence_hotspot"
],
operator_context["follow_through_just_reearned_confidence_hotspot"],
),
(
"Holding Re-Earned Confidence Hotspot",
operator_context[
"follow_through_holding_reearned_confidence_hotspot"
],
operator_context["follow_through_holding_reearned_confidence_hotspot"],
),
]
)
Expand Down Expand Up @@ -435,9 +421,7 @@ def build_dashboard_operator_rows(
),
(
"Reset Re-entry Rebuild Re-Entry Restore Re-Re-Re-Restore Churn Controls",
operator_context[
"reset_reentry_rebuild_reentry_restore_rerererestore_churn"
],
operator_context["reset_reentry_rebuild_reentry_restore_rerererestore_churn"],
),
(
"Closure Forecast Summary",
Expand Down Expand Up @@ -505,9 +489,7 @@ def build_dashboard_top_attention_rows(

def build_dashboard_top_opportunity_rows(audits: list[dict[str, Any]]) -> list[list[Any]]:
rows: list[list[Any]] = []
audits_sorted = sorted(
audits, key=lambda audit: audit.get("overall_score", 0), reverse=True
)
audits_sorted = sorted(audits, key=lambda audit: audit.get("overall_score", 0), reverse=True)
for audit in audits_sorted:
action = (audit.get("action_candidates") or [{}])[0]
hotspots = audit.get("hotspots") or []
Expand Down Expand Up @@ -545,8 +527,9 @@ def build_dashboard_kpi_specs(
grade: str,
average_score: float,
tiers: dict[str, Any],
risk_lookup: dict[str, str] | None = None,
) -> list[tuple[str, Any, str | None, str | None]]:
return [
specs: list[tuple[str, Any, str | None, str | None]] = [
("Portfolio Grade", grade, None, None),
("Avg Score", f"{average_score:.2f}", None, None),
("Shipped", tiers.get("shipped", 0), "166534", "Tier Breakdown"),
Expand All @@ -559,6 +542,10 @@ def build_dashboard_kpi_specs(
None,
),
]
if risk_lookup:
elevated = sum(1 for tier in risk_lookup.values() if tier == "elevated")
specs.append(("Elevated Risk", elevated, "DC2626", "Risk Summary"))
return specs


def build_dashboard_sidebar_rows(
Expand Down Expand Up @@ -666,22 +653,16 @@ def build_dashboard_portfolio_trend_sparkline(


def build_dashboard_dna_tiers(audits: list[dict[str, Any]]) -> list[str]:
audits_sorted = sorted(
audits, key=lambda audit: audit.get("overall_score", 0), reverse=True
)
return [
audit.get("completeness_tier", "abandoned") for audit in audits_sorted[:24]
]
audits_sorted = sorted(audits, key=lambda audit: audit.get("overall_score", 0), reverse=True)
return [audit.get("completeness_tier", "abandoned") for audit in audits_sorted[:24]]


def build_dashboard_highlight_rows(data: dict[str, Any]) -> dict[str, list[str]]:
return {
"best_work": list(
data.get("best_work") or data.get("summary", {}).get("highest_scored", [])
)[:5],
"needs_attention": list(
data.get("summary", {}).get("lowest_scored", [])
)[:5],
"needs_attention": list(data.get("summary", {}).get("lowest_scored", []))[:5],
}


Expand Down Expand Up @@ -734,7 +715,9 @@ def write_dashboard_visual_sections(
ws.column_dimensions[get_column_letter(column_index)].hidden = True
for column_letter, width in dashboard_preferred_widths().items():
ws.column_dimensions[column_letter].width = width
_write_dashboard_language_chart(ws, build_dashboard_language_rows(data.get("language_distribution", {})))
_write_dashboard_language_chart(
ws, build_dashboard_language_rows(data.get("language_distribution", {}))
)
_write_dashboard_scatter_section(ws, data.get("audits", []), section_font, subheader_font)


Expand Down Expand Up @@ -861,7 +844,9 @@ def _write_dashboard_language_chart(ws, language_rows: list[tuple[str, Any]]) ->
ws.add_chart(lang_bar, "A35")


def _write_dashboard_scatter_section(ws, audits: list[dict[str, Any]], section_font, subheader_font) -> None:
def _write_dashboard_scatter_section(
ws, audits: list[dict[str, Any]], section_font, subheader_font
) -> None:
if len(audits) < 2:
return

Expand Down Expand Up @@ -931,10 +916,14 @@ def _write_dashboard_scatter_section(ws, audits: list[dict[str, Any]], section_f
chart.width = 10.5
chart.height = 8
ws.add_chart(chart, "J34")
_write_quadrant_table(ws, audits, legend_row=52, section_font=section_font, subheader_font=subheader_font)
_write_quadrant_table(
ws, audits, legend_row=52, section_font=section_font, subheader_font=subheader_font
)


def _write_quadrant_table(ws, audits: list[dict[str, Any]], *, legend_row: int, section_font, subheader_font) -> None:
def _write_quadrant_table(
ws, audits: list[dict[str, Any]], *, legend_row: int, section_font, subheader_font
) -> None:
buckets: list[list[str]] = [[], [], [], []]
for audit in audits:
x = audit.get("overall_score", 0)
Expand Down Expand Up @@ -970,6 +959,7 @@ def build_dashboard_sheet(
score_history: dict[str, list[float]] | None = None,
*,
excel_mode: str,
risk_lookup: dict[str, str] | None = None,
get_or_create_sheet,
clear_worksheet,
configure_sheet_view,
Expand Down Expand Up @@ -1052,6 +1042,7 @@ def build_dashboard_sheet(
grade=grade,
average_score=data["average_score"],
tiers=tiers,
risk_lookup=risk_lookup,
)
):
link = f"#{sheet_location(target_sheet)}" if target_sheet else None
Expand Down Expand Up @@ -1191,6 +1182,7 @@ def build_dashboard_workbook_sheet(
score_history: dict[str, list[float]] | None = None,
*,
excel_mode: str,
risk_lookup: dict[str, str] | None = None,
get_or_create_sheet,
clear_worksheet,
configure_sheet_view,
Expand Down Expand Up @@ -1248,6 +1240,7 @@ def build_dashboard_workbook_sheet(
diff_data,
score_history,
excel_mode=excel_mode,
risk_lookup=risk_lookup,
get_or_create_sheet=get_or_create_sheet,
clear_worksheet=clear_worksheet,
configure_sheet_view=configure_sheet_view,
Expand Down
4 changes: 4 additions & 0 deletions src/excel_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -611,13 +611,15 @@ def _build_dashboard(
score_history: dict[str, list[float]] | None = None,
*,
excel_mode: str = "standard",
risk_lookup: dict[str, str] | None = None,
) -> None:
_build_dashboard_workbook_sheet(
wb,
data,
diff_data,
score_history,
excel_mode=excel_mode,
risk_lookup=risk_lookup,
get_or_create_sheet=_get_or_create_sheet,
clear_worksheet=_clear_worksheet,
configure_sheet_view=_configure_sheet_view,
Expand Down Expand Up @@ -1173,6 +1175,7 @@ def _build_hidden_data_sheets(
trend_data: list[dict] | None = None,
score_history: dict[str, list[float]] | None = None,
diff_data: dict | None = None,
risk_lookup: dict[str, str] | None = None,
) -> None:
_build_hidden_data_sheets_helper(
wb,
Expand All @@ -1189,6 +1192,7 @@ def _build_hidden_data_sheets(
build_core_hidden_rows=_build_core_hidden_rows,
trend_history_window=TREND_HISTORY_WINDOW,
tier_order=TIER_ORDER,
risk_lookup=risk_lookup,
)


Expand Down
12 changes: 10 additions & 2 deletions src/excel_export_runner_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,11 @@ def build_excel_workbook(
) -> None:
run_workbook_build_steps(
[
(build_dashboard, (wb, data, diff_data, score_history), {"excel_mode": excel_mode}),
(
build_dashboard,
(wb, data, diff_data, score_history),
{"excel_mode": excel_mode, "risk_lookup": risk_lookup},
),
(build_all_repos, (wb, data, score_history), {"risk_lookup": risk_lookup}),
(
build_portfolio_explorer,
Expand Down Expand Up @@ -145,7 +149,11 @@ def build_excel_workbook(
(build_dependency_graph, (wb, data), {}),
(build_score_explainer, (wb,), {}),
(build_action_items, (wb, data), {}),
(build_hidden_data_sheets, (wb, data, trend_data, score_history, diff_data), {}),
(
build_hidden_data_sheets,
(wb, data, trend_data, score_history, diff_data),
{"risk_lookup": risk_lookup},
),
(
build_navigation,
(wb, data),
Expand Down
8 changes: 7 additions & 1 deletion src/excel_hidden_data_content_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,13 @@ def build_hidden_data_payload(
build_core_hidden_rows,
trend_history_window: int,
tier_order: list[str],
risk_lookup: dict[str, str] | None = None,
) -> dict[str, Any]:
repo_detail_rows, repo_dimension_rollup_rows, repo_history_rollup_rows = build_repo_detail_rows(
data, score_history, ghas_lookup=data.get("ghas_alerts") or None
data,
score_history,
ghas_lookup=data.get("ghas_alerts") or None,
risk_lookup=risk_lookup,
)
run_change_rollup_rows, run_change_repo_rows = build_run_change_rows(data, diff_data)
operator_queue_rows, operator_repo_rollups, material_rollups = build_workbook_rollups(data)
Expand Down Expand Up @@ -76,6 +80,7 @@ def build_hidden_data_sheets(
build_core_hidden_rows,
trend_history_window: int,
tier_order: list[str],
risk_lookup: dict[str, str] | None = None,
) -> None:
payload = build_hidden_data_payload_fn(
data=data,
Expand All @@ -89,6 +94,7 @@ def build_hidden_data_sheets(
build_core_hidden_rows=build_core_hidden_rows,
trend_history_window=trend_history_window,
tier_order=tier_order,
risk_lookup=risk_lookup,
)
write_hidden_data_tables(
wb,
Expand Down
2 changes: 2 additions & 0 deletions src/excel_repo_data_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ def repo_detail_rows(
data: dict,
score_history: dict[str, list[float]] | None,
ghas_lookup: dict | None = None,
risk_lookup: dict[str, str] | None = None,
) -> tuple[list[list[object]], list[list[object]], list[list[object]]]:
memberships = collection_memberships(data)
history = extend_score_history_with_current(data, score_history)
Expand Down Expand Up @@ -281,6 +282,7 @@ def repo_detail_rows(
if len(implementation_hotspot_lines) > 2
else "No third implementation hotspot is currently surfaced.",
*_arc_f_detail_cols(audit, ghas_lookup),
(risk_lookup or {}).get(repo_name, "") or "—",
]
)

Expand Down
15 changes: 14 additions & 1 deletion src/excel_repo_detail_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -572,5 +572,18 @@ def build_repo_detail_sheet(
"left",
)

# Risk Posture section (column 79 in Data_RepoDetail, from portfolio truth)
ws["A43"] = "Risk Posture"
ws["A43"].font = section_font
ws.cell(row=44, column=1, value="Risk Tier").font = subheader_font
style_data_cell(
ws.cell(
row=44,
column=2,
value=repo_detail_lookup_formula(79, "—", allow_blank=True),
),
"left",
)

nav_row = populate_repo_detail_handoff_section(ws, style_data_cell, subheader_font)
auto_width(ws, 8, nav_row)
auto_width(ws, 8, max(nav_row, 44))
42 changes: 29 additions & 13 deletions src/report_enrichment.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,8 +154,14 @@ def _metadata(audit: Any) -> dict[str, Any]:
return metadata


def _extract_risk_posture(output_dir: Path | None) -> dict[str, Any]:
"""Load portfolio-truth-latest.json and extract risk tier summary. Returns {} if unavailable."""
def build_risk_lookup(output_dir: Path | None) -> dict[str, dict[str, str]]:
"""Canonical per-repo risk reader: display_name -> {risk_tier, risk_summary}.

Single source of truth for risk data loaded from portfolio-truth-latest.json.
Render surfaces that need a per-repo risk tier consume this directly, and
_extract_risk_posture derives its aggregate counts from it. Returns {} when the
truth snapshot is missing, unreadable, or has no named projects.
"""
if not output_dir:
return {}
truth_path = output_dir / "portfolio-truth-latest.json"
Expand All @@ -165,21 +171,31 @@ def _extract_risk_posture(output_dir: Path | None) -> dict[str, Any]:
truth = json.loads(truth_path.read_text())
except Exception:
return {}
projects = truth.get("projects") or []
lookup: dict[str, dict[str, str]] = {}
for project in truth.get("projects") or []:
name = str((project.get("identity") or {}).get("display_name") or "")
if not name:
continue
risk = project.get("risk") or {}
lookup[name] = {
"risk_tier": str(risk.get("risk_tier") or "baseline"),
"risk_summary": str(risk.get("risk_summary") or ""),
}
return lookup


def _extract_risk_posture(output_dir: Path | None) -> dict[str, Any]:
"""Aggregate risk tier summary, derived from build_risk_lookup. Returns {} if unavailable."""
lookup = build_risk_lookup(output_dir)
if not lookup:
return {}
tier_counts: dict[str, int] = {}
top_elevated: list[dict[str, Any]] = []
for project in projects:
risk = project.get("risk") or {}
tier = str(risk.get("risk_tier") or "baseline")
for name, entry in lookup.items():
tier = entry["risk_tier"]
tier_counts[tier] = tier_counts.get(tier, 0) + 1
if tier == "elevated":
identity = project.get("identity") or {}
top_elevated.append(
{
"repo": str(identity.get("display_name") or ""),
"risk_summary": str(risk.get("risk_summary") or ""),
}
)
top_elevated.append({"repo": name, "risk_summary": entry["risk_summary"]})
return {
"elevated_count": tier_counts.get("elevated", 0),
"moderate_count": tier_counts.get("moderate", 0),
Expand Down
Loading