Akiba So
198da00ecd
All checks were successful
CI / Windows build (push) Successful in 7m47s
Redesign the optional FastAPI companion around vault files (manifest / PUT/GET/DELETE + OCR jobs) instead of legacy strokes_json notes. Wire a client Server settings panel for health/login. Polish shell UX: l10n for settings/home/board, sticky-board empty state, and a narrow-screen diagnostics FAB. Co-authored-by: Cursor <cursoragent@cursor.com>
75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
"""v1 OCR job API — upload ink raster, poll status, fetch text."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
|
|
|
|
from ..auth import get_current_user
|
|
from ..config import settings
|
|
from ..ocr import queue as ocr_queue
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/jobs", status_code=status.HTTP_202_ACCEPTED)
|
|
async def create_ocr_job(
|
|
image: UploadFile = File(...),
|
|
source_path: str = Form(""),
|
|
page_index: int = Form(0),
|
|
user_id: str = Depends(get_current_user),
|
|
) -> dict:
|
|
"""Enqueue an OCR job from an uploaded PNG/JPEG of handwriting."""
|
|
job_id = str(uuid.uuid4())
|
|
blob_dir = os.path.join(settings.storage_path, "ocr_blobs", user_id)
|
|
os.makedirs(blob_dir, exist_ok=True)
|
|
ext = os.path.splitext(image.filename or "ink.png")[1] or ".png"
|
|
blob_path = os.path.join(blob_dir, f"{job_id}{ext}")
|
|
data = await image.read()
|
|
if not data:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="empty image")
|
|
with open(blob_path, "wb") as f:
|
|
f.write(data)
|
|
|
|
ocr_queue.enqueue(
|
|
{
|
|
"id": job_id,
|
|
"user_id": user_id,
|
|
"source_path": source_path,
|
|
"note_id": source_path or job_id,
|
|
"page_index": page_index,
|
|
"image_path": blob_path,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"job_id": job_id,
|
|
"status": "pending",
|
|
"source_path": source_path,
|
|
"page_index": page_index,
|
|
}
|
|
|
|
|
|
@router.get("/jobs/{job_id}")
|
|
async def get_ocr_job(
|
|
job_id: str,
|
|
user_id: str = Depends(get_current_user),
|
|
) -> dict:
|
|
job = ocr_queue.get_status(job_id)
|
|
if job is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="job not found")
|
|
if job.get("user_id") not in (None, user_id):
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, detail="forbidden")
|
|
return {
|
|
"job_id": job_id,
|
|
"status": job.get("status"),
|
|
"result": job.get("result_text") or job.get("result") or job.get("text"),
|
|
"error": job.get("error_message") or job.get("error"),
|
|
"source_path": job.get("source_path"),
|
|
"page_index": job.get("page_index"),
|
|
}
|