import { createHash } from "node:crypto";
import { createReadStream, createWriteStream } from "node:fs";
import { stat, rm } from "node:fs/promises";
import { Transform, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import path from "node:path";
import { ProviderTaskUncertainError } from "../video-task-lifecycle";
import { audioJobInputSchema, audioJobStatusSchema } from "./audio-contract";
const sha = (s: string) => createHash("sha256").update(s).digest("hex");
export function remoteAudioConfig() {
  const value = process.env.SMART_EDIT_AUDIO_URL;
  if (!value) return null;
  const url = new URL(value),
    token = process.env.SMART_EDIT_AUDIO_TOKEN;
  if (
    !["http:", "https:"].includes(url.protocol) ||
    url.username ||
    url.password ||
    url.search ||
    url.hash ||
    !token ||
    token.length < 32
  )
    throw Error("远程音频服务配置无效");
  return { base: value.replace(/\/$/, ""), token };
}
export async function remoteAudioCapabilities() {
  const config = remoteAudioConfig();
  if (!config) return null;
  try {
    const r = await fetch(config.base + "/health", {
      headers: { authorization: "Bearer " + config.token },
      signal: AbortSignal.timeout(4000),
      redirect: "error",
      cache: "no-store",
    });
    if (!r.ok) throw Error();
    const data = await r.json();
    return {
      asr: data.status === "ok" && data.capabilities?.asr === true,
      separation:
        data.status === "ok" && data.capabilities?.separation === true,
    };
  } catch {
    return { asr: false, separation: false };
  }
}
export async function audioFileHash(file: string) {
  const h = createHash("sha256");
  for await (const chunk of createReadStream(file)) h.update(chunk);
  return h.digest("hex");
}
export async function remoteAudio(
  operation: "transcribe" | "separate",
  file: string,
  ownerKey: string,
  requestKey: string,
  dir: string,
  progress: (stage: string) => Promise<void> = async () => {}
) {
  const config = remoteAudioConfig();
  if (!config) throw Error("远程音频服务未配置");
  const plan = audioJobInputSchema.parse({
    operation,
    bytes: (await stat(file)).size,
    sha256: await audioFileHash(file),
  });
  const tenant = sha(ownerKey),
    id = sha(
      "audio-v1:" + tenant + ":" + requestKey + ":" + JSON.stringify(plan)
    );
  const headers = { authorization: "Bearer " + config.token };
  async function request(
    route: string,
    init: RequestInit = {},
    timeoutMs?: number
  ) {
    let r: Response;
    try {
      r = await fetch(config!.base + route, {
        ...init,
        headers: { ...headers, ...init.headers },
        signal: AbortSignal.timeout(
          timeoutMs || (init.method === "PUT" ? 180000 : 20000)
        ),
        redirect: "error",
        cache: "no-store",
      });
    } catch {
      throw new ProviderTaskUncertainError(
        "音频服务连接中断，任务已保留，可继续处理同一音频"
      );
    }
    if (
      (init.method === "PUT" && r.status === 409) ||
      [429, 500, 502, 503, 504].includes(r.status)
    )
      throw new ProviderTaskUncertainError(
        "音频服务繁忙，等待继续处理同一任务"
      );
    if (!r.ok) throw Error(`音频服务请求失败（${r.status}）`);
    return r;
  }
  async function status(r: Response) {
    let value: unknown;
    try {
      value = await r.json();
    } catch {
      throw new ProviderTaskUncertainError(
        "音频服务响应中断，将继续查询同一任务"
      );
    }
    const s = audioJobStatusSchema.parse(value);
    if (s.id !== id) throw Error("音频任务编号不匹配");
    return s;
  }
  let current = await status(
    await request("/v1/audio-jobs", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ id, tenant, plan }),
    })
  );
  if (current.status === "awaiting-input") {
    await progress("传送音频到算力服务器");
    current = await status(
      await request("/v1/audio-jobs/" + id + "/input", {
        method: "PUT",
        headers: {
          "content-type": "audio/wav",
          "content-length": String(plan.bytes),
        },
        body: createReadStream(file) as any,
        duplex: "half",
      } as RequestInit)
    );
  }
  const deadline = Date.now() + 8 * 60000;
  while (current.status === "queued" || current.status === "running") {
    await progress(current.stage);
    if (Date.now() > deadline)
      throw new ProviderTaskUncertainError(
        "远程音频仍在处理中，任务已保存，稍后继续查询"
      );
    await new Promise((r) => setTimeout(r, 2000));
    current = await status(await request("/v1/audio-jobs/" + id));
  }
  if (current.status === "failed") throw Error(current.error || "音频处理失败");
  if (current.status !== "succeeded" || !current.result)
    throw Error("音频服务没有返回有效结果");
  const result = current.result;
  if (operation === "transcribe") {
    if (!result.subtitles) throw Error("音频服务缺少转写结果");
    if (
      result.subtitles.some(
        (s) => s.end > result.durationSeconds * 1e6 + 100000
      )
    )
      throw Error("转写时间戳超出音频范围");
  } else {
    if (!result.stems) throw Error("音频服务缺少分离结果");
    for (const stem of ["vocals", "background"] as const) {
      await progress(stem === "vocals" ? "读取人声音轨" : "读取背景声音轨");
      const destination = path.join(dir, stem + ".wav"),
        temp = destination + ".part";
      try {
        const r = await request(
          "/v1/audio-jobs/" + id + "/" + stem,
          {},
          180000
        );
        if (!r.body) throw Error();
        let bytes = 0;
        const hash = createHash("sha256"),
          expected = result.stems[stem];
        await pipeline(
          Readable.fromWeb(r.body as any),
          new Transform({
            transform(chunk, _enc, cb) {
              bytes += chunk.length;
              hash.update(chunk);
              cb(bytes > expected.bytes ? Error("音轨超过上限") : null, chunk);
            },
          }),
          createWriteStream(temp, { mode: 0o600 })
        );
        if (bytes !== expected.bytes || hash.digest("hex") !== expected.sha256)
          throw Error("分离音轨校验失败");
        const { rename } = await import("node:fs/promises");
        await rename(temp, destination);
      } catch {
        throw new ProviderTaskUncertainError(
          "分离音轨读取中断，将继续读取同一结果"
        );
      } finally {
        await rm(temp, { force: true });
      }
    }
  }
  return result;
}