John Wilganowski
4ecc7c5cef
APP_BASE_URL default (config, .env.example, both compose files) is now https://conchecker.scoutitsystems.com with no port, so review-required and final-report email links use the public site. CI bakes the short commit SHA into the image as APP_BUILD via a Docker build-arg; /health returns version+build and the site header shows the build so it's easy to confirm which image is deployed. Local runs default to 'dev'.
238 lines
9.1 KiB
Python
238 lines
9.1 KiB
Python
"""
|
|
main.py - Thin FastAPI app: upload a PDF, run the pipeline, return the report.
|
|
|
|
Run from the project root:
|
|
uvicorn backend.main:app --reload
|
|
|
|
The pipeline is synchronous and can take minutes on a large set; for a single
|
|
architect checking one set at a time that's fine. Move to a job queue if this
|
|
ever needs concurrency.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
|
|
from fastapi.responses import HTMLResponse, JSONResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
import backend.jobs
|
|
from backend import config
|
|
from backend.jobs import PIPELINE_MODES, create_job, get_job, _set
|
|
from backend.pipeline.pdf_processor import render_page_jpeg
|
|
from backend.review.feedback import decision_to_label, write_label
|
|
from backend.review.finalizer import finalize_review
|
|
from backend.review.store import ReviewStore
|
|
|
|
app = FastAPI(title=config.APP_TITLE, version=config.APP_VERSION)
|
|
|
|
_FRONTEND_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "frontend")
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok", "model": config.MODEL,
|
|
"version": config.APP_VERSION,
|
|
"build": config.APP_BUILD,
|
|
"key_configured": bool(config.AI_API_KEY),
|
|
"email_configured": bool(config.SMTP_HOST and config.SMTP_USER and config.SMTP_PASSWORD)}
|
|
|
|
|
|
@app.post("/check")
|
|
async def check(
|
|
file: UploadFile = File(...),
|
|
notification_email: Optional[str] = Form(None),
|
|
project_name: Optional[str] = Form(None),
|
|
address: Optional[str] = Form(None),
|
|
occupancy: Optional[str] = Form(None),
|
|
work_type: Optional[str] = Form(None),
|
|
text_local: bool = Form(False),
|
|
pipeline_mode: str = Form("classic"),
|
|
):
|
|
"""
|
|
Accept a PDF, start a background conflict check, and return a job_id
|
|
immediately. The client can poll GET /jobs/{id} or just wait for the email.
|
|
|
|
Optional intake fields (project_name/address/occupancy/work_type) feed the
|
|
Stage 0 jurisdiction profile; anything left blank is derived from the cover
|
|
sheet.
|
|
"""
|
|
if not file.filename.lower().endswith(".pdf"):
|
|
raise HTTPException(status_code=400, detail="Please upload a PDF.")
|
|
pipeline_mode = pipeline_mode.strip().lower()
|
|
if pipeline_mode not in PIPELINE_MODES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"pipeline_mode must be one of: {', '.join(sorted(PIPELINE_MODES))}",
|
|
)
|
|
|
|
os.makedirs(config.UPLOAD_DIR, exist_ok=True)
|
|
suffix = "_" + os.path.basename(file.filename)
|
|
fd, tmp_path = tempfile.mkstemp(suffix=suffix, dir=config.UPLOAD_DIR)
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(await file.read())
|
|
|
|
email = (notification_email or "").strip() or None
|
|
project_input = {
|
|
k: v.strip()
|
|
for k, v in {
|
|
"project_name": project_name, "address": address,
|
|
"occupancy": occupancy, "work_type": work_type,
|
|
}.items()
|
|
if v and v.strip()
|
|
}
|
|
job_id = create_job(tmp_path, source_filename=file.filename, email=email,
|
|
project_input=project_input, text_local=text_local,
|
|
pipeline_mode=pipeline_mode)
|
|
return JSONResponse({
|
|
"job_id": job_id,
|
|
"status": "queued",
|
|
"email": email,
|
|
"pipeline_mode": pipeline_mode,
|
|
})
|
|
|
|
|
|
@app.get("/jobs/{job_id}")
|
|
def job_status(job_id: str):
|
|
job = get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
return JSONResponse(job)
|
|
|
|
|
|
@app.get("/jobs/{job_id}/review")
|
|
def review_queue(job_id: str):
|
|
job = get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
|
|
# Read-only endpoint: don't create review/ dirs just by looking at them
|
|
# (readers already degrade to empty on missing files).
|
|
store = ReviewStore(out_dir, create=False)
|
|
queue = store.read_queue()
|
|
return {"queue": queue, "progress": store.progress(queue),
|
|
"decisions": store.read_decisions()}
|
|
|
|
|
|
@app.post("/jobs/{job_id}/review-decisions")
|
|
def save_review_decisions(job_id: str, payload: dict):
|
|
job = get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
if job.get("status") not in ("needs_review", "reviewing"):
|
|
# Positive state guard, mirroring the finalize endpoint: only jobs
|
|
# sitting at (or working through) the review gate accept decisions.
|
|
raise HTTPException(status_code=409, detail={
|
|
"detail": f"cannot save review decisions for a job in status {job.get('status')}",
|
|
})
|
|
out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
|
|
store = ReviewStore(out_dir)
|
|
queue = store.read_queue()
|
|
items_by_id = {item.get("review_item_id"): item for item in queue}
|
|
saved = 0
|
|
try:
|
|
for decision in payload.get("decisions") or []:
|
|
store.append_decision(decision)
|
|
queue_item = items_by_id.get(decision.get("review_item_id"))
|
|
if queue_item is not None:
|
|
write_label(out_dir, decision_to_label(queue_item, decision, job))
|
|
saved += 1
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=422, detail=str(e))
|
|
progress = store.progress(queue)
|
|
if job.get("status") == "needs_review" and saved > 0 and progress["remaining"] > 0:
|
|
try:
|
|
_set(job_id, status="reviewing")
|
|
except KeyError:
|
|
pass # job not in the in-memory registry (e.g. loaded from disk)
|
|
return {"progress": progress}
|
|
|
|
|
|
def _finalize_job(job_id: str, out_dir: str) -> None:
|
|
"""Background finalization: the ONE place the final report email may fire."""
|
|
try:
|
|
report = finalize_review(job_id, out_dir)
|
|
except Exception as e:
|
|
try:
|
|
_set(job_id, status="finalization_error", error=str(e),
|
|
finished_at=time.time(), stage=None)
|
|
except KeyError:
|
|
pass # job not in the in-memory registry
|
|
return
|
|
try:
|
|
_set(job_id, status="done", report=report,
|
|
finished_at=time.time(), stage=None)
|
|
except KeyError:
|
|
pass
|
|
try:
|
|
backend.jobs._notify(job_id, report, out_dir)
|
|
except Exception as e:
|
|
print(f"[Jobs] Final notification for {job_id} failed: {e}")
|
|
|
|
|
|
@app.post("/jobs/{job_id}/finalize-review")
|
|
def finalize_review_endpoint(job_id: str):
|
|
job = get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
if job.get("status") in ("done", "finalizing"):
|
|
raise HTTPException(status_code=409, detail={
|
|
"detail": f"job is already {job['status']}",
|
|
})
|
|
if job.get("status") not in ("needs_review", "reviewing", "finalization_error"):
|
|
# Positive state-machine guard: finalization (and the final email) is
|
|
# only reachable after the job has passed through the review gate.
|
|
raise HTTPException(status_code=409, detail={
|
|
"detail": f"cannot finalize a job in status {job.get('status')}",
|
|
})
|
|
out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
|
|
store = ReviewStore(out_dir)
|
|
queue = store.read_queue()
|
|
decisions = store.read_decisions()
|
|
if any(item.get("blocking") and item.get("review_item_id") not in decisions
|
|
for item in queue):
|
|
# 409 detail shape: {"detail": <message>, "progress": <store.progress()>}
|
|
raise HTTPException(status_code=409, detail={
|
|
"detail": "incomplete review",
|
|
"progress": store.progress(queue),
|
|
})
|
|
try:
|
|
_set(job_id, status="finalizing")
|
|
except KeyError:
|
|
pass # job not in the in-memory registry (e.g. loaded from disk)
|
|
threading.Thread(target=_finalize_job, args=(job_id, out_dir), daemon=True).start()
|
|
return {"status": "finalizing"}
|
|
|
|
|
|
@app.get("/jobs/{job_id}/sheet-image/{page}")
|
|
def sheet_image(job_id: str, page: int):
|
|
"""Render one page of a completed job's source PDF as JPEG (sheet viewer)."""
|
|
pdf = os.path.join(config.OUTPUT_DIR, job_id, "source.pdf")
|
|
if not os.path.isfile(pdf):
|
|
raise HTTPException(status_code=404, detail="Source PDF not found for this job")
|
|
try:
|
|
data = render_page_jpeg(pdf, page)
|
|
except IndexError:
|
|
raise HTTPException(status_code=404, detail=f"Page {page} out of range")
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Could not render page: {e}")
|
|
return Response(content=data, media_type="image/jpeg",
|
|
headers={"Cache-Control": "max-age=3600"})
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index():
|
|
index_path = os.path.join(_FRONTEND_DIR, "index.html")
|
|
if os.path.isfile(index_path):
|
|
with open(index_path) as f:
|
|
return f.read()
|
|
return "<h1>Conflict Checker</h1><p>frontend/index.html not found.</p>"
|
|
|
|
|
|
# Serve any other static assets (none required for the single-file UI).
|
|
if os.path.isdir(_FRONTEND_DIR):
|
|
app.mount("/static", StaticFiles(directory=_FRONTEND_DIR), name="static")
|