xiaobailing-turn-detection/labels/eot-studio/app/api/workspace/route.ts

import { randomUUID } from "node:crypto";
import { promises as fs } from "node:fs";
import path from "node:path";
import {
  exportAnnotations,
  exportTranscripts,
  parseAnnotations,
  parseTranscripts,
  type Annotation,
  type Transcript,
} from "@/lib/annotations";
import { isSameOriginRequest } from "@/lib/request-origin";
import type { CallReference } from "@/lib/call-reference";

export const runtime = "nodejs";
export const dynamic = "force-dynamic";

const projectRoot = path.resolve(process.cwd(), "..", "..");
const dataRoot = path.join(projectRoot, "data");
const labelsRoot = path.join(projectRoot, "labels", "EOT标注");
const savedRoot = path.join(labelsRoot, "工作数据");
const exampleRoot = path.join(labelsRoot, "示例");
const callReferencesPath = path.join(process.cwd(), "data", "hantai-reference.json");
const lockPath = path.join(savedRoot, ".workspace.lock");
const extensions = new Set([".mp3", ".wav", ".m4a", ".ogg", ".flac", ".webm"]);
const maxPayloadBytes = 20 * 1024 * 1024;

type Recording = { call_id: string; name: string; path: string; url: string };
type WorkspaceRuntime = typeof globalThis & { __eotWorkspaceQueue?: Promise<void> };

class WorkspaceError extends Error {
  constructor(message: string, readonly status = 422) {
    super(message);
  }
}

function hasCode(error: unknown, code: string): boolean {
  return error instanceof Error && "code" in error && error.code === code;
}

function json(value: unknown, status = 200): Response {
  return Response.json(value, { status, headers: { "Cache-Control": "no-store" } });
}

function errorResponse(error: unknown): Response {
  if (error instanceof WorkspaceError) return json({ error: error.message }, error.status);
  console.error("EOT workspace request failed", error);
  return json({ error: "无法读取或保存项目文件，请检查目录和文件权限。" }, 500);
}

// Share the queue across development hot reloads. The disk lock also prevents
// two server processes from reading or replacing different halves of a save.
function serialize<T>(operation: () => Promise<T>): Promise<T> {
  const state = globalThis as WorkspaceRuntime;
  const pending = (state.__eotWorkspaceQueue ?? Promise.resolve()).then(operation);
  state.__eotWorkspaceQueue = pending.then(() => undefined, () => undefined);
  return pending;
}

async function exists(file: string): Promise<boolean> {
  try {
    await fs.access(file);
    return true;
  } catch (error) {
    if (hasCode(error, "ENOENT")) return false;
    throw error;
  }
}

async function withDiskLock<T>(operation: () => Promise<T>): Promise<T> {
  let handle;
  try {
    handle = await fs.open(lockPath, "wx", 0o600);
  } catch (error) {
    if (hasCode(error, "EEXIST")) {
      throw new WorkspaceError("另一个操作正在读取或保存工作数据，请稍后重试。", 409);
    }
    throw error;
  }
  try {
    await handle.writeFile(JSON.stringify({ pid: process.pid, startedAt: new Date().toISOString() }));
    return await operation();
  } finally {
    try {
      await handle.close();
    } finally {
      await fs.unlink(lockPath).catch((error: unknown) => {
        if (!hasCode(error, "ENOENT")) throw error;
      });
    }
  }
}

async function discoverRecordings(): Promise<Recording[]> {
  const result: Recording[] = [];
  async function visit(directory: string): Promise<void> {
    const entries = await fs.readdir(directory, { withFileTypes: true });
    entries.sort((left, right) => left.name.localeCompare(right.name, "en"));
    for (const entry of entries) {
      const absolute = path.join(directory, entry.name);
      if (entry.isDirectory()) {
        await visit(absolute);
      } else if (entry.isFile() && extensions.has(path.extname(entry.name).toLowerCase())) {
        const relative = path.relative(dataRoot, absolute).split(path.sep).join("/");
        result.push({
          call_id: path.basename(entry.name, path.extname(entry.name)),
          name: entry.name,
          path: relative,
          url: `/api/audio?path=${encodeURIComponent(relative)}`,
        });
      }
    }
  }
  try {
    await visit(dataRoot);
  } catch (error) {
    if (!hasCode(error, "ENOENT")) throw error;
  }
  return result;
}

