US05-03: Parse and Persist Uploader Outcomes (#73)

photo_pipeline/services/upload_batches.py

@@ -18,10 +18,11 @@ cannot undo, so the discipline is:
   vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
   blindly. Resolving that is US05-04.
 
-Per-asset upload *results* are deliberately not interpreted here — parsing the
-report is US05-03. What this story guarantees is that the report exists, is bounded,
-carries no credential, and that the batch's state is an honest description of what
-the process did.
+Per-asset upload *results* are not interpreted here: after the attempt ends the
+report is handed to :class:`~photo_pipeline.services.upload_reports.
+UploadReportService` (US05-03), which classifies every file. The batch ``state``
+stays an honest description of what the *process* did; ``outcome_state`` says
+whether the report's evidence is complete enough to trust.
 """
 
 from __future__ import annotations
@@ -38,6 +39,7 @@ from photo_pipeline.config import Config
 from photo_pipeline.integrations import immich_go
 from photo_pipeline.models import UploadBatch, UploadItem
 from photo_pipeline.services.hashing import sha1_file
+from photo_pipeline.services.upload_reports import UploadReportService
 from photo_pipeline.services.uploads import UploadService
 
 
@@ -238,6 +240,13 @@ class UploadBatchService:
         self._finish(batch_id, token=token, state=state, error=error, result=result)
         if item_state:
             self._set_items(batch_id, item_state)
+        # The report is the only evidence of what happened per file, so it is read
+        # while it is fresh (US05-03). A parse failure must not lose the batch
+        # outcome that was just recorded; the items simply stay unknown.
+        try:
+            UploadReportService(self._session_factory).ingest(batch_id)
+        except OSError:
+            pass
         return self.get(batch_id)
 
     def cancel(self, batch_id: str) -> dict:
@@ -409,6 +418,14 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
         "exit_code": row.exit_code,
         "error_code": row.error_code,
         "error_message": row.error_message,
+        # Parsed report evidence (US05-03): what the uploader said per file, and
+        # whether that evidence is complete enough to be trusted.
+        "parser": row.parser,
+        "parser_version": row.parser_version,
+        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
+        "outcome_state": row.outcome_state,
+        "outcome_counts": json.loads(row.outcome_counts) if row.outcome_counts else None,
+        "report_counts": json.loads(row.report_counts) if row.report_counts else None,
         "started_at": row.started_at.isoformat() if row.started_at else None,
         "finished_at": row.finished_at.isoformat() if row.finished_at else None,
         "items": [
@@ -418,6 +435,9 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
                 "sha256": item.sha256,
                 "sha1": item.sha1,
                 "state": item.state,
+                "outcome": item.outcome,
+                "evidence": item.evidence,
+                "outcome_at": item.outcome_at.isoformat() if item.outcome_at else None,
             }
             for item in items
         ],

photo_pipeline/services/upload_reports.py

@@ -0,0 +1,140 @@
+"""UploadReportService — durable per-item upload outcomes (US05-03).
+
+US05-02 proves the uploader ran and kept a bounded, credential-free report. This
+service is the step that reads that report and answers the operator's actual
+question — *what happened to each photo?* — without ever guessing in the app's
+favour:
+
+- **an item is ``unknown`` until the report says otherwise.** A file the report
+  never mentions, a report from an unpinned uploader version, and a line the
+  grammar does not recognise all leave the item ``unknown``. Exit code 0 is
+  evidence the *process* ended well, never evidence that a particular file reached
+  Immich (concept §15 "wrong upload state").
+- **counts are reconciled, not trusted.** The uploader's own summary is stored
+  next to the totals derived from the items. A disagreement makes the batch
+  ``requires_verification`` even when every line parsed.
+- **reprocessing is idempotent.** Outcomes are keyed by ``(batch_id, asset_id)``
+  and rewritten in place, so parsing the same report again — after a restart, or
+  because the operator asked — converges on the same rows and the same counts.
+
+The batch's own ``state`` still describes the process (US05-02's contract);
+``outcome_state`` describes the evidence. Turning ``requires_verification`` into
+verification, retry, and manual resolution is US05-04.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.integrations import immich_go_report as report_parser
+from photo_pipeline.models import UploadBatch, UploadItem
+
+VERIFIED = "verified"
+REQUIRES_VERIFICATION = "requires_verification"
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _key(path: str) -> str:
+    return os.path.normpath(path)
+
+
+class UploadReportService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    def ingest(self, batch_id: str) -> dict:
+        """Parse this batch's report and persist an outcome for every item.
+
+        Returns ``{"parser", "parser_version", "outcome_state", "counts",
+        "report_counts", "unmatched", "unparsed"}``. Safe to call repeatedly.
+        """
+        with self._session_factory() as session:
+            batch = session.get(UploadBatch, batch_id)
+            if batch is None:
+                raise KeyError(f"unknown upload batch {batch_id!r}")
+            items = list(session.scalars(select(UploadItem).where(UploadItem.batch_id == batch_id)))
+            parsed = report_parser.parse(_read_report(batch.report_path), batch.uploader_version)
+
+            by_path = {_key(entry["path"]): entry for entry in parsed["entries"]}
+            # Basenames are the fallback: an uploader may log a path relative to the
+            # folder it was given. Ambiguous basenames are dropped rather than
+            # guessed at.
+            by_name: dict[str, dict] = {}
+            for key, entry in by_path.items():
+                name = os.path.basename(key)
+                by_name[name] = None if name in by_name else entry
+
+            matched: set[str] = set()
+            counts = {outcome: 0 for outcome in report_parser.OUTCOMES}
+            for item in items:
+                key = _key(item.path)
+                entry = by_path.get(key) or by_name.get(os.path.basename(key))
+                if entry is not None:
+                    matched.add(_key(entry["path"]))
+                item.outcome = entry["outcome"] if entry else report_parser.UNKNOWN
+                item.evidence = entry["evidence"] if entry else None
+                item.outcome_at = _now()
+                counts[item.outcome] += 1
+
+            unmatched = sorted(set(by_path) - matched)
+            reconciled = _reconciles(parsed["counts"], counts)
+            state = (
+                VERIFIED
+                if (
+                    parsed["supported"]
+                    and items
+                    and not counts[report_parser.UNKNOWN]
+                    and not unmatched
+                    and not parsed["entries_truncated"]
+                    and reconciled
+                )
+                else REQUIRES_VERIFICATION
+            )
+
+            batch.parser = parsed["parser"]
+            batch.parser_version = parsed["parser_version"]
+            batch.parsed_at = _now()
+            batch.outcome_state = state
+            batch.outcome_counts = json.dumps(counts, sort_keys=True)
+            batch.report_counts = (
+                json.dumps(parsed["counts"], sort_keys=True) if parsed["counts"] else None
+            )
+            session.commit()
+
+            return {
+                "parser": parsed["parser"],
+                "parser_version": parsed["parser_version"],
+                "outcome_state": state,
+                "counts": counts,
+                "report_counts": parsed["counts"],
+                "unmatched": unmatched,
+                "unparsed": parsed["unparsed"],
+            }
+
+
+def _read_report(report_path: str | None) -> str:
+    """The raw report, or ``""`` when the attempt never produced one."""
+    if not report_path:
+        return ""
+    path = Path(report_path)
+    return path.read_text(errors="replace") if path.exists() else ""
+
+
+def _reconciles(report_counts: dict | None, derived: dict) -> bool:
+    """Whether the uploader's own totals agree with the per-item outcomes.
+
+    No summary is not a disagreement — most of the uncertainty it would catch is
+    already caught by unmatched entries and unknown items.
+    """
+    if not report_counts:
+        return True
+    return all(derived.get(outcome) == total for outcome, total in report_counts.items())