video-create/.claude/skills/video-from-script/scripts/lib/video-poll-utils.js

/**
 * 共享视频轮询重试工具
 *
 * 提供 pollWithRetry 工厂函数，供 kling/veo/grok 三个视频生成器共用。
 * 两层重试：轮询级（同一 taskId，处理网络瞬断/超时）→ 任务级（仅 API 明确返回 failed 才创建新 task）
 *
 * 铁律：超时 ≠ 失败。超时说明服务端还在跑，创建新任务会重复计费。
 */

const path = require('path')
const fs = require('fs')
const https = require('https')
const http = require('http')

// "fetch failed" = Node fetch threw before receiving a response (network error, NOT a Kling task failure)
const TRANSIENT_RE = /timeout|ECONNRESET|ETIMEDOUT|network|socket|超时|processing|pending|fetch failed/i

const TASK_FAILED_RE = /\bfailed\b|失败/i

const POLL_RETRIES = 5      // 同一 task 轮询重试次数（含超时等待）
const POLL_RETRY_DELAY = 5000  // 轮询重试间隔 ms
const TASK_RETRY_DELAY = 5000  // 任务级重试间隔 ms

function isTransientError(err) {
  return TRANSIENT_RE.test(err.message || '')
}

function isTaskFailed(err) {
  return TASK_FAILED_RE.test(err.message || '')
}

async function download(url, outputPath) {
  const protocol = url.startsWith('https') ? https : http
  return new Promise((resolve, reject) => {
    const file = fs.createWriteStream(outputPath)
    protocol.get(url, (response) => {
      if (response.statusCode >= 300 && response.statusCode < 400 && response.headers.location) {
        file.close()
        fs.unlinkSync(outputPath)
        return download(response.headers.location, outputPath).then(resolve).catch(reject)
      }
      response.pipe(file)
      file.on('finish', () => { file.close(); resolve(outputPath) })
    }).on('error', (err) => {
      file.close()
      if (fs.existsSync(outputPath)) fs.unlinkSync(outputPath)
      reject(err)
    })
  })
}

/**
 * 创建 pollWithRetry 函数
 *
 * @param {object} opts
 * @param {object} opts.Api - 有 create() 和 poll() 方法的 API 对象
 * @param {string} opts.suffix - 输出文件后缀（如 '_kling'）
 * @param {number} opts.duration - 视频时长（秒）
 * @param {number} [opts.maxRetries=3] - 任务级最大重试次数
 * @param {function} [opts.optimizePrompt] - 提示词优化函数 (prompt, failReason, attempt) => optimizedPrompt
 * @param {function} opts.buildCreateOpts - (item_options) => create() 的第三个参数
 * @returns {function} pollWithRetry(taskId, prompt, options)
 */
function makePollWithRetry({ Api, suffix, duration, maxRetries = 3, optimizePrompt, buildCreateOpts }) {
  return async function pollWithRetry(taskId, prompt, options = {}) {
    let currentTaskId = taskId
    let currentPrompt = prompt
    let lastError = null
    let lastErrorWasTaskFailure = false

    for (let attempt = 0; attempt <= maxRetries; attempt++) {
      if (attempt > 0) {
        // 只有 API 明确返回 failed 才创建新任务。超时/网络问题禁止创建新任务（原任务仍在服务端运行，重复计费）
        if (!lastErrorWasTaskFailure) {
          const err = new Error(`视频生成超时，放弃等待: ${lastError}`)
          err.isTaskFailure = false
          throw err
        }
        if (optimizePrompt) {
          currentPrompt = optimizePrompt(prompt, lastError, attempt)
        }
        console.log(`\n   🔄 重试 (任务 ${currentTaskId.substring(0, 8)}...): ${currentPrompt.substring(0, 50)}`)
        const createOpts = buildCreateOpts(options)
        currentTaskId = await Api.create(options.imageUrl || '', currentPrompt, createOpts)
      }

      const outputDir = options.outputDir || './output'

      for (let pollAttempt = 0; pollAttempt <= POLL_RETRIES; pollAttempt++) {
        try {
          const result = await Api.poll(currentTaskId)

          const timestamp = new Date().toISOString().replace(/[:.]/g, '-')
          const videoFile = path.join(outputDir, `${timestamp}${suffix}.mp4`)
          await download(result.videoUrl, videoFile)

          return {
            taskId: currentTaskId,
            prompt: currentPrompt,
            originalPrompt: prompt,
            attempts: attempt + 1,
            file: videoFile,
            files: [videoFile],
            duration,
          }
        } catch (err) {
          lastError = err.message
          // Network errors (transient) must never be treated as task failures.
          // isTaskFailed("fetch failed") would be true without this guard because "failed" matches TASK_FAILED_RE.
          const transient = isTransientError(err)
          lastErrorWasTaskFailure = !transient && isTaskFailed(err)
          if (transient && pollAttempt < POLL_RETRIES) {
            const tag = isTaskFailed(err) ? '失败' : '超时/瞬断'
            console.log(`   ⚠ 轮询${tag} (${pollAttempt + 1}/${POLL_RETRIES}): ${err.message.slice(0, 60)}`)
            await new Promise(r => setTimeout(r, POLL_RETRY_DELAY))
            continue
          }
          break
        }
      }

      if (attempt < maxRetries && lastErrorWasTaskFailure) {
        await new Promise(r => setTimeout(r, TASK_RETRY_DELAY))
      }
    }

    const finalErr = new Error(`重试 ${maxRetries} 次后仍失败: ${lastError}`)
    finalErr.isTaskFailure = true
    throw finalErr
  }
}

module.exports = { makePollWithRetry, isTaskFailed, POLL_RETRIES, POLL_RETRY_DELAY, TASK_RETRY_DELAY }