generate_sample_report.py

#!/usr/bin/env python3
"""Regenerate the checked-in sample assessment report.

``examples/sample_assessment_report.html`` is the report a prospective user opens
before installing anything, and the source for the README screenshot. It is built
here from the *live* check registry rather than captured from a real run, so it
cannot go stale as checks are added, and so it never contains real customer data.

Findings are assigned deterministically from a seeded RNG, which keeps the output
byte-stable across runs — regenerating after a styling change produces a diff that
only reflects the styling change.

    python scripts/generate_sample_report.py
"""

from __future__ import annotations

import argparse
import random
import sys
from collections import Counter
from datetime import datetime, timedelta
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from sample_contact_flows import (  # noqa: E402
    BILLING_FLOW,
    BILLING_ID,
    MAIN_IVR_FLOW,
    MAIN_IVR_ID,
    OVERFLOW_FLOW,
    OVERFLOW_ID,
    dormant_flow_specs,
)

from amazon_connect_assessment import __version__  # noqa: E402
from amazon_connect_assessment.checks.base import BaseCheck  # noqa: E402
from amazon_connect_assessment.checks.control_registry import (  # noqa: E402
    AtomicControl,
    ExecutionSource,
)
from amazon_connect_assessment.checks.registration import register_all_checks  # noqa: E402
from amazon_connect_assessment.checks.registry import CheckRegistry  # noqa: E402
from amazon_connect_assessment.journey.journey_scorer import (  # noqa: E402
    generate_journey_findings,
    score_journeys,
)
from amazon_connect_assessment.journey.models import (  # noqa: E402
    JourneyMapResult,
    PhoneNumberEntry,
    TierAssignment,
)
from amazon_connect_assessment.journey.path_enumerator import enumerate_journeys  # noqa: E402
from amazon_connect_assessment.journey.renderer import flow_to_diagram_artifacts  # noqa: E402
from amazon_connect_assessment.journey.super_graph import build_super_graph  # noqa: E402
from amazon_connect_assessment.models import (  # noqa: E402
    AssessmentMetadata,
    AssessmentResult,
    AssessmentSummary,
    CheckStatus,
    ConnectInstance,
    ContactFlow,
    Finding,
    FindingDisposition,
    Pillar,
    Severity,
)
from amazon_connect_assessment.parsers import ContactFlowParser  # noqa: E402
from amazon_connect_assessment.report_generator import ReportGenerator  # noqa: E402
from amazon_connect_assessment.score_policy import (  # noqa: E402
    FindingScoreClassification,
    classify_finding,
    compute_scored_control_counts,
    is_control_failure,
)

DEFAULT_OUTPUT = REPO_ROOT / "examples" / "sample_assessment_report.html"

# Documentation-only placeholders. 111122223333 is the account ID AWS reserves for
# examples, and the instance IDs are fixed so the sample diff stays readable.
SAMPLE_ACCOUNT_ID = "111122223333"
SAMPLE_REGION = "us-east-1"
SAMPLE_ASSESSMENT_ID = "00000000-0000-4000-8000-000000000001"
SAMPLE_TIMESTAMP = datetime(2026, 1, 15, 14, 30, 0)
SAMPLE_SEED = 20260115

# Weighted so the sample shows a realistic mix: mostly passing, a meaningful tail of
# failures to demonstrate the remediation guidance, and a few access-denied skips.
STATUS_WEIGHTS: list[tuple[CheckStatus, int]] = [
    (CheckStatus.PASS, 62),
    (CheckStatus.FAIL, 27),
    (CheckStatus.SKIPPED, 6),
    (CheckStatus.NOT_APPLICABLE, 4),
    (CheckStatus.ERROR, 1),
]

PRIMARY_INSTANCE_ID = "11111111-1111-4111-8111-111111111111"
OVERFLOW_INSTANCE_ID = "22222222-2222-4222-8222-222222222222"

