render_weekly_report.py

#!/usr/bin/env python3
"""Render a compact weekly GitHub issue body from github-security-agent latest.json."""

from __future__ import annotations

import argparse
import json
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlparse


DEFAULT_HEADING = "Weekly Security Report"

SAFE_REASON_CODES = {
    "superseded",
    "advisory_withdrawn",
    "dependency_removed",
    "unsupported_alert_class",
    "unsupported_ecosystem",
    "verification_unavailable",
    "verification_failed",
    "checks_pending",
    "checks_red",
    "branch_protection_block",
    "manifest_change_required",
    "no_patch_available",
    "clone_missing",
    "rate_limited",
    "env_mismatch",
    "auth_insufficient",
    "registry_auth_missing",
    "lock_contended",
    "policy_blocked",
    "unsupported_rule",
    "code_scanning_disabled",
    "secret_scanning_disabled",
    "no_analysis_found",
    "history_rewrite_required",
    "manual_only",
    "manual_only_repository",
    "report_only",
}


@dataclass(frozen=True)
class SecurityOverview:
    dependabot: int = 0
    code_scanning: int = 0
    secret_scanning: int = 0

    @property
    def total(self) -> int:
        return self.dependabot + self.code_scanning + self.secret_scanning


def load_latest_json(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data, dict):
        raise ValueError("latest.json must contain a JSON object")
    return data


def load_security_overview_json(path: str | Path) -> SecurityOverview:
    with Path(path).open("r", encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data, dict):
        raise ValueError("security overview JSON must contain an object")
    return security_overview_from_summary(data)


def render_weekly_report(
    summary: dict[str, Any],
    heading: str = DEFAULT_HEADING,
    security_overview: SecurityOverview | None = None,
) -> str:
    units = list(_iter_units(summary))
    repo_counts = summary.get("repo_counts") or {}
    active_count = _first_int(repo_counts, "active", "active_repos", "active_repositories")
    manual_count = _first_int(repo_counts, "manual_only", "manual_only_repos", "manual_only_repositories")
    overview = security_overview or security_overview_from_summary(summary)

    dependabot = _counts_for(units, "dependabot")
    code_scanning = _counts_for(units, "code_scanning")
    secret_scanning = _counts_for(units, "secret_scanning")
    manual_by_class = _manual_actions_by_class(units)

    lines = [
        f"## {heading}",
        "",
    ]
    lines.extend(_run_summary_lines(summary, units, overview))
    lines.extend(
        [
            "",
            "By alert class:",
        ]
    )
    lines.extend(
        [
        (
            "Dependabot: "
            f"{dependabot['merged']} merged, "
            f"{dependabot['opened_pr']} PR, "
            f"{dependabot['blocked']} manual review"
        ),
        (
            "Code scanning: "
            f"{code_scanning['merged']} fixed, "
            f"{code_scanning['opened_pr']} PR, "
            f"{code_scanning['manual']} manual"
        ),
        (
            "Secret scanning: "
            f"{secret_scanning['opened_pr']} cleanup PR, "
            f"{secret_scanning['manual']} manual"
        ),
        "",
        ]
    )

    patched_lines = _patched_by_automation_lines(summary, units)
    if patched_lines:
        lines.extend(patched_lines)
        lines.append("")

    if manual_by_class:
        lines.append("Manual review required:")
        for alert_class in ("dependabot", "code_scanning", "secret_scanning"):
            count = manual_by_class.get(alert_class, 0)
            if count:
                lines.append(f"- {_class_label(alert_class)}: {count}")
        for alert_class, count in sorted(manual_by_class.items()):
            if alert_class not in {"dependabot", "code_scanning", "secret_scanning"}:
                lines.append(f"- {_class_label(alert_class)}: {count}")
        lines.extend(_manual_review_detail_lines(units))
        lines.append("")
    else:
        lines.append(f"manual repos: {manual_count} checked, no current reportable alerts")
        lines.append("")

    lines.append("Notes:")
    lines.append(f"- {active_count} active repos scanned")
    lines.append(f"- {manual_count} manual-only repos checked")

    return "\n".join(lines).strip() + "\n"


