WYF-koubo/electron/mapi/runninghub/main.ts

import { ipcMain } from 'electron';
import { RunningHubClient } from './client';
import logger from '../log/main';
import { getElectronSystemConfigSync, fetchElectronSystemConfig, clearElectronConfigCache } from '../systemConfig';
import * as path from 'path';
import * as fs from 'fs';

let clientInstance1: RunningHubClient | null = null;
let clientInstance2: RunningHubClient | null = null;
let clientInstance1Key = '';
let clientInstance2Key = '';

type CloudBeautySessionData = {
  task_id: string;
  video_upload_url?: string;
  audio_upload_url?: string;
  status_url?: string;
  status?: string;
  queue_position?: number | null;
  assigned_node_id?: string | null;
};

type CloudBeautyAuthMode = 'none' | 'legacy' | 'full';

function buildCloudBeautyHeaders(
  apiKey: string,
  extraHeaders: Record<string, string> = {},
  mode: CloudBeautyAuthMode = 'full'
) {
  const headers: Record<string, string> = { ...extraHeaders };
  if (mode === 'none' || !apiKey) {
    return headers;
  }

  headers['X-API-Key'] = apiKey;
  if (mode === 'full') {
    headers['ApiKey'] = apiKey;
    headers['Authorization'] = `Bearer ${apiKey}`;
  }
  return headers;
}

function isCloudBeautyAuthError(error: any) {
  const status = error?.response?.status;
  return status === 401 || status === 403;
}

async function cloudBeautyPostWithFallback(
  axios: any,
  url: string,
  data: any,
  params: {
    apiKey: string;
    extraHeaders?: Record<string, string>;
    timeout?: number;
    label: string;
    modes?: CloudBeautyAuthMode[];
    maxBodyLength?: number;
    maxContentLength?: number;
  }
) {
  const modes = params.modes || ['full', 'legacy'];
  let lastError: any = null;

  for (let i = 0; i < modes.length; i++) {
    const mode = modes[i];
    try {
      return await axios.post(url, data, {
        headers: buildCloudBeautyHeaders(params.apiKey, params.extraHeaders || {}, mode),
        timeout: params.timeout,
        maxBodyLength: params.maxBodyLength,
        maxContentLength: params.maxContentLength,
      });
    } catch (error: any) {
      lastError = error;
      if (!isCloudBeautyAuthError(error) || i === modes.length - 1) {
        throw error;
      }
      logger.warn(`[CloudBeauty] ${params.label} 鉴权模式回退`, {
        url,
        mode,
        status: error?.response?.status,
      });
    }
  }

  throw lastError;
}

async function cloudBeautyGetWithFallback(
  axios: any,
  url: string,
  params: {
    apiKey: string;
    extraHeaders?: Record<string, string>;
    timeout?: number;
    label: string;
    modes?: CloudBeautyAuthMode[];
  }
) {
  const modes = params.modes || ['full', 'legacy'];
  let lastError: any = null;

  for (let i = 0; i < modes.length; i++) {
    const mode = modes[i];
    try {
      return await axios.get(url, {
        headers: buildCloudBeautyHeaders(params.apiKey, params.extraHeaders || {}, mode),
        timeout: params.timeout,
      });
    } catch (error: any) {
      lastError = error;
      if (!isCloudBeautyAuthError(error) || i === modes.length - 1) {
        throw error;
      }
      logger.warn(`[CloudBeauty] ${params.label} 鉴权模式回退`, {
        url,
        mode,
        status: error?.response?.status,
      });
    }
  }

  throw lastError;
}

function unwrapCloudBeautyPayload<T = any>(payload: any): T {
  return (payload?.data || payload || {}) as T;
}

function isCloudBeautyFailureStatus(status?: string) {
  const normalized = String(status || '').toLowerCase();
  return ['failed', 'fail', 'error', 'rejected', 'cancelled'].includes(normalized);
}

