133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Web-MVP für PDF zu ICS Konverter
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
import os
|
|
import secrets
|
|
|
|
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile, status
|
|
from fastapi.responses import FileResponse, HTMLResponse
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
from fastapi.templating import Jinja2Templates
|
|
from starlette.background import BackgroundTask
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from pdf_to_ics import create_ics_from_dienstplan, extract_dienstplan_data
|
|
|
|
app = FastAPI(title="PDF zu ICS Web")
|
|
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
|
|
security = HTTPBasic()
|
|
|
|
|
|
def require_auth(credentials: HTTPBasicCredentials = Depends(security)):
|
|
expected_user = os.getenv("WEB_AUTH_USER")
|
|
expected_password = os.getenv("WEB_AUTH_PASSWORD")
|
|
|
|
if not expected_user or not expected_password:
|
|
return
|
|
|
|
valid_user = secrets.compare_digest(credentials.username, expected_user)
|
|
valid_password = secrets.compare_digest(credentials.password, expected_password)
|
|
if not (valid_user and valid_password):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Unauthorized",
|
|
headers={"WWW-Authenticate": "Basic"},
|
|
)
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index(request: Request, _: None = Depends(require_auth)):
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"error": None,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post("/convert")
|
|
async def convert_pdf(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
exclude_rest: bool = Form(False),
|
|
exclude_vacation: bool = Form(False),
|
|
_: None = Depends(require_auth),
|
|
):
|
|
filename = file.filename or "dienstplan.pdf"
|
|
if not filename.lower().endswith(".pdf"):
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"error": "Bitte eine PDF-Datei hochladen.",
|
|
},
|
|
status_code=400,
|
|
)
|
|
|
|
data = await file.read()
|
|
if not data:
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"error": "Die Datei ist leer.",
|
|
},
|
|
status_code=400,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory(prefix="pdf_to_ics_web_") as temp_dir:
|
|
temp_dir_path = Path(temp_dir)
|
|
pdf_path = temp_dir_path / "upload.pdf"
|
|
pdf_path.write_bytes(data)
|
|
|
|
dienstplan = extract_dienstplan_data(str(pdf_path))
|
|
if not dienstplan.get("events"):
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"error": "Keine Dienstplan-Events in der PDF gefunden.",
|
|
},
|
|
status_code=422,
|
|
)
|
|
|
|
safe_name = re.sub(r"[^A-Za-z0-9._-]", "_", Path(filename).stem)
|
|
ics_filename = f"{safe_name}.ics"
|
|
ics_path = temp_dir_path / ics_filename
|
|
|
|
create_ics_from_dienstplan(
|
|
dienstplan,
|
|
str(ics_path),
|
|
exclude_rest=exclude_rest,
|
|
exclude_vacation=exclude_vacation,
|
|
)
|
|
|
|
content = ics_path.read_bytes()
|
|
|
|
response_tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".ics", prefix="pdf_to_ics_out_")
|
|
response_tmp.write(content)
|
|
response_tmp.close()
|
|
|
|
return FileResponse(
|
|
path=response_tmp.name,
|
|
media_type="text/calendar",
|
|
filename=ics_filename,
|
|
headers={"Cache-Control": "no-store"},
|
|
background=BackgroundTask(lambda path: os.remove(path) if os.path.exists(path) else None, response_tmp.name),
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
def health(_: None = Depends(require_auth)):
|
|
return {"status": "ok"}
|