diff --git a/backend/service/app.py b/backend/service/app.py index 0742632..cc0f3d8 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -27,6 +27,8 @@ from .job_manager import JobManager from .line_spacing import analyze_svg_line_spacing_file from .log_config import get_logger from .metadata_store import MetadataStore +from .product_archive_service import ProductArchiveService, ProductPendingCleanupError +from .product_archive_store import ProductArchiveStore from .runner import JobRunner from .schemas import ( Asset, @@ -41,6 +43,9 @@ from .schemas import ( LineSpacingAnalysisSummary, Project, ProjectSummary, + ProductInput, + ProductRecord, + ProductVersionArchiveResponse, Template, WordLocation, ) @@ -62,17 +67,20 @@ FONTS_DIR = PROJECT_ROOT / "service_fonts" DESIGN_TEMPLATES_DIR = PROJECT_ROOT / "service_design_templates" METADATA_DIR = PROJECT_ROOT / "service_metadata" ORDERS_DIR = PROJECT_ROOT / "service_orders" +PRODUCT_ARCHIVES_DIR = PROJECT_ROOT / "service_product_archives" ASSETS_DIR.mkdir(parents=True, exist_ok=True) PROJECTS_DIR.mkdir(parents=True, exist_ok=True) FONTS_DIR.mkdir(parents=True, exist_ok=True) DESIGN_TEMPLATES_DIR.mkdir(parents=True, exist_ok=True) METADATA_DIR.mkdir(parents=True, exist_ok=True) ORDERS_DIR.mkdir(parents=True, exist_ok=True) +PRODUCT_ARCHIVES_DIR.mkdir(parents=True, exist_ok=True) metadata_store = MetadataStore(METADATA_DIR / "app.db") manager = JobManager(metadata_store) storage = Storage(WORKSPACE_DIR) runner = JobRunner(PROJECT_ROOT, manager) +product_archive_store = ProductArchiveStore(PRODUCT_ARCHIVES_DIR / "metadata") app = FastAPI(title="WordCloud Test Service", version="0.1.0") @@ -1026,6 +1034,133 @@ async def login(request: Request) -> dict: return {"token": _orders_token()} +# ═══════════════════════════════════════════════════════════ +# 3.10 产品及设计版本归档（需登录） +# ═══════════════════════════════════════════════════════════ + +def _get_product_or_404(product_id: str) -> ProductRecord: + try: + return product_archive_store.get_product(product_id) + except ValueError as exc: + raise HTTPException(status_code=404, detail="product not found") from exc + + +def _product_archive_service() -> ProductArchiveService: + """Build against current globals so isolated tests can substitute local storage.""" + return ProductArchiveService( + store=product_archive_store, + storage=storage, + load_asset_meta=lambda asset_id: _read_asset_meta(_asset_dir(asset_id)), + resolve_job_status=_resolve_job_status, + ) + + +@app.post("/api/products", response_model=ProductRecord, status_code=201) +def create_product(request: Request, product: ProductInput) -> ProductRecord: + _require_orders_auth(request) + try: + return product_archive_store.upsert_product(product) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.get("/api/products", response_model=list[ProductRecord]) +def list_products(request: Request, query: str = Query("")) -> list[ProductRecord]: + _require_orders_auth(request) + return product_archive_store.list_products(query=query) + + +@app.get("/api/products/{product_id}", response_model=ProductRecord) +def get_product(request: Request, product_id: str) -> ProductRecord: + _require_orders_auth(request) + return _get_product_or_404(product_id) + + +@app.post( + "/api/products/{product_id}/versions", + response_model=ProductVersionArchiveResponse, + status_code=201, +) +async def create_product_version( + request: Request, + product_id: str, + document_json: str = Form(""), + preview: UploadFile | None = File(None), +) -> ProductVersionArchiveResponse: + _require_orders_auth(request) + _get_product_or_404(product_id) + if preview is None or preview.content_type != "image/png": + raise HTTPException(status_code=400, detail="preview must be image/png") + try: + document = json.loads(document_json) + except json.JSONDecodeError as exc: + raise HTTPException(status_code=400, detail="document_json must be valid JSON") from exc + if not isinstance(document, dict): + raise HTTPException(status_code=400, detail="document_json must be a JSON object") + + try: + version = _product_archive_service().archive_version( + product_id=product_id, + document=document, + preview_bytes=await preview.read(), + now=datetime.now(timezone.utc), + ) + except ProductPendingCleanupError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + source_job_ids = [ + str(source.get("source_job_id") or "") + for source in version.metadata.get("wordcloud_sources", []) + if isinstance(source, dict) + ] + rows = product_archive_store._fetchall( + "SELECT * FROM product_wordcloud_archives WHERE version_id = ? ORDER BY created_at, archive_id", + (version.version_id,), + ) + archives = [ + { + "archive_id": row["archive_id"], + "product_id": row["product_id"], + "version_id": row["version_id"], + "archive_path": row["archive_path"], + "created_at": datetime.fromisoformat(row["created_at"]), + "source_job_id": source_job_id, + "db_path": row["archive_path"], + } + for source_job_id, row in zip(source_job_ids, rows, strict=True) + ] + return ProductVersionArchiveResponse( + **version.model_dump(), + design_preview_path=str(PRODUCT_ARCHIVES_DIR / product_id / version.version_id / "design-preview.png"), + wordcloud_count=len(archives), + wordcloud_archives=archives, + ) + + +@app.delete("/api/products/{product_id}", response_model=ProductRecord) +def delete_product(request: Request, product_id: str) -> ProductRecord: + _require_orders_auth(request) + product = _get_product_or_404(product_id) + if product.status == "pending_cleanup": + raise HTTPException(status_code=409, detail="product is pending cleanup") + try: + return product_archive_store.mark_pending_cleanup(product_id, now=datetime.now(timezone.utc)) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/products/{product_id}/restore", response_model=ProductRecord) +def restore_product(request: Request, product_id: str) -> ProductRecord: + _require_orders_auth(request) + _get_product_or_404(product_id) + try: + return product_archive_store.restore_product(product_id, now=datetime.now(timezone.utc)) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + @app.get("/api/orders") def list_orders(request: Request) -> list[dict]: """生产订单列表：来自小程序下单派单投递到 wordcloud 的 WCD 生产任务（需登录）。""" diff --git a/backend/service/product_archive_service.py b/backend/service/product_archive_service.py new file mode 100644 index 0000000..3187ff7 --- /dev/null +++ b/backend/service/product_archive_service.py @@ -0,0 +1,148 @@ +"""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 soft-deleted product is changed before restoration.""" + + +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 + + def _rollback_metadata(self, version_id: str) -> None: + # ProductArchiveStore deliberately keeps its public API small. These rows + # all belong to the just-created version and can be removed safely here. + self.store._execute("DELETE FROM product_wordcloud_archives WHERE version_id = ?", (version_id,)) + self.store._execute("DELETE FROM product_images WHERE version_id = ?", (version_id,)) + self.store._execute("DELETE FROM product_versions WHERE version_id = ?", (version_id,)) + + 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 == "pending_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" + version: ProductVersionRecord | None = None + 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]] = [] + 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" + ) + copy_word_locations_snapshot(source_db, snapshot_path) + snapshots.append((source, snapshot_path)) + + version = self.store.create_version( + product_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 + ], + }, + now=now, + ) + target_dir = product_dir / version.version_id + staging_dir.replace(target_dir) + final_dir = target_dir + + final_preview = final_dir / "design-preview.png" + self.store.add_image( + product_id, version.version_id, image_path=str(final_preview), image_type="design_preview", now=now + ) + for _, snapshot_path in snapshots: + final_snapshot = final_dir / snapshot_path.relative_to(staging_dir) + self.store.add_wordcloud_archive( + product_id, version.version_id, archive_path=str(final_snapshot), now=now + ) + return version + except Exception: + if version is not None: + self._rollback_metadata(version.version_id) + shutil.rmtree(final_dir or staging_dir, ignore_errors=True) + raise diff --git a/backend/service/schemas.py b/backend/service/schemas.py index 329d65b..977fd9b 100644 --- a/backend/service/schemas.py +++ b/backend/service/schemas.py @@ -74,6 +74,17 @@ class ProductWordcloudArchiveRecord(BaseModel): created_at: datetime +class ProductWordcloudArchiveResponse(ProductWordcloudArchiveRecord): + source_job_id: str + db_path: str + + +class ProductVersionArchiveResponse(ProductVersionRecord): + design_preview_path: str + wordcloud_count: int + wordcloud_archives: list[ProductWordcloudArchiveResponse] = Field(default_factory=list) + + class JobCreateResponse(BaseModel): job_id: str diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index 78850ac..aed1b07 100644 --- a/backend/tests/test_product_archive.py +++ b/backend/tests/test_product_archive.py @@ -1,9 +1,15 @@ +import json import hashlib import sqlite3 import sys +import shutil +from io import BytesIO from pathlib import Path +from types import SimpleNamespace import pytest +from fastapi.testclient import TestClient +from PIL import Image BACKEND_DIR = Path(__file__).resolve().parents[1] @@ -16,6 +22,71 @@ from service.product_archive import ( # noqa: E402 find_visible_wordcloud_sources, validate_word_locations_db, ) +from service.product_archive_store import ProductArchiveStore # noqa: E402 +from service import app as service_app # noqa: E402 + + +def _png_bytes() -> bytes: + image = BytesIO() + Image.new("RGB", (1, 1), "white").save(image, format="PNG") + return image.getvalue() + + +PNG_BYTES = _png_bytes() + + +def orders_auth_header() -> dict[str, str]: + return {"Authorization": f"Bearer {service_app._orders_token()}"} + + +def create_product(client): + response = client.post( + "/api/products", json={"name": "笔盒", "source": "manual"}, headers=orders_auth_header() + ) + assert response.status_code == 201 + return response.json() + + +@pytest.fixture +def product_archive_client(tmp_path, monkeypatch): + archive_root = tmp_path / "product_archives" + store = ProductArchiveStore(archive_root / "metadata") + monkeypatch.setattr(service_app, "PRODUCT_ARCHIVES_DIR", archive_root, raising=False) + monkeypatch.setattr(service_app, "product_archive_store", store, raising=False) + client = TestClient(service_app.app) + client.archive_root = archive_root + return client + + +@pytest.fixture +def prepared_wordcloud_job(tmp_path, monkeypatch): + workspace = tmp_path / "source-workspace" + db_path = workspace / "output" / "word_locations.sqlite" + db_path.parent.mkdir(parents=True) + with sqlite3.connect(db_path) as connection: + connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)") + connection.execute("INSERT INTO word_locations (name) VALUES ('hello')") + + asset_id = "asset_wordcloud" + assets_dir = tmp_path / "assets" + asset_dir = assets_dir / asset_id[:2] / asset_id + asset_dir.mkdir(parents=True) + (asset_dir / "meta.json").write_text( + json.dumps({"asset_id": asset_id, "type": "wordcloud", "job_id": "job-success"}), + encoding="utf-8", + ) + monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir) + monkeypatch.setattr( + service_app, + "_resolve_job_status", + lambda job_id: SimpleNamespace(status="success", artifacts={"db": str(db_path)}), + ) + return SimpleNamespace( + workspace=workspace, + asset_id=asset_id, + document={"elements": [{"type": "sticker", "assetId": asset_id}]}, + db_path=db_path, + ) def test_scanner_keeps_only_visible_wordcloud_assets_and_deduplicates(): @@ -84,3 +155,100 @@ def test_snapshot_validates_table_copies_atomically_and_returns_checksum(tmp_pat assert checksum == hashlib.sha256(destination.read_bytes()).hexdigest() with sqlite3.connect(destination) as connection: assert connection.execute("SELECT name FROM word_locations").fetchone() == ("hello",) + + +def test_archive_version_copies_db_after_source_workspace_is_removed( + product_archive_client, prepared_wordcloud_job +): + product = create_product(product_archive_client) + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps(prepared_wordcloud_job.document)}, + files={"preview": ("design-preview.png", PNG_BYTES, "image/png")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 201 + archive = response.json()["wordcloud_archives"][0] + shutil.rmtree(prepared_wordcloud_job.workspace) + assert Path(archive["db_path"]).exists() + + +def test_archive_version_rejects_preview_with_non_png_mime_type( + product_archive_client, prepared_wordcloud_job +): + product = create_product(product_archive_client) + + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps(prepared_wordcloud_job.document)}, + files={"preview": ("design-preview.jpg", PNG_BYTES, "image/jpeg")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 400 + + +def test_archive_version_rejects_unavailable_source_db(product_archive_client, prepared_wordcloud_job): + prepared_wordcloud_job.db_path.unlink() + product = create_product(product_archive_client) + + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps(prepared_wordcloud_job.document)}, + files={"preview": ("design-preview.png", PNG_BYTES, "image/png")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 400 + assert not list(product_archive_client.archive_root.glob("prod_*")) + + +def test_archive_version_rejects_non_success_source_job(product_archive_client, prepared_wordcloud_job, monkeypatch): + monkeypatch.setattr( + service_app, + "_resolve_job_status", + lambda job_id: SimpleNamespace(status="running", artifacts={"db": str(prepared_wordcloud_job.db_path)}), + ) + product = create_product(product_archive_client) + + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps(prepared_wordcloud_job.document)}, + files={"preview": ("design-preview.png", PNG_BYTES, "image/png")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 400 + + +def test_archive_version_ignores_hidden_wordclouds(product_archive_client, prepared_wordcloud_job): + document = { + "layers": [{"id": "hidden", "visible": False}], + "elements": [{"type": "sticker", "assetId": prepared_wordcloud_job.asset_id, "layerId": "hidden"}], + } + product = create_product(product_archive_client) + + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps(document)}, + files={"preview": ("design-preview.png", PNG_BYTES, "image/png")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 201 + assert response.json()["wordcloud_count"] == 0 + + +def test_archive_version_handles_document_without_wordcloud_sources(product_archive_client): + product = create_product(product_archive_client) + + response = product_archive_client.post( + f"/api/products/{product['product_id']}/versions", + data={"document_json": json.dumps({"elements": []})}, + files={"preview": ("design-preview.png", PNG_BYTES, "image/png")}, + headers=orders_auth_header(), + ) + + assert response.status_code == 201 + assert response.json()["wordcloud_count"] == 0