feat: add product archive metadata store

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)

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
 

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)