feat: openai账号401自动停止调度

2026-01-22 16:43:35 +00:00 · 2025-09-24 11:35:48 +08:00
parent 00faa21e4b
commit f56d1edce0
5 changed files with 364 additions and 14 deletions
--- a/src/routes/openaiRoutes.js
+++ b/src/routes/openaiRoutes.js
@@ -135,6 +135,13 @@ async function getOpenAIAuthToken(apiKeyData, sessionId = null, requestedModel =
 // 主处理函数，供两个路由共享
 const handleResponses = async (req, res) => {
  let upstream = null
+  let accountId = null
+  let accountType = 'openai'
+  let sessionHash = null
+  let account = null
+  let proxy = null
+  let accessToken = null
+
  try {
    // 从中间件获取 API Key 数据
    const apiKeyData = req.apiKey || {}
@@ -147,6 +154,8 @@ const handleResponses = async (req, res) => {
      req.body?.conversation_id ||
      null

+    sessionHash = sessionId ? crypto.createHash('sha256').update(sessionId).digest('hex') : null
+
    // 从请求体中提取模型和流式标志
    let requestedModel = req.body?.model || null

@@ -191,14 +200,11 @@ const handleResponses = async (req, res) => {
    }

    // 使用调度器选择账户
-    const {
-      accessToken,
-      accountId,
-      accountName: _accountName,
-      accountType,
-      proxy,
-      account
-    } = await getOpenAIAuthToken(apiKeyData, sessionId, requestedModel)
+    ;({ accessToken, accountId, accountType, proxy, account } = await getOpenAIAuthToken(
+      apiKeyData,
+      sessionId,
+      requestedModel
+    ))

    // 如果是 OpenAI-Responses 账户，使用专门的中继服务处理
    if (accountType === 'openai-responses') {
@@ -312,7 +318,7 @@ const handleResponses = async (req, res) => {
      await unifiedOpenAIScheduler.markAccountRateLimited(
        accountId,
        'openai',
-        sessionId ? crypto.createHash('sha256').update(sessionId).digest('hex') : null,
+        sessionHash,
        resetsInSeconds
      )

@@ -337,6 +343,77 @@ const handleResponses = async (req, res) => {
        res.status(429).json(errorResponse)
      }

+      return
+    } else if (upstream.status === 401) {
+      logger.warn(`🔐 Unauthorized error detected for OpenAI account ${accountId} (Codex API)`)
+
+      let errorData = null
+
+      try {
+        if (isStream && upstream.data && typeof upstream.data.on === 'function') {
+          const chunks = []
+          await new Promise((resolve, reject) => {
+            upstream.data.on('data', (chunk) => chunks.push(chunk))
+            upstream.data.on('end', resolve)
+            upstream.data.on('error', reject)
+            setTimeout(resolve, 5000)
+          })
+
+          const fullResponse = Buffer.concat(chunks).toString()
+          try {
+            errorData = JSON.parse(fullResponse)
+          } catch (parseError) {
+            logger.error('Failed to parse 401 error response:', parseError)
+            logger.debug('Raw 401 response:', fullResponse)
+            errorData = { error: { message: fullResponse || 'Unauthorized' } }
+          }
+        } else {
+          errorData = upstream.data
+        }
+      } catch (parseError) {
+        logger.error('⚠️ Failed to handle 401 error response:', parseError)
+      }
+
+      let reason = 'OpenAI账号认证失败（401错误）'
+      if (errorData) {
+        const messageCandidate =
+          errorData.error &&
+          typeof errorData.error.message === 'string' &&
+          errorData.error.message.trim()
+            ? errorData.error.message.trim()
+            : typeof errorData.message === 'string' && errorData.message.trim()
+              ? errorData.message.trim()
+              : null
+        if (messageCandidate) {
+          reason = `OpenAI账号认证失败（401错误）：${messageCandidate}`
+        }
+      }
+
+      try {
+        await unifiedOpenAIScheduler.markAccountUnauthorized(
+          accountId,
+          'openai',
+          sessionHash,
+          reason
+        )
+      } catch (markError) {
+        logger.error('❌ Failed to mark OpenAI account unauthorized after 401:', markError)
+      }
+
+      let errorResponse = errorData
+      if (!errorResponse || typeof errorResponse !== 'object' || Buffer.isBuffer(errorResponse)) {
+        const fallbackMessage =
+          typeof errorData === 'string' && errorData.trim() ? errorData.trim() : 'Unauthorized'
+        errorResponse = {
+          error: {
+            message: fallbackMessage,
+            type: 'unauthorized',
+            code: 'unauthorized'
+          }
+        }
+      }
+
+      res.status(401).json(errorResponse)
      return
    } else if (upstream.status === 200 || upstream.status === 201) {
      // 请求成功，检查并移除限流状态
@@ -553,7 +630,7 @@ const handleResponses = async (req, res) => {
        await unifiedOpenAIScheduler.markAccountRateLimited(
          accountId,
          'openai',
-          sessionId ? crypto.createHash('sha256').update(sessionId).digest('hex') : null,
+          sessionHash,
          rateLimitResetsInSeconds
        )
      } else if (upstream.status === 200) {
@@ -594,9 +671,51 @@ const handleResponses = async (req, res) => {
    logger.error('Proxy to ChatGPT codex/responses failed:', error)
    // 优先使用主动设置的 statusCode，然后是上游响应的状态码，最后默认 500
    const status = error.statusCode || error.response?.status || 500
-    const message = error.response?.data || error.message || 'Internal server error'
+
+    if (status === 401 && accountId) {
+      let reason = 'OpenAI账号认证失败（401错误）'
+      const errorData = error.response?.data
+      if (errorData) {
+        if (typeof errorData === 'string' && errorData.trim()) {
+          reason = `OpenAI账号认证失败（401错误）：${errorData.trim()}`
+        } else if (
+          errorData.error &&
+          typeof errorData.error.message === 'string' &&
+          errorData.error.message.trim()
+        ) {
+          reason = `OpenAI账号认证失败（401错误）：${errorData.error.message.trim()}`
+        } else if (typeof errorData.message === 'string' && errorData.message.trim()) {
+          reason = `OpenAI账号认证失败（401错误）：${errorData.message.trim()}`
+        }
+      } else if (error.message) {
+        reason = `OpenAI账号认证失败（401错误）：${error.message}`
+      }
+
+      try {
+        await unifiedOpenAIScheduler.markAccountUnauthorized(
+          accountId,
+          accountType || 'openai',
+          sessionHash,
+          reason
+        )
+      } catch (markError) {
+        logger.error('❌ Failed to mark OpenAI account unauthorized in catch handler:', markError)
+      }
+    }
+
+    let responsePayload = error.response?.data
+    if (!responsePayload) {
+      responsePayload = { error: { message: error.message || 'Internal server error' } }
+    } else if (typeof responsePayload === 'string') {
+      responsePayload = { error: { message: responsePayload } }
+    } else if (typeof responsePayload === 'object' && !responsePayload.error) {
+      responsePayload = {
+        error: { message: responsePayload.message || error.message || 'Internal server error' }
+      }
+    }
+
    if (!res.headersSent) {
-      res.status(status).json({ error: { message } })
+      res.status(status).json(responsePayload)
    }
  }
 }
--- a/src/services/openaiAccountService.js
+++ b/src/services/openaiAccountService.js
@@ -865,6 +865,49 @@ async function setAccountRateLimited(accountId, isLimited, resetsInSeconds = nul
  }
 }

+// 🚫 标记账户为未授权状态（401错误）
+async function markAccountUnauthorized(accountId, reason = 'OpenAI账号认证失败（401错误）') {
+  const account = await getAccount(accountId)
+  if (!account) {
+    throw new Error('Account not found')
+  }
+
+  const now = new Date().toISOString()
+  const currentCount = parseInt(account.unauthorizedCount || '0', 10)
+  const unauthorizedCount = Number.isFinite(currentCount) ? currentCount + 1 : 1
+
+  const updates = {
+    status: 'unauthorized',
+    schedulable: 'false',
+    errorMessage: reason,
+    unauthorizedAt: now,
+    unauthorizedCount: unauthorizedCount.toString()
+  }
+
+  await updateAccount(accountId, updates)
+  logger.warn(
+    `🚫 Marked OpenAI account ${account.name || accountId} as unauthorized due to 401 error`
+  )
+
+  try {
+    const webhookNotifier = require('../utils/webhookNotifier')
+    await webhookNotifier.sendAccountAnomalyNotification({
+      accountId,
+      accountName: account.name || accountId,
+      platform: 'openai',
+      status: 'unauthorized',
+      errorCode: 'OPENAI_UNAUTHORIZED',
+      reason,
+      timestamp: now
+    })
+    logger.info(
+      `📢 Webhook notification sent for OpenAI account ${account.name} unauthorized state`
+    )
+  } catch (webhookError) {
+    logger.error('Failed to send unauthorized webhook notification:', webhookError)
+  }
+}
+
 // 🔄 重置账户所有异常状态
 async function resetAccountStatus(accountId) {
  const account = await getAccount(accountId)
@@ -1001,6 +1044,7 @@ module.exports = {
  refreshAccountToken,
  isTokenExpired,
  setAccountRateLimited,
+  markAccountUnauthorized,
  resetAccountStatus,
  toggleSchedulable,
  getAccountRateLimitInfo,
--- a/src/services/openaiResponsesAccountService.js
+++ b/src/services/openaiResponsesAccountService.js
@@ -293,6 +293,48 @@ class OpenAIResponsesAccountService {
    )
  }

+  // 🚫 标记账户为未授权状态（401错误）
+  async markAccountUnauthorized(accountId, reason = 'OpenAI Responses账号认证失败（401错误）') {
+    const account = await this.getAccount(accountId)
+    if (!account) {
+      return
+    }
+
+    const now = new Date().toISOString()
+    const currentCount = parseInt(account.unauthorizedCount || '0', 10)
+    const unauthorizedCount = Number.isFinite(currentCount) ? currentCount + 1 : 1
+
+    await this.updateAccount(accountId, {
+      status: 'unauthorized',
+      schedulable: 'false',
+      errorMessage: reason,
+      unauthorizedAt: now,
+      unauthorizedCount: unauthorizedCount.toString()
+    })
+
+    logger.warn(
+      `🚫 OpenAI-Responses account ${account.name || accountId} marked as unauthorized due to 401 error`
+    )
+
+    try {
+      const webhookNotifier = require('../utils/webhookNotifier')
+      await webhookNotifier.sendAccountAnomalyNotification({
+        accountId,
+        accountName: account.name || accountId,
+        platform: 'openai',
+        status: 'unauthorized',
+        errorCode: 'OPENAI_UNAUTHORIZED',
+        reason,
+        timestamp: now
+      })
+      logger.info(
+        `📢 Webhook notification sent for OpenAI-Responses account ${account.name || accountId} unauthorized state`
+      )
+    } catch (webhookError) {
+      logger.error('Failed to send unauthorized webhook notification:', webhookError)
+    }
+  }
+
  // 检查并清除过期的限流状态
  async checkAndClearRateLimit(accountId) {
    const account = await this.getAccount(accountId)
--- a/src/services/openaiResponsesRelayService.js
+++ b/src/services/openaiResponsesRelayService.js
@@ -169,6 +169,61 @@ class OpenAIResponsesRelayService {
          errorData
        })

+        if (response.status === 401) {
+          let reason = 'OpenAI Responses账号认证失败（401错误）'
+          if (errorData) {
+            if (typeof errorData === 'string' && errorData.trim()) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.trim()}`
+            } else if (
+              errorData.error &&
+              typeof errorData.error.message === 'string' &&
+              errorData.error.message.trim()
+            ) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.error.message.trim()}`
+            } else if (typeof errorData.message === 'string' && errorData.message.trim()) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.message.trim()}`
+            }
+          }
+
+          try {
+            await unifiedOpenAIScheduler.markAccountUnauthorized(
+              account.id,
+              'openai-responses',
+              sessionHash,
+              reason
+            )
+          } catch (markError) {
+            logger.error(
+              '❌ Failed to mark OpenAI-Responses account unauthorized after 401:',
+              markError
+            )
+          }
+
+          let unauthorizedResponse = errorData
+          if (
+            !unauthorizedResponse ||
+            typeof unauthorizedResponse !== 'object' ||
+            unauthorizedResponse.pipe ||
+            Buffer.isBuffer(unauthorizedResponse)
+          ) {
+            const fallbackMessage =
+              typeof errorData === 'string' && errorData.trim() ? errorData.trim() : 'Unauthorized'
+            unauthorizedResponse = {
+              error: {
+                message: fallbackMessage,
+                type: 'unauthorized',
+                code: 'unauthorized'
+              }
+            }
+          }
+
+          // 清理监听器
+          req.removeListener('close', handleClientDisconnect)
+          res.removeListener('close', handleClientDisconnect)
+
+          return res.status(401).json(unauthorizedResponse)
+        }
+
        // 清理监听器
        req.removeListener('close', handleClientDisconnect)
        res.removeListener('close', handleClientDisconnect)
@@ -250,6 +305,57 @@ class OpenAIResponsesRelayService {
          }
        }

+        if (status === 401) {
+          let reason = 'OpenAI Responses账号认证失败（401错误）'
+          if (errorData) {
+            if (typeof errorData === 'string' && errorData.trim()) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.trim()}`
+            } else if (
+              errorData.error &&
+              typeof errorData.error.message === 'string' &&
+              errorData.error.message.trim()
+            ) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.error.message.trim()}`
+            } else if (typeof errorData.message === 'string' && errorData.message.trim()) {
+              reason = `OpenAI Responses账号认证失败（401错误）：${errorData.message.trim()}`
+            }
+          }
+
+          try {
+            await unifiedOpenAIScheduler.markAccountUnauthorized(
+              account.id,
+              'openai-responses',
+              sessionHash,
+              reason
+            )
+          } catch (markError) {
+            logger.error(
+              '❌ Failed to mark OpenAI-Responses account unauthorized in catch handler:',
+              markError
+            )
+          }
+
+          let unauthorizedResponse = errorData
+          if (
+            !unauthorizedResponse ||
+            typeof unauthorizedResponse !== 'object' ||
+            unauthorizedResponse.pipe ||
+            Buffer.isBuffer(unauthorizedResponse)
+          ) {
+            const fallbackMessage =
+              typeof errorData === 'string' && errorData.trim() ? errorData.trim() : 'Unauthorized'
+            unauthorizedResponse = {
+              error: {
+                message: fallbackMessage,
+                type: 'unauthorized',
+                code: 'unauthorized'
+              }
+            }
+          }
+
+          return res.status(401).json(unauthorizedResponse)
+        }
+
        return res.status(status).json(errorData)
      }

--- a/src/services/unifiedOpenAIScheduler.js
+++ b/src/services/unifiedOpenAIScheduler.js
@@ -356,7 +356,12 @@ class UnifiedOpenAIScheduler {
    try {
      if (accountType === 'openai') {
        const account = await openaiAccountService.getAccount(accountId)
-        if (!account || !account.isActive || account.status === 'error') {
+        if (
+          !account ||
+          !account.isActive ||
+          account.status === 'error' ||
+          account.status === 'unauthorized'
+        ) {
          return false
        }
        // 检查是否可调度
@@ -370,7 +375,8 @@ class UnifiedOpenAIScheduler {
        if (
          !account ||
          (account.isActive !== true && account.isActive !== 'true') ||
-          account.status === 'error'
+          account.status === 'error' ||
+          account.status === 'unauthorized'
        ) {
          return false
        }
@@ -500,6 +506,39 @@ class UnifiedOpenAIScheduler {
    }
  }

+  // 🚫 标记账户为未授权状态
+  async markAccountUnauthorized(
+    accountId,
+    accountType,
+    sessionHash = null,
+    reason = 'OpenAI账号认证失败（401错误）'
+  ) {
+    try {
+      if (accountType === 'openai') {
+        await openaiAccountService.markAccountUnauthorized(accountId, reason)
+      } else if (accountType === 'openai-responses') {
+        await openaiResponsesAccountService.markAccountUnauthorized(accountId, reason)
+      } else {
+        logger.warn(
+          `⚠️ Unsupported account type ${accountType} when marking unauthorized for account ${accountId}`
+        )
+        return { success: false }
+      }
+
+      if (sessionHash) {
+        await this._deleteSessionMapping(sessionHash)
+      }
+
+      return { success: true }
+    } catch (error) {
+      logger.error(
+        `❌ Failed to mark account as unauthorized: ${accountId} (${accountType})`,
+        error
+      )
+      throw error
+    }
+  }
+
  // ✅ 移除账户的限流状态
  async removeAccountRateLimit(accountId, accountType) {
    try {