ic-ctbu-backend/lib/Lepao/qgOutboundAxios.js

/**
 * 与 lepaoSchoolHttp 一致的青果 HTTP 出站：HttpsProxyAgent + 先代理（短超时）再可回退直连。
 * 供电费、选课书单等非 Worker 场景的对外 GET/POST 使用。
 */
const axios = require('axios')
const QgProxyManager = require('./QgProxyManager')
const {
    buildAxiosOutboundConfig
} = require('./outboundAxiosConfig')

const PROXY_FIRST_TIMEOUT_MS = 20000

function sleep(ms) {
    return new Promise(r => setTimeout(r, ms))
}

async function getOutboundWithBackoff(qgOpts, rounds = 2) {
    let lastErr
    for (let i = 0; i < rounds; i++) {
        try {
            if (i > 0) await sleep(380 * i * i)
            return await QgProxyManager.getOutboundAxiosFragment(qgOpts)
        } catch (e) {
            lastErr = e
        }
    }
    throw lastErr
}


function debugProxyEnabled() {
    return String(process.env.LEPAO_DEBUG_PROXY || '').trim() === '1'
}

function debugProxyAxiosFragment() {
    const host = process.env.LEPAO_DEBUG_PROXY_HOST || '127.0.0.1'
    const port = Number(process.env.LEPAO_DEBUG_PROXY_PORT || 9000)
    return {
        proxy: {
            host,
            port,
            protocol: 'http'
        }
    }
}

function briefUrlPath(fullUrl) {
    try {
        const u = new URL(fullUrl)
        return `${u.pathname}${u.search}`
    } catch {
        return fullUrl
    }
}

function isQgProxyEligibleFailure(err) {
    if (!err) return false
    const status = err.response?.status
    if (status === 407 || status === 408 || status === 500) return true
    if (status === 502 || status === 503 || status === 504) return true
    if (
        err.code &&
        ['ECONNRESET', 'ECONNABORTED', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN', 'ECONNREFUSED', 'EPROTO', 'ERR_FR_TOO_MANY_REDIRECTS', 'ERR_INVALID_PROTOCOL'].includes(
            err.code
        )
    ) {
        return true
    }
    if (err.isAxiosError && !err.response) return true
    const msg = (err.message || '').toLowerCase()
    if (msg.includes('timeout') || msg.includes('socket') || msg.includes('network')) return true
    return false
}

function summarizeAxiosError(err) {
    if (!err) return {}
    return {
        message: err.message,
        code: err.code,
        status: err.response?.status,
        isAxiosError: err.isAxiosError
    }
}

function isProxyTlsHandshakeReset(err) {
    if (!err) return false
    const code = err.code
    if (code !== 'ECONNRESET' && code !== 'ECONNABORTED') return false
    const msg = String(err.message || '')
    return /tls|secure\s+tls|handshake/i.test(msg)
}

function logLabel(traceId, mqTaskId) {
    let s = ''
    if (traceId) s += `[${traceId}] `
    s += '[qgOutboundAxios]'
    if (mqTaskId) s += ` [${mqTaskId}]`
    return s
}

/**
 * @param {{
 *   method?: 'get'|'post'
 *   url: string
 *   data?: any
 *   headers?: object
 *   timeout?: number
 *   outboundMode?: 'auto'|'direct'|'proxy'
 *   logger?: { info?: Function, warn?: Function, error?: Function }
 *   traceId?: string|null
 *   mqTaskId?: string|null
 *   scene?: string
 *   validateStatus?: (status:number)=>boolean
 *   responseType?: string
 *   transformResponse?: Function[]
 * }} opts
 */