async function waitForCloudBeautyUploadUrls(params: {
  axios: any;
  baseUrl: string;
  apiKey: string;
  taskId: string;
  statusUrl?: string;
  maxAttempts?: number;
  intervalMs?: number;
}) {
  const pollUrl = params.statusUrl || `${params.baseUrl}/api/task/status/${params.taskId}`;
  const maxAttempts = params.maxAttempts ?? 120;
  const intervalMs = params.intervalMs ?? 2000;

  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    const resp = await cloudBeautyGetWithFallback(params.axios, pollUrl, {
      apiKey: params.apiKey,
      timeout: 30000,
      label: '查询上传地址',
      modes: ['full', 'legacy'],
    });
    const statusData = unwrapCloudBeautyPayload<CloudBeautySessionData>(resp.data);

    const uploadReady = !!statusData.video_upload_url && !!statusData.audio_upload_url;
    const nodeAssigned = !!statusData.assigned_node_id;

    if (uploadReady && nodeAssigned) {
      return statusData;
    }

    if (isCloudBeautyFailureStatus(statusData.status)) {
      throw new Error(`等待上传地址失败: ${JSON.stringify(statusData)}`);
    }

    logger.info('[CloudBeauty] 上传地址尚未就绪，继续等待', {
      taskId: params.taskId,
      status: statusData.status,
      queuePosition: statusData.queue_position,
      uploadReady,
      assignedNodeId: statusData.assigned_node_id,
      attempt,
      maxAttempts,
    });

    if (attempt < maxAttempts) {
      await new Promise(resolve => setTimeout(resolve, intervalMs));
    }
  }

  throw new Error(`等待上传地址超时: taskId=${params.taskId}`);
}

function normalizeBaseUrl(url: string): string {
  return (url || '').trim().replace(/\/+$/, '');
}

function getMimeType(filePath: string, kind: 'video' | 'audio'): string {
  const ext = path.extname(filePath).toLowerCase();
  if (kind === 'video') {
    if (ext === '.mov') return 'video/quicktime';
    if (ext === '.mkv') return 'video/x-matroska';
    if (ext === '.avi') return 'video/x-msvideo';
    if (ext === '.webm') return 'video/webm';
    return 'video/mp4';
  }

  if (ext === '.wav') return 'audio/wav';
  if (ext === '.m4a') return 'audio/mp4';
  if (ext === '.aac') return 'audio/aac';
  if (ext === '.flac') return 'audio/flac';
  if (ext === '.ogg') return 'audio/ogg';
  return 'audio/mpeg';
}

const cloudBeautyApiModeCache = new Map<string, 'direct' | 'legacy'>();

async function refreshRunningHubConfig() {
  clearElectronConfigCache();
  clientInstance1 = null;
  clientInstance2 = null;
  clientInstance1Key = '';
  clientInstance2Key = '';
  cloudBeautyApiModeCache.clear();
  const config = await fetchElectronSystemConfig();
  logger.info('[RunningHub] 配置已刷新，客户端缓存已清空', {
    workflowId1: config.runninghub_workflow_id_1,
    workflowId2: config.runninghub_workflow_id_2,
    hasApiKey: !!config.runninghub_api_key,
  });
  return {
    success: true,
    workflowId1: config.runninghub_workflow_id_1,
    workflowId2: config.runninghub_workflow_id_2,
    hasApiKey: !!config.runninghub_api_key,
  };
}

async function detectCloudBeautyApiMode(baseUrl: string): Promise<'direct' | 'legacy'> {
  const normalizedBaseUrl = normalizeBaseUrl(baseUrl);
  const cachedMode = cloudBeautyApiModeCache.get(normalizedBaseUrl);
  if (cachedMode) {
    return cachedMode;
  }

  const axios = (await import('axios')).default;
  try {
    const openapiResp = await axios.get(`${normalizedBaseUrl}/openapi.json`, {
      timeout: 10000,
    });

    const paths = openapiResp.data?.paths || {};
    const mode = paths['/api/session/create'] ? 'direct' : 'legacy';
    cloudBeautyApiModeCache.set(normalizedBaseUrl, mode);
    return mode;
  } catch (error: any) {
    const status = error?.response?.status;
    if (status === 404 || status === 405) {
      logger.info('[CloudBeauty] openapi.json unavailable, assume legacy mode', {
        baseUrl: normalizedBaseUrl,
        status,
      });
      cloudBeautyApiModeCache.set(normalizedBaseUrl, 'legacy');
      return 'legacy';
    }

    throw error;
  }
}

