/**
 * ============================================================
 * server.ts — Express API 服务（供前端 Dashboard 调用）
 * ============================================================
 * 功能：
 *   - POST /api/upload       — 接收上传的 CSV 文件，检查重复，返回预览
 *   - POST /api/upload/confirm — 确认写入 Supabase（支持 append 参数）
 *   - GET  /api/state        — 返回当前服务状态（兼容前端轮询）
 * 
 * 端口：3030（与前端 automation-api.ts 默认 base URL 一致）
 * ============================================================
 */

import "dotenv/config"; // 自动加载 .env 环境变量
import "../core/env.ts"; // 启动期立即校验 env，缺失/格式错时 fail-fast
import { execFile, spawn, type ChildProcess } from "node:child_process";
import { existsSync, writeFileSync } from "node:fs";
import { mkdir, readFile, readdir, rm } from "node:fs/promises";
import path from "path";
import { promisify } from "node:util";
import { fileURLToPath } from "url";
import express, { type NextFunction, type Request, type Response } from "express";
import cors from "cors";
import multer from "multer";
import { parseWangwangContactIdentity } from "../../chatbot/src/contact-identity.js";
import { parseCsvBuffer } from "../core/csv-parser.ts";
import { processCsvBuffer } from "./pipeline.ts";
import {
  analyzeProductsCombined,
  isAbortError,
  resolveLlmRuntimeConfig,
  resolveOnlineStandardizationModel,
} from "../standardization/standardization-llm.ts";
import { mergeRuntimeModelOverrides, setRuntimeModelOverrides } from "../standardization/model-number-override.ts";
import {
} from "../keywords/keyword-generator.ts";
import {
  createBatchTask,
  downloadBatchFileContent,
  resolveBigModelBatchModel,
  resolveBigModelBatchRuntimeConfig,
  retrieveBatchTask,
  type BigModelBatchRecord,
} from "../standardization/bigmodel-batch-client.ts";
import {
  buildOfflineBatchJsonl,
  parseOfflineBatchCustomId,
} from "../standardization/offline-batch-request.ts";
import {
  extractBatchLineMessage,
  extractBatchLineStatusCode,
  parseJsonlLines,
  parseOfflineCombinedResult,
  formatParameterInterpretationsSummary,
  toAiKeywordAnalysis,
  toLlmAnalysisResult,
} from "../standardization/offline-batch-result.ts";
import { buildCategorizedHumanFixHint } from "../standardization/human-fix-hint.ts";
import {
  deleteStandardizedResultsByMaterialIds,
  getLatestKeywordProcess,
  getImageSearchResultByMaterialId,
  getBestMatchImageByMaterialId,
  getKeywordSelectionByMaterialId,
  upsertKeywordProcess,
  deleteKeywordProcessesByMaterialIds,
  createStandardizationTask,
  getStandardizationTaskItemsByTaskId,
  getStandardizationTasksByModeAndStatuses,
  updateStandardizationTask,
  insertStandardizationTaskItems,
  updateStandardizationTaskItem,
  resolveOutstandingStandardizationFailures,
  failPendingStandardizationTaskItems,
  createSearchTask,
  deleteSearchTaskById,
  deleteSearchTaskItemsByTaskId,
  getSearchTaskById,
  getSearchTaskItemsByTaskId,
  getSearchTaskItemsByTaskIds,
  getSearchTasksByStatuses,
  getBatchCreatedBy,
  getProcurementAccountById,
  getProcurementAccountsByIds,
  insertSearchTaskItems,
  loadModelNumberOverridesFromDb,
  markSearchTaskItemsStopped,
  updateSearchTask,
  updateSearchTaskItem,
  upsertModelNumberOverride,
  upsertProcurementDictionaryEntries,
  upsertStandardizedResults,
  supabase,
} from "./db.ts";
import type { ProcurementAccount1688Row } from "./db.ts";
import type {
  RawCsvRow,
  ParsedProduct,
  OfflineCombinedStandardizationResult,
  StandardizationSourceSnapshot,
  StandardizationExecutionMode,
  StandardizedResultRow,
  StandardizationTaskRow,
  StandardizationTaskItemRow,
  SearchPipelineConfig,
  SearchTaskRow,
  SearchTaskItemRow,
  SearchTaskStatus,
} from "../core/types.ts";
import {
  DEFAULT_SEARCH_PIPELINE_CONFIG,
  normalizeSearchPipelineConfig,
} from "../core/types.ts";
import { extractAndInsertMaterialsFromUploadedFile } from "./extract-materials-api.ts";
import { buildExtractMaterialsErrorResponse } from "./extract-materials-errors.ts";
import { resolveDocumentExtractionModel } from "../standardization/document-extraction-llm.ts";
import {
  createBusySearchExecutorStatus,
  createIdleSearchExecutorStatus,
} from "./search-executor-state.ts";
import {
  isSearchAccountExecutionLocked,
  withSearchAccountExecutionLock,
} from "./search-account-lock.ts";
import {
  getSearchTaskRuntimePaths,
  prepareSearchTaskRuntimeProfile,
  syncSearchTaskRuntimeProfileToAccount,
} from "./search-profile-workspace.ts";
import { getEnv } from "../core/env.ts";
import {
  isApiValidationError,
  parseBody,
} from "../core/http-validation.ts";
import { z } from "zod";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const CHATBOT_DIR = path.join(__dirname, "..", "..", "chatbot");
const CHATBOT_WORKER_SCRIPT = path.join(CHATBOT_DIR, "src", "chat-worker.js");
const LOGIN_1688_SCRIPT = path.join(__dirname, "login-1688.js");
const ROW_PIPELINE_SYNC_SCRIPT = path.join(__dirname, "..", "pipelines", "row-pipeline-sync.ts");
const DIST_DIR = path.join(__dirname, "..", "..", "dist");
const DIST_INDEX_FILE = path.join(DIST_DIR, "index.html");
const SEARCH_TASK_RUNTIME_ROOT = path.join(__dirname, "..", ".runtime", "search-tasks");
const execFileAsync = promisify(execFile);

const app = express();
const PORT = getEnv().automationPort;
const hasBuiltFrontend = existsSync(DIST_INDEX_FILE);

type ApiUserRole = "admin" | "user";

interface ApiUserContext {
  userId: string;
  role: ApiUserRole;
  isAdmin: boolean;
}

interface AuthenticatedRequest extends Request {
  apiUser?: ApiUserContext;
}

class ApiAuthorizationError extends Error {
  statusCode: number;

  constructor(statusCode: number, message: string) {
    super(message);
    this.name = "ApiAuthorizationError";
    this.statusCode = statusCode;
  }
}

function isApiAuthorizationError(error: unknown): error is ApiAuthorizationError {
  return error instanceof ApiAuthorizationError;
}

function getBearerToken(req: Request) {
  const authHeader = req.header("authorization")?.trim();
  if (!authHeader) {
    return null;
  }

  const [scheme, token] = authHeader.split(/\s+/, 2);
  if (!/^Bearer$/i.test(scheme) || !token?.trim()) {
    return null;
  }

  return token.trim();
}

async function resolveApiUserContext(req: Request): Promise<ApiUserContext | null> {
  const token = getBearerToken(req);
  if (!token) {
    return null;
  }

  const { data: authData, error: authError } = await supabase.auth.getUser(token);
  if (authError || !authData.user?.id) {
    return null;
  }

  const userId = authData.user.id;
  const { data: profileData, error: profileError } = await supabase
    .from("profiles")
    .select("role")
    .eq("id", userId)
    .maybeSingle();

  if (profileError) {
    console.warn(`[auth] 读取 profile 角色失败: ${profileError.message}`);
  }

  const role: ApiUserRole = profileData?.role === "admin" ? "admin" : "user";
  return {
    userId,
    role,
    isAdmin: role === "admin",
  };
}

async function requireApiUser(req: AuthenticatedRequest, res: Response, next: NextFunction) {
  try {
    const apiUser = await resolveApiUserContext(req);
    if (!apiUser) {
      res.status(401).json({ error: "未登录或登录态已失效" });
      return;
    }

    req.apiUser = apiUser;
    next();
  } catch (error) {
    next(error);
  }
}

function getRequestApiUser(req: AuthenticatedRequest): ApiUserContext {
  if (!req.apiUser) {
    throw new ApiAuthorizationError(401, "未登录或登录态已失效");
  }

  return req.apiUser;
}

function sendApiError(res: Response, error: unknown) {
  if (isApiAuthorizationError(error)) {
    res.status(error.statusCode).json({ error: error.message });
    return;
  }

  if (isApiValidationError(error)) {
    res.status(error.statusCode).json({
      error: error.message,
      issues: error.issues,
    });
    return;
  }

  const message = error instanceof Error ? error.message : String(error);
  res.status(500).json({ error: message });
}

async function getAccessibleBatchIdSet(user: ApiUserContext, batchIds: string[]): Promise<Set<string>> {
  const normalizedBatchIds = [...new Set(batchIds.map((batchId) => batchId.trim()).filter(Boolean))];
  if (normalizedBatchIds.length === 0) {
    return new Set<string>();
  }

  if (user.isAdmin) {
    return new Set(normalizedBatchIds);
  }

  const { data, error } = await supabase
    .from("procurement_batches")
    .select("batch_id")
    .eq("created_by", user.userId)
    .in("batch_id", normalizedBatchIds);

  if (error) {
    throw new Error(`[procurement_batches] 查询可访问批次失败: ${error.message}`);
  }

  return new Set(
    ((data ?? []) as Array<{ batch_id?: string | null }>)
      .map((row) => String(row.batch_id ?? "").trim())
      .filter(Boolean),
  );
}

async function getAccessibleSearchAccountIdSet(user: ApiUserContext, accountIds: string[]): Promise<Set<string>> {
  const normalizedAccountIds = [...new Set(accountIds.map((accountId) => accountId.trim()).filter(Boolean))];
  if (normalizedAccountIds.length === 0) {
    return new Set<string>();
  }

  if (user.isAdmin) {
    return new Set(normalizedAccountIds);
  }

  const { data, error } = await supabase
    .from("procurement_accounts_1688")
    .select("id")
    .eq("user_id", user.userId)
    .in("id", normalizedAccountIds);

  if (error) {
    throw new Error(`[procurement_accounts_1688] 查询可访问账户失败: ${error.message}`);
  }

  return new Set(
    ((data ?? []) as Array<{ id?: string | null }>)
      .map((row) => String(row.id ?? "").trim())
      .filter(Boolean),
  );
}

async function assertBatchAccess(user: ApiUserContext, batchId: string) {
  const normalizedBatchId = batchId.trim();
  if (!normalizedBatchId) {
    throw new ApiAuthorizationError(400, "缺少 batch_id 参数");
  }

  if (user.isAdmin) {
    return;
  }

  const createdBy = await getBatchCreatedBy(normalizedBatchId);
  if (!createdBy) {
    throw new ApiAuthorizationError(404, `未找到批次 ${normalizedBatchId}`);
  }

  if (createdBy !== user.userId) {
    throw new ApiAuthorizationError(403, `当前用户无权访问批次 ${normalizedBatchId}`);
  }
}

async function getAuthorizedProcurementAccount(user: ApiUserContext, accountId: string): Promise<ProcurementAccount1688Row> {
  const account = await getProcurementAccountById(accountId);
  if (!account) {
    throw new ApiAuthorizationError(404, `未找到 id=${accountId} 的 1688 账户`);
  }

  if (!user.isAdmin && account.user_id !== user.userId) {
    throw new ApiAuthorizationError(403, `当前用户无权使用 1688 账户 ${accountId}`);
  }

  return account;
}

// ── 自动录入采购字典 helper ────────────────────────────────────
/**
 * 当 LLM 结果中携带 identified_unique_model 或参数解读时，fire-and-forget 写入采购字典。
 * 并同步更新当前任务的 runtimeOverrides，使同次任务后续行立即生效。
 */
async function autoSaveDictEntry(
  result: OfflineCombinedStandardizationResult,
  currentRuntimeKeys: Set<string>,
  context: {
    batchId: string;
    source: StandardizationSourceSnapshot;
    llmModel: string;
  },
): Promise<void> {
  const model = result.identified_unique_model?.trim();
  const entries = result.parameter_interpretations
    .map((entry) => ({
      ...entry,
      suggested_keyword: "",
      source_batch_id: context.batchId,
      source_material_id: context.source.id,
      source_llm_model: context.llmModel,
      created_by: context.source.created_by,
    }))
    .filter((entry) => {
      const key = `${entry.term.trim().toUpperCase()}::${entry.term_type}::${entry.product_context.trim().toLowerCase()}`;
      if (currentRuntimeKeys.has(key)) {
        return false;
      }
      currentRuntimeKeys.add(key);
      return true;
    });

  try {
    if (entries.length > 0) {
      await upsertProcurementDictionaryEntries(entries);
      console.log(`[dict-auto] 自动录入采购字典参数解读: ${entries.map((entry) => entry.term).join(", ")}`);
    }

    if (model) {
      const upperModel = model.toUpperCase();
      const modelKey = `${upperModel}::unique_model::`;
      if (!currentRuntimeKeys.has(modelKey)) {
        currentRuntimeKeys.add(modelKey);
        await upsertModelNumberOverride(model, result.suggested_keyword, result.reasoning.slice(0, 200));
        mergeRuntimeModelOverrides({
          [upperModel]: { suggested_keyword: result.suggested_keyword, note: "" },
        });
        console.log(`[dict-auto] 自动录入采购字典唯一型号: ${model} → ${result.suggested_keyword}`);
      }
    }
  } catch (e) {
    console.warn("[dict-auto] 自动录入采购字典失败（不影响标准化）:", e);
  }
}

// ── 中间件 ────────────────────────────────────────────────────
app.use(cors());                          // 允许跨域（前端在不同端口）
app.use(express.json());                  // 解析 JSON body
const upload = multer({ storage: multer.memoryStorage() }); // 内存存储上传文件

// ── 服务状态 ──────────────────────────────────────────────────
type ServerStatus = "idle" | "processing" | "completed" | "failed";
let currentStatus: ServerStatus = "idle";
let lastResult: Record<string, unknown> | null = null;
let lastUploadedBuffer: Buffer | undefined;

// ── 标准化批次状态（独立于 CSV 上传状态）──────────────────────
type StandardizeProgressStatus = "idle" | "pending" | "processing" | "stopping" | "completed" | "failed" | "stopped";

interface StandardizeProgress {
  taskId: string;
  status: StandardizeProgressStatus;
  batchId: string;
  executionMode: StandardizationExecutionMode;
  providerStatus: string;
  totalRows: number;
  processedRows: number;
  progressCurrent: number;
  progressTotal: number;
  successRows: number;
  failedRows: number;
  currentItem: string;
  currentPhase: string;
  errors: string[];
  result: {
    success?: boolean;
    stopped?: boolean;
    taskId?: string;
    batchId?: string;
    totalRows?: number;
    processedRows?: number;
    successRows?: number;
    failedRows?: number;
    keywordInputRowsGenerated?: number;
    searchPlanRowsGenerated?: number;
    keywordProcessed?: number;
    error?: string;
  } | null;
  startedAt: string | null;
  finishedAt: string | null;
}

interface StandardizeRunState {
  task: StandardizationTaskRow;
  progress: StandardizeProgress;
  abortController?: AbortController | null;
  stopMessage?: string | null;
}

const ACTIVE_STANDARDIZE_STATUSES = new Set<StandardizeProgressStatus>(["processing", "stopping"]);
const standardizeRuns = new Map<string, StandardizeRunState>();

const _STANDARDIZATION_EDITABLE_FIELDS: Array<keyof StandardizationSourceSnapshot> = [
  "material_name",
  "specification",
  "quantity",
  "unit",
  "technical_standard",
  "material_requirement",
  "technical_requirement",
  "applicable_conditions",
  "test_requirement",
  "acceptance_note",
  "warranty_period",
  "deadline",
  "invoice_type",
  "brand",
  "remarks",
  "delivery_address",
  "source_file",
  "created_by",
];

function createStandardizeProgress(overrides: Partial<StandardizeProgress> = {}): StandardizeProgress {
  return {
    taskId: "",
    status: "idle",
    batchId: "",
    executionMode: "online",
    providerStatus: "",
    totalRows: 0,
    processedRows: 0,
    progressCurrent: 0,
    progressTotal: 0,
    successRows: 0,
    failedRows: 0,
    currentItem: "",
    currentPhase: "",
    errors: [...(overrides.errors ?? [])],
    result: overrides.result ? { ...overrides.result } : null,
    startedAt: null,
    finishedAt: null,
    ...overrides,
  };
}

function serializeStandardizeProgress(progress: StandardizeProgress): StandardizeProgress {
  return {
    ...progress,
    errors: [...progress.errors],
    result: progress.result ? { ...progress.result } : null,
  };
}

function isActiveStandardizeRun(runState: StandardizeRunState | null | undefined) {
  return Boolean(runState && ACTIVE_STANDARDIZE_STATUSES.has(runState.progress.status));
}

function getStandardizeProgressSortTime(progress: StandardizeProgress) {
  const timestamps = [progress.finishedAt, progress.startedAt]
    .filter((value): value is string => Boolean(value))
    .map((value) => {
      const time = Date.parse(value);
      return Number.isNaN(time) ? 0 : time;
    });

  return timestamps.length > 0 ? Math.max(...timestamps) : 0;
}

function sleep(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function parsePositiveInteger(value: string | undefined, fallback: number): number {
  const parsed = Number.parseInt(value ?? "", 10);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}

function getOfflineBatchPollIntervalMs() {
  return parsePositiveInteger(process.env.BIGMODEL_BATCH_POLL_INTERVAL_MS, 30_000);
}

function getTaskItemPatchedFields(item: Pick<StandardizationTaskItemRow, "patched_fields">) {
  if (!item.patched_fields || typeof item.patched_fields !== "object" || Array.isArray(item.patched_fields)) {
    return {} as Record<string, unknown>;
  }

  return { ...(item.patched_fields as Record<string, unknown>) };
}

function mergeTaskItemPatchedFields(
  item: Pick<StandardizationTaskItemRow, "patched_fields">,
  updates: Record<string, unknown>,
) {
  return {
    ...getTaskItemPatchedFields(item),
    ...updates,
  };
}

const OFFLINE_PHASE_PREFIX = "离线批处理";

function buildOfflinePhase(providerStatus: string) {
  return `${OFFLINE_PHASE_PREFIX} · ${providerStatus}`;
}

function extractOfflineProviderStatus(currentPhase: string) {
  const match = /^离线批处理\s*·\s*(.+)$/u.exec(currentPhase.trim());
  return match?.[1]?.trim() ?? "";
}

function extractOfflineBatchId(value: string) {
  const match = /(batch_[A-Za-z0-9_-]+)/.exec(value);
  return match?.[1] ?? "";
}

function inferTaskExecutionMode(
  taskRow: Pick<StandardizationTaskRow, "execution_mode" | "current_phase" | "current_item">,
  taskItem?: StandardizationTaskItemRow,
): StandardizationExecutionMode {
  if (taskRow.execution_mode === "online" || taskRow.execution_mode === "offline") {
    return taskRow.execution_mode;
  }

  const patchedFields = taskItem ? getTaskItemPatchedFields(taskItem) : {};
  if (patchedFields.execution_mode === "offline") {
    return "offline";
  }

  if (taskRow.current_phase.startsWith(OFFLINE_PHASE_PREFIX) || extractOfflineBatchId(taskRow.current_item)) {
    return "offline";
  }

  return "online";
}

function getProgressTotal(row: StandardizationTaskRow, executionMode: StandardizationExecutionMode) {
  return executionMode === "offline"
    ? Math.max(0, row.total_rows)
    : Math.max(0, row.total_rows * 2);
}

function getProgressCurrent(row: StandardizationTaskRow, executionMode: StandardizationExecutionMode) {
  const total = getProgressTotal(row, executionMode);
  return Math.min(total, Math.max(0, row.processed_rows));
}

function buildOfflineCurrentItem(params: {
  batchId: string;
  totalRows: number;
  completedRows: number;
  failedRows: number;
}) {
  return `Batch ${params.batchId}：已完成 ${params.completedRows}/${params.totalRows}，失败 ${params.failedRows} 条`;
}

function listStandardizeProgresses() {
  return Array.from(standardizeRuns.values())
    .map((runState) => serializeStandardizeProgress(runState.progress))
    .filter((progress) => progress.status !== "idle" || Boolean(progress.batchId))
    .sort((left, right) => getStandardizeProgressSortTime(right) - getStandardizeProgressSortTime(left));
}

function getLatestStandardizeProgress() {
  return listStandardizeProgresses()[0] ?? createStandardizeProgress();
}

function _findActiveStandardizeRunByBatch(batchId: string) {
  return Array.from(standardizeRuns.values()).find((runState) => (
    runState.progress.batchId === batchId && isActiveStandardizeRun(runState)
  ));
}

function pushStandardizeError(runState: StandardizeRunState, message: string) {
  const trimmed = message.trim();
  if (!trimmed) {
    return;
  }

  runState.progress.errors = [...new Set([...runState.progress.errors, trimmed])].slice(-20);
}

function looksLikeUuid(value: string) {
  return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value);
}

function looksLikeSearchTaskCode(value: string) {
  return /^[A-Za-z0-9][A-Za-z0-9-]*-ss-\d{3,}$/i.test(value.trim());
}

function _buildStandardizationSourceKey(source: Pick<StandardizationSourceSnapshot, "id" | "batch_id">) {
  return `${source.batch_id}:${source.id}`;
}

function normalizeStandardizationSourceSnapshot(
  source: Record<string, unknown>,
  index: number,
  batchIdFallback: string,
): StandardizationSourceSnapshot {
  const quantity = Number(source.quantity ?? 0);
  const materialId = String(source.id ?? source.material_id ?? `selected-${index + 1}`);

  return {
    id: materialId,
    batch_id: String(source.batch_id ?? batchIdFallback),
    material_name: String(source.material_name ?? source.name ?? ""),
    specification: String(source.specification ?? source.spec ?? ""),
    quantity: Number.isFinite(quantity) ? quantity : 0,
    unit: String(source.unit ?? ""),
    technical_standard: String(source.technical_standard ?? ""),
    material_requirement: String(source.material_requirement ?? ""),
    technical_requirement: String(source.technical_requirement ?? ""),
    applicable_conditions: String(source.applicable_conditions ?? ""),
    test_requirement: String(source.test_requirement ?? ""),
    acceptance_note: String(source.acceptance_note ?? ""),
    warranty_period: String(source.warranty_period ?? ""),
    deadline: String(source.deadline ?? ""),
    invoice_type: String(source.invoice_type ?? ""),
    brand: String(source.brand ?? ""),
    remarks: String(source.remarks ?? source.notes ?? ""),
    delivery_address: String(source.delivery_address ?? ""),
    source_file: String(source.source_file ?? ""),
    created_at: String(source.created_at ?? new Date().toISOString()),
    created_by: String(source.created_by ?? ""),
  };
}

function getStandardizeRunsByMode(executionMode: StandardizationExecutionMode) {
  return Array.from(standardizeRuns.values()).filter((runState) => runState.progress.executionMode === executionMode);
}

function getActiveStandardizeRunByMode(executionMode: StandardizationExecutionMode) {
  return getStandardizeRunsByMode(executionMode).find((runState) => isActiveStandardizeRun(runState));
}

function getOfflineTaskLlmModel(taskItems: StandardizationTaskItemRow[]) {
  return taskItems.find((item) => item.llm_model)?.llm_model
    || resolveBigModelBatchModel();
}

function getStandardizeTaskProviderBatchId(
  task: Pick<StandardizationTaskRow, "current_item">,
  taskItems: StandardizationTaskItemRow[],
) {
  return taskItems
    .map((item) => getTaskItemPatchedFields(item).provider_batch_id)
    .find((value): value is string => typeof value === "string" && value.trim().length > 0)
    ?? extractOfflineBatchId(task.current_item);
}

async function loadStandardizeTaskRunPayload(task: StandardizationTaskRow) {
  const taskId = task.id ?? "";
  const taskItems = taskId ? await getStandardizationTaskItemsByTaskId(taskId) : [];
  const sources = taskItems
    .map((item, index) => normalizeStandardizationSourceSnapshot(item.source_snapshot as Record<string, unknown>, index, task.batch_id))
    .sort((left, right) => left.created_at.localeCompare(right.created_at) || left.id.localeCompare(right.id));

  return {
    taskItems,
    sources,
    taskItemsByMaterialId: new Map(taskItems.map((item) => [item.source_material_id, item])),
    providerBatchId: getStandardizeTaskProviderBatchId(task, taskItems),
    llmModel: getOfflineTaskLlmModel(taskItems),
  };
}

function buildRawCsvRowFromSource(source: StandardizationSourceSnapshot): RawCsvRow {
  return {
    发票种类: source.invoice_type,
    名称: source.material_name,
    规格型号: source.specification,
    品牌: source.brand,
    数量: String(source.quantity ?? ""),
    单位: source.unit,
    "备注/附件": source.remarks,
    收货地址: source.delivery_address,
    技术标准: source.technical_standard,
    材质要求: source.material_requirement,
    技术要求: source.technical_requirement,
    适用条件: source.applicable_conditions,
    检验要求: source.test_requirement,
    验收说明: source.acceptance_note,
    质保期: source.warranty_period,
    交货期限: source.deadline,
    来源文件: source.source_file,
    批次ID: source.batch_id,
  };
}

