/** * QC Engine — 4-stage AI pipeline. * IDENTICAL port from Python qc-report-generator/src/qc/checker.py + src/engines/claude_engine.py */ import type Anthropic from "@anthropic-ai/sdk"; import { getClaudeClient } from "../ai/claude-client"; import type { QCIssue, QCLine, QCModule, QCSeverity, QCSummary, QCResult } from "./types"; import { runRuleChecks } from "./rule-checks"; import { runNetflixChecks } from "./netflix-checks"; import { parseNeonStep3, getSourceTextForTimeRange } from "../neon/parser"; import { type KNPData, formatKNPForPrompt } from "./knp-parser"; import { QC_SYSTEM_PROMPT, VALIDATION_SYSTEM_PROMPT, SECOND_PASS_SYSTEM_PROMPT, AMAZON_RULES, AMAZON_SYSTEM_EXTRA, NETFLIX_RULES, buildCheckPrompt, buildValidationPrompt, buildSecondPassPrompt, } from "./prompts"; const BATCH_SIZE = 15; const CONTEXT_LINES = 3; const MAX_RETRIES = 8; const BACKOFF_BASE_429 = 5; const BACKOFF_CAP_429 = 60; const BACKOFF_BASE_ERR = 2; const BACKOFF_CAP_ERR = 15; const JITTER_FACTOR = 0.3; const log = (msg: string) => console.log(`[QC:engine] ${msg}`); // Global semaphore — max 50 concurrent Claude API calls across all files const MAX_CONCURRENCY = 50; let activeCalls = 0; const waitQueue: Array<() => void> = []; async function withSemaphore(fn: () => Promise): Promise { if (activeCalls >= MAX_CONCURRENCY) { await new Promise((resolve) => waitQueue.push(resolve)); } activeCalls++; try { return await fn(); } finally { activeCalls--; const next = waitQueue.shift(); if (next) next(); } } /** Parse timecode string to milliseconds (supports HH:MM:SS:FF and HH:MM:SS,mmm) */ function tcToMsHelper(tc: string, fps: number = 23.976): number { const frameParts = tc.match(/^(\d{2}):(\d{2}):(\d{2}):(\d{2})$/); if (frameParts) { const [, h, m, s, f] = frameParts.map(Number); return Math.round((h * 3600 + m * 60 + s) * 1000 + (f / fps) * 1000); } const msParts = tc.match(/^(\d{2}):(\d{2}):(\d{2})[.,](\d{3})$/); if (msParts) { const [, h, m, s, ms] = msParts.map(Number); return h * 3600000 + m * 60000 + s * 1000 + ms; } return 0; } // ─── Retry Logic (identical to Python claude_engine.py) ───────────────────── function isRetryable(err: Error): { retryable: boolean; isRateLimit: boolean } { const s = err.message.toLowerCase(); if (s.includes("429") || s.includes("rate") || s.includes("overloaded")) return { retryable: true, isRateLimit: true }; for (const code of ["500", "502", "503", "504", "529"]) if (s.includes(code)) return { retryable: true, isRateLimit: false }; if (s.includes("timeout")) return { retryable: true, isRateLimit: false }; if (s.includes("could not resolve authentication")) return { retryable: true, isRateLimit: true }; return { retryable: false, isRateLimit: false }; } function backoffSeconds(attempt: number, isRateLimit: boolean): number { const base = isRateLimit ? Math.min(BACKOFF_CAP_429, BACKOFF_BASE_429 * 2 ** Math.min(attempt, 6)) : Math.min(BACKOFF_CAP_ERR, BACKOFF_BASE_ERR * 2 ** Math.min(attempt, 4)); const jitter = base * JITTER_FACTOR * (2 * Math.random() - 1); return Math.max(1, base + jitter); } async function callWithRetry( client: any, // Anthropic | AnthropicVertex — both have .messages.create() system: string, user: string, label: string, ): Promise { let lastError: Error | null = null; for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { try { const t0 = Date.now(); const response: any = await withSemaphore(() => client.messages.create({ model: "claude-opus-4-6", max_tokens: 16384, system, messages: [{ role: "user", content: user }], })); const text = response.content .filter((b: any) => b.type === "text") .map((b: any) => b.text) .join(""); const dur = ((Date.now() - t0) / 1000).toFixed(1); log(`${label} — ${dur}s, prompt ${user.length} chars → response ${text.length} chars`); return text; } catch (err) { lastError = err as Error; const { retryable, isRateLimit } = isRetryable(lastError); if (!retryable || attempt >= MAX_RETRIES) { log(`${label} FAILED after ${attempt + 1} attempts: ${lastError.message}`); throw lastError; } const wait = backoffSeconds(attempt, isRateLimit); const tag = isRateLimit ? "429 rate limit" : "transient error"; log(`${tag} on ${label} (attempt ${attempt + 1}/${MAX_RETRIES}): ${lastError.message.substring(0, 80)}... retrying in ${wait.toFixed(1)}s`); await new Promise((r) => setTimeout(r, wait * 1000)); } } throw lastError!; } // ─── JSON Parsing (identical to Python base.py parse_issues / parse_validation_response) ─ function parseIssues(responseText: string, lines: QCLine[]): QCIssue[] { let text = responseText.trim(); if (text.startsWith("```")) { text = text.split("\n").filter((l) => !l.trim().startsWith("```")).join("\n"); } const jsonMatch = text.match(/\[[\s\S]*\]/); if (!jsonMatch) return []; try { const raw = JSON.parse(jsonMatch[0]); if (!Array.isArray(raw)) return []; const lineLookup = new Map(lines.map((l) => [l.id, l.text])); return raw .filter((r: any) => typeof r === "object" && r.line_id != null) .map((r: any) => ({ lineId: r.line_id, severity: (r.severity || "warning").replace("error", "critical") as QCSeverity, category: r.category || "grammar", originalText: lineLookup.get(r.line_id) || "", description: r.description || "", suggestedFix: r.suggested_fix || "", flaggedText: r.flagged_text || "", })); } catch { return []; } } function parseValidationResponse(responseText: string): Array<{ issue_index: number; verdict: string; reason?: string; severity?: string }> { let text = responseText.trim(); if (text.startsWith("```")) { text = text.split("\n").filter((l) => !l.trim().startsWith("```")).join("\n"); } const jsonMatch = text.match(/\[[\s\S]*\]/); if (!jsonMatch) return []; try { const verdicts = JSON.parse(jsonMatch[0]); if (!Array.isArray(verdicts)) return []; return verdicts.filter((v: any) => typeof v === "object" && "verdict" in v); } catch { return []; } } // ─── Module Config ────────────────────────────────────────────────────────── function getModuleConfig(module: QCModule) { if (module === "amazon") return { rules: AMAZON_RULES, systemExtra: AMAZON_SYSTEM_EXTRA }; return { rules: NETFLIX_RULES, systemExtra: "" }; } function sevRank(severity: string): number { return { critical: 0, warning: 1, info: 2 }[severity] ?? 1; } function buildSummary(issues: QCIssue[], totalLines: number): QCSummary { const byCategory: Record = {}; let critical = 0, warnings = 0, info = 0; for (const issue of issues) { byCategory[issue.category] = (byCategory[issue.category] || 0) + 1; if (issue.severity === "critical") critical++; else if (issue.severity === "warning") warnings++; else info++; } return { totalLines, totalIssues: issues.length, critical, warnings, info, byCategory }; } // ─── Pipeline (identical to Python checker.py) ────────────────────────────── export type LogCallback = (message: string, data?: any) => void; export async function runQCPipelineWithLogs( lines: QCLine[], language: string, module: QCModule, filename: string, deliverableType: string = "sub", neonSource?: any, onLog?: LogCallback, knpData?: KNPData, ): Promise { return runQCPipeline(lines, language, module, filename, deliverableType, neonSource, onLog, knpData); } export async function runQCPipeline( lines: QCLine[], language: string, module: QCModule, filename: string, deliverableType: string = "sub", neonSource?: any, onLog?: LogCallback, knpData?: KNPData, ): Promise { const start = Date.now(); const emit = (msg: string, data?: any) => { log(msg); onLog?.(msg, data); }; if (!lines.length) { return { filename, issues: [], summary: buildSummary([], 0), module, language, engine: "claude", durationMs: 0 }; } emit(`Pipeline start: ${filename}, ${lines.length} lines, module=${module}, lang=${language}, type=${deliverableType}`); const { client, authMethod } = getClaudeClient(); emit(`Claude auth: ${authMethod}`); const { rules, systemExtra } = getModuleConfig(module); // ─── Stage 1: AI Batch QC ─── const batches: QCLine[][] = []; for (let i = 0; i < lines.length; i += BATCH_SIZE) { batches.push(lines.slice(i, i + BATCH_SIZE)); } emit(`Stage 1: AI Batch QC — ${batches.length} batches of ${BATCH_SIZE}`); const s1 = Date.now(); const batchContexts = batches.map((batch, i) => { const startIdx = i * BATCH_SIZE; return { batch, before: lines.slice(Math.max(0, startIdx - CONTEXT_LINES), startIdx), after: lines.slice(startIdx + batch.length, Math.min(lines.length, startIdx + batch.length + CONTEXT_LINES)), }; }); // Parse Neon source for source context (if provided) const parsedNeon = neonSource ? parseNeonStep3(neonSource) : null; const aiIssues: QCIssue[] = []; // All batches in parallel (like Python asyncio.gather) const batchResults = await Promise.all( batchContexts.map(async ({ batch, before, after }, idx) => { const lineRange = `L${batch[0].id}-L${batch[batch.length - 1].id}`; const system = systemExtra ? `${QC_SYSTEM_PROMPT}\n\n${systemExtra}` : QC_SYSTEM_PROMPT; // Build source context from Neon transcription (time-aligned per line) let sourceContext: Array<{ lineId: number; sourceText: string }> | undefined; if (parsedNeon && batch[0]?.tcIn) { sourceContext = batch .filter((l) => l.tcIn && l.tcOut) .map((l) => { // Parse timecodes — lines may have HH:MM:SS:FF or HH:MM:SS,mmm format const tcIn = l.tcIn!; const tcOut = l.tcOut!; const startMs = tcToMsHelper(tcIn); const endMs = tcToMsHelper(tcOut); const sourceText = getSourceTextForTimeRange(parsedNeon.transcription, startMs, endMs); return sourceText ? { lineId: l.id, sourceText } : null; }) .filter((s): s is { lineId: number; sourceText: string } => s !== null); } const prompt = buildCheckPrompt( batch, language, deliverableType, rules, systemExtra, before.length > 0 ? before : undefined, after.length > 0 ? after : undefined, sourceContext, ); try { const text = await callWithRetry(client, system, prompt, `batch ${idx + 1}/${batches.length} ${lineRange}`); return parseIssues(text, lines); } catch (err) { emit(`Batch ${idx + 1} error (continuing): ${(err as Error).message.substring(0, 80)}`); return []; } }), ); for (const r of batchResults) aiIssues.push(...r); emit(`Stage 1 done: ${aiIssues.length} AI issues in ${((Date.now() - s1) / 1000).toFixed(1)}s`); // ─── Stage 2: Rule-based checks ─── emit(`Stage 2: Rule-based checks`); const s2 = Date.now(); let ruleIssues = runRuleChecks(lines, language); if (module === "netflix") { const nfx = runNetflixChecks(lines, language); emit(`Stage 2: Netflix checks found ${nfx.length} issues`); ruleIssues.push(...nfx); } emit(`Stage 2 done: ${ruleIssues.length} rule issues in ${Date.now() - s2}ms`); // Split rule issues: OFFENSIVE terms (critical sensitive) go direct to final, rest go through AI const directIssues = ruleIssues.filter( (i) => i.category === "sensitive" && i.severity === "critical" && i.description.startsWith("Netflix offensive"), ); ruleIssues = ruleIssues.filter( (i) => !(i.category === "sensitive" && i.severity === "critical" && i.description.startsWith("Netflix offensive")), ); if (directIssues.length > 0) { emit(`${directIssues.length} offensive terms added directly (skip AI)`); } // ─── Stage 2b: AI validation of rule issues (POTENTIALLY_OFFENSIVE + other rule issues) ─── if (ruleIssues.length > 0) { emit(`Stage 2b: AI validation of ${ruleIssues.length} rule issues`); const s2b = Date.now(); const beforeCount = ruleIssues.length; try { const prompt = buildValidationPrompt(ruleIssues, language, deliverableType, lines); const text = await callWithRetry(client, VALIDATION_SYSTEM_PROMPT, prompt, "rule validation"); const verdicts = parseValidationResponse(text); const verdictMap = new Map(); for (const v of verdicts) { if (typeof v.issue_index === "number" && v.issue_index >= 0 && v.issue_index < ruleIssues.length) { verdictMap.set(v.issue_index, v); } } const validated: QCIssue[] = []; for (let i = 0; i < ruleIssues.length; i++) { const v = verdictMap.get(i); if (!v) { validated.push(ruleIssues[i]); continue; } if (v.verdict === "reject") continue; if (v.verdict === "downgrade") { const downgradeMap: Record = { critical: "warning", warning: "info", info: "info" }; ruleIssues[i].severity = downgradeMap[ruleIssues[i].severity] || "info"; if (v.reason) ruleIssues[i].description += ` [Review note: ${v.reason}]`; } validated.push(ruleIssues[i]); } ruleIssues = validated; emit(`Stage 2b done: ${ruleIssues.length} confirmed, ${beforeCount - ruleIssues.length} rejected in ${((Date.now() - s2b) / 1000).toFixed(1)}s`); } catch (err) { emit(`Stage 2b validation error (keeping unvalidated): ${(err as Error).message.substring(0, 80)}`); } } // ─── Stage 3: Merge & deduplicate (identical to Python) ─── emit(`Stage 3: Merge & deduplicate`); const issueMap = new Map(); // AI issues first for (const issue of aiIssues) { const key = `${issue.lineId}:${issue.category}`; const existing = issueMap.get(key); if (!existing || sevRank(issue.severity) < sevRank(existing.severity)) { issueMap.set(key, issue); } } // Rule issues — only add if not already covered by AI or higher severity for (const issue of ruleIssues) { const key = `${issue.lineId}:${issue.category}`; const existing = issueMap.get(key); if (!existing || sevRank(issue.severity) <= sevRank(existing.severity)) { issueMap.set(key, issue); } } // Add direct offensive terms (they bypassed AI) for (const issue of directIssues) { const key = `${issue.lineId}:${issue.category}:offensive`; issueMap.set(key, issue); } let merged = [...issueMap.values()]; merged.sort((a, b) => a.lineId !== b.lineId ? a.lineId - b.lineId : sevRank(a.severity) - sevRank(b.severity)); emit(`Stage 3 done: ${aiIssues.length} AI + ${ruleIssues.length} rule + ${directIssues.length} offensive → ${merged.length} merged`); // ─── Stage 4: Second validation pass (identical to Python) ─── emit(`Stage 4: Second validation pass on ${merged.length} issues`); const s4 = Date.now(); try { const prompt = buildSecondPassPrompt(merged, lines, language, deliverableType); const text = await callWithRetry(client, SECOND_PASS_SYSTEM_PROMPT, prompt, "second pass"); const verdicts = parseValidationResponse(text); const verdictMap = new Map(); for (const v of verdicts) { if (typeof v.issue_index === "number" && v.issue_index >= 0 && v.issue_index < merged.length) { verdictMap.set(v.issue_index, v); } } const validated: QCIssue[] = []; let confirmed = 0, rejected = 0, downgraded = 0; for (let i = 0; i < merged.length; i++) { const v = verdictMap.get(i); if (!v) { validated.push(merged[i]); confirmed++; continue; } if (v.verdict === "reject") { rejected++; continue; } if (v.verdict === "downgrade") { downgraded++; // Downgrade by one level: critical→warning, warning→info const currentSev = merged[i].severity; const downgradeMap: Record = { critical: "warning", warning: "info", info: "info" }; merged[i].severity = downgradeMap[currentSev] || "info"; if (v.reason) merged[i].description += ` [Note: ${v.reason}]`; } else { confirmed++; } validated.push(merged[i]); } merged = validated; emit(`Stage 4 done: ${confirmed} confirmed, ${rejected} rejected, ${downgraded} downgraded in ${((Date.now() - s4) / 1000).toFixed(1)}s`); } catch (err) { emit(`Stage 4 second pass error (keeping unvalidated): ${(err as Error).message.substring(0, 80)}`); } // ─── Stage 5 (optional): KNP Terminology Consistency Check ─── if (knpData && knpData.entries.length > 0) { emit(`Stage 5: KNP consistency check (${knpData.entries.length} terms, ${knpData.sourceLanguage}→${knpData.targetLanguage})`); const s5 = Date.now(); try { const knpPrompt = formatKNPForPrompt(knpData); const knpBatchSize = 15; const knpBatches: QCLine[][] = []; for (let i = 0; i < lines.length; i += knpBatchSize) { knpBatches.push(lines.slice(i, i + knpBatchSize)); } const knpResults = await Promise.all( knpBatches.map(async (batch, idx) => { const lineText = batch.map((l) => `Line ${l.id}: ${l.text}`).join("\n"); const prompt = `${knpPrompt}\n\nCheck these subtitles for KNP terminology consistency:\n\n${lineText}\n\nFor each line where a KNP source term should appear but the approved target translation is NOT used (or is inconsistent), return a JSON array of issues:\n\`\`\`json\n[\n { "line_id": , "severity": "warning", "category": "consistency", "flagged_text": "", "description": "", "suggested_fix": "" }\n]\n\`\`\`\nIf no KNP issues found, return [].`; try { const text = await callWithRetry(client, "You are a terminology consistency checker for subtitle QC.", prompt, `knp batch ${idx + 1}/${knpBatches.length}`); return parseIssues(text, lines); } catch { return []; } }), ); const knpIssues = knpResults.flat(); emit(`Stage 5 done: ${knpIssues.length} KNP issues in ${((Date.now() - s5) / 1000).toFixed(1)}s`); merged.push(...knpIssues); } catch (err) { emit(`Stage 5 KNP error (non-fatal): ${(err as Error).message.substring(0, 80)}`); } } // ─── Stage 6: AI Translator Credit Check (last 3 subtitles) ─── if (module === "netflix" && lines.length > 0) { emit(`Stage 6: AI translator credit check`); const s6 = Date.now(); try { const lastLines = lines.slice(-3); const lastText = lastLines.map((l) => `Line ${l.id}: ${l.text}`).join("\n"); const creditPrompt = `Check if any of these last subtitles contain a translator/subtitler credit. Credits typically say "Translated by", "Subtitled by", "Subtitle translation by" or equivalent in any language.\n\nSubtitles:\n${lastText}\n\nRespond with JSON:\n\`\`\`json\n[\n { "line_id": , "severity": "warning", "category": "formatting", "flagged_text": "", "description": "Translator credit detected — verify it follows Netflix credit guidelines", "suggested_fix": "Verify against Netflix Originals Credit Translation document" }\n]\n\`\`\`\nIf no credits found, return [].`; const creditText = await callWithRetry(client, "You detect translator credits in subtitle files. Be precise — only flag actual credits, not normal dialogue.", creditPrompt, "credit check"); const creditIssues = parseIssues(creditText, lines); if (creditIssues.length > 0) { emit(`Stage 6 done: ${creditIssues.length} credit issue(s) in ${((Date.now() - s6) / 1000).toFixed(1)}s`); merged.push(...creditIssues); } else { emit(`Stage 6 done: no credits detected in ${((Date.now() - s6) / 1000).toFixed(1)}s`); } } catch (err) { emit(`Stage 6 credit check error (non-fatal): ${(err as Error).message.substring(0, 80)}`); } } // Final sort merged.sort((a, b) => a.lineId !== b.lineId ? a.lineId - b.lineId : sevRank(a.severity) - sevRank(b.severity)); const durationMs = Date.now() - start; emit(`Pipeline complete: ${merged.length} final issues in ${(durationMs / 1000).toFixed(1)}s`); // Build timecode lookup for report const lineTimecodes: Record = {}; for (const line of lines) { if (line.tcIn && line.tcOut) { lineTimecodes[line.id] = { tcIn: line.tcIn, tcOut: line.tcOut }; } } return { filename, issues: merged, summary: buildSummary(merged, lines.length), module, language, engine: "claude", durationMs, lineTimecodes, }; }