async function readCallReferences(): Promise<CallReference[]> {
  let source: string;
  try {
    source = await fs.readFile(callReferencesPath, "utf8");
  } catch (error) {
    if (hasCode(error, "ENOENT")) return [];
    throw error;
  }
  try {
    const references: unknown = JSON.parse(source);
    if (!Array.isArray(references)) throw new Error("顶层必须为数组。");
    return references.map((value, index) => {
      const message = `第 ${index + 1} 条原表记录格式无效。`;
      if (!value || typeof value !== "object") throw new Error(message);
      const row = value as Partial<CallReference>;
      if (typeof row.call_id !== "string" || !row.call_id.trim()
        || !row.source || typeof row.source.file !== "string" || typeof row.source.sheet !== "string"
        || !Number.isSafeInteger(row.source.row) || row.source.row < 1
        || !Array.isArray(row.fields)
        || row.fields.some((field) => !field || typeof field.label !== "string" || typeof field.value !== "string")) {
        throw new Error(message);
      }
      return {
        call_id: row.call_id,
        source: { file: row.source.file, sheet: row.source.sheet, row: row.source.row },
        fields: row.fields.map((field) => ({ label: field.label, value: field.value })),
      };
    });
  } catch (error) {
    throw new WorkspaceError(`原表参考数据格式有误：${error instanceof Error ? error.message : "JSON 格式错误。"}`);
  }
}

async function readCsvPair(directory: string): Promise<{ transcripts: Transcript[]; annotations: Annotation[] }> {
  const [transcriptsCsv, annotationsCsv] = await Promise.all([
    fs.readFile(path.join(directory, "asr_history.csv"), "utf8"),
    fs.readFile(path.join(directory, "eot_labels.csv"), "utf8"),
  ]);
  try {
    return { transcripts: parseTranscripts(transcriptsCsv), annotations: parseAnnotations(annotationsCsv) };
  } catch (error) {
    const detail = error instanceof Error ? error.message : "CSV 格式错误";
    throw new WorkspaceError(`工作数据格式有误：${detail}`);
  }
}

async function savedAt(): Promise<string> {
  const stats = await Promise.all([
    fs.stat(path.join(savedRoot, "asr_history.csv")),
    fs.stat(path.join(savedRoot, "eot_labels.csv")),
  ]);
  return new Date(Math.max(...stats.map((stat) => stat.mtimeMs))).toISOString();
}

export async function GET(): Promise<Response> {
  try {
    return await serialize(async () => {
      if (await exists(lockPath)) {
        throw new WorkspaceError("工作数据正在保存，请稍后重试。", 409);
      }
      const [hasTranscripts, hasAnnotations, recordings, callReferences] = await Promise.all([
        exists(path.join(savedRoot, "asr_history.csv")),
        exists(path.join(savedRoot, "eot_labels.csv")),
        discoverRecordings(),
        readCallReferences(),
      ]);
      if (hasTranscripts && hasAnnotations) {
        return await withDiskLock(async () => json({
          ...await readCsvPair(savedRoot),
          recordings,
          callReferences,
          source: "saved",
          savedAt: await savedAt(),
        }));
      }
      return json({ ...await readCsvPair(exampleRoot), recordings, callReferences, source: "example" });
    });
  } catch (error) {
    return errorResponse(error);
  }
}

function requireString(row: Record<string, unknown>, name: string, label: string): void {
  if (typeof row[name] !== "string") throw new WorkspaceError(`${label} 的 ${name} 必须是文本。`);
}

function requireMillis(row: Record<string, unknown>, name: string, label: string): void {
  if (!Number.isSafeInteger(row[name]) || Number(row[name]) < 0) {
    throw new WorkspaceError(`${label} 的 ${name} 必须是非负整数毫秒。`);
  }
}

function validateRow(value: unknown, label: string): Record<string, unknown> {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    throw new WorkspaceError(`${label} 必须是一个记录对象。`);
  }
  return value as Record<string, unknown>;
}

