140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""
|
|
Signatur-Plotter-Service (FastAPI).
|
|
|
|
Zustandsloser Service: nimmt ein PDF mit unsichtbaren Marker-Token, bis zu drei
|
|
Signatur-SVGs und eine Job-Definition entgegen und erzeugt A4-Plot-SVGs bzw. ein
|
|
Vorschau-PDF. Keine Datenbank, keine Persistenz, keine Plotter-Kalibrierung.
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import zipfile
|
|
from typing import List
|
|
|
|
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import ValidationError
|
|
|
|
from core import build_preview_pdf, place_all
|
|
from models import SignatureJob
|
|
|
|
app = FastAPI(title="Signatur-Plotter-Service", version="1.0.0")
|
|
|
|
# CORS offen halten, damit das Upload-Frontend (anderer Port/Origin) den Service erreicht.
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
MAX_SVGS = 3
|
|
|
|
|
|
# ── Eingaben aufbereiten ────────────────────────────────────────────────
|
|
def _parse_jobs(job_raw: str) -> List[SignatureJob]:
|
|
"""Parst das Job-Definition-JSON-Formfeld zu einer Liste von SignatureJob."""
|
|
try:
|
|
data = json.loads(job_raw)
|
|
except json.JSONDecodeError as e:
|
|
raise HTTPException(400, f"Job-Definition ist kein gültiges JSON: {e}")
|
|
|
|
# Sowohl eine reine Liste als auch ein Wrapper-Objekt akzeptieren.
|
|
if isinstance(data, dict):
|
|
data = data.get("signaturen") or data.get("jobs") or [data]
|
|
if not isinstance(data, list) or not data:
|
|
raise HTTPException(400, "Job-Definition muss eine nicht-leere Liste sein.")
|
|
|
|
jobs: List[SignatureJob] = []
|
|
errors: List[str] = []
|
|
for i, entry in enumerate(data, start=1):
|
|
try:
|
|
jobs.append(SignatureJob(**entry))
|
|
except ValidationError as e:
|
|
for err in e.errors():
|
|
loc = ".".join(str(x) for x in err["loc"])
|
|
errors.append(f"Eintrag {i} ({loc}): {err['msg']}")
|
|
except TypeError:
|
|
errors.append(f"Eintrag {i}: ungültiges Format.")
|
|
if errors:
|
|
# 422 bei Validierungsfehlern der Job-Definition.
|
|
raise HTTPException(422, detail=errors)
|
|
|
|
# Doppelte Token in der Job-Definition selbst sind ein Eingabefehler.
|
|
seen = {}
|
|
for j in jobs:
|
|
seen[j.token] = seen.get(j.token, 0) + 1
|
|
dups = [t for t, c in seen.items() if c > 1]
|
|
if dups:
|
|
raise HTTPException(422, detail=[f"Token '{t}' mehrfach in der Job-Definition." for t in dups])
|
|
|
|
return jobs
|
|
|
|
|
|
async def _read_svgs(svgs: List[UploadFile]) -> dict:
|
|
"""Liest die hochgeladenen SVGs in ein {dateiname: bytes}-Dict. Max. 3."""
|
|
if len(svgs) > MAX_SVGS:
|
|
raise HTTPException(400, f"Maximal {MAX_SVGS} SVG-Dateien erlaubt (es waren {len(svgs)}).")
|
|
out = {}
|
|
for f in svgs:
|
|
out[f.filename] = await f.read()
|
|
return out
|
|
|
|
|
|
async def _prepare(pdf: UploadFile, svgs: List[UploadFile], job: str):
|
|
"""Gemeinsame Aufbereitung für /render und /preview."""
|
|
jobs = _parse_jobs(job)
|
|
svg_map = await _read_svgs(svgs)
|
|
pdf_bytes = await pdf.read()
|
|
if not pdf_bytes:
|
|
raise HTTPException(400, "PDF-Datei ist leer.")
|
|
pages, errors = place_all(pdf_bytes, svg_map, jobs)
|
|
if errors:
|
|
# 422: fachliche Validierung (Token nicht gefunden, SVG fehlt, keine viewBox …)
|
|
raise HTTPException(422, detail=errors)
|
|
return pdf_bytes, pages
|
|
|
|
|
|
# ── Endpunkte ───────────────────────────────────────────────────────────
|
|
@app.get("/")
|
|
def health():
|
|
return {"service": "Signatur-Plotter-Service", "status": "ok"}
|
|
|
|
|
|
@app.post("/render")
|
|
async def render(
|
|
pdf: UploadFile = File(...),
|
|
svgs: List[UploadFile] = File(default=[]),
|
|
job: str = Form(...),
|
|
):
|
|
"""Erzeugt pro betroffener Seite eine A4-SVG und liefert alle als ZIP."""
|
|
_, pages = await _prepare(pdf, svgs, job)
|
|
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for p in pages:
|
|
zf.writestr(p.filename, p.a4_svg)
|
|
buf.seek(0)
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/zip",
|
|
headers={"Content-Disposition": 'attachment; filename="signaturen.zip"'},
|
|
)
|
|
|
|
|
|
@app.post("/preview")
|
|
async def preview(
|
|
pdf: UploadFile = File(...),
|
|
svgs: List[UploadFile] = File(default=[]),
|
|
job: str = Form(...),
|
|
):
|
|
"""Erzeugt ein Vorschau-PDF mit allen betroffenen Seiten und Signatur-Overlay."""
|
|
pdf_bytes, pages = await _prepare(pdf, svgs, job)
|
|
out_pdf = build_preview_pdf(pdf_bytes, pages)
|
|
return StreamingResponse(
|
|
io.BytesIO(out_pdf),
|
|
media_type="application/pdf",
|
|
headers={"Content-Disposition": 'inline; filename="vorschau.pdf"'},
|
|
)
|