File size: 1,809 Bytes
0cb7559 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | from typing import List
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from scanner.engine import PIIScannerEngine
from scanner.zip_handler import ZipExtractionError, extract_uploaded_items
MAX_UPLOAD_BYTES = 25 * 1024 * 1024
MAX_BATCH_FILES = 50
app = FastAPI(title="PII Scanner API", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
scanner = PIIScannerEngine()
@app.get("/health")
def health():
return {"status": "ok", "model_loaded": scanner.model_loaded}
@app.post("/scan")
async def scan(files: List[UploadFile] = File(...), redact: bool = Form(True)):
if not files:
raise HTTPException(status_code=400, detail="No files uploaded")
if len(files) > MAX_BATCH_FILES:
raise HTTPException(status_code=400, detail=f"Too many uploaded files. Max {MAX_BATCH_FILES}")
uploaded_items: List[tuple[str, bytes]] = []
for file in files:
content = await file.read()
if len(content) > MAX_UPLOAD_BYTES:
raise HTTPException(status_code=400, detail=f"File too large: {file.filename}")
uploaded_items.append((file.filename or "uploaded_file", content))
try:
extracted_files = extract_uploaded_items(uploaded_items)
except ZipExtractionError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(status_code=400, detail="Invalid ZIP or unsupported archive content") from exc
if not extracted_files:
raise HTTPException(status_code=400, detail="No supported files found in upload")
return scanner.scan_files(extracted_files, redact=redact)
|