# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """What an import did, in buckets that have to add up. Split from pack_import.py at the shared 250-line cap, along the seam between deciding what happens to a file and saying what happened. The arithmetic is the point: every source file lands in exactly one bucket, and the printed total is computed from the decisions rather than tracked alongside them, so a report that does not add up is not expressible. Every bucket prints even at zero. A suppressed zero is a silently missing category -- "0 could not be stripped" is how an operator sees that the video pipeline is still working. """ from __future__ import annotations from dataclasses import dataclass from enum import StrEnum from typing import TYPE_CHECKING, Final from signal_bot.pack import MEDIA_DIR if TYPE_CHECKING: from pathlib import Path # Imported for typing only: pack_import imports this module at run time, # and a real import here would close the loop. from signal_bot.pack_import import Decision SKIPPED_DIR: Final = "_skipped" class Outcome(StrEnum): """Why one source file did or did not end up in the album.""" NOT_MEDIA = "not-media" UNSTRIPPABLE = "unstrippable" ALREADY_IN_PACK = "already-in-pack" EXACT_DUP = "exact-dup" IMPORTED = "imported" # Printed in this order, so the report reads as a funnel. ORDER: Final = ( Outcome.NOT_MEDIA, Outcome.UNSTRIPPABLE, Outcome.ALREADY_IN_PACK, Outcome.EXACT_DUP, Outcome.IMPORTED, ) LABELS: Final[dict[Outcome, str]] = { Outcome.NOT_MEDIA: "not media", Outcome.UNSTRIPPABLE: "could not be stripped", Outcome.ALREADY_IN_PACK: "already in pack", Outcome.EXACT_DUP: "duplicate in batch", Outcome.IMPORTED: "imported", } @dataclass(frozen=True, slots=True) class ImportReport: """Every source file, bucketed, plus what it does to the album.""" source: Path directory: Path decisions: tuple[Decision, ...] before: int @property def counts(self) -> dict[Outcome, int]: """How many files fell into each bucket, every bucket present.""" tally = dict.fromkeys(ORDER, 0) for decision in self.decisions: tally[decision.outcome] += 1 return tally @property def imported(self) -> tuple[Decision, ...]: """The decisions that put a new file in the album.""" return tuple(d for d in self.decisions if d.outcome is Outcome.IMPORTED) @property def after(self) -> int: """How many files media/ holds once this import is applied.""" return self.before + len(self.imported) def lines(self) -> list[str]: """The report, with a total that has to add up to what was seen.""" counts = self.counts lines = [f"== {self.source} -> {self.directory}", ""] lines.append(f"{len(self.decisions):6} files seen") for outcome in ORDER: note = "" if outcome is Outcome.IMPORTED else f" -> {SKIPPED_DIR}/{outcome}/" lines.append(f"{counts[outcome]:6} {LABELS[outcome]:<21}{note}") lines.append(" ---") lines.append( f"{sum(counts.values()):6} accounted for" f" {self.before} -> {self.after} in {MEDIA_DIR}/" ) return lines