async function submitCloudBeautyDirectTask(params: {
  baseUrl: string;
  apiKey: string;
  videoPath: string;
  audioPath: string;
  humanWeight: string;
}) {
  const axios = (await import('axios')).default;
  const FormData = (await import('form-data')).default;

  const sessionBody = new URLSearchParams({
    human_weight: params.humanWeight,
    weight_sync: '0.5',
  });

  logger.info('[CloudBeauty] 使用新接口创建直传会话', {
    baseUrl: params.baseUrl,
    humanWeight: params.humanWeight,
  });

  const sessionResp = await cloudBeautyPostWithFallback(
    axios,
    `${params.baseUrl}/api/session/create`,
    sessionBody.toString(),
    {
      apiKey: params.apiKey,
      extraHeaders: {
        'Content-Type': 'application/x-www-form-urlencoded',
      },
      timeout: 30000,
      label: '创建直传会话',
      modes: ['full', 'legacy'],
    }
  );

  let sessionData = unwrapCloudBeautyPayload<CloudBeautySessionData>(sessionResp.data);
  if (
    sessionData.task_id &&
    (
      !sessionData.video_upload_url ||
      !sessionData.audio_upload_url ||
      !sessionData.assigned_node_id ||
      sessionData.status === 'queued_upload'
    ) &&
    (sessionData.status_url || sessionData.status === 'queued_upload' || typeof sessionData.queue_position === 'number')
  ) {
    logger.info('[CloudBeauty] 直传会话进入排队，等待上传地址', {
      taskId: sessionData.task_id,
      status: sessionData.status,
      queuePosition: sessionData.queue_position,
      assignedNodeId: sessionData.assigned_node_id,
      hasVideoUploadUrl: !!sessionData.video_upload_url,
      hasAudioUploadUrl: !!sessionData.audio_upload_url,
      statusUrl: sessionData.status_url,
    });
    sessionData = await waitForCloudBeautyUploadUrls({
      axios,
      baseUrl: params.baseUrl,
      apiKey: params.apiKey,
      taskId: sessionData.task_id,
      statusUrl: sessionData.status_url,
    });
  }

  if (!sessionData.task_id || !sessionData.video_upload_url || !sessionData.audio_upload_url || !sessionData.assigned_node_id) {
    throw new Error(`新接口返回数据不完整: ${JSON.stringify(sessionResp.data)}`);
  }

  const uploadFile = async (uploadUrl: string, filePath: string, kind: 'video' | 'audio') => {
    logger.info('[CloudBeauty] 直传文件到 worker', {
      kind,
      uploadUrl,
      filePath,
    });

    const createUploadForm = () => {
      const form = new FormData();
      form.append('file', fs.createReadStream(filePath), {
        filename: path.basename(filePath),
        contentType: getMimeType(filePath, kind),
      });
      return form;
    };

    const modes: CloudBeautyAuthMode[] = ['none', 'legacy', 'full'];
    let lastError: any = null;

    for (let i = 0; i < modes.length; i++) {
      const mode = modes[i];
      const form = createUploadForm();
      try {
        const uploadResp = await axios.post(uploadUrl, form, {
          headers: buildCloudBeautyHeaders(params.apiKey, form.getHeaders() as Record<string, string>, mode),
          maxBodyLength: Infinity,
          maxContentLength: Infinity,
          timeout: 10 * 60 * 1000,
        });
        return uploadResp.data;
      } catch (error: any) {
        lastError = error;
        if (!isCloudBeautyAuthError(error) || i === modes.length - 1) {
          throw error;
        }
        logger.warn('[CloudBeauty] worker 上传鉴权模式回退', {
          kind,
          uploadUrl,
          mode,
          status: error?.response?.status,
        });
      }
    }

    throw lastError;
  };

  const videoUploadResult = await uploadFile(sessionData.video_upload_url, params.videoPath, 'video');
  const audioUploadResult = await uploadFile(sessionData.audio_upload_url, params.audioPath, 'audio');

  logger.info('[CloudBeauty] 新接口直传完成', {
    taskId: sessionData.task_id,
    videoUploadResult,
    audioUploadResult,
  });

  return {
    taskId: sessionData.task_id,
    mode: 'direct' as const,
  };
}

