Document Pipeline¶

POST /api/v1/uploads/presign

class PresignRequest(BaseModel):
    case_id: UUID
    filename: str
    content_type: str           # "application/pdf", "image/jpeg", etc.
    file_size_bytes: int

class PresignResponse(BaseModel):
    document_id: UUID
    presigned_url: str          # R2 presigned PUT URL (15-minute expiry)
    expires_at: datetime

// Frontend upload (Next.js)
const upload = async (file: File, presignedUrl: string) => {
  const response = await fetch(presignedUrl, {
    method: "PUT",
    body: file,
    headers: { "Content-Type": file.type },
  });
  if (!response.ok) throw new UploadError("R2 upload failed");
};

POST /api/v1/uploads/confirm

class ConfirmRequest(BaseModel):
    document_id: UUID
    sha256_hash: str            # Client-computed hash for dedup

@router.post("/internal/callbacks/ocr")
async def ocr_callback(
    request: QStashCallbackRequest,
    x_qstash_signature: str = Header(),
):
    """Process OCR for an uploaded document (QStash callback)."""
    verify_qstash_signature(request, x_qstash_signature)
    document = await document_service.get(request.document_id)
    await ocr_pipeline.process(document)

import fitz  # PyMuPDF

async def extract_with_pymupdf(document_path: str) -> OCRResult:
    """Primary OCR: fast, free, synchronous text extraction."""
    doc = fitz.open(document_path)
    pages = []
    for page in doc:
        text = page.get_text("text")
        pages.append(PageResult(
            page_number=page.number + 1,
            text=text,
            confidence=estimate_text_quality(text),
        ))

    overall_confidence = sum(p.confidence for p in pages) / len(pages)

    if overall_confidence < 0.6:
        raise LowQualityError(f"PyMuPDF confidence {overall_confidence:.2f} below threshold")

    return OCRResult(
        method="pymupdf",
        pages=pages,
        confidence=overall_confidence,
    )

from unstructured.partition.auto import partition

async def extract_with_unstructured(document_path: str) -> OCRResult:
    """Fallback 1: Unstructured.io for complex layouts."""
    elements = partition(filename=document_path)
    text = "\n".join(str(el) for el in elements)

    return OCRResult(
        method="unstructured",
        pages=[PageResult(page_number=1, text=text, confidence=0.8)],
        confidence=0.8,
    )

async def extract_with_claude_vision(document_path: str) -> OCRResult:
    """Fallback 2: Claude Vision for scanned/handwritten documents."""
    images = convert_pdf_to_images(document_path)

    pages = []
    for i, image in enumerate(images):
        response = await anthropic_client.messages.create(
            model="claude-sonnet-4-6-20250514",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image", "source": {"type": "base64", "data": image}},
                    {"type": "text", "text": VISION_OCR_PROMPT},
                ],
            }],
            max_tokens=4000,
        )
        pages.append(PageResult(
            page_number=i + 1,
            text=response.content[0].text,
            confidence=0.85,
        ))

    return OCRResult(method="claude_vision", pages=pages, confidence=0.85)

class ExtractedEntities(BaseModel):
    """Clinical entities extracted from a medical document."""
    conditions: list[ConditionEntity]       # Diagnoses, findings
    procedures: list[ProcedureEntity]       # Past/recommended procedures
    medications: list[MedicationEntity]     # Current medications
    lab_results: list[LabResult]            # Lab values with ranges
    vitals: list[VitalSign]                 # Blood pressure, heart rate, etc.
    allergies: list[AllergyEntity]
    imaging_findings: list[ImagingFinding]
    physician_notes: list[str]              # Free-text clinical notes
    document_date: Optional[date]
    patient_name_in_doc: Optional[str]      # For cross-reference validation
    laterality: Optional[str]               # "left", "right", "bilateral"

RERANKER_PROMPT = """
You are a medical document classifier. Given an extracted document summary
and a list of candidate requirements, determine which requirement (if any)
the document satisfies.

Document summary: {document_summary}

Candidate requirements:
{candidates}

For each candidate, respond with:
- requirement_id: The ID of the requirement
- confidence: 0.0 to 1.0
- reasoning: Brief explanation

Only return matches with confidence >= 0.70.
"""

class DocumentValidator:
    """Validates extracted document data for anomalies."""

    async def validate(
        self,
        document: Document,
        extracted: ExtractedEntities,
        case: Case,
    ) -> list[ValidationIssue]:
        issues = []
        issues.extend(self._check_laterality_mismatch(extracted, case))
        issues.extend(self._check_document_age(extracted))
        issues.extend(self._check_wrong_body_part(extracted, case))
        issues.extend(self._check_patient_name_mismatch(extracted, case))
        issues.extend(self._check_ocr_quality(document))
        return issues

def _check_laterality_mismatch(
    self,
    extracted: ExtractedEntities,
    case: Case,
) -> list[ValidationIssue]:
    """Detect left/right mismatch between document and case."""
    if not extracted.laterality or not case.laterality:
        return []

    if extracted.laterality != case.laterality:
        return [ValidationIssue(
            type="laterality_mismatch",
            severity="error",
            message=(
                f"Document indicates '{extracted.laterality}' side, "
                f"but case is for '{case.laterality}' side. "
                f"Please verify and upload the correct document."
            ),
        )]
    return []

async def check_duplicate(
    tenant_id: str,
    case_id: UUID,
    sha256_hash: str,
) -> Optional[Document]:
    """Check if this exact document has already been uploaded."""
    existing = await db.fetch_one(
        """
        SELECT * FROM documents
        WHERE tenant_id = $1 AND case_id = $2 AND sha256_hash = $3
        AND status != 'deleted'
        """,
        tenant_id, case_id, sha256_hash,
    )
    return Document(**existing) if existing else None

async def supersede_document(
    document_id: UUID,
    new_document_id: UUID,
    tenant_id: str,
):
    """Mark an existing document as superseded by a newer version."""
    await db.execute(
        """
        UPDATE documents
        SET status = 'superseded',
            superseded_by = $1,
            updated_at = NOW()
        WHERE id = $2 AND tenant_id = $3
        """,
        new_document_id, document_id, tenant_id,
    )

const UPLOAD_CONFIG = {
  allowedExtensions: [".pdf", ".jpg", ".jpeg", ".png", ".tiff"],
  minSizeBytes: 5 * 1024,        // 5 KB
  maxSizeBytes: 20 * 1024 * 1024, // 20 MB
};

function validateFile(file: File): ValidationResult {
  const ext = getExtension(file.name).toLowerCase();

  if (!UPLOAD_CONFIG.allowedExtensions.includes(ext)) {
    return { valid: false, error: `Unsupported file type: ${ext}` };
  }
  if (file.size < UPLOAD_CONFIG.minSizeBytes) {
    return { valid: false, error: "File is too small (minimum 5 KB)" };
  }
  if (file.size > UPLOAD_CONFIG.maxSizeBytes) {
    return { valid: false, error: "File is too large (maximum 20 MB)" };
  }
  return { valid: true };
}

const handleUpload = async (file: File) => {
  const validation = validateFile(file);
  if (!validation.valid) {
    toast.error(validation.error, {
      position: "bottom-center",
      duration: 5000,
    });
    return;
  }
  // Proceed with presign + upload flow
};