def render_no_completed_run(
    heading: str = DEFAULT_HEADING,
    security_overview: SecurityOverview | None = None,
) -> str:
    lines = [
        f"## {heading}",
        "",
        "No completed security-agent run this week.",
    ]
    _append_security_overview(lines, security_overview)
    _append_blank(lines)
    lines.extend(
        [
            "Action needed:",
            "- Review automation runner: remediation report latest.json is missing.",
            "",
            "Notes:",
            "- GitHub alert counts are dashboard counts only; no remediation pass completed.",
        ]
    )
    return "\n".join(lines).strip() + "\n"


def render_stale_report(
    completed_at: str,
    heading: str = DEFAULT_HEADING,
    security_overview: SecurityOverview | None = None,
) -> str:
    lines = [
        f"## {heading}",
        "",
        "Stale report.",
    ]
    _append_security_overview(lines, security_overview)
    _append_blank(lines)
    lines.extend(
        [
            "Action needed:",
            f"- Review automation runner: latest report is stale. Last completed run: {completed_at}",
            "",
            "Notes:",
            "- GitHub alert counts are dashboard counts only; remediation details are stale.",
        ]
    )
    return "\n".join(lines).strip() + "\n"


def security_overview_from_summary(summary: dict[str, Any]) -> SecurityOverview:
    counts = summary.get("open_alert_counts")
    if not isinstance(counts, dict):
        counts = summary
    return SecurityOverview(
        dependabot=_int_value(counts.get("dependabot")),
        code_scanning=_int_value(counts.get("code_scanning")),
        secret_scanning=_int_value(counts.get("secret_scanning")),
    )


def _append_security_overview(lines: list[str], overview: SecurityOverview | None) -> None:
    if overview is None or overview.total == 0:
        return
    _append_blank(lines)
    lines.extend(
        [
            "GitHub open alerts:",
            f"- Dependabot: {overview.dependabot}",
            f"- Code scanning: {overview.code_scanning}",
            f"- Secret scanning: {overview.secret_scanning}",
            f"- Total: {overview.total}",
            "",
        ]
    )


def _append_blank(lines: list[str]) -> None:
    if lines and lines[-1] != "":
        lines.append("")


def _run_summary_lines(
    summary: dict[str, Any],
    units: list[dict[str, Any]],
    overview: SecurityOverview,
) -> list[str]:
    result_totals = summary.get("result_totals")
    if not isinstance(result_totals, dict):
        result_totals = {}

    initial = _summary_counts(summary, "initial_open_alert_counts", "open_alert_counts")
    if initial.total == 0:
        initial = _unit_alert_counts(units)
    current = overview

    patched_alerts = _nested_int(result_totals, "patched_by_automation", "alerts")
    if patched_alerts == 0:
        patched_alerts = _sum_alerts(
            units,
            lambda unit: str(unit.get("outcome", "")).lower() in {"merged", "opened_pr"},
        )
    patched_prs = _nested_int(result_totals, "patched_by_automation", "units")
    if patched_prs == 0:
        patched_prs = sum(
            1
            for unit in units
            if str(unit.get("outcome", "")).lower() in {"merged", "opened_pr"}
        )

    opened_or_updated_prs = _nested_int(result_totals, "opened_or_updated_prs", "total")
    if opened_or_updated_prs == 0:
        opened_or_updated_prs = patched_prs

    merged_prs = _nested_int(result_totals, "merged_prs", "total")
    if merged_prs == 0:
        merged_prs = sum(1 for unit in units if str(unit.get("outcome", "")).lower() == "merged")
    merged_alerts = _sum_alerts(
        units,
        lambda unit: str(unit.get("outcome", "")).lower() == "merged",
    )

    manual_alerts = _nested_int(result_totals, "remaining_manual_review", "alerts")
    if manual_alerts == 0:
        manual_alerts = _sum_alerts(units, _is_blocked_or_manual)
    manual_units = _nested_int(result_totals, "remaining_manual_review", "units")
    if manual_units == 0:
        manual_units = sum(1 for unit in units if _is_blocked_or_manual(unit))

    return [
        "Run summary:",
        f"- Initial alerts: {initial.total} ({_inline_counts(initial)})",
        (
            f"- Patched by automation: {patched_alerts} {_plural(patched_alerts, 'alert')} "
            f"across {patched_prs} {_plural(patched_prs, 'PR')}"
        ),
        f"- PRs created or updated: {opened_or_updated_prs}",
        (
            f"- Auto-merged: {merged_alerts} {_plural(merged_alerts, 'alert')} "
            f"across {merged_prs} {_plural(merged_prs, 'PR')}"
        ),
        (
            f"- Manual review required: {manual_alerts} {_plural(manual_alerts, 'alert')} "
            f"across {manual_units} {_plural(manual_units, 'item')}"
        ),
        f"- Current GitHub open alerts: {current.total} ({_inline_counts(current)})",
    ]