async function submitCloudBeautyLegacyTask(params: {
  baseUrl: string;
  apiKey: string;
  videoPath: string;
  audioPath: string;
  humanWeight: string;
  sysCfg: any;
}) {
  const ossConfig = {
    accessKeyId: (params.sysCfg.aliyun_oss_access_key_id || '').trim(),
    accessKeySecret: (params.sysCfg.aliyun_oss_access_key_secret || '').trim(),
    bucket: (params.sysCfg.aliyun_oss_bucket || '').trim(),
    region: (params.sysCfg.aliyun_oss_region || '').trim(),
    endpoint: `https://${(params.sysCfg.aliyun_oss_region || '').trim()}.aliyuncs.com`
  };

  if (!ossConfig.accessKeyId || !ossConfig.bucket) {
    return { success: false, error: 'OSS未配置，无法回退到旧接口模式' };
  }

  logger.info('[CloudBeauty] 回退旧接口 URL 提交模式', {
    bucket: ossConfig.bucket,
    region: ossConfig.region,
    endpoint: ossConfig.endpoint,
  });

  const { uploadToOss } = await import('../aliyun/oss');
  const axios = (await import('axios')).default;
  const FormData = (await import('form-data')).default;

  const videoExt = path.extname(params.videoPath) || '.mp4';
  const audioExt = path.extname(params.audioPath) || '.mp3';
  const timestamp = Date.now();
  const randomStr = Math.random().toString(36).substring(2, 8);

  const videoOssResult = await uploadToOss(params.videoPath, ossConfig, `cloudBeauty/video/${timestamp}_${randomStr}${videoExt}`);
  if (!videoOssResult.success || !videoOssResult.url) {
    return { success: false, error: `视频上传OSS失败: ${videoOssResult.error}` };
  }

  const audioOssResult = await uploadToOss(params.audioPath, ossConfig, `cloudBeauty/audio/${timestamp}_${randomStr}${audioExt}`);
  if (!audioOssResult.success || !audioOssResult.url) {
    return { success: false, error: `音频上传OSS失败: ${audioOssResult.error}` };
  }

  const form = new FormData();
  form.append('video_url', videoOssResult.url);
  form.append('audio_url', audioOssResult.url);
  form.append('human_weight', params.humanWeight);
  form.append('weight_sync', '0.5');

  const resp = await cloudBeautyPostWithFallback(
    axios,
    `${params.baseUrl}/api/task/submit`,
    form,
    {
      apiKey: params.apiKey,
      extraHeaders: form.getHeaders() as Record<string, string>,
      timeout: 30000,
      label: '旧接口提交任务',
      modes: ['full', 'legacy'],
      maxBodyLength: Infinity,
      maxContentLength: Infinity,
    }
  );

  const taskId = resp.data?.data?.task_id;
  if (!taskId) {
    return { success: false, error: '未获取到task_id: ' + JSON.stringify(resp.data) };
  }

  return { success: true, taskId };
}