async function axiosWithQgOutbound(opts) {
    const {
        method = 'get',
        url,
        data,
        headers = {},
        timeout = 15000,
        outboundMode = 'auto',
        logger = null,
        traceId = null,
        mqTaskId = null,
        scene = 'outbound',
        validateStatus,
        responseType,
        transformResponse
    } = opts

    const m = String(method).toLowerCase()
    const lbl = () => logLabel(traceId, mqTaskId)

    const baseAxiosOpts = {
        headers,
        timeout,
        proxy: false,
        ...(validateStatus ? { validateStatus } : {}),
        ...(responseType ? { responseType } : {}),
        ...(transformResponse ? { transformResponse } : {})
    }

    const exec = (fragment, requestTimeout) => {
        const outbound = buildAxiosOutboundConfig(fragment)
        const merged = { ...baseAxiosOpts, ...outbound, timeout: requestTimeout }
        if (m === 'post') {
            return axios.post(url, data, merged)
        }
        return axios.get(url, merged)
    }

    if (outboundMode === 'direct') {
        logger?.info?.(`${lbl()} (${scene}) 强制直连 ${m.toUpperCase()} ${briefUrlPath(url)}`)
        return exec({ proxy: false }, timeout)
    }

    if (debugProxyEnabled()) {
        const dbg = debugProxyAxiosFragment()
        logger?.info?.(`${lbl()} (${scene}) Charles 调试代理 LEPAO_DEBUG_PROXY ${briefUrlPath(url)}`)
        return exec(dbg, timeout)
    }

    const qgOn = await QgProxyManager.isOutboundProxyEnabled()
    if (!qgOn) {
        logger?.info?.(`${lbl()} (${scene}) 青果未启用 直连 ${briefUrlPath(url)}`)
        return exec({ proxy: false }, timeout)
    }

    let frag
    try {
        frag = await getOutboundWithBackoff({ forceRefresh: false }, 2)
    } catch (e0) {
        if (outboundMode === 'proxy') {
            const err = new Error(`代理模式提取失败: ${e0.message || e0}`)
            err.code = 'PROXY_REQUIRED_EXTRACT_FAILED'
            err.retryable = true
            throw err
        }
        logger?.error?.(`${lbl()} (${scene}) 青果提取失败改直连: ${e0.message || e0}`)
        await QgProxyManager.recordFallbackDirect({
            reason: 'qg_extract_error',
            path: briefUrlPath(url),
            scene,
            mq_task_id: mqTaskId,
            trace_id: traceId,
            ...summarizeAxiosError(e0)
        })
        return exec({ proxy: false }, timeout)
    }

    if (frag.proxy === false) {
        try {
            await sleep(400)
            frag = await getOutboundWithBackoff({ forceRefresh: true }, 2)
        } catch {
            /* keep */
        }
    }

    if (frag.proxy === false) {
        if (outboundMode === 'proxy') {
            const err = new Error('代理模式无可用节点')
            err.code = 'PROXY_REQUIRED_NO_NODE'
            err.retryable = true
            throw err
        }
        logger?.warn?.(`${lbl()} (${scene}) 无可用节点 直连 ${briefUrlPath(url)}`)
        await QgProxyManager.recordFallbackDirect({
            reason: 'no_proxy_available',
            path: briefUrlPath(url),
            scene,
            mq_task_id: mqTaskId,
            trace_id: traceId
        })
        return exec({ proxy: false }, timeout)
    }

    logger?.info?.(`${lbl()} (${scene}) 经代理 ${m.toUpperCase()} ${briefUrlPath(url)}`)
    try {
        return await exec(frag, PROXY_FIRST_TIMEOUT_MS)
    } catch (e1) {
        if (outboundMode === 'proxy') {
            const err = new Error(`代理模式请求失败: ${e1.message || e1}`)
            err.code = 'PROXY_REQUIRED_REQUEST_FAILED'
            err.retryable = true
            throw err
        }
        if (!isQgProxyEligibleFailure(e1)) throw e1

        logger?.warn?.(
            `${lbl()} (${scene}) 代理失败回退直连 err=${e1.message || e1} ${JSON.stringify(summarizeAxiosError(e1))}`
        )

        if (isProxyTlsHandshakeReset(e1)) {
            await QgProxyManager.recordFallbackDirect({
                reason: 'tls_prefinish_reset_direct',
                path: briefUrlPath(url),
                scene,
                mq_task_id: mqTaskId,
                trace_id: traceId,
                ...summarizeAxiosError(e1)
            })
            return exec({ proxy: false }, timeout)
        }

        await QgProxyManager.recordFallbackDirect({
            reason: 'proxy_post_failed_then_direct',
            path: briefUrlPath(url),
            scene,
            mq_task_id: mqTaskId,
            trace_id: traceId,
            ...summarizeAxiosError(e1)
        })
        return exec({ proxy: false }, timeout)
    }
}

module.exports = {
    axiosWithQgOutbound,
    buildAxiosOutboundConfig,
    getOutboundWithBackoff,
    isQgProxyEligibleFailure,
    isProxyTlsHandshakeReset,
    summarizeAxiosError,
    debugProxyEnabled,
    debugProxyAxiosFragment,
    sleep,
    PROXY_FIRST_TIMEOUT_MS
}