def _summary_counts(summary: dict[str, Any], *keys: str) -> SecurityOverview:
    for key in keys:
        value = summary.get(key)
        if isinstance(value, dict):
            return security_overview_from_summary(value)
    return SecurityOverview()


def _unit_alert_counts(units: list[dict[str, Any]]) -> SecurityOverview:
    counts: Counter[str] = Counter()
    for unit in units:
        counts[_alert_class(unit)] += _alert_count(unit)
    return SecurityOverview(
        dependabot=counts["dependabot"],
        code_scanning=counts["code_scanning"],
        secret_scanning=counts["secret_scanning"],
    )


def _sum_alerts(units: list[dict[str, Any]], predicate: Any) -> int:
    return sum(_alert_count(unit) for unit in units if predicate(unit))


def _alert_count(unit: dict[str, Any]) -> int:
    value = _int_value(unit.get("alert_count"))
    if value:
        return value
    for key in ("advisory_ids", "alert_numbers"):
        items = unit.get(key)
        if isinstance(items, list) and items:
            return len(items)
    return 1


def _nested_int(data: dict[str, Any], key: str, nested_key: str) -> int:
    value = data.get(key)
    if not isinstance(value, dict):
        return 0
    return _int_value(value.get(nested_key))


def _inline_counts(counts: SecurityOverview) -> str:
    return (
        f"Dependabot {counts.dependabot}, "
        f"code scanning {counts.code_scanning}, "
        f"secret scanning {counts.secret_scanning}"
    )


def _iter_units(summary: dict[str, Any]) -> list[dict[str, Any]]:
    for key in ("units", "remediation_units", "items", "records", "summaries"):
        value = summary.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
    return []


def _counts_for(units: list[dict[str, Any]], alert_class: str) -> Counter[str]:
    counts: Counter[str] = Counter()
    for unit in units:
        if _alert_class(unit) != alert_class:
            continue
        outcome = str(unit.get("outcome", "")).lower()
        if outcome == "merged":
            counts["merged"] += 1
        elif outcome == "opened_pr":
            counts["opened_pr"] += 1
        if _is_blocked_or_manual(unit):
            counts["blocked"] += 1
            counts["manual"] += 1
    return counts


def _manual_actions_by_class(units: list[dict[str, Any]]) -> Counter[str]:
    by_class: Counter[str] = Counter()
    for unit in units:
        if not _is_blocked_or_manual(unit):
            continue
        by_class[_alert_class(unit)] += 1
    return by_class


