alex
12407b74e4
Saving current working state before proceeding to Stage 2. Includes: - Backend: Python-based QC validator with shapefile processing - Frontend: Drag-and-drop file upload interface - Sample files for testing - Documentation and revision history 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
71 lines
1.9 KiB
Python
71 lines
1.9 KiB
Python
from fastapi import FastAPI, File, UploadFile
|
|
from fastapi.responses import FileResponse, PlainTextResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import zipfile
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from qc_validator import validate_shapefiles
|
|
|
|
app = FastAPI()
|
|
|
|
# Enable CORS for frontend
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
TEMP_DIR = Path("../temp")
|
|
TEMP_DIR.mkdir(exist_ok=True)
|
|
|
|
@app.post("/upload")
|
|
async def upload_shapefile(file: UploadFile = File(...)):
|
|
"""Handle shapefile ZIP upload and QC validation"""
|
|
|
|
# Clear temp directory
|
|
for item in TEMP_DIR.glob("*"):
|
|
if item.is_file():
|
|
item.unlink()
|
|
elif item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Save uploaded file
|
|
zip_path = TEMP_DIR / file.filename
|
|
with open(zip_path, "wb") as f:
|
|
content = await file.read()
|
|
f.write(content)
|
|
|
|
# Unzip file
|
|
try:
|
|
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
|
|
zip_ref.extractall(TEMP_DIR)
|
|
except Exception as e:
|
|
return PlainTextResponse(f"Error extracting ZIP file: {str(e)}", status_code=400)
|
|
|
|
# Run QC validation
|
|
qc_result = validate_shapefiles(TEMP_DIR)
|
|
|
|
if qc_result["passed"]:
|
|
return {"message": "success"}
|
|
else:
|
|
# Generate QC report
|
|
report_path = TEMP_DIR / "QC_report.txt"
|
|
with open(report_path, "w") as f:
|
|
f.write("QC VALIDATION FAILED\n")
|
|
f.write("=" * 50 + "\n\n")
|
|
for error in qc_result["errors"]:
|
|
f.write(f"{error}\n")
|
|
|
|
return FileResponse(
|
|
path=report_path,
|
|
media_type="text/plain",
|
|
filename="QC_report.txt"
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|