diff --git a/backend/service/product_archive.py b/backend/service/product_archive.py new file mode 100644 index 0000000..696b4ab --- /dev/null +++ b/backend/service/product_archive.py @@ -0,0 +1,74 @@ +"""Helpers for locating and safely archiving word-cloud layout databases.""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + + +@dataclass(frozen=True) +class WordcloudSource: + asset_id: str + source_job_id: str + + +def find_visible_wordcloud_sources( + document: dict[str, Any], load_asset_meta: Callable[[str], dict[str, Any]] +) -> list[WordcloudSource]: + """Return unique word-cloud sources referenced by visible sticker elements.""" + visible = { + str(layer.get("id")): layer.get("visible") is not False + for layer in document.get("layers") or [] + if isinstance(layer, dict) + } + seen: set[str] = set() + result: list[WordcloudSource] = [] + for element in document.get("elements") or []: + if not isinstance(element, dict) or element.get("type") != "sticker": + continue + if visible and visible.get(str(element.get("layerId")), True) is False: + continue + asset_id = str(element.get("assetId") or "") + meta = load_asset_meta(asset_id) + job_id = str(meta.get("job_id") or "") + if meta.get("type") == "wordcloud" and job_id and job_id not in seen: + seen.add(job_id) + result.append(WordcloudSource(asset_id, job_id)) + return result + + +def validate_word_locations_db(db_path: Path) -> None: + """Ensure a read-only SQLite database contains the expected layout table.""" + if not db_path.is_file(): + raise ValueError(f"word locations database does not exist: {db_path}") + try: + with sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True) as connection: + exists = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'word_locations'" + ).fetchone() + except sqlite3.Error as exc: + raise ValueError(f"invalid word locations database: {db_path}") from exc + if exists is None: + raise ValueError(f"word locations database is missing word_locations: {db_path}") + + +def copy_word_locations_snapshot(source: Path, destination: Path) -> str: + """Copy a validated layout database atomically and return its SHA-256 checksum.""" + validate_word_locations_db(source) + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_name(f"{destination.name}.tmp") + digest = hashlib.sha256() + try: + with source.open("rb") as source_file, temporary.open("wb") as temporary_file: + while chunk := source_file.read(1024 * 1024): + temporary_file.write(chunk) + digest.update(chunk) + validate_word_locations_db(temporary) + temporary.replace(destination) + except Exception: + temporary.unlink(missing_ok=True) + raise + return digest.hexdigest() diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py new file mode 100644 index 0000000..78850ac --- /dev/null +++ b/backend/tests/test_product_archive.py @@ -0,0 +1,86 @@ +import hashlib +import sqlite3 +import sys +from pathlib import Path + +import pytest + + +BACKEND_DIR = Path(__file__).resolve().parents[1] +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +from service.product_archive import ( # noqa: E402 + WordcloudSource, + copy_word_locations_snapshot, + find_visible_wordcloud_sources, + validate_word_locations_db, +) + + +def test_scanner_keeps_only_visible_wordcloud_assets_and_deduplicates(): + document = { + "layers": [{"id": "shown", "visible": True}, {"id": "hidden", "visible": False}], + "elements": [ + {"type": "sticker", "assetId": "wc-a", "layerId": "shown"}, + {"type": "sticker", "assetId": "wc-a", "layerId": "shown"}, + {"type": "sticker", "assetId": "wc-b", "layerId": "hidden"}, + {"type": "sticker", "assetId": "photo", "layerId": "shown"}, + ], + } + assets = { + "wc-a": {"type": "wordcloud", "job_id": "a" * 32}, + "wc-b": {"type": "wordcloud", "job_id": "b" * 32}, + "photo": {"type": "upload", "job_id": ""}, + } + + assert find_visible_wordcloud_sources(document, assets.__getitem__) == [ + WordcloudSource("wc-a", "a" * 32) + ] + + +def test_scanner_treats_elements_as_visible_without_layers(): + document = {"elements": [{"type": "sticker", "assetId": "wc-a"}]} + assets = {"wc-a": {"type": "wordcloud", "job_id": "a" * 32}} + + assert find_visible_wordcloud_sources(document, assets.__getitem__) == [ + WordcloudSource("wc-a", "a" * 32) + ] + + +def test_scanner_returns_empty_list_without_elements(): + assert find_visible_wordcloud_sources({"layers": []}, lambda _: {}) == [] + + +def test_scanner_skips_wordcloud_without_job_id(): + document = {"elements": [{"type": "sticker", "assetId": "wc-a"}]} + assets = {"wc-a": {"type": "wordcloud"}} + + assert find_visible_wordcloud_sources(document, assets.__getitem__) == [] + + +def test_snapshot_rejects_non_sqlite_file_without_creating_destination(tmp_path): + source = tmp_path / "not-a-database.sqlite" + destination = tmp_path / "archive" / "word_locations.sqlite" + source.write_text("not a SQLite database", encoding="utf-8") + + with pytest.raises(ValueError): + copy_word_locations_snapshot(source, destination) + + assert not destination.exists() + assert not destination.with_name(f"{destination.name}.tmp").exists() + + +def test_snapshot_validates_table_copies_atomically_and_returns_checksum(tmp_path): + source = tmp_path / "word_locations.sqlite" + destination = tmp_path / "archive" / "word_locations.sqlite" + with sqlite3.connect(source) as connection: + connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)") + connection.execute("INSERT INTO word_locations (name) VALUES ('hello')") + + validate_word_locations_db(source) + checksum = copy_word_locations_snapshot(source, destination) + + assert checksum == hashlib.sha256(destination.read_bytes()).hexdigest() + with sqlite3.connect(destination) as connection: + assert connection.execute("SELECT name FROM word_locations").fetchone() == ("hello",)