def _patched_by_automation_lines(summary: dict[str, Any], units: list[dict[str, Any]]) -> list[str]:
    owner = _summary_owner(summary)
    public_lines: list[str] = []
    private_count = 0

    for unit in units:
        if str(unit.get("outcome", "")).lower() not in {"opened_pr", "merged"}:
            continue
        if _is_explicit_public(unit):
            public_lines.append(_public_pr_line(unit, owner))
        else:
            private_count += 1

    if not public_lines and private_count == 0:
        return []

    lines = ["Patched by automation:"]
    lines.extend(public_lines)
    if private_count:
        lines.append(
            f"- Private or undisclosed repos: {private_count} {_plural(private_count, 'PR')} created, updated, or merged"
        )
    return lines


def _manual_review_detail_lines(units: list[dict[str, Any]]) -> list[str]:
    public_by_repo: dict[str, Counter[str]] = {}
    reasons_by_repo: dict[str, set[str]] = {}
    private_count = 0

    for unit in units:
        if not _is_blocked_or_manual(unit):
            continue
        if not _is_explicit_public(unit):
            private_count += 1
            continue
        repo = _repo_name(unit)
        public_by_repo.setdefault(repo, Counter())[_alert_class(unit)] += 1
        reasons_by_repo.setdefault(repo, set()).add(_manual_reason(unit))

    lines: list[str] = []
    for repo in sorted(public_by_repo):
        class_counts = public_by_repo[repo]
        parts = []
        for alert_class in ("dependabot", "code_scanning", "secret_scanning"):
            count = class_counts.get(alert_class, 0)
            if count:
                parts.append(f"{_class_label(alert_class)} {count}")
        for alert_class, count in sorted(class_counts.items()):
            if alert_class not in {"dependabot", "code_scanning", "secret_scanning"}:
                parts.append(f"{_class_label(alert_class)} {count}")
        reasons = ", ".join(sorted(reason for reason in reasons_by_repo.get(repo, set()) if reason))
        suffix = f" ({reasons})" if reasons else ""
        lines.append(f"- {repo}: {', '.join(parts)}{suffix}")

    if private_count:
        lines.append(
            f"- Private or undisclosed repos: {private_count} {_plural(private_count, 'manual-review item')}"
        )
    return lines


def _is_blocked_or_manual(unit: dict[str, Any]) -> bool:
    outcome = str(unit.get("outcome", "")).lower()
    repository_mode = str(unit.get("repository_mode", "")).lower()
    follow_up = unit.get("manual_follow_up_actions") or unit.get("manual_actions") or []
    return (
        repository_mode == "manual_only"
        or bool(follow_up)
        or outcome in {"blocked", "failed"}
    )


def _manual_reason(unit: dict[str, Any]) -> str:
    reason_code = str(unit.get("reason_code") or "").strip()
    if reason_code:
        return reason_code if reason_code in SAFE_REASON_CODES else "manual_follow_up"
    follow_up = unit.get("manual_follow_up_actions") or unit.get("manual_actions") or []
    if follow_up:
        return "manual_follow_up"
    outcome = str(unit.get("outcome") or "").lower()
    if outcome in {"blocked", "failed"}:
        return outcome
    return "review_required"


def _alert_class(unit: dict[str, Any]) -> str:
    raw = str(unit.get("alert_class") or unit.get("class") or "").lower()
    return raw.replace("-", "_").replace(" ", "_")


def _repo_name(unit: dict[str, Any]) -> str:
    for key in ("repository", "repo", "repository_name"):
        value = unit.get(key)
        if isinstance(value, str) and value:
            return value.split("/")[-1]
        if isinstance(value, dict):
            nested = value.get("name") or value.get("full_name")
            if isinstance(nested, str) and nested:
                return nested.split("/")[-1]
    return "unknown"


def _summary_owner(summary: dict[str, Any]) -> str:
    owner = summary.get("owner")
    if isinstance(owner, str) and owner:
        return owner.lower()
    profile = summary.get("profile")
    if isinstance(profile, dict):
        owner = profile.get("owner")
        if isinstance(owner, str) and owner:
            return owner.lower()
    return ""


