76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""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 "")
|
|
asset_type = str(meta.get("type") or "")
|
|
if asset_type in {"wordcloud", "sticker"} 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()
|