function normalizeSearchTerm(value: string | null | undefined) {
  return String(value ?? "").replace(/\s+/g, " ").trim();
}

function dedupeAlternativeTerms(primarySearchTerm: string, values: string[]) {
  const compactPrimary = normalizeSearchTerm(primarySearchTerm).replace(/\s+/g, "").toLowerCase();
  const seen = new Set<string>();
  const results: string[] = [];

  for (const value of values) {
    const normalized = normalizeSearchTerm(value);
    if (!normalized) {
      continue;
    }

    const compactValue = normalized.replace(/\s+/g, "").toLowerCase();
    if (!compactValue || compactValue === compactPrimary || seen.has(compactValue)) {
      continue;
    }

    seen.add(compactValue);
    results.push(normalized);
  }

  return results;
}

function buildStandardizedResultRow(params: {
  batchId: string;
  source: StandardizationSourceSnapshot;
  raw: RawCsvRow;
  result: OfflineCombinedStandardizationResult;
  llmModel: string;
}): Omit<StandardizedResultRow, "id" | "created_at" | "updated_at"> {
  const primarySearchTerm = normalizeSearchTerm(params.result.suggested_keyword)
    || normalizeSearchTerm(params.source.material_name)
    || normalizeSearchTerm(params.raw.名称);
  const createdBy = looksLikeUuid(params.source.created_by) ? params.source.created_by : null;
  const reasoningSummary = dedupeAlternativeTerms("", [
    normalizeSearchTerm(params.result.reasoning),
    formatParameterInterpretationsSummary(params.result.parameter_interpretations),
    normalizeSearchTerm(params.result.structure_summary),
    normalizeSearchTerm(params.result.required_attributes_summary),
  ]).join("\n");

  return {
    batch_id: params.batchId,
    material_id: params.source.id,
    source_type: "ai",
    invoice_type: params.source.invoice_type || normalizeSearchTerm(params.raw.发票种类),
    original_name: normalizeSearchTerm(params.raw.名称),
    original_spec: normalizeSearchTerm(params.raw.规格型号),
    standardized_name: normalizeSearchTerm(params.source.material_name) || normalizeSearchTerm(params.raw.名称),
    category: normalizeSearchTerm(params.result.material) || normalizeSearchTerm(params.result.usage),
    brand: normalizeSearchTerm(params.result.brand_term) || params.source.brand || normalizeSearchTerm(params.raw.品牌),
    notes: params.source.remarks || normalizeSearchTerm(params.raw["备注/附件"]),
    delivery_address: params.source.delivery_address || normalizeSearchTerm(params.raw.收货地址),
    primary_search_term: primarySearchTerm,
    alt_search_terms: dedupeAlternativeTerms(primarySearchTerm, params.result.alternative_terms),
    standardized_spec: normalizeSearchTerm(params.result.simplified_spec) || normalizeSearchTerm(params.raw.规格型号),
    standardized_unit: params.source.unit || normalizeSearchTerm(params.raw.单位),
    quantity: Number.isFinite(params.source.quantity) ? params.source.quantity : Number(params.raw.数量 ?? 0) || 0,
    product_type: normalizeSearchTerm(params.result.product_type),
    llm_model: params.llmModel,
    ai_reasoning: reasoningSummary,
    created_by: createdBy,
    updated_by: createdBy,
  };
}

function buildKeywordProcessRow(params: {
  batchId: string;
  source: StandardizationSourceSnapshot;
  product: ParsedProduct;
  llmModel: string;
  keywordAnalysisPayload: Record<string, unknown>;
  standardizedResult: Omit<StandardizedResultRow, "id" | "created_at" | "updated_at">;
}) {
  return {
    batch_id: params.batchId,
    material_id: params.source.id,
    row_name: params.standardizedResult.standardized_name || params.standardizedResult.original_name,
    row_spec: params.standardizedResult.standardized_spec || params.standardizedResult.original_spec,
    row_brand: params.standardizedResult.brand,
    llm_model: params.llmModel,
    ai_keyword_analysis: params.keywordAnalysisPayload,
    full_report: {
      batch_id: params.batchId,
      material_id: params.source.id,
      source_snapshot: params.source,
      raw_row: params.product.raw,
      standardized_result: params.standardizedResult,
      ai_keyword_analysis: params.keywordAnalysisPayload,
      llm_analysis: params.product.analysis,
    },
  };
}

async function deleteRelatedSearchArtifactsByMaterialIds(batchId: string, materialIds: string[]) {
  if (materialIds.length === 0) {
    return;
  }

  const { data: finalRows, error: finalRowsError } = await supabase
    .from("final_result_products")
    .select("id")
    .eq("batch_id", batchId)
    .in("material_id", materialIds);

  if (finalRowsError) {
    throw new Error(`[final_result_products] 查询失败: ${finalRowsError.message}`);
  }

  const searchResultIds = (finalRows ?? []).map((row) => String((row as { id: string }).id));
  if (searchResultIds.length > 0) {
    const [contactLogsRes, customerChatsRes] = await Promise.all([
      supabase.from("contact_logs").delete().in("search_result_id", searchResultIds),
      supabase.from("customer_service_chats").delete().in("search_result_id", searchResultIds),
    ]);

    if (contactLogsRes.error) {
      throw new Error(`[contact_logs] 删除失败: ${contactLogsRes.error.message}`);
    }

    if (customerChatsRes.error) {
      throw new Error(`[customer_service_chats] 删除失败: ${customerChatsRes.error.message}`);
    }
  }

  const deletions = await Promise.all([
    supabase.from("search_rounds").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("best_match_images").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("extended_best_matches").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("image_search_results").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("keyword_selections").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("final_results").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("final_result_products").delete().eq("batch_id", batchId).in("material_id", materialIds),
    supabase.from("extended_best_match_records").delete().eq("batch_id", batchId).in("material_id", materialIds),
  ]);

  for (const result of deletions) {
    if (result.error) {
      throw new Error(result.error.message);
    }
  }
}

function mapTaskRowToStandardizeProgress(
  row: StandardizationTaskRow,
  executionMode: StandardizationExecutionMode = "online",
): StandardizeProgress {
  const progressCurrent = getProgressCurrent(row, executionMode);
  const progressTotal = getProgressTotal(row, executionMode);

  return createStandardizeProgress({
    taskId: row.id ?? "",
    status: row.status,
    batchId: row.batch_id,
    executionMode,
    providerStatus: executionMode === "offline" ? extractOfflineProviderStatus(row.current_phase) : "",
    totalRows: row.total_rows,
    processedRows: row.processed_rows,
    progressCurrent,
    progressTotal,
    successRows: row.success_rows,
    failedRows: row.failed_rows,
    currentItem: row.current_item,
    currentPhase: row.current_phase,
    result: row.status === "failed"
      ? {
          taskId: row.id,
          batchId: row.batch_id,
          totalRows: row.total_rows,
          processedRows: row.processed_rows,
          successRows: row.success_rows,
          failedRows: row.failed_rows,
          error: row.error_message,
        }
      : row.status === "stopped"
        ? {
            stopped: true,
            taskId: row.id,
            batchId: row.batch_id,
            totalRows: row.total_rows,
            processedRows: row.processed_rows,
            successRows: row.success_rows,
            failedRows: row.failed_rows,
            error: row.error_message,
          }
      : row.status === "completed"
        ? {
            success: row.failed_rows === 0,
            taskId: row.id,
            batchId: row.batch_id,
            totalRows: row.total_rows,
            processedRows: row.processed_rows,
            successRows: row.success_rows,
            failedRows: row.failed_rows,
          }
        : null,
    startedAt: row.started_at ?? null,
    finishedAt: row.finished_at ?? null,
  });
}

async function syncStandardizeTaskRow(runState: StandardizeRunState) {
  try {
    runState.task = await updateStandardizationTask(runState.progress.taskId, {
      status: runState.progress.status === "idle" ? "completed" : runState.progress.status,
      execution_mode: runState.progress.executionMode,
      total_rows: runState.progress.totalRows,
      processed_rows: runState.progress.processedRows,
      success_rows: runState.progress.successRows,
      failed_rows: runState.progress.failedRows,
      current_item: runState.progress.currentItem,
      current_phase: runState.progress.currentPhase,
      error_message: runState.progress.errors[0] ?? "",
      started_at: runState.progress.startedAt,
      finished_at: runState.progress.finishedAt,
    });
  } catch (err) {
    pushStandardizeError(runState, err instanceof Error ? err.message : String(err));
  }
}

async function patchStandardizeProgress(
  runState: StandardizeRunState,
  updates: Partial<StandardizeProgress>,
) {
  runState.progress = {
    ...runState.progress,
    ...updates,
    errors: updates.errors ? [...updates.errors] : runState.progress.errors,
    result: updates.result === undefined
      ? runState.progress.result
      : updates.result
        ? { ...updates.result }
        : null,
  };

  await syncStandardizeTaskRow(runState);
}

async function finalizeStandardizeRun(
  runState: StandardizeRunState,
  status: Extract<StandardizeProgressStatus, "completed" | "failed" | "stopped">,
  result: StandardizeProgress["result"],
) {
  await patchStandardizeProgress(runState, {
    status,
    currentPhase: runState.progress.executionMode === "offline"
      ? buildOfflinePhase(status === "completed" ? "completed" : status === "stopped" ? "cancelled" : "failed")
      : status === "completed"
        ? "已完成"
        : status === "stopped"
          ? "已停止"
        : "任务失败",
    providerStatus: runState.progress.executionMode === "offline"
      ? status === "completed"
        ? "completed"
        : status === "stopped"
          ? "cancelled"
        : "failed"
      : runState.progress.providerStatus,
    finishedAt: new Date().toISOString(),
    result,
  });
}

async function getPersistedLatestStandardizeProgress() {
  const { data, error } = await supabase
    .from("standardization_tasks")
    .select("*")
    .order("updated_at", { ascending: false })
    .limit(1)
    .maybeSingle();

  if (error) {
    throw new Error(`[standardization_tasks] 查询最新任务失败: ${error.message}`);
  }

  if (!data) {
    return createStandardizeProgress();
  }

  const taskRow = data as StandardizationTaskRow;
  const taskId = taskRow.id ?? "";
  const taskItems = taskId ? await getStandardizationTaskItemsByTaskId(taskId).catch(() => []) : [];
  const executionMode = inferTaskExecutionMode(taskRow, taskItems[0]);

  return mapTaskRowToStandardizeProgress(taskRow, executionMode);
}

function getOnlineStandardizeBatchSize() {
  const value = Number.parseInt(process.env.STANDARDIZE_ONLINE_BATCH_SIZE ?? "", 10);
  if (!Number.isInteger(value) || value <= 0) {
    return 10;
  }
  return Math.min(value, 50);
}

function chunkArray<T>(items: T[], chunkSize: number): T[][] {
  const chunks: T[][] = [];
  for (let index = 0; index < items.length; index += chunkSize) {
    chunks.push(items.slice(index, index + chunkSize));
  }
  return chunks;
}

function getStandardizeStopMessage(runState: StandardizeRunState) {
  return runState.stopMessage?.trim() || "用户手动停止在线标准化任务";
}

function throwIfStandardizeRunShouldStop(runState: StandardizeRunState) {
  if (runState.progress.executionMode !== "online") {
    return;
  }

  if (runState.progress.status === "stopping" || runState.abortController?.signal.aborted) {
    const error = new Error(getStandardizeStopMessage(runState));
    error.name = "AbortError";
    throw error;
  }
}

async function requestOnlineStandardizeStop(runState: StandardizeRunState, message: string) {
  runState.stopMessage = message;

  if (runState.progress.status !== "stopping") {
    await patchStandardizeProgress(runState, {
      status: "stopping",
      currentPhase: "停止中",
      currentItem: "正在停止当前在线标准化任务...",
      result: null,
    });
  }

  if (runState.abortController && !runState.abortController.signal.aborted) {
    runState.abortController.abort(message);
  }
}

async function finalizeStoppedOnlineStandardizeRun(
  runState: StandardizeRunState,
  totalRows: number,
) {
  const taskId = runState.progress.taskId;
  const batchId = runState.progress.batchId;
  const message = getStandardizeStopMessage(runState);

  await failPendingStandardizationTaskItems(taskId, message).catch(() => {
    // best effort
  });

  const taskItems = await getStandardizationTaskItemsByTaskId(taskId).catch(() => []);
  const successRows = taskItems.filter((item) => item.status === "success").length;
  const failedRows = taskItems.filter((item) => item.status === "failed").length;

  await finalizeStandardizeRun(runState, "stopped", {
    success: false,
    stopped: true,
    taskId,
    batchId,
    totalRows,
    processedRows: runState.progress.processedRows,
    successRows,
    failedRows,
    error: message,
  });
}

async function runOnlineStandardizationTask(params: {
  runState: StandardizeRunState;
  sources: StandardizationSourceSnapshot[];
  taskItemsByMaterialId: Map<string, StandardizationTaskItemRow>;
  llmModel: string;
}) {
  const { runState, sources, taskItemsByMaterialId, llmModel } = params;
  const batchId = runState.progress.batchId;
  const taskId = runState.progress.taskId;
  runState.abortController ??= new AbortController();
  let stageOneProcessed = 0;
  let failedRows = 0;
  let successRows = 0;
  let keywordProcessed = 0;
  const parsedProducts: ParsedProduct[] = [];
  const combinedResultsByMaterialId = new Map<string, OfflineCombinedStandardizationResult>();
  const standardizedRowsByMaterialId = new Map<string, Omit<StandardizedResultRow, "id" | "created_at" | "updated_at">>();
  const sourcesByMaterialId = new Map(sources.map((source) => [source.id, source]));
  const genericFixHint = buildCategorizedHumanFixHint([]);
  /** 本轮任务内已自动录入字典的型号 key 集合（大写），用于去重 */
  const autoSavedModelKeys = new Set<string>();

  // 从 DB 加载最新型号字典，失败时静默降级到静态字典
  try {
    setRuntimeModelOverrides(await loadModelNumberOverridesFromDb());
  } catch (e) {
    console.warn("[standardize-online] 型号字典 DB 加载失败，使用静态字典:", e);
  }

  try {
    await patchStandardizeProgress(runState, {
      status: "processing",
      executionMode: "online",
      providerStatus: "",
      batchId,
      totalRows: sources.length,
      processedRows: 0,
      progressCurrent: 0,
      progressTotal: sources.length,
      successRows: 0,
      failedRows: 0,
      currentItem: "正在准备标准化任务...",
      currentPhase: "标准化分析",
      errors: [],
      result: null,
      startedAt: runState.progress.startedAt,
      finishedAt: null,
    });

    const sourceChunks = chunkArray(sources, getOnlineStandardizeBatchSize());

    for (let chunkIndex = 0; chunkIndex < sourceChunks.length; chunkIndex += 1) {
      throwIfStandardizeRunShouldStop(runState);

      const chunk = sourceChunks[chunkIndex]!;
      const chunkStart = stageOneProcessed + 1;
      const chunkEnd = Math.min(sources.length, stageOneProcessed + chunk.length);
      const chunkRows = chunk.map((source) => ({
        materialId: source.id,
        row: buildRawCsvRowFromSource(source),
      }));

      await patchStandardizeProgress(runState, {
        currentItem: `正在批量分析: 第 ${chunkStart}-${chunkEnd} 行，共 ${chunk.length} 条`,
        currentPhase: "标准化分析",
        processedRows: stageOneProcessed,
        progressCurrent: stageOneProcessed,
        progressTotal: sources.length,
        failedRows,
      });

      for (const source of chunk) {
        const materialId = source.id;
        const taskItem = taskItemsByMaterialId.get(materialId);
        if (!taskItem?.id) {
          continue;
        }

        taskItemsByMaterialId.set(
          materialId,
          await updateStandardizationTaskItem(taskItem.id, {
            status: "processing",
            failure_reason: "",
            human_fix_hint: "",
            missing_required_fields: [],
            patched_fields: mergeTaskItemPatchedFields(taskItem, {
              execution_mode: "online",
            }),
            llm_model: llmModel,
          }),
        );
      }

      try {
        console.log(`[standardize] 批量分析 ${chunkIndex + 1}/${sourceChunks.length}: ${chunk.length} 条`);
        const combinedResults = await analyzeProductsCombined(chunkRows, { signal: runState.abortController?.signal });
        throwIfStandardizeRunShouldStop(runState);

        for (let offset = 0; offset < chunk.length; offset += 1) {
          throwIfStandardizeRunShouldStop(runState);

          const source = chunk[offset]!;
          const raw = chunkRows[offset]!.row;
          const materialId = source.id;
          const combinedResult = combinedResults.get(materialId);
          const taskItem = taskItemsByMaterialId.get(materialId);
          if (!combinedResult) {
            throw new Error(`批量标准化结果缺少物料 ${materialId}`);
          }

          const analysis = toLlmAnalysisResult(combinedResult);
          const parsedProduct: ParsedProduct = { material_id: materialId, raw, analysis };
          const standardizedPreview = buildStandardizedResultRow({
            batchId,
            source,
            raw,
            result: combinedResult,
            llmModel,
          });
          combinedResultsByMaterialId.set(materialId, combinedResult);
          standardizedRowsByMaterialId.set(materialId, standardizedPreview);
          parsedProducts.push(parsedProduct);
          // 自动录入采购字典（fire-and-forget）
          if (combinedResult.outcome === "success") {
            void autoSaveDictEntry(combinedResult, autoSavedModelKeys, { batchId, source, llmModel });
          }

          if (taskItem?.id) {
            taskItemsByMaterialId.set(
              materialId,
              await updateStandardizationTaskItem(taskItem.id, {
                status: "processing",
                analysis_result: analysis as unknown as Record<string, unknown>,
                standardized_name: standardizedPreview.standardized_name,
                patched_fields: mergeTaskItemPatchedFields(taskItem, {
                  execution_mode: "online",
                }),
                llm_model: llmModel,
              }),
            );
          }

          stageOneProcessed += 1;
          await patchStandardizeProgress(runState, {
            currentItem: `已分析: ${source.material_name || raw.名称 || `第 ${chunkStart + offset} 行`}`,
            currentPhase: "标准化分析",
            processedRows: stageOneProcessed,
            progressCurrent: stageOneProcessed,
            progressTotal: sources.length,
            failedRows,
          });
        }
      } catch (err) {
        if (isAbortError(err)) {
          throw err;
        }

        const message = err instanceof Error ? err.message : String(err);
        for (let offset = 0; offset < chunk.length; offset += 1) {
          const source = chunk[offset]!;
          const raw = chunkRows[offset]!.row;
          const materialId = source.id;
          const taskItem = taskItemsByMaterialId.get(materialId);
          failedRows += 1;
          stageOneProcessed += 1;
          pushStandardizeError(runState, `物料 ${materialId} 标准化失败: ${message}`);

          if (taskItem?.id) {
            const failureMessage = formatRetryFailureMessage(taskItem, message);
            taskItemsByMaterialId.set(
              materialId,
              await updateStandardizationTaskItem(taskItem.id, {
                status: "failed",
                analysis_result: {},
                keyword_analysis: {},
                standardized_name: source.material_name || raw.名称,
                suggested_keyword: "",
                can_search_1688: null,
                missing_required_fields: [],
                failure_reason: failureMessage,
                human_fix_hint: genericFixHint,
                patched_fields: mergeTaskItemPatchedFields(taskItem, {
                  execution_mode: "online",
                }),
                llm_model: llmModel,
              }),
            );
          }

          await patchStandardizeProgress(runState, {
            currentItem: `批量分析失败: ${source.material_name || raw.名称 || `第 ${chunkStart + offset} 行`}`,
            currentPhase: "标准化分析",
            processedRows: stageOneProcessed,
            progressCurrent: stageOneProcessed,
            progressTotal: sources.length,
            failedRows,
          });
        }
      }
    }

    throwIfStandardizeRunShouldStop(runState);

    const selectedMaterialIds = sources.map((source) => source.id);

    await patchStandardizeProgress(runState, {
      currentItem: "正在清理旧标准化结果...",
      currentPhase: "清理旧结果",
      processedRows: stageOneProcessed,
      progressCurrent: stageOneProcessed,
      progressTotal: sources.length,
      failedRows,
    });

    throwIfStandardizeRunShouldStop(runState);

    await Promise.all([
      deleteStandardizedResultsByMaterialIds(batchId, selectedMaterialIds),
      deleteKeywordProcessesByMaterialIds(selectedMaterialIds, batchId),
      deleteRelatedSearchArtifactsByMaterialIds(batchId, selectedMaterialIds),
    ]);

    throwIfStandardizeRunShouldStop(runState);

    parsedProducts.sort((left, right) => left.material_id.localeCompare(right.material_id));

    if (parsedProducts.length === 0) {
      const message = "全部物料标准化失败，请补充关键字段后重试";
      await failPendingStandardizationTaskItems(taskId, message);
      await finalizeStandardizeRun(runState, "failed", {
        success: false,
        taskId,
        batchId,
        totalRows: sources.length,
        processedRows: stageOneProcessed,
        successRows: 0,
        failedRows,
        error: message,
      });
      return;
    }

    await patchStandardizeProgress(runState, {
      currentItem: "正在写入标准化结果...",
      currentPhase: "写入标准化结果",
      processedRows: stageOneProcessed,
      progressCurrent: stageOneProcessed,
      progressTotal: sources.length,
      failedRows,
    });

    throwIfStandardizeRunShouldStop(runState);

    for (let i = 0; i < parsedProducts.length; i += 1) {
      throwIfStandardizeRunShouldStop(runState);

      const product = parsedProducts[i]!;
      const combinedResult = combinedResultsByMaterialId.get(product.material_id);
      const taskItem = taskItemsByMaterialId.get(product.material_id);
      const source = sourcesByMaterialId.get(product.material_id);
      const standardizedRow = standardizedRowsByMaterialId.get(product.material_id);
      if (!combinedResult || !source || !standardizedRow) {
        throw new Error(`material_id=${product.material_id} 缺少统一标准化结果`);
      }

      await patchStandardizeProgress(runState, {
        currentItem: `回写统一结果: ${standardizedRow.standardized_name} (${i + 1}/${parsedProducts.length})`,
        currentPhase: "回写统一结果",
        processedRows: stageOneProcessed,
        progressCurrent: stageOneProcessed,
        progressTotal: sources.length,
        successRows,
        failedRows,
      });

      console.log(`[standardize] 回写统一结果 ${i + 1}/${parsedProducts.length}: ${standardizedRow.standardized_name}`);
      try {
        const kwAnalysis = toAiKeywordAnalysis(combinedResult);
        const keywordAnalysisPayload = {
          ...kwAnalysis,
          alt_search_terms: [...product.analysis.alternative_terms],
          alternative_keywords: [...product.analysis.alternative_terms],
        };
        throwIfStandardizeRunShouldStop(runState);

        await upsertKeywordProcess(buildKeywordProcessRow({
          batchId,
          source,
          product,
          llmModel,
          keywordAnalysisPayload: keywordAnalysisPayload as unknown as Record<string, unknown>,
          standardizedResult: standardizedRow,
        }));
        keywordProcessed += 1;

        const itemFailed = kwAnalysis.outcome === "failed" || kwAnalysis.can_search_1688 === false;
        if (!itemFailed) {
          await upsertStandardizedResults([standardizedRow]);
        }

        if (taskItem?.id) {
          const failureReason = formatRetryFailureMessage(taskItem, kwAnalysis.failure_reason);
          taskItemsByMaterialId.set(
            product.material_id,
            await updateStandardizationTaskItem(taskItem.id, {
              status: itemFailed ? "failed" : "success",
              keyword_analysis: keywordAnalysisPayload as unknown as Record<string, unknown>,
              standardized_name: standardizedRow.standardized_name,
              suggested_keyword: kwAnalysis.suggested_keyword,
              can_search_1688: kwAnalysis.can_search_1688,
              missing_required_fields: kwAnalysis.missing_required_fields,
              failure_reason: itemFailed ? failureReason : kwAnalysis.failure_reason,
              human_fix_hint: kwAnalysis.human_fix_hint,
              patched_fields: mergeTaskItemPatchedFields(taskItem, {
                execution_mode: "online",
              }),
              llm_model: llmModel,
            }),
          );
        }

        if (itemFailed) {
          failedRows += 1;
          pushStandardizeError(runState, `物料 ${product.material_id} ${formatRetryFailureMessage(taskItem, kwAnalysis.failure_reason || "缺少关键采购信息")}`);
        } else {
          successRows += 1;
          await resolveOutstandingStandardizationFailures({
            batchId,
            sourceMaterialId: taskItem?.source_material_id || undefined,
            resolvedByTaskId: taskId,
          });
        }
      } catch (err) {
        if (isAbortError(err)) {
          throw err;
        }

        const message = err instanceof Error ? err.message : String(err);
        failedRows += 1;
        pushStandardizeError(runState, `物料 ${product.material_id} 结果回写失败: ${message}`);

        if (taskItem?.id) {
          const failureMessage = formatRetryFailureMessage(taskItem, message);
          taskItemsByMaterialId.set(
            product.material_id,
            await updateStandardizationTaskItem(taskItem.id, {
              status: "failed",
              keyword_analysis: {},
              suggested_keyword: "",
              can_search_1688: null,
              missing_required_fields: [],
              failure_reason: failureMessage,
              human_fix_hint: genericFixHint,
              patched_fields: mergeTaskItemPatchedFields(taskItem, {
                execution_mode: "online",
              }),
              llm_model: llmModel,
            }),
          );
        }
      }
    }

    const completionResult = {
      success: failedRows === 0,
      taskId,
      batchId,
      totalRows: sources.length,
      processedRows: sources.length,
      successRows,
      failedRows,
      keywordInputRowsGenerated: 0,
      searchPlanRowsGenerated: 0,
      keywordProcessed,
    };

    await patchStandardizeProgress(runState, {
      currentItem: failedRows === 0
        ? "标准化完成"
        : `标准化完成，${successRows} 条成功，${failedRows} 条需处理`,
      currentPhase: failedRows === 0 ? "标准化完成" : "标准化完成（部分失败）",
      processedRows: sources.length,
      progressCurrent: sources.length,
      progressTotal: sources.length,
      successRows,
      failedRows,
    });
    await finalizeStandardizeRun(runState, "completed", completionResult);

    console.log(`[standardize] 批次 ${batchId} 标准化完成: standardized=${successRows}, keyword=${keywordProcessed}`);
  } catch (err) {
    if (isAbortError(err) && runState.abortController?.signal.aborted) {
      await finalizeStoppedOnlineStandardizeRun(runState, sources.length);
      console.warn(`[standardize] 批次 ${batchId} 已停止: ${getStandardizeStopMessage(runState)}`);
      return;
    }

    const msg = err instanceof Error ? err.message : String(err);

    pushStandardizeError(runState, msg);
    await failPendingStandardizationTaskItems(taskId, msg).catch(() => {
      // 任务头会记录最终错误，这里仅做 best-effort 的行级失败收敛。
    });
    await finalizeStandardizeRun(runState, "failed", {
      success: false,
      taskId,
      batchId,
      totalRows: sources.length,
      processedRows: runState.progress.processedRows,
      successRows: runState.progress.successRows,
      failedRows: Math.max(runState.progress.failedRows, failedRows),
      error: msg,
    });
    console.error(`[standardize] 批次 ${batchId} 标准化失败: ${msg}`);
  }
}