# Reserved-for-documentation ranges: +1 800 555 01xx is the North American
# fictional-number block, and 206-555-01xx is its DID equivalent.
SAMPLE_PHONE_NUMBERS: dict[str, list[dict[str, str]]] = {
    PRIMARY_INSTANCE_ID: [
        {
            "PhoneNumber": "+18005550100",
            "PhoneNumberType": "TOLL_FREE",
            "PhoneNumberCountryCode": "US",
            "PhoneNumberDescription": "Main customer service line",
            "flow_id": MAIN_IVR_ID,
        },
        {
            "PhoneNumber": "+12065550142",
            "PhoneNumberType": "DID",
            "PhoneNumberCountryCode": "US",
            "PhoneNumberDescription": "Seattle local number",
            "flow_id": MAIN_IVR_ID,
        },
    ],
    OVERFLOW_INSTANCE_ID: [
        {
            "PhoneNumber": "+18005550199",
            "PhoneNumberType": "TOLL_FREE",
            "PhoneNumberCountryCode": "US",
            "PhoneNumberDescription": "Peak-season overflow line",
            "flow_id": OVERFLOW_ID,
        },
    ],
}


def _flow(instance_id: str, flow_id: str, name: str, flow_type: str, content: dict) -> ContactFlow:
    """Build a ContactFlow the way instance discovery would return it."""
    return ContactFlow(
        id=flow_id,
        arn=(
            f"arn:aws:connect:{SAMPLE_REGION}:{SAMPLE_ACCOUNT_ID}:instance/"
            f"{instance_id}/contact-flow/{flow_id}"
        ),
        name=name,
        type=flow_type,
        state="ACTIVE",
        content=content,
    )


def _build_instances() -> list[ConnectInstance]:
    """Build the two synthetic instances, including their contact flows."""
    primary_flows = [
        _flow(PRIMARY_INSTANCE_ID, MAIN_IVR_ID, "Main IVR", "CONTACT_FLOW", MAIN_IVR_FLOW),
        _flow(PRIMARY_INSTANCE_ID, BILLING_ID, "Billing Support", "CONTACT_FLOW", BILLING_FLOW),
    ]
    primary_flows.extend(
        _flow(PRIMARY_INSTANCE_ID, spec["id"], spec["name"], spec["type"], spec["content"])
        for spec in dormant_flow_specs(1)
    )

    return [
        ConnectInstance(
            instance_id=PRIMARY_INSTANCE_ID,
            instance_arn=(
                f"arn:aws:connect:{SAMPLE_REGION}:{SAMPLE_ACCOUNT_ID}:instance/"
                f"{PRIMARY_INSTANCE_ID}"
            ),
            identity_management_type="SAML",
            inbound_calls_enabled=True,
            outbound_calls_enabled=True,
            instance_alias="example-contact-center",
            service_role=(
                f"arn:aws:iam::{SAMPLE_ACCOUNT_ID}:role/aws-service-role/connect.amazonaws.com"
            ),
            status="ACTIVE",
            contact_flows=primary_flows,
        ),
        ConnectInstance(
            instance_id=OVERFLOW_INSTANCE_ID,
            instance_arn=(
                f"arn:aws:connect:{SAMPLE_REGION}:{SAMPLE_ACCOUNT_ID}:instance/"
                f"{OVERFLOW_INSTANCE_ID}"
            ),
            identity_management_type="CONNECT_MANAGED",
            inbound_calls_enabled=True,
            outbound_calls_enabled=False,
            instance_alias="example-overflow",
            status="ACTIVE",
            contact_flows=[
                _flow(
                    OVERFLOW_INSTANCE_ID,
                    OVERFLOW_ID,
                    "Overflow Routing",
                    "CONTACT_FLOW",
                    OVERFLOW_FLOW,
                )
            ],
        ),
    ]


SAMPLE_INSTANCES = _build_instances()


