feat: extract _run_analysis helper and add /api/redaction/spot endpoint

Extract inline analysis logic from the analyze endpoint into a reusable
_run_analysis() helper function, then add a new /api/redaction/spot
endpoint that chains spot_redaction detection with analysis in one call.

Co-Authored-By: Claude Opus 4.6 <[email protected]>
This commit is contained in:
Alex-GilbertandClaude Opus 4.6 committed 2026-02-16 17:56:48 -06:00
1 parent 5b395a6a8f
commit 7fb5a85634
1 file changed
+45 -19
+45 -19
View File
@@ -17,7 +17,7 @@ from PIL import Image, ImageFont
from sse_starlette.sse import EventSourceResponse
from unredact.pipeline.rasterize import rasterize_pdf
from unredact.pipeline.detect_redactions import detect_redactions
from unredact.pipeline.detect_redactions import detect_redactions, spot_redaction
from unredact.pipeline.ocr import ocr_page
from unredact.pipeline.font_detect import detect_font_for_line, CANDIDATE_FONTS, _find_font_path
from unredact.pipeline.solver import build_constraint, SolveResult
@@ -147,34 +147,22 @@ class AnalyzeRequest(BaseModel):
redaction: dict # {x, y, w, h}
@app.post("/api/redaction/analyze")
async def analyze_redaction(req: AnalyzeRequest):
doc = _docs.get(req.doc_id)
if not doc or req.page not in doc["pages"]:
return JSONResponse({"error": "not found"}, status_code=404)
page_img = doc["pages"][req.page]["original"]
rx, ry, rw, rh = req.redaction["x"], req.redaction["y"], req.redaction["w"], req.redaction["h"]
# Expand vertically to capture the full line, horizontally to page edges
pad_y = rh # one redaction-height of vertical padding
def _run_analysis(page_img: Image.Image, rx: int, ry: int, rw: int, rh: int) -> dict | None:
"""Run OCR + font detection for a redaction box. Returns analysis dict or None."""
pad_y = rh
crop_y1 = max(0, ry - pad_y)
crop_y2 = min(page_img.height, ry + rh + pad_y)
line_crop = page_img.crop((0, crop_y1, page_img.width, crop_y2))
# OCR just this line crop
lines = ocr_page(line_crop)
if not lines:
return JSONResponse({"error": "no text detected near redaction"}, status_code=422)
return None
# Find the line closest to the redaction's y-center
redaction_cy = (ry - crop_y1) + rh / 2
best_line = min(lines, key=lambda l: abs((l.y + l.h / 2) - redaction_cy))
# Detect font for this line
font_match = detect_font_for_line(best_line)
# Build segments: text before redaction, gap, text after redaction
segments = []
gap = {"x": rx, "w": rw}
@@ -184,7 +172,6 @@ async def analyze_redaction(req: AnalyzeRequest):
left_text = "".join(c.text for c in left_chars).rstrip()
right_text = "".join(c.text for c in right_chars).lstrip()
# Compute initial offset guess: align end of left text with gap start
pil_font = font_match.to_pil_font()
if left_text:
left_rendered_width = pil_font.getlength(left_text)
@@ -202,7 +189,6 @@ async def analyze_redaction(req: AnalyzeRequest):
rw2 = (right_chars[-1].x + right_chars[-1].w) - rx2
segments.append({"text": right_text, "x": rx2, "w": rw2})
# Adjust segment coordinates back to page-relative y
chars_json = [
{"text": c.text, "x": c.x, "y": c.y + crop_y1, "w": c.w, "h": c.h, "conf": c.conf}
for c in best_line.chars
@@ -230,6 +216,46 @@ async def analyze_redaction(req: AnalyzeRequest):
}
@app.post("/api/redaction/analyze")
async def analyze_redaction(req: AnalyzeRequest):
doc = _docs.get(req.doc_id)
if not doc or req.page not in doc["pages"]:
return JSONResponse({"error": "not found"}, status_code=404)
page_img = doc["pages"][req.page]["original"]
rx, ry, rw, rh = req.redaction["x"], req.redaction["y"], req.redaction["w"], req.redaction["h"]
result = _run_analysis(page_img, rx, ry, rw, rh)
if result is None:
return JSONResponse({"error": "no text detected near redaction"}, status_code=422)
return result
class SpotRequest(BaseModel):
doc_id: str
page: int
click_x: int
click_y: int
@app.post("/api/redaction/spot")
async def spot_redaction_endpoint(req: SpotRequest):
doc = _docs.get(req.doc_id)
if not doc or req.page not in doc["pages"]:
return JSONResponse({"error": "not found"}, status_code=404)
page_img = doc["pages"][req.page]["original"]
redaction = spot_redaction(page_img, req.click_x, req.click_y)
if redaction is None:
return JSONResponse({"error": "no_redaction_found"}, status_code=422)
result = _run_analysis(page_img, redaction.x, redaction.y, redaction.w, redaction.h)
response = {"box": {"x": redaction.x, "y": redaction.y, "w": redaction.w, "h": redaction.h}}
if result:
response.update(result)
return response
# In-memory dictionary store
_dictionary_store = DictionaryStore()