// lib/api.mjs — 网关 HTTP 封装(chat / images / 异步 task 轮询 / 文件下载) // 零依赖,仅用 Node 内置 fetch(Node >= 18)。 import { loadConfig, requireToken, mask } from './config.mjs'; /** 通道未开通/上游不可用时抛出,携带可操作指引 */ export class ProviderUnavailableError extends Error { constructor(message, { endpoint, status, hint, cause } = {}) { super(message); this.name = 'ProviderUnavailableError'; this.endpoint = endpoint; this.status = status; this.hint = hint; if (cause) this.cause = cause; } } /** 网关返回了业务错误(参数错/额度等) */ export class GatewayError extends Error { constructor(message, { endpoint, status, body } = {}) { super(message); this.name = 'GatewayError'; this.endpoint = endpoint; this.status = status; this.body = body; } } const DEFAULT_TIMEOUT = 120_000; function joinUrl(base, path) { return `${String(base).replace(/\/+$/, '')}/${String(path).replace(/^\/+/, '')}`; } /** * 底层请求。自动带 token、超时、把网关错误翻译成可读异常。 * @returns {Promise<{status:number, headers:Headers, body:any, text:string}>} */ export async function request(path, opts = {}) { const { method = 'GET', body, timeout = DEFAULT_TIMEOUT, base, // 覆盖 base(默认 config.baseUrl) headers = {}, expectJson = true, raw = false, retries = 0, retryDelay = 1500, } = opts; const cfg = loadConfig(); const token = requireToken(); const url = /^https?:\/\//.test(path) ? path : joinUrl(base || cfg.baseUrl, path); let lastErr; for (let attempt = 0; attempt <= retries; attempt++) { if (attempt > 0) { await sleep(retryDelay * 2 ** (attempt - 1)); } try { const res = await fetch(url, { method, headers: { Authorization: `Bearer ${token}`, Accept: 'application/json', ...(body !== undefined ? { 'Content-Type': 'application/json' } : {}), ...headers, }, body: body !== undefined ? (typeof body === 'string' ? body : JSON.stringify(body)) : undefined, signal: AbortSignal.timeout(timeout), }); const text = raw ? '' : await res.text(); let parsed = null; if (expectJson && text) { try { parsed = JSON.parse(text); } catch { parsed = null; } } if (!res.ok) { const err = classifyHttpError(res.status, text, url, parsed); // 5xx / 429 可重试 if ((res.status >= 500 || res.status === 429) && attempt < retries) { lastErr = err; continue; } throw err; } return { status: res.status, headers: res.headers, body: parsed, text }; } catch (e) { if (e instanceof ProviderUnavailableError || e instanceof GatewayError) throw e; lastErr = new GatewayError(`${e.name === 'TimeoutError' ? '请求超时' : '网络错误'}: ${e.message}`, { endpoint: url, }); if (attempt >= retries) throw lastErr; } } throw lastErr; } /** 把 HTTP 状态码 / 网关错误体翻译成「能照着做」的错误 */ function classifyHttpError(status, text, url, parsed) { const raw = (parsed && (parsed.error?.message || parsed.message)) || text.slice(0, 300); const lower = String(raw).toLowerCase(); // 渠道未开通的典型特征 const unavailable = lower.includes('no available channel') || lower.includes('model_not_found') || lower.includes('channel') && lower.includes('not') || lower.includes('task request failed') || lower.includes('upstream') && (status === 404 || status === 502 || status === 503) || lower.includes('not implemented'); const hint = status === 401 || status === 403 ? '凭据无效或权限不足:检查 FEME_API_TOKEN / /opt/data/fmode-identity.json 的 newapi_token' : unavailable ? '该模型/通道在当前网关未开通。可用 `fmode-mv providers` 查看替代通道,或在 provider 配置里改换模型名。' : status === 429 ? '触发限流,稍后重试或降低并发' : undefined; const message = `网关 ${status} @ ${url}\n ${raw}` + (hint ? `\n → ${hint}` : ''); if (unavailable) { return new ProviderUnavailableError(message, { endpoint: url, status, hint }); } return new GatewayError(message, { endpoint: url, status, body: raw }); } export function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); } /** 下载远程文件为 Buffer */ export async function download(url, { timeout = 180_000 } = {}) { const res = await fetch(url, { signal: AbortSignal.timeout(timeout) }); if (!res.ok) throw new GatewayError(`下载失败 ${res.status}: ${url}`); return Buffer.from(await res.arrayBuffer()); } // ==================== Chat ==================== /** * 文本对话(作曲填词 / 分镜 / 口播稿都走这里)。 * @param {{model?:string, messages:Array, system?:string, temperature?:number, max_tokens?:number, json?:boolean, timeout?:number}} o */ export async function chat(o = {}) { const { model = process.env.FEME_CHAT_MODEL || 'deepseek-v4-flash', messages = [], system, temperature = 0.8, max_tokens = 4096, json = false, timeout = 180_000, retries = 1, } = o; const msgs = system ? [{ role: 'system', content: system }, ...messages] : messages; const body = { model, messages: msgs, temperature, max_tokens, stream: false, }; if (json) body.response_format = { type: 'json_object' }; const res = await request('/chat/completions', { method: 'POST', body, timeout, retries, }); const choice = res.body?.choices?.[0]; // 部分推理模型(如 deepseek-v4)会把内容放在 reasoning_content,content 可能为空 let content = choice?.message?.content; if ((content == null || content === '') && choice?.message?.reasoning_content) { if (json) { // 尝试从 reasoning 里抠出 JSON const m = String(choice.message.reasoning_content).match(/\{[\s\S]*\}/); content = m ? m[0] : choice.message.reasoning_content; } else { content = choice.message.reasoning_content; } } if (content == null) { throw new GatewayError(`chat 返回为空(model=${model})`, { body: res.body }); } return { content: String(content), model: res.body?.model || model, usage: res.body?.usage, finishReason: choice?.finish_reason || null, truncated: choice?.finish_reason === 'length', }; } /** * 让模型产出 JSON 并解析。 * * 现实里有两类坑,这里都自动处理: * 1) 推理型模型(如 deepseek-v4-flash)会把大量 token 花在推理上, * 导致 max_tokens 被吃满、JSON 被截断 —— finish_reason === 'length'。 * → 自动把 max_tokens 翻倍重试,最多升到上限。 * 2) 输出带 markdown 围栏 / 前后废话 / 轻微格式错误。 * → extractJSON 已能剥离围栏;仍失败则带着错误提示让模型重出一次。 */ export async function chatJSON(o = {}) { const { model = process.env.FEME_CHAT_MODEL || 'deepseek-v4-flash', messages = [], system, temperature = 0.7, max_tokens = 4096, maxTokensCap = 32768, escalationSteps = 2, repairAttempts = 1, timeout = 300_000, } = o; const msgs = system ? [{ role: 'system', content: system }, ...messages] : [...messages]; let budget = max_tokens; let lastText = ''; let lastErr = null; for (let step = 0; step <= escalationSteps; step++) { const r = await chat({ model, messages: msgs, temperature, max_tokens: budget, json: true, timeout }); lastText = r.content; const parsed = extractJSON(r.content); if (parsed !== null) return { ...r, json: parsed, maxTokensUsed: budget }; // 截断 → 加大预算重来 if (r.truncated && budget < maxTokensCap) { budget = Math.min(maxTokensCap, budget * 2); continue; } lastErr = r; break; } // 修复式重试:把失败输出回喂,要求只吐合法 JSON for (let i = 0; i < repairAttempts; i++) { const tail = lastText.slice(-1200); const r = await chat({ model, messages: [ ...msgs, { role: 'assistant', content: tail }, { role: 'user', content: '上面的 JSON 不完整或格式非法,无法解析。请重新输出**完整且合法**的 JSON:' + '不要省略任何数组元素、不要写 markdown 围栏、不要在 JSON 前后添加任何解释文字。' + '如果内容太长,请压缩每个字段的文字长度,但必须保证 JSON 结构完整闭合。', }, ], temperature: 0.3, max_tokens: maxTokensCap, json: true, timeout, }); lastText = r.content; const parsed = extractJSON(r.content); if (parsed !== null) return { ...r, json: parsed, maxTokensUsed: maxTokensCap, repaired: true }; } const hint = lastErr?.truncated || lastText.length > 3000 ? '\n → 疑似输出被 max_tokens 截断。可提高 --max-tokens,或减少镜头数(--shots),或换用非推理型模型(FEME_CHAT_MODEL=doubao-seed-2-1-pro)。' : ''; throw new GatewayError( `模型未返回可解析的 JSON(model=${model},已尝试升配与修复)。输出尾部片段:\n${lastText.slice(-400)}${hint}` ); } /** 从可能带 ```json 围栏或前后缀说明的文本中抠出 JSON */ export function extractJSON(text) { if (!text) return null; const s = String(text).trim(); const tryParse = (t) => { try { return JSON.parse(t); } catch { return null; } }; let v = tryParse(s); if (v !== null) return v; const fence = s.match(/```(?:json)?\s*([\s\S]*?)```/i); if (fence) { v = tryParse(fence[1].trim()); if (v !== null) return v; } const obj = s.match(/[{[][\s\S]*[}\]]/); if (obj) { v = tryParse(obj[0]); if (v !== null) return v; } return null; } // ==================== Images ==================== /** * 文生图 / 图生图(关键帧)。 * @param {{prompt:string, model?:string, size?:string, n?:number, refImage?:string|Buffer, timeout?:number}} o * @returns {Promise<{buffers:Buffer[], model:string, size:string}>} */ export async function generateImage(o = {}) { const { prompt, model = process.env.FEME_IMAGE_MODEL || 'gpt-image-2.5-sunburst', size = '1024x1024', n = 1, refImage, // 参考图(URL 或本地路径/base64)—— 用于人物一致性 timeout = 300_000, retries = 1, } = o; if (!prompt) throw new GatewayError('generateImage 需要 prompt'); const payload = { model, prompt, n, size }; if (refImage) payload.image = await toImageInput(refImage); // 兼容 OpenAI 风格图生图 const res = await request('/images/generations', { method: 'POST', body: payload, timeout, retries, }); const items = res.body?.data || []; const buffers = []; for (const it of items) { if (it.b64_json) buffers.push(Buffer.from(it.b64_json, 'base64')); else if (it.url) buffers.push(await download(it.url)); } if (!buffers.length) { throw new GatewayError(`图片通道未返回图像数据(model=${model})`, { body: JSON.stringify(res.body).slice(0, 300) }); } return { buffers, model: res.body?.model || model, size }; } async function toImageInput(ref) { if (Buffer.isBuffer(ref)) return `data:image/png;base64,${ref.toString('base64')}`; const s = String(ref); if (/^https?:\/\//.test(s) || s.startsWith('data:')) return s; const { readFileSync } = await import('node:fs'); const buf = readFileSync(s); const ext = s.toLowerCase().endsWith('.jpg') || s.toLowerCase().endsWith('.jpeg') ? 'jpeg' : 'png'; return `data:image/${ext};base64,${buf.toString('base64')}`; } // ==================== 异步任务(音乐 / 视频 / 唇形) ==================== /** * 提交一个异步任务并轮询至完成。 * 不同通道的字段名差异用 pick 函数吸收。 * * @param {object} o * @param {string} o.submitPath 提交路径 * @param {string} o.fetchPath 查询路径(返回单任务) * @param {string} o.listPath 查询路径(返回任务列表,多数通道用这个) * @param {object} o.payload 提交体 * @param {number} o.interval 轮询间隔 ms * @param {number} o.maxWait 最长等待 ms * @param {(t:any)=>({status:string, progress?:number, result?:any, error?:string})} o.pick 状态解析 */ export async function runAsyncTask(o) { const { submitPath, fetchPath, listPath, payload, interval = 5000, maxWait = 600_000, pick, submitBase, fetchBase, taskLabel = 'task', onTick, } = o; const cfg = loadConfig(); const submitRes = await request(submitPath, { method: 'POST', body: payload, base: submitBase || cfg.rootUrl, timeout: 60_000, retries: 0, }); const taskId = submitRes.body?.data ?? submitRes.body?.id ?? submitRes.body?.task_id ?? submitRes.body?.taskId ?? submitRes.body?.data?.task_id ?? submitRes.body?.data?.id; if (!taskId) { throw new GatewayError(`提交${taskLabel}未返回任务 ID`, { body: JSON.stringify(submitRes.body).slice(0, 300), }); } const started = Date.now(); let last = null; while (Date.now() - started < maxWait) { await sleep(interval); const id = typeof taskId === 'object' ? taskId.id : taskId; const res = listPath ? await request(listPath, { method: 'POST', body: { ids: [id] }, base: fetchBase || cfg.rootUrl, timeout: 60_000, }) : await request(`${fetchPath}?ids=${encodeURIComponent(id)}`, { base: fetchBase || cfg.rootUrl, timeout: 60_000, }); const raw = listPath ? res.body?.data?.[0] ?? res.body?.data ?? res.body : res.body?.data ?? res.body; const st = pick ? pick(raw) : { status: normalStatus(raw?.status) }; last = st; if (onTick) onTick({ ...st, taskId: id, elapsed: Date.now() - started }); if (st.status === 'succeeded' || st.status === 'complete' || st.status === 'completed') { return { taskId: id, result: st.result ?? raw, elapsed: Date.now() - started }; } if (st.status === 'failed' || st.status === 'error') { throw new GatewayError(`${taskLabel}执行失败:${st.error || '未知原因'}`, { body: JSON.stringify(raw).slice(0, 500), }); } } throw new GatewayError( `${taskLabel}超时(${Math.round(maxWait / 1000)}s 未完成,最后状态 ${last?.status || 'unknown'},taskId=${last?.taskId || '-'})` ); } function normalStatus(s) { const v = String(s || '').toLowerCase(); if (['success', 'succeeded', 'succeed', 'complete', 'completed', 'done'].includes(v)) return 'succeeded'; if (['fail', 'failed', 'error', 'cancelled', 'canceled'].includes(v)) return 'failed'; if (!v) return 'unknown'; return 'running'; }