def parse_args() -> argparse.Namespace:
    """Parse command-line options for the sample report build."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output",
        type=Path,
        default=DEFAULT_OUTPUT,
        help=f"Destination HTML file (default: {DEFAULT_OUTPUT})",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=SAMPLE_SEED,
        help=f"RNG seed controlling the finding mix (default: {SAMPLE_SEED})",
    )
    return parser.parse_args()


def _selected_execution_plan() -> tuple[list[AtomicControl], dict[str, BaseCheck]]:
    """Return the unified selected controls and their BaseCheck executors."""
    registry = CheckRegistry()
    register_all_checks(registry)
    return registry.get_selected_controls(), {
        check.check_id: check for check in registry.get_all_checks()
    }


def _status_for(rng: random.Random, outcome_index: int) -> CheckStatus:
    """Pick a deterministic status while guaranteeing every status is represented."""
    required_statuses = tuple(CheckStatus)
    if outcome_index < len(required_statuses):
        return required_statuses[outcome_index]
    statuses = [status for status, _ in STATUS_WEIGHTS]
    weights = [weight for _, weight in STATUS_WEIGHTS]
    return rng.choices(statuses, weights=weights, k=1)[0]


def _describe(control: AtomicControl, check: BaseCheck, status: CheckStatus) -> str:
    """Build a status-appropriate description for a sample finding."""
    if status is CheckStatus.PASS:
        return f"{control.name} passed. {check.description}"
    if status is CheckStatus.SKIPPED:
        return (
            f"{control.name} was skipped because the assessing identity lacked a required "
            "read permission. Grant the permissions listed in the check catalog and re-run."
        )
    if status is CheckStatus.NOT_APPLICABLE:
        return f"{control.name} does not apply to this instance's configuration."
    if status is CheckStatus.ERROR:
        return f"{control.name} could not complete because an AWS API call was throttled."
    if control.disposition is FindingDisposition.MANUAL_REVIEW:
        return f"{control.name} found a review candidate. {check.description}"
    if control.disposition is FindingDisposition.INFORMATIONAL:
        return f"{control.name} recorded inventory context. {check.description}"
    return f"{control.name} found a measured control gap. {check.description}"


def _parse_flows(instance: ConnectInstance) -> dict:
    """Parse every flow on an instance, keyed by flow id.

    Mirrors ``AssessmentEngine._compute_journey_findings``, including the
    flow_id/flow_name backfill — ``Content`` never carries Identifier/Name, so
    without it every flow collapses onto the same super-graph key.
    """
    parser = ContactFlowParser()
    graphs = {}
    for flow in instance.contact_flows:
        graph = parser.parse(flow.content)
        graph.flow_id = flow.id
        graph.flow_name = flow.name
        graphs[flow.id] = graph
    return graphs


def _tier_assignments(instance: ConnectInstance, entry_flow_ids: set[str]) -> list[TierAssignment]:
    """Classify flows the way ``journey.topology.resolve_topology`` would.

    Flows a phone number terminates on are tier 1; flows reachable only via a
    transfer are tier 2; everything else is dormant. Assigned here rather than
    resolved from AWS because ``resolve_topology`` is the one step of the
    pipeline that needs live ``ListPhoneNumbersV2`` / ``ListFlowAssociations``
    calls. Every downstream step below is the production code.
    """
    transfer_targets = {BILLING_ID}
    assignments: list[TierAssignment] = []
    for flow in instance.contact_flows:
        if flow.id in entry_flow_ids:
            tier, rationale = "tier1_did", "Inbound phone number terminates on this flow"
        elif flow.id in transfer_targets:
            tier, rationale = "tier2_traffic", "Reached by a transfer from a tier 1 flow"
        else:
            tier, rationale = "tier3_dormant", "No phone number association and no inbound traffic"
        assignments.append(
            TierAssignment(flow_id=flow.id, flow_name=flow.name, tier=tier, rationale=rationale)
        )
    return assignments


def build_journey_data(
    instances: list[ConnectInstance],
) -> tuple[list[dict], list[Finding]]:
    """Run the journey pipeline over the sample flows.

    Returns ``(journey_map_entries, journey_findings)`` — the same two things
    ``AssessmentEngine`` produces, built by the same parser, super-graph, path
    enumerator, scorer and renderer.
    """
    entries: list[dict] = []
    findings: list[Finding] = []

    for instance in instances:
        numbers = SAMPLE_PHONE_NUMBERS[instance.instance_id]
        graphs = _parse_flows(instance)
        entry_flow_ids = {n["flow_id"] for n in numbers}

        # --- Caller Journey Map diagrams (one per number, deduped per flow) ---
        rendered: dict = {}
        for num in numbers:
            flow_id = num["flow_id"]
            flow = next(f for f in instance.contact_flows if f.id == flow_id)
            if flow_id not in rendered:
                rendered[flow_id] = flow_to_diagram_artifacts(graphs[flow_id])
            artifacts = rendered[flow_id]
            entries.append(
                {
                    "instance_id": instance.instance_id,
                    "instance_display_name": instance.display_name,
                    "phone_number": num["PhoneNumber"],
                    "phone_type": num["PhoneNumberType"],
                    "phone_country_code": num["PhoneNumberCountryCode"],
                    "phone_description": num["PhoneNumberDescription"],
                    "flow_id": flow_id,
                    "flow_name": flow.name,
                    "flow_type": flow.type,
                    "diagram_html": artifacts.diagram_html,
                    "diagram_model": artifacts.diagram_model,
                    "exports": artifacts.export_payload(),
                }
            )

        # --- journey-* findings ---
        assignments = _tier_assignments(instance, entry_flow_ids)
        super_graph = build_super_graph(graphs, assignments)
        phone_entries = [
            PhoneNumberEntry(
                phone_number=num["PhoneNumber"],
                number_type=num["PhoneNumberType"],
                country_code=num["PhoneNumberCountryCode"],
                contact_flow_id=num["flow_id"],
                contact_flow_name=next(
                    f.name for f in instance.contact_flows if f.id == num["flow_id"]
                ),
                target_arn=instance.instance_arn,
            )
            for num in numbers
        ]
        journeys = enumerate_journeys(super_graph=super_graph, phone_entries=phone_entries)
        scores = score_journeys(journeys, {})
        dormant = [a.flow_id for a in assignments if a.tier == "tier3_dormant"]
        containment = {
            e.phone_number: (
                sum(
                    1
                    for j in journeys
                    if j.entry_number == e.phone_number and j.terminal_type != "agent_queue"
                )
                / max(1, sum(1 for j in journeys if j.entry_number == e.phone_number))
                * 100
            )
            for e in phone_entries
        }
        result = JourneyMapResult(
            phone_entries=phone_entries,
            tier_assignments=assignments,
            journeys=journeys,
            scores=scores,
            dormant_flows=dormant,
            dynamic_edges=super_graph.dynamic_references,
            containment_scores=containment,
        )
        findings.extend(generate_journey_findings(result, instance_id=instance.instance_id))

    entries.sort(key=lambda e: (e["instance_display_name"], e["phone_number"]))
    return entries, findings


def _sample_evidence(control: AtomicControl, instance: ConnectInstance) -> dict:
    """Return deterministic evidence that exercises the report's generic renderers."""
    evidence = {
        "instance_alias": instance.instance_alias,
        "sample_data": True,
        "root_condition_key": control.root_condition_key,
    }
    if control.control_id == "perf-flow-complexity-001":
        evidence["flow_structural_metrics"] = [
            {
                "flow": "Main customer service IVR",
                "flow_id": "8f14e45f-ea6b-4f9d-93b3-9f8f477a2e17",
                "total_actions": 24,
                "reachable_actions": 22,
                "longest_route_transitions": 9,
                "route_analysis_capped": False,
                "integration_points": 3,
                "cycles": 1,
                "paths_enumerated": 18,
                "path_enumeration_capped": False,
                "module_invocations": 2,
            },
            {
                "flow": "Billing self-service",
                "flow_id": "45c48cce-2e2d-4c4f-9179-0f6a01c12345",
                "total_actions": 16,
                "reachable_actions": 16,
                "longest_route_transitions": 7,
                "route_analysis_capped": False,
                "integration_points": 2,
                "cycles": 0,
                "paths_enumerated": 12,
                "path_enumeration_capped": False,
                "module_invocations": 1,
            },
        ]
    return evidence


