webapp: export Blancco reports for the last 1/3/6/12 months

/reports/export?range=1m|3m|6m|12m&format=csv|zip. CSV has one row per erased drive (system and drive serials, standard, result, times, licence consumption id). ZIP holds the original XML/PDF reports plus summary.csv. Reports are selected by the date inside each XML, falling back to mtime only when it is missing. Exports are audit-logged.
2026-10-05 09:32:32 -04:00
parent 678d10d5c2
commit 96c4e5dd11
3 changed files with 159 additions and 0 deletions
--- a/webapp/services/blancco_report.py
+++ b/webapp/services/blancco_report.py
@@ -103,3 +103,69 @@ def parse(path: str) -> dict:
            out["hardware"][sub.attrib.get("name", "")] = _walk(sub)

    return out
+
+
+def report_datetime(data: dict):
+    """Report creation time from the report itself, or None if missing.
+
+    Blancco writes ISO 8601 with a numeric offset, e.g.
+    2026-09-04T17:00:22+0000. File mtime is not reliable for audit ranges:
+    copying reports onto the share resets it.
+    """
+    from datetime import datetime
+    raw = (data.get("meta") or {}).get("date") or ""
+    try:
+        return datetime.strptime(raw, "%Y-%m-%dT%H:%M:%S%z")
+    except ValueError:
+        return None
+
+
+EXPORT_COLUMNS = [
+    "report_file", "report_date", "document_id", "product_version",
+    "system_manufacturer", "system_model", "system_serial",
+    "drive_model", "drive_serial", "drive_capacity_gb", "drive_interface",
+    "erasure_standard", "state", "start_time", "end_time", "elapsed_time",
+    "total_errors", "license_consumption_id", "business_name",
+]
+
+
+def export_rows(filename: str, data: dict, when: str) -> list[dict]:
+    """One CSV row per erased drive; a report with no erasures still gets a row."""
+    meta = data.get("meta") or {}
+    system = (data.get("hardware") or {}).get("system") or {}
+    if isinstance(system, list):
+        system = system[0] if system else {}
+    base = {
+        "report_file": filename,
+        "report_date": when,
+        "document_id": meta.get("document_id", ""),
+        "product_version": meta.get("product_version", ""),
+        "system_manufacturer": system.get("manufacturer", ""),
+        "system_model": system.get("model", ""),
+        "system_serial": system.get("serial", ""),
+        "license_consumption_id": ";".join(data.get("license_consumption_ids") or []),
+        "business_name": (data.get("company") or {}).get("business_name", ""),
+    }
+    rows = []
+    for er in data.get("erasures") or []:
+        target = er.get("target") or {}
+        if isinstance(target, list):
+            target = target[0] if target else {}
+        try:
+            cap = f"{int(target.get('capacity', '')) / 1e9:.1f}"
+        except ValueError:
+            cap = ""
+        rows.append({
+            **base,
+            "drive_model": target.get("model", ""),
+            "drive_serial": target.get("serial", ""),
+            "drive_capacity_gb": cap,
+            "drive_interface": target.get("interface_type", ""),
+            "erasure_standard": er.get("erasure_standard_name", ""),
+            "state": er.get("state", ""),
+            "start_time": er.get("start_time", ""),
+            "end_time": er.get("end_time", ""),
+            "elapsed_time": er.get("elapsed_time", ""),
+            "total_errors": er.get("total_errors", ""),
+        })
+    return rows or [base]