pdf_to_ics/web/app.py

#!/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, JSONResponse
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("/preview")
async def preview_pdf(
    file: UploadFile = File(...),
    _: None = Depends(require_auth),
):
    filename = file.filename or "dienstplan.pdf"
    if not filename.lower().endswith(".pdf"):
        return JSONResponse(
            {"error": "Bitte eine PDF-Datei hochladen."},
            status_code=400,
        )

    data = await file.read()
    if not data:
        return JSONResponse(
            {"error": "Die Datei ist leer."},
            status_code=400,
        )

    with tempfile.TemporaryDirectory(prefix="pdf_to_ics_web_preview_") 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 JSONResponse(
                {"error": "Keine Dienstplan-Events in der PDF gefunden."},
                status_code=422,
            )

        events_preview = []
        for event in dienstplan["events"]:
            events_preview.append({
                "date": event["date"].strftime("%d.%m.%Y") if event["date"] else "",
                "service": event["service"] or "—",
                "time": f"{event['start_time']}-{event['end_time']}" if event["start_time"] and event["end_time"] else "—",
            })

        return JSONResponse({
            "success": True,
            "filename": filename,
            "metadata": {
                "name": f"{dienstplan.get('vorname', '')} {dienstplan.get('name', '')}".strip(),
                "personalnummer": dienstplan.get("personalnummer", "—"),
                "betriebshof": dienstplan.get("betriebshof", "—"),
                "count": len(dienstplan["events"]),
            },
            "events": events_preview,
        })


@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"}