From bf24d20be9e349a75eefdc472d705bc9ba728408 Mon Sep 17 00:00:00 2001 From: obroccolio Date: Sat, 12 Sep 2026 13:01:35 +0800 Subject: [PATCH] feat: add product archive metadata store --- backend/service/product_archive_store.py | 294 ++++++++++++++++++++ backend/service/schemas.py | 69 ++++- backend/tests/test_product_archive_store.py | 115 ++++++++ 3 files changed, 477 insertions(+), 1 deletion(-) create mode 100644 backend/service/product_archive_store.py create mode 100644 backend/tests/test_product_archive_store.py diff --git a/backend/service/product_archive_store.py b/backend/service/product_archive_store.py new file mode 100644 index 0000000..108fab3 --- /dev/null +++ b/backend/service/product_archive_store.py @@ -0,0 +1,294 @@ +"""Durable SQLite metadata store for archived product word clouds.""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from .schemas import ( + ProductImageRecord, + ProductInput, + ProductRecord, + ProductVersionRecord, + ProductWordcloudArchiveRecord, +) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class ProductArchiveStore: + """Owns durable product metadata and logical cleanup state.""" + + def __init__(self, root: Path) -> None: + self.root = root + self.root.mkdir(parents=True, exist_ok=True) + self.db_path = self.root / "product_archive.db" + self._lock = threading.Lock() + self._init_db() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(str(self.db_path), check_same_thread=False) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + return conn + + def _init_db(self) -> None: + with self._lock, self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS products ( + product_id TEXT PRIMARY KEY, + source TEXT NOT NULL, + external_product_id TEXT, + name TEXT NOT NULL, + sku TEXT NOT NULL DEFAULT '', + specification TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'active', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + purge_after TEXT + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_products_source_external_id + ON products(source, external_product_id); + CREATE INDEX IF NOT EXISTS idx_products_name ON products(name); + CREATE INDEX IF NOT EXISTS idx_products_purge_after ON products(purge_after); + + CREATE TABLE IF NOT EXISTS product_versions ( + version_id TEXT PRIMARY KEY, + product_id TEXT NOT NULL, + version TEXT NOT NULL DEFAULT '', + metadata TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_product_versions_product_id + ON product_versions(product_id); + + CREATE TABLE IF NOT EXISTS product_images ( + image_id TEXT PRIMARY KEY, + product_id TEXT NOT NULL, + version_id TEXT NOT NULL, + image_path TEXT NOT NULL DEFAULT '', + image_type TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_product_images_product_id + ON product_images(product_id); + CREATE INDEX IF NOT EXISTS idx_product_images_version_id + ON product_images(version_id); + + CREATE TABLE IF NOT EXISTS product_wordcloud_archives ( + archive_id TEXT PRIMARY KEY, + product_id TEXT NOT NULL, + version_id TEXT NOT NULL, + archive_path TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_product_wordcloud_archives_product_id + ON product_wordcloud_archives(product_id); + CREATE INDEX IF NOT EXISTS idx_product_wordcloud_archives_version_id + ON product_wordcloud_archives(version_id); + + CREATE TABLE IF NOT EXISTS cleanup_records ( + cleanup_id TEXT PRIMARY KEY, + product_id TEXT NOT NULL, + status TEXT NOT NULL, + purge_after TEXT, + created_at TEXT NOT NULL, + completed_at TEXT + ); + CREATE INDEX IF NOT EXISTS idx_cleanup_records_product_id + ON cleanup_records(product_id); + CREATE INDEX IF NOT EXISTS idx_cleanup_records_purge_after + ON cleanup_records(purge_after); + """ + ) + + def _execute(self, sql: str, parameters: tuple[Any, ...] = ()) -> None: + with self._lock, self._connect() as conn: + conn.execute(sql, parameters) + + def _fetchone(self, sql: str, parameters: tuple[Any, ...] = ()) -> sqlite3.Row | None: + with self._lock, self._connect() as conn: + return conn.execute(sql, parameters).fetchone() + + def _fetchall(self, sql: str, parameters: tuple[Any, ...] = ()) -> list[sqlite3.Row]: + with self._lock, self._connect() as conn: + return conn.execute(sql, parameters).fetchall() + + @staticmethod + def _product_from_row(row: sqlite3.Row) -> ProductRecord: + return ProductRecord( + product_id=row["product_id"], source=row["source"], + external_product_id=row["external_product_id"], name=row["name"], + sku=row["sku"], specification=row["specification"], status=row["status"], + created_at=datetime.fromisoformat(row["created_at"]), + updated_at=datetime.fromisoformat(row["updated_at"]), + purge_after=datetime.fromisoformat(row["purge_after"]) if row["purge_after"] else None, + ) + + def upsert_product(self, product: ProductInput) -> ProductRecord: + if not product.name.strip(): + raise ValueError("product name must not be blank") + if product.source == "external" and not (product.external_product_id or "").strip(): + raise ValueError("external products require external_product_id") + + now = _now() + existing = None + if product.source == "external": + existing = self._fetchone( + "SELECT * FROM products WHERE source = ? AND external_product_id = ?", + (product.source, product.external_product_id), + ) + if existing: + self._execute( + """UPDATE products SET name = ?, sku = ?, specification = ?, updated_at = ? + WHERE product_id = ?""", + (product.name, product.sku, product.specification, now.isoformat(), existing["product_id"]), + ) + return self.get_product(existing["product_id"]) + + product_id = f"prod_{uuid.uuid4().hex}" + self._execute( + """INSERT INTO products ( + product_id, source, external_product_id, name, sku, specification, + status, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, 'active', ?, ?)""", + (product_id, product.source, product.external_product_id, product.name, + product.sku, product.specification, now.isoformat(), now.isoformat()), + ) + return self.get_product(product_id) + + def get_product(self, product_id: str) -> ProductRecord: + row = self._fetchone("SELECT * FROM products WHERE product_id = ?", (product_id,)) + if row is None: + raise ValueError(f"unknown product_id: {product_id}") + return self._product_from_row(row) + + def list_products(self, query: str = "") -> list[ProductRecord]: + if query: + rows = self._fetchall( + "SELECT * FROM products WHERE name LIKE ? ORDER BY updated_at DESC, product_id", + (f"%{query}%",), + ) + else: + rows = self._fetchall("SELECT * FROM products ORDER BY updated_at DESC, product_id") + return [self._product_from_row(row) for row in rows] + + def create_version( + self, product_id: str, version: str = "", metadata: dict[str, Any] | None = None, + now: datetime | None = None, + ) -> ProductVersionRecord: + self.get_product(product_id) + created_at = now or _now() + record = ProductVersionRecord( + version_id=f"ver_{uuid.uuid4().hex}", product_id=product_id, version=version, + metadata=metadata or {}, created_at=created_at, + ) + self._execute( + "INSERT INTO product_versions (version_id, product_id, version, metadata, created_at) VALUES (?, ?, ?, ?, ?)", + (record.version_id, record.product_id, record.version, json.dumps(record.metadata, ensure_ascii=False), record.created_at.isoformat()), + ) + return record + + def add_image( + self, product_id: str, version_id: str, image_path: str = "", image_type: str = "", + now: datetime | None = None, + ) -> ProductImageRecord: + self._require_version(product_id, version_id) + record = ProductImageRecord( + image_id=f"img_{uuid.uuid4().hex}", product_id=product_id, version_id=version_id, + image_path=image_path, image_type=image_type, created_at=now or _now(), + ) + self._execute( + """INSERT INTO product_images (image_id, product_id, version_id, image_path, image_type, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (record.image_id, record.product_id, record.version_id, record.image_path, record.image_type, record.created_at.isoformat()), + ) + return record + + def add_wordcloud_archive( + self, product_id: str, version_id: str, archive_path: str = "", now: datetime | None = None, + ) -> ProductWordcloudArchiveRecord: + self._require_version(product_id, version_id) + record = ProductWordcloudArchiveRecord( + archive_id=f"wca_{uuid.uuid4().hex}", product_id=product_id, version_id=version_id, + archive_path=archive_path, created_at=now or _now(), + ) + self._execute( + """INSERT INTO product_wordcloud_archives (archive_id, product_id, version_id, archive_path, created_at) + VALUES (?, ?, ?, ?, ?)""", + (record.archive_id, record.product_id, record.version_id, record.archive_path, record.created_at.isoformat()), + ) + return record + + def _require_version(self, product_id: str, version_id: str) -> None: + row = self._fetchone( + "SELECT version_id FROM product_versions WHERE product_id = ? AND version_id = ?", + (product_id, version_id), + ) + if row is None: + raise ValueError(f"unknown version_id for product: {version_id}") + + def mark_pending_cleanup(self, product_id: str, now: datetime) -> ProductRecord: + self.get_product(product_id) + purge_after = now + timedelta(days=30) + self._execute( + "UPDATE products SET status = ?, purge_after = ?, updated_at = ? WHERE product_id = ?", + ("pending_cleanup", purge_after.isoformat(), now.isoformat(), product_id), + ) + self._execute( + """INSERT INTO cleanup_records (cleanup_id, product_id, status, purge_after, created_at) + VALUES (?, ?, ?, ?, ?)""", + (f"cln_{uuid.uuid4().hex}", product_id, "pending_cleanup", purge_after.isoformat(), now.isoformat()), + ) + return self.get_product(product_id) + + def restore_product(self, product_id: str, now: datetime | None = None) -> ProductRecord: + self.get_product(product_id) + restored_at = now or _now() + self._execute( + "UPDATE products SET status = ?, purge_after = NULL, updated_at = ? WHERE product_id = ?", + ("active", restored_at.isoformat(), product_id), + ) + self._execute( + """UPDATE cleanup_records SET status = ?, completed_at = ? + WHERE product_id = ? AND status = ?""", + ("restored", restored_at.isoformat(), product_id, "pending_cleanup"), + ) + return self.get_product(product_id) + + def due_product_cleanups(self, now: datetime) -> list[ProductRecord]: + rows = self._fetchall( + """SELECT * FROM products WHERE status = ? AND purge_after IS NOT NULL AND purge_after <= ? + ORDER BY purge_after, product_id""", + ("pending_cleanup", now.isoformat()), + ) + return [self._product_from_row(row) for row in rows] + + def purge_product(self, product_id: str, now: datetime | None = None) -> ProductRecord: + purged_at = now or _now() + product = self.get_product(product_id) + if ( + product.status != "pending_cleanup" + or product.purge_after is None + or product.purge_after > purged_at + ): + raise ValueError("product is not due for purge") + self._execute( + "UPDATE products SET status = ?, updated_at = ? WHERE product_id = ?", + ("purged", purged_at.isoformat(), product_id), + ) + self._execute( + """UPDATE cleanup_records SET status = ?, completed_at = ? + WHERE product_id = ? AND status = ?""", + ("purged", purged_at.isoformat(), product_id, "pending_cleanup"), + ) + return self.get_product(product_id) diff --git a/backend/service/schemas.py b/backend/service/schemas.py index ce954cd..329d65b 100644 --- a/backend/service/schemas.py +++ b/backend/service/schemas.py @@ -2,11 +2,78 @@ from __future__ import annotations from datetime import datetime from pathlib import Path -from typing import Literal +from typing import Any, Literal from pydantic import BaseModel, Field +class ProductInput(BaseModel): + source: Literal["manual", "external"] + external_product_id: str | None = None + name: str + sku: str = "" + specification: str = "" + + def __init__( + self, + source: Literal["manual", "external"] | None = None, + external_product_id: str | None = None, + name: str | None = None, + sku: str = "", + specification: str = "", + **data: Any, + ) -> None: + if source is not None: + data["source"] = source + if external_product_id is not None or "external_product_id" not in data: + data["external_product_id"] = external_product_id + if name is not None: + data["name"] = name + if sku or "sku" not in data: + data["sku"] = sku + if specification or "specification" not in data: + data["specification"] = specification + super().__init__(**data) + + +class ProductRecord(BaseModel): + product_id: str + source: Literal["manual", "external"] + external_product_id: str | None = None + name: str + sku: str = "" + specification: str = "" + status: Literal["active", "pending_cleanup", "purged"] = "active" + created_at: datetime + updated_at: datetime + purge_after: datetime | None = None + + +class ProductVersionRecord(BaseModel): + version_id: str + product_id: str + version: str = "" + metadata: dict[str, Any] = Field(default_factory=dict) + created_at: datetime + + +class ProductImageRecord(BaseModel): + image_id: str + product_id: str + version_id: str + image_path: str = "" + image_type: str = "" + created_at: datetime + + +class ProductWordcloudArchiveRecord(BaseModel): + archive_id: str + product_id: str + version_id: str + archive_path: str = "" + created_at: datetime + + class JobCreateResponse(BaseModel): job_id: str diff --git a/backend/tests/test_product_archive_store.py b/backend/tests/test_product_archive_store.py new file mode 100644 index 0000000..7494d5d --- /dev/null +++ b/backend/tests/test_product_archive_store.py @@ -0,0 +1,115 @@ +import sys +from datetime import datetime, timedelta, timezone +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_store import ProductArchiveStore # noqa: E402 +from service.schemas import ProductInput # noqa: E402 + + +NOW = datetime(2026, 9, 12, 8, 0, tzinfo=timezone.utc) + + +def test_external_product_id_updates_one_product(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + + first = store.upsert_product( + ProductInput("external", "sku-42", "笔盒", "B-42", "黄色") + ) + second = store.upsert_product( + ProductInput("external", "sku-42", "笔盒新版", "B-42", "黄色") + ) + + assert first.product_id == second.product_id + assert store.list_products(query="新版")[0].name == "笔盒新版" + + +def test_manual_product_has_internal_id_and_soft_delete_window(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + + product = store.upsert_product( + ProductInput("manual", None, "校长笔盒", "", "166 × 47 mm") + ) + store.mark_pending_cleanup(product.product_id, now=NOW) + + assert product.product_id.startswith("prod_") + assert store.get_product(product.product_id).status == "pending_cleanup" + assert store.get_product(product.product_id).purge_after == NOW + timedelta(days=30) + + +@pytest.mark.parametrize( + "product_input", + [ + ProductInput("manual", None, "", "B-42", "黄色"), + ProductInput("external", None, "笔盒", "B-42", "黄色"), + ProductInput("external", "", "笔盒", "B-42", "黄色"), + ], +) +def test_upsert_rejects_invalid_product_inputs(tmp_path, product_input): + store = ProductArchiveStore(tmp_path / "service_products") + + with pytest.raises(ValueError): + store.upsert_product(product_input) + + +def test_unknown_product_operations_are_rejected(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + + with pytest.raises(ValueError): + store.get_product("prod_missing") + with pytest.raises(ValueError): + store.mark_pending_cleanup("prod_missing", now=NOW) + + +def test_purge_requires_due_pending_cleanup_and_lists_due_products(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + product = store.upsert_product(ProductInput("manual", None, "待归档笔盒")) + + with pytest.raises(ValueError): + store.purge_product(product.product_id, now=NOW) + + store.mark_pending_cleanup(product.product_id, now=NOW) + future = NOW + timedelta(days=29) + with pytest.raises(ValueError): + store.purge_product(product.product_id, now=future) + assert store.due_product_cleanups(now=future) == [] + + due = NOW + timedelta(days=30) + assert [record.product_id for record in store.due_product_cleanups(now=due)] == [ + product.product_id + ] + assert store.purge_product(product.product_id, now=due).status == "purged" + + +def test_restore_removes_product_from_due_cleanup_selection(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + product = store.upsert_product(ProductInput("manual", None, "恢复笔盒")) + store.mark_pending_cleanup(product.product_id, now=NOW) + + restored = store.restore_product(product.product_id, now=NOW + timedelta(days=1)) + + assert restored.status == "active" + assert restored.purge_after is None + assert store.due_product_cleanups(now=NOW + timedelta(days=31)) == [] + + +def test_version_bound_operations_reject_unknown_or_mismatched_versions(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + first = store.upsert_product(ProductInput("manual", None, "第一个笔盒")) + second = store.upsert_product(ProductInput("manual", None, "第二个笔盒")) + version = store.create_version(first.product_id) + + with pytest.raises(ValueError): + store.add_image(first.product_id, "ver_missing") + with pytest.raises(ValueError): + store.add_wordcloud_archive(first.product_id, "ver_missing") + with pytest.raises(ValueError): + store.add_image(second.product_id, version.version_id) + with pytest.raises(ValueError): + store.add_wordcloud_archive(second.product_id, version.version_id)