Skip to content

report_service

report_service

Service layer for multi-device reporting.

ReportService

Service for handling multi-device reports.

Source code in hier_config_api/services/report_service.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class ReportService:
    """Service for handling multi-device reports."""

    @staticmethod
    def create_report(remediations: list[DeviceRemediation]) -> dict[str, Any]:
        """Create a report from multiple device remediations."""
        report_data: dict[str, Any] = {
            "devices": [],
            "total_devices": len(remediations),
            "devices_with_changes": 0,
            "total_changes": 0,
            "changes_by_tag": {},
            "change_details": [],
        }

        # Track changes across devices
        change_tracker: dict[str, dict[str, Any]] = defaultdict(
            lambda: {"count": 0, "devices": [], "tags": []}
        )

        for device_rem in remediations:
            # Generate remediation for this device
            remediation_result = RemediationService.generate_remediation(
                platform=device_rem.platform,
                running_config=device_rem.running_config,
                intended_config=device_rem.intended_config,
            )

            remediation_config = remediation_result["remediation_config"]
            has_changes = bool(remediation_config.strip())

            if has_changes:
                report_data["devices_with_changes"] += 1

            # Track changes
            changes = remediation_config.splitlines()
            report_data["total_changes"] += len(changes)

            # Store device data
            device_data = {
                "device_id": device_rem.device_id,
                "platform": device_rem.platform,
                "has_changes": has_changes,
                "change_count": len(changes),
                "remediation": remediation_config,
                "rollback": remediation_result["rollback_config"],
            }
            report_data["devices"].append(device_data)

            # Track individual changes
            for change in changes:
                if change.strip():
                    change_tracker[change]["count"] += 1
                    change_tracker[change]["devices"].append(device_rem.device_id)

        # Convert change tracker to change details
        report_data["change_details"] = [
            {
                "change_text": change_text,
                "device_count": data["count"],
                "device_ids": data["devices"],
                "tags": data["tags"],
            }
            for change_text, data in change_tracker.items()
        ]

        return report_data

    @staticmethod
    def get_summary(report_data: dict[str, Any]) -> ReportSummary:
        """Get summary statistics from report."""
        return ReportSummary(
            total_devices=report_data["total_devices"],
            devices_with_changes=report_data["devices_with_changes"],
            total_changes=report_data["total_changes"],
            changes_by_tag=report_data.get("changes_by_tag", {}),
        )

    @staticmethod
    def get_changes(
        report_data: dict[str, Any],
        tag_filter: str | None = None,
        min_devices: int = 1,
    ) -> list[ChangeDetail]:
        """Get detailed change analysis."""
        changes = report_data.get("change_details", [])

        # Filter by tag if specified
        if tag_filter:
            changes = [c for c in changes if tag_filter in c.get("tags", [])]

        # Filter by minimum devices
        changes = [c for c in changes if c["device_count"] >= min_devices]

        # Convert to ChangeDetail objects
        return [
            ChangeDetail(
                change_text=c["change_text"],
                device_count=c["device_count"],
                device_ids=c["device_ids"],
                tags=c.get("tags", []),
            )
            for c in changes
        ]

    @staticmethod
    def export_report(report_data: dict[str, Any], format_type: str = "json") -> str:
        """Export report in specified format."""
        if format_type == "json":
            return json.dumps(report_data, indent=2)

        elif format_type == "yaml":
            return yaml.dump(report_data, default_flow_style=False)

        elif format_type == "csv":
            output = io.StringIO()
            writer = csv.writer(output)

            # Write header
            writer.writerow(
                ["Device ID", "Platform", "Has Changes", "Change Count", "Remediation Summary"]
            )

            # Write device rows
            for device in report_data.get("devices", []):
                writer.writerow(
                    [
                        device["device_id"],
                        device["platform"],
                        device["has_changes"],
                        device["change_count"],
                        (
                            device["remediation"][:50] + "..."
                            if len(device["remediation"]) > 50
                            else device["remediation"]
                        ),
                    ]
                )

            return output.getvalue()

        else:
            raise ValueError(f"Unsupported format: {format_type}")

create_report(remediations) staticmethod

Create a report from multiple device remediations.