def _is_explicit_public(unit: dict[str, Any]) -> bool:
    for key in ("repository_visibility", "visibility", "repo_visibility"):
        value = unit.get(key)
        if isinstance(value, str):
            return value.lower() == "public"
    repository = unit.get("repository")
    if isinstance(repository, dict):
        visibility = repository.get("visibility")
        if isinstance(visibility, str):
            return visibility.lower() == "public"
        private = repository.get("private")
        if isinstance(private, bool):
            return not private
    for key in ("repository_private", "private", "is_private"):
        value = unit.get(key)
        if isinstance(value, bool):
            return not value
    return False


def _public_pr_line(unit: dict[str, Any], owner: str) -> str:
    repo = _repo_name(unit)
    label = _pr_label(unit)
    url = _safe_pr_url(unit, owner, repo)
    if url:
        return f"- {repo}: [{label}]({url})"
    return f"- {repo}: {label}"


def _pr_label(unit: dict[str, Any]) -> str:
    alert_class = _alert_class(unit)
    if alert_class == "dependabot":
        return "Dependabot remediation PR"
    if alert_class == "code_scanning":
        return "Code scanning remediation PR"
    if alert_class == "secret_scanning":
        return "Secret scanning cleanup PR"
    return "Security remediation PR"


def _safe_pr_url(unit: dict[str, Any], owner: str, repo: str) -> str:
    if not owner or not repo:
        return ""
    for key in ("pull_request_url", "pr_url", "pull_request_link", "pr_link"):
        value = unit.get(key)
        if isinstance(value, str) and _is_allowed_pr_url(value, owner, repo):
            return value
    pull_request = unit.get("pull_request")
    if isinstance(pull_request, dict):
        for key in ("html_url", "url"):
            value = pull_request.get(key)
            if isinstance(value, str) and _is_allowed_pr_url(value, owner, repo):
                return value
    return ""


def _is_allowed_pr_url(url: str, owner: str, repo: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme != "https" or parsed.netloc.lower() != "github.com":
        return False
    parts = [part for part in parsed.path.split("/") if part]
    return (
        len(parts) >= 4
        and parts[0].lower() == owner.lower()
        and parts[1].lower() == repo.lower()
        and parts[2] == "pull"
        and parts[3].isdigit()
    )


def _plural(count: int, noun: str) -> str:
    if count == 1:
        return noun
    if noun == "PR":
        return "PRs"
    return f"{noun}s"


def _class_label(alert_class: str) -> str:
    if alert_class == "dependabot":
        return "Dependabot"
    if alert_class == "code_scanning":
        return "code scanning"
    if alert_class == "secret_scanning":
        return "secret scanning"
    return alert_class.replace("_", " ")


def _first_int(data: Any, *keys: str) -> int:
    if not isinstance(data, dict):
        return 0
    for key in keys:
        value = data.get(key)
        if isinstance(value, int):
            return value
    return 0


def _int_value(value: Any) -> int:
    if isinstance(value, bool):
        return 0
    if isinstance(value, int):
        return value
    if isinstance(value, str) and value.isdigit():
        return int(value)
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("latest_json", help="Path to github-security-agent latest.json")
    parser.add_argument("--heading", default=DEFAULT_HEADING, help="Markdown heading text")
    parser.add_argument("--security-overview-json", help="Optional sanitized GitHub open-alert counts JSON")
    parser.add_argument("--output", help="Write Markdown to this file instead of stdout")
    args = parser.parse_args()

    overview = load_security_overview_json(args.security_overview_json) if args.security_overview_json else None
    markdown = render_weekly_report(
        load_latest_json(args.latest_json),
        heading=args.heading,
        security_overview=overview,
    )
    if args.output:
        Path(args.output).write_text(markdown, encoding="utf-8")
    else:
        print(markdown, end="")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())