from __future__ import annotations import json from pathlib import Path from uuid import uuid4 from app.core.config import settings from app.repositories.scheme_artifacts import create_scheme_artifact, list_scheme_artifacts def _preview_storage_dir() -> Path: path = Path(settings.storage_preview_dir) path.mkdir(parents=True, exist_ok=True) return path async def save_publish_preview_artifact( *, scheme_id: str, scheme_version_id: str, payload: dict, baseline_scheme_version_id: str | None, ) -> dict: artifact_dir = _preview_storage_dir() / uuid4().hex artifact_dir.mkdir(parents=True, exist_ok=True) storage_path = artifact_dir / "publish-preview.json" storage_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") artifact = await create_scheme_artifact( scheme_id=scheme_id, scheme_version_id=scheme_version_id, artifact_type="publish_preview", artifact_variant=baseline_scheme_version_id or "default", storage_path=str(storage_path), meta_json={ "baseline_scheme_version_id": baseline_scheme_version_id, "summary": payload.get("summary"), }, ) return artifact async def get_latest_publish_preview_artifact( *, scheme_version_id: str, baseline_scheme_version_id: str | None, ): rows = await list_scheme_artifacts(scheme_version_id=scheme_version_id) variant = baseline_scheme_version_id or "default" matching = [ row for row in rows if row.artifact_type == "publish_preview" and row.artifact_variant == variant ] if not matching: return None matching.sort(key=lambda row: (row.created_at, row.id), reverse=True) return matching[0]