async function applyOfflineBatchResults(params: {
  runState: StandardizeRunState;
  sources: StandardizationSourceSnapshot[];
  taskItemsByMaterialId: Map<string, StandardizationTaskItemRow>;
  batchRecord: BigModelBatchRecord;
  llmModel: string;
}) {
  const { runState, sources, taskItemsByMaterialId, batchRecord, llmModel } = params;
  const batchId = runState.progress.batchId;
  const taskId = runState.progress.taskId;
  const genericFixHint = buildCategorizedHumanFixHint([]);

  await patchStandardizeProgress(runState, {
    currentPhase: buildOfflinePhase("写回结果"),
    providerStatus: "writing-results",
    currentItem: `正在回写 Batch ${batchRecord.id} 的结果...`,
    progressCurrent: runState.progress.progressCurrent,
    progressTotal: sources.length,
  });

  const rawRowsByMaterialId = new Map(sources.map((source) => [source.id, buildRawCsvRowFromSource(source)]));
  const parsedProductsByMaterialId = new Map<string, ParsedProduct>();
  const combinedResultsByMaterialId = new Map<string, OfflineCombinedStandardizationResult>();
  const providerHandledMaterialIds = new Set<string>();
  const failureReasonsByMaterialId = new Map<string, string>();
  const responseIdsByMaterialId = new Map<string, string>();
  /** 本轮任务内已自动录入字典的型号 key 集合（大写），用于去重 */
  const autoSavedModelKeys = new Set<string>();

  const handleBatchLine = (lineContent: string) => {
    const line = JSON.parse(lineContent) as Record<string, unknown>;
    const customId = typeof line.custom_id === "string" ? line.custom_id : "";
    const parsedCustomId = parseOfflineBatchCustomId(customId);
    if (!parsedCustomId) {
      return;
    }

    providerHandledMaterialIds.add(parsedCustomId.materialId);
    const responseId = typeof line.id === "string" ? line.id : "";
    if (responseId) {
      responseIdsByMaterialId.set(parsedCustomId.materialId, responseId);
    }

    const statusCode = extractBatchLineStatusCode(line as Parameters<typeof extractBatchLineStatusCode>[0]);
    if (statusCode !== 200) {
      try {
        extractBatchLineMessage(line as Parameters<typeof extractBatchLineMessage>[0]);
      } catch (error) {
        failureReasonsByMaterialId.set(parsedCustomId.materialId, error instanceof Error ? error.message : String(error));
      }
      return;
    }

    const raw = rawRowsByMaterialId.get(parsedCustomId.materialId);
    if (!raw) {
      failureReasonsByMaterialId.set(parsedCustomId.materialId, "离线结果无法匹配原始物料");
      return;
    }

    try {
      const message = extractBatchLineMessage(line as Parameters<typeof extractBatchLineMessage>[0]);
      const combined = parseOfflineCombinedResult(message, raw);
      combinedResultsByMaterialId.set(parsedCustomId.materialId, combined);
      parsedProductsByMaterialId.set(parsedCustomId.materialId, {
        material_id: parsedCustomId.materialId,
        raw,
        analysis: toLlmAnalysisResult(combined),
      });
      // 自动录入采购字典（fire-and-forget）
      if (combined.outcome === "success") {
        const source = sources.find((item) => item.id === parsedCustomId.materialId);
        if (source) {
          void autoSaveDictEntry(combined, autoSavedModelKeys, { batchId, source, llmModel });
        }
      }
    } catch (error) {
      failureReasonsByMaterialId.set(parsedCustomId.materialId, error instanceof Error ? error.message : String(error));
    }
  };

  const outputLines = batchRecord.output_file_id
    ? downloadBatchFileContent(batchRecord.output_file_id).then(parseJsonlLines)
    : Promise.resolve([]);
  const errorLines = batchRecord.error_file_id
    ? downloadBatchFileContent(batchRecord.error_file_id).then(parseJsonlLines)
    : Promise.resolve([]);
  const [successfulLines, failedLines] = await Promise.all([outputLines, errorLines]);

  for (const line of successfulLines) {
    handleBatchLine(JSON.stringify(line));
  }
  for (const line of failedLines) {
    handleBatchLine(JSON.stringify(line));
  }

  for (const source of sources) {
    if (!providerHandledMaterialIds.has(source.id)) {
      failureReasonsByMaterialId.set(source.id, "离线批处理未返回该物料结果");
    }
  }

  await Promise.all([
    deleteStandardizedResultsByMaterialIds(batchId, sources.map((source) => source.id)),
    deleteKeywordProcessesByMaterialIds(sources.map((source) => source.id), batchId),
    deleteRelatedSearchArtifactsByMaterialIds(batchId, sources.map((source) => source.id)),
  ]);

  let successRows = 0;
  let failedRows = 0;
  let keywordProcessed = 0;

  for (const source of sources) {
    const materialId = source.id;
    const taskItem = taskItemsByMaterialId.get(materialId);
    const parsedProduct = parsedProductsByMaterialId.get(materialId);
    const combinedResult = combinedResultsByMaterialId.get(materialId);
    const responseId = responseIdsByMaterialId.get(materialId) ?? "";

    if (!taskItem?.id) {
      continue;
    }

    if (!parsedProduct || !combinedResult) {
      const failureReason = failureReasonsByMaterialId.get(materialId) ?? "离线批处理未返回有效结果";
      failedRows += 1;
      pushStandardizeError(runState, `物料 ${materialId} 离线处理失败: ${failureReason}`);
      taskItemsByMaterialId.set(
        materialId,
        await updateStandardizationTaskItem(taskItem.id, {
          status: "failed",
          analysis_result: {},
          keyword_analysis: {},
          standardized_name: source.material_name,
          suggested_keyword: "",
          can_search_1688: null,
          missing_required_fields: [],
          failure_reason: failureReason,
          human_fix_hint: genericFixHint,
          patched_fields: mergeTaskItemPatchedFields(taskItem, {
            execution_mode: "offline",
            provider_response_id: responseId,
          }),
          llm_model: llmModel,
        }),
      );
      continue;
    }

    const standardizedRow = buildStandardizedResultRow({
      batchId,
      source,
      raw: parsedProduct.raw,
      result: combinedResult,
      llmModel,
    });
    const keywordAnalysis = toAiKeywordAnalysis(combinedResult);
    keywordProcessed += 1;

    const keywordAnalysisPayload = {
      ...keywordAnalysis,
      alt_search_terms: [...parsedProduct.analysis.alternative_terms],
      alternative_keywords: [...parsedProduct.analysis.alternative_terms],
    } as unknown as Record<string, unknown>;

    await upsertKeywordProcess(buildKeywordProcessRow({
      batchId,
      source,
      product: parsedProduct,
      llmModel,
      keywordAnalysisPayload,
      standardizedResult: standardizedRow,
    }));

    const itemFailed = keywordAnalysis.outcome === "failed" || keywordAnalysis.can_search_1688 === false;

    if (!itemFailed) {
      await upsertStandardizedResults([standardizedRow]);
    }

    taskItemsByMaterialId.set(
      materialId,
      await updateStandardizationTaskItem(taskItem.id, {
        status: itemFailed ? "failed" : "success",
        analysis_result: parsedProduct.analysis as unknown as Record<string, unknown>,
        keyword_analysis: keywordAnalysisPayload,
        standardized_name: standardizedRow.standardized_name,
        suggested_keyword: keywordAnalysis.suggested_keyword,
        can_search_1688: keywordAnalysis.can_search_1688,
        missing_required_fields: keywordAnalysis.missing_required_fields,
        failure_reason: keywordAnalysis.failure_reason,
        human_fix_hint: keywordAnalysis.human_fix_hint,
        patched_fields: mergeTaskItemPatchedFields(taskItem, {
          execution_mode: "offline",
          provider_response_id: responseId,
        }),
        llm_model: llmModel,
      }),
    );

    if (itemFailed) {
      failedRows += 1;
      pushStandardizeError(runState, `物料 ${materialId} 需要人工补充: ${keywordAnalysis.failure_reason || "缺少关键采购信息"}`);
    } else {
      successRows += 1;
      await resolveOutstandingStandardizationFailures({
        batchId,
        sourceMaterialId: taskItem.source_material_id || undefined,
        resolvedByTaskId: taskId,
      });
    }
  }

  return {
    successRows,
    failedRows,
    keywordInputRowsGenerated: 0,
    searchPlanRowsGenerated: 0,
    keywordProcessed,
  };
}

async function runOfflineStandardizationTask(params: {
  runState: StandardizeRunState;
  sources: StandardizationSourceSnapshot[];
  taskItemsByMaterialId: Map<string, StandardizationTaskItemRow>;
  llmModel: string;
  existingBatchId?: string;
}) {
  const { runState, sources, taskItemsByMaterialId, llmModel } = params;
  const batchId = runState.progress.batchId;
  const taskId = runState.progress.taskId;
  let providerBatchId = params.existingBatchId?.trim() ?? "";

  // 从 DB 加载最新型号字典，失败时静默降级到静态字典
  try {
    setRuntimeModelOverrides(await loadModelNumberOverridesFromDb());
  } catch (e) {
    console.warn("[standardize-offline] 型号字典 DB 加载失败，使用静态字典:", e);
  }

  try {
    await patchStandardizeProgress(runState, {
      status: "processing",
      executionMode: "offline",
      batchId,
      totalRows: sources.length,
      progressTotal: sources.length,
      progressCurrent: runState.progress.progressCurrent,
      processedRows: runState.progress.processedRows,
      successRows: runState.progress.successRows,
      failedRows: runState.progress.failedRows,
      result: null,
      errors: runState.progress.errors,
      startedAt: runState.progress.startedAt,
      finishedAt: null,
      providerStatus: runState.progress.providerStatus,
      currentPhase: runState.progress.currentPhase,
      currentItem: runState.progress.currentItem,
    });

    if (!providerBatchId) {
      const batchConfig = await resolveBigModelBatchRuntimeConfig();
      const jsonl = await buildOfflineBatchJsonl({
        taskId,
        model: batchConfig.model,
        sources,
      });

      await patchStandardizeProgress(runState, {
        providerStatus: "preparing",
        currentPhase: buildOfflinePhase("preparing"),
        currentItem: `正在生成 ${sources.length} 条离线标准化请求...`,
        progressCurrent: 0,
        progressTotal: sources.length,
        processedRows: 0,
      });

      const uploadedFile = await import("../standardization/bigmodel-batch-client.ts").then((mod) => mod.uploadBatchFile({
        filename: `standardization-${taskId}.jsonl`,
        content: jsonl,
      }));

      await patchStandardizeProgress(runState, {
        providerStatus: "uploaded",
        currentPhase: buildOfflinePhase("uploaded"),
        currentItem: `输入文件 ${uploadedFile.id} 已上传，正在创建 Batch 任务...`,
      });

      const batchRecord = await createBatchTask({
        inputFileId: uploadedFile.id,
        endpoint: "/v4/chat/completions",
        autoDeleteInputFile: true,
        metadata: {
          task_id: taskId,
          batch_id: batchId,
          description: `standardization-${batchId}`,
        },
      });
      providerBatchId = batchRecord.id;

      for (const taskItem of taskItemsByMaterialId.values()) {
        if (!taskItem.id) {
          continue;
        }

        const nextItem = await updateStandardizationTaskItem(taskItem.id, {
          patched_fields: mergeTaskItemPatchedFields(taskItem, {
            execution_mode: "offline",
            provider: "bigmodel-batch",
            provider_batch_id: providerBatchId,
            provider_input_file_id: uploadedFile.id,
          }),
          llm_model: batchConfig.model,
        });
        taskItemsByMaterialId.set(nextItem.source_material_id, nextItem);
      }

      await patchStandardizeProgress(runState, {
        providerStatus: String(batchRecord.status),
        currentPhase: buildOfflinePhase(String(batchRecord.status)),
        currentItem: buildOfflineCurrentItem({
          batchId: providerBatchId,
          totalRows: sources.length,
          completedRows: Math.max(0, Number(batchRecord.completed ?? 0) + Number(batchRecord.failed ?? 0)),
          failedRows: Math.max(0, Number(batchRecord.failed ?? 0)),
        }),
      });
    }

    while (true) {
      const batchRecord = await retrieveBatchTask(providerBatchId);
      const completedRows = Math.max(0, Number(batchRecord.completed ?? 0) + Number(batchRecord.failed ?? 0));
      const failedRows = Math.max(0, Number(batchRecord.failed ?? 0));

      await patchStandardizeProgress(runState, {
        executionMode: "offline",
        providerStatus: String(batchRecord.status),
        currentPhase: buildOfflinePhase(String(batchRecord.status)),
        currentItem: buildOfflineCurrentItem({
          batchId: providerBatchId,
          totalRows: sources.length,
          completedRows,
          failedRows,
        }),
        processedRows: Math.min(sources.length, completedRows),
        progressCurrent: Math.min(sources.length, completedRows),
        progressTotal: sources.length,
      });

      if (batchRecord.status === "completed") {
        const resultSummary = await applyOfflineBatchResults({
          runState,
          sources,
          taskItemsByMaterialId,
          batchRecord,
          llmModel,
        });

        await patchStandardizeProgress(runState, {
          processedRows: sources.length,
          progressCurrent: sources.length,
          progressTotal: sources.length,
          successRows: resultSummary.successRows,
          failedRows: resultSummary.failedRows,
          currentItem: resultSummary.failedRows === 0
            ? "离线标准化完成"
            : `离线标准化完成，${resultSummary.successRows} 条成功，${resultSummary.failedRows} 条需处理`,
        });

        await finalizeStandardizeRun(runState, "completed", {
          success: resultSummary.failedRows === 0,
          taskId,
          batchId,
          totalRows: sources.length,
          processedRows: sources.length,
          successRows: resultSummary.successRows,
          failedRows: resultSummary.failedRows,
          keywordInputRowsGenerated: resultSummary.keywordInputRowsGenerated,
          searchPlanRowsGenerated: resultSummary.searchPlanRowsGenerated,
          keywordProcessed: resultSummary.keywordProcessed,
        });
        return;
      }

      if (["failed", "expired", "cancelled"].includes(String(batchRecord.status))) {
        const reason = `离线批处理任务结束于状态 ${batchRecord.status}`;
        pushStandardizeError(runState, reason);
        await failPendingStandardizationTaskItems(taskId, reason);
        await finalizeStandardizeRun(runState, "failed", {
          success: false,
          taskId,
          batchId,
          totalRows: sources.length,
          processedRows: runState.progress.processedRows,
          successRows: runState.progress.successRows,
          failedRows: Math.max(runState.progress.failedRows, Number(batchRecord.failed ?? 0)),
          error: reason,
        });
        return;
      }

      await sleep(getOfflineBatchPollIntervalMs());
    }
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    pushStandardizeError(runState, message);
    await failPendingStandardizationTaskItems(taskId, message).catch(() => {
      // best effort
    });
    await finalizeStandardizeRun(runState, "failed", {
      success: false,
      taskId,
      batchId,
      totalRows: sources.length,
      processedRows: runState.progress.processedRows,
      successRows: runState.progress.successRows,
      failedRows: runState.progress.failedRows,
      error: message,
    });
  }
}

async function restorePendingStandardizationRuns() {
  const [onlineTasks, offlineTasks] = await Promise.all([
    getStandardizationTasksByModeAndStatuses("online", ["pending", "processing", "stopping"]),
    getStandardizationTasksByModeAndStatuses("offline", ["pending", "processing"], {
      orderBy: "created_at",
      ascending: true,
    }),
  ]);

  for (const task of onlineTasks) {
    if (!task.id) {
      continue;
    }

    const interruptedMessage = task.error_message || "服务已重启，在线任务未完成，请重新发起标准化";
    await failPendingStandardizationTaskItems(task.id, interruptedMessage).catch(() => {
      // best effort
    });
    await updateStandardizationTask(task.id, {
      execution_mode: "online",
      status: "failed",
      current_phase: "任务中断",
      error_message: interruptedMessage,
      finished_at: new Date().toISOString(),
    });
  }

  for (const task of offlineTasks) {
    if (!task.id) {
      continue;
    }

    if (process.env.ALLOW_LEGACY_OFFLINE_STANDARDIZATION_RESTORE !== "true") {
      const disabledMessage = "离线标准化已停用，请重新发起在线 GLM-5.1 标准化";
      const failedRunState: StandardizeRunState = {
        task,
        progress: mapTaskRowToStandardizeProgress(task, "offline"),
      };
      standardizeRuns.set(task.id, failedRunState);
      await failPendingStandardizationTaskItems(task.id, disabledMessage).catch(() => {
        // best effort
      });
      await finalizeStandardizeRun(failedRunState, "failed", {
        success: false,
        taskId: task.id,
        batchId: task.batch_id,
        totalRows: task.total_rows,
        processedRows: task.processed_rows,
        successRows: task.success_rows,
        failedRows: task.failed_rows,
        error: disabledMessage,
      });
      continue;
    }

    let restorableTask = task;
    if (task.status === "pending") {
      restorableTask = await updateStandardizationTask(task.id, {
        execution_mode: "offline",
        status: "processing",
        current_phase: buildOfflinePhase("preparing"),
        current_item: `正在准备 ${task.total_rows} 条离线标准化任务...`,
        error_message: "",
        started_at: task.started_at ?? new Date().toISOString(),
        finished_at: null,
      });
    }

    const runState: StandardizeRunState = {
      task: restorableTask,
      progress: mapTaskRowToStandardizeProgress(restorableTask, "offline"),
    };
    standardizeRuns.set(restorableTask.id ?? task.id, runState);

    const { sources, taskItemsByMaterialId, providerBatchId, llmModel } = await loadStandardizeTaskRunPayload(restorableTask);

    if (task.status !== "pending" && !providerBatchId) {
      const interruptedMessage = restorableTask.error_message || "离线任务缺少 Batch 标识，无法恢复，请重新发起标准化";
      await failPendingStandardizationTaskItems(task.id, interruptedMessage).catch(() => {
        // best effort
      });
      await finalizeStandardizeRun(runState, "failed", {
        success: false,
        taskId: task.id,
        batchId: task.batch_id,
        totalRows: task.total_rows,
        processedRows: task.processed_rows,
        successRows: task.success_rows,
        failedRows: task.failed_rows,
        error: interruptedMessage,
      });
      continue;
    }

    void runOfflineStandardizationTask({
      runState,
      sources,
      taskItemsByMaterialId,
      llmModel,
      existingBatchId: providerBatchId,
    });
  }
}

// ── GET /api/state — 返回当前状态（兼容前端轮询） ─────────────
app.get("/api/state", (_req, res) => {
  res.json({
    status: currentStatus,
    lastResult,
  });
});

// ── POST /api/upload — 上传 CSV 文件，检查重复，返回预览 ──────
app.post("/api/upload", upload.single("file"), async (req, res) => {
  try {
    // 检查文件是否存在
    if (!req.file) {
      res.status(400).json({ error: "请上传一个 CSV 文件" });
      return;
    }

    // 解析 CSV
    const rows = parseCsvBuffer(req.file.buffer);

    const existingIds: string[] = [];

    res.json({
      totalRows: rows.length,
      existingRows: existingIds.length,
      existingMaterialIds: existingIds,
      isNew: existingIds.length === 0,
      // 返回前3行预览
      preview: rows.slice(0, 3).map((row, i) => ({
        row_index: i + 1,
        名称: row.名称,
        规格型号: row.规格型号,
        数量: row.数量,
        单位: row.单位,
      })),
      message:
        existingIds.length > 0
          ? `检测到 ${existingIds.length} 行已存在。请确认是否追加新数据。`
          : `共 ${rows.length} 行新数据，可以直接写入。`,
      // 把 buffer 暂存到内存供 confirm 使用
      _buffered: true,
    });

    // 暂存文件 buffer 供后续 confirm 使用
    lastUploadedBuffer = req.file.buffer;
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    res.status(500).json({ error: msg });
  }
});

// ── POST /api/upload/confirm — 确认写入 Supabase ─────────────
const UploadConfirmBody = z.object({
  append: z.boolean().optional(),
});

app.post("/api/upload/confirm", async (req, res) => {
  try {
    const buffer = lastUploadedBuffer;

    if (!buffer) {
      res.status(400).json({ error: "没有待处理的文件，请先上传" });
      return;
    }

    // 是否追加模式（默认 true）
    const append = parseBody(req, UploadConfirmBody).append !== false;

    currentStatus = "processing";

    // 执行完整管道
    const result = await processCsvBuffer(buffer, { append });

    currentStatus = result.success ? "completed" : "failed";
    lastResult = result as unknown as Record<string, unknown>;

    // 清除暂存的 buffer
    lastUploadedBuffer = undefined;

    if (!result.success && result.errors.some((message) => message.includes("已弃用"))) {
      res.status(410).json(result);
      return;
    }

    res.json(result);
  } catch (err) {
    currentStatus = "failed";
    const msg = err instanceof Error ? err.message : String(err);
    lastResult = { error: msg };
    sendApiError(res, err);
  }
});

// ── GET /api/keyword-analysis — 读取 Supabase 最新阶段一分析报告 ─
app.get("/api/keyword-analysis", async (_req, res) => {
  try {
    const row = await getLatestKeywordProcess();

    if (!row) {
      res.status(404).json({ error: "Supabase 中暂无关键词分析数据，请先运行阶段一" });
      return;
    }

    const fullReport = row.full_report as Record<string, unknown>;

    res.json({
      row_name: row.row_name,
      row_spec: row.row_spec,
      ai_keyword_analysis: row.ai_keyword_analysis,
      search_plan_preview: fullReport.search_plan_preview ?? [],
    });
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    res.status(500).json({ error: msg });
  }
});

