119 lines
3.3 KiB
Python
119 lines
3.3 KiB
Python
"""Result objects describing what a download run did."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
|
|
|
|
class FileStatus(str, Enum):
|
|
"""Outcome of a single file within a pack."""
|
|
|
|
DOWNLOADED = "downloaded"
|
|
SKIPPED_EXISTING = "skipped_existing"
|
|
FAILED = "failed"
|
|
PLANNED = "planned" # --dry-run only
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileOutcome:
|
|
"""What happened to one sticker file."""
|
|
|
|
path: Path
|
|
status: FileStatus
|
|
size: int = 0
|
|
error: str | None = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.path.name
|
|
|
|
|
|
@dataclass
|
|
class PackResult:
|
|
"""Aggregated outcome for one sticker pack."""
|
|
|
|
reference: str
|
|
name: str | None = None
|
|
title: str | None = None
|
|
directory: Path | None = None
|
|
total_stickers: int = 0
|
|
outcomes: list[FileOutcome] = field(default_factory=list)
|
|
error: str | None = None
|
|
|
|
@property
|
|
def downloaded(self) -> int:
|
|
return sum(1 for o in self.outcomes if o.status is FileStatus.DOWNLOADED)
|
|
|
|
@property
|
|
def skipped(self) -> int:
|
|
return sum(1 for o in self.outcomes if o.status is FileStatus.SKIPPED_EXISTING)
|
|
|
|
@property
|
|
def planned(self) -> int:
|
|
return sum(1 for o in self.outcomes if o.status is FileStatus.PLANNED)
|
|
|
|
@property
|
|
def failed(self) -> int:
|
|
return sum(1 for o in self.outcomes if o.status is FileStatus.FAILED)
|
|
|
|
@property
|
|
def bytes_written(self) -> int:
|
|
return sum(o.size for o in self.outcomes if o.status is FileStatus.DOWNLOADED)
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
"""True when the pack was processed without a fatal or per-file error."""
|
|
return self.error is None and self.failed == 0
|
|
|
|
def summary(self) -> str:
|
|
if self.error is not None:
|
|
return f"failed: {self.error}"
|
|
parts = [f"{self.downloaded} downloaded"]
|
|
if self.planned:
|
|
parts.append(f"{self.planned} planned")
|
|
if self.skipped:
|
|
parts.append(f"{self.skipped} already present")
|
|
if self.failed:
|
|
parts.append(f"{self.failed} failed")
|
|
return ", ".join(parts)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunTotals:
|
|
"""Totals across every pack in a run."""
|
|
|
|
packs: int
|
|
packs_ok: int
|
|
packs_failed: int
|
|
downloaded: int
|
|
skipped: int
|
|
planned: int
|
|
failed: int
|
|
bytes_written: int
|
|
|
|
@classmethod
|
|
def from_results(cls, results: list[PackResult]) -> RunTotals:
|
|
return cls(
|
|
packs=len(results),
|
|
packs_ok=sum(1 for r in results if r.ok),
|
|
packs_failed=sum(1 for r in results if not r.ok),
|
|
downloaded=sum(r.downloaded for r in results),
|
|
skipped=sum(r.skipped for r in results),
|
|
planned=sum(r.planned for r in results),
|
|
failed=sum(r.failed for r in results),
|
|
bytes_written=sum(r.bytes_written for r in results),
|
|
)
|
|
|
|
|
|
def format_size(num_bytes: int) -> str:
|
|
"""Human-readable byte count, e.g. ``1.4 MiB``."""
|
|
size = float(num_bytes)
|
|
for unit in ("B", "KiB", "MiB", "GiB"):
|
|
if size < 1024 or unit == "GiB":
|
|
precision = 0 if unit == "B" else 1
|
|
return f"{size:.{precision}f} {unit}"
|
|
size /= 1024
|
|
raise AssertionError("unreachable") # pragma: no cover
|