def build_sample_result(seed: int) -> AssessmentResult:
    """Construct one canonical outcome per control and synthetic instance."""
    rng = random.Random(seed)
    controls, base_checks = _selected_execution_plan()
    control_by_id = {control.control_id: control for control in controls}

    journey_map_entries, generated_journey_findings = build_journey_data(SAMPLE_INSTANCES)
    journey_by_instance_and_id = {
        (finding.instance_id, finding.check_id): finding for finding in generated_journey_findings
    }

    findings: list[Finding] = []
    for instance in SAMPLE_INSTANCES:
        for control in controls:
            if control.execution_source is ExecutionSource.JOURNEY:
                finding = journey_by_instance_and_id[(instance.instance_id, control.control_id)]
            else:
                check = base_checks[control.control_id]
                status = _status_for(rng, len(findings))
                finding = Finding(
                    check_id=control.control_id,
                    check_name=control.name,
                    pillar=control.pillar,
                    severity=control.default_severity,
                    status=status,
                    resource_id=instance.instance_id,
                    resource_type="ConnectInstance",
                    description=_describe(control, check, status),
                    remediation=(
                        check.remediation_template
                        or "See the check catalog for remediation or review guidance."
                    ),
                    evidence=_sample_evidence(control, instance),
                    disposition=control.disposition,
                    methodology=control.methodology,
                    instance_id=instance.instance_id,
                )
            finding.timestamp = SAMPLE_TIMESTAMP + timedelta(seconds=len(findings))
            findings.append(finding)

    by_status = Counter(finding.status for finding in findings)
    by_disposition = Counter(finding.disposition for finding in findings)
    classifications = Counter(classify_finding(finding) for finding in findings)
    scored_numerator, scored_denominator = compute_scored_control_counts(findings)
    failed_controls = [finding for finding in findings if is_control_failure(finding)]
    by_severity = Counter(finding.severity for finding in failed_controls)
    journey_count = sum(
        control_by_id[finding.check_id].execution_source is ExecutionSource.JOURNEY
        for finding in findings
    )

    summary = AssessmentSummary(
        total_checks=len(findings),
        passed_checks=by_status[CheckStatus.PASS],
        failed_checks=by_status[CheckStatus.FAIL],
        error_checks=by_status[CheckStatus.ERROR],
        skipped_checks=by_status[CheckStatus.SKIPPED],
        not_applicable_checks=by_status[CheckStatus.NOT_APPLICABLE],
        critical_findings=by_severity[Severity.CRITICAL],
        high_findings=by_severity[Severity.HIGH],
        medium_findings=by_severity[Severity.MEDIUM],
        low_findings=by_severity[Severity.LOW],
        registered_checks=len(findings) - journey_count,
        journey_findings=journey_count,
        control_findings=by_disposition[FindingDisposition.CONTROL],
        manual_review_findings=by_disposition[FindingDisposition.MANUAL_REVIEW],
        informational_findings=by_disposition[FindingDisposition.INFORMATIONAL],
        scored_control_passes=classifications[FindingScoreClassification.SCORED_PASS],
        scored_control_failures=classifications[FindingScoreClassification.SCORED_FAIL],
        unevaluated_controls=classifications[FindingScoreClassification.UNEVALUATED_CONTROL],
        not_applicable_controls=sum(
            finding.disposition is FindingDisposition.CONTROL
            and finding.status is CheckStatus.NOT_APPLICABLE
            for finding in findings
        ),
        scored_control_numerator=scored_numerator,
        scored_control_denominator=scored_denominator,
    )

    metadata = AssessmentMetadata(
        tool_version=__version__,
        execution_time_seconds=42.7,
        aws_account_id=SAMPLE_ACCOUNT_ID,
        aws_region=SAMPLE_REGION,
        execution_environment="sample",
        python_version="3.12",
    )

    return AssessmentResult(
        assessment_id=SAMPLE_ASSESSMENT_ID,
        timestamp=SAMPLE_TIMESTAMP,
        account_id=SAMPLE_ACCOUNT_ID,
        region=SAMPLE_REGION,
        instances=SAMPLE_INSTANCES,
        findings=findings,
        summary=summary,
        metadata=metadata,
        execution_errors=[],
        journey_map_entries=journey_map_entries,
        journey_map_status=None,
    )


