Implement functionality in order to generate report directly from the artifacts folder.

report_generator/sources.py

@@ -1,14 +1,23 @@
 """Collects everything a scan's report is built from.
 
-The one place that knows where the bytes come from. Swapping the store -- persistent_storage today,
-archon tomorrow -- is a change to :func:`collect` and to nothing else: what it returns is files by
-name, and a renderer cannot tell which store handed them over.
+The one place that knows where the bytes come from. Two ways in:
+
+* :func:`collect` reads a scan out of the store -- Postgres for the rows, the bucket for the files.
+* :func:`collect_from_directory` reads a folder of exported artifacts, and needs neither.
+
+Both return the same thing: files by name. Nothing downstream can tell which of the two handed them
+over, which is what makes the second one worth having rather than a second pipeline -- a report of
+an exported run is the same report, built by the same code.
 """
 
 from __future__ import annotations
 
+import json
+import re
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, cast
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, cast
 from uuid import UUID
 
 from sqlalchemy import create_engine, text
@@ -20,7 +29,6 @@ from persistent_storage.utils.s3_utils import fetch_artifact_files
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
-    from datetime import datetime
 
     from mypy_boto3_s3 import S3Client
     from sqlalchemy.sql.elements import TextClause
@@ -254,3 +262,132 @@ def collect(configuration: StorageConfiguration, scan_id: UUID) -> ReportSources
         artifacts=scan.artifacts,
         files=dict(zip(names, contents, strict=True)),
     )
+
+
+# A UUID wherever it appears inside a longer string, which is how the pipeline writes its own id
+# into the paths it records.
+_UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.IGNORECASE)
+
+# The metadata fields that carry a path ending in the run's own id. Both are written by the
+# pipeline itself, so a folder exported from a run identifies itself without being told.
+_ID_BEARING = ("export_folder", "repository_path")
+
+# Read whole rather than streamed: the report needs every file at once, and the largest export in
+# the sample set comes to a few megabytes. A folder that does not fit in memory is not an export.
+_METADATA = ("metadata.json", "meta.json")
+
+
+def collect_from_directory(directory: Path, scan_id: UUID | None = None) -> ReportSources:
+    """Read a folder of exported artifacts, without touching Postgres or the bucket.
+
+    The same report from a directory somebody was sent: a scan that was exported, a pipeline run on
+    a machine with no store attached, a bug report that arrived as a zip. Nothing downstream knows
+    the difference.
+
+    Args:
+        directory: The folder holding the artifacts, as the pipeline exported them.
+        scan_id: The scan to report under. Read out of the artifacts when not
+            given -- the pipeline writes its own id into the paths it records in
+            ``metadata.json``, and failing that the folder may be named by it.
+
+    Returns:
+        The files, under the id the run is known by.
+
+    Raises:
+        PipelineError: The folder does not exist, holds no files, or holds nothing
+            that names the run and none was given.
+    """
+    context.logger.info(f"Collecting the artifacts in {directory}...")
+
+    if not directory.is_dir():
+        message = f"'{directory}' is not a directory."
+
+        raise PipelineError(message)
+
+    files = {entry.name: entry.read_bytes() for entry in sorted(directory.iterdir()) if entry.is_file()}
+
+    if not files:
+        message = f"'{directory}' holds no files."
+
+        raise PipelineError(message)
+
+    context.logger.info(f"Collected {len(files)} file(s) ✔")
+
+    metadata = _metadata_of(files)
+    reported = scan_id or _scan_id_of(metadata, directory)
+
+    if reported is None:
+        message = (
+            f"Nothing in '{directory}' names the scan it belongs to. Pass --scan_id, or keep the "
+            f"metadata.json the pipeline exported alongside the other artifacts."
+        )
+
+        raise PipelineError(message)
+
+    return ReportSources(
+        scan_id=reported,
+        stored_scan_id=reported,
+        run_at=_run_at_of(metadata, directory),
+        artifacts=[],
+        files=files,
+    )
+
+
+def _metadata_of(files: dict[str, bytes]) -> dict[str, Any]:
+    """The pipeline metadata among the collected files, or nothing.
+
+    Read here as well as in the renderer because the id and the date are wanted before a context is
+    built, and a folder without metadata still has to produce a report.
+    """
+    for name in _METADATA:
+        content = files.get(name)
+
+        if content is None:
+            continue
+
+        try:
+            parsed = json.loads(content)
+        except ValueError:
+            return {}
+
+        return parsed if isinstance(parsed, dict) else {}
+
+    return {}
+
+
+def _scan_id_of(metadata: dict[str, Any], directory: Path) -> UUID | None:
+    """Work out which run a folder of artifacts belongs to.
+
+    The pipeline records where it worked and where it exported to, and both paths end in its own
+    id, so an export identifies itself. Where the metadata is gone, the folder may still be named
+    by the run -- which is how these are usually kept.
+    """
+    for field in _ID_BEARING:
+        stated = metadata.get(field)
+
+        if isinstance(stated, str):
+            found = _UUID.search(stated)
+
+            if found:
+                return UUID(found.group(0))
+
+    found = _UUID.fullmatch(directory.resolve().name)
+
+    return UUID(found.group(0)) if found else None
+
+
+def _run_at_of(metadata: dict[str, Any], directory: Path) -> datetime:
+    """When the run happened.
+
+    Its own timestamp where the metadata kept one; otherwise the folder's, which is when it was
+    exported rather than when it ran -- close enough to date a report by, and better than now.
+    """
+    stated = metadata.get("pipeline_start_iso")
+
+    if isinstance(stated, str):
+        try:
+            return datetime.fromisoformat(stated)
+        except ValueError:
+            pass
+
+    return datetime.fromtimestamp(directory.stat().st_mtime, tz=UTC)