function validatePayload(value: unknown): { transcriptsCsv: string; annotationsCsv: string; transcripts: Transcript[]; annotations: Annotation[] } {
  const payload = validateRow(value, "提交内容");
  if (!Array.isArray(payload.transcripts) || !Array.isArray(payload.annotations)) {
    throw new WorkspaceError("提交内容必须包含 transcripts 和 annotations 两个数组。");
  }
  payload.transcripts.forEach((value, index) => {
    const label = `转写第 ${index + 1} 行`;
    const row = validateRow(value, label);
    for (const name of ["call_id", "role", "transcript"]) requireString(row, name, label);
    for (const name of ["start_ms", "end_ms"]) requireMillis(row, name, label);
    if (row.role !== "agent" && row.role !== "user") throw new WorkspaceError(`${label} 的 role 必须是 agent 或 user。`);
  });
  payload.annotations.forEach((value, index) => {
    const label = `EOT 第 ${index + 1} 行`;
    const row = validateRow(value, label);
    for (const name of ["call_id", "user_text_at_decision", "can_reply_now", "confidence", "reason", "annotator"]) {
      requireString(row, name, label);
    }
    requireMillis(row, "decision_ms", label);
    if (!["", "0", "1"].includes(String(row.can_reply_now))) throw new WorkspaceError(`${label} 的 can_reply_now 必须是 0、1 或空。`);
    if (!["", "high", "medium", "low"].includes(String(row.confidence))) throw new WorkspaceError(`${label} 的 confidence 取值无效。`);
  });
  try {
    const transcriptsCsv = exportTranscripts(payload.transcripts as Transcript[]);
    const annotationsCsv = exportAnnotations(payload.annotations as Annotation[]);
    return {
      transcriptsCsv,
      annotationsCsv,
      transcripts: parseTranscripts(transcriptsCsv),
      annotations: parseAnnotations(annotationsCsv),
    };
  } catch (error) {
    throw new WorkspaceError(error instanceof Error ? error.message : "数据格式有误。");
  }
}

async function replaceCsvPair(transcriptsCsv: string, annotationsCsv: string): Promise<void> {
  const token = `${process.pid}-${randomUUID()}`;
  const entries = [
    { name: "asr_history.csv", content: transcriptsCsv },
    { name: "eot_labels.csv", content: annotationsCsv },
  ].map((entry) => ({
    ...entry,
    destination: path.join(savedRoot, entry.name),
    temporary: path.join(savedRoot, `.${entry.name}.${token}.tmp`),
    backup: path.join(savedRoot, `.${entry.name}.${token}.backup`),
    existed: false,
    replaced: false,
  }));
  let restored = true;
  try {
    for (const entry of entries) {
      entry.existed = await exists(entry.destination);
      if (entry.existed) await fs.copyFile(entry.destination, entry.backup);
      await fs.writeFile(entry.temporary, entry.content, { encoding: "utf8", flag: "wx", mode: 0o600 });
    }
    for (const entry of entries) {
      await fs.rename(entry.temporary, entry.destination);
      entry.replaced = true;
    }
  } catch (error) {
    for (const entry of entries.filter((entry) => entry.replaced)) {
      try {
        if (entry.existed) await fs.rename(entry.backup, entry.destination);
        else await fs.unlink(entry.destination);
      } catch (rollbackError) {
        restored = false;
        console.error("EOT workspace rollback failed; backup retained", entry.backup, rollbackError);
      }
    }
    if (!restored) throw new WorkspaceError("保存未完成，恢复备份时发生错误。请检查工作数据目录中的 backup 文件。", 500);
    throw error;
  } finally {
    for (const entry of entries) {
      await fs.rm(entry.temporary, { force: true });
      if (restored) await fs.rm(entry.backup, { force: true });
    }
  }
}

export async function POST(request: Request): Promise<Response> {
  try {
    if (!isSameOriginRequest(request)) {
      throw new WorkspaceError("仅允许当前工具页面保存工作数据。", 403);
    }
    const raw = await request.text();
    if (Buffer.byteLength(raw, "utf8") > maxPayloadBytes) throw new WorkspaceError("提交数据超过 20 MB，请分批整理。", 413);
    let payload;
    try {
      payload = JSON.parse(raw);
    } catch {
      throw new WorkspaceError("提交内容不是有效 JSON。", 400);
    }
    const validated = validatePayload(payload);
    return await serialize(async () => {
      await fs.mkdir(savedRoot, { recursive: true });
      return await withDiskLock(async () => {
        await replaceCsvPair(validated.transcriptsCsv, validated.annotationsCsv);
        return json({
          transcripts: validated.transcripts,
          annotations: validated.annotations,
          source: "saved",
          savedAt: await savedAt(),
        });
      });
    });
  } catch (error) {
    return errorResponse(error);
  }
}