// ── GET /api/search-evaluation — 从 Supabase 读取最新搜索轮次 ───────
app.get("/api/search-evaluation", async (req, res) => {
  try {
    const materialIdParam = typeof req.query.material_id === "string" ? req.query.material_id.trim() : "";
    const batchIdParam = typeof req.query.batch_id === "string" ? req.query.batch_id.trim() : "";

    let keywordProcessRow: { batch_id: string; material_id: string; row_name: string } | null = null;

    if (batchIdParam && materialIdParam) {
      const { data, error } = await supabase
        .from("keyword_process")
        .select("batch_id, material_id, row_name")
        .eq("batch_id", batchIdParam)
        .eq("material_id", materialIdParam)
        .maybeSingle();
      if (error) throw error;
      keywordProcessRow = data;
    } else {
      const latest = await getLatestKeywordProcess();
      keywordProcessRow = latest
        ? {
            batch_id: latest.batch_id,
            material_id: latest.material_id,
            row_name: latest.row_name,
          }
        : null;
    }

    if (!keywordProcessRow) {
      res.status(404).json({ error: "暂无搜索记录" });
      return;
    }

    const { data: rounds, error: roundsError } = await supabase
      .from("search_rounds")
      .select("material_id, search_round, run_type, primary_keyword, search_strategy, is_valid_search, visual_confidence, visual_reason, stop_reason, results_count, results, screenshot_url, keyword_hit_rate, created_at")
      .eq("batch_id", keywordProcessRow.batch_id)
      .eq("material_id", keywordProcessRow.material_id)
      .order("created_at", { ascending: true })
      .order("search_round", { ascending: true });

    if (roundsError) throw roundsError;

    res.json({
      batch_id: keywordProcessRow.batch_id,
      material_id: keywordProcessRow.material_id,
      row_name: keywordProcessRow.row_name ?? "",
      rounds: (rounds ?? []).map((round) => ({
        material_id: round.material_id,
        search_round: round.search_round,
        run_type: round.run_type,
        primary_keyword: round.primary_keyword,
        search_strategy: round.search_strategy,
        entered_results_page: Number(round.results_count ?? 0) > 0,
        results_count: round.results_count,
        is_valid_search: round.is_valid_search,
        visual_confidence: round.visual_confidence,
        visual_reason: round.visual_reason,
        stop_reason: round.stop_reason,
        top_results: Array.isArray(round.results) ? round.results : [],
        results_path: "",
        page_screenshot_path: round.screenshot_url ?? "",
        keyword_hit_rate: round.keyword_hit_rate ?? null,
      })),
    });
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── GET /api/image-search — 图搜阶段数据 ───────────────────────
app.get("/api/image-search", async (_req, res) => {
  try {
    const kp = await getLatestKeywordProcess();
    if (!kp) {
      res.status(404).json({ error: "暂无数据" });
      return;
    }
    const [imageSearchRaw, bestMatchRaw] = await Promise.all([
      getImageSearchResultByMaterialId(kp.material_id),
      getBestMatchImageByMaterialId(kp.material_id),
    ]);
    if (!imageSearchRaw && !bestMatchRaw) {
      res.status(404).json({ error: "图搜阶段尚未执行" });
      return;
    }

    // ─ 表1：从 full_report.similar_products 补充 bestMatch.image_url ─
    const bestMatch = bestMatchRaw ? { ...bestMatchRaw } as Record<string, unknown> : null;
    if (bestMatch && !bestMatch.image_url && imageSearchRaw?.full_report) {
      const fr = imageSearchRaw.full_report as Record<string, unknown>;
      const prods = (fr.similar_products ?? []) as Array<{ offer_url?: string; image_url?: string }>;
      const offerUrl = String(bestMatch.offer_url ?? "");
      const offerId = offerUrl.match(/(\d{10,})/)?.[1];
      if (offerId) {
        const matched = prods.find((p) => p.offer_url?.includes(offerId));
        if (matched?.image_url) bestMatch.image_url = matched.image_url;
      }
    }

    // ─ 表2：如果 screenshot_url 为空，从 Storage 搜索截图 ─
    const imageSearch = imageSearchRaw ? { ...imageSearchRaw } as Record<string, unknown> : null;
    if (imageSearch && !imageSearch.screenshot_url) {
      const prefix = `material-${kp.material_id}`;
      const findImgInDir = async (dir: string): Promise<string> => {
        const { data: items } = await supabase.storage.from("screenshots").list(dir, { limit: 100 });
        if (!items) return "";
        for (const item of items) {
          if (/\.(jpg|jpeg|png|webp)$/i.test(item.name)) {
            return supabase.storage.from("screenshots").getPublicUrl(`${dir}/${item.name}`).data.publicUrl;
          }
          // 子目录递归找
          if (!item.id) {
            const url = await findImgInDir(`${dir}/${item.name}`);
            if (url) return url;
          }
        }
        return "";
      };
      const url = await findImgInDir(prefix);
      if (url) imageSearch.screenshot_url = url;
    }

    // ─ 表3：提取图搜失败原因 ─
    const FAILURE_REASON_MAP: Record<string, string> = {
      insufficient_image_samples: "图搜取回的图片样本不足，无法完成相似度较更",
      low_cluster_confidence: "商品图片聚类置信度过低，未找到明确匹配",
      text_rule_failed: "标题文本相关性检查未通过",
      no_similar_products: "未找到相似商品",
    };
    if (imageSearch && !imageSearchRaw?.is_valid_image_search) {
      const imgSim = (imageSearchRaw?.image_similarity ?? {}) as Record<string, unknown>;
      const rawReason = String(imgSim.reason ?? "");
      const failureReason = FAILURE_REASON_MAP[rawReason] ?? rawReason ?? "原因未知";
      (imageSearch as Record<string, unknown>).failure_reason = failureReason;
    }

    res.json({ imageSearch, bestMatch });
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

app.get("/api/keyword-selection", async (_req, res) => {
  try {
    const kp = await getLatestKeywordProcess();
    if (!kp) { res.status(404).json({ error: "暂无数据" }); return; }
    const sel = await getKeywordSelectionByMaterialId(kp.material_id);
    if (!sel) { res.status(404).json({ error: "关键词选定阶段尚未执行" }); return; }
    const debug = sel.debug_data as Record<string, unknown>;
    const isFallback = Boolean(debug.is_fallback);
    const fallbackReason = String(debug.fallback_reason ?? "");
    // is_fallback=true 意味着多模态判定失败，回退到阶段一 AI 分析结果
    const isSameAsFirstRound = isFallback;
    // 阶段一 AI 建议关键词（从 keyword_process 表的 ai_keyword_analysis 字段读取）
    const aiAnalysis = kp.ai_keyword_analysis as Record<string, unknown>;
    const aiSuggestedKeyword = String(aiAnalysis.suggested_keyword ?? "");
    res.json({
      main_keyword: sel.main_keyword,
      judgment_reason: sel.judgment_reason,
      is_fallback: isFallback,
      fallback_reason: fallbackReason,
      is_same_as_first_round: isSameAsFirstRound,
      ai_suggested_keyword: aiSuggestedKeyword,
      row_name: kp.row_name,
    });
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/standardize-batch — 读取采购清单调用 LLM 标准化 ──
const StandardizeBatchMaterial = z
  .object({
    id: z.string().optional(),
    material_name: z.string().optional(),
    specification: z.string().optional(),
    quantity: z.number().optional(),
    unit: z.string().optional(),
    invoice_type: z.string().optional(),
    brand: z.string().optional(),
    remarks: z.string().optional(),
    delivery_address: z.string().optional(),
  })
  .passthrough();

const StandardizeBatchBody = z.object({
  batch_id: z.string().trim().min(1, "缺少 batch_id 参数"),
  material_ids: z.array(z.string()).optional(),
  materials: z.array(StandardizeBatchMaterial).optional(),
});

const FailureSupplementBody = z.object({
  spec_model: z.string().optional(),
  brand: z.string().optional(),
  material: z.string().optional(),
  usage_scene: z.string().optional(),
  execution_standard: z.string().optional(),
  supplement_notes: z.string().optional(),
}).partial();

type FailureSupplementBody = z.infer<typeof FailureSupplementBody>;

const RetryStandardizationFailureBody = z.object({
  failure_id: z.string().trim().min(1, "缺少 failure_id"),
  supplement: FailureSupplementBody.optional(),
});

const RetryStandardizationFailuresBody = z.object({
  failures: z.array(RetryStandardizationFailureBody).min(1, "至少需要 1 条失败记录"),
});

function normalizeSupplementInput(value: unknown) {
  return String(value ?? "").replace(/\s+/g, " ").trim();
}

function buildHumanSupplementContext(supplement: FailureSupplementBody) {
  return [
    ["规格型号", supplement.spec_model],
    ["品牌", supplement.brand],
    ["材质", supplement.material],
    ["用途/场景", supplement.usage_scene],
    ["执行标准", supplement.execution_standard],
    ["补充说明", supplement.supplement_notes],
  ]
    .map(([label, value]) => {
      const text = normalizeSupplementInput(value);
      return text ? `${label}：${text}` : "";
    })
    .filter(Boolean)
    .join("\n");
}

function joinSupplementContext(values: string[]) {
  return values
    .map((value) => value.trim())
    .filter(Boolean)
    .join("\n");
}

function buildRetryStandardizationSource(params: {
  item: StandardizationTaskItemRow;
  index: number;
  supplement: FailureSupplementBody;
}): StandardizationSourceSnapshot {
  const source = normalizeStandardizationSourceSnapshot(
    params.item.source_snapshot as Record<string, unknown>,
    params.index,
    params.item.batch_id,
  );
  const supplement = params.supplement;
  const context = buildHumanSupplementContext(supplement);

  return {
    ...source,
    id: params.item.source_material_id || source.id,
    batch_id: params.item.batch_id || source.batch_id,
    specification: normalizeSupplementInput(supplement.spec_model) || source.specification,
    brand: normalizeSupplementInput(supplement.brand) || source.brand,
    technical_standard: normalizeSupplementInput(supplement.execution_standard) || source.technical_standard,
    material_requirement: joinSupplementContext([
      normalizeSupplementInput(supplement.material),
      source.material_requirement,
    ]),
    applicable_conditions: joinSupplementContext([
      normalizeSupplementInput(supplement.usage_scene),
      source.applicable_conditions,
    ]),
    remarks: joinSupplementContext([
      source.remarks,
      context ? `人工补充信息：\n${context}` : "",
    ]),
  };
}

function formatRetryFailureMessage(taskItem: Pick<StandardizationTaskItemRow, "retry_of_item_id"> | undefined, message: string) {
  const normalized = message.trim() || "缺少关键采购信息";
  return taskItem?.retry_of_item_id
    ? `重跑失败，仍需处理：${normalized}`
    : normalized;
}

app.post("/api/standardize-batch", async (req, res) => {
  try {
    const { batch_id, material_ids, materials: selectedMaterials } = parseBody(
      req,
      StandardizeBatchBody,
    );

    let sourceMaterials: Record<string, unknown>[] = [];
    const fallbackSelectedMaterials = (selectedMaterials ?? []).map((item, index) => ({
      id: String(item.id ?? `selected-${index + 1}`),
      material_name: String(item.material_name ?? ""),
      specification: String(item.specification ?? ""),
      quantity: Number(item.quantity ?? 0),
      unit: String(item.unit ?? ""),
      invoice_type: String(item.invoice_type ?? ""),
      brand: String(item.brand ?? ""),
      remarks: String(item.remarks ?? ""),
      delivery_address: String(item.delivery_address ?? ""),
    }));

    if (material_ids && material_ids.length > 0) {
      const { data: allMaterials, error: matError } = await supabase
        .from("procurement_materials")
        .select("*")
        .eq("batch_id", batch_id)
        .order("created_at", { ascending: true });

      if (matError) {
        res.status(500).json({ error: `读取采购清单失败: ${matError.message}` });
        return;
      }

      if (allMaterials && allMaterials.length > 0) {
        const selectedFromDb = allMaterials
          .filter((m: Record<string, unknown>) => material_ids.includes(String(m.id)));

        if (selectedFromDb.length === material_ids.length) {
          sourceMaterials = selectedFromDb;
        }
      }

      if (sourceMaterials.length === 0 && fallbackSelectedMaterials.length > 0) {
        sourceMaterials = fallbackSelectedMaterials;
      }
    } else if (fallbackSelectedMaterials.length > 0) {
      sourceMaterials = fallbackSelectedMaterials;
    } else {
      // 从 procurement_materials 读取该批次的全部物料（保持 created_at 顺序稳定）
      const { data: allMaterials, error: matError } = await supabase
        .from("procurement_materials")
        .select("*")
        .eq("batch_id", batch_id)
        .order("created_at", { ascending: true });

      if (matError) {
        res.status(500).json({ error: `读取采购清单失败: ${matError.message}` });
        return;
      }

      if (!allMaterials || allMaterials.length === 0) {
        res.status(404).json({ error: `批次 ${batch_id} 下没有采购物料` });
        return;
      }

      sourceMaterials = allMaterials;
    }

    if (sourceMaterials.length === 0) {
      res.status(404).json({ error: "未找到符合条件的采购物料" });
      return;
    }

    const sources = sourceMaterials
      .map((material, index) => normalizeStandardizationSourceSnapshot(material, index, batch_id))
      .sort((left, right) => left.created_at.localeCompare(right.created_at) || left.id.localeCompare(right.id));
    const executionMode: StandardizationExecutionMode = "online";

    if (executionMode === "online") {
      const activeOnlineRun = getActiveStandardizeRunByMode("online");
      if (activeOnlineRun) {
        res.status(409).json({
          error: "已有在线标准化任务正在执行",
          batchId: activeOnlineRun.progress.batchId,
          executionMode: "online",
        });
        return;
      }

      const persistedOnlineTask = (await getStandardizationTasksByModeAndStatuses("online", ["processing", "stopping"]))[0];
      if (persistedOnlineTask?.id) {
        res.status(409).json({
          error: "已有在线标准化任务正在执行",
          batchId: persistedOnlineTask.batch_id,
          executionMode: "online",
        });
        return;
      }
    }
    const runtimeConfig = await resolveLlmRuntimeConfig();
    const llmModel = resolveOnlineStandardizationModel(runtimeConfig.model);

    const startedAt = new Date().toISOString();
    const task = await createStandardizationTask({
      batch_id,
      execution_mode: executionMode,
      status: "processing",
      total_rows: sources.length,
      processed_rows: 0,
      success_rows: 0,
      failed_rows: 0,
      current_item: "正在准备标准化任务...",
      current_phase: "准备任务",
      error_message: "",
      started_at: startedAt,
      finished_at: null,
    });

    if (!task.id) {
      throw new Error("标准化任务创建成功但缺少任务 ID");
    }
    const taskId = task.id;

    const runState: StandardizeRunState = {
      task,
      progress: mapTaskRowToStandardizeProgress(task, executionMode),
      abortController: new AbortController(),
      stopMessage: null,
    };
    standardizeRuns.set(taskId, runState);

    const taskItems = await insertStandardizationTaskItems(
      sources.map((source) => ({
        task_id: taskId,
        batch_id,
        source_material_id: source.id,
        status: "pending",
        source_snapshot: source,
        analysis_result: {},
        keyword_analysis: {},
        standardized_name: "",
        suggested_keyword: "",
        can_search_1688: null,
        missing_required_fields: [],
        failure_reason: "",
        human_fix_hint: "",
        patched_fields: {
          execution_mode: "online",
        },
        llm_model: llmModel,
        retry_of_item_id: null,
        resolved_by_task_id: null,
        resolved_at: null,
      })),
    );
    const taskItemsByMaterialId = new Map(taskItems.map((item) => [item.source_material_id, item]));

    // 立即返回，后台异步执行
    res.json({
      message: `在线标准化已提交，共 ${sources.length} 条物料，将使用 ${llmModel} 批量分析`,
      totalRows: sources.length,
      taskId: taskId,
      executionMode,
      taskStatus: "processing",
    });

    void runOnlineStandardizationTask({
      runState,
      sources,
      taskItemsByMaterialId,
      llmModel,
    });
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    res.status(500).json({ error: msg });
  }
});

// ── POST /api/standardization-failures/retry — 人工补充后重跑失败项 ──
app.post("/api/standardization-failures/retry", async (req, res) => {
  try {
    const { failures } = parseBody(req, RetryStandardizationFailuresBody);
    const requested = failures.map((failure) => ({
      failureId: failure.failure_id.trim(),
      supplement: failure.supplement ?? {},
    }));
    const requestedIds = [...new Set(requested.map((failure) => failure.failureId))];

    const { data, error } = await supabase
      .from("standardization_task_items")
      .select("*")
      .eq("status", "failed")
      .in("id", requestedIds);

    if (error) {
      throw new Error(`读取标准化失败记录失败: ${error.message}`);
    }

    const itemMap = new Map(
      ((data ?? []) as StandardizationTaskItemRow[]).map((item) => [String(item.id ?? ""), item]),
    );
    const rejected: Array<{ failureId: string; error: string }> = [];
    const accepted: Array<{
      failureId: string;
      item: StandardizationTaskItemRow;
      supplement: FailureSupplementBody;
    }> = [];
    const acceptedMaterialIds = new Set<string>();

    for (const requestItem of requested) {
      const item = itemMap.get(requestItem.failureId);
      if (!item) {
        rejected.push({ failureId: requestItem.failureId, error: "未找到待处理的失败记录" });
        continue;
      }

      if (item.resolved_at) {
        rejected.push({ failureId: requestItem.failureId, error: "失败记录已处理，不能重跑" });
        continue;
      }

      if (!item.source_material_id) {
        rejected.push({ failureId: requestItem.failureId, error: "失败记录缺少物料 ID" });
        continue;
      }

      if (acceptedMaterialIds.has(item.source_material_id)) {
        rejected.push({ failureId: requestItem.failureId, error: "同一物料已在本次重跑任务中" });
        continue;
      }

      acceptedMaterialIds.add(item.source_material_id);
      accepted.push({
        failureId: requestItem.failureId,
        item,
        supplement: requestItem.supplement,
      });
    }

    if (accepted.length === 0) {
      res.status(400).json({
        error: "没有可重跑的标准化失败记录",
        acceptedIds: [],
        rejected,
      });
      return;
    }

    const batchIds = [...new Set(accepted.map(({ item }) => item.batch_id).filter(Boolean))];
    if (batchIds.length !== 1) {
      res.status(400).json({
        error: "一次只能重跑同一批次的失败记录",
        acceptedIds: [],
        rejected: [
          ...rejected,
          ...accepted.map(({ failureId }) => ({ failureId, error: "所选失败记录跨批次，请按批次分别重跑" })),
        ],
      });
      return;
    }

    const batchId = batchIds[0]!;
    const activeOnlineRun = getActiveStandardizeRunByMode("online");
    if (activeOnlineRun) {
      res.status(409).json({
        error: "已有在线标准化任务正在执行",
        batchId: activeOnlineRun.progress.batchId,
        executionMode: "online",
        acceptedIds: [],
        rejected,
      });
      return;
    }

    const persistedOnlineTask = (await getStandardizationTasksByModeAndStatuses("online", ["processing", "stopping"]))[0];
    if (persistedOnlineTask?.id) {
      res.status(409).json({
        error: "已有在线标准化任务正在执行",
        batchId: persistedOnlineTask.batch_id,
        executionMode: "online",
        acceptedIds: [],
        rejected,
      });
      return;
    }

    const runtimeConfig = await resolveLlmRuntimeConfig();
    const llmModel = resolveOnlineStandardizationModel(runtimeConfig.model);
    const startedAt = new Date().toISOString();
    const sources = accepted
      .map((entry, index) => buildRetryStandardizationSource({
        item: entry.item,
        index,
        supplement: entry.supplement,
      }))
      .sort((left, right) => left.created_at.localeCompare(right.created_at) || left.id.localeCompare(right.id));
    const acceptedBySourceId = new Map(accepted.map((entry) => [entry.item.source_material_id, entry]));

    const task = await createStandardizationTask({
      batch_id: batchId,
      execution_mode: "online",
      status: "processing",
      total_rows: sources.length,
      processed_rows: 0,
      success_rows: 0,
      failed_rows: 0,
      current_item: "正在准备失败项重跑任务...",
      current_phase: "准备重跑",
      error_message: "",
      started_at: startedAt,
      finished_at: null,
    });

    if (!task.id) {
      throw new Error("重跑任务创建成功但缺少任务 ID");
    }

    const runState: StandardizeRunState = {
      task,
      progress: mapTaskRowToStandardizeProgress(task, "online"),
      abortController: new AbortController(),
      stopMessage: null,
    };
    standardizeRuns.set(task.id, runState);

    const taskItems = await insertStandardizationTaskItems(
      sources.map((source) => {
        const entry = acceptedBySourceId.get(source.id);
        if (!entry) {
          throw new Error(`重跑任务缺少失败记录映射: ${source.id}`);
        }

        const humanSupplementContext = buildHumanSupplementContext(entry.supplement);
        return {
          task_id: task.id!,
          batch_id: batchId,
          source_material_id: source.id,
          status: "pending" as const,
          source_snapshot: source,
          analysis_result: {},
          keyword_analysis: {},
          standardized_name: "",
          suggested_keyword: "",
          can_search_1688: null,
          missing_required_fields: [],
          failure_reason: "",
          human_fix_hint: "",
          patched_fields: {
            execution_mode: "online",
            retry_of_task_id: entry.item.task_id,
            retry_of_item_id: entry.failureId,
            human_supplement: entry.supplement,
            human_supplement_context: humanSupplementContext,
          },
          llm_model: llmModel,
          retry_of_item_id: entry.failureId,
          resolved_by_task_id: null,
          resolved_at: null,
        };
      }),
    );
    const taskItemsByMaterialId = new Map(taskItems.map((item) => [item.source_material_id, item]));

    res.json({
      message: `失败项重跑已提交，共 ${sources.length} 条物料，将使用 ${llmModel} 批量分析`,
      totalRows: sources.length,
      taskId: task.id,
      executionMode: "online",
      taskStatus: "processing",
      acceptedIds: accepted.map((entry) => entry.failureId),
      rejected,
    });

    void runOnlineStandardizationTask({
      runState,
      sources,
      taskItemsByMaterialId,
      llmModel,
    });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── GET /api/standardize-status — 轮询标准化进度 ──────────────
app.get("/api/standardize-status", async (_req, res) => {
  try {
    const progress = listStandardizeProgresses().length > 0
      ? getLatestStandardizeProgress()
      : await getPersistedLatestStandardizeProgress();

    res.json(progress);
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/standardize-stop — 停止当前在线标准化任务 ──────────
const StandardizeStopBody = z.object({
  task_id: z.string().trim().min(1).optional(),
});

app.post("/api/standardize-stop", async (req, res) => {
  try {
    const { task_id } = parseBody(req, StandardizeStopBody);
    const runState = task_id
      ? standardizeRuns.get(task_id)
      : getActiveStandardizeRunByMode("online");

    if (!runState || runState.progress.executionMode !== "online") {
      res.status(400).json({ error: "当前没有正在运行的在线标准化任务" });
      return;
    }

    if (runState.progress.status === "stopping") {
      res.json({
        message: "在线标准化任务正在停止，请稍候刷新状态",
        taskId: runState.progress.taskId,
        batchId: runState.progress.batchId,
      });
      return;
    }

    await requestOnlineStandardizeStop(runState, "用户手动停止在线标准化任务");
    res.json({
      message: "在线标准化停止信号已发送，正在中断当前请求",
      taskId: runState.progress.taskId,
      batchId: runState.progress.batchId,
    });
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── 1688 自动化搜索状态 ───────────────────────────────────────
type SearchProgressStatus = "idle" | SearchTaskStatus;

interface SearchProgress {
  taskId: string;
  status: SearchProgressStatus;
  batchId: string;
  accountId: string;
  totalRows: number;
  processedRows: number;
  completedRows: number;
  failedRows: number;
  stoppedRows: number;
  currentRow: string;
  currentItem: string;
  errors: string[];
  result: Record<string, unknown> | null;
  displayProgressPercent?: number;
  currentStepLabel?: string;
  searchProgress?: SearchProgressSnapshot | null;
  startedAt: string | null;
  pausedAt: string | null;
  finishedAt: string | null;
}

type SearchChildTerminationReason = "pause" | "stop";

interface SearchRunState {
  task: SearchTaskRow;
  progress: SearchProgress;
  childProcess?: ChildProcess | null;
  childKillTimer?: ReturnType<typeof setTimeout> | null;
  childTerminationReason?: SearchChildTerminationReason | null;
  stopMessage?: string | null;
}

const SEARCH_ALL_STATUSES: SearchTaskStatus[] = [
  "pending",
  "processing",
  "waiting_for_manual",
  "pausing",
  "paused",
  "stopping",
  "stopped",
  "completed",
  "failed",
];
const SEARCH_BLOCKING_STATUSES: SearchTaskStatus[] = ["pending", "processing", "waiting_for_manual", "pausing", "paused", "stopping"];
const SEARCH_RECOVERABLE_STATUSES: SearchTaskStatus[] = ["pending", "processing", "waiting_for_manual", "pausing", "stopping"];
const SEARCH_CHILD_OUTPUT_LIMIT_BYTES = 20 * 1024 * 1024;
const SEARCH_CHILD_KILL_TIMEOUT_MS = 2_000;
const SEARCH_CHILD_EVENT_PREFIX = "__SEARCH_EVENT__:";
const searchRuns = new Map<string, SearchRunState>();

type SearchChildEventType = "manual_intervention_required" | "manual_intervention_message" | "manual_intervention_resolved" | "progress_update";

type SearchChildEvent = {
  type: SearchChildEventType;
  alertId?: string;
  kind?: string;
  message?: string;
  materialId?: string;
  materialName?: string;
  accountId?: string;
  pageUrl?: string;
  taskId?: string;
  detectedAt?: string;
  resolvedAt?: string;
  itemProgressPercent?: number;
  softCapPercent?: number;
  stageStartedAt?: string;
};

type SearchProgressSnapshot = {
  message: string;
  materialId: string;
  materialName: string;
  itemProgressPercent: number;
  softCapPercent: number | null;
  stageStartedAt: string;
  updatedAt: string;
};

function clampPercent(value: unknown) {
  const numeric = Number(value ?? 0);
  if (!Number.isFinite(numeric)) return 0;
  return Math.min(100, Math.max(0, Math.round(numeric)));
}

function readSearchProgressSnapshot(result: Record<string, unknown> | null): SearchProgressSnapshot | null {
  const progress = result?.searchProgress;
  if (!progress || typeof progress !== "object" || Array.isArray(progress)) {
    return null;
  }
  const record = progress as Record<string, unknown>;
  return {
    message: String(record.message ?? ""),
    materialId: String(record.materialId ?? ""),
    materialName: String(record.materialName ?? ""),
    itemProgressPercent: clampPercent(record.itemProgressPercent),
    softCapPercent: record.softCapPercent == null ? null : clampPercent(record.softCapPercent),
    stageStartedAt: String(record.stageStartedAt ?? ""),
    updatedAt: String(record.updatedAt ?? ""),
  };
}

function computeSearchDisplayProgress(params: {
  status: SearchProgressStatus | SearchTaskStatus;
  totalRows: number;
  processedRows: number;
  result: Record<string, unknown> | null;
}) {
  if (params.status === "completed") {
    return 100;
  }
  if (params.totalRows <= 0) {
    return 0;
  }

  const snapshot = readSearchProgressSnapshot(params.result);
  const currentItemFraction = params.status === "processing" || params.status === "waiting_for_manual"
    ? (snapshot?.itemProgressPercent ?? 0) / 100
    : 0;
  return clampPercent(((params.processedRows + currentItemFraction) / params.totalRows) * 100);
}

function buildSearchProgressResult(
  runState: SearchRunState,
  event: SearchChildEvent,
): Record<string, unknown> {
  const previous = runState.progress.result && typeof runState.progress.result === "object"
    ? runState.progress.result
    : {};
  const message = event.message || runState.progress.currentItem || "搜索任务处理中";
  return {
    ...previous,
    searchProgress: {
      message,
      materialId: event.materialId || runState.progress.currentRow,
      materialName: event.materialName || "",
      itemProgressPercent: clampPercent(event.itemProgressPercent),
      softCapPercent: event.softCapPercent == null ? null : clampPercent(event.softCapPercent),
      stageStartedAt: event.stageStartedAt || new Date().toISOString(),
      updatedAt: new Date().toISOString(),
    },
  };
}

function getSearchManualInterventionCommandPath(taskId: string) {
  return path.join(getSearchTaskRuntimePaths(taskId).rootDir, "manual-intervention-command.json");
}

async function cleanupSearchTaskRuntimeWorkspace(taskId: string) {
  if (!taskId) {
    return;
  }

  if (shouldPreserveSearchTaskRuntimeWorkspace()) {
    console.log(`[search] 已保留任务运行目录: ${getSearchTaskRuntimePaths(taskId).rootDir}`);
    return;
  }

  await rm(getSearchTaskRuntimePaths(taskId).rootDir, { recursive: true, force: true });
}

function normalizeBooleanEnv(value: string | undefined | null) {
  const normalized = String(value ?? "").trim().toLowerCase();
  if (!normalized) {
    return null;
  }
  if (["1", "true", "yes", "y", "on"].includes(normalized)) {
    return true;
  }
  if (["0", "false", "no", "n", "off"].includes(normalized)) {
    return false;
  }
  return null;
}

function shouldPreserveSearchTaskRuntimeWorkspace() {
  const explicitPreserve = normalizeBooleanEnv(process.env.SEARCH_PRESERVE_TASK_RUNTIME);
  if (explicitPreserve != null) {
    return explicitPreserve;
  }

  const explicitCleanup = normalizeBooleanEnv(process.env.SEARCH_CLEANUP_TASK_RUNTIME);
  if (explicitCleanup != null) {
    return !explicitCleanup;
  }

  return true;
}

async function listSearchRuntimeFiles(rootDir: string) {
  const files: string[] = [];

  async function walk(currentDir: string) {
    let entries: Array<{ name: string; isDirectory: () => boolean; isFile: () => boolean }>;
    try {
      entries = await readdir(currentDir, { withFileTypes: true });
    } catch {
      return;
    }

    for (const entry of entries) {
      const fullPath = path.join(currentDir, entry.name);
      if (entry.isDirectory()) {
        await walk(fullPath);
      } else if (entry.isFile()) {
        files.push(fullPath);
      }
    }
  }

  await walk(rootDir);
  return files;
}

async function readJsonArtifact(filePath: string): Promise<unknown | null> {
  try {
    return JSON.parse(await readFile(filePath, "utf8"));
  } catch {
    return null;
  }
}

function readFiniteNumber(value: unknown) {
  const numeric = Number(value);
  return Number.isFinite(numeric) ? numeric : 0;
}

function pickSearchShortageReason(reasons: string[]) {
  if (reasons.includes("no_search_candidates")) {
    return "no_search_candidates";
  }
  return reasons.find(Boolean) || "";
}

function buildSearchShortageMessage(summary: {
  targetResultCount: number;
  qualifiedResultCount: number;
  shortageReason: string;
}) {
  if (!summary.shortageReason || summary.qualifiedResultCount >= summary.targetResultCount) {
    return "";
  }
  if (summary.qualifiedResultCount <= 0) {
    if (summary.shortageReason === "no_search_candidates") {
      return "搜索完成但未获得有效商家：没有搜索候选商品";
    }
    if (summary.shortageReason === "llm_batch_failed") {
      return `搜索完成但未获得有效商家：${resolveBigModelBatchModel()} 离线 Batch 商品匹配失败`;
    }
    return "搜索完成但未获得有效商家：LLM 无匹配或有效电话不足";
  }
  return `搜索完成，有效商家不足：${summary.qualifiedResultCount}/${summary.targetResultCount}`;
}

async function buildSearchRuntimeArtifactSummary(taskId: string) {
  const runtimePaths = getSearchTaskRuntimePaths(taskId);
  const files = await listSearchRuntimeFiles(runtimePaths.outputRoot);
  const byName = (fileName: string) => files.filter((filePath) => path.basename(filePath) === fileName);
  const llmMatchResults = byName("llm-match-results.json");
  const llmBatchInputs = byName("llm-batch-input.jsonl");
  const llmBatchUploadedFiles = byName("llm-batch-uploaded-file.json");
  const llmBatchRecords = byName("llm-batch-record.json");
  const llmBatchOutputs = byName("llm-batch-output.jsonl");
  const llmBatchErrors = [
    ...byName("llm-batch-error.jsonl"),
    ...byName("llm-batch-error.json"),
  ];
  const contactEnrichmentResults = byName("contact-enrichment-results.json");
  const finalSelectedProducts = byName("final-selected-products.json");
  const runSummaries = byName("run-summary.json");

  let llmCandidateCount = 0;
  let llmMatchedCount = 0;
  for (const filePath of llmMatchResults) {
    const parsed = await readJsonArtifact(filePath);
    const rows = Array.isArray(parsed) ? parsed : [];
    llmCandidateCount += rows.length;
    llmMatchedCount += rows.filter((row) => Boolean((row as Record<string, unknown>)?.is_match)).length;
  }

  let attemptedContactCount = 0;
  let qualifiedResultCount = 0;
  let targetResultCount = 0;
  const shortageReasons: string[] = [];
  for (const filePath of contactEnrichmentResults) {
    const parsed = await readJsonArtifact(filePath);
    if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
      continue;
    }
    const record = parsed as Record<string, unknown>;
    attemptedContactCount += readFiniteNumber(record.attemptedCount);
    qualifiedResultCount += readFiniteNumber(record.qualifiedCount);
    targetResultCount += readFiniteNumber(record.targetQualifiedCount);
    const shortageReason = String(record.shortageReason ?? "").trim();
    if (shortageReason) {
      shortageReasons.push(shortageReason);
    }
  }

  const shortageReason = pickSearchShortageReason(shortageReasons);
  const shortageSummary = targetResultCount > 0 && shortageReason
    ? {
      targetResultCount,
      qualifiedResultCount,
      shortageReason,
      message: buildSearchShortageMessage({
        targetResultCount,
        qualifiedResultCount,
        shortageReason,
      }),
    }
    : null;

  return {
    runtimeRoot: runtimePaths.rootDir,
    outputRoot: runtimePaths.outputRoot,
    preserved: shouldPreserveSearchTaskRuntimeWorkspace(),
    llmMatchResults,
    llmBatchInputs,
    llmBatchUploadedFiles,
    llmBatchRecords,
    llmBatchOutputs,
    llmBatchErrors,
    contactEnrichmentResults,
    finalSelectedProducts,
    runSummaries,
    counts: {
      llmCandidateCount,
      llmMatchedCount,
      attemptedContactCount,
      qualifiedResultCount,
      targetResultCount,
    },
    shortageSummary,
  };
}

async function attachSearchRuntimeArtifacts(taskId: string, result: Record<string, unknown>) {
  const searchArtifacts = await buildSearchRuntimeArtifactSummary(taskId);
  return {
    ...result,
    searchArtifacts,
    ...(searchArtifacts.shortageSummary ? { searchShortage: searchArtifacts.shortageSummary } : {}),
  };
}

function createSearchProgress(overrides: Partial<SearchProgress> = {}): SearchProgress {
  return {
    taskId: "",
    status: "idle",
    batchId: "",
    accountId: "",
    totalRows: 0,
    processedRows: 0,
    completedRows: 0,
    failedRows: 0,
    stoppedRows: 0,
    currentRow: "",
    currentItem: "",
    errors: [...(overrides.errors ?? [])],
    result: overrides.result ? { ...overrides.result } : null,
    startedAt: null,
    pausedAt: null,
    finishedAt: null,
    ...overrides,
  };
}

function mapSearchTaskRowToProgress(task: SearchTaskRow): SearchProgress {
  const result = task.result && Object.keys(task.result).length > 0
    ? { ...task.result }
    : null;
  const searchProgress = readSearchProgressSnapshot(result);
  const displayProgressPercent = computeSearchDisplayProgress({
    status: task.status,
    totalRows: task.total_rows,
    processedRows: task.processed_rows,
    result,
  });

  return createSearchProgress({
    taskId: task.id ?? "",
    status: task.status,
    batchId: task.batch_id,
    accountId: task.account_id,
    totalRows: task.total_rows,
    processedRows: task.processed_rows,
    completedRows: task.completed_rows,
    failedRows: task.failed_rows,
    stoppedRows: task.stopped_rows,
    currentRow: task.current_material_id ?? "",
    currentItem: task.current_item,
    errors: [...task.errors],
    result,
    displayProgressPercent,
    currentStepLabel: searchProgress?.message || task.current_item,
    searchProgress,
    startedAt: task.started_at ?? null,
    pausedAt: task.paused_at ?? null,
    finishedAt: task.finished_at ?? null,
  });
}

function getSearchRunState(task: SearchTaskRow): SearchRunState {
  const taskId = task.id ?? "";
  const existing = taskId ? searchRuns.get(taskId) : undefined;

  if (existing) {
    existing.task = task;
    if (!existing.childProcess) {
      existing.progress = mapSearchTaskRowToProgress(task);
    }
    return existing;
  }

  const runState: SearchRunState = {
    task,
    progress: mapSearchTaskRowToProgress(task),
    childProcess: null,
    childKillTimer: null,
    childTerminationReason: null,
    stopMessage: null,
  };

  if (taskId) {
    searchRuns.set(taskId, runState);
  }

  return runState;
}

async function getLatestSearchTask(statuses: SearchTaskStatus[] = SEARCH_ALL_STATUSES): Promise<SearchTaskRow | null> {
  const tasks = await getSearchTasksByStatuses(statuses);
  return tasks[0] ?? null;
}

async function getLatestVisibleSearchTask(
  user: ApiUserContext,
  statuses: SearchTaskStatus[] = SEARCH_ALL_STATUSES,
): Promise<SearchTaskRow | null> {
  const tasks = await getSearchTasksByStatuses(statuses);
  if (tasks.length === 0 || user.isAdmin) {
    return tasks[0] ?? null;
  }

  const [accessibleBatchIds, accessibleAccountIds] = await Promise.all([
    getAccessibleBatchIdSet(user, tasks.map((task) => task.batch_id)),
    getAccessibleSearchAccountIdSet(user, tasks.map((task) => task.account_id)),
  ]);

  return tasks.find(
    (task) => accessibleBatchIds.has(task.batch_id) && accessibleAccountIds.has(task.account_id),
  ) ?? null;
}

async function canManageSearchTask(user: ApiUserContext, task: SearchTaskRow): Promise<boolean> {
  if (user.isAdmin) {
    return true;
  }

  const [accessibleBatchIds, accessibleAccountIds] = await Promise.all([
    getAccessibleBatchIdSet(user, [task.batch_id]),
    getAccessibleSearchAccountIdSet(user, [task.account_id]),
  ]);

  return accessibleBatchIds.has(task.batch_id) && accessibleAccountIds.has(task.account_id);
}

async function requireVisibleSearchTask(user: ApiUserContext, taskId: string): Promise<SearchTaskRow> {
  const task = await getSearchTaskById(taskId);
  if (!task?.id) {
    throw new ApiAuthorizationError(404, `未找到搜索任务 ${taskId}`);
  }

  if (!(await canManageSearchTask(user, task))) {
    throw new ApiAuthorizationError(403, `当前用户无权访问搜索任务 ${taskId}`);
  }

  return task;
}

function parseSearchTaskStatuses(raw: unknown): SearchTaskStatus[] {
  const values = Array.isArray(raw)
    ? raw.flatMap((value) => String(value).split(","))
    : String(raw ?? "").split(",");
  const requested = values
    .map((value) => value.trim())
    .filter((value): value is SearchTaskStatus => SEARCH_ALL_STATUSES.includes(value as SearchTaskStatus));

  return requested.length > 0 ? [...new Set(requested)] : [...SEARCH_ALL_STATUSES];
}

function parseSearchTaskPageSize(raw: unknown, fallback: number) {
  const numeric = Number.parseInt(String(raw ?? ""), 10);
  if (!Number.isInteger(numeric) || numeric <= 0) {
    return fallback;
  }

  return Math.min(numeric, 200);
}

function parseSearchTaskOffset(raw: unknown) {
  const numeric = Number.parseInt(String(raw ?? ""), 10);
  if (!Number.isInteger(numeric) || numeric < 0) {
    return 0;
  }

  return numeric;
}

function isSearchTaskDeletable(task: SearchTaskRow) {
  return task.status === "paused"
    || task.status === "stopped"
    || task.status === "completed"
    || task.status === "failed";
}

async function buildSearchExecutorStatusForUser(user: ApiUserContext) {
  const blockingTask = await getLatestSearchTask(SEARCH_BLOCKING_STATUSES);
  if (!blockingTask) {
    return createIdleSearchExecutorStatus();
  }

  return createBusySearchExecutorStatus(blockingTask, await canManageSearchTask(user, blockingTask));
}

async function getPersistedSearchProgressForUser(user: ApiUserContext, taskId?: string): Promise<SearchProgress> {
  const task = taskId
    ? await requireVisibleSearchTask(user, taskId)
    : await getLatestVisibleSearchTask(user);
  return task ? mapSearchTaskRowToProgress(task) : createSearchProgress();
}

function pushSearchError(runState: SearchRunState, message: string) {
  const trimmed = message.trim();
  if (!trimmed) {
    return;
  }

  runState.progress.errors = [...new Set([...runState.progress.errors, trimmed])].slice(-20);
}

async function syncSearchTaskRow(runState: SearchRunState) {
  try {
    runState.task = await updateSearchTask(runState.progress.taskId, {
      batch_id: runState.progress.batchId,
      account_id: runState.progress.accountId,
      pipeline_config: normalizeSearchPipelineConfig(runState.task.pipeline_config),
      contact_limit: runState.task.contact_limit,
      headless: runState.task.headless,
      status: (runState.progress.status === "idle" ? "completed" : runState.progress.status) as SearchTaskStatus,
      total_rows: runState.progress.totalRows,
      processed_rows: runState.progress.processedRows,
      completed_rows: runState.progress.completedRows,
      failed_rows: runState.progress.failedRows,
      stopped_rows: runState.progress.stoppedRows,
      current_material_id: runState.progress.currentRow || null,
      current_item: runState.progress.currentItem,
      error_message: runState.progress.errors[0] ?? "",
      errors: runState.progress.errors,
      result: runState.progress.result ?? {},
      started_at: runState.progress.startedAt,
      paused_at: runState.progress.pausedAt,
      finished_at: runState.progress.finishedAt,
    });
  } catch (err) {
    console.error("[search] 同步 search_tasks 失败:", err);
    pushSearchError(runState, err instanceof Error ? err.message : String(err));
  }
}

async function patchSearchProgress(
  runState: SearchRunState,
  updates: Partial<SearchProgress>,
) {
  runState.progress = {
    ...runState.progress,
    ...updates,
    errors: updates.errors ? [...updates.errors] : runState.progress.errors,
    result: updates.result === undefined
      ? runState.progress.result
      : updates.result
        ? { ...updates.result }
        : null,
  };

  await syncSearchTaskRow(runState);
}

function clearSearchChildKillTimer(runState: SearchRunState) {
  if (runState.childKillTimer) {
    clearTimeout(runState.childKillTimer);
    runState.childKillTimer = null;
  }
}

function clearSearchChildState(runState: SearchRunState, child?: ChildProcess | null) {
  clearSearchChildKillTimer(runState);
  if (!child || runState.childProcess === child) {
    runState.childProcess = null;
    runState.childTerminationReason = null;
  }
}

function sendSignalToSearchChild(child: ChildProcess, signal: NodeJS.Signals) {
  if (process.platform !== "win32" && typeof child.pid === "number" && child.pid > 0) {
    process.kill(-child.pid, signal);
    return;
  }

  child.kill(signal);
}

function terminateSearchChildProcess(runState: SearchRunState, reason: SearchChildTerminationReason) {
  const child = runState.childProcess;
  if (!child || child.killed) {
    return false;
  }

  try {
    runState.childTerminationReason = reason;
    sendSignalToSearchChild(child, "SIGTERM");
  } catch (err) {
    runState.childTerminationReason = null;
    console.warn("[search] 发送 SIGTERM 失败:", err);
    return false;
  }

  clearSearchChildKillTimer(runState);
  runState.childKillTimer = setTimeout(() => {
    if (runState.childProcess === child && !child.killed) {
      try {
        sendSignalToSearchChild(child, "SIGKILL");
      } catch (err) {
        console.warn("[search] 发送 SIGKILL 失败:", err);
      }
    }
  }, SEARCH_CHILD_KILL_TIMEOUT_MS);

  return true;
}

type SearchAbortError = Error & { reason: SearchChildTerminationReason };

function createSearchAbortError(reason: SearchChildTerminationReason, message: string): SearchAbortError {
  const error = new Error(message) as SearchAbortError;
  error.name = "AbortError";
  error.reason = reason;
  return error;
}

function isSearchAbortError(error: unknown, reason?: SearchChildTerminationReason): error is SearchAbortError {
  if (!isAbortError(error)) {
    return false;
  }

  const candidateReason = (error as Partial<SearchAbortError>).reason;
  if (candidateReason !== "pause" && candidateReason !== "stop") {
    return false;
  }

  return reason ? candidateReason === reason : true;
}

function summarizeSearchFailureMessage(message: string) {
  const trimmed = message.trim();
  if (!trimmed) {
    return "搜索脚本执行失败";
  }

  const headline = trimmed
    .split(/\n\[(?:stdout|stderr)\]\n/i)[0]
    .split("\n")
    .map((line) => line.trim())
    .find(Boolean);

  return headline || "搜索脚本执行失败";
}

function parseSearchChildEventLine(line: string): SearchChildEvent | null {
  const trimmed = line.trim();
  if (!trimmed.startsWith(SEARCH_CHILD_EVENT_PREFIX)) {
    return null;
  }

  try {
    const parsed = JSON.parse(trimmed.slice(SEARCH_CHILD_EVENT_PREFIX.length)) as Partial<SearchChildEvent>;
    const type = typeof parsed.type === "string" ? parsed.type : "";
    if (
      type !== "manual_intervention_required"
      && type !== "manual_intervention_message"
      && type !== "manual_intervention_resolved"
      && type !== "progress_update"
    ) {
      return null;
    }

    return {
      type,
      alertId: typeof parsed.alertId === "string" ? parsed.alertId : "",
      kind: typeof parsed.kind === "string" ? parsed.kind : "captcha",
      message: typeof parsed.message === "string" ? parsed.message : "",
      materialId: typeof parsed.materialId === "string" ? parsed.materialId : "",
      materialName: typeof parsed.materialName === "string" ? parsed.materialName : "",
      accountId: typeof parsed.accountId === "string" ? parsed.accountId : "",
      pageUrl: typeof parsed.pageUrl === "string" ? parsed.pageUrl : "",
      taskId: typeof parsed.taskId === "string" ? parsed.taskId : "",
      detectedAt: typeof parsed.detectedAt === "string" ? parsed.detectedAt : "",
      resolvedAt: typeof parsed.resolvedAt === "string" ? parsed.resolvedAt : "",
      itemProgressPercent: typeof parsed.itemProgressPercent === "number" ? parsed.itemProgressPercent : undefined,
      softCapPercent: typeof parsed.softCapPercent === "number" ? parsed.softCapPercent : undefined,
      stageStartedAt: typeof parsed.stageStartedAt === "string" ? parsed.stageStartedAt : "",
    };
  } catch (error) {
    console.warn("[search] 解析子进程事件失败:", error);
    return null;
  }
}

function buildManualInterventionResult(
  runState: SearchRunState,
  event: SearchChildEvent,
  active: boolean,
) {
  const previous = runState.progress.result && typeof runState.progress.result === "object"
    ? runState.progress.result
    : {};
  const previousManual = previous.manualIntervention
    && typeof previous.manualIntervention === "object"
    && !Array.isArray(previous.manualIntervention)
      ? previous.manualIntervention as Record<string, unknown>
      : {};

  return {
    ...previous,
    manualIntervention: {
      ...previousManual,
      active,
      alertId: event.alertId || String(previousManual.alertId ?? ""),
      kind: event.kind || String(previousManual.kind ?? "captcha"),
      message: event.message || String(previousManual.message ?? (active ? "等待人工处理验证码" : "验证码已解除，继续执行")),
      taskId: runState.progress.taskId,
      batchId: runState.progress.batchId,
      headless: runState.task.headless,
      accountId: event.accountId || runState.progress.accountId || String(previousManual.accountId ?? ""),
      materialId: event.materialId || runState.progress.currentRow || String(previousManual.materialId ?? ""),
      materialName: event.materialName || String(previousManual.materialName ?? ""),
      pageUrl: event.pageUrl || String(previousManual.pageUrl ?? ""),
      detectedAt: event.detectedAt || String(previousManual.detectedAt ?? new Date().toISOString()),
      resolvedAt: active ? null : (event.resolvedAt || new Date().toISOString()),
      updatedAt: new Date().toISOString(),
    },
  };
}

function readTaskManualIntervention(task: SearchTaskRow) {
  const result = task.result && typeof task.result === "object" ? task.result : {};
  const manualIntervention = result.manualIntervention;
  if (!manualIntervention || typeof manualIntervention !== "object" || Array.isArray(manualIntervention)) {
    return null;
  }

  return manualIntervention as Record<string, unknown>;
}

async function applySearchChildEvent(runState: SearchRunState, event: SearchChildEvent) {
  if (event.taskId && event.taskId !== runState.progress.taskId) {
    return;
  }

  if (event.type === "progress_update") {
    const message = event.message || "搜索任务处理中";
    await patchSearchProgress(runState, {
      status: runState.progress.status === "waiting_for_manual" ? "waiting_for_manual" : "processing",
      currentItem: message,
      currentRow: event.materialId || runState.progress.currentRow,
      result: buildSearchProgressResult(runState, event),
    });
    return;
  }

  if (event.type === "manual_intervention_resolved") {
    await patchSearchProgress(runState, {
      status: runState.progress.status === "stopping" ? "stopping" : "processing",
      currentItem: event.message || "验证码已解除，继续执行自动搜索",
      result: buildManualInterventionResult(runState, event, false),
    });
    return;
  }

  await patchSearchProgress(runState, {
    status: "waiting_for_manual",
    currentItem: event.message || "检测到验证码，等待人工处理",
    result: buildManualInterventionResult(runState, event, true),
  });
}

async function runManagedSearchChild(
  runState: SearchRunState,
  params: {
    label: string;
    args: string[];
    cwd: string;
    env: NodeJS.ProcessEnv;
  },
): Promise<{ stdout: string; stderr: string }> {
  return new Promise((resolve, reject) => {
    let settled = false;
    let stdout = "";
    let stderr = "";
    let stdoutLineBuffer = "";
    let stderrLineBuffer = "";
    let eventChain = Promise.resolve();

    const appendOutput = (current: string, chunk: Buffer | string) => {
      const text = typeof chunk === "string" ? chunk : chunk.toString("utf8");
      if (!text) {
        return current;
      }

      const remainingBytes = SEARCH_CHILD_OUTPUT_LIMIT_BYTES - Buffer.byteLength(current, "utf8");
      if (remainingBytes <= 0) {
        return current;
      }

      const chunkBuffer = Buffer.from(text, "utf8");
      if (chunkBuffer.byteLength <= remainingBytes) {
        return current + text;
      }

      return current + chunkBuffer.subarray(0, remainingBytes).toString("utf8");
    };

    const queueSearchChildEvents = (text: string, channel: "stdout" | "stderr") => {
      const sourceBuffer = channel === "stdout" ? stdoutLineBuffer : stderrLineBuffer;
      const combined = sourceBuffer + text;
      const lines = combined.split(/\r?\n/);
      const remainder = lines.pop() ?? "";

      for (const line of lines) {
        const event = parseSearchChildEventLine(line);
        if (!event) {
          continue;
        }

        eventChain = eventChain
          .then(() => applySearchChildEvent(runState, event))
          .catch((error) => {
            console.warn("[search] 应用子进程事件失败:", error);
          });
      }

      if (channel === "stdout") {
        stdoutLineBuffer = remainder;
      } else {
        stderrLineBuffer = remainder;
      }
    };

    const flushSearchChildEventBuffer = (channel: "stdout" | "stderr") => {
      const line = (channel === "stdout" ? stdoutLineBuffer : stderrLineBuffer).trim();
      if (!line) {
        return;
      }

      const event = parseSearchChildEventLine(line);
      if (!event) {
        return;
      }

      eventChain = eventChain
        .then(() => applySearchChildEvent(runState, event))
        .catch((error) => {
          console.warn("[search] 应用子进程事件失败:", error);
        });

      if (channel === "stdout") {
        stdoutLineBuffer = "";
      } else {
        stderrLineBuffer = "";
      }
    };

    const child = spawn(process.execPath, params.args, {
      cwd: params.cwd,
      detached: process.platform !== "win32",
      env: params.env,
      stdio: ["ignore", "pipe", "pipe"],
    });

    runState.childProcess = child;
    clearSearchChildKillTimer(runState);

    child.stdout?.on("data", (chunk) => {
      stdout = appendOutput(stdout, chunk);
      queueSearchChildEvents(typeof chunk === "string" ? chunk : chunk.toString("utf8"), "stdout");
    });

    child.stderr?.on("data", (chunk) => {
      stderr = appendOutput(stderr, chunk);
      queueSearchChildEvents(typeof chunk === "string" ? chunk : chunk.toString("utf8"), "stderr");
    });

    child.once("error", (error) => {
      if (settled) {
        return;
      }

      settled = true;
      clearSearchChildState(runState, child);
      reject(error);
    });

    child.once("close", (code, signal) => {
      if (settled) {
        return;
      }

      settled = true;
      void (async () => {
        flushSearchChildEventBuffer("stdout");
        flushSearchChildEventBuffer("stderr");
        await eventChain;

        const terminationReason = runState.childTerminationReason;
        clearSearchChildState(runState, child);

        if (code === 0) {
          resolve({
            stdout: stdout.trim(),
            stderr: stderr.trim(),
          });
          return;
        }

        if (terminationReason) {
          reject(createSearchAbortError(
            terminationReason,
            terminationReason === "pause"
              ? "搜索任务已暂停，恢复后会重试当前未完成物料"
              : getSearchStopMessage(runState),
          ));
          return;
        }

        const lines = [`${params.label} 执行失败: exit ${code ?? "null"}${signal ? ` signal ${signal}` : ""}`];
        if (stdout.trim()) {
          lines.push(`[stdout]\n${stdout.trim()}`);
        }
        if (stderr.trim()) {
          lines.push(`[stderr]\n${stderr.trim()}`);
        }
        reject(new Error(lines.join("\n")));
      })();
    });
  });
}

function buildSearchChildEnv(params: {
  taskId: string;
  runtimeRootDir: string;
  account: Awaited<ReturnType<typeof getProcurementAccountById>> extends infer T
    ? T extends { id: string; account_name: string; password: string }
      ? T
      : never
    : never;
  accountProfileDir: string;
  outputRoot: string;
  contactLimit: number;
  headless: boolean;
  pipelineConfig: SearchPipelineConfig;
}) {
  return {
    ...process.env,
    SEARCH_TASK_ID: params.taskId,
    SEARCH_TASK_RUNTIME_ROOT_DIR: params.runtimeRootDir,
    SEARCH_1688_ACCOUNT_ID: params.account.id,
    SEARCH_1688_ACCOUNT_NAME: params.account.account_name,
    LOGIN_1688_USER: params.account.account_name,
    LOGIN_1688_PASS: params.account.password,
    SEARCH_1688_PROFILE_DIR: params.accountProfileDir,
    SEARCH_TASK_OUTPUT_ROOT: params.outputRoot,
    SEARCH_CONTACT_LIMIT: String(params.contactLimit),
    SEARCH_RESULT_LIMIT: String(params.contactLimit),
    SEARCH_HEADLESS: params.headless ? "1" : "0",
    SEARCH_PIPELINE_CONFIG: JSON.stringify(params.pipelineConfig),
  } satisfies NodeJS.ProcessEnv;
}

function summarizeSearchTaskItems(items: SearchTaskItemRow[]) {
  const summary = {
    total: items.length,
    processed: 0,
    completed: 0,
    failed: 0,
    stopped: 0,
  };

  for (const item of items) {
    if (item.status === "completed") {
      summary.completed += 1;
      summary.processed += 1;
      continue;
    }

    if (item.status === "failed") {
      summary.failed += 1;
      summary.processed += 1;
      continue;
    }

    if (item.status === "stopped") {
      summary.stopped += 1;
      summary.processed += 1;
    }
  }

  return summary;
}

function buildSearchResultSummary(
  runState: SearchRunState,
  overrides: {
    success?: boolean;
    stopped?: boolean;
    paused?: boolean;
    error?: string;
    processedRows?: number;
    completedRows?: number;
    failedRows?: number;
    stoppedRows?: number;
    errorCount?: number;
  } = {},
) {
  const processedRows = overrides.processedRows ?? runState.progress.processedRows;
  const completedRows = overrides.completedRows ?? runState.progress.completedRows;
  const failedRows = overrides.failedRows ?? runState.progress.failedRows;
  const stoppedRows = overrides.stoppedRows ?? runState.progress.stoppedRows;
  const summary: Record<string, unknown> = {
    success: overrides.success ?? (failedRows === 0 && stoppedRows === 0),
    taskId: runState.progress.taskId,
    batchId: runState.progress.batchId,
    processedRows,
    totalRows: runState.progress.totalRows,
    completedRows,
    failedRows,
    stoppedRows,
    errorCount: overrides.errorCount ?? runState.progress.errors.length,
  };

  if (overrides.stopped) {
    summary.stopped = true;
  }
  if (overrides.paused) {
    summary.paused = true;
  }
  if (overrides.error) {
    summary.error = overrides.error;
  }

  return summary;
}

function getSearchStopMessage(runState: SearchRunState) {
  return runState.stopMessage?.trim() || "用户手动停止搜索任务";
}

async function finalizeSearchRun(
  runState: SearchRunState,
  status: Extract<SearchTaskStatus, "completed" | "failed" | "paused" | "stopped">,
  options: {
    currentItem?: string;
    processedRows?: number;
    completedRows?: number;
    failedRows?: number;
    stoppedRows?: number;
    errors?: string[];
    result?: Record<string, unknown> | null;
  } = {},
) {
  const now = new Date().toISOString();
  const baseResult = options.result ?? buildSearchResultSummary(runState, {
    success: status === "completed",
    stopped: status === "stopped",
    paused: status === "paused",
    processedRows: options.processedRows,
    completedRows: options.completedRows,
    failedRows: options.failedRows,
    stoppedRows: options.stoppedRows,
  });
  const result = status === "paused"
    ? baseResult
    : await attachSearchRuntimeArtifacts(runState.progress.taskId, baseResult).catch((error) => {
      console.warn("[search] 汇总任务运行产物失败:", error instanceof Error ? error.message : error);
      return baseResult;
    });
  const shortage = result.searchShortage && typeof result.searchShortage === "object" && !Array.isArray(result.searchShortage)
    ? result.searchShortage as Record<string, unknown>
    : null;
  const shortageMessage = String(shortage?.message ?? "").trim();
  const defaultCurrentItem = status === "paused"
    ? "搜索已暂停，可从下一条未完成物料继续"
    : status === "stopped"
      ? "搜索已停止"
      : status === "completed"
        ? "搜索完成"
        : "搜索失败";
  const requestedCurrentItem = options.currentItem ?? defaultCurrentItem;
  const currentItem = status === "completed"
    && shortageMessage
    && (!options.currentItem || options.currentItem === "搜索完成")
      ? shortageMessage
      : requestedCurrentItem;

  await patchSearchProgress(runState, {
    status,
    currentRow: "",
    currentItem,
    processedRows: options.processedRows ?? runState.progress.processedRows,
    completedRows: options.completedRows ?? runState.progress.completedRows,
    failedRows: options.failedRows ?? runState.progress.failedRows,
    stoppedRows: options.stoppedRows ?? runState.progress.stoppedRows,
    errors: options.errors ? [...options.errors] : runState.progress.errors,
    result,
    pausedAt: status === "paused" ? now : null,
    finishedAt: status === "paused" ? null : now,
  });

  if (status !== "paused") {
    await cleanupSearchTaskRuntimeWorkspace(runState.progress.taskId).catch((error) => {
      console.warn("[search] 清理任务运行目录失败:", error instanceof Error ? error.message : error);
    });
    runState.stopMessage = null;
  }
}

async function finalizePausedSearchRun(runState: SearchRunState) {
  await finalizeSearchRun(runState, "paused", {
    currentItem: "搜索已暂停，可恢复后继续未完成物料",
    result: buildSearchResultSummary(runState, {
      success: false,
      paused: true,
    }),
  });
}

async function finalizeStoppedSearchRun(runState: SearchRunState) {
  const taskId = runState.progress.taskId;
  const message = getSearchStopMessage(runState);
  await markSearchTaskItemsStopped(taskId, message).catch(() => {
    // best effort
  });

  const items = await getSearchTaskItemsByTaskId(taskId).catch(() => []);
  const summary = summarizeSearchTaskItems(items);

  await finalizeSearchRun(runState, "stopped", {
    processedRows: summary.processed,
    completedRows: summary.completed,
    failedRows: summary.failed,
    stoppedRows: summary.stopped,
    currentItem: "搜索已停止",
    result: buildSearchResultSummary(runState, {
      success: false,
      stopped: true,
      error: message,
      processedRows: summary.processed,
      completedRows: summary.completed,
      failedRows: summary.failed,
      stoppedRows: summary.stopped,
    }),
  });
}

async function requestSearchPause(runState: SearchRunState) {
  if (runState.progress.status === "paused" || runState.progress.status === "pausing") {
    return;
  }

  const hasRunningChild = Boolean(runState.childProcess);
  await patchSearchProgress(runState, {
    status: "pausing",
    currentItem: hasRunningChild
      ? "正在暂停当前搜索脚本..."
      : runState.progress.currentRow
        ? "当前物料收口后将暂停"
        : "正在准备暂停搜索任务...",
    result: null,
  });

  if (hasRunningChild) {
    terminateSearchChildProcess(runState, "pause");
  }
}

async function requestSearchStop(runState: SearchRunState, message: string) {
  runState.stopMessage = message;

  if (runState.progress.status !== "stopping") {
    await patchSearchProgress(runState, {
      status: "stopping",
      currentItem: "正在停止当前搜索任务...",
      pausedAt: null,
      result: null,
    });
  }

  const signaled = terminateSearchChildProcess(runState, "stop");
  if (!signaled) {
    await finalizeStoppedSearchRun(runState);
  }
}

async function recoverOrphanSearchTasks() {
  const tasks = await getSearchTasksByStatuses(SEARCH_RECOVERABLE_STATUSES).catch(() => []);
  if (tasks.length === 0) {
    return;
  }

  const message = "自动化服务重启，已回收未完成搜索任务";
  for (const task of tasks) {
    const taskId = task.id ?? "";
    if (!taskId) {
      continue;
    }

    await markSearchTaskItemsStopped(taskId, message).catch(() => {
      // best effort
    });

    const items = await getSearchTaskItemsByTaskId(taskId).catch(() => []);
    const summary = summarizeSearchTaskItems(items);
    const updated = await updateSearchTask(taskId, {
      status: "stopped",
      processed_rows: summary.processed,
      completed_rows: summary.completed,
      failed_rows: summary.failed,
      stopped_rows: summary.stopped,
      current_material_id: null,
      current_item: "搜索任务因服务重启已停止",
      error_message: message,
      errors: [message],
      result: {
        success: false,
        stopped: true,
        taskId,
        batchId: task.batch_id,
        processedRows: summary.processed,
        totalRows: task.total_rows,
        completedRows: summary.completed,
        failedRows: summary.failed,
        stoppedRows: summary.stopped,
        error: message,
      },
      paused_at: null,
      finished_at: new Date().toISOString(),
    }).catch((err) => {
      console.error(`[search] 回收僵尸任务失败: ${taskId}`, err);
      return null;
    });

    if (updated) {
      const runState = getSearchRunState(updated);
      runState.progress = mapSearchTaskRowToProgress(updated);
      runState.stopMessage = null;
    }

    await cleanupSearchTaskRuntimeWorkspace(taskId).catch(() => {
      // best effort
    });
  }

  console.log(`[search] 已回收 ${tasks.length} 个未完成搜索任务`);
}

async function runSearchTask(runState: SearchRunState) {
  const taskId = runState.progress.taskId;
  if (!taskId || runState.childProcess) {
    return;
  }

  try {
    const task = await getSearchTaskById(taskId);
    if (!task) {
      throw new Error(`未找到搜索任务 ${taskId}`);
    }
    runState.task = task;

    if (task.status === "stopped" || task.status === "completed" || task.status === "failed") {
      runState.progress = mapSearchTaskRowToProgress(task);
      return;
    }

    const account = await getProcurementAccountById(task.account_id);
    if (!account) {
      throw new Error(`未找到 id=${task.account_id} 的 1688 账户`);
    }
    if (!account.account_name || !account.password) {
      throw new Error(`1688 账户 ${task.account_id} 缺少账号名或密码`);
    }

    const taskItems = await getSearchTaskItemsByTaskId(taskId);
    const pendingItems = taskItems.filter((item) => item.status === "pending");
    const pipelineConfig = normalizeSearchPipelineConfig(task.pipeline_config);
    const accountProfileDir = path.join(__dirname, "..", ".browser", "1688-profiles", account.id);
    if (isSearchAccountExecutionLocked(account.id)) {
      await patchSearchProgress(runState, {
        status: "processing",
        currentRow: "",
        currentItem: `等待 1688 账户空闲: ${account.account_name}`,
        startedAt: runState.progress.startedAt ?? task.started_at ?? new Date().toISOString(),
        pausedAt: null,
        finishedAt: null,
        result: null,
      });
    }

    await withSearchAccountExecutionLock(account.id, async () => {
      if (runState.progress.status === "stopping") {
        await finalizeStoppedSearchRun(runState);
        return;
      }
      if (runState.progress.status === "pausing") {
        await finalizePausedSearchRun(runState);
        return;
      }

      const runtimePaths = await prepareSearchTaskRuntimeProfile({
        taskId,
        accountProfileDir,
        runtimeRootDir: SEARCH_TASK_RUNTIME_ROOT,
      });
      const childEnv = buildSearchChildEnv({
        taskId,
        runtimeRootDir: runtimePaths.rootDir,
        account,
        accountProfileDir: runtimePaths.profileDir,
        outputRoot: runtimePaths.outputRoot,
        contactLimit: task.contact_limit,
        headless: task.headless,
        pipelineConfig,
      });
      const automationDir = path.join(__dirname, "..");

    await patchSearchProgress(runState, {
      status: "processing",
      currentRow: "",
      currentItem: `正在检查登录态并准备搜索: ${account.account_name}`,
      startedAt: runState.progress.startedAt ?? task.started_at ?? new Date().toISOString(),
      pausedAt: null,
      finishedAt: null,
      result: null,
    });

    console.log(`[search] 1688 登录态将由搜索子流程接管并与首轮搜索复用同一页面: ${account.account_name}`);

    if (runState.progress.status === "stopping") {
      await finalizeStoppedSearchRun(runState);
      return;
    }
    if (runState.progress.status === "pausing") {
      await finalizePausedSearchRun(runState);
      return;
    }

    if (pendingItems.length === 0) {
      const hasFailures = runState.progress.failedRows > 0;
      await finalizeSearchRun(runState, hasFailures ? "failed" : "completed", {
        currentItem: hasFailures ? "搜索结束（有错误）" : "搜索完成",
        result: buildSearchResultSummary(runState, {
          success: !hasFailures,
        }),
      });
      return;
    }

    for (const item of pendingItems) {
      const itemId = item.id ?? "";
      if (!itemId) {
        throw new Error(`搜索任务明细缺少 id: material_id=${item.material_id}`);
      }

      if (runState.progress.status === "stopping") {
        await finalizeStoppedSearchRun(runState);
        return;
      }
      if (runState.progress.status === "pausing") {
        await finalizePausedSearchRun(runState);
        return;
      }

      await updateSearchTaskItem(itemId, {
        status: "processing",
        error_message: "",
        started_at: new Date().toISOString(),
        finished_at: null,
      });

      await patchSearchProgress(runState, {
        status: "processing",
        currentRow: item.material_id,
        currentItem: `正在搜索: ${item.material_name} (${item.material_id})`,
        result: null,
      });

      console.log(`[search] 开始处理 material_id=${item.material_id} name=${item.material_name}`);

      try {
        const childResult = await runManagedSearchChild(runState, {
          label: `物料 ${item.material_id} 搜索`,
          args: [
            "--import", "tsx",
            ROW_PIPELINE_SYNC_SCRIPT,
            `--material-id=${item.material_id}`,
            `--batch-id=${runState.progress.batchId}`,
            `--headless=${runState.task.headless ? 1 : 0}`,
            `--contact-limit=${runState.task.contact_limit}`,
          ],
          cwd: automationDir,
          env: {
            ...childEnv,
            SEARCH_TASK_CURRENT_MATERIAL_ID: item.material_id,
            SEARCH_TASK_CURRENT_MATERIAL_NAME: item.material_name,
          },
        });

        if (childResult.stdout) {
          console.log(childResult.stdout);
        }
        if (childResult.stderr) {
          console.error(childResult.stderr);
        }

        await syncSearchTaskRuntimeProfileToAccount({
          taskId,
          accountProfileDir,
          runtimeRootDir: SEARCH_TASK_RUNTIME_ROOT,
        });

        await updateSearchTaskItem(itemId, {
          status: "completed",
          error_message: "",
          finished_at: new Date().toISOString(),
        });

        await patchSearchProgress(runState, {
          processedRows: runState.progress.processedRows + 1,
          completedRows: runState.progress.completedRows + 1,
        });

        console.log(`[search] material_id=${item.material_id} 完成 (${runState.progress.processedRows}/${runState.progress.totalRows})`);
      } catch (err) {
        if (isSearchAbortError(err, "pause")) {
          await updateSearchTaskItem(itemId, {
            status: "pending",
            error_message: "",
            started_at: null,
            finished_at: null,
          }).catch(() => {
            // best effort
          });

          await patchSearchProgress(runState, {
            currentRow: "",
            currentItem: `搜索已暂停，待恢复后重试: ${item.material_name} (${item.material_id})`,
            result: null,
          });
          await finalizePausedSearchRun(runState);
          return;
        }

        if (runState.progress.status === "stopping") {
          const stopMessage = getSearchStopMessage(runState);
          await updateSearchTaskItem(itemId, {
            status: "stopped",
            error_message: stopMessage,
            finished_at: new Date().toISOString(),
          }).catch(() => {
            // best effort
          });

          await patchSearchProgress(runState, {
            processedRows: runState.progress.processedRows + 1,
            stoppedRows: runState.progress.stoppedRows + 1,
          });
          await finalizeStoppedSearchRun(runState);
          return;
        }

        const rawMessage = err instanceof Error ? err.message : String(err);
        const message = summarizeSearchFailureMessage(rawMessage);
        const taskMessage = `物料 ${item.material_id} (${item.material_name}) 失败: ${message}`;
        pushSearchError(runState, taskMessage);
        await updateSearchTaskItem(itemId, {
          status: "failed",
          error_message: message,
          finished_at: new Date().toISOString(),
        }).catch(() => {
          // best effort
        });

        await patchSearchProgress(runState, {
          processedRows: runState.progress.processedRows + 1,
          failedRows: runState.progress.failedRows + 1,
          errors: runState.progress.errors,
        });

        console.error(`[search] ${taskMessage}`, err);
      }

      if (runState.progress.status === "stopping") {
        await finalizeStoppedSearchRun(runState);
        return;
      }
      if (runState.progress.status === "pausing") {
        await finalizePausedSearchRun(runState);
        return;
      }
    }

    const hasFailures = runState.progress.failedRows > 0;
    await finalizeSearchRun(runState, hasFailures ? "failed" : "completed", {
      currentItem: hasFailures ? "搜索结束（有错误）" : "搜索完成",
      result: buildSearchResultSummary(runState, {
        success: !hasFailures,
      }),
    });

    console.log(`[search] 任务 ${taskId} 搜索结束: ${runState.progress.processedRows}/${runState.progress.totalRows}, 错误=${runState.progress.errors.length}`);
    });
  } catch (err) {
    if (isSearchAbortError(err, "pause")) {
      await finalizePausedSearchRun(runState);
      return;
    }

    if (runState.progress.status === "stopping") {
      await finalizeStoppedSearchRun(runState);
      return;
    }

    const rawMessage = err instanceof Error ? err.message : String(err);
    const message = summarizeSearchFailureMessage(rawMessage);
    pushSearchError(runState, message);
    await finalizeSearchRun(runState, "failed", {
      currentItem: "搜索失败",
      errors: runState.progress.errors,
      result: buildSearchResultSummary(runState, {
        success: false,
        error: message,
      }),
    });
    console.error(`[search] 任务 ${taskId} 异常终止: ${message}`, err);
  } finally {
    clearSearchChildState(runState);
  }
}

const SearchTaskIdBody = z.object({
  task_id: z.string().optional(),
});

function readSearchTaskId(raw: unknown) {
  const taskId = String(raw ?? "").trim();
  if (!taskId) {
    return "";
  }
  if (!looksLikeUuid(taskId) && !looksLikeSearchTaskCode(taskId)) {
    throw new Error("task_id 格式无效");
  }
  return taskId;
}

function isInvalidSearchTaskIdError(error: unknown) {
  return error instanceof Error && error.message === "task_id 格式无效";
}

// ── POST /api/search-batch — 从标准化结果启动 1688 自动化搜索 ──
const SearchBatchBody = z.object({
  batch_id: z.string().trim().min(1, "缺少 batch_id 参数"),
  account_id: z.string().trim().min(1, "缺少 account_id 参数"),
  material_ids: z.array(z.string()).optional(),
  contact_limit: z
    .number()
    .int("contact_limit 必须是大于等于 0 的整数")
    .min(0, "contact_limit 必须是大于等于 0 的整数")
    .optional(),
  headless: z.boolean().optional(),
});

app.post("/api/search-batch", requireApiUser, async (req: AuthenticatedRequest, res) => {
  const createdTasks: SearchTaskRow[] = [];

  try {
    const apiUser = getRequestApiUser(req);
    const { batch_id, material_ids, account_id, contact_limit, headless } =
      parseBody(req, SearchBatchBody);

    await assertBatchAccess(apiUser, batch_id);
    const account = await getAuthorizedProcurementAccount(apiUser, account_id);

    let query = supabase
      .from("standardized_results")
      .select("material_id, standardized_name, original_name")
      .eq("batch_id", batch_id);
    if (material_ids && material_ids.length > 0) {
      query = query.in("material_id", material_ids);
    }
    const { data: rows, error: queryError } = await query.order("created_at");
    if (queryError) throw queryError;
    if (!rows || rows.length === 0) {
      res.status(404).json({ error: "未找到符合条件的标准化数据行" });
      return;
    }

    if (!account.account_name || !account.password) {
      res.status(400).json({ error: `1688 账户 ${account_id} 缺少账号名或密码` });
      return;
    }

    const rowList = (rows ?? [])
      .map((row) => ({
        material_id: String((row as { material_id?: string }).material_id ?? ""),
        name: String((row as { standardized_name?: string; original_name?: string }).standardized_name ?? (row as { original_name?: string }).original_name ?? ""),
      }))
      .filter((row) => looksLikeUuid(row.material_id));

    if (rowList.length === 0) {
      res.status(404).json({ error: "未找到可执行搜索的有效物料" });
      return;
    }

    const pipelineConfig = normalizeSearchPipelineConfig(DEFAULT_SEARCH_PIPELINE_CONFIG);
    const searchHeadless = headless === true;
    for (const row of rowList) {
      const createdTask = await createSearchTask({
        batch_id,
        account_id,
        pipeline_config: pipelineConfig,
        contact_limit: contact_limit ?? 0,
        headless: searchHeadless,
        status: "pending",
        total_rows: 1,
        processed_rows: 0,
        completed_rows: 0,
        failed_rows: 0,
        stopped_rows: 0,
        current_material_id: row.material_id,
        current_item: "正在准备...",
        error_message: "",
        errors: [],
        result: {},
        started_at: null,
        paused_at: null,
        finished_at: null,
      });

      const taskId = createdTask.id ?? "";
      if (!taskId) {
        throw new Error("创建搜索任务后未返回 task id");
      }

      await insertSearchTaskItems([{
        task_id: taskId,
        batch_id,
        material_id: row.material_id,
        sort_order: 1,
        material_name: row.name,
        status: "pending",
        item_snapshot: {
          material_id: row.material_id,
          name: row.name,
        },
        error_message: "",
        started_at: null,
        finished_at: null,
      }]);

      createdTasks.push(createdTask);
    }

    const taskIds = createdTasks
      .map((task) => task.id ?? "")
      .filter(Boolean);

    res.json({
      message: "搜索任务已启动",
      totalRows: rowList.length,
      createdCount: createdTasks.length,
      taskId: taskIds[0] ?? "",
      taskIds,
    });

    createdTasks.forEach((task) => {
      const runState = getSearchRunState(task);
      void runSearchTask(runState);
    });
  } catch (err) {
    if (createdTasks.length > 0) {
      const message = err instanceof Error ? err.message : String(err);
      await Promise.allSettled(createdTasks.map((task) => (
        task.id
          ? updateSearchTask(task.id, {
            status: "failed",
            current_item: "搜索任务创建失败",
            error_message: message,
            errors: [message],
            result: {
              success: false,
              taskId: task.id,
              batchId: task.batch_id,
              processedRows: 0,
              totalRows: task.total_rows,
              completedRows: 0,
              failedRows: 0,
              stoppedRows: 0,
              errorCount: 1,
              error: message,
            },
            finished_at: new Date().toISOString(),
          })
          : Promise.resolve(null)
      )));
    }

    const msg = err instanceof Error ? err.message : String(err);
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    res.status(500).json({ error: msg });
  }
});

// ── GET /api/search-manual-interventions — 人工介入提醒列表 ────
app.get("/api/search-manual-interventions", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const tasks = await getSearchTasksByStatuses(["waiting_for_manual"], {
      orderBy: "updated_at",
      ascending: false,
    });

    if (tasks.length === 0) {
      res.json({ alerts: [] });
      return;
    }

    const [accessibleBatchIds, accessibleAccountIds] = await Promise.all([
      getAccessibleBatchIdSet(apiUser, tasks.map((task) => task.batch_id)),
      getAccessibleSearchAccountIdSet(apiUser, tasks.map((task) => task.account_id)),
    ]);

    const visibleTasks = tasks.filter((task) => (
      accessibleBatchIds.has(task.batch_id)
      && accessibleAccountIds.has(task.account_id)
    ));

    const accountNameById = new Map(
      (await getProcurementAccountsByIds(visibleTasks.map((task) => task.account_id)))
        .map((account) => [account.id, account.account_name]),
    );

    const alerts = visibleTasks
      .map((task) => {
        const manualIntervention = readTaskManualIntervention(task);
        if (!manualIntervention) {
          return null;
        }

        return {
          alertId: String(manualIntervention.alertId ?? task.id ?? ""),
          taskId: task.id ?? "",
          batchId: task.batch_id,
          accountId: task.account_id,
          accountName: accountNameById.get(task.account_id) ?? "",
          status: task.status,
          headless: task.headless,
          message: String(manualIntervention.message ?? task.current_item ?? "检测到验证码，等待人工处理"),
          kind: String(manualIntervention.kind ?? "captcha"),
          materialId: String(manualIntervention.materialId ?? task.current_material_id ?? ""),
          materialName: String(manualIntervention.materialName ?? task.current_item ?? ""),
          pageUrl: String(manualIntervention.pageUrl ?? ""),
          detectedAt: String(manualIntervention.detectedAt ?? task.updated_at ?? task.created_at ?? ""),
          resolvedAt: manualIntervention.resolvedAt == null ? null : String(manualIntervention.resolvedAt),
          updatedAt: String(manualIntervention.updatedAt ?? task.updated_at ?? task.created_at ?? ""),
          active: manualIntervention.active !== false,
        };
      })
      .filter((item): item is NonNullable<typeof item> => Boolean(item));

    res.json({ alerts });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── POST /api/search-manual-interventions/open — 打开验证码会话 ─
app.post("/api/search-manual-interventions/open", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(parseBody(req, SearchTaskIdBody).task_id);
    const task = await requireVisibleSearchTask(apiUser, taskId);

    if (task.status !== "waiting_for_manual") {
      res.status(400).json({ error: "当前搜索任务不在等待人工处理状态" });
      return;
    }

    const runState = getSearchRunState(task);
    if (!runState.childProcess) {
      res.status(409).json({ error: "当前验证码等待会话已不在线，无法打开人工验证窗口" });
      return;
    }

    const manualIntervention = readTaskManualIntervention(task);
    const pageUrl = String(manualIntervention?.pageUrl ?? "");
    const requestId = `open-${Date.now()}`;
    const commandPath = getSearchManualInterventionCommandPath(taskId);
    await mkdir(path.dirname(commandPath), { recursive: true });
    writeFileSync(commandPath, JSON.stringify({
      type: "open_manual_browser",
      requestId,
      taskId,
      pageUrl,
      createdAt: new Date().toISOString(),
    }, null, 2), "utf8");

    res.json({
      message: task.headless
        ? "已请求打开验证码可视窗口，请在弹出的 1688 页面中完成人工验证"
        : "已请求切换到当前验证码页面，请在自动化浏览器窗口中完成人工验证",
      taskId,
      requestId,
      pageUrl,
      headless: task.headless,
    });
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── GET /api/search-executor-status — 查询搜索执行器占用状态 ─────
app.get("/api/search-executor-status", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    res.json(await buildSearchExecutorStatusForUser(apiUser));
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── GET /api/search-tasks — 搜索任务列表 ─────────────────────
app.get("/api/search-tasks", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const batchId = String(req.query.batch_id ?? "").trim();
    const accountId = String(req.query.account_id ?? "").trim();
    const search = String(req.query.search ?? "").trim().toLowerCase();
    const statuses = parseSearchTaskStatuses(req.query.statuses);
    const limit = parseSearchTaskPageSize(req.query.limit, 100);
    const offset = parseSearchTaskOffset(req.query.offset);

    const tasks = await getSearchTasksByStatuses(statuses, {
      orderBy: "updated_at",
      ascending: false,
    });

    if (tasks.length === 0) {
      res.json({
        tasks: [],
        total: 0,
        summary: {
          total: 0,
          pending: 0,
          processing: 0,
          waiting_for_manual: 0,
          pausing: 0,
          paused: 0,
          stopping: 0,
          stopped: 0,
          completed: 0,
          failed: 0,
        },
      });
      return;
    }

    const [accessibleBatchIds, accessibleAccountIds] = await Promise.all([
      getAccessibleBatchIdSet(apiUser, tasks.map((task) => task.batch_id)),
      getAccessibleSearchAccountIdSet(apiUser, tasks.map((task) => task.account_id)),
    ]);

    const visibleTasks = tasks.filter((task) => (
      accessibleBatchIds.has(task.batch_id)
      && accessibleAccountIds.has(task.account_id)
    ));

    const taskIds = visibleTasks
      .map((task) => task.id ?? "")
      .filter(Boolean);
    const [taskItems, accounts] = await Promise.all([
      getSearchTaskItemsByTaskIds(taskIds),
      getProcurementAccountsByIds(visibleTasks.map((task) => task.account_id)),
    ]);

    const firstTaskItemByTaskId = new Map<string, SearchTaskItemRow>();
    for (const item of taskItems) {
      if (!firstTaskItemByTaskId.has(item.task_id)) {
        firstTaskItemByTaskId.set(item.task_id, item);
      }
    }

    const accountNameById = new Map(accounts.map((account) => [account.id, account.account_name]));

    const filteredTasks = visibleTasks
      .filter((task) => !batchId || task.batch_id === batchId)
      .filter((task) => !accountId || task.account_id === accountId)
      .map((task) => {
        const taskId = task.id ?? "";
        const taskItem = firstTaskItemByTaskId.get(taskId);
        const accountName = accountNameById.get(task.account_id) ?? "";

        return {
          id: taskId,
          batch_id: task.batch_id,
          account_id: task.account_id,
          account_name: accountName,
          material_id: taskItem?.material_id ?? task.current_material_id ?? "",
          material_name: taskItem?.material_name ?? task.current_item,
          status: task.status,
          total_rows: task.total_rows,
          processed_rows: task.processed_rows,
          completed_rows: task.completed_rows,
          failed_rows: task.failed_rows,
          stopped_rows: task.stopped_rows,
          current_material_id: task.current_material_id ?? "",
          current_item: task.current_item,
          current_step_label: readSearchProgressSnapshot(task.result)?.message || task.current_item,
          display_progress_percent: computeSearchDisplayProgress({
            status: task.status,
            totalRows: task.total_rows,
            processedRows: task.processed_rows,
            result: task.result,
          }),
          search_progress: readSearchProgressSnapshot(task.result),
          error_message: task.error_message,
          errors: task.errors,
          started_at: task.started_at ?? null,
          paused_at: task.paused_at ?? null,
          finished_at: task.finished_at ?? null,
          created_at: task.created_at ?? null,
          updated_at: task.updated_at ?? null,
          deletable: isSearchTaskDeletable(task),
        };
      })
      .filter((task) => {
        if (!search) {
          return true;
        }

        return [
          task.id,
          task.batch_id,
          task.account_name,
          task.account_id,
          task.material_id,
          task.material_name,
          task.current_item,
          task.status,
        ].some((value) => String(value ?? "").toLowerCase().includes(search));
      });

    const summary = filteredTasks.reduce((accumulator, task) => {
      accumulator.total += 1;
      accumulator[task.status] += 1;
      return accumulator;
    }, {
      total: 0,
      pending: 0,
      processing: 0,
      waiting_for_manual: 0,
      pausing: 0,
      paused: 0,
      stopping: 0,
      stopped: 0,
      completed: 0,
      failed: 0,
    });

    res.json({
      tasks: filteredTasks.slice(offset, offset + limit),
      total: filteredTasks.length,
      summary,
    });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── DELETE /api/search-tasks/:taskId — 删除搜索任务记录 ───────
app.delete("/api/search-tasks/:taskId", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(req.params.taskId);
    const task = await requireVisibleSearchTask(apiUser, taskId);

    if (!isSearchTaskDeletable(task)) {
      res.status(400).json({ error: "仅支持删除已暂停或已结束的搜索任务记录" });
      return;
    }

    await deleteSearchTaskItemsByTaskId(taskId);
    await deleteSearchTaskById(taskId);
    searchRuns.delete(taskId);
    await cleanupSearchTaskRuntimeWorkspace(taskId).catch(() => {
      // best effort
    });

    res.json({ message: "搜索任务记录已删除", taskId });
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── GET /api/search-status — 轮询搜索进度 ─────────────────────
app.get("/api/search-status", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(req.query.task_id);
    const progress = await getPersistedSearchProgressForUser(apiUser, taskId || undefined);
    res.json(progress);
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/search-pause — 安全暂停搜索任务 ─────────────────
app.post("/api/search-pause", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(parseBody(req, SearchTaskIdBody).task_id);
    const targetTask = taskId
      ? await requireVisibleSearchTask(apiUser, taskId)
      : await getLatestVisibleSearchTask(apiUser, ["processing", "waiting_for_manual", "pausing"]);

    const runState = targetTask?.id ? getSearchRunState(targetTask) : null;

    if (!runState) {
      res.status(400).json({ error: "当前没有可暂停的搜索任务" });
      return;
    }

    if (runState.progress.status === "paused") {
      res.json({ message: "搜索任务已处于暂停状态", taskId: runState.progress.taskId });
      return;
    }

    if (
      runState.progress.status !== "processing"
      && runState.progress.status !== "waiting_for_manual"
      && runState.progress.status !== "pausing"
    ) {
      res.status(400).json({ error: "当前搜索任务不支持暂停" });
      return;
    }

    await requestSearchPause(runState);
    if (!runState.childProcess && !runState.progress.currentRow) {
      await finalizePausedSearchRun(runState);
      res.json({ message: "搜索任务已暂停", taskId: runState.progress.taskId });
      return;
    }

    res.json({
      message: runState.childProcess
        ? "暂停信号已发送，正在终止当前搜索脚本"
        : "暂停信号已发送，等待当前步骤收口",
      taskId: runState.progress.taskId,
    });
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/search-resume — 恢复暂停的搜索任务 ──────────────
app.post("/api/search-resume", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(parseBody(req, SearchTaskIdBody).task_id);
    const targetTask = taskId
      ? await requireVisibleSearchTask(apiUser, taskId)
      : await getLatestVisibleSearchTask(apiUser, ["paused"]);
    if (!targetTask?.id) {
      res.status(400).json({ error: "当前没有可恢复的暂停搜索任务" });
      return;
    }
    if (targetTask.status !== "paused") {
      res.status(400).json({ error: "只有 paused 状态的搜索任务才能恢复" });
      return;
    }

    const runState = getSearchRunState(targetTask);
    runState.stopMessage = null;
    await patchSearchProgress(runState, {
      status: "processing",
      currentItem: "正在恢复搜索任务...",
      pausedAt: null,
      finishedAt: null,
      result: null,
    });

    res.json({ message: "搜索任务恢复中", taskId: runState.progress.taskId });
    void runSearchTask(runState);
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/search-stop — 停止搜索任务 ──────────────────────
app.post("/api/search-stop", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const taskId = readSearchTaskId(parseBody(req, SearchTaskIdBody).task_id);
    const targetTask = taskId
      ? await requireVisibleSearchTask(apiUser, taskId)
      : await getLatestVisibleSearchTask(apiUser, ["processing", "pausing", "paused", "stopping"]);
    if (!targetTask?.id) {
      res.status(400).json({ error: "当前没有正在运行或暂停中的搜索任务" });
      return;
    }

    const runState = getSearchRunState(targetTask);

    if (runState.progress.status === "stopping") {
      res.json({ message: "搜索任务正在停止，请稍候刷新状态", taskId: runState.progress.taskId });
      return;
    }

    if (runState.progress.status === "completed" || runState.progress.status === "failed" || runState.progress.status === "stopped") {
      res.status(400).json({ error: "当前搜索任务已经结束" });
      return;
    }

    await requestSearchStop(runState, "用户手动停止搜索任务");
    res.json({
      message: runState.progress.status === "stopped"
        ? "搜索任务已停止"
        : "停止信号已发送，正在终止当前搜索脚本",
      taskId: runState.progress.taskId,
    });
  } catch (err) {
    if (isApiAuthorizationError(err)) {
      sendApiError(res, err);
      return;
    }
    if (isInvalidSearchTaskIdError(err)) {
      res.status(400).json({ error: err instanceof Error ? err.message : String(err) });
      return;
    }
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── GET /api/search-presets — 获取用户搜索预设 ────────────────
app.get("/api/search-presets", async (req, res) => {
  try {
    const userId = String(req.query.user_id ?? "");
    if (!userId) {
      res.status(400).json({ error: "缺少 user_id 参数" });
      return;
    }
    const { data, error } = await supabase
      .from("search_presets")
      .select("*")
      .eq("user_id", userId)
      .order("created_at", { ascending: false });
    if (error) throw error;
    res.json((data ?? []).map((preset) => ({
      ...preset,
      config: normalizeSearchPipelineConfig((preset as { config?: unknown }).config),
    })));
  } catch (err) {
    res.status(500).json({ error: err instanceof Error ? err.message : String(err) });
  }
});

// ── POST /api/search-presets — 创建搜索预设 ───────────────────
const SearchPresetCreateBody = z.object({
  user_id: z.string().trim().min(1, "缺少必要参数 user_id"),
  name: z.string().trim().min(1, "缺少必要参数 name"),
  config: z.record(z.unknown()),
  is_default: z.boolean().optional(),
});

app.post("/api/search-presets", async (req, res) => {
  try {
    const { user_id, name, config, is_default } = parseBody(req, SearchPresetCreateBody);
    // 如果设为默认，先取消当前默认
    if (is_default) {
      await supabase.from("search_presets").update({ is_default: false }).eq("user_id", user_id).eq("is_default", true);
    }
    const normalizedConfig = normalizeSearchPipelineConfig(config);
    const { data, error } = await supabase
      .from("search_presets")
      .insert({ user_id, name, config: normalizedConfig, is_default: is_default ?? false })
      .select()
      .single();
    if (error) throw error;
    res.json({
      ...data,
      config: normalizeSearchPipelineConfig((data as { config?: unknown }).config),
    });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── PUT /api/search-presets/:id — 更新搜索预设 ────────────────
const SearchPresetUpdateBody = z.object({
  name: z.string().optional(),
  config: z.record(z.unknown()).optional(),
  is_default: z.boolean().optional(),
});

app.put("/api/search-presets/:id", async (req, res) => {
  try {
    const { id } = req.params;
    const { name, config, is_default } = parseBody(req, SearchPresetUpdateBody);
    const updates: Record<string, unknown> = { updated_at: new Date().toISOString() };
    if (name !== undefined) updates.name = name;
    if (config !== undefined) updates.config = normalizeSearchPipelineConfig(config);
    if (is_default !== undefined) {
      updates.is_default = is_default;
      // 如果设为默认，先取消该用户的其他默认
      if (is_default) {
        const { data: existing } = await supabase.from("search_presets").select("user_id").eq("id", id).single();
        if (existing) {
          await supabase.from("search_presets").update({ is_default: false }).eq("user_id", existing.user_id).eq("is_default", true);
        }
      }
    }
    const { data, error } = await supabase
      .from("search_presets")
      .update(updates)
      .eq("id", id)
      .select()
      .single();
    if (error) throw error;
    res.json({
      ...data,
      config: normalizeSearchPipelineConfig((data as { config?: unknown }).config),
    });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── DELETE /api/search-presets/:id — 删除搜索预设 ─────────────
app.delete("/api/search-presets/:id", async (req, res) => {
  try {
    const { id } = req.params;
    const { error } = await supabase.from("search_presets").delete().eq("id", id);
    if (error) throw error;
    res.json({ message: "预设已删除" });
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── 智能询价状态 ─────────────────────────────────────────────
type InquiryProgressStatus = "idle" | "processing" | "completed" | "failed" | "stopping" | "stopped";

interface InquiryProgress {
  status: InquiryProgressStatus;
  batchId: string;
  accountId: string;
  totalTasks: number;
  completedTasks: number;
  rejectedTasks: number;
  humanTakeoverTasks: number;
  currentItem: string;
  errors: string[];
  result: Record<string, unknown> | null;
  startedAt: string | null;
  finishedAt: string | null;
}

interface InquiryTaskSummary {
  total: number;
  completed: number;
  rejected: number;
  humanTakeover: number;
  active: number;
}

interface InquiryBaseRow {
  material_id: string;
  original_name: string;
  standardized_name: string;
  original_spec: string;
  standardized_spec: string;
  brand: string;
  quantity: string;
  standardized_unit: string;
  category: string;
  notes: string;
  delivery_address: string;
  primary_search_term: string;
  alt_search_terms: string;
  invoice_type: string;
  product_type: string;
  source_type: string;
}

interface InquiryFinalCandidateRow {
  id: string;
  batch_id: string;
  material_id: string;
  title: string;
  offer_url: string;
  merchant_name: string;
  merchant_url: string;
  wangwang_url: string;
  contact_page_url: string;
}

interface InquiryCandidate {
  search_result_id: string;
  batch_id: string;
  material_id: string | null;
  merchant_name: string;
  merchant_url: string;
  contact_url: string;
  contact_handle: string;
  contact_site: string;
  offer_url: string;
  title: string;
  ask_context: Record<string, string>;
}

interface ExistingInquiryTaskRow {
  id: string;
  search_result_id: string;
  merchant_id: string;
  merchant_name: string;
  merchant_url: string;
  contact_url: string;
  contact_handle: string;
  contact_site: string;
  offer_url: string;
  title: string;
  ask_context: Record<string, string> | null;
}

interface InquiryFunnelRoundPayload {
  round: number;
  name: string;
  templateText: string;
  passCondition: string;
}

interface InquiryFunnelConfigPayload {
  id: string;
  name: string;
  rounds: InquiryFunnelRoundPayload[];
}

const MAX_INQUIRY_FUNNEL_ROUNDS = 100;

function normalizeInquiryFunnelConfig(input: unknown): InquiryFunnelConfigPayload | null {
  if (!input || typeof input !== "object") {
    return null;
  }

  const payload = input as {
    id?: unknown;
    name?: unknown;
    rounds?: unknown;
  };

  if (!Array.isArray(payload.rounds)) {
    return null;
  }

  const rounds = payload.rounds
    .map((round, index) => {
      if (!round || typeof round !== "object") {
        return null;
      }

      const value = round as {
        round?: unknown;
        name?: unknown;
        templateText?: unknown;
        passCondition?: unknown;
        enabled?: unknown;
      };

      if (value.enabled === false) {
        return null;
      }

      const templateText = String(value.templateText ?? "").trim();
      if (!templateText) {
        return null;
      }

      const parsedRound = Number.parseInt(String(value.round ?? index + 1), 10);
      return {
        round: Number.isFinite(parsedRound) && parsedRound > 0 ? parsedRound : index + 1,
        name: String(value.name ?? `第 ${index + 1} 轮`).trim() || `第 ${index + 1} 轮`,
        templateText,
        passCondition: String(value.passCondition ?? "").trim(),
      };
    })
    .filter((round): round is InquiryFunnelRoundPayload => round !== null)
    .sort((left, right) => left.round - right.round)
    .slice(0, MAX_INQUIRY_FUNNEL_ROUNDS)
    .map((round, index) => ({
      ...round,
      round: index + 1,
    }));

  if (rounds.length === 0) {
    return null;
  }

  return {
    id: String(payload.id ?? "runtime").trim() || "runtime",
    name: String(payload.name ?? "运行时询价漏斗").trim() || "运行时询价漏斗",
    rounds,
  };
}

const ACTIVE_INQUIRY_STATUSES = new Set<InquiryProgressStatus>(["processing", "stopping"]);

interface InquiryRunState {
  progress: InquiryProgress;
  worker: ChildProcess | null;
  summaryTimer: ReturnType<typeof setInterval> | null;
  stopRequested: boolean;
}

const inquiryRuns = new Map<string, InquiryRunState>();

function createInquiryProgress(overrides: Partial<InquiryProgress> = {}): InquiryProgress {
  return {
    status: "idle",
    batchId: "",
    accountId: "",
    totalTasks: 0,
    completedTasks: 0,
    rejectedTasks: 0,
    humanTakeoverTasks: 0,
    currentItem: "",
    errors: [...(overrides.errors ?? [])],
    result: overrides.result ? { ...overrides.result } : null,
    startedAt: null,
    finishedAt: null,
    ...overrides,
  };
}

function isActiveInquiryRun(runState: InquiryRunState | null | undefined) {
  return Boolean(runState && ACTIVE_INQUIRY_STATUSES.has(runState.progress.status));
}

function clearInquirySummaryTimer(runState: InquiryRunState) {
  if (runState.summaryTimer) {
    clearInterval(runState.summaryTimer);
    runState.summaryTimer = null;
  }
}

function pushInquiryError(runState: InquiryRunState, message: string) {
  const trimmed = message.trim();
  if (!trimmed) {
    return;
  }

  runState.progress.errors = [...new Set([...runState.progress.errors, trimmed])].slice(-20);
}

function updateInquiryCurrentItem(runState: InquiryRunState, output: string, isError = false) {
  const lines = output
    .split(/\r?\n/)
    .map(line => line.trim())
    .filter(Boolean);

  if (lines.length === 0) {
    return;
  }

  const lastLine = lines[lines.length - 1];
  runState.progress.currentItem = lastLine;

  if (isError) {
    pushInquiryError(runState, lastLine);
  }
}

function serializeInquiryProgress(progress: InquiryProgress): InquiryProgress {
  return {
    ...progress,
    errors: [...progress.errors],
    result: progress.result ? { ...progress.result } : null,
  };
}

function getInquiryProgressSortTime(progress: InquiryProgress) {
  const timestamps = [progress.finishedAt, progress.startedAt]
    .filter((value): value is string => Boolean(value))
    .map((value) => {
      const time = Date.parse(value);
      return Number.isNaN(time) ? 0 : time;
    });

  return timestamps.length > 0 ? Math.max(...timestamps) : 0;
}

function listInquiryProgresses(): InquiryProgress[] {
  return Array.from(inquiryRuns.values())
    .map(runState => serializeInquiryProgress(runState.progress))
    .filter(progress => progress.status !== "idle" || Boolean(progress.batchId))
    .sort((left, right) => getInquiryProgressSortTime(right) - getInquiryProgressSortTime(left));
}

async function listInquiryProgressesForUser(user: ApiUserContext): Promise<InquiryProgress[]> {
  const progresses = listInquiryProgresses();
  if (user.isAdmin || progresses.length === 0) {
    return progresses;
  }

  const accessibleBatchIds = await getAccessibleBatchIdSet(user, progresses.map((progress) => progress.batchId));
  return progresses.filter((progress) => progress.batchId && accessibleBatchIds.has(progress.batchId));
}

async function getLatestInquiryProgressForUser(user: ApiUserContext): Promise<InquiryProgress> {
  return (await listInquiryProgressesForUser(user))[0] ?? createInquiryProgress();
}

function getActiveInquiryRuns() {
  return Array.from(inquiryRuns.values()).filter(runState => isActiveInquiryRun(runState));
}

async function getAccessibleActiveInquiryRuns(user: ApiUserContext) {
  const activeRuns = getActiveInquiryRuns();
  if (user.isAdmin || activeRuns.length === 0) {
    return activeRuns;
  }

  const accessibleBatchIds = await getAccessibleBatchIdSet(user, activeRuns.map((runState) => runState.progress.batchId));
  return activeRuns.filter((runState) => accessibleBatchIds.has(runState.progress.batchId));
}

function findActiveInquiryRunByAccount(accountId: string, excludedBatchId?: string) {
  return Array.from(inquiryRuns.values()).find(
    runState => runState.progress.accountId === accountId
      && runState.progress.batchId !== excludedBatchId
      && isActiveInquiryRun(runState),
  );
}

function formatInquiryNumericValue(value: number | null | undefined) {
  return typeof value === "number" && Number.isFinite(value) && value > 0 ? String(value) : "";
}

function normalizeInquirySearchTerms(value: unknown) {
  if (!Array.isArray(value)) {
    return "";
  }

  return value
    .map((item) => String(item ?? "").trim())
    .filter(Boolean)
    .join(" / ");
}

function buildInquiryAskContext(baseRow: InquiryBaseRow | undefined, title: string) {
  return {
    standardized_name: baseRow?.standardized_name || baseRow?.original_name || title || "",
    original_name: baseRow?.original_name || baseRow?.standardized_name || title || "",
    standardized_spec: baseRow?.standardized_spec || baseRow?.original_spec || "",
    original_spec: baseRow?.original_spec || baseRow?.standardized_spec || "",
    quantity: baseRow?.quantity || "",
    standardized_unit: baseRow?.standardized_unit || "",
    brand: baseRow?.brand || "",
    category: baseRow?.category || "",
    primary_search_term: baseRow?.primary_search_term || "",
    alt_search_terms: baseRow?.alt_search_terms || "",
    notes: baseRow?.notes || "",
    delivery_address: baseRow?.delivery_address || "",
    invoice_type: baseRow?.invoice_type || "",
    product_type: baseRow?.product_type || "",
    source_type: baseRow?.source_type || "",
  };
}

function buildInquiryCandidate(
  row: InquiryFinalCandidateRow,
  baseMap: Map<string, InquiryBaseRow>,
): InquiryCandidate {
  const contactUrl = row.wangwang_url || row.contact_page_url || "";
  const contactIdentity = parseWangwangContactIdentity(contactUrl);

  return {
    search_result_id: row.id,
    batch_id: row.batch_id,
    material_id: row.material_id,
    merchant_name: row.merchant_name || "",
    merchant_url: row.merchant_url || "",
    contact_url: contactUrl,
    contact_handle: contactIdentity.handle,
    contact_site: contactIdentity.site,
    offer_url: row.offer_url || "",
    title: row.title || "",
    ask_context: buildInquiryAskContext(baseMap.get(row.material_id), row.title || ""),
  };
}

function shouldPatchInquiryTask(existingRow: ExistingInquiryTaskRow, candidate: InquiryCandidate) {
  const nextContactHandle = existingRow.contact_handle || candidate.contact_handle;
  const nextContactSite = existingRow.contact_site && existingRow.contact_site !== "1688"
    ? existingRow.contact_site
    : (candidate.contact_site || existingRow.contact_site || "");

  return existingRow.merchant_id !== (nextContactHandle || existingRow.merchant_id)
    || existingRow.merchant_name !== candidate.merchant_name
    || existingRow.merchant_url !== candidate.merchant_url
    || existingRow.contact_url !== candidate.contact_url
    || existingRow.contact_handle !== nextContactHandle
    || existingRow.contact_site !== nextContactSite
    || existingRow.offer_url !== candidate.offer_url
    || existingRow.title !== candidate.title
    || JSON.stringify(existingRow.ask_context ?? {}) !== JSON.stringify(candidate.ask_context);
}

function buildPatchedInquiryTask(existingRow: ExistingInquiryTaskRow, candidate: InquiryCandidate) {
  const nextContactHandle = existingRow.contact_handle || candidate.contact_handle;
  const nextContactSite = existingRow.contact_site && existingRow.contact_site !== "1688"
    ? existingRow.contact_site
    : (candidate.contact_site || existingRow.contact_site || "");

  return {
    merchant_id: nextContactHandle || existingRow.merchant_id || "",
    merchant_name: candidate.merchant_name,
    merchant_url: candidate.merchant_url,
    contact_url: candidate.contact_url,
    contact_handle: nextContactHandle,
    contact_site: nextContactSite,
    offer_url: candidate.offer_url,
    title: candidate.title,
    ask_context: candidate.ask_context,
  };
}

async function getInquiryTaskSummary(batchId: string): Promise<InquiryTaskSummary> {
  const { data, error } = await supabase
    .from("customer_service_chats")
    .select("status")
    .eq("batch_id", batchId);

  if (error) {
    throw new Error(`[customer_service_chats] 汇总失败: ${error.message}`);
  }

  const summary: InquiryTaskSummary = {
    total: 0,
    completed: 0,
    rejected: 0,
    humanTakeover: 0,
    active: 0,
  };

  for (const row of data ?? []) {
    summary.total += 1;

    if (row.status === "COMPLETED") {
      summary.completed += 1;
      continue;
    }

    if (row.status === "REJECTED" || row.status === "TIMEOUT_REJECTED") {
      summary.rejected += 1;
      continue;
    }

    if (row.status === "HUMAN_TAKEOVER") {
      summary.humanTakeover += 1;
      continue;
    }

    summary.active += 1;
  }

  return summary;
}

async function refreshInquiryTaskSummary(batchId: string) {
  const runState = inquiryRuns.get(batchId);
  if (!runState) {
    return {
      total: 0,
      completed: 0,
      rejected: 0,
      humanTakeover: 0,
      active: 0,
    } satisfies InquiryTaskSummary;
  }

  const summary = await getInquiryTaskSummary(batchId);
  runState.progress.totalTasks = summary.total;
  runState.progress.completedTasks = summary.completed;
  runState.progress.rejectedTasks = summary.rejected;
  runState.progress.humanTakeoverTasks = summary.humanTakeover;
  return summary;
}

function startInquirySummaryPolling(runState: InquiryRunState) {
  const { batchId } = runState.progress;

  clearInquirySummaryTimer(runState);
  void refreshInquiryTaskSummary(batchId).catch((err) => {
    pushInquiryError(runState, err instanceof Error ? err.message : String(err));
  });

  runState.summaryTimer = setInterval(() => {
    void refreshInquiryTaskSummary(batchId).catch((err) => {
      pushInquiryError(runState, err instanceof Error ? err.message : String(err));
    });
  }, 2000);
}

async function finalizeInquiryProgress(
  batchId: string,
  status: InquiryProgressStatus,
  currentItem: string,
  result: Record<string, unknown>,
) {
  const runState = inquiryRuns.get(batchId);
  if (!runState) {
    return;
  }

  clearInquirySummaryTimer(runState);

  if (runState.progress.batchId) {
    try {
      await refreshInquiryTaskSummary(runState.progress.batchId);
    } catch (err) {
      pushInquiryError(runState, err instanceof Error ? err.message : String(err));
    }
  }

  runState.progress.status = status;
  runState.progress.currentItem = currentItem;
  runState.progress.finishedAt = new Date().toISOString();
  runState.progress.result = {
    totalTasks: runState.progress.totalTasks,
    completedTasks: runState.progress.completedTasks,
    rejectedTasks: runState.progress.rejectedTasks,
    humanTakeoverTasks: runState.progress.humanTakeoverTasks,
    ...result,
  };
  runState.worker = null;
}

async function loadInquiryCandidates(batchId: string): Promise<{ source: string; candidates: InquiryCandidate[] }> {
  const [finalRes, standardizedRes] = await Promise.all([
    supabase
      .from("final_result_products")
      .select("id, batch_id, material_id, title, offer_url, merchant_name, merchant_url, wangwang_url, contact_page_url")
      .eq("batch_id", batchId)
      .order("created_at", { ascending: true }),
    supabase
      .from("standardized_results")
      .select("material_id, source_type, invoice_type, original_name, standardized_name, category, brand, notes, delivery_address, primary_search_term, alt_search_terms, standardized_spec, standardized_unit, quantity, product_type, original_spec")
      .eq("batch_id", batchId)
      .order("created_at", { ascending: true }),
  ]);

  if (finalRes.error) throw new Error(`[final_result_products] 查询失败: ${finalRes.error.message}`);
  if (standardizedRes.error) throw new Error(`[standardized_results] 查询失败: ${standardizedRes.error.message}`);

  const baseMap = new Map(
    ((standardizedRes.data ?? []) as Array<{
      material_id: string;
      source_type: string;
      original_name: string;
      standardized_name: string;
      category: string;
      original_spec: string;
      brand: string;
      notes: string;
      delivery_address: string;
      primary_search_term: string;
      alt_search_terms: unknown;
      standardized_spec: string;
      standardized_unit: string;
      quantity: number;
      product_type: string;
      invoice_type: string;
    }>).map((row) => [row.material_id, {
      material_id: row.material_id,
      original_name: row.original_name || row.standardized_name || "",
      standardized_name: row.standardized_name || row.original_name || "",
      original_spec: row.original_spec || row.standardized_spec || "",
      standardized_spec: row.standardized_spec || row.original_spec || "",
      brand: row.brand || "",
      quantity: formatInquiryNumericValue(row.quantity),
      standardized_unit: row.standardized_unit || "",
      category: row.category || "",
      notes: row.notes || "",
      delivery_address: row.delivery_address || "",
      primary_search_term: row.primary_search_term || "",
      alt_search_terms: normalizeInquirySearchTerms(row.alt_search_terms),
      invoice_type: row.invoice_type || "",
      product_type: row.product_type || "",
      source_type: row.source_type || "",
    } satisfies InquiryBaseRow]),
  );

  const finalCandidates = ((finalRes.data ?? []) as InquiryFinalCandidateRow[])
    .map((row) => buildInquiryCandidate(row, baseMap))
    .filter(candidate => Boolean(candidate.contact_url));

  if (finalCandidates.length > 0) {
    return { source: "final_result_products", candidates: finalCandidates };
  }

  return { source: "final_result_products", candidates: [] };
}

async function ensureInquiryTasks(batchId: string) {
  const { source, candidates } = await loadInquiryCandidates(batchId);
  if (candidates.length === 0) {
    return { source, insertedTasks: 0, totalTasks: 0 };
  }

  const { data: existingRows, error: existingError } = await supabase
    .from("customer_service_chats")
    .select("id, search_result_id, merchant_id, merchant_name, merchant_url, contact_url, contact_handle, contact_site, offer_url, title, ask_context")
    .eq("batch_id", batchId);

  if (existingError) {
    throw new Error(`[customer_service_chats] 查询失败: ${existingError.message}`);
  }

  const existingRowsBySearchResultId = new Map(
    ((existingRows ?? []) as ExistingInquiryTaskRow[])
      .filter((row) => Boolean(row.search_result_id))
      .map((row) => [String(row.search_result_id), row]),
  );

  const existingIds = new Set(existingRowsBySearchResultId.keys());

  const rowsToPatch = candidates
    .map((candidate) => {
      const existingRow = existingRowsBySearchResultId.get(candidate.search_result_id);
      if (!existingRow || !shouldPatchInquiryTask(existingRow, candidate)) {
        return null;
      }

      return {
        id: existingRow.id,
        payload: buildPatchedInquiryTask(existingRow, candidate),
      };
    })
    .filter((row): row is { id: string; payload: ReturnType<typeof buildPatchedInquiryTask> } => row !== null);

  if (rowsToPatch.length > 0) {
    await Promise.all(rowsToPatch.map(({ id, payload }) => supabase
      .from("customer_service_chats")
      .update(payload)
      .eq("id", id)
      .then(({ error }) => {
        if (error) {
          throw new Error(`[customer_service_chats] 修复失败: ${error.message}`);
        }
      })));
  }

  const insertRows = candidates
    .filter(candidate => !existingIds.has(candidate.search_result_id))
    .map(candidate => ({
      intent_id: null,
      search_result_id: candidate.search_result_id,
      batch_id: candidate.batch_id,
      merchant_id: candidate.contact_handle || "",
      merchant_name: candidate.merchant_name,
      merchant_url: candidate.merchant_url,
      contact_url: candidate.contact_url,
      contact_handle: candidate.contact_handle,
      contact_site: candidate.contact_site,
      offer_url: candidate.offer_url,
      title: candidate.title,
      current_round: 1,
      status: "PENDING_SEND",
      ask_context: candidate.ask_context,
    }));

  if (insertRows.length > 0) {
    const { error: insertError } = await supabase
      .from("customer_service_chats")
      .insert(insertRows);

    if (insertError) {
      throw new Error(`[customer_service_chats] 创建失败: ${insertError.message}`);
    }
  }

  const summary = await getInquiryTaskSummary(batchId);
  return { source, insertedTasks: insertRows.length, totalTasks: summary.total };
}

// ── POST /api/inquiry-batch — 启动批次智能询价 ───────────────
const InquiryBatchBody = z.object({
  batch_id: z.string().trim().min(1, "缺少 batch_id 参数"),
  account_id: z.string().trim().min(1, "缺少 account_id 参数"),
  headless: z.boolean().optional(),
  funnel_config: z.unknown().optional(),
});

app.post("/api/inquiry-batch", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const { batch_id, account_id, headless, funnel_config } = parseBody(req, InquiryBatchBody);

    await assertBatchAccess(apiUser, batch_id);

    const existingRun = inquiryRuns.get(batch_id);
    if (isActiveInquiryRun(existingRun)) {
      res.status(409).json({ error: `批次 ${batch_id} 的询价任务已在运行中` });
      return;
    }

    const conflictingRun = findActiveInquiryRunByAccount(account_id, batch_id);
    if (conflictingRun) {
      res.status(409).json({
        error: `1688 账户 ${account_id} 已用于批次 ${conflictingRun.progress.batchId}，请更换账户或先停止该任务`,
      });
      return;
    }

    const account = await getAuthorizedProcurementAccount(apiUser, account_id);
    if (!account.account_name || !account.password) {
      res.status(400).json({ error: `1688 账户 ${account_id} 缺少账号名或密码` });
      return;
    }

    const { source, insertedTasks, totalTasks } = await ensureInquiryTasks(batch_id);
    if (totalTasks === 0) {
      res.status(404).json({ error: "当前批次暂无可自动询价的商家结果，请先完成 1688 搜索并确保存在旺旺联系入口" });
      return;
    }

    const accountProfileDir = path.join(__dirname, "..", ".browser", "1688-profiles", account.id);
    const inquiryHeadless = headless === true;
    const runtimeFunnelConfig = normalizeInquiryFunnelConfig(funnel_config);
    const childEnv = {
      ...process.env,
      SUPABASE_service_key: process.env.SUPABASE_service_key
        ?? process.env.SUPABASE_SERVICE_ROLE_KEY
        ?? process.env.SUPABASE_ANON_KEY
        ?? "",
      SEARCH_1688_ACCOUNT_ID: account.id,
      SEARCH_1688_ACCOUNT_NAME: account.account_name,
      LOGIN_1688_USER: account.account_name,
      LOGIN_1688_PASS: account.password,
      SEARCH_1688_PROFILE_DIR: accountProfileDir,
      CHAT_1688_PROFILE_DIR: accountProfileDir,
      CHAT_BATCH_ID: batch_id,
      CHAT_FUNNEL_CONFIG_JSON: runtimeFunnelConfig ? JSON.stringify(runtimeFunnelConfig) : "",
      WORKER_EXIT_ON_IDLE: "1",
      HEADLESS: inquiryHeadless ? "1" : "0",
    };

    const runState: InquiryRunState = {
      progress: createInquiryProgress({
        status: "processing",
        batchId: batch_id,
        accountId: account_id,
        totalTasks,
        completedTasks: 0,
        rejectedTasks: 0,
        humanTakeoverTasks: 0,
        currentItem: "正在准备询价任务...",
        errors: [],
        result: null,
        startedAt: new Date().toISOString(),
        finishedAt: null,
      }),
      worker: null,
      summaryTimer: null,
      stopRequested: false,
    };
    inquiryRuns.set(batch_id, runState);
    startInquirySummaryPolling(runState);

    res.json({
      message: "询价任务已启动",
      batchId: batch_id,
      totalTasks,
      insertedTasks,
      source,
    });

    (async () => {
      try {
        runState.progress.currentItem = `正在登录账号: ${account.account_name}`;
        await execFileAsync(
          process.execPath,
          [
            LOGIN_1688_SCRIPT,
            `--account-id=${account.id}`,
            `--profile-dir=${accountProfileDir}`,
            "--headless=0",
            "--timeout-ms=90000",
          ],
          {
            cwd: path.join(__dirname, ".."),
            maxBuffer: 20 * 1024 * 1024,
            env: childEnv,
          },
        );

        if (runState.stopRequested) {
          await finalizeInquiryProgress(batch_id, "stopped", "询价已停止", { stopped: true });
          return;
        }

        runState.progress.currentItem = `正在启动批次 ${batch_id} 的询价 Worker`;

        const worker = spawn(process.execPath, [CHATBOT_WORKER_SCRIPT], {
          cwd: CHATBOT_DIR,
          env: childEnv,
          stdio: ["ignore", "pipe", "pipe"],
        });
        runState.worker = worker;
        let workerSettled = false;

        worker.stdout.on("data", (chunk: Buffer) => {
          updateInquiryCurrentItem(runState, chunk.toString(), false);
        });

        worker.stderr.on("data", (chunk: Buffer) => {
          updateInquiryCurrentItem(runState, chunk.toString(), true);
        });

        worker.on("error", async (error) => {
          if (workerSettled) {
            return;
          }
          workerSettled = true;
          runState.worker = null;
          pushInquiryError(runState, `询价 Worker 启动失败: ${error.message}`);
          await finalizeInquiryProgress(batch_id, "failed", "询价 Worker 启动失败", { error: error.message });
        });

        worker.on("exit", async (code, signal) => {
          if (workerSettled) {
            return;
          }
          workerSettled = true;
          runState.worker = null;

          if (runState.stopRequested) {
            await finalizeInquiryProgress(batch_id, "stopped", "询价已停止", {
              stopped: true,
              exitCode: code,
              signal,
            });
            return;
          }

          const summary = await getInquiryTaskSummary(batch_id).catch(() => null);
          const isFinished = summary ? summary.total > 0 && summary.active === 0 : false;

          if (code === 0 && isFinished) {
            await finalizeInquiryProgress(batch_id, "completed", "询价完成", { success: true });
            return;
          }

          const reason = `询价 Worker 异常退出 (code=${code ?? "null"}, signal=${signal ?? "null"})`;
          pushInquiryError(runState, reason);
          await finalizeInquiryProgress(batch_id, "failed", "询价失败", { error: reason });
        });
      } catch (err) {
        runState.worker = null;
        const message = err instanceof Error ? err.message : String(err);
        pushInquiryError(runState, message);
        await finalizeInquiryProgress(batch_id, "failed", "询价失败", { error: message });
      }
    })();
  } catch (err) {
    sendApiError(res, err);
  }
});

// ── GET /api/inquiry-statuses — 返回询价任务列表 ──────────────
app.get("/api/inquiry-statuses", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    res.json(await listInquiryProgressesForUser(apiUser));
  } catch (error) {
    sendApiError(res, error);
  }
});

// ── GET /api/inquiry-status — 轮询询价进度 ───────────────────
app.get("/api/inquiry-status", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const batchId = typeof req.query.batch_id === "string" ? req.query.batch_id.trim() : "";

    if (batchId) {
      await assertBatchAccess(apiUser, batchId);
      const runState = inquiryRuns.get(batchId);
      res.json(runState ? serializeInquiryProgress(runState.progress) : createInquiryProgress({ batchId }));
      return;
    }

    res.json(await getLatestInquiryProgressForUser(apiUser));
  } catch (error) {
    sendApiError(res, error);
  }
});

// ── POST /api/inquiry-stop — 停止当前询价任务 ───────────────
const InquiryStopBody = z.object({
  batch_id: z.string().optional(),
});

app.post("/api/inquiry-stop", requireApiUser, async (req: AuthenticatedRequest, res) => {
  try {
    const apiUser = getRequestApiUser(req);
    const batchId = parseBody(req, InquiryStopBody).batch_id?.trim() ?? "";

    let runState: InquiryRunState | undefined;

    if (batchId) {
      await assertBatchAccess(apiUser, batchId);
      runState = inquiryRuns.get(batchId);
      if (!isActiveInquiryRun(runState)) {
        res.status(400).json({ error: `批次 ${batchId} 当前没有正在运行的询价任务` });
        return;
      }
    } else {
      const accessibleRuns = await getAccessibleActiveInquiryRuns(apiUser);
      if (accessibleRuns.length === 0) {
        res.status(400).json({ error: "当前没有正在运行的询价任务" });
        return;
      }
      if (accessibleRuns.length > 1) {
        res.status(409).json({ error: "当前存在多个询价任务，请指定 batch_id 后再停止" });
        return;
      }
      [runState] = accessibleRuns;
    }

    if (!runState) {
      res.status(400).json({ error: "当前没有正在运行的询价任务" });
      return;
    }

    runState.stopRequested = true;
    runState.progress.status = "stopping";
    runState.progress.currentItem = "正在停止询价任务...";

    if (runState.worker) {
      runState.worker.kill("SIGTERM");
    }

    res.json({ message: "询价停止信号已发送", batchId: runState.progress.batchId });
  } catch (error) {
    sendApiError(res, error);
  }
});

// ── POST /api/extract-materials — 原始文件 → 文档提取阶段 LLM 提取物料 → 落库 ──
// multipart/form-data：multer 把非文件字段都放进 req.body 字符串字段。
const ExtractMaterialsBody = z.object({
  batch_id: z.string().trim().min(1, "缺少 batch_id 参数"),
  source_file_id: z.string().trim().min(1, "缺少 source_file_id 参数"),
  source_file_name: z.string().trim().min(1, "缺少 source_file_name 参数"),
  created_by: z.string().trim().min(1, "created_by 不能为空").optional(),
});

app.post("/api/extract-materials", upload.single("file"), async (req, res) => {
  const startTime = Date.now();

  if (!req.file) {
    res.status(400).json({ error: "请上传一个文件" });
    return;
  }

  let batchId: string;
  let sourceFileId: string;
  let sourceFileName: string;
  let createdBy: string | undefined;
  try {
    const parsed = parseBody(req, ExtractMaterialsBody);
    batchId = parsed.batch_id;
    sourceFileId = parsed.source_file_id;
    sourceFileName = parsed.source_file_name;
    createdBy = parsed.created_by;
  } catch (err) {
    sendApiError(res, err);
    return;
  }

  const fileSizeKB = Math.round(req.file.size / 1024);
  console.log(`[extract-materials] 收到请求: batch_id=${batchId}, source_file_id=${sourceFileId}, file=${sourceFileName}, size=${fileSizeKB}KB, mimeType=${req.file.mimetype}`);

  try {
    const result = await extractAndInsertMaterialsFromUploadedFile({
      fileBuffer: req.file.buffer,
      mimeType: req.file.mimetype,
      batchId,
      sourceFileId,
      sourceFileName,
      createdBy,
    });
    const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
    console.log(`[extract-materials] 完成: file=${sourceFileName}, batch=${batchId}, model=${result.extraction_model}, items=${result.materials.length}, elapsed=${elapsed}s`);

    res.json({
      ...result,
      elapsed: `${elapsed}s`,
    });
  } catch (error: unknown) {
    const message = error instanceof Error ? error.message : String(error);
    const fallbackExtractionModel = await resolveLlmRuntimeConfig("document_extraction")
      .then((config) => config.model)
      .catch(() => resolveDocumentExtractionModel());
    const structuredError = buildExtractMaterialsErrorResponse(error, {
      batchId,
      sourceFileId,
      sourceFileName,
      extractionModel: fallbackExtractionModel,
    });
    console.error(`[extract-materials] 失败: file=${sourceFileName}, batch=${batchId}`, message);
    res.status(structuredError.status).json(structuredError.body);
  }
});

if (hasBuiltFrontend) {
  app.use(express.static(DIST_DIR));

  app.get(/^(?!\/api(?:\/|$)).*/, (_req, res) => {
    res.sendFile(DIST_INDEX_FILE);
  });
}

// ── 启动服务 ──────────────────────────────────────────────────
app.listen(PORT, () => {
  console.log(`[server] 自动化 API 服务启动: http://localhost:${PORT}`);
  if (hasBuiltFrontend) {
    console.log(`[server] 前端静态资源托管: http://localhost:${PORT}`);
  } else {
    console.log(`[server] 未检测到 dist/index.html，当前仅启动 API 服务`);
  }
  console.log(`[server] 路由:`);
  console.log(`  GET  /api/state`);
  console.log(`  GET  /api/keyword-analysis   (阶段一分析报告)`);
  console.log(`  GET  /api/search-evaluation  (文字搜索结果)`);
  console.log(`  GET  /api/image-search       (图搜阶段)`);
  console.log(`  GET  /api/keyword-selection  (主关键词选定)`);
  console.log(`  POST /api/upload             (上传 CSV)`);
  console.log(`  POST /api/upload/confirm     (确认写入 Supabase)`);
  console.log(`  POST /api/standardize-batch  (批次标准化)`);
  console.log(`  GET  /api/standardize-status (标准化进度)`);
  console.log(`  POST /api/standardize-stop   (停止在线标准化)`);
  console.log(`  POST /api/search-batch       (启动 1688 搜索)`);
  console.log(`  GET  /api/search-executor-status (搜索执行器占用状态)`);
  console.log(`  GET  /api/search-status      (搜索进度)`);
  console.log(`  POST /api/search-pause       (安全暂停搜索)`);
  console.log(`  POST /api/search-resume      (恢复暂停搜索)`);
  console.log(`  POST /api/search-stop        (停止搜索)`);
  console.log(`  CRUD /api/search-presets     (搜索预设管理)`);
  console.log(`  POST /api/inquiry-batch      (启动智能询价)`);
  console.log(`  GET  /api/inquiry-statuses   (询价任务列表)`);
  console.log(`  GET  /api/inquiry-status     (询价进度)`);
  console.log(`  POST /api/inquiry-stop       (停止询价)`);
  console.log(`  POST /api/extract-materials   (文档提取物料)`);
  void recoverOrphanSearchTasks().catch((error) => {
    console.error("[server] 回收未完成搜索任务失败:", error);
  });
  void restorePendingStandardizationRuns().catch((error) => {
    console.error("[server] 处理未完成标准化任务失败:", error);
  });
});
