98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
"""草稿预览：未发布蓝图的内部可访问快照（供 /#/preview/{id} 与真页面截图）。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_lock = threading.Lock()
|
|
_store: dict[str, dict[str, Any]] = {}
|
|
_TTL_SEC = 3600
|
|
_disk = (Path(__file__).resolve().parent / ".runtime" / "preview").resolve()
|
|
_alt_disk = (Path(__file__).resolve().parents[1] / ".runtime" / "preview").resolve()
|
|
|
|
|
|
def _preview_dirs() -> list[Path]:
|
|
dirs = [_disk]
|
|
if _alt_disk != _disk:
|
|
dirs.append(_alt_disk)
|
|
return dirs
|
|
|
|
|
|
def _purge() -> None:
|
|
now = time.time()
|
|
dead = [k for k, v in _store.items() if now - float(v.get("ts") or 0) > _TTL_SEC]
|
|
for k in dead:
|
|
_store.pop(k, None)
|
|
for d in _preview_dirs():
|
|
p = d / f"{k}.json"
|
|
if p.is_file():
|
|
try:
|
|
p.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def create_preview(
|
|
blueprint: dict[str, Any],
|
|
*,
|
|
rows: list[dict[str, Any]] | None = None,
|
|
resource: str = "",
|
|
) -> str:
|
|
"""写入预览包，返回 preview_id。"""
|
|
pid = uuid.uuid4().hex[:12]
|
|
res = resource
|
|
if not res:
|
|
apis = (blueprint.get("apis") or {}).get("resources") or []
|
|
if apis and isinstance(apis[0], dict):
|
|
res = str(apis[0].get("path") or "records").lstrip("/")
|
|
else:
|
|
res = "records"
|
|
pack = {
|
|
"id": pid,
|
|
"ts": time.time(),
|
|
"blueprint": blueprint,
|
|
"rows": list(rows or [])[:800],
|
|
"resource": res,
|
|
}
|
|
with _lock:
|
|
_purge()
|
|
_store[pid] = pack
|
|
for d in _preview_dirs():
|
|
try:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
(d / f"{pid}.json").write_text(
|
|
json.dumps(pack, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
except OSError:
|
|
continue
|
|
return pid
|
|
|
|
|
|
def get_preview(preview_id: str) -> dict[str, Any] | None:
|
|
pid = (preview_id or "").strip()
|
|
if not pid:
|
|
return None
|
|
with _lock:
|
|
_purge()
|
|
pack = _store.get(pid)
|
|
if pack:
|
|
return pack
|
|
for d in _preview_dirs():
|
|
p = d / f"{pid}.json"
|
|
if not p.is_file():
|
|
continue
|
|
try:
|
|
pack = json.loads(p.read_text(encoding="utf-8"))
|
|
if isinstance(pack, dict):
|
|
with _lock:
|
|
_store[pid] = pack
|
|
return pack
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
return None
|