async function getClientAsync(mode?: string): Promise<RunningHubClient> {
  await fetchElectronSystemConfig();
  const sysCfg = getElectronSystemConfigSync();
  const workflowId1 = sysCfg.runninghub_workflow_id_1 || '2013514129943826433';
  const workflowId2 = sysCfg.runninghub_workflow_id_2 || '2013514129943826433';

  logger.info('[RunningHub] 当前配置', {
    mode,
    workflowId1,
    workflowId2,
    source: sysCfg.runninghub_workflow_id_1 ? '后台' : '默认值'
  });

  const commonNodes = {
    videoNode: '88',
    audioNode: '15',
    durationNode: '60',
    fpsNode: '63',
    widthNode: '67',
    heightNode: '69'
  };

  if (mode === '__CLOUD_1942022968606355458__' || mode === 'mode2') {
    const configKey = `${sysCfg.runninghub_api_key}|${workflowId2}|plus`;
    if (!clientInstance2 || clientInstance2Key !== configKey) {
      clientInstance2 = new RunningHubClient({
        apiKey: sysCfg.runninghub_api_key,
        workflowId: workflowId2,
        apiVersion: 'v2',
        instanceType: 'plus',
        nodes: commonNodes
      });
      clientInstance2Key = configKey;
      logger.info('[RunningHub] 创建v2客户端(mode2)', { workflowId: workflowId2 });
    }
    return clientInstance2;
  }

  const configKey = `${sysCfg.runninghub_api_key}|${workflowId1}|default`;
  if (!clientInstance1 || clientInstance1Key !== configKey) {
    clientInstance1 = new RunningHubClient({
      apiKey: sysCfg.runninghub_api_key,
      workflowId: workflowId1,
      apiVersion: 'v2',
      instanceType: 'default',
      nodes: commonNodes
    });
    clientInstance1Key = configKey;
    logger.info('[RunningHub] 创建v2客户端(mode1)', { workflowId: workflowId1 });
  }
  return clientInstance1;
}

/**
 * 注册RunningHub IPC handlers
 */