def main() -> int:
    """Render the sample report to disk."""
    args = parse_args()
    result = build_sample_result(args.seed)

    html = ReportGenerator().generate_html_report(
        assessment_result=result,
        include_raw_data=False,
        generated_at=SAMPLE_TIMESTAMP,
    )

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(html, encoding="utf-8")

    pillar_counts = {pillar: 0 for pillar in Pillar}
    for finding in result.findings:
        pillar_counts[finding.pillar] += 1
    breakdown = ", ".join(
        f"{pillar.value}={count}" for pillar, count in pillar_counts.items() if count
    )

    print(f"Sample report written: {args.output} ({args.output.stat().st_size / 1024:.1f} KiB)")
    print(f"  {len(result.findings)} findings ({breakdown})")
    print(
        f"  pass={result.summary.passed_checks} fail={result.summary.failed_checks} "
        f"skipped={result.summary.skipped_checks} error={result.summary.error_checks} "
        f"not_applicable={result.summary.not_applicable_checks}"
    )
    print(
        f"  scored controls={result.summary.scored_control_numerator}/"
        f"{result.summary.scored_control_denominator}; "
        f"manual reviews={result.summary.manual_review_findings}; "
        f"informational={result.summary.informational_findings}"
    )
    print(
        f"  {result.summary.journey_findings} Journey-backed outcome(s), "
        f"{len(result.journey_map_entries)} journey map entr(ies)"
    )
    return 0


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