import "server-only";
import { processEditAudio } from "./audio-process";
import { sql } from "drizzle-orm";
import { db } from "@/lib/db";
import { uploadFile } from "@/lib/storage/service";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { randomUUID } from "node:crypto";
import {
  SECOND,
  smartEditRequestSchema,
  buildEditPlans,
  planTimeline,
  editPlanSchema,
  type EditAsset,
  type EditCandidate,
  type EditPlan,
} from "./contract";
import {
  materialize,
  probeMedia,
  prepareAnalysis,
  pythonAnalysis,
  prepareAudio,
  ffmpeg,
  pruneSmartEditCache,
} from "./local";
import { selectEditCandidates } from "./semantic";
import { selectSequenceHighlights } from "./highlights";
import { HighlightValidationError } from "./highlight-contract";
import { renderJob } from "./render-job";
import { remoteRender, remoteRenderConfig, type RemoteRenderCheckpoint } from "./remote-render";
import { ProviderTaskUncertainError } from "../video-task-lifecycle";
export type SmartEditCheckpoint = {
  operation: "smart-edit";
  action: "plan" | "render" | "separate" | "transcribe";
  phase: "local" | "analyzed" | "selecting" | "selected" | "remote-render";
  remote?: RemoteRenderCheckpoint;
  stage: string;
  assets?: EditAsset[];
  candidates?: EditCandidate[];
  beats?: number[];
  warnings?: string[];
  plans?: EditPlan[];
};
export async function executeSmartEdit(
  value: unknown,
  userId: string,
  lifecycle: {
    taskId: string;
    resume?: SmartEditCheckpoint | null;
    assertActive: () => Promise<void>;
    checkpoint: (result: Record<string, unknown>) => Promise<void>;
  },
) {
  const params = smartEditRequestSchema.parse(value);
  let state: SmartEditCheckpoint = lifecycle.resume || {
    operation: "smart-edit",
    action: params.operation,
    phase: "local",
    stage: "等待处理",
  };
  const save = async (patch: Partial<SmartEditCheckpoint>) => {
    await lifecycle.assertActive();
    state = { ...state, ...patch };
    await lifecycle.checkpoint({ smartEdit: state });
  };
  await save({});
  const upload = async (file: string, suffix: string) => {
    await lifecycle.assertActive();
    const bytes = await readFile(file);
    return (
      await uploadFile(`smart-edit-${randomUUID()}.${suffix}`, Uint8Array.from(bytes).buffer)
    ).url;
  };
  // New edits can use local FFmpeg while already submitted remote jobs keep
  // their original endpoint identity for safe recovery.
  if (params.operation === "render" &&
      (state.remote || (process.env.SMART_EDIT_RENDER_FORCE_LOCAL !== "1" && remoteRenderConfig()))) {
    const dir = await mkdtemp(path.join(tmpdir(), "remote-render-"));
    try {
      const rendered = await remoteRender(params.plan, userId, lifecycle.taskId, dir, {
        resume: state.remote,
        active: lifecycle.assertActive,
        save: (remote, stage) => save({ phase: "remote-render", remote, stage }),
      });
      const actual = await probeMedia(rendered.file, 300.2);
      if (
        Math.abs(actual.duration / SECOND - planTimeline(params.plan).duration / SECOND) >
          0.15 ||
        actual.width !== rendered.width ||
        actual.height !== rendered.height
      )
        throw Error("远程成片与剪辑方案不一致");
      await save({ stage: "转存成片" });
      const url = await upload(rendered.file, "mp4"),
        subtitleUrl = await upload(rendered.srt, "srt");
      return {
        url,
        subtitleUrl,
        plan: params.plan,
        metadata: {
          width: actual.width,
          height: actual.height,
          durationSeconds: actual.duration / SECOND,
          storageStatus: "archived",
          renderBackend: "remote",
        },
      };
    } finally {
      await rm(dir, { recursive: true, force: true });
    }
  }
  return db.transaction(async (tx) => {
    const rows = await tx.execute(sql`select pg_try_advisory_xact_lock(91827, 1) as acquired`);
    if (!(rows as any)[0]?.acquired)
      throw new ProviderTaskUncertainError("剪辑处理器繁忙，任务已保存并等待继续");
    await pruneSmartEditCache();
    const dir = await mkdtemp(path.join(tmpdir(), "smart-edit-"));
    try {
      if (params.operation === "plan") {
        if (state.phase === "selected" && state.plans)
          return { plans: state.plans, summary: "剪辑方案已生成" };
        if (state.phase === "selecting")
          throw new ProviderTaskUncertainError(
            "智能选片请求状态未确认，禁止自动重发；请核对模型记录",
          );
        let assets = state.assets || [],
          candidates = state.candidates || [],
          beats = state.beats || [],
          warnings = state.warnings || [];
        if (state.phase === "local") {
          assets = [];
          candidates = [];
          warnings = [];
          beats = [];
          let total = 0;
          for (const input of params.assets) {
            await save({ stage: `解析素材 ${assets.length + 1}/${params.assets.length}` });
            const media = await materialize(input.url, userId),
              meta = await probeMedia(media.file);
            if (!meta.width || !meta.height) throw Error(`${input.name} 不包含视频画面`);
            total += meta.duration;
            if (total > 600 * SECOND) throw Error("素材总时长不能超过 10 分钟");
            // Opening a timeline only probes source files; it must not run scene analysis or models.
            const analysis =
              params.mode === "sequence"
                ? {
                    scenes: [[0, meta.duration]],
                    active: [[0, meta.duration]],
                    subtitles: input.subtitles,
                    warnings: [],
                  }
                : await (async () => {
                    const prepared = await prepareAnalysis(
                      media.file,
                      media.hash,
                      media.dir,
                      meta.hasAudio,
                    );
                    return pythonAnalysis(
                      {
                        version: 1,
                        video: prepared.proxy,
                        audio: prepared.audio,
                        duration: meta.duration,
                        transcribe: params.transcribe && !input.subtitles.length,
                      },
                      media.dir,
                      lifecycle.taskId,
                      (stage) => save({stage}),
                    );
                  })();
            const subtitles = input.subtitles.length ? input.subtitles : analysis.subtitles;
            if (subtitles.some((s) => s.end > meta.duration + 100000))
              throw Error(`${input.name} 的字幕时间超出视频长度`);
            const asset: EditAsset = { ...input, ...meta, hash: media.hash, subtitles };
            assets.push(asset);
            warnings.push(...analysis.warnings.map((w) => `${input.name}：${w}`));
            const speech = params.mode === "talk" || params.mode === "explain";
            if (speech && !subtitles.length)
              warnings.push(
                `${input.name} 没有语句时间戳，使用画面镜头选片。可导入 SRT 或开启自动转写。`,
              );
            const regions =
              speech && subtitles.length
                ? subtitles.map((s) => ({ start: s.start, end: s.end, text: s.text }))
                : analysis.scenes.map(([start, end]) => ({
                    start,
                    end: Math.min(end, meta.duration),
                    text: "",
                  }));
            for (const region of regions) {
              const ranges =
                params.removeSilence && meta.hasAudio && !(speech && subtitles.length)
                  ? analysis.active.map(([a, b]) => ({
                      start: Math.max(region.start, a),
                      end: Math.min(region.end, b),
                      text: region.text,
                    }))
                  : [region];
              for (const range of ranges) {
                // Keep full speech sentences; split long visual scenes into bounded candidates.
                for (let start = range.start; start < range.end; ) {
                  const end =
                    params.mode === "sequence" || (speech && subtitles.length)
                      ? range.end
                      : Math.min(
                          range.end,
                          start +
                            Math.max(5, Math.min(30, (params.targetDuration || 20) / 12)) *
                              SECOND,
                        );
                  if (end - start >= 200000)
                    candidates.push({
                      id: `candidate-${candidates.length}`,
                      assetId: asset.id,
                      start,
                      end,
                      text: range.text,
                    });
                  start = end;
                }
              }
            }
          }
          // Bound model context while sampling every source fairly.
          const fair: EditCandidate[] = [];
          const grouped = assets.map((a) => candidates.filter((c) => c.assetId === a.id));
          for (let i = 0; fair.length < 24 && grouped.some((g) => i < g.length); i++)
            for (const group of grouped) if (group[i] && fair.length < 24) fair.push(group[i]);
          if (fair.length < candidates.length)
            warnings.push(
              `共识别 ${candidates.length} 个候选，本轮按素材轮询选取 24 个进入选片。`,
            );
          candidates = fair;
          if (params.beatSync && params.music) {
            await save({ stage: "检测音乐节拍" });
            const music = await materialize(params.music.url, userId),
              meta = await probeMedia(music.file);
            if (!meta.hasAudio) throw Error("背景音乐没有音轨");
            if (params.music.start >= meta.duration) throw Error("音乐开始时间超出文件");
            const audio = await prepareAudio(
              music.file,
              path.join(music.dir, music.hash + ".wav"),
            );
            const analysis = await pythonAnalysis(
              { version: 1, audio, beats: true },
              music.dir,
            );
            beats = analysis.beats;
            warnings.push(...analysis.warnings);
          } else if (params.beatSync) warnings.push("未选择音乐，跳过音乐卡点。");
          await save({
            phase: "analyzed",
            stage: "素材解析完成",
            assets,
            candidates,
            beats,
            warnings,
          });
        }
        if (!candidates.length) throw Error("没有可用片段，请关闭去静音或更换素材");
        let order = candidates.map((c) => c.id);
        let sequenceTransitions: ("cut" | "fade")[] = [];
        if (params.semantic && params.mode === "sequence" && params.shotDurations) {
          if (candidates.length !== params.shotDurations.length)
            throw Error("分镜数量和生成视频数量不一致，无法逐镜头选片");
          const contactSheets: string[] = [];
          for (const [index, candidate] of candidates.entries()) {
            await lifecycle.assertActive();
            await save({ stage: `分析视频画面 ${index + 1}/${candidates.length}` });
            const asset = assets.find((item) => item.id === candidate.assetId)!;
            const media = await materialize(asset.url, userId);
            if (media.hash !== asset.hash) throw Error("原素材内容已变化，请重新解析");
            const sheet = path.join(dir, `shot-${index}.jpg`);
            const seconds = Math.max(0.2, (candidate.end - candidate.start) / SECOND);
            const margin = Math.max(0.02, Math.min(0.5, seconds * 0.05));
            const span = seconds - margin * 2;
            await ffmpeg([
              "-ss", String(candidate.start / SECOND), "-i", media.file,
              "-vf", `trim=start=${margin.toFixed(3)}:end=${(seconds - margin + 0.03).toFixed(3)},setpts=PTS-STARTPTS,fps=5/${span.toFixed(3)},scale=240:136:force_original_aspect_ratio=decrease:force_divisible_by=2,pad=240:136:(ow-iw)/2:(oh-ih)/2,tile=3x2:nb_frames=6`,
              "-frames:v", "1", sheet,
            ]);
            contactSheets.push("data:image/jpeg;base64," + (await readFile(sheet)).toString("base64"));
          }
          await save({ phase: "selecting", stage: "AI 分析高光与镜头衔接" });
          try {
            const selected = await selectSequenceHighlights(
              candidates, contactSheets, params.prompt, params.shotDurations,
              params.shotDescriptions || [], params.preferredModelRef,
              candidates.map(candidate => assets.find(asset => asset.id === candidate.assetId)!.url),
            );
            candidates = candidates.map((candidate, index) => ({
              ...candidate,
              start: candidate.start + Math.round(selected.selections[index].startSeconds * SECOND),
              text: [params.shotDescriptions?.[index], selected.selections[index].reason].filter(Boolean).join("；").slice(0, 1500),
            }));
            sequenceTransitions = selected.selections.map((item) => item.transition);
            warnings.push(`AI 选片：${selected.summary.slice(0, 440)}`);
            selected.selections.forEach((item, index) => {
              warnings.push(`镜头 ${index + 1} 画面分析匹配评分 ${item.matchScore}/100：${item.reason.slice(0, 300)}`);
            });
          } catch (error) {
            if (error instanceof HighlightValidationError) throw error;
            throw new ProviderTaskUncertainError(
              `AI 视频高光选片未完成，系统不会自动重复调用：${error instanceof Error ? error.message : "请求中断"}`,
            );
          }
        }
        if (params.semantic && params.mode !== "sequence") {
          const images: string[] = [];
          for (const candidate of candidates) {
            await lifecycle.assertActive();
            const asset = assets.find((a) => a.id === candidate.assetId)!;
            const media = await materialize(asset.url, userId);
            if (media.hash !== asset.hash) throw Error("原素材内容已变化，请重新解析");
            await prepareAnalysis(media.file, media.hash, media.dir, false);
            const frame = path.join(dir, `${images.length}.jpg`);
            await ffmpeg([
              "-ss",
              String((candidate.start + candidate.end) / 2 / SECOND),
              "-i",
              path.join(media.dir, media.hash + ".proxy.mp4"),
              "-frames:v",
              "1",
              "-vf",
              "scale=320:180:force_original_aspect_ratio=decrease",
              frame,
            ]);
            images.push("data:image/jpeg;base64," + (await readFile(frame)).toString("base64"));
          }
          await save({ phase: "selecting", stage: "千问智能选片" });
          try {
            order = (
              await selectEditCandidates(
                candidates,
                images,
                params.prompt,
                params.preferredModelRef,
              )
            ).order;
          } catch (error) {
            throw new ProviderTaskUncertainError(
              `智能选片未完成，系统不会自动重复调用：${error instanceof Error ? error.message : "请求中断"}`,
            );
          }
        }
        const plans = buildEditPlans(params, assets, candidates, order, beats, warnings, sequenceTransitions);
        await save({ phase: "selected", stage: "剪辑方案已生成", plans });
        return { plans, summary: "剪辑方案已生成" };
      }
      if (params.operation === "separate" || params.operation === "transcribe") {
        if (state.phase === "selected" && state.plans) return { plans: state.plans };
        const plan = await processEditAudio(
          params.operation,
          params.plan,
          params.operation === "separate" ? params.clipId : undefined,
          userId,
          dir,
          (stage) => save({ stage }),
          upload,
          params.operation === "separate" ? params.track : "video",
          params.operation === "separate" ? params.keep : "both",
          lifecycle.taskId,
        );
        await save({ phase: "selected", stage: "音频处理完成", plans: [plan] });
        return { plans: [plan] };
      }
      const plan = editPlanSchema.parse(params.plan);
      planTimeline(plan);
      const rendered = await renderJob(plan, userId, dir, (stage) => save({ stage }));
      const actual = await probeMedia(rendered.file, 300.2);
      if (Math.abs(actual.duration / SECOND - rendered.duration) > 0.15)
        throw Error("导出时长与剪辑方案不一致，未发布结果");
      await save({ stage: "转存成片" });
      const url = await upload(rendered.file, "mp4"),
        subtitleUrl = await upload(rendered.srt, "srt");
      return {
        url,
        subtitleUrl,
        plan,
        metadata: {
          width: rendered.width,
          height: rendered.height,
          durationSeconds: actual.duration / SECOND,
          storageStatus: "archived",
        },
      };
    } finally {
      await rm(dir, { recursive: true, force: true });
    }
  });
}