Source code in hier_config_api/services/report_service.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@staticmethod
def create_report(remediations: list[DeviceRemediation]) -> dict[str, Any]:
    """Create a report from multiple device remediations."""
    report_data: dict[str, Any] = {
        "devices": [],
        "total_devices": len(remediations),
        "devices_with_changes": 0,
        "total_changes": 0,
        "changes_by_tag": {},
        "change_details": [],
    }

    # Track changes across devices
    change_tracker: dict[str, dict[str, Any]] = defaultdict(
        lambda: {"count": 0, "devices": [], "tags": []}
    )

    for device_rem in remediations:
        # Generate remediation for this device
        remediation_result = RemediationService.generate_remediation(
            platform=device_rem.platform,
            running_config=device_rem.running_config,
            intended_config=device_rem.intended_config,
        )

        remediation_config = remediation_result["remediation_config"]
        has_changes = bool(remediation_config.strip())

        if has_changes:
            report_data["devices_with_changes"] += 1

        # Track changes
        changes = remediation_config.splitlines()
        report_data["total_changes"] += len(changes)

        # Store device data
        device_data = {
            "device_id": device_rem.device_id,
            "platform": device_rem.platform,
            "has_changes": has_changes,
            "change_count": len(changes),
            "remediation": remediation_config,
            "rollback": remediation_result["rollback_config"],
        }
        report_data["devices"].append(device_data)

        # Track individual changes
        for change in changes:
            if change.strip():
                change_tracker[change]["count"] += 1
                change_tracker[change]["devices"].append(device_rem.device_id)

    # Convert change tracker to change details
    report_data["change_details"] = [
        {
            "change_text": change_text,
            "device_count": data["count"],
            "device_ids": data["devices"],
            "tags": data["tags"],
        }
        for change_text, data in change_tracker.items()
    ]

    return report_data

get_summary(report_data) staticmethod

Get summary statistics from report.

Source code in hier_config_api/services/report_service.py 
83
84
85
86
87
88
89
90
91
@staticmethod
def get_summary(report_data: dict[str, Any]) -> ReportSummary:
    """Get summary statistics from report."""
    return ReportSummary(
        total_devices=report_data["total_devices"],
        devices_with_changes=report_data["devices_with_changes"],
        total_changes=report_data["total_changes"],
        changes_by_tag=report_data.get("changes_by_tag", {}),
    )

get_changes(report_data, tag_filter=None, min_devices=1) staticmethod

Get detailed change analysis.

Source code in hier_config_api/services/report_service.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@staticmethod
def get_changes(
    report_data: dict[str, Any],
    tag_filter: str | None = None,
    min_devices: int = 1,
) -> list[ChangeDetail]:
    """Get detailed change analysis."""
    changes = report_data.get("change_details", [])

    # Filter by tag if specified
    if tag_filter:
        changes = [c for c in changes if tag_filter in c.get("tags", [])]

    # Filter by minimum devices
    changes = [c for c in changes if c["device_count"] >= min_devices]

    # Convert to ChangeDetail objects
    return [
        ChangeDetail(
            change_text=c["change_text"],
            device_count=c["device_count"],
            device_ids=c["device_ids"],
            tags=c.get("tags", []),
        )
        for c in changes
    ]

export_report(report_data, format_type='json') staticmethod

Export report in specified format.

Source code in hier_config_api/services/report_service.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@staticmethod
def export_report(report_data: dict[str, Any], format_type: str = "json") -> str:
    """Export report in specified format."""
    if format_type == "json":
        return json.dumps(report_data, indent=2)

    elif format_type == "yaml":
        return yaml.dump(report_data, default_flow_style=False)

    elif format_type == "csv":
        output = io.StringIO()
        writer = csv.writer(output)

        # Write header
        writer.writerow(
            ["Device ID", "Platform", "Has Changes", "Change Count", "Remediation Summary"]
        )

        # Write device rows
        for device in report_data.get("devices", []):
            writer.writerow(
                [
                    device["device_id"],
                    device["platform"],
                    device["has_changes"],
                    device["change_count"],
                    (
                        device["remediation"][:50] + "..."
                        if len(device["remediation"]) > 50
                        else device["remediation"]
                    ),
                ]
            )

        return output.getvalue()

    else:
        raise ValueError(f"Unsupported format: {format_type}")