147 lines
5.8 KiB
Python
147 lines
5.8 KiB
Python
"""Transactional product-version archives for generated word clouds."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import shutil
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from PIL import Image, UnidentifiedImageError
|
|
|
|
from .product_archive import (
|
|
WordcloudSource,
|
|
copy_word_locations_snapshot,
|
|
find_visible_wordcloud_sources,
|
|
)
|
|
from .product_archive_store import ProductArchiveStore
|
|
from .schemas import ProductVersionRecord
|
|
|
|
|
|
class ProductPendingCleanupError(ValueError):
|
|
"""Raised when a pending or failed-cleanup product is changed before recovery."""
|
|
|
|
|
|
class ProductArchiveService:
|
|
"""Materialize a self-contained design preview and word-location snapshots."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: ProductArchiveStore,
|
|
storage: Any,
|
|
load_asset_meta: Callable[[str], dict[str, Any]],
|
|
resolve_job_status: Callable[[str], Any],
|
|
) -> None:
|
|
self.store = store
|
|
self.storage = storage
|
|
self.load_asset_meta = load_asset_meta
|
|
self.resolve_job_status = resolve_job_status
|
|
self.archive_root = store.root.parent
|
|
|
|
@staticmethod
|
|
def _validate_preview(preview_bytes: bytes) -> None:
|
|
if not preview_bytes.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
raise ValueError("preview must be a PNG image")
|
|
try:
|
|
with Image.open(io.BytesIO(preview_bytes)) as image:
|
|
image.verify()
|
|
if image.format != "PNG":
|
|
raise ValueError("preview must be a PNG image")
|
|
except (UnidentifiedImageError, OSError, ValueError) as exc:
|
|
raise ValueError("preview must be a valid PNG image") from exc
|
|
|
|
def _source_db_paths(self, sources: list[WordcloudSource]) -> list[tuple[WordcloudSource, Path]]:
|
|
resolved: list[tuple[WordcloudSource, Path]] = []
|
|
for source in sources:
|
|
status = self.resolve_job_status(source.source_job_id)
|
|
if status is None or getattr(status, "status", "") != "success":
|
|
raise ValueError(f"wordcloud source job is not successful: {source.source_job_id}")
|
|
artifacts = getattr(status, "artifacts", {}) or {}
|
|
raw_path = artifacts.get("db", "")
|
|
if not raw_path:
|
|
raise ValueError(f"wordcloud source database is unavailable: {source.source_job_id}")
|
|
db_path = Path(raw_path)
|
|
if not db_path.is_file():
|
|
raise ValueError(f"wordcloud source database is unavailable: {source.source_job_id}")
|
|
resolved.append((source, db_path))
|
|
return resolved
|
|
|
|
@staticmethod
|
|
def _move_staging(staging_dir: Path, final_dir: Path) -> None:
|
|
staging_dir.replace(final_dir)
|
|
|
|
def archive_version(
|
|
self,
|
|
product_id: str,
|
|
document: dict[str, Any],
|
|
preview_bytes: bytes,
|
|
now: datetime | None = None,
|
|
) -> ProductVersionRecord:
|
|
"""Archive visible, successful word clouds without trusting client job IDs."""
|
|
if not isinstance(document, dict):
|
|
raise ValueError("document must be a JSON object")
|
|
product = self.store.get_product(product_id)
|
|
if product.status in ("pending_cleanup", "failed_cleanup"):
|
|
raise ProductPendingCleanupError("product is pending cleanup")
|
|
if product.status != "active":
|
|
raise ValueError("product is not active")
|
|
|
|
self._validate_preview(preview_bytes)
|
|
sources = find_visible_wordcloud_sources(document, self.load_asset_meta)
|
|
source_dbs = self._source_db_paths(sources)
|
|
|
|
product_dir = self.archive_root / product_id
|
|
staging_dir = product_dir / f".{uuid.uuid4().hex}.staging"
|
|
final_dir: Path | None = None
|
|
try:
|
|
staging_dir.mkdir(parents=True, exist_ok=False)
|
|
preview_path = staging_dir / "design-preview.png"
|
|
preview_path.write_bytes(preview_bytes)
|
|
|
|
snapshots: list[tuple[WordcloudSource, Path, str]] = []
|
|
for index, (source, source_db) in enumerate(source_dbs, start=1):
|
|
snapshot_path = (
|
|
staging_dir
|
|
/ "wordclouds"
|
|
/ f"{index:02d}-{source.source_job_id}"
|
|
/ "word_locations.sqlite"
|
|
)
|
|
checksum = copy_word_locations_snapshot(source_db, snapshot_path)
|
|
snapshots.append((source, snapshot_path, checksum))
|
|
|
|
version_id = f"ver_{uuid.uuid4().hex}"
|
|
target_dir = product_dir / version_id
|
|
self._move_staging(staging_dir, target_dir)
|
|
final_dir = target_dir
|
|
preview = final_dir / "design-preview.png"
|
|
archive_rows = [
|
|
{
|
|
"archive_path": str(final_dir / snapshot_path.relative_to(staging_dir)),
|
|
"source_job_id": source.source_job_id,
|
|
"source_asset_id": source.asset_id,
|
|
"db_checksum": checksum,
|
|
}
|
|
for source, snapshot_path, checksum in snapshots
|
|
]
|
|
version, _, _ = self.store.write_archive_version(
|
|
product_id,
|
|
version_id=version_id,
|
|
metadata={
|
|
"document": document,
|
|
"wordcloud_count": len(snapshots),
|
|
"wordcloud_sources": [
|
|
{"asset_id": source.asset_id, "source_job_id": source.source_job_id}
|
|
for source, _, _ in snapshots
|
|
],
|
|
},
|
|
preview_path=str(preview),
|
|
archives=archive_rows,
|
|
now=now,
|
|
)
|
|
return version
|
|
except Exception:
|
|
shutil.rmtree(final_dir or staging_dir, ignore_errors=True)
|
|
raise
|