Add OCR rate limit handling and batch reprocess action
Rate limiting: - Retry with exponential backoff (5s, 10s, 20s, 40s) on rate limit errors - Reduce concurrent OCR jobs from 2 to 1 - Add 2s delay between AI calls within a card (response vs survey side) - Add 3s delay between cards in a multi-page PDF - Add 3s delay between jobs in the processing queue Batch reprocess: - New /api/cards/reprocess-batch endpoint processes cards sequentially with delays to respect rate limits - Reprocess button added to selection toolbar on dashboard Made-with: Cursor
This commit is contained in:
parent
9b27041970
commit
1d20932ef5
6 changed files with 155 additions and 21 deletions
62
src/app/api/cards/reprocess-batch/route.ts
Normal file
62
src/app/api/cards/reprocess-batch/route.ts
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
import { NextRequest, NextResponse } from "next/server";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { reprocessCard } from "@/lib/ocr";
|
||||
|
||||
const DELAY_BETWEEN_CARDS_MS = 3_000;
|
||||
|
||||
export async function POST(request: NextRequest) {
|
||||
try {
|
||||
const body = await request.json().catch(() => ({}));
|
||||
const ids = body.ids as string[] | undefined;
|
||||
|
||||
if (!ids || !Array.isArray(ids) || ids.length === 0) {
|
||||
return NextResponse.json({ error: "No card IDs provided" }, { status: 400 });
|
||||
}
|
||||
|
||||
const cards = await prisma.responseCard.findMany({
|
||||
where: { id: { in: ids } },
|
||||
select: { id: true, ocrStatus: true, backImagePath: true, frontImagePath: true },
|
||||
});
|
||||
|
||||
const eligible = cards.filter(
|
||||
(c) => c.ocrStatus !== "processing" && (c.backImagePath || c.frontImagePath)
|
||||
);
|
||||
|
||||
if (eligible.length === 0) {
|
||||
return NextResponse.json(
|
||||
{ error: "No eligible cards (already processing or no images)" },
|
||||
{ status: 400 }
|
||||
);
|
||||
}
|
||||
|
||||
await prisma.responseCard.updateMany({
|
||||
where: { id: { in: eligible.map((c) => c.id) } },
|
||||
data: { ocrStatus: "processing", ocrError: null },
|
||||
});
|
||||
|
||||
(async () => {
|
||||
for (let i = 0; i < eligible.length; i++) {
|
||||
try {
|
||||
await reprocessCard(eligible[i].id);
|
||||
} catch (err) {
|
||||
console.error(`[reprocess-batch] Card ${eligible[i].id} failed:`, err);
|
||||
}
|
||||
if (i < eligible.length - 1) {
|
||||
await new Promise((r) => setTimeout(r, DELAY_BETWEEN_CARDS_MS));
|
||||
}
|
||||
}
|
||||
console.log(`[reprocess-batch] Completed ${eligible.length} card(s)`);
|
||||
})().catch((err) => {
|
||||
console.error("[reprocess-batch] Background task failed:", err);
|
||||
});
|
||||
|
||||
return NextResponse.json({
|
||||
ok: true,
|
||||
queued: eligible.length,
|
||||
skipped: ids.length - eligible.length,
|
||||
});
|
||||
} catch (error) {
|
||||
console.error("[reprocess-batch POST]", error);
|
||||
return NextResponse.json({ error: "Failed to start batch reprocessing" }, { status: 500 });
|
||||
}
|
||||
}
|
||||
|
|
@ -145,6 +145,28 @@ export function DashboardContent() {
|
|||
}
|
||||
};
|
||||
|
||||
const handleBulkReprocess = async (ids: string[]) => {
|
||||
try {
|
||||
const res = await fetch("/api/cards/reprocess-batch", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ ids }),
|
||||
});
|
||||
const data = await res.json();
|
||||
if (res.ok) {
|
||||
toast.success(
|
||||
`Reprocessing ${data.queued} card(s)${data.skipped > 0 ? ` (${data.skipped} skipped)` : ""}`
|
||||
);
|
||||
setSelectedIds([]);
|
||||
fetchCards();
|
||||
} else {
|
||||
toast.error(data.error || "Batch reprocess failed");
|
||||
}
|
||||
} catch {
|
||||
toast.error("Failed to start batch reprocessing");
|
||||
}
|
||||
};
|
||||
|
||||
const handleExportCsv = () => {
|
||||
if (data.length === 0) {
|
||||
toast.error("No data to export");
|
||||
|
|
@ -344,6 +366,7 @@ export function DashboardContent() {
|
|||
selectedIds={selectedIds}
|
||||
onMarkReviewed={(ids) => handleBulkAction(ids, "reviewed")}
|
||||
onMarkExported={(ids) => handleBulkAction(ids, "exported")}
|
||||
onReprocess={handleBulkReprocess}
|
||||
onDelete={(ids) => handleBulkAction(ids, "delete")}
|
||||
onClear={() => setSelectedIds([])}
|
||||
/>
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
import {
|
||||
CheckCircle,
|
||||
Download,
|
||||
RefreshCw,
|
||||
Trash2,
|
||||
X,
|
||||
} from "lucide-react";
|
||||
|
|
@ -13,6 +14,7 @@ interface SelectionToolbarProps {
|
|||
selectedIds: string[];
|
||||
onMarkReviewed?: (ids: string[]) => void;
|
||||
onMarkExported?: (ids: string[]) => void;
|
||||
onReprocess?: (ids: string[]) => void;
|
||||
onDelete?: (ids: string[]) => void;
|
||||
onClear: () => void;
|
||||
}
|
||||
|
|
@ -21,6 +23,7 @@ export function SelectionToolbar({
|
|||
selectedIds,
|
||||
onMarkReviewed,
|
||||
onMarkExported,
|
||||
onReprocess,
|
||||
onDelete,
|
||||
onClear,
|
||||
}: SelectionToolbarProps) {
|
||||
|
|
@ -63,6 +66,17 @@ export function SelectionToolbar({
|
|||
<span className="hidden sm:inline ml-1">Export</span>
|
||||
</Button>
|
||||
)}
|
||||
{onReprocess && (
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
className="rounded-xl"
|
||||
onClick={() => onReprocess(selectedIds)}
|
||||
>
|
||||
<RefreshCw className="size-4" />
|
||||
<span className="hidden sm:inline ml-1">Reprocess</span>
|
||||
</Button>
|
||||
)}
|
||||
{onDelete && (
|
||||
<Button
|
||||
variant="ghost"
|
||||
|
|
|
|||
|
|
@ -99,32 +99,53 @@ export interface OcrResult {
|
|||
side: "response" | "survey";
|
||||
}
|
||||
|
||||
const MAX_RETRIES = 4;
|
||||
const INITIAL_BACKOFF_MS = 5_000;
|
||||
|
||||
async function extractStructured<T extends Record<string, unknown>>(
|
||||
model: LanguageModel,
|
||||
schema: z.ZodType<T>,
|
||||
systemPrompt: string,
|
||||
imageBase64: string
|
||||
): Promise<T> {
|
||||
const { output, text } = await generateText({
|
||||
model,
|
||||
output: Output.object({ schema }),
|
||||
messages: [
|
||||
{ role: "system", content: systemPrompt },
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "Extract all data from this scanned card image." },
|
||||
{ type: "image", image: Buffer.from(imageBase64, "base64") },
|
||||
],
|
||||
},
|
||||
],
|
||||
});
|
||||
let lastError: unknown;
|
||||
|
||||
if (!output) {
|
||||
throw new Error(`AI model did not return structured output. Raw text: ${(text || "").slice(0, 500)}`);
|
||||
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
|
||||
try {
|
||||
const { output, text } = await generateText({
|
||||
model,
|
||||
output: Output.object({ schema }),
|
||||
messages: [
|
||||
{ role: "system", content: systemPrompt },
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "Extract all data from this scanned card image." },
|
||||
{ type: "image", image: Buffer.from(imageBase64, "base64") },
|
||||
],
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
if (!output) {
|
||||
throw new Error(`AI model did not return structured output. Raw text: ${(text || "").slice(0, 500)}`);
|
||||
}
|
||||
|
||||
return output;
|
||||
} catch (err) {
|
||||
lastError = err;
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
const isRateLimit = /rate.?limit|too many requests|429|temporarily|free credits/i.test(msg);
|
||||
|
||||
if (!isRateLimit || attempt === MAX_RETRIES) throw err;
|
||||
|
||||
const backoff = INITIAL_BACKOFF_MS * Math.pow(2, attempt);
|
||||
console.log(`[ai-ocr] Rate limited (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${backoff / 1000}s...`);
|
||||
await new Promise((r) => setTimeout(r, backoff));
|
||||
}
|
||||
}
|
||||
|
||||
return output;
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
export async function ocrImage(
|
||||
|
|
|
|||
|
|
@ -5,6 +5,11 @@ import { ocrImage } from "./ai-ocr";
|
|||
import { pdfToImages, imageToBase64, processUploadedImage } from "./pdf";
|
||||
import { fireIntegrationEvent } from "./integrations";
|
||||
|
||||
const DELAY_BETWEEN_OCR_CALLS_MS = 2_000;
|
||||
function ocrDelay() {
|
||||
return new Promise((r) => setTimeout(r, DELAY_BETWEEN_OCR_CALLS_MS));
|
||||
}
|
||||
|
||||
export async function processFile(
|
||||
jobId: string,
|
||||
fileName: string,
|
||||
|
|
@ -54,7 +59,9 @@ export async function processFile(
|
|||
pairs.push({ response: pageImages[0] });
|
||||
}
|
||||
|
||||
for (const pair of pairs) {
|
||||
for (let pairIdx = 0; pairIdx < pairs.length; pairIdx++) {
|
||||
if (pairIdx > 0) await ocrDelay();
|
||||
const pair = pairs[pairIdx];
|
||||
const card = await prisma.responseCard.create({
|
||||
data: {
|
||||
sourceFile: sourceKey,
|
||||
|
|
@ -85,6 +92,8 @@ export async function processFile(
|
|||
}
|
||||
|
||||
if (pair.survey) {
|
||||
if (pair.response) await ocrDelay();
|
||||
|
||||
const imgKey = `images/${card.id}/survey.jpg`;
|
||||
await uploadBuffer(imgKey, pair.survey.buffer, "image/jpeg");
|
||||
await prisma.responseCard.update({
|
||||
|
|
@ -196,6 +205,8 @@ export async function reprocessCard(cardId: string): Promise<void> {
|
|||
}
|
||||
|
||||
if (card.frontImagePath) {
|
||||
if (card.backImagePath) await ocrDelay();
|
||||
|
||||
const imgBuffer = await getBuffer(card.frontImagePath);
|
||||
const base64 = await imageToBase64(imgBuffer);
|
||||
const result = await ocrImage(base64, "survey");
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import { processFile } from "./ocr";
|
||||
|
||||
const MAX_CONCURRENT = 2;
|
||||
const MAX_CONCURRENT = 1;
|
||||
const DELAY_BETWEEN_JOBS_MS = 3_000;
|
||||
const queue: QueueItem[] = [];
|
||||
let running = 0;
|
||||
|
||||
|
|
@ -22,7 +23,9 @@ function drain() {
|
|||
.catch(item.reject)
|
||||
.finally(() => {
|
||||
running--;
|
||||
drain();
|
||||
if (queue.length > 0) {
|
||||
setTimeout(drain, DELAY_BETWEEN_JOBS_MS);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -35,7 +38,7 @@ export function enqueueProcessing(
|
|||
): Promise<string[]> {
|
||||
return new Promise((resolve, reject) => {
|
||||
queue.push({ jobId, fileName, fileBuffer, isPdf, resolve, reject });
|
||||
drain();
|
||||
if (running < MAX_CONCURRENT) drain();
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue