70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
CAPTURE_DIR = Path(tempfile.gettempdir()) / "chart-captures"
|
|
CAPTURE_TTL_SECONDS = 24 * 60 * 60
|
|
CAPTURE_LIMIT = 50
|
|
CAPTURE_MAX_BYTES = 10 * 1024 * 1024
|
|
CAPTURE_ID = re.compile(r"^c-[A-Za-z0-9_-]{12}$")
|
|
|
|
|
|
def _remove(capture_id: str) -> None:
|
|
for suffix in (".png", ".json"):
|
|
(CAPTURE_DIR / f"{capture_id}{suffix}").unlink(missing_ok=True)
|
|
|
|
|
|
def cleanup_captures(now: float | None = None) -> None:
|
|
current = time.time() if now is None else now
|
|
images = sorted(CAPTURE_DIR.glob("c-*.png"), key=lambda path: path.stat().st_mtime)
|
|
for image in images:
|
|
if current - image.stat().st_mtime > CAPTURE_TTL_SECONDS:
|
|
_remove(image.stem)
|
|
images = sorted(CAPTURE_DIR.glob("c-*.png"), key=lambda path: path.stat().st_mtime)
|
|
for image in images[: max(0, len(images) - CAPTURE_LIMIT + 1)]:
|
|
_remove(image.stem)
|
|
|
|
|
|
def save_capture(image: bytes, metadata: dict) -> tuple[str, dict]:
|
|
if not image.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
raise ValueError("Capture is not a PNG")
|
|
if len(image) > CAPTURE_MAX_BYTES:
|
|
raise ValueError("Capture exceeds 10 MB")
|
|
CAPTURE_DIR.mkdir(parents=True, exist_ok=True)
|
|
cleanup_captures()
|
|
capture_id = f"c-{secrets.token_urlsafe(9)}"
|
|
details = {
|
|
**metadata,
|
|
"id": capture_id,
|
|
"received_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
|
|
"commit": os.environ.get("SOURCE_COMMIT", "dev"),
|
|
"bytes": len(image),
|
|
}
|
|
image_path = CAPTURE_DIR / f"{capture_id}.png"
|
|
metadata_path = CAPTURE_DIR / f"{capture_id}.json"
|
|
image_path.write_bytes(image)
|
|
metadata_path.write_text(json.dumps(details, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
return capture_id, details
|
|
|
|
|
|
def capture_path(capture_id: str, suffix: str) -> Path | None:
|
|
if not CAPTURE_ID.fullmatch(capture_id):
|
|
return None
|
|
path = CAPTURE_DIR / f"{capture_id}{suffix}"
|
|
return path if path.is_file() else None
|
|
|
|
|
|
def delete_capture(capture_id: str) -> bool:
|
|
if not CAPTURE_ID.fullmatch(capture_id):
|
|
return False
|
|
image = CAPTURE_DIR / f"{capture_id}.png"
|
|
metadata = CAPTURE_DIR / f"{capture_id}.json"
|
|
if not image.is_file() and not metadata.is_file():
|
|
return False
|
|
_remove(capture_id)
|
|
return True
|