export function registerRunningHubHandlers() {

  ipcMain.handle('runninghub:refreshConfig', async () => {
    try {
      return await refreshRunningHubConfig();
    } catch (error: any) {
      logger.error('[RunningHub] 刷新配置失败', error);
      return { success: false, error: error.message };
    }
  });

  // 上传文件
  ipcMain.handle('runninghub:uploadFile', async (event, params) => {
    try {
      const { filePath, fileType, mode } = params;
      logger.info('[RunningHub:IPC] 上传文件', { filePath, fileType, mode });

      const fileName = await (await getClientAsync(mode)).uploadFile(filePath, fileType);
      return { success: true, fileName };
    } catch (error: any) {
      logger.error('[RunningHub:IPC] 上传文件失败', error);
      return { success: false, error: error.message };
    }
  });

  // 创建任务
  ipcMain.handle('runninghub:createTask', async (event, params) => {
    try {
      const { videoFileName, audioFileName, audioDuration, mode } = params;
      logger.info('[RunningHub:IPC] 创建任务', { videoFileName, audioFileName, audioDuration, mode });

      const taskId = await (await getClientAsync(mode)).createTask(videoFileName, audioFileName, audioDuration);
      return { success: true, taskId };
    } catch (error: any) {
      logger.error('[RunningHub:IPC] 创建任务失败', error);
      return { success: false, error: error.message };
    }
  });

  // 查询任务状态
  ipcMain.handle('runninghub:queryTask', async (event, params) => {
    try {
      const { taskId, mode } = params;

      const result = await (await getClientAsync(mode)).queryTask(taskId);
      return { success: true, result };
    } catch (error: any) {
      logger.error('[RunningHub:IPC] 查询任务失败', error);
      return { success: false, error: error.message };
    }
  });

  // 等待任务完成（带轮询）
  ipcMain.handle('runninghub:waitForTaskComplete', async (event, params) => {
    try {
      const { taskId, maxRetries, interval, mode } = params;
      logger.info('[RunningHub:IPC] 等待任务完成', { taskId, maxRetries, mode });

      const result = await (await getClientAsync(mode)).waitForTaskComplete(
        taskId,
        maxRetries,
        interval,
        (status, retryCount) => {
          // 发送进度更新到渲染进程
          event.sender.send('runninghub:task:progress', {
            taskId,
            status,
            retryCount
          });
        }
      );

      return { success: true, result };
    } catch (error: any) {
      logger.error('[RunningHub:IPC] 等待任务完成失败', error);
      return { success: false, error: error.message };
    }
  });

  // 🆕 RunningHub TTS 语音合成
  ipcMain.handle('runninghub:tts:synthesize', async (event, params) => {
    try {
      const { synthesizeSpeech } = await import('./index');
      logger.info('[RunningHub:IPC] 开始TTS语音合成', {
        text: params.text?.substring(0, 50),
        emotion: params.config?.emotion
      });

      const result = await synthesizeSpeech(params.text, params.config);
      return result;
    } catch (error: any) {
      logger.error('[RunningHub:IPC] TTS语音合成失败', error);
      return { success: false, error: error.message };
    }
  });

  // 🆕 获取RunningHub可用情感列表
  ipcMain.handle('runninghub:tts:getEmotions', async (event) => {
    try {
      const { getAvailableEmotions } = await import('./index');
      return {
        success: true,
        emotions: getAvailableEmotions()
      };
    } catch (error: any) {
      logger.error('[RunningHub:IPC] 获取情感列表失败', error);
      return { success: false, error: error.message };
    }
  });

  logger.info('[RunningHub:IPC] Handlers已注册');

  ipcMain.handle('cloudBeauty:submitTask', async (event, params: {
    videoPath: string; audioPath: string; humanWeight: string;
  }) => {
    try {
      const sysCfg = await fetchElectronSystemConfig();
      const baseUrl = normalizeBaseUrl(sysCfg.cloud_beauty_base_url || '');
      const apiKey = sysCfg.cloud_beauty_api_key || '';
      if (!baseUrl || !apiKey) return { success: false, error: '云端美化服务未配置，请确认后台已填写 Base URL 和 API Key' };

      const { videoPath, audioPath, humanWeight } = params;
      logger.info('[CloudBeauty] 提交任务', { videoPath, audioPath, humanWeight });

      const detectedMode = await detectCloudBeautyApiMode(baseUrl);
      logger.info('[CloudBeauty] API mode detected', { baseUrl, detectedMode });
      if (detectedMode === 'direct') {
        const directResult = await submitCloudBeautyDirectTask({
          baseUrl,
          apiKey,
          videoPath,
          audioPath,
          humanWeight,
        });
        return { success: true, taskId: directResult.taskId };
      }

      return await submitCloudBeautyLegacyTask({
        baseUrl,
        apiKey,
        videoPath,
        audioPath,
        humanWeight,
        sysCfg,
      });
    } catch (error: any) {
      logger.error('[CloudBeauty] 提交任务失败', error);
      const errMsg = error.response
        ? `提交失败(${error.response.status}): ${JSON.stringify(error.response.data)}`
        : error.message;
      return { success: false, error: errMsg };
    }
  });

  ipcMain.handle('cloudBeauty:queryStatus', async (event, params: { taskId: string }) => {
    try {
      const sysCfg = await fetchElectronSystemConfig();
      const baseUrl = normalizeBaseUrl(sysCfg.cloud_beauty_base_url || '');

      const axios = (await import('axios')).default;
      const apiKey = sysCfg.cloud_beauty_api_key || '';
      const resp = await cloudBeautyGetWithFallback(axios, `${baseUrl}/api/task/status/${params.taskId}`, {
        apiKey,
        timeout: 30000,
        label: '查询任务状态',
        modes: ['full', 'legacy'],
      });
      const data = unwrapCloudBeautyPayload(resp.data);

      if (data?.download_url && data.download_url.startsWith('/')) {
        data.download_url = baseUrl + data.download_url;
      }

      return { success: true, data };
    } catch (error: any) {
      return { success: false, error: error.message };
    }
  });
}