Merge branch 'mq' of Pchen0/ic-ctbu-backend into master

apis/Corn/StartAutoLepao.js

@@ -1,7 +1,9 @@
 const API = require("../../lib/API");
 const db = require('../../plugin/DataBase/db')
 const Redis = require('../../plugin/DataBase/Redis')
-const lepao = require("../../lib/Lepao/Lepao.js").lepao
+const mq = require('../../plugin/mq')
+const { assertRunforgeTaskIngress, publishRunforgeTask } = require('../../plugin/mq/runforgeTaskMq')
+const { scheduleDelayedRunforgeTask } = require('../../plugin/mq/lepaoAutoScheduleRedis')
 
 const { BaseStdResponse } = require("../../BaseStdResponse");
 
@@ -24,7 +26,7 @@ class StartAutoLepao extends API {
             const hour = new Date().getHours()
             this.logger.info('开始执行自动乐跑任务')
             let sql = `
-                        SELECT name, create_user, student_num, token, uid, school_id, state
+                        SELECT name, student_num
                         FROM lepao_account
                         WHERE auto_run = 1
                             AND (auto_time = ? OR (auto_time = -1 AND today_auto_time = ?))
@@ -34,9 +36,23 @@ class StartAutoLepao extends API {
             if (!r)
                 return this.logger.error('获取自动乐跑账号失败！')
 
+            // 为本小时内随机打散投递时间，减轻瞬时并发（0 ~ 当前小时剩余毫秒数）
+            const nowMs = Date.now()
+            const hourEnd = new Date()
+            hourEnd.setHours(hourEnd.getHours() + 1, 0, 0, 0)
+            const spreadWindowMs = Math.max(0, hourEnd.getTime() - nowMs)
+
+            let channel
+            try {
+                channel = await mq.getChannel('lepao_corn')
+                await assertRunforgeTaskIngress(channel, this.logger)
+            } catch (err) {
+                this.logger.error(`自动乐跑：连接 MQ 或声明拓扑失败：${err.message || err}`)
+                return
+            }
+
             for (const item of r) {
-                const { name, create_user, student_num, token, uid, school_id, state } = item
-                this.logger.info(`${name}(${student_num})开始乐跑`)
+                const { name, student_num } = item
 
                 const isSuccess = await Redis.get(`lepaoSuccess:${student_num}`)
                 if (isSuccess) {
@@ -44,11 +60,40 @@ class StartAutoLepao extends API {
                     continue
                 }
 
+                const delayMs = spreadWindowMs > 0 ? Math.floor(Math.random() * spreadWindowMs) : 0
+                const fireAt = nowMs + delayMs
+                const taskId = `lepao:auto:${fireAt}:${student_num}`
+                const payload = {
+                    id: taskId,
+                    type: 'lepao.startRun',
+                    data: {
+                        taskId,
+                        account: student_num
+                    },
+                    retry: 0
+                }
+
+                if (delayMs > 0) {
+                    try {
+                        await scheduleDelayedRunforgeTask(fireAt, payload, {
+                            name,
+                            account: student_num,
+                            delayMs
+                        })
+                        this.logger.info(
+                            `${name}(${student_num})已写入 Redis 调度（约 ${Math.round(delayMs / 1000)}s 后进 MQ 主队列）`
+                        )
+                    } catch (err) {
+                        this.logger.error(`${name}(${student_num})Redis 调度失败：${err.message || err}`)
+                    }
+                    continue
+                }
+
                 try {
-                    await lepao.beginLepao(create_user, student_num, token, uid, school_id, state)
-                    this.logger.info(`${name}(${student_num})乐跑完成`)
+                    publishRunforgeTask(channel, payload)
+                    this.logger.info(`${name}(${student_num})已投递自动乐跑任务`)
                 } catch (err) {
-                    this.logger.error(`${name}(${student_num})乐跑失败：${err.message || err}`)
+                    this.logger.error(`${name}(${student_num})乐跑投递失败：${err.message || err}`)
                 }
             }
         } catch (error) {

apis/Corn/StartLepao.js

@@ -1,7 +1,11 @@
 const API = require("../../lib/API");
 const db = require('../../plugin/DataBase/db')
 const Redis = require('../../plugin/DataBase/Redis')
-const lepao = require("../../lib/Lepao/Lepao.js").lepao
+const mq = require('../../plugin/mq')
+const {
+    assertRunforgeTaskIngress,
+    publishRunforgeTask
+} = require('../../plugin/mq/runforgeTaskMq')
 const { BaseStdResponse } = require("../../BaseStdResponse")
 
 // 出现异常情况时补充乐跑
@@ -31,9 +35,9 @@ class StartLepao extends API {
 
             const day = new Date().getDay()
             let sql = `
-                        SELECT name, create_user, student_num, token, uid, school_id, state
+                        SELECT name, student_num
                         FROM lepao_account
-                        WHERE auto_run = 1
+                        WHERE auto_run = 1 AND state = 1
                             AND JSON_CONTAINS(auto_day, CAST(? AS JSON))
                             AND (auto_time = ? OR (auto_time = -1 AND today_auto_time = ?))
                         `
@@ -42,7 +46,7 @@ class StartLepao extends API {
                 return this.logger.error('获取补充乐跑账号失败！')
 
             for (const item of r) {
-                const { name, create_user, student_num, token, uid, school_id, state } = item
+                const { name, student_num } = item
                 this.logger.info(`${name}(${student_num})开始补充乐跑`)
 
                 const isSuccess = await Redis.get(`lepaoSuccess:${student_num}`)
@@ -52,8 +56,22 @@ class StartLepao extends API {
                 }
 
                 try {
-                    await lepao.beginLepao(create_user, student_num, token, uid, school_id, state)
-                    this.logger.info(`${name}(${student_num})补充乐跑完成`)
+                    const channel = await mq.getChannel('lepao_corn')
+                    await assertRunforgeTaskIngress(channel, this.logger)
+
+                    const taskId = `lepao:repair:${Date.now()}:${student_num}`
+                    const payload = {
+                        id: taskId,
+                        type: 'lepao.startRun',
+                        data: {
+                            taskId,
+                            account: student_num
+                        },
+                        retry: 0
+                    }
+
+                    publishRunforgeTask(channel, payload)
+                    this.logger.info(`${name}(${student_num})已投递补充乐跑任务`)
                 } catch (err) {
                     this.logger.error(`${name}(${student_num})补充乐跑失败：${err.message || err}`)
                 }

apis/Corn/UpdateState.js

@@ -1,8 +1,11 @@
 const API = require("../../lib/API.js")
 const db = require('../../plugin/DataBase/db.js')
-const axios = require('axios')
-const config = require('../../config.json')
 const { BaseStdResponse } = require("../../BaseStdResponse.js")
+const {
+    probeSetZone,
+    isProbeSetZoneOk,
+    getProbeFailMessage
+} = require('../../plugin/Lepao/runforgeSetZoneProbe')
 
 const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms))
 
@@ -13,8 +16,6 @@ class UpdateState extends API {
         this.noEncrypt()
         this.setPath('/Corn/UpdateState')
         this.setMethod('GET')
-
-        this.runpy = config.runpy
     }
 
     async onRequest(req, res) {
@@ -25,41 +26,42 @@ class UpdateState extends API {
 
             this.logger.info('开始更新乐跑账号登录状态')
 
-            const zoneUrl = this.runpy + '/set_zone'
-
-            let sql = `SELECT id, uid, token, school_id, name, student_num FROM lepao_account WHERE state = 0 AND auto_run = 1 AND token IS NOT NULL`
+            let sql = `SELECT id, uid, token, school_id, name, student_num, userAgent FROM lepao_account WHERE state = 0 AND auto_run = 1 AND token IS NOT NULL`
 
             let r = await db.query(sql)
             if (!r)
                 return this.logger.error('更新乐跑账号登录状态失败！')
 
             for (const item of r) {
-                const { name, student_num, token, uid, school_id } = item
+                const { name, student_num, token, uid, school_id, userAgent } = item
                 this.logger.info(`${name}(${student_num})开始更新乐跑登录状态`)
 
                 await sleep(2000)
 
                 try {
-                    const ossData = { uid, token, school_id, student_id: student_num, random_id: 1, run_end_time: 1 }
-                    const zoneRes = await axios.post(zoneUrl, ossData)
-                    const { data } = zoneRes
+                    const data = await probeSetZone({
+                        uid,
+                        token,
+                        school_id,
+                        student_num,
+                        random_id: 1,
+                        userAgent
+                    })
 
                     this.logger.info(`${student_num}更新乐跑登录状态返回结果: ${JSON.stringify(data)}`)
 
-                    if (!data || data.status != 1 || !data.data) {
+                    if (!isProbeSetZoneOk(data)) {
                         this.logger.info(`${name}(${student_num})数据获取失败，不更新`)
                         continue
                     }
 
-                    const sql = 'UPDATE lepao_account SET state = 1 WHERE student_num  = ?'
-                    await db.query(sql, [student_num])
+                    const updateSql = 'UPDATE lepao_account SET state = 1 WHERE student_num  = ?'
+                    await db.query(updateSql, [student_num])
 
                     this.logger.info(`${name}(${student_num})数据获取成果，状态更新为正常`)
                 } catch (err) {
                     this.logger.error(`${name}(${student_num})更新乐跑登录状态失败：${err.message || err}`)
                 }
-
-                
             }
 
             this.logger.info('更新乐跑账号登录状态完成')

apis/Corn/UpdateStateAll.js

@@ -1,8 +1,11 @@
 const API = require("../../lib/API.js")
 const db = require('../../plugin/DataBase/db.js')
-const axios = require('axios')
-const config = require('../../config.json')
 const { BaseStdResponse } = require("../../BaseStdResponse.js")
+const {
+    probeSetZone,
+    isProbeSetZoneOk,
+    getProbeFailMessage
+} = require('../../plugin/Lepao/runforgeSetZoneProbe')
 
 const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms))
 
@@ -13,8 +16,6 @@ class UpdateStateAll extends API {
         this.noEncrypt()
         this.setPath('/Corn/UpdateStateAll')
         this.setMethod('GET')
-
-        this.runpy = config.runpy
     }
 
     async onRequest(req, res) {
@@ -25,48 +26,50 @@ class UpdateStateAll extends API {
 
             this.logger.info('开始更新乐跑账号登录状态')
 
-            const zoneUrl = this.runpy + '/set_zone'
-
-            let sql = `SELECT id, uid, token, school_id, name, student_num FROM lepao_account WHERE token IS NOT NULL`
+            let sql = `SELECT id, uid, token, school_id, name, student_num, userAgent FROM lepao_account WHERE token IS NOT NULL`
 
             let r = await db.query(sql)
             if (!r)
                 return this.logger.error('更新乐跑账号登录状态失败！')
 
             for (const item of r) {
-                const { name, student_num, token, uid, school_id } = item
+                const { name, student_num, token, uid, school_id, userAgent } = item
                 this.logger.info(`${name}(${student_num})开始更新乐跑登录状态`)
 
                 await sleep(2000)
 
                 try {
-                    const ossData = { uid, token, school_id, student_id: student_num, random_id: 1, run_end_time: 1 }
-                    const zoneRes = await axios.post(zoneUrl, ossData)
-                    const { data } = zoneRes
+                    const data = await probeSetZone({
+                        uid,
+                        token,
+                        school_id,
+                        student_num,
+                        random_id: 1,
+                        userAgent
+                    })
 
                     this.logger.info(`${student_num}更新乐跑登录状态返回结果: ${JSON.stringify(data)}`)
 
-                    if (!data || data.status != 1 || !data.data) {
-                        if (data?.info.includes('重新登录')) {
-                            const sql = 'UPDATE lepao_account SET state = 0 WHERE student_num  = ?'
-                            await db.query(sql, [student_num])
+                    if (!isProbeSetZoneOk(data)) {
+                        const msg = getProbeFailMessage(data)
+                        if (msg && msg.includes('重新登录')) {
+                            const failSql = 'UPDATE lepao_account SET state = 0 WHERE student_num  = ?'
+                            await db.query(failSql, [student_num])
                             this.logger.info(`${name}(${student_num})状态更新为待登录`)
+                        } else {
+                            this.logger.info(`${name}(${student_num})数据获取失败`)
                         }
 
-                        else this.logger.info(`${name}(${student_num})数据获取失败`)
-
                         continue
                     }
 
-                    const sql = 'UPDATE lepao_account SET state = 1 WHERE student_num  = ?'
-                    await db.query(sql, [student_num])
+                    const okSql = 'UPDATE lepao_account SET state = 1 WHERE student_num  = ?'
+                    await db.query(okSql, [student_num])
 
                     this.logger.info(`${name}(${student_num})状态更新为正常`)
                 } catch (err) {
                     this.logger.error(`${name}(${student_num})更新乐跑登录状态失败：${err.message || err}`)
                 }
-
-
             }
 
             this.logger.info('更新乐跑账号登录状态完成')

apis/Lepao/Account/UpdateAccount/UpdateAccount.js

@@ -1,11 +1,10 @@
 const API = require("../../../../lib/API.js")
 const db = require("../../../../plugin/DataBase/db.js")
-const axios = require('axios')
 const EmailTemplate = require('../../../../plugin/Email/emailTemplate.js')
-const config = require('../../../../config.json')
-const lepao = require("../../../../lib/Lepao/Lepao.js").lepao
+const { enqueueLepaoStartRun } = require('../../../../plugin/mq/enqueueLepaoStartRun')
 const mq = require('../../../../plugin/mq')
 const { BaseStdResponse } = require("../../../../BaseStdResponse.js")
+const { dataDecrypt } = require('../../../../plugin/Lepao/Crypto')
 
 // 客户端上传数据接口
 class UpdateAccount extends API {
@@ -14,7 +13,6 @@ class UpdateAccount extends API {
 
         this.messageQueue = 'runforge_message_queue'
 
-        this.runpy = config.runpy
         this.noEncrypt()
         this.setPath('/Lepao/UpdateAccount')
         this.setMethod('POST')
@@ -29,29 +27,26 @@ class UpdateAccount extends API {
             })
 
         try {
-            const endpoint = config.runpy + '/decrypted_data'
-            const ReqData = await axios.post(endpoint, { data: reqData })
-            let userData = ReqData.data
-            if (userData?.code !== 200 || !userData.data)
+            const userData = JSON.parse(dataDecrypt(reqData) || '{}')
+            if (!userData || Object.keys(userData).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
-            const { token } = userData.data
+            const { token } = userData
             if ([token].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '未提取出用户登录信息，请重试'
                 })
 
-            const ResData = await axios.post(endpoint, { data: resData })
-            let userData2 = ResData.data
-            if (userData2?.code !== 200 || !userData2.data)
+            const userData2 = JSON.parse(dataDecrypt(resData) || '{}')
+            if (!userData2 || Object.keys(userData2).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
-            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2.data
+            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2
             if ([uid, student_num, school_id, grade_id, class_id, sex, name, academy_name].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
@@ -139,25 +134,25 @@ class UpdateAccount extends API {
                 }
 
                 if (findRows[0].auto_run === 1 && Array.isArray(findRows[0].auto_day) && findRows[0].auto_day.includes(new Date().getDay())) {
-                    lepao.beginLepao(findRows[0].create_user, student_num, token, uid, school_id, 1)
+                    enqueueLepaoStartRun(student_num, this.logger)
                 }
             }
 
             // 获取新加账号中存在的路径
-            try {
-                let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
-                let rows = await db.query(sql, [student_num])
-
-                // 不是老帐号就不获取
-                if (!rows || rows.length !== 0) return
-
-                const reqData = { uid, token, school_id, student_id: student_num }
-                this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
-                const recordUrl = this.runpy + '/get_path_data'
-                await axios.post(recordUrl, reqData)
-            } catch (error) {
-                this.logger.info(`获取跑步记录出错 ${error.stack}`)
-            }
+            // try {
+            //     let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
+            //     let rows = await db.query(sql, [student_num])
+
+            //     // 不是老帐号就不获取
+            //     if (!rows || rows.length !== 0) return
+
+            //     const reqData = { uid, token, school_id, student_id: student_num }
+            //     this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
+                
+
+            // } catch (error) {
+            //     this.logger.info(`获取跑步记录出错 ${error.stack}`)
+            // }
 
         } catch (error) {
             this.logger.error(`更新用户信息时出错。${error.stack}`)

apis/Lepao/Account/UpdateAccount/UpdateAccountAndroidApp.js

@@ -1,11 +1,10 @@
 const API = require("../../../../lib/API.js")
 const db = require("../../../../plugin/DataBase/db.js")
-const axios = require('axios')
 const EmailTemplate = require('../../../../plugin/Email/emailTemplate.js')
-const config = require('../../../../config.json')
-const lepao = require("../../../../lib/Lepao/Lepao.js").lepao
+const { enqueueLepaoStartRun } = require('../../../../plugin/mq/enqueueLepaoStartRun')
 const mq = require('../../../../plugin/mq')
 const { BaseStdResponse } = require("../../../../BaseStdResponse.js")
+const { dataDecrypt } = require('../../../../plugin/Lepao/Crypto')
 
 // 客户端上传数据接口
 class UpdateAccountAndroidApp extends API {
@@ -13,7 +12,6 @@ class UpdateAccountAndroidApp extends API {
         super()
 
         this.messageQueue = 'runforge_message_queue'
-        this.runpy = config.runpy
 
         this.setPath('/Lepao/UpdateAccountAndroidApp')
         this.setMethod('POST')
@@ -29,29 +27,26 @@ class UpdateAccountAndroidApp extends API {
             })
 
         try {
-            const endpoint = config.runpy + '/decrypted_data'
-            const ReqData = await axios.post(endpoint, { data: reqData })
-            let userData = ReqData.data
-            if (userData?.code !== 200 || !userData.data)
+            const userData = JSON.parse(dataDecrypt(reqData) || '{}')
+            if (!userData || Object.keys(userData).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
-            const { token } = userData.data
+            const { token } = userData
             if ([token].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '未提取出用户登录信息，请重试'
                 })
 
-            const ResData = await axios.post(endpoint, { data: resData })
-            let userData2 = ResData.data
-            if (userData2?.code !== 200 || !userData2.data)
+            const userData2 = JSON.parse(dataDecrypt(resData) || '{}')
+            if (!userData2 || Object.keys(userData2).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
-            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2.data
+            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2
             if ([uid, student_num, school_id, grade_id, class_id, sex, name, academy_name].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
@@ -141,25 +136,25 @@ class UpdateAccountAndroidApp extends API {
                 }
 
                 if (findRows[0].auto_run === 1 && Array.isArray(findRows[0].auto_day) && findRows[0].auto_day.includes(new Date().getDay())) {
-                    lepao.beginLepao(findRows[0].create_user, student_num, token, uid, school_id, 1)
+                    enqueueLepaoStartRun(student_num, this.logger)
                 }
             }
 
             // 获取新加账号中存在的路径
-            try {
-                let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
-                let rows = await db.query(sql, [student_num])
-
-                // 不是老帐号就不获取
-                if (!rows || rows.length !== 0) return
-
-                const reqData = { uid, token, school_id, student_id: student_num }
-                this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
-                const recordUrl = this.runpy + '/get_path_data'
-                await axios.post(recordUrl, reqData)
-            } catch (error) {
-                this.logger.info(`获取跑步记录出错 ${error.stack}`)
-            }
+            // try {
+            //     let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
+            //     let rows = await db.query(sql, [student_num])
+
+            //     // 不是老帐号就不获取
+            //     if (!rows || rows.length !== 0) return
+
+            //     const reqData = { uid, token, school_id, student_id: student_num }
+            //     this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
+                
+
+            // } catch (error) {
+            //     this.logger.info(`获取跑步记录出错 ${error.stack}`)
+            // }
 
         } catch (error) {
             this.logger.error(`更新用户信息时出错。${error.stack}`)

apis/Lepao/Account/UpdateAccount/UpdateAccountiPhone.js

@@ -1,11 +1,10 @@
 const API = require("../../../../lib/API.js")
 const db = require("../../../../plugin/DataBase/db.js")
-const axios = require('axios')
 const EmailTemplate = require('../../../../plugin/Email/emailTemplate.js')
-const config = require('../../../../config.json')
-const lepao = require("../../../../lib/Lepao/Lepao.js").lepao
+const { enqueueLepaoStartRun } = require('../../../../plugin/mq/enqueueLepaoStartRun')
 const mq = require('../../../../plugin/mq')
 const { BaseStdResponse } = require("../../../../BaseStdResponse.js")
+const { dataDecrypt } = require('../../../../plugin/Lepao/Crypto')
 
 // 客户端上传数据接口
 class UpdateAccountiPhone extends API {
@@ -13,7 +12,6 @@ class UpdateAccountiPhone extends API {
         super()
 
         this.messageQueue = 'runforge_message_queue'
-        this.runpy = config.runpy
         this.noEncrypt()
         this.setPath('/Lepao/UpdateAccountiPhone')
         this.setMethod('POST')
@@ -29,30 +27,27 @@ class UpdateAccountiPhone extends API {
             })
 
         try {
-            const endpoint = config.runpy + '/decrypted_data'
-            const ReqData = await axios.post(endpoint, { data: reqData })
-            let userData = ReqData.data
-            if (userData?.code !== 200 || !userData.data)
+            const userData = JSON.parse(dataDecrypt(reqData) || '{}')
+            if (!userData || Object.keys(userData).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
-            const { token } = userData.data
+            const { token } = userData
             if ([token].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '未提取出用户登录信息，请重试'
                 })
 
-            const ResData = await axios.post(endpoint, { data: resData })
-            let userData2 = ResData.data
-            if (userData2?.code !== 200 || !userData2.data)
+            const userData2 = JSON.parse(dataDecrypt(resData) || '{}')
+            if (!userData2 || Object.keys(userData2).length === 0)
                 return res.json({
                     ...BaseStdResponse.ERR,
                     msg: '无法解析用户数据，请重试'
                 })
 
-            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2.data
+            const { uid, user_avatar, student_num, school_id, grade_id, class_id, sex, name, academy_name } = userData2
             if ([uid, student_num, school_id, grade_id, class_id, sex, name, academy_name].some(value => value === '' || value === null || value === undefined))
                 return res.json({
                     ...BaseStdResponse.ERR,
@@ -143,25 +138,25 @@ class UpdateAccountiPhone extends API {
                 }
 
                 if (findRows[0].auto_run === 1 && Array.isArray(findRows[0].auto_day) && findRows[0].auto_day.includes(new Date().getDay())) {
-                    lepao.beginLepao(findRows[0].create_user, student_num, token, uid, school_id, 1)
+                    enqueueLepaoStartRun(student_num, this.logger)
                 }
             }
 
             // 获取新加账号中存在的路径
-            try {
-                let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
-                let rows = await db.query(sql, [student_num])
-
-                // 不是老帐号就不获取
-                if (!rows || rows.length !== 0) return
-
-                const reqData = { uid, token, school_id, student_id: student_num }
-                this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
-                const recordUrl = this.runpy + '/get_path_data'
-                await axios.post(recordUrl, reqData)
-            } catch (error) {
-                this.logger.info(`获取跑步记录出错 ${error.stack}`)
-            }
+            // try {
+            //     let sql = 'SELECT id FROM lepao_record WHERE lepao_account = ?'
+            //     let rows = await db.query(sql, [student_num])
+
+            //     // 不是老帐号就不获取
+            //     if (!rows || rows.length !== 0) return
+
+            //     const reqData = { uid, token, school_id, student_id: student_num }
+            //     this.logger.info(`开始请求获取跑步记录 uid=${uid} student_id=${student_num}`)
+                
+
+            // } catch (error) {
+            //     this.logger.info(`获取跑步记录出错 ${error.stack}`)
+            // }
 
         } catch (error) {
             this.logger.error(`更新用户信息时出错。${error.stack}`)

apis/Lepao/Account/UpdateSelfAccount.js

@@ -0,0 +1,135 @@
+const API = require("../../../lib/API.js")
+const db = require("../../../plugin/DataBase/db.js")
+const axios = require("axios")
+const AccessControl = require("../../../lib/AccessControl.js")
+const { BaseStdResponse } = require("../../../BaseStdResponse.js")
+const { dataEncrypt, dataDecrypt, dataSign } = require("../../../plugin/Lepao/Crypto")
+const { URLSearchParams } = require("url")
+
+class UpdateSelfAccount extends API {
+    constructor() {
+        super()
+
+        this.setPath('/Lepao/Account/UpdateSelfAccount')
+        this.setMethod('POST')
+    }
+
+    async onRequest(req, res) {
+        const { uuid, session, student_num } = req.body
+
+        if ([uuid, session, student_num].some(v => v === '' || v === null || v === undefined)) {
+            return res.json({
+                ...BaseStdResponse.MISSING_PARAMETER
+            })
+        }
+
+        if (!await AccessControl.checkSession(uuid, session)) {
+            return res.status(401).json({
+                ...BaseStdResponse.ACCESS_DENIED
+            })
+        }
+
+        try {
+            const rows = await db.query(
+                'SELECT uid, token, school_id, userAgent, state FROM lepao_account WHERE student_num = ? AND create_user = ?',
+                [student_num, uuid]
+            )
+            if (!rows || rows.length !== 1) {
+                return res.json({
+                    ...BaseStdResponse.ERR,
+                    msg: '未找到该乐跑账号或无权限操作'
+                })
+            }
+
+            const account = rows[0]
+            const raw = {
+                uid: account.uid,
+                token: account.token,
+                school_id: account.school_id,
+                term_id: 0,
+                course_id: 0,
+                class_id: 0,
+                student_num,
+                card_id: student_num,
+                timestamp: Number((Date.now() / 1000).toFixed(3)),
+                version: 1,
+                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+                ostype: 5
+            }
+            raw.sign = dataSign(raw)
+
+            const form = new URLSearchParams()
+            form.append('ostype', '5')
+            form.append('data', dataEncrypt(JSON.stringify(raw)))
+
+            const headers = {
+                'Content-Type': 'application/x-www-form-urlencoded',
+                'Accept': '*/*',
+                'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
+                'Accept-Encoding': 'gzip, deflate, br',
+                'Referer': 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
+                'User-Agent': account.userAgent || 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'
+            }
+
+            const apiRes = await axios.post(
+                'https://lepao.ctbu.edu.cn/v3/api.php/Run2/beforeRunV260',
+                form,
+                { headers, proxy: false }
+            )
+
+            let result = apiRes.data
+            if (result?.data && result?.is_encrypt === 1) {
+                result.data = JSON.parse(dataDecrypt(result.data))
+            }
+
+            const info = result?.info || result?.msg || '系统繁忙，请稍后再试'
+            const updateTime = Date.now()
+
+            // 登录失效：更新 state=0
+            if (String(info).includes('重新登录') || Number(result?.status) === 101) {
+                await db.query('UPDATE lepao_account SET state = 0, update_time = ? WHERE student_num = ? AND create_user = ?', [updateTime, student_num, uuid])
+                return res.json({
+                    ...BaseStdResponse.ERR,
+                    msg: info
+                })
+            }
+
+            if (!result || Number(result.status) !== 1 || !result.data) {
+                return res.json({
+                    ...BaseStdResponse.ERR,
+                    msg: info
+                })
+            }
+
+            const term_num = Number(result.data.term_num ?? 0)
+            const total_num = Number(result.data.total_num ?? 30)
+
+            const updateRows = await db.query(
+                'UPDATE lepao_account SET term_num = ?, total_num = ?, state = 1, update_time = ? WHERE student_num = ? AND create_user = ?',
+                [term_num, total_num, updateTime, student_num, uuid]
+            )
+            if (!updateRows || updateRows.affectedRows !== 1) {
+                return res.json({
+                    ...BaseStdResponse.DATABASE_ERR
+                })
+            }
+
+            return res.json({
+                ...BaseStdResponse.OK,
+                data: {
+                    term_num,
+                    total_num,
+                    state: 1
+                }
+            })
+        } catch (error) {
+            this.logger.error(`用户自助同步乐跑账号失败: ${error.stack || error}`)
+            return res.json({
+                ...BaseStdResponse.ERR,
+                msg: '同步失败，请稍后再试'
+            })
+        }
+    }
+}
+
+module.exports.UpdateSelfAccount = UpdateSelfAccount

apis/Lepao/Face/BeginFaceReco.js

@@ -3,15 +3,12 @@ const db = require("../../../plugin/DataBase/db.js")
 const axios = require('axios')
 const { v4: uuidv4 } = require('uuid')
 const Redis = require('../../../plugin/DataBase/Redis')
-const config = require('../../../config.json')
 const { BaseStdResponse } = require("../../../BaseStdResponse.js")
 
 class BeginFaceReco extends API {
     constructor() {
         super()
 
-        this.runpy = config.runpy
-
         this.setPath('/Face/BeginFaceReco')
         this.setMethod('POST')
     }

apis/Lepao/SingleRun.js

@@ -3,12 +3,16 @@ const Redis = require('../../plugin/DataBase/Redis')
 const db = require("../../plugin/DataBase/db.js")
 const { BaseStdResponse } = require("../../BaseStdResponse.js")
 const AccessControl = require("../../lib/AccessControl.js")
-const lepao = require("../../lib/Lepao/Lepao.js").lepao
+const mq = require('../../plugin/mq')
+const {
+    assertRunforgeTaskIngress,
+    publishRunforgeTask
+} = require('../../plugin/mq/runforgeTaskMq')
 
 // 单次乐跑
 class SingleRun extends API {
     constructor() {
-        super();
+        super()
 
         this.setPath('/Lepao/SingleRun')
         this.setMethod('GET')
@@ -86,7 +90,21 @@ class SingleRun extends API {
             })
 
             try {
-                await lepao.beginLepao(selectRows[0].create_user, student_num, rows[0].token, rows[0].uid, rows[0].school_id, rows[0].state)
+                const channel = await mq.getChannel('lepao_api')
+                await assertRunforgeTaskIngress(channel, this.logger)
+
+                const taskId = `lepao:${Date.now()}:${student_num}`
+                const payload = {
+                    id: taskId,
+                    type: 'lepao.startRun',
+                    data: {
+                        taskId,
+                        account: student_num
+                    },
+                    retry: 0
+                }
+
+                publishRunforgeTask(channel, payload)
             } catch (err) {
                 this.logger.error(`后台乐跑任务异常：${err.stack}`)
             }

apis/Public/GetAppVersion.js

@@ -13,8 +13,8 @@ class GetAppVersion extends API {
         res.json({
             ...BaseStdResponse.OK,
             data: {
-                version: '1.9',
-                msg: '\n更新内容：\n修复部分bug，优化用户体验'
+                version: '2.0',
+                msg: '\n更新内容：\n重构乐跑逻辑 优化乐跑任务调度功能'
             }
         })
     }

apis/User/Admin/GetQueueTasks.js

@@ -0,0 +1,294 @@
+const API = require('../../../lib/API')
+const axios = require('axios')
+const mq = require('../../../plugin/mq')
+const config = require('../../../config.json')
+const AccessControl = require('../../../lib/AccessControl')
+const { BaseStdResponse } = require('../../../BaseStdResponse')
+const {
+    SCHEDULE_KEY,
+    listPendingScheduledForAdmin,
+    countPendingScheduled
+} = require('../../../plugin/mq/lepaoAutoScheduleRedis')
+
+/** 允许通过管理接口查看的队列（防任意队列名探测） */
+const ALLOWED_QUEUES = [
+    'runforge_task_queue',
+    'runforge_task_result_queue',
+    'runforge_task_dead_queue',
+    'runforge_message_queue',
+    'order_payment_check',
+    'mq_health_check'
+]
+
+function parseAmqpHttpBase(amqpUrl) {
+    const u = new URL(String(amqpUrl).replace(/^amqp:/, 'http:'))
+    return {
+        user: decodeURIComponent(u.username || ''),
+        password: decodeURIComponent(u.password || ''),
+        hostname: u.hostname
+    }
+}
+
+function managementEndpoint(queueName) {
+    const rm = config.rabbitmq || {}
+    const creds = parseAmqpHttpBase(rm.url || '')
+    const base = (rm.managementBaseUrl || `http://${creds.hostname}:15672`).replace(/\/$/, '')
+    const vhost = encodeURIComponent(rm.vhost != null ? rm.vhost : '/')
+    const q = encodeURIComponent(queueName)
+    return {
+        url: `${base}/api/queues/${vhost}/${q}/get`,
+        auth: {
+            username: rm.managementUser || creds.user,
+            password: rm.managementPassword || creds.password
+        }
+    }
+}
+
+function decodePayload(msg) {
+    const enc = msg.payload_encoding || 'string'
+    let raw = msg.payload
+    if (enc === 'base64' && typeof raw === 'string') {
+        try {
+            raw = Buffer.from(raw, 'base64').toString('utf8')
+        } catch {
+            return { raw: msg.payload, encoding: enc, parseError: true }
+        }
+    }
+    if (typeof raw === 'string') {
+        try {
+            return { body: JSON.parse(raw), encoding: enc }
+        } catch {
+            return { body: raw, encoding: enc }
+        }
+    }
+    return { body: raw, encoding: enc }
+}
+
+/**
+ * 通过 Management API 窥视队列消息（reject_requeue_true：看完后重新入队，不消费）
+ */
+async function peekQueueMessages(queueName, limit) {
+    const { url, auth } = managementEndpoint(queueName)
+    const { data, status } = await axios.post(
+        url,
+        {
+            count: limit,
+            ackmode: 'reject_requeue_true',
+            encoding: 'auto'
+        },
+        {
+            auth,
+            timeout: 12000,
+            validateStatus: () => true
+        }
+    )
+
+    if (status >= 400) {
+        const reason =
+            typeof data === 'object' && data !== null
+                ? data.reason || data.error || JSON.stringify(data)
+                : String(data)
+        const err = new Error(reason || `Management HTTP ${status}`)
+        err.status = status
+        throw err
+    }
+
+    const list = Array.isArray(data) ? data : []
+    return list.map((m) => {
+        const decoded = decodePayload(m)
+        return {
+            redelivered: m.redelivered,
+            routing_key: m.routing_key,
+            exchange: m.exchange,
+            properties: m.properties
+                ? {
+                      messageId: m.properties.message_id,
+                      timestamp: m.properties.timestamp,
+                      contentType: m.properties.content_type,
+                      headers: m.properties.headers
+                  }
+                : undefined,
+            payload: decoded.body,
+            payload_encoding: decoded.encoding
+        }
+    })
+}
+
+class GetQueueTasks extends API {
+    constructor() {
+        super()
+
+        this.setPath('/Admin/MQ/GetQueueTasks')
+        this.setMethod('get')
+    }
+
+    async onRequest(req, res) {
+        const {
+            uuid,
+            session,
+            queue,
+            limit: limitStr,
+            summary,
+            includeScheduled,
+            scheduledLimit: scheduledLimitStr
+        } = req.query
+
+        if ([uuid, session].some((v) => v === '' || v == null))
+            return res.json({
+                ...BaseStdResponse.MISSING_PARAMETER
+            })
+
+        if (!(await AccessControl.checkSession(uuid, session)))
+            return res.status(401).json({
+                ...BaseStdResponse.ACCESS_DENIED
+            })
+
+        const permission = await AccessControl.getPermission(uuid)
+        if (!permission.includes('admin') && !permission.includes('service'))
+            return res.json({
+                ...BaseStdResponse.PERMISSION_DENIED
+            })
+
+        const wantSummary = summary === '1' || summary === 'true'
+
+        try {
+            const ch = await mq.getChannel('admin_queue_inspect')
+
+            if (wantSummary) {
+                const queues = {}
+                for (const name of ALLOWED_QUEUES) {
+                    try {
+                        const info = await ch.checkQueue(name)
+                        queues[name] = {
+                            messageCount: info.messageCount,
+                            consumerCount: info.consumerCount
+                        }
+                    } catch (e) {
+                        queues[name] = {
+                            messageCount: null,
+                            consumerCount: null,
+                            error: e.message || String(e)
+                        }
+                    }
+                }
+
+                const slimit = Math.min(
+                    2000,
+                    Math.max(1, parseInt(scheduledLimitStr, 10) || 800)
+                )
+                const pendingCount = await countPendingScheduled(Date.now())
+                const scheduledMirror = await listPendingScheduledForAdmin(Date.now(), slimit)
+
+                return res.json({
+                    ...BaseStdResponse.OK,
+                    data: {
+                        summary: true,
+                        queues,
+                        redisScheduler: {
+                            key: SCHEDULE_KEY,
+                            pendingCount,
+                            note: '到期任务由本服务定时写入 runforge_task_queue；多实例共享同一 Redis ZSET。'
+                        },
+                        autoRunScheduledMirror: {
+                            pendingCount,
+                            note: scheduledMirror.note,
+                            sample: scheduledMirror.items.slice(0, 20)
+                        },
+                        fetchedAt: Date.now()
+                    }
+                })
+            }
+
+            const queueName = queue || 'runforge_task_queue'
+            if (!ALLOWED_QUEUES.includes(queueName))
+                return res.json({
+                    ...BaseStdResponse.ERR,
+                    msg: '不支持的队列名称'
+                })
+
+            let limit = parseInt(limitStr, 10)
+            if (Number.isNaN(limit) || limit < 1) limit = 30
+            if (limit > 100) limit = 100
+
+            let messageCount = null
+            let consumerCount = null
+            try {
+                const info = await ch.checkQueue(queueName)
+                messageCount = info.messageCount
+                consumerCount = info.consumerCount
+            } catch (e) {
+                return res.json({
+                    ...BaseStdResponse.ERR,
+                    msg: `无法访问队列：${e.message || e}`
+                })
+            }
+
+            let tasks = []
+            let managementError = null
+            try {
+                tasks = await peekQueueMessages(queueName, limit)
+            } catch (e) {
+                managementError =
+                    e.status === 401 || e.status === 403
+                        ? 'Management 鉴权失败，请在 config.json 的 rabbitmq 中配置 managementUser / managementPassword，或检查管理插件用户权限'
+                        : e.code === 'ECONNREFUSED' || e.code === 'ETIMEDOUT'
+                          ? '无法连接 RabbitMQ Management（默认 15672）。请开启管理插件并开放端口，或配置 rabbitmq.managementBaseUrl'
+                          : (e.message || String(e))
+                this.logger.warn(`[GetQueueTasks] Management 窥视失败: ${managementError}`)
+            }
+
+            const wantScheduled =
+                includeScheduled !== '0' &&
+                includeScheduled !== 'false' &&
+                queueName === 'runforge_task_queue'
+
+            let autoRunScheduledMirror = null
+            let pendingScheduledCount = null
+            if (queueName === 'runforge_task_queue') {
+                pendingScheduledCount = await countPendingScheduled(Date.now())
+                if (wantScheduled) {
+                    const slimit = Math.min(
+                        500,
+                        Math.max(1, parseInt(scheduledLimitStr, 10) || 200)
+                    )
+                    autoRunScheduledMirror = await listPendingScheduledForAdmin(Date.now(), slimit)
+                }
+            }
+
+            const detail = {
+                queue: queueName,
+                messageCount,
+                consumerCount,
+                peekLimit: limit,
+                tasks,
+                managementError,
+                redisScheduler:
+                    queueName === 'runforge_task_queue'
+                        ? {
+                              key: SCHEDULE_KEY,
+                              pendingCount: pendingScheduledCount
+                          }
+                        : undefined,
+                autoRunScheduledMirror,
+                fetchedAt: Date.now()
+            }
+            if (queueName === 'runforge_task_queue') {
+                detail.peekNote =
+                    'tasks：已在主队列中的消息；autoRunScheduledMirror：尚未到 fireAt、仍只在 Redis 中的调度（到期后由服务写入 MQ）。'
+            }
+
+            return res.json({
+                ...BaseStdResponse.OK,
+                data: detail
+            })
+        } catch (error) {
+            this.logger.error(`GetQueueTasks: ${error.stack || error}`)
+            return res.json({
+                ...BaseStdResponse.ERR,
+                msg: error.message || '查询失败'
+            })
+        }
+    }
+}
+
+module.exports.GetQueueTasks = GetQueueTasks

lib/Lepao/Lepao.js

@@ -1,512 +0,0 @@
-const axios = require('axios')
-const Redis = require('../../plugin/DataBase/Redis')
-const db = require('../../plugin/DataBase/db')
-const Logger = require('../Logger')
-const path = require('path')
-const EmailTemplate = require('../../plugin/Email/emailTemplate')
-const config = require('../../config.json')
-const mq = require('../../plugin/mq')
-
-class Lepao {
-    constructor() {
-        this.logger = new Logger(path.join(__dirname, '../logs/Lepao.log'), 'INFO')
-        this.runpy = config.runpy
-
-        this.messageQueue = 'runforge_message_queue'
-    }
-
-    async getPath(account, vip) {
-        this.logger.info(`${account}开始获取路径`)
-        const accountSql = 'SELECT area, sex FROM lepao_account WHERE student_num = ?'
-        const rows = await db.query(accountSql, [account])
-        if (!rows || rows.length === 0) {
-            this.logger.error(`${account}无法获取账号数据`)
-            throw new Error('无法获取账号数据')
-        }
-
-        const { area, sex } = rows[0]
-
-        let max = 4.00
-        let min = 2.00
-        if (sex === 2) {
-            max = 2.00
-            min = 1.60
-        }
-
-        this.logger.info(`${account}路径参数: area=${area ?? '随机'}, max_distance=${max}, min_distance=${min}`)
-
-        let pathSql = 'SELECT id FROM path_data WHERE state = 1 AND distance < ? AND distance > ? '
-        const pathParams = [max, min]
-
-        if (area) {
-            pathSql += ' AND run_zone_name = ?'
-            pathParams.push(area)
-        }
-
-        pathSql += ' ORDER BY count ASC LIMIT 1'
-
-        const paths = await db.query(pathSql, pathParams)
-        if (!paths || paths.length === 0) {
-            this.logger.error(`${account}未找到符合条件的路线`)
-            throw new Error('未找到符合条件的路线，请改变路径选择条件')
-        }
-
-        const randomPath = paths[0]
-
-        const updateSql = 'UPDATE path_data SET count = count + 1 WHERE id = ?'
-        await db.query(updateSql, [randomPath.id])
-        this.logger.info(`${account}路径选中id=${randomPath.id}，计数加1成功`)
-        return randomPath.id
-    }
-
-    async getRecord(uid, token, school_id, student_id, userAgent) {
-        try {
-            const reqData = { uid, token, school_id, student_id, userAgent }
-            this.logger.info(`开始请求获取跑步次数 uid=${uid} student_id=${student_id}`)
-            const recordUrl = this.runpy + '/get_record'
-            let recordRes = await axios.post(recordUrl, reqData)
-            const { data } = recordRes
-            this.logger.info(`获取跑步次数返回结果: ${JSON.stringify(data)}`)
-            if (!data || data.status !== 1 || !data.data) {
-                this.logger.warn('获取剩余跑步次数失败，接口返回异常')
-                return
-            }
-            return data.data
-        } catch (error) {
-            this.logger.error(`获取跑步次数失败: ${error.stack || error.message}`)
-            return
-        }
-    }
-
-    async writeRedis(account) {
-        try {
-            // 计算至明日0时过期的秒数
-            const now = new Date()
-            const tomorrow = new Date().setHours(24, 0, 0, 0)
-            const exp = Math.floor((tomorrow - now) / 1000)
-
-            await Redis.set(`lepaoSuccess:${account}`, account, {
-                EX: exp
-            })
-        } catch (error) {
-            this.logger.error(`redis缓存乐跑记录失败: ${error.stack || '未知错误'}`)
-        }
-    }
-
-    async beginLepao(uuid, account, token, uid, school_id, state) {
-        try {
-            this.logger.info(`${account}开始执行乐跑流程`)
-
-            // 检查redis是否存在当天乐跑成功记录
-            const isSuccess = await Redis.get(`lepaoSuccess:${account}`)
-            if (isSuccess)
-                throw new Error('该账号当天已存在成功乐跑记录')
-            const isProgress = await Redis.get(`lepaoProgress:${account}`)
-            if (isProgress)
-                throw new Error('该账号已进入乐跑任务队列，请等待乐跑完成后再进行乐跑操作')
-
-            //已开始乐跑，存入Redis
-            await Redis.set(`lepaoProgress:${account}`, account, {
-                EX: 120
-            })
-
-            const userPermissionSql = 'SELECT vip, lepao_count FROM users WHERE uuid = ?'
-            const userPermissionData = await db.query(userPermissionSql, [uuid])
-            if (!userPermissionData || userPermissionData.length !== 1) {
-                this.logger.error(`${account}无法获取用户信息`)
-                throw new Error('无法获取用户信息，请重试或联系RunForge客服')
-            }
-
-            if (userPermissionData[0].lepao_count < 1) {
-                this.logger.warn(`${account}乐跑次数不足`)
-                throw new Error('用户乐跑次数不足，请购买乐跑套餐！')
-            }
-
-            if (state !== 1) {
-                this.logger.warn(`${account}登录状态异常 state=${state}`)
-                return this.sendFailEmail(account, '乐跑账号登录已过期，请尝试使用登录器重新登录')
-            }
-
-            // 获取路径 ID
-            const path_id = await this.getPath(account, userPermissionData[0].vip)
-
-            // 更换跑区
-            this.logger.info(`${account}开始更换跑区，path_id=${path_id}`)
-            const zoneUrl = this.runpy + '/set_zone'
-
-            // 晚上10点后提前
-            let run_end_time = Math.floor(Date.now() / 1000) - 300 // 提前5分钟
-            let hour = new Date().getHours()
-
-            if (hour < 7)
-                throw new Error('当前不在有效乐跑时间范围内。RunForge支持乐跑时间段为7:00~24:00')
-
-            if (hour >= 22) {
-                this.logger.info(`${account}当前时间为${hour}点，调整run_end_time提前5小时`)
-                run_end_time -= 18000
-            }
-
-            let phoneSql = 'SELECT userAgent, deviceModel FROM lepao_account WHERE student_num = ?'
-            let phoneData = await db.query(phoneSql, [account])
-            let userAgent = (phoneData && phoneData.length > 0 && phoneData[0].userAgent) ? phoneData[0].userAgent : 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/5G Language/zh_CN ABI/arm64'
-            let deviceModel = (phoneData && phoneData.length > 0 && phoneData[0].deviceModel) ? phoneData[0].deviceModel : '2211133C'
-
-            const ossData = { uid, token, school_id, student_id: account, random_id: path_id, run_end_time, userAgent }
-
-            try {
-                const zoneRes = await axios.post(zoneUrl, ossData)
-                const { data } = zoneRes
-
-                this.logger.info(`${account}更换跑区返回结果: ${JSON.stringify(data)}`)
-
-                if (!data || data.status !== 1 || !data.data) {
-                    // 10.17更新，只有明确说明登录失效才会更新状态
-                    if (data && data.info && data.info.includes('请重新登录'))
-                        this.setStatusFail(account)
-                    throw new Error(data?.info || '系统繁忙，请联系客服或稍后再试')
-                }
-            } catch (error) {
-                this.logger.error(`${account}更换跑区失败: ${error.stack || error.message}`)
-                throw error
-            }
-
-            // 临时修复3.12~3.18期间未入库记录，后续可删除
-            // await this.fixRecords(uuid, ossData)
-
-            // 上传 OSS
-            this.logger.info(`${account}开始上传OSS记录`)
-            const ossUrl = this.runpy + '/upload_oss_file'
-            let oss_path, point_data
-
-            try {
-                const ossRes = await axios.post(ossUrl, ossData, {
-                    proxy: false
-                })
-
-                const { data } = ossRes
-                this.logger.info(`${account}上传OSS记录返回结果: ${JSON.stringify(data)}`)
-                if (!data || data.code !== 200 || !data.oss_path || !data.point_data) {
-                    if (data.code == -200) {
-                        this.logger.info(`${account}分配打卡点数量不足，重新执行乐跑流程`)
-                        return this.beginLepao(uuid, account, token, uid, school_id, state)
-                    }
-
-                    throw new Error('系统繁忙，请联系客服或稍后再试')
-                }
-                oss_path = data.oss_path
-                point_data = data.point_data
-                this.logger.info(`${account}上传OSS记录成功！oss_path:${oss_path}`)
-            } catch (error) {
-                // this.setStatusFail(account)
-                this.logger.error(`${account}上传OSS记录失败，请检查登录是否过期。${error.stack || error.message}`)
-                throw new Error('系统繁忙，请联系客服或稍后再试')
-            }
-
-            // 扣除乐跑次数
-            this.logger.info(`${account}开始扣减乐跑次数`)
-            const useLepaoCountSql = 'UPDATE users SET lepao_count = lepao_count - 1 WHERE uuid  = ?'
-            await db.query(useLepaoCountSql, [uuid])
-            this.logger.info(`${account}扣减乐跑次数完成`)
-
-            const lepaoData = {
-                uid,
-                token,
-                school_id,
-                student_id: account,
-                random_id: path_id,
-                record_file: oss_path,
-                run_end_time,
-                point_data,
-                userAgent,
-                deviceModel
-            }
-
-            this.logger.info(`${account}乐跑请求参数构造完成:`)
-            this.logger.info(JSON.stringify(lepaoData))
-
-            // 绑定乐跑数据
-            this.logger.info(`${account}开始绑定乐跑数据`)
-            const lepaoUrl = this.runpy + '/bind_data'
-            try {
-                const lepaoRes = await axios.post(lepaoUrl, lepaoData)
-                const { data } = lepaoRes
-                this.logger.info(`${account}绑定乐跑数据返回结果: ${JSON.stringify(data)}`)
-
-                if (!data || data.status !== 1 || !data.data) {
-                    // 10.17更新，只有明确说明登录失效才会更新状态
-                    if (data && data.info && data.info.includes('请重新登录'))
-                        this.setStatusFail(account)
-
-                    throw new Error(data?.info || '系统繁忙，请联系客服或稍后再试')
-                }
-
-                await this.addRecord(uuid, account, data.data, path_id, point_data)
-
-                // 获取剩余跑步次数
-                const recordData = await this.getRecord(uid, token, school_id, account, userAgent)
-                this.logger.info(`${account}获取剩余跑步次数结果: ${JSON.stringify(recordData)}`)
-
-                let term_num = recordData?.term_num || 0
-                let total_num = recordData?.total_num || 30
-
-                if (data.data.record_failed_reason === '自动确认有效' || data.data.record_failed_reason === '') {
-                    // 成功记录存入Redis
-                    await this.writeRedis(account)
-                    await this.sendSuccessEmail(account, data.data, term_num, total_num)
-                } else {
-                    this.logger.warn(`${account}乐跑失败，原因: ${data.data.record_failed_reason}`)
-
-                    // 已存在记录也存redis
-                    if (data.data.record_failed_reason === '当天关联成绩次数已达到上限')
-                        await this.writeRedis(account)
-
-                    await this.sendFailEmail(account, data.data.record_failed_reason)
-                    await this.lepaoFail(uuid)
-                }
-
-                let recordSql = 'UPDATE lepao_account SET term_num = ?, total_num = ? WHERE student_num = ?'
-                let recordRows = await db.query(recordSql, [term_num, total_num, account])
-                if (!recordRows || recordRows.affectedRows !== 1)
-                    this.logger.warn(`${account}更新乐跑次数失败`)
-                else
-                    this.logger.info(`${account}更新乐跑次数成功 term_num=${term_num}, total_num=${total_num}`)
-            } catch (error) {
-                this.logger.error(`${account}绑定乐跑数据失败: ${error.stack || error.message}`)
-                await this.lepaoFail(uuid)
-                throw error
-            }
-        } catch (error) {
-            this.logger.error(`${account}乐跑流程异常: ${error.stack || error.message}`)
-            await this.sendFailEmail(account, error.message || '系统繁忙，请联系客服或稍后再试')
-        } finally {
-            await Redis.del(`lepaoProgress:${account}`)
-        }
-    }
-
-    async addRecord(uuid, account, result, path_id, point_data) {
-        try {
-            const time = Date.now()
-            this.logger.info(`${account}添加乐跑记录，path_id=${path_id}`)
-            const sql = 'INSERT INTO lepao_record (uuid, time, lepao_account, result, path_id, point_data) VALUES (?, ?, ?, ?, ?, ?)'
-            await db.query(sql, [uuid, time, account, result, path_id, point_data])
-            this.logger.info(`${account}添加乐跑记录成功`)
-        } catch (error) {
-            this.logger.error(`添加乐跑记录失败: ${error.stack || error.message}`)
-        }
-    }
-
-    async sendSuccessEmail(account, lepaoData, term_num, total_num) {
-        try {
-            this.logger.info(`${account}发送乐跑成功邮件`)
-            const emailSql = `
-                SELECT 
-                    a.name, 
-                    a.email, 
-                    a.target_count,
-                    a.notice_type,
-                    e.bot_umo 
-                FROM 
-                    lepao_account a
-                LEFT JOIN
-                    lepao_extra e
-                ON 
-                    a.student_num = e.student_num
-                WHERE
-                    a.student_num = ?
-            `
-
-            const rows = await db.query(emailSql, [account])
-            if (!rows || rows.length === 0) {
-                this.logger.error(`${account}查找用户信息失败`)
-                throw new Error('查找用户信息失败')
-            }
-
-            let data = {
-                ...lepaoData,
-                type: 'lepao_success',
-                umo: rows[0].bot_umo,
-                term_num: rows[0].target_count,
-                total_num,
-                name: rows[0].name,
-                account
-            }
-
-            if (rows[0].notice_type === 'bot' && rows[0].bot_umo) {
-                this.logger.info(`${account}发送乐跑成功Bot通知，UMO=${rows[0].bot_umo}`)
-                const ch = await mq.getChannel(this.messageQueue)
-
-                await ch.assertQueue(this.messageQueue, {
-                    durable: true
-                })
-
-                ch.sendToQueue(
-                    this.messageQueue,
-                    Buffer.from(JSON.stringify(data)),
-                    {
-                        persistent: true,
-                        contentType: 'application/json'
-                    }
-                )
-
-                this.logger.info(`${account}乐跑成功Bot通知发送完成`)
-            } else if (rows[0].notice_type === 'email' && rows[0].email) {
-                await EmailTemplate.lepaoSuccess(rows[0].email, data)
-                this.logger.info(`${account}乐跑成功邮件发送完成`)
-            }
-
-            if (rows[0].target_count !== 0 && total_num >= rows[0].target_count) {
-                this.logger.info(`${account}乐跑目标完成，发送乐跑结束通知并关闭自动乐跑`)
-
-                if (rows[0].notice_type === 'bot' && rows[0].bot_umo) {
-                    this.logger.info(`${account}发送乐跑完成Bot通知，UMO=${rows[0].bot_umo}`)
-                    const ch = await mq.getChannel(this.messageQueue)
-
-                    await ch.assertQueue(this.messageQueue, {
-                        durable: true
-                    })
-
-                    data.type = 'lepao_over'
-
-                    ch.sendToQueue(
-                        this.messageQueue,
-                        Buffer.from(JSON.stringify(data)),
-                        {
-                            persistent: true,
-                            contentType: 'application/json'
-                        }
-                    )
-
-                    this.logger.info(`${account}乐跑完成Bot通知发送完成`)
-                } else if (rows[0].notice_type === 'email' && rows[0].email) {
-                    await EmailTemplate.lepaoOver(rows[0].email, data)
-                    this.logger.info(`${account}乐跑完成邮件发送完成`)
-                }
-
-                let overSql = 'UPDATE lepao_account SET auto_run = 0 WHERE student_num = ?'
-                let overRows = await db.query(overSql, [account])
-                if (!overRows || overRows.affectedRows !== 1)
-                    this.logger.warn(`${account}乐跑结束后关闭自动乐跑失败`)
-                else
-                    this.logger.info(`${account}自动乐跑关闭成功`)
-            }
-        } catch (error) {
-            this.logger.error(`发送成功邮件失败: ${error.stack || error.message}`)
-        }
-    }
-
-    async sendFailEmail(account, reason) {
-        try {
-            this.logger.info(`${account}发送乐跑失败通知，原因: ${reason}`)
-            const emailSql = `
-                SELECT 
-                    a.name, 
-                    a.email, 
-                    a.target_count,
-                    a.notice_type,
-                    e.bot_umo 
-                FROM 
-                    lepao_account a
-                LEFT JOIN
-                    lepao_extra e
-                ON 
-                    a.student_num = e.student_num
-                WHERE
-                    a.student_num = ?
-            `
-            const rows = await db.query(emailSql, [account])
-            if (!rows || rows.length == 0) {
-                this.logger.error(`${account}查找用户邮箱失败`)
-                throw new Error('查找用户邮箱失败')
-            }
-
-            const data = {
-                type: 'lepao_fail',
-                umo: rows[0].bot_umo,
-                name: rows[0].name,
-                account,
-                reason: (reason === 'Request failed with status code 503' || reason === 'Request failed with status code 500') ? '系统繁忙，请稍后再试' : reason
-            }
-
-            if (rows[0].notice_type === 'bot' && rows[0].bot_umo) {
-                this.logger.info(`${account}发送乐跑失败Bot通知，UMO=${rows[0].bot_umo}`)
-                const ch = await mq.getChannel(this.messageQueue)
-
-                await ch.assertQueue(this.messageQueue, {
-                    durable: true
-                })
-
-                ch.sendToQueue(
-                    this.messageQueue,
-                    Buffer.from(JSON.stringify(data)),
-                    {
-                        persistent: true,
-                        contentType: 'application/json'
-                    }
-                )
-
-                this.logger.info(`${account}Bot通知发送完成`)
-            } else if (rows[0].notice_type === 'email' && rows[0].email) {
-                await EmailTemplate.lepaoFail(rows[0].email, data)
-                this.logger.info(`${account}乐跑失败邮件发送完成`)
-            }
-        } catch (error) {
-            this.logger.error(`发送失败邮件失败: ${error.stack || error.message}`)
-        }
-    }
-
-    async lepaoFail(uuid) {
-        try {
-            this.logger.info(`返还用户 ${uuid} 乐跑次数`)
-            const sql = 'UPDATE users SET lepao_count = lepao_count + 1 WHERE uuid  = ?'
-            await db.query(sql, [uuid])
-            this.logger.info(`返还用户 ${uuid} 乐跑次数成功`)
-        } catch (error) {
-            this.logger.error(`返还用户 ${uuid} 乐跑次数时出错: ${error.stack || error.message}`)
-        }
-    }
-
-    async setStatusFail(account) {
-        try {
-            this.logger.info(`${account}设置账号为未启用`)
-            const sql = 'UPDATE lepao_account SET state = 0 WHERE student_num  = ?'
-            await db.query(sql, [account])
-            this.logger.info(`${account}账号状态设置为未启用成功`)
-        } catch (error) {
-            this.logger.error(`设置用户 ${account} state时出错: ${error.stack || error.message}`)
-        }
-    }
-
-    // 修复3.12~3.18期间未入库记录
-    async fixRecords(uuid, reqData) {
-        try {
-            //已开始乐跑，存入Redis
-            const isFix = await Redis.get(`fixRecord:${reqData.student_id}`)
-            if (isFix) {
-                this.logger.info(`${reqData.student_id}已存在修复记录，跳过修复流程`)
-                return
-            }
-
-            await Redis.set(`fixRecord:${reqData.student_id}`, reqData.student_id)
-
-            const countUrl = this.runpy + '/get_term_record'
-
-            const recordRes = await axios.post(countUrl, reqData)
-            const { data } = recordRes
-            if (!data || !data.counts) {
-                this.logger.warn(`修复乐跑记录失败，接口返回异常`)
-                return
-            }
-
-            const count = data.counts || 0
-            let fixSql = 'UPDATE users SET lepao_count = lepao_count - ? WHERE uuid  = ?'
-            await db.query(fixSql, [count, uuid])
-            this.logger.info(`修复乐跑记录成功，扣除用户 ${uuid} 乐跑次数 ${count} 次`)
-
-        } catch (error) {
-            this.logger.error(`修复乐跑记录失败: ${error.stack || error.message}`)
-        }
-    }
-}
-
-const lepao = new Lepao()
-module.exports.lepao = lepao

lib/Lepao/Worker.js

@@ -0,0 +1,1239 @@
+const path = require('path')
+const axios = require('axios')
+const OSS = require('ali-oss')
+const mq = require('../../plugin/mq')
+const { assertRunforgeTaskIngress } = require('../../plugin/mq/runforgeTaskMq')
+const db = require('../../plugin/DataBase/db')
+const Redis = require('../../plugin/DataBase/Redis')
+const EmailTemplate = require('../../plugin/Email/emailTemplate')
+const { URLSearchParams } = require('url')
+const {
+    getPathData,
+    selectCheckpoints,
+    generateCadence
+} = require('../../plugin/Lepao/Path')
+const {
+    dataEncrypt,
+    dataDecrypt,
+    dataSign
+} = require('../../plugin/Lepao/Crypto')
+
+const Logger = require('../Logger')
+
+class Worker {
+    constructor() {
+        this.logger = new Logger(
+            path.join(__dirname, '../logs/LepaoWorker.log'),
+            'INFO'
+        )
+
+        this.handlers = {}
+        this.running = false
+
+        this.baseUrl = 'https://lepao.ctbu.edu.cn/v3/api.php'
+
+        this.taskQueue = 'runforge_task_queue'
+        this.resultQueue = 'runforge_task_result_queue'
+        this.deadQueue = 'runforge_task_dead_queue'
+        this.noticeQueue = 'runforge_message_queue'
+
+        this.channelName = 'lepao_worker'
+
+        this.maxRetry = 3
+        this.timeout = 15000
+
+        this.defaultUserAgent = 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'
+
+        // 调试模式：将 axios 请求走本地代理（例如 charles/fiddler）
+        // 开启方式：设置环境变量 LEPAO_DEBUG_PROXY=1
+        this.debugProxyEnabled = String(process.env.LEPAO_DEBUG_PROXY || '').trim() === '1'
+        this.debugProxyHost = process.env.LEPAO_DEBUG_PROXY_HOST || '127.0.0.1'
+        this.debugProxyPort = Number(process.env.LEPAO_DEBUG_PROXY_PORT || 9000)
+    }
+
+    /* ================= 工具 ================= */
+
+    api(path) {
+        return this.baseUrl + path
+    }
+
+    traceId() {
+        return Date.now() + '_' + Math.random().toString(36).slice(2, 8)
+    }
+
+    sleep(ms) {
+        return new Promise(r => setTimeout(r, ms))
+    }
+
+    isRunSuccess(bindResponse) {
+        const payload = bindResponse?.data
+        if (!bindResponse || bindResponse.status !== 1 || !payload) {
+            return {
+                ok: false,
+                reason: bindResponse?.info || '系统繁忙，请联系客服或稍后再试'
+            }
+        }
+
+        const failedReason = payload.record_failed_reason || ''
+        if (failedReason === '' || failedReason === '自动确认有效') {
+            return { ok: true, payload }
+        }
+
+        return {
+            ok: false,
+            reason: failedReason,
+            payload
+        }
+    }
+
+    extractApiErrorMessage(name, result) {
+        if (!result) {
+            this.logger.error(`${name} 接口无响应数据: ${this.safeStringify(result)}`)
+            return `系统繁忙，请联系客服或稍后再试`
+        }
+
+        const candidates = [
+            result.info,
+            result.msg,
+            result.message,
+            result?.data?.info,
+            result?.data?.msg,
+            result?.data?.message,
+            result?.data?.record_failed_reason
+        ]
+
+        const reason = candidates.find(v => typeof v === 'string' && v.trim() !== '')
+        if (reason) {
+            return reason
+        }
+
+        if (result.code !== undefined || result.status !== undefined) {
+            this.logger.error(`${name} 接口返回异常: ${this.safeStringify(result)}`)
+            return `系统繁忙，请联系客服或稍后再试`
+        }
+        return `系统繁忙，请联系客服或稍后再试`
+    }
+
+    async markLoginExpired(account) {
+        if (!account) return
+        try {
+            const sql = 'UPDATE lepao_account SET state = 0 WHERE student_num = ?'
+            await db.query(sql, [account])
+            this.logger.warn(`${account} 登录状态已失效，已自动更新为未登录`)
+        } catch (error) {
+            this.logger.error(`更新账号登录状态失败：${error.stack || error}`)
+        }
+    }
+
+    async writeSuccessRedis(account) {
+        if (!account) return
+        try {
+            const now = new Date()
+            const tomorrow = new Date().setHours(24, 0, 0, 0)
+            const exp = Math.floor((tomorrow - now) / 1000)
+            await Redis.set(`lepaoSuccess:${account}`, account, { EX: exp })
+        } catch (error) {
+            this.logger.error(`写入乐跑成功缓存失败: ${error.stack || error}`)
+        }
+    }
+
+    async addLepaoRecord(uuid, account, result, pathId, pointData) {
+        if (!uuid || !account || !result || !pathId) return
+        try {
+            const time = Date.now()
+            const sql = 'INSERT INTO lepao_record (uuid, time, lepao_account, result, path_id, point_data, state) VALUES (?, ?, ?, ?, ?, ?, ?)'
+            await db.query(sql, [uuid, time, account, result, pathId, JSON.stringify(pointData || []), 1])
+        } catch (error) {
+            this.logger.error(`写入乐跑记录失败: ${error.stack || error}`)
+        }
+    }
+
+    async syncRunCount(req, ctx) {
+        try {
+            const sid = req?.student_id || req?.account
+            if (
+                req?.uid == null ||
+                req?.token == null ||
+                String(req.token).trim() === '' ||
+                req?.school_id == null ||
+                !sid
+            ) {
+                return { ok: false, reason: '缺少同步乐跑次数所需凭证' }
+            }
+
+            const recordData = await this.handlers['lepao.getRecord'](req, ctx)
+            const data = recordData?.data
+            if (!data) {
+                return { ok: false, reason: 'getRecord 无有效 data' }
+            }
+
+            const term_num = Number(data.term_num ?? 30)
+            const total_num = Number(data.total_num ?? 0)
+
+            const sql = 'UPDATE lepao_account SET term_num = ?, total_num = ? WHERE student_num = ?'
+            const rows = await db.query(sql, [term_num, total_num, req.account])
+            if (!rows || rows.affectedRows !== 1) {
+                this.logger.warn(`${req.account}更新乐跑次数失败`)
+                return { ok: false, reason: '数据库更新 lepao_account 失败', term_num, total_num }
+            }
+            this.logger.info(`${req.account}更新乐跑次数成功 term_num=${term_num}, total_num=${total_num}`)
+            return { ok: true, term_num, total_num }
+        } catch (error) {
+            this.logger.warn(`${req?.account || 'unknown'}同步乐跑次数失败: ${error.message || error}`)
+            return { ok: false, reason: error.message || String(error) }
+        }
+    }
+
+    lepaoTimestamp() {
+        return Number((Date.now() / 1000).toFixed(3))
+    }
+
+    axiosProxyConfig() {
+        if (!this.debugProxyEnabled) {
+            return { proxy: false }
+        }
+        this.logger.info(`使用本地代理: ${this.debugProxyHost}:${this.debugProxyPort}`)
+        return {
+            proxy: {
+                host: this.debugProxyHost,
+                port: this.debugProxyPort,
+                protocol: 'http'
+            }
+        }
+    }
+
+    async enqueueTask(channel, type, data, options = {}) {
+        const payload = {
+            id: options.id || this.traceId(),
+            type,
+            data,
+            retry: options.retry ?? 0
+        }
+
+        await channel.sendToQueue(
+            this.taskQueue,
+            Buffer.from(JSON.stringify(payload)),
+            { persistent: true, contentType: 'application/json' }
+        )
+
+        return payload.id
+    }
+
+    async withTimeout(promise, name) {
+        return Promise.race([
+            promise,
+            new Promise((_, reject) =>
+                setTimeout(() => reject(new Error(`${name} 超时`)), this.timeout)
+            )
+        ])
+    }
+
+    async retry(fn, name) {
+        let lastErr
+
+        for (let i = 0; i < this.maxRetry; i++) {
+            try {
+                return await fn()
+            } catch (err) {
+                lastErr = err
+                if (!this.isRetryableTaskError(err)) {
+                    throw err
+                }
+                this.logger.warn(`[RETRY] ${name} 第${i + 1}次失败`)
+
+                await this.sleep(1000 * (i + 1)) // 指数退避
+            }
+        }
+
+        throw lastErr
+    }
+
+    isNetworkError(err) {
+        if (!err) return false
+        if (err.code && ['ECONNRESET', 'ECONNABORTED', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN'].includes(err.code)) {
+            return true
+        }
+        if (err.isAxiosError && !err.response) return true
+        const msg = (err.message || '').toLowerCase()
+        return msg.includes('timeout') || msg.includes('network')
+    }
+
+    isRetryableTaskError(err) {
+        if (!err) return false
+        if (err.retryable === true) return true
+        if (this.isNetworkError(err)) return true
+        return ['PATH_SELECT_FAILED', 'CHECKPOINT_FETCH_FAILED', 'CHECKPOINT_INSUFFICIENT'].includes(err.code)
+    }
+
+    safeStringify(obj) {
+        const seen = new WeakSet();
+        return JSON.stringify(obj, (key, value) => {
+            if (typeof value === 'object' && value !== null) {
+                if (seen.has(value)) return '[Circular]';
+                seen.add(value);
+            }
+            return value;
+        })
+    }
+
+    log(traceId, type, msg, data) {
+        this.logger.info(`[${traceId}] [${type}] ${msg} ${data ? this.safeStringify(data) : ''}`)
+    }
+
+    logErr(traceId, msg, err) {
+        this.logger.error(`[${traceId}] ${msg} ${err.stack || err}`)
+    }
+
+    async request(traceId, name, url, raw, headers = {}) {
+        return this.retry(async () => {
+            this.log(traceId, 'REQ', name, raw)
+
+            const mergedHeaders = {
+                'Content-Type': 'application/x-www-form-urlencoded',
+                'Accept': '*/*',
+                'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
+                'Accept-Encoding': 'gzip, deflate, br',
+                'Referer': 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
+                ...headers
+            }
+            if (!mergedHeaders['User-Agent']) {
+                mergedHeaders['User-Agent'] = this.defaultUserAgent
+            }
+
+            const form = new URLSearchParams()
+            form.append('ostype', '5')
+            form.append('data', dataEncrypt(JSON.stringify(raw)))
+
+            const res = await this.withTimeout(
+                axios.post(
+                    url,
+                    form,
+                    {
+                        headers: mergedHeaders,
+                        ...this.axiosProxyConfig()
+                    }
+                ),
+                name
+            )
+
+            let result = res.data
+
+            if (result?.data && result?.is_encrypt === 1) {
+                result.data = JSON.parse(dataDecrypt(result.data))
+            }
+
+            this.log(traceId, 'RES', name, result)
+
+            // 除 bindData 外，其余调用若接口已明确返回失败，直接抛出该失败原因
+            // bindData 需要保留完整响应由 isRunSuccess 统一判定。
+            if (name !== 'bindData') {
+                const hasCode = result && Object.prototype.hasOwnProperty.call(result, 'code')
+                const hasStatus = result && Object.prototype.hasOwnProperty.call(result, 'status')
+                const failedByCode = hasCode && Number(result.code) !== 1 && Number(result.code) !== 200
+                const failedByStatus = hasStatus && Number(result.status) !== 1
+                if (failedByCode || failedByStatus) {
+                    const message = this.extractApiErrorMessage(name, result)
+                    const err = new Error(message)
+                    // 学习 Lepao.js：若明确提示重新登录，自动标记账号失效
+                    if (message.includes('重新登录')) {
+                        await this.markLoginExpired(raw?.student_num)
+                    }
+                    // 接口已返回业务错误，禁止重试
+                    err.retryable = false
+                    throw err
+                }
+            }
+
+            return result
+        }, name)
+    }
+
+    /**
+     * 累计完成次数 >= 跑友目标（且目标>0）时：关闭 auto_run，并发送乐跑目标完成邮件 / Bot 通知
+     */
+    async handleLepaoTargetComplete(account, user, totalNum, traceId) {
+        const target = Number(user?.target_count) || 0
+        const total = Number(totalNum) || 0
+        if (target <= 0 || total < target) return
+
+        try {
+            const up = await db.query(
+                'UPDATE lepao_account SET auto_run = 0 WHERE student_num = ? AND auto_run = 1',
+                [account]
+            )
+            if (up?.affectedRows !== 1) {
+                return
+            }
+            this.logger.info(
+                `${account} 已达目标次数（${total}/${target}），关闭自动乐跑`
+            )
+        } catch (e) {
+            this.logger.error(`关闭自动乐跑失败 ${account}: ${e.message || e}`)
+            return
+        }
+
+        const noticeType = user.notice_type || 'none'
+        const overPayload = {
+            type: 'lepao_over',
+            umo: user.bot_umo,
+            name: user.name,
+            account,
+            total_num: total,
+            target_count: target,
+            traceId
+        }
+
+        if (noticeType === 'email' && user.email) {
+            try {
+                await EmailTemplate.lepaoOver(user.email, {
+                    name: user.name,
+                    account
+                })
+            } catch (e) {
+                this.logger.error(`lepaoOver 邮件发送失败: ${e.message || e}`)
+            }
+        }
+
+        if (noticeType === 'bot' && user.bot_umo) {
+            try {
+                const ch = await mq.getChannel(this.noticeQueue)
+                await ch.assertQueue(this.noticeQueue, { durable: true })
+                ch.sendToQueue(
+                    this.noticeQueue,
+                    Buffer.from(JSON.stringify(overPayload)),
+                    {
+                        persistent: true,
+                        contentType: 'application/json'
+                    }
+                )
+            } catch (e) {
+                this.logger.error(`lepao_over Bot 通知失败: ${e.message || e}`)
+            }
+        }
+    }
+
+    register(type, handler) {
+        this.handlers[type] = handler
+        this.logger.info(`注册任务: ${type}`)
+    }
+
+    /* ================= 业务 ================= */
+
+    initHandlers() {
+        /* ---------------- 开始乐跑 ---------------- */
+        this.register('lepao.startRun', async (req, ctx) => {
+            const traceId = ctx.traceId
+            const maxPathRetry = 20  // 自动获取路径失败最大重试次数
+            let pathRetry = 0
+            let pointData = null
+            let ossPath = null
+            let userData = null
+            let pathId = null
+            let runZoneId = 0
+            let bindRes = null
+
+            try {
+                // 检查redis是否存在当天乐跑成功记录
+                const isSuccess = await Redis.get(`lepaoSuccess:${req.account}`)
+                if (isSuccess)
+                    throw new Error('该账号当天已乐跑成功！请勿重复乐跑')
+
+                userData = await this.handlers['lepao.getUserData'](req, ctx)
+
+                // 立刻合并账号凭证，保证后续任意 throw 时 finally 里 syncRunCount 不会用空 token 调 getRecord
+                req = {
+                    ...req,
+                    ...userData,
+                    student_id: req.account
+                }
+
+                // 进入乐跑进程后写入进行中缓存
+                const progressKey = `lepaoProgress:${req.account}`
+                const inProgress = await Redis.get(progressKey)
+                if (inProgress) {
+                    throw new Error('该账号已进入乐跑任务队列，请等待乐跑完成后再进行乐跑操作')
+                }
+                await Redis.set(progressKey, req.account, { EX: 1800 })
+
+                // 晚上10点后提前
+                let run_end_time = Math.floor(Date.now() / 1000) - 300 // 提前5分钟
+                let hour = new Date().getHours()
+
+                if (hour < 7)
+                    throw new Error('当前不在有效乐跑时间范围内。RunForge支持乐跑时间段为7:00~24:00')
+
+                if (hour >= 22) {
+                    this.logger.info(`${req.account}当前时间为${hour}点，调整run_end_time提前5小时`)
+                    run_end_time -= 18000
+                }
+
+                req = {
+                    ...req,
+                    run_end_time
+                }
+
+                // 1.5️⃣ 乐跑开始前扣减次数（失败会返还，且有幂等保护）
+                await this.handlers['lepao.consumeCount']({
+                    account: req.account,
+                    uuid: userData?.create_user
+                }, ctx)
+
+                while (pathRetry < maxPathRetry) {
+                    try {
+                        // 2️⃣ 获取路径（仅路径选择失败时重试）
+                        const pathRes = await this.handlers['lepao.getPath'](req, ctx)
+                        pathId = pathRes.path_id
+
+                        // 3️⃣ 切换跑区
+                        const zoneRes = await this.handlers['lepao.setZone']({ ...req, random_id: pathId }, ctx)
+                        runZoneId = zoneRes?.run_zone_id || 0
+
+                        // 4️⃣ 上传 OSS 文件、生成打卡点
+                        const uploadRes = await this.handlers['lepao.uploadOssFile']({ ...req, random_id: pathId }, ctx)
+                        ossPath = uploadRes.oss_path
+                        pointData = uploadRes.point_data
+
+                        if (!pointData) {
+                            pathRetry++
+                            this.logger.warn(`[${traceId}] 打卡点不满足要求，重新获取路径 第${pathRetry}次`)
+                            continue
+                        }
+
+                        // 打卡点符合要求，跳出循环
+                        break
+                    } catch (err) {
+                        if (!this.isRetryableTaskError(err)) {
+                            throw err
+                        }
+                        this.logger.warn(`[${traceId}] 可重试错误，重新获取路径 第${pathRetry + 1}次，原因：${err.message}`)
+                        pathRetry++
+                        await this.sleep(1000 * pathRetry)
+                    }
+                }
+
+                if (!pointData) {
+                    throw new Error('打卡点获取失败，乐跑任务终止')
+                }
+
+                // 5️⃣ 提交跑步数据
+                bindRes = await this.handlers['lepao.bindData']({
+                    ...req,
+                    random_id: pathId,
+                    run_zone_id: runZoneId,
+                    record_file: ossPath,
+                    point_data: pointData
+                }, ctx)
+
+                // 绑定接口有返回即入库
+                if (bindRes && bindRes.data) {
+                    await this.addLepaoRecord(userData?.create_user, req.account, bindRes.data, pathId, pointData)
+                }
+
+                const runResult = this.isRunSuccess(bindRes)
+                if (runResult.ok || runResult.reason === '当天关联成绩次数已达到上限') {
+                    await this.writeSuccessRedis(req.account)
+                }
+                if (!runResult.ok) {
+                    throw new Error(runResult.reason)
+                }
+
+                // 同步乐跑次数（通知里要带 total_num / term_num，与 getRecord 一致）
+                const syncResult = await this.syncRunCount(req, ctx)
+                if (!syncResult?.ok) {
+                    this.logger.error(
+                        `[${traceId}] 同步乐跑次数失败：${syncResult?.reason || 'unknown'}`
+                    )
+                }
+
+                // 6️⃣ 发送通知（把同步后的学期次数、累计完成次数传给 Bot / 邮件）
+                if (ctx.channel) {
+                    await this.enqueueTask(
+                        ctx.channel,
+                        'lepao.sendNotice',
+                        {
+                            account: req.account,
+                            success: true,
+                            data: runResult.payload,
+                            traceId,
+                            total_num: syncResult?.ok ? syncResult.total_num : undefined
+                        },
+                        { id: `${traceId}:notice:success` }
+                    )
+                }
+
+                return { traceId, ossPath, pointData, bindRes }
+
+            } catch (err) {
+                this.logger.error(`[${traceId}] 乐跑流程失败:`, err)
+
+                // 若已扣减次数，则失败时返还（幂等）
+                try {
+                    await this.handlers['lepao.refundCount']({
+                        account: req.account,
+                        uuid: userData?.create_user
+                    }, ctx)
+                } catch (e) {
+                    this.logger.error(`[${traceId}] 返还乐跑次数失败：${e.stack || e}`)
+                }
+
+                if (ctx.channel) {
+                    await this.enqueueTask(ctx.channel, 'lepao.sendNotice', {
+                        account: req.account,
+                        success: false,
+                        reason: err.message || '未知错误',
+                        traceId
+                    }, { id: `${traceId}:notice:fail` })
+                }
+
+                // 将失败消息发送到结果队列或死信队列
+                if (ctx.channel) {
+                    await this.sendResult(ctx.channel, {
+                        id: req.taskId,
+                        success: false,
+                        error: err.message
+                    })
+                    await ctx.channel.sendToQueue(
+                        this.deadQueue,
+                        Buffer.from(JSON.stringify({ ...req, error: err.message })),
+                        { persistent: true }
+                    )
+                }
+                throw err
+            } finally {
+                await Redis.del(`lepaoProgress:${req.account}`)
+            }
+        })
+
+        /* ---------------- 发送通知（独立 MQ 任务） ---------------- */
+        this.register('lepao.sendNotice', async (req, ctx) => {
+            const { account, success, data, reason, traceId, total_num: totalNumArg } = req || {}
+
+            if (!account) {
+                throw new Error('发送通知失败：缺少 account')
+            }
+
+            const emailSql = `
+                SELECT 
+                    a.name, 
+                    a.email, 
+                    a.target_count,
+                    a.notice_type,
+                    e.bot_umo
+                FROM 
+                    lepao_account a
+                LEFT JOIN
+                    lepao_extra e
+                ON 
+                    a.student_num = e.student_num
+                WHERE
+                    a.student_num = ?
+            `
+            const rows = await db.query(emailSql, [account])
+            if (!rows || rows.length === 0) {
+                throw new Error('发送通知失败：未找到用户通知配置')
+            }
+
+            const user = rows[0]
+            const noticeType = user.notice_type || 'none'
+
+            let totalForNotice = totalNumArg
+            if (
+                success &&
+                (totalForNotice === undefined || totalForNotice === null)
+            ) {
+                const accRows = await db.query(
+                    'SELECT total_num FROM lepao_account WHERE student_num = ?',
+                    [account]
+                )
+                if (accRows && accRows[0]) {
+                    totalForNotice = accRows[0].total_num
+                }
+            }
+            if (totalForNotice === undefined || totalForNotice === null) {
+                totalForNotice = 0
+            }
+            totalForNotice = Number(totalForNotice)
+            const targetCount = Number(user.target_count) || 0
+
+            const payload = success ? {
+                ...(data && typeof data === 'object' ? data : {}),
+                type: 'lepao_success',
+                umo: user.bot_umo,
+                total_num: totalForNotice,
+                target_count: targetCount,
+                name: user.name,
+                account,
+                traceId
+            } : {
+                type: 'lepao_fail',
+                umo: user.bot_umo,
+                name: user.name,
+                account,
+                reason,
+                traceId
+            }
+
+            const afterSuccessNotify = async () => {
+                if (success) {
+                    await this.handleLepaoTargetComplete(account, user, totalForNotice, traceId)
+                }
+            }
+
+            if (noticeType === 'bot' && user.bot_umo) {
+                const ch = await mq.getChannel(this.noticeQueue)
+                await ch.assertQueue(this.noticeQueue, { durable: true })
+                ch.sendToQueue(
+                    this.noticeQueue,
+                    Buffer.from(JSON.stringify(payload)),
+                    {
+                        persistent: true,
+                        contentType: 'application/json'
+                    }
+                )
+                await afterSuccessNotify()
+                return { delivered: true, via: 'bot' }
+            }
+
+            if (noticeType === 'email' && user.email) {
+                if (success) {
+                    await EmailTemplate.lepaoSuccess(user.email, payload)
+                    await afterSuccessNotify()
+                    return { delivered: true, via: 'email' }
+                }
+
+                await EmailTemplate.lepaoFail(user.email, {
+                    name: user.name,
+                    account,
+                    reason: reason || '系统繁忙，请联系客服或稍后再试',
+                    traceId
+                })
+                return { delivered: true, via: 'email' }
+            }
+
+            await afterSuccessNotify()
+            return { delivered: false, via: 'none' }
+        })
+
+        /* ---------------- 扣减次数 ---------------- */
+        this.register('lepao.consumeCount', async (req, ctx) => {
+            const account = req?.account
+            const uuid = req?.uuid
+            if (!uuid) {
+                throw new Error('扣减乐跑次数失败：缺少 uuid')
+            }
+
+            // 幂等：同一 taskId 只扣一次
+            const consumeKey = `lepao:consume:${ctx?.taskId || ctx?.traceId || account || uuid}`
+            const existed = await Redis.get(consumeKey)
+            if (existed) {
+                return true
+            }
+
+            this.logger.info(`${account || uuid}开始扣减乐跑次数`)
+            const useLepaoCountSql = 'UPDATE users SET lepao_count = lepao_count - 1 WHERE uuid  = ?'
+            const r = await db.query(useLepaoCountSql, [uuid])
+            if (!r || r.affectedRows !== 1) {
+                throw new Error('扣减乐跑次数失败：数据库更新失败')
+            }
+            this.logger.info(`${account || uuid}扣减乐跑次数完成`)
+
+            await Redis.set(consumeKey, '1', { EX: 3600 })
+            return true
+        })
+
+        /* ---------------- 返还次数（失败时执行） ---------------- */
+        this.register('lepao.refundCount', async (req, ctx) => {
+            const account = req?.account
+            const uuid = req?.uuid
+            if (!uuid) {
+                return true
+            }
+
+            const baseKey = `${ctx?.taskId || ctx?.traceId || account || uuid}`
+            const consumeKey = `lepao:consume:${baseKey}`
+            const refundKey = `lepao:refund:${baseKey}`
+
+            const consumed = await Redis.get(consumeKey)
+            if (!consumed) {
+                return true
+            }
+            const refunded = await Redis.get(refundKey)
+            if (refunded) {
+                return true
+            }
+
+            this.logger.info(`${account || uuid}开始返还乐跑次数`)
+            const sql = 'UPDATE users SET lepao_count = lepao_count + 1 WHERE uuid  = ?'
+            const r = await db.query(sql, [uuid])
+            if (!r || r.affectedRows !== 1) {
+                throw new Error('返还乐跑次数失败：数据库更新失败')
+            }
+            this.logger.info(`${account || uuid}返还乐跑次数完成`)
+            await Redis.set(refundKey, '1', { EX: 3600 })
+            return true
+        })
+
+        this.register('lepao.getUserData', async (req, ctx) => {
+            const account = req.account
+            this.logger.info(`${account}开始获取用户数据`)
+            const accountSql = `
+                SELECT
+                    u.uuid, 
+                    u.lepao_count,
+                    l.create_user,
+                    l.name, 
+                    l.student_num,
+                    l.area, 
+                    l.sex, 
+                    l.state,  
+                    l.token, 
+                    l.uid, 
+                    l.school_id, 
+                    l.userAgent,
+                    l.deviceModel,
+                    l.notice_type, 
+                    l.email,
+                    e.bot_account
+                FROM 
+                    lepao_account l
+                LEFT JOIN
+                    users u
+                ON
+                    l.create_user = u.uuid
+                LEFT JOIN
+                    lepao_extra e
+                ON 
+                    l.student_num = e.student_num
+                WHERE 
+                    l.student_num = ?
+            `
+            const rows = await db.query(accountSql, [account])
+            if (!rows || rows.length === 0) {
+                this.logger.error(`${account}无法获取账号数据`)
+                throw new Error('无法获取账号数据，请联系客服或稍后再试')
+            }
+
+            let userData = rows[0]
+
+            if (!userData.create_user || !userData.uuid) {
+                this.logger.warn(`${account}账号状态异常`)
+                throw new Error('当前账号状态异常，请联系客服')
+            }
+
+            if (userData.state !== 1) {
+                this.logger.warn(`${account}登录状态异常 state=${userData.state}`)
+                throw new Error('乐跑账号登录已过期，请尝试使用登录器重新登录')
+            }
+
+            if (userData.lepao_count < 1) {
+                this.logger.warn(`${account}乐跑次数不足`)
+                throw new Error('用户乐跑次数不足，请购买乐跑次数后重试！')
+            }
+
+            if (!userData.userAgent)
+                userData.userAgent = 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'
+
+            if (!userData.deviceModel)
+                userData.deviceModel = '2211133C'
+
+            return userData
+        })
+
+        this.register('lepao.getPath', async (req, ctx) => {
+            const account = req.account
+            this.logger.info(`${account}开始获取路径`)
+            const accountSql = 'SELECT area, sex FROM lepao_account WHERE student_num = ?'
+            const rows = await db.query(accountSql, [account])
+            if (!rows || rows.length === 0) {
+                this.logger.error(`${account}无法获取账号数据`)
+                throw new Error('无法获取账号数据')
+            }
+
+            const { area, sex } = rows[0]
+
+            let max = 4.00
+            let min = 2.00
+            if (sex === 2) {
+                max = 2.00
+                min = 1.60
+            }
+
+            this.logger.info(`${account}路径参数: area=${area ?? '随机'}, max_distance=${max}, min_distance=${min}`)
+
+            let pathSql = 'SELECT id FROM path_data WHERE state = 1 AND distance < ? AND distance > ? '
+            const pathParams = [max, min]
+
+            if (area) {
+                pathSql += ' AND run_zone_name = ?'
+                pathParams.push(area)
+            }
+
+            pathSql += ' ORDER BY count ASC LIMIT 1'
+
+            const paths = await db.query(pathSql, pathParams)
+            if (!paths || paths.length === 0) {
+                this.logger.error(`${account}未找到符合条件的路线`)
+                const err = new Error('未找到符合条件的路线，请改变路径选择条件')
+                err.code = 'PATH_SELECT_FAILED'
+                err.retryable = true
+                throw err
+            }
+
+            const randomPath = paths[0]
+
+            const updateSql = 'UPDATE path_data SET count = count + 1 WHERE id = ?'
+            await db.query(updateSql, [randomPath.id])
+
+            this.logger.info(`${account}路径选中id=${randomPath.id}，计数加1成功`)
+
+            return { path_id: randomPath.id }
+        })
+
+        /* ---------------- 获取跑步记录 ---------------- */
+        this.register('lepao.getRecord', async (req, ctx) => {
+            const now = this.lepaoTimestamp()
+            const raw = {
+                uid: req.uid,
+                token: req.token,
+                school_id: req.school_id,
+                term_id: 0,
+                course_id: 0,
+                class_id: 0,
+                student_num: req.student_id,
+                card_id: req.student_id,
+                timestamp: now,
+                version: 1,
+                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+                ostype: 5
+            }
+            raw.sign = dataSign(raw)
+
+            return this.request(
+                ctx.traceId,
+                'getRecord',
+                this.api('/Run2/beforeRunV260'),
+                raw,
+                {
+                    'User-Agent': req.userAgent,
+                    'charset': 'utf-8',
+                    'Referer': 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
+                }
+            )
+        })
+
+        /* ---------------- 切换跑区 ---------------- */
+        this.register('lepao.setZone', async (req, ctx) => {
+            const runZoneMap = {
+                '兰花湖校区跑区': 2,
+                '主校区北跑区': 3,
+                '主校区南跑区': 5,
+                '重庆工商大学茶园校区': 6
+            }
+
+            const record = await db.query(
+                'SELECT run_zone_name FROM path_data WHERE id = ?',
+                [req.random_id]
+            )
+            if (!record || record.length === 0) {
+                throw new Error('跑区不存在')
+            }
+
+            const runZoneId = runZoneMap[record[0].run_zone_name]
+            if (!runZoneId) throw new Error('跑区不存在')
+
+            const raw = {
+                uid: req.uid,
+                token: req.token,
+                school_id: req.school_id,
+                term_id: 0,
+                course_id: 0,
+                class_id: 0,
+                student_num: req.student_id,
+                card_id: req.student_id,
+                timestamp: this.lepaoTimestamp(),
+                version: 1,
+                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+                ostype: 5,
+                run_zone_id: String(runZoneId)
+            }
+            raw.sign = dataSign(raw)
+
+            await this.request(
+                ctx.traceId,
+                'setZone',
+                this.api('/Run/setRunZone'),
+                raw
+            )
+            return { run_zone_id: runZoneId }
+        })
+
+        /* ---------------- 获取 OSS STS ---------------- */
+        this.register('lepao.getOssSts', async (req, ctx) => {
+            const raw = {
+                uid: req.uid,
+                token: req.token,
+                school_id: req.school_id,
+                term_id: 0,
+                course_id: 0,
+                class_id: 0,
+                student_num: req.student_id,
+                card_id: req.student_id,
+                timestamp: this.lepaoTimestamp(),
+                version: 1,
+                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+                ostype: 5
+            }
+            raw.sign = dataSign(raw)
+
+            const res = await this.request(
+                ctx.traceId,
+                'getOssSts',
+                this.api('/WpIndex/getOssSts'),
+                raw
+            )
+
+            return res.data
+        })
+
+        /* ---------------- 上传 OSS 文件 ---------------- */
+        this.register('lepao.uploadOssFile', async (req, ctx) => {
+            const pathRow = await db.query(
+                'SELECT * FROM path_data WHERE id=?',
+                [req.random_id]
+            )
+            if (!pathRow || pathRow.length === 0) {
+                throw new Error('路径数据不存在')
+            }
+            const pathData = pathRow[0]
+
+            // 处理跑步路径
+            const newPathData = getPathData(pathData.data, req.run_end_time, pathData.time)
+            const pathResult = dataEncrypt(JSON.stringify(newPathData))
+
+            // 获取跑步规则参数
+            const runRule = await this.handlers['lepao.getRecord'](req, ctx)
+            const ruleData = runRule?.data
+            if (!ruleData?.run_line_info?.point_list || !ruleData?.time_rule_arr?.length) {
+                const err = new Error('获取打卡点规则失败')
+                err.code = 'CHECKPOINT_FETCH_FAILED'
+                err.retryable = true
+                throw err
+            }
+            const check_points = ruleData.run_line_info.point_list
+            let min_log_num = ruleData.time_rule_arr[0]?.min_log_num || 4
+            const point_update_distance = parseFloat(ruleData.run_line_info.point_update_distance || 0) * 1000
+            const log_max_distance = Number(ruleData.run_line_info.log_max_distance || 0)
+
+            // 生成打卡点
+            const point_data = selectCheckpoints(newPathData, check_points, min_log_num, point_update_distance, log_max_distance, req.run_end_time, pathData.time)
+            if (!point_data) {
+                this.logger.warn(`[RETRY] 打卡点数量不足，重新更换路径`)
+                const err = new Error('打卡点数量不足')
+                err.code = 'CHECKPOINT_INSUFFICIENT'
+                err.retryable = true
+                throw err
+            }
+
+            const sts = await this.handlers['lepao.getOssSts'](req, ctx)
+            if (!sts?.bucket || !sts?.AccessKeyId || !sts?.AccessKeySecret || !sts?.SecurityToken) {
+                throw new Error('获取 OSS STS 失败')
+            }
+
+            const now = new Date()
+            const yyyy = now.getFullYear()
+            const mm = String(now.getMonth() + 1).padStart(2, '0')
+            const dd = String(now.getDate()).padStart(2, '0')
+            const formattedToday = `${yyyy}-${mm}-${dd}`
+            const boundary = String(Date.now())
+            const timestamp = String(Date.now())
+            const ossPath = `Public/Upload/file/run_record/${boundary.slice(-3)}/${formattedToday}/${timestamp}-${Math.floor(Math.random() * 150)}.txt`
+            const client = new OSS({
+                bucket: sts.bucket,
+                region: sts.region || 'oss-cn-hangzhou',
+                accessKeyId: sts.AccessKeyId,
+                accessKeySecret: sts.AccessKeySecret,
+                stsToken: sts.SecurityToken,
+                secure: true
+            })
+            await client.put(ossPath, Buffer.from(pathResult, 'utf-8'))
+
+            return { oss_path: ossPath, point_data: point_data }
+        })
+
+        /* ---------------- 提交跑步数据 ---------------- */
+        this.register('lepao.bindData', async (req, ctx) => {
+            if (req?.random_id === undefined || req?.random_id === null || req?.random_id === '') {
+                throw new Error('提交跑步数据失败：缺少 random_id')
+            }
+
+            const pathRow = await db.query(
+                'SELECT * FROM path_data WHERE id=?',
+                [req.random_id]
+            )
+            if (!pathRow || pathRow.length === 0) {
+                throw new Error(`提交跑步数据失败：未找到路径数据(random_id=${req.random_id})`)
+            }
+            const pathData = pathRow[0]
+
+            const distance = Number(Number(pathData.distance || 0).toFixed(2))
+            const stepData = generateCadence(distance, pathData.time)
+            const stepInfo = JSON.stringify({ interval: 60, list: stepData.cadence_list })
+
+            let points = req.point_data.map(({ address, jingwei, ...rest }) => rest)
+            points = JSON.stringify(points)
+
+            const data = {
+                uid: req.uid,
+                token: req.token,
+                school_id: req.school_id,
+                term_id: 1,
+                course_id: 0,
+                class_id: 0,
+                student_num: req.student_id,
+                card_id: req.student_id,
+                timestamp: this.lepaoTimestamp(),
+                version: 1,
+                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+                ostype: 5,
+                game_id: String(req.run_zone_id || 0),
+                start_time: req.run_end_time - Number(pathData.time),
+                end_time: req.run_end_time,
+                distance,
+                record_img: "",
+                log_data: points,
+                file_img: "",
+                is_running_area_valid: 1,
+                mobileDeviceId: 1,
+                mobileModel: req.deviceModel,
+                step_info: stepInfo,
+                step_num: stepData.total_steps,
+                used_time: pathData.time,
+                mobileOsVersion: 1,
+                record_file: req.record_file
+            }
+
+            data.sign = dataSign(data)
+
+            return this.request(
+                ctx.traceId,
+                'bindData',
+                this.api('/Run/stopRunV278'),
+                data
+            )
+        })
+    }
+
+    /* ================= Worker核心 ================= */
+
+    async start() {
+        if (this.running) return
+        this.running = true
+
+        this.logger.info('Worker 启动中...')
+
+        try {
+            this.initHandlers()
+
+            const channel = await mq.getChannel(this.channelName)
+
+            await channel.prefetch(5)
+
+            await assertRunforgeTaskIngress(channel, this.logger)
+            await channel.assertQueue(this.resultQueue, { durable: true })
+            await channel.assertQueue(this.deadQueue, { durable: true })
+
+            await channel.consume(this.taskQueue, async (msg) => {
+                if (!msg) return
+
+                let content
+
+                try {
+                    content = JSON.parse(msg.content.toString())
+                } catch {
+                    return channel.ack(msg)
+                }
+
+                const { id, type, data, retry = 0 } = content
+
+                const traceId = this.traceId()
+
+                const handler = this.handlers[type]
+
+                if (!handler) {
+                    this.log(traceId, 'ERROR', '未知任务', { type })
+                    return channel.ack(msg)
+                }
+
+                try {
+                    const result = await this.withTimeout(
+                        handler(data, { traceId, channel, taskId: id }),
+                        type
+                    )
+
+                    await this.sendResult(channel, {
+                        id,
+                        success: true,
+                        result
+                    })
+
+                    this.log(traceId, 'DONE', `任务完成 ${type}`)
+                    channel.ack(msg)
+
+                } catch (err) {
+                    this.logErr(traceId, `任务失败 ${type}`, err)
+
+                    if (retry < this.maxRetry && this.isRetryableTaskError(err)) {
+                        // 重试
+                        await channel.sendToQueue(
+                            this.taskQueue,
+                            Buffer.from(JSON.stringify({
+                                ...content,
+                                retry: retry + 1
+                            })),
+                            { persistent: true }
+                        )
+
+                        this.log(traceId, 'RETRY', `重试第${retry + 1}次`)
+                    } else {
+                        // 死信
+                        await channel.sendToQueue(
+                            this.deadQueue,
+                            Buffer.from(JSON.stringify(content)),
+                            { persistent: true }
+                        )
+
+                        this.log(traceId, 'DEAD', '进入死信队列')
+                    }
+
+                    await this.sendResult(channel, {
+                        id,
+                        success: false,
+                        error: err.message
+                    })
+
+                    channel.ack(msg)
+                }
+            })
+
+            this.logger.info('RunForge Worker 启动成功')
+        } catch (err) {
+            this.logger.error('RunForge Worker 启动失败: ' + err.stack)
+        }
+    }
+
+    async sendResult(channel, data) {
+        channel.sendToQueue(
+            this.resultQueue,
+            Buffer.from(JSON.stringify(data)),
+            { persistent: true }
+        )
+    }
+
+    async stop() {
+        this.running = false
+        await mq.close()
+        this.logger.info('RunForge Worker 已停止')
+    }
+}
+
+module.exports = Worker

lib/Server.js

@@ -5,7 +5,9 @@ const fs = require('fs')
 const config = require('../config.json')
 const Logger = require('./Logger')
 const MySQL = require('../plugin/DataBase/MySQL')
+const Worker = require('./Lepao/Worker')
 const mq = require('../plugin/mq')
+const { startLepaoSchedulePublisher } = require('../plugin/mq/lepaoSchedulePublisher')
 
 class SERVER {
     constructor() {
@@ -52,6 +54,20 @@ class SERVER {
 
             await ch.assertQueue('mq_health_check', { durable: false })
             this.logger.info('✅ RabbitMQ 初始化 & 测试成功')
+
+            const worker = new Worker()
+            try {
+                await worker.start()
+                this.logger.info('RunForge Worker 已启动，正在监听 MQ 任务...')
+                startLepaoSchedulePublisher({
+                    logger: this.logger,
+                    intervalMs: config.rabbitmq?.lepaoScheduleTickMs ?? 2000,
+                    batch: config.rabbitmq?.lepaoScheduleBatch ?? 100
+                })
+            } catch (err) {
+                console.error('RunForge Worker 启动失败:', err)
+                process.exit(1)
+            }
         } catch (e) {
             this.logger.error('❌ RabbitMQ 初始化失败')
             process.exit(1)

plugin/Email/Email.js

@@ -36,7 +36,7 @@ async function sendEmail(email, subject, content) {
 
             try {
                 await transporter.sendMail(mail)
-                logger.info(`邮件发送成功，使用账号: ${currentConfig.user}`)
+                logger.info(`${email}邮件发送成功，使用账号: ${currentConfig.user}`)
                 return resolve()
             } catch (error) {
                 logger.error(`邮件发送失败 (${currentConfig.user})，错误:`, error.stack)

plugin/Email/emailTemplate.js

@@ -377,6 +377,29 @@ class emailTemplate {
     }
 
     async lepaoSuccess(email, data) {
+        const target_count = Number(data.target_count) || 0
+        const total_num = Number(data.total_num) || 0
+        const timeSec = Number(data.time) || 0
+        const distanceKm = Number(data.distance) || 0
+        const passTit = data.pass_tit != null && data.pass_tit !== '' ? data.pass_tit : '—'
+        const paceStr =
+            distanceKm > 0 && timeSec > 0
+                ? this.calculatePace(timeSec, distanceKm)
+                : '—'
+        const timeStr = timeSec > 0 ? this.formatSecondsToMinSec(timeSec) : '—'
+
+        let goalHtml = ''
+        if (target_count === 0) {
+            goalHtml = `
+                <p><strong>累计次数：</strong> ${total_num} 次 ✨</p>`
+        } else {
+            const remain = Math.max(0, target_count - total_num)
+            const hitGoal = total_num >= target_count
+            goalHtml = `
+                <p><strong>目标次数：</strong> ${target_count} 次 🎯</p>
+                <p><strong>累计次数：</strong> ${total_num} 次 ✨</p>`
+        }
+
         await sendEmail(email, '乐跑成功提醒',
             `<html lang="zh-CN">
             <head>
@@ -424,6 +447,7 @@ class emailTemplate {
 
                     .info p {
                         margin: 5px 0;
+                        text-indent: 0;
                     }
 
                     .important {
@@ -451,16 +475,11 @@ class emailTemplate {
 
                 <div class="info">
                 <p><strong>学号：</strong> ${data.account}</p>
-                <p><strong>跑区：</strong> ${data.pass_tit} 🌈</p>
-                <p><strong>跑步时间：</strong> ${this.formatSecondsToMinSec(data.time)} ⏱️</p>
-                <p><strong>平均配速：</strong> ${this.calculatePace(data.time, data.distance)} 🐇</p>
-                <p><strong>跑步距离：</strong> ${data.distance} Km 💕</p>
-                <p><strong>累计次数：</strong> ${data.total_num} 次 ✨</p>
-                <p><strong>剩余次数：</strong>  ${data.term_num === 0 
-                                                ? '∞' 
-                                                : (data.term_num - data.total_num >= 0 
-                                                    ? (data.term_num - data.total_num) 
-                                                    : '已完成')} 次 🎯</p>
+                <p><strong>跑区：</strong> ${passTit} 🌈</p>
+                <p><strong>跑步时间：</strong> ${timeStr} ⏱️</p>
+                <p><strong>平均配速：</strong> ${paceStr} 🐇</p>
+                <p><strong>跑步距离：</strong> ${distanceKm || '—'} Km 💕</p>
+                ${goalHtml}
                 </div>
 
                 <p class="important">如果宝宝开启了自动乐跑，要记得不要在其他设备上登录“智慧体育”小程序哦 🚫📱，不然登录就会失效，要重新来一次啦~</p>

plugin/Lepao/Crypto.js

@@ -0,0 +1,89 @@
+const crypto = require('crypto')
+
+const KEY_STR = "Wet2C8d34f62ndi3"
+const IV_STR = "K6iv85jBD8jgf32D"
+const SALT = "rDJiNB9j7vD2"
+
+const KEY = Buffer.from(KEY_STR, 'utf-8')
+const IV = Buffer.from(IV_STR, 'utf-8')
+
+/**
+ * AES 加密
+ */
+function dataEncrypt(plainText) {
+    try {
+        const cipher = crypto.createCipheriv('aes-128-cbc', KEY, IV)
+        cipher.setAutoPadding(true)
+
+        let encrypted = cipher.update(plainText, 'utf-8')
+        encrypted = Buffer.concat([encrypted, cipher.final()])
+
+        return encrypted.toString('base64')
+    } catch (err) {
+        console.error('加密失败:', err.message)
+        return null
+    }
+}
+
+/**
+ * AES 解密
+ */
+function dataDecrypt(encryptedBase64Text) {
+    try {
+        const encryptedBuffer = Buffer.from(encryptedBase64Text, 'base64')
+
+        const decipher = crypto.createDecipheriv('aes-128-cbc', KEY, IV)
+        decipher.setAutoPadding(true)
+
+        let decrypted = decipher.update(encryptedBuffer)
+        decrypted = Buffer.concat([decrypted, decipher.final()])
+
+        return decrypted.toString('utf-8')
+    } catch (err) {
+        console.error('解密失败:', err.message)
+        return null
+    }
+}
+
+/**
+ * MD5 签名
+ */
+function dataSign(dataObj) {
+    if (typeof dataObj !== 'object') {
+        throw new TypeError('data must be object')
+    }
+
+    const sortedKeys = Object.keys(dataObj).sort()
+
+    let str = ''
+    for (const key of sortedKeys) {
+        str += key + String(dataObj[key])
+    }
+
+    str += SALT
+
+    return crypto.createHash('md5').update(str, 'utf-8').digest('hex')
+}
+
+/**
+ * OSS POST 签名
+ */
+function ossPostSign(accessKeySecret, policyDocument) {
+    const policyStr = JSON.stringify(policyDocument)
+
+    const base64Policy = Buffer.from(policyStr).toString('base64')
+
+    const signature = crypto
+        .createHmac('sha1', accessKeySecret)
+        .update(base64Policy)
+        .digest('base64')
+
+    return [base64Policy, signature]
+}
+
+module.exports = {
+    dataEncrypt,
+    dataDecrypt,
+    dataSign,
+    ossPostSign
+}

plugin/Lepao/Path.js

@@ -0,0 +1,165 @@
+// 随机扰动函数
+function randomPerturbation(scale = 1e-7) {
+    return (Math.random() * 2 - 1) * scale
+}
+
+// Haversine 公式计算两点距离（米）
+function haversine(lat1, lon1, lat2, lon2) {
+    const R = 6371000
+    const toRad = (deg) => deg * Math.PI / 180
+    const phi1 = toRad(lat1)
+    const phi2 = toRad(lat2)
+    const dphi = toRad(lat2 - lat1)
+    const dlambda = toRad(lon2 - lon1)
+
+    const a = Math.sin(dphi / 2) ** 2 + Math.cos(phi1) * Math.cos(phi2) * Math.sin(dlambda / 2) ** 2
+    return 2 * R * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a))
+}
+
+/**
+ * 根据原路径时间，重新生成新的时间路径数据
+ */
+function getPathData(pathlist, runEndTime, useTime) {
+    const startTime = (runEndTime - useTime) * 1000
+    const oldStartTime = parseInt(pathlist[0].d.split(' ')[0])
+
+    const newPathlist = pathlist.map((item, i) => {
+        const newItem = { ...item }
+        const [oldTimeStr, suffix] = item.d.split(' ', 2)
+        const oldTime = parseInt(oldTimeStr)
+
+        if (i === 0) {
+            newItem.d = `${startTime} ${suffix}`
+        } else {
+            const newTime = startTime + oldTime - oldStartTime
+            newItem.d = `${newTime} ${suffix}`
+            newItem.a += randomPerturbation()
+            newItem.o += randomPerturbation()
+        }
+        return newItem
+    })
+
+    return newPathlist
+}
+
+/**
+ * 选择打卡点
+ */
+function selectCheckpoints(path, checkpoints, runLogNum, pointUpdateDistance, logMaxDistance, runEndTime, pathTime) {
+    const results = []
+    let totalDistance = 0
+    let lastCheckpointDistance = 0
+
+    // 筛选有效打卡点
+    const filteredCheckpoints = checkpoints
+        .filter(cp => cp.is_del === "0" && cp.is_online === 1 && cp.ctrl_status === "1")
+    // 打乱顺序
+    for (let i = filteredCheckpoints.length - 1; i > 0; i--) {
+        const j = Math.floor(Math.random() * (i + 1))
+        // 避免解构交换在部分运行环境下触发 TDZ 相关异常
+        const temp = filteredCheckpoints[i]
+        filteredCheckpoints[i] = filteredCheckpoints[j]
+        filteredCheckpoints[j] = temp
+    }
+
+    const usedCheckpoints = new Set()
+
+    for (const p of path) {
+        const parts = p.d.split(' ')
+        const tsMs = parseInt(parts[0])
+        const stepDistance = parseFloat(parts[1].split('_')[0]) || 0
+
+        totalDistance += stepDistance
+
+        for (const cp of filteredCheckpoints.slice(0, -1)) {
+            if (usedCheckpoints.has(cp.id)) continue
+
+            const [lat, lon] = cp.jingwei.split(',').map(Number)
+            const dist = haversine(p.a, p.o, lat, lon)
+
+            if (dist < logMaxDistance && (totalDistance - lastCheckpointDistance) > 200) {
+                console.log(`选中打卡点 ${cp.id} ${cp.address}，距离：${dist}`)
+                results.push({
+                    point_id: cp.id,
+                    distance: +(totalDistance / 1000).toFixed(2),
+                    longitude: p.o,
+                    longtitude: p.o, // 保留原字段
+                    latitude: p.a,
+                    address: cp.address,
+                    jingwei: cp.jingwei.split(','),
+                    time: Math.floor(tsMs / 1000)
+                })
+                usedCheckpoints.add(cp.id)
+                lastCheckpointDistance = totalDistance
+                break
+            }
+        }
+    }
+
+    if (results.length >= runLogNum) {
+        // 随机抽取 runLogNum 个打卡点
+        const indices = []
+        while (indices.length < runLogNum) {
+            const idx = Math.floor(Math.random() * results.length)
+            if (!indices.includes(idx)) indices.push(idx)
+        }
+        indices.sort((a, b) => a - b)
+        const selectedResults = indices.map(i => results[i])
+
+        const n = selectedResults.length
+        let d = 0
+        for (let i = 0; i < n; i++) {
+            if (selectedResults[i].time >= runEndTime - 5) {
+                console.log("打卡点时间异常，重新分配时间")
+                for (let j = 0; j < n; j++) {
+                    selectedResults[j].time = Math.floor(runEndTime - (pathTime / n) * (n - j))
+                    selectedResults[j].distance = d + 0.3 * (j + 1)
+                }
+                break
+            }
+        }
+        console.log(`选中打卡点：${JSON.stringify(selectedResults)}`)
+        return selectedResults
+    } else {
+        console.log(`选中打卡点数量不足：${JSON.stringify(results)}`)
+        return false
+    }
+}
+
+/**
+ * 根据距离和用时生成步频数据
+ */
+function generateCadence(distanceKm, usedTime) {
+    const paceSec = usedTime / distanceKm
+    const paceMin = paceSec / 60
+
+    let spmRange
+    if (paceMin >= 8) spmRange = [85, 110]
+    else if (paceMin >= 5) spmRange = [150, 170]
+    else spmRange = [170, 190]
+
+    const minutes = Math.ceil(usedTime / 60)
+    const cadenceList = []
+
+    for (let i = 0; i < minutes; i++) {
+        let spm = Math.floor(Math.random() * (spmRange[1] - spmRange[0] + 1) + spmRange[0])
+        if (i === minutes - 1) {
+            const lastDuration = usedTime - (minutes - 1) * 60
+            spm = Math.round(spm * (lastDuration / 60))
+        }
+        cadenceList.push(spm)
+    }
+
+    const totalSteps = cadenceList.reduce((a, b) => a + b, 0)
+
+    return {
+        cadence_list: cadenceList,
+        total_steps: totalSteps
+    }
+}
+
+module.exports = {
+    getPathData,
+    selectCheckpoints,
+    generateCadence
+}

plugin/Lepao/runforgeSetZoneProbe.js

@@ -0,0 +1,115 @@
+/**
+ * 直连 RunForge 切换跑区（与 Worker lepao.setZone 一致），用于 token 探活，不经过 runpy。
+ */
+const axios = require('axios')
+const { URLSearchParams } = require('url')
+const db = require('../DataBase/db')
+const { dataEncrypt, dataDecrypt, dataSign } = require('./Crypto')
+
+const BASE_URL = 'https://lepao.ctbu.edu.cn/v3/api.php'
+
+const DEFAULT_UA =
+    'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'
+
+const runZoneMap = {
+    兰花湖校区跑区: 2,
+    主校区北跑区: 3,
+    主校区南跑区: 5,
+    重庆工商大学茶园校区: 6
+}
+
+function lepaoTimestamp() {
+    return Number((Date.now() / 1000).toFixed(3))
+}
+
+/**
+ * @param {object} p
+ * @param {string} p.uid
+ * @param {string} p.token
+ * @param {string|number} p.school_id
+ * @param {string} p.student_num
+ * @param {number} [p.random_id=1] path_data.id，需存在且 run_zone_name 可映射
+ * @param {string} [p.userAgent]
+ * @returns {Promise<object>} RunForge 响应体（与 Worker request 解密后形态一致）
+ */
+async function probeSetZone(p) {
+    const { uid, token, school_id, student_num, random_id = 1, userAgent } = p
+
+    const record = await db.query('SELECT run_zone_name FROM path_data WHERE id = ?', [random_id])
+    if (!record || record.length === 0) {
+        throw new Error('跑区不存在')
+    }
+
+    const runZoneId = runZoneMap[record[0].run_zone_name]
+    if (!runZoneId) throw new Error('跑区不存在')
+
+    const raw = {
+        uid,
+        token,
+        school_id,
+        term_id: 0,
+        course_id: 0,
+        class_id: 0,
+        student_num,
+        card_id: student_num,
+        timestamp: lepaoTimestamp(),
+        version: 1,
+        nonce: String(Math.floor(Math.random() * 900000 + 100000)),
+        ostype: 5,
+        run_zone_id: String(runZoneId)
+    }
+    raw.sign = dataSign(raw)
+
+    const form = new URLSearchParams()
+    form.append('ostype', '5')
+    form.append('data', dataEncrypt(JSON.stringify(raw)))
+
+    const res = await axios.post(`${BASE_URL}/Run/setRunZone`, form, {
+        headers: {
+            'Content-Type': 'application/x-www-form-urlencoded',
+            Accept: '*/*',
+            'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
+            'User-Agent': userAgent || DEFAULT_UA,
+            Referer: 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
+            charset: 'utf-8'
+        },
+        timeout: 20000,
+        proxy: false
+    })
+
+    let result = res.data
+    if (result?.data && result?.is_encrypt === 1) {
+        result = { ...result, data: JSON.parse(dataDecrypt(result.data)) }
+    }
+    return result
+}
+
+function isProbeSetZoneOk(result) {
+    if (!result) return false
+    const hasData = result.data != null && result.data !== ''
+    if (!hasData) return false
+    if (Number(result.status) === 1) return true
+    const cd = Number(result.code)
+    if (cd === 1 || cd === 200) return true
+    return false
+}
+
+function getProbeFailMessage(result) {
+    if (!result) return ''
+    return (
+        result.info ||
+        result.msg ||
+        result.message ||
+        (typeof result.data === 'object' && result.data
+            ? result.data.info || result.data.msg || result.data.message
+            : '') ||
+        ''
+    )
+}
+
+module.exports = {
+    probeSetZone,
+    isProbeSetZoneOk,
+    getProbeFailMessage,
+    BASE_URL
+}

plugin/mq/Worker.js

@@ -0,0 +1,143 @@
+const db = require('../DataBase/db')
+const path = require('path')
+const Logger = require('../../lib/Logger')
+const mq = require('.')
+
+class Worker {
+    constructor() {
+        this.logger = new Logger(
+            path.join(__dirname, '../logs/Worker.log'),
+            'INFO'
+        )
+
+        this.handlers = {}
+        this.running = false
+
+        // 队列名
+        this.taskQueue = 'task_queue'
+        this.resultQueue = 'task_result_queue'
+
+        // channel 名称（避免和别的模块冲突）
+        this.channelName = 'worker_channel'
+    }
+
+    /**
+     * 注册任务处理器
+     */
+    register(type, handler) {
+        this.handlers[type] = handler
+        this.logger.info(`注册处理器: ${type}`)
+    }
+
+    /**
+     * 启动 Worker
+     */
+    async start() {
+        if (this.running) return
+        this.running = true
+
+        this.logger.info('Worker 启动中...')
+
+        try {
+            const channel = await mq.getChannel(this.channelName)
+
+            // 控制并发（重要）
+            await channel.prefetch(5)
+
+            // 确保队列存在
+            await channel.assertQueue(this.taskQueue, { durable: true })
+            await channel.assertQueue(this.resultQueue, { durable: true })
+
+            // 开始消费
+            await channel.consume(
+                this.taskQueue,
+                async (msg) => {
+                    if (!msg) return
+
+                    let content
+
+                    try {
+                        content = JSON.parse(msg.content.toString())
+                    } catch (err) {
+                        this.logger.error('消息解析失败: ' + err.message)
+                        channel.ack(msg)
+                        return
+                    }
+
+                    const { id, type, data } = content
+
+                    this.logger.info(`收到任务: ${id} 类型: ${type}`)
+
+                    const handler = this.handlers[type]
+
+                    if (!handler) {
+                        this.logger.error(`未找到处理器: ${type}`)
+                        channel.ack(msg)
+                        return
+                    }
+
+                    try {
+                        const result = await handler(data, {
+                            db,
+                            logger: this.logger
+                        })
+
+                        this.logger.info(`任务完成: ${id}`)
+
+                        await this.sendResult(channel, {
+                            id,
+                            success: true,
+                            result
+                        })
+
+                        channel.ack(msg)
+                    } catch (err) {
+                        this.logger.error(`任务失败: ${id} - ${err.stack}`)
+
+                        await this.sendResult(channel, {
+                            id,
+                            success: false,
+                            error: err.message
+                        })
+
+                        // 简单策略：失败直接 ack（避免死循环）
+                        channel.ack(msg)
+                    }
+                },
+                {
+                    noAck: false
+                }
+            )
+
+            this.logger.info('Worker 启动成功')
+        } catch (err) {
+            this.logger.error('Worker 启动失败: ' + err.stack)
+        }
+    }
+
+    /**
+     * 发送结果
+     */
+    async sendResult(channel, data) {
+        try {
+            channel.sendToQueue(
+                this.resultQueue,
+                Buffer.from(JSON.stringify(data)),
+                { persistent: true }
+            )
+        } catch (err) {
+            this.logger.error('结果发送失败: ' + err.message)
+        }
+    }
+
+    /**
+     * 停止 Worker
+     */
+    async stop() {
+        this.running = false
+        await mq.close()
+        this.logger.info('Worker 已停止')
+    }
+}
+
+module.exports = Worker

plugin/mq/enqueueLepaoStartRun.js

@@ -0,0 +1,26 @@
+const mq = require('./index')
+const { assertRunforgeTaskIngress, publishRunforgeTask } = require('./runforgeTaskMq')
+
+/**
+ * 将乐跑任务写入 MQ，由 Worker 执行（与 SingleRun 一致）
+ */
+async function enqueueLepaoStartRun(studentNum, logger) {
+    try {
+        const channel = await mq.getChannel('lepao_account_autorun')
+        await assertRunforgeTaskIngress(channel, logger)
+        const taskId = `lepao:account:${Date.now()}:${studentNum}`
+        publishRunforgeTask(channel, {
+            id: taskId,
+            type: 'lepao.startRun',
+            data: {
+                taskId,
+                account: studentNum
+            },
+            retry: 0
+        })
+    } catch (e) {
+        logger?.error?.(`投递自动乐跑任务失败 ${studentNum}: ${e.message || e}`)
+    }
+}
+
+module.exports = { enqueueLepaoStartRun }

plugin/mq/lepaoAutoScheduleRedis.js

@@ -0,0 +1,135 @@
+const Redis = require('../DataBase/Redis')
+
+/** 到期后由进程定时器投递到 runforge_task_queue（重启不丢，依赖 Redis） */
+const SCHEDULE_KEY = 'lepao:mq:scheduled'
+
+const POP_DUE_LUA = `
+local key = KEYS[1]
+local now = tonumber(ARGV[1])
+local limit = tonumber(ARGV[2])
+local items = redis.call('ZRANGEBYSCORE', key, '-inf', now, 'LIMIT', 0, limit)
+for i = 1, #items do
+  redis.call('ZREM', key, items[i])
+end
+return items
+`
+
+function stripScheduleMeta(msg) {
+    if (!msg || typeof msg !== 'object') return msg
+    const copy = { ...msg }
+    delete copy._scheduleMeta
+    return copy
+}
+
+/**
+ * 延迟投递：整消息入 ZSET，score = fireAt（毫秒时间戳）
+ * @param {object} meta 可选，管理端展示用：{ name, account, delayMs }
+ */
+async function scheduleDelayedRunforgeTask(fireAt, messageObject, meta = null) {
+    const toStore =
+        meta != null
+            ? {
+                  ...messageObject,
+                  _scheduleMeta: {
+                      ...meta,
+                      fireAt
+                  }
+              }
+            : messageObject
+    const member = JSON.stringify(toStore)
+    await Redis.sendCommand(['ZADD', SCHEDULE_KEY, String(fireAt), member])
+}
+
+async function requeueAt(fireAt, messageObject) {
+    const member = JSON.stringify(messageObject)
+    await Redis.sendCommand(['ZADD', SCHEDULE_KEY, String(fireAt), member])
+}
+
+/**
+ * 原子取出已到期的成员（原始 JSON 字符串数组）
+ */
+async function popDueMessages(now = Date.now(), limit = 100) {
+    const res = await Redis.sendCommand([
+        'EVAL',
+        POP_DUE_LUA,
+        '1',
+        SCHEDULE_KEY,
+        String(now),
+        String(limit)
+    ])
+    if (!Array.isArray(res)) return []
+    return res.map((x) => (Buffer.isBuffer(x) ? x.toString('utf8') : String(x)))
+}
+
+/** 清理过久未弹出项（异常场景） */
+async function pruneStaleScheduled(beforeScore, now = Date.now()) {
+    await Redis.sendCommand(['ZREMRANGEBYSCORE', SCHEDULE_KEY, '-inf', String(beforeScore)])
+}
+
+/**
+ * 管理端：尚未到期的调度（score > now）
+ */
+async function listPendingScheduledForAdmin(now = Date.now(), limitTotal = 800) {
+    await pruneStaleScheduled(now - 48 * 3600 * 1000, now)
+
+    const raw = await Redis.sendCommand([
+        'ZRANGEBYSCORE',
+        SCHEDULE_KEY,
+        `(${String(now)}`,
+        '+inf',
+        'WITHSCORES',
+        'LIMIT',
+        '0',
+        String(limitTotal)
+    ])
+
+    const items = []
+    for (let i = 0; i < raw.length; i += 2) {
+        const score = Number(raw[i + 1])
+        const value = Buffer.isBuffer(raw[i]) ? raw[i].toString('utf8') : raw[i]
+        let parsed
+        try {
+            parsed = JSON.parse(value)
+        } catch {
+            parsed = { raw: value }
+        }
+        const meta = parsed._scheduleMeta || {}
+        const fireAt = meta.fireAt != null ? meta.fireAt : score
+        items.push({
+            taskId: parsed.id,
+            type: parsed.type,
+            account: parsed.data?.account,
+            name: meta.name,
+            fireAt,
+            score,
+            delayMs: meta.delayMs,
+            remainMs: Math.max(0, fireAt - now),
+            payloadPreview: stripScheduleMeta(parsed)
+        })
+    }
+
+    return {
+        items,
+        note: 'Redis 调度：未到 fireAt 前不会进入 runforge_task_queue；到期由本机定时任务写入 MQ。'
+    }
+}
+
+async function countPendingScheduled(now = Date.now()) {
+    const n = await Redis.sendCommand([
+        'ZCOUNT',
+        SCHEDULE_KEY,
+        `(${String(now)}`,
+        '+inf'
+    ])
+    return Number(n) || 0
+}
+
+module.exports = {
+    SCHEDULE_KEY,
+    stripScheduleMeta,
+    scheduleDelayedRunforgeTask,
+    requeueAt,
+    popDueMessages,
+    listPendingScheduledForAdmin,
+    countPendingScheduled
+}

plugin/mq/lepaoSchedulePublisher.js

@@ -0,0 +1,63 @@
+const mq = require('./index')
+const { publishRunforgeTask } = require('./runforgeTaskMq')
+const {
+    popDueMessages,
+    stripScheduleMeta,
+    requeueAt,
+    SCHEDULE_KEY
+} = require('./lepaoAutoScheduleRedis')
+
+let intervalHandle = null
+
+/**
+ * 定时将 Redis 中已到期的乐跑任务写入 runforge_task_queue
+ */
+function startLepaoSchedulePublisher(options = {}) {
+    const logger = options.logger || console
+    const intervalMs = options.intervalMs ?? 2000
+    const batch = options.batch ?? 100
+
+    if (intervalHandle) return
+
+    intervalHandle = setInterval(async () => {
+        try {
+            const now = Date.now()
+            const rawList = await popDueMessages(now, batch)
+            if (!rawList.length) return
+
+            const channel = await mq.getChannel('lepao_schedule_tick')
+
+            for (const raw of rawList) {
+                let msg
+                try {
+                    msg = JSON.parse(raw)
+                } catch (e) {
+                    logger.error?.(
+                        `[LepaoSchedule] 调度 JSON 无效已丢弃 key=${SCHEDULE_KEY}: ${String(raw).slice(0, 120)}`
+                    )
+                    continue
+                }
+
+                try {
+                    publishRunforgeTask(channel, stripScheduleMeta(msg))
+                } catch (e) {
+                    logger.error?.(`[LepaoSchedule] MQ 投递失败，5s 后重试: ${e.message || e}`)
+                    const retryAt = now + 5000
+                    try {
+                        await requeueAt(retryAt, msg)
+                    } catch (re2) {
+                        logger.error?.(`[LepaoSchedule] 写回 Redis 失败: ${re2.message || re2}`)
+                    }
+                }
+            }
+        } catch (e) {
+            logger.error?.(`[LepaoSchedule] tick 异常: ${e.message || e}`)
+        }
+    }, intervalMs)
+
+    logger.info?.(
+        `[LepaoSchedule] 已启动 Redis→MQ 调度（间隔 ${intervalMs}ms，每批最多 ${batch} 条，key=${SCHEDULE_KEY}）`
+    )
+}
+
+module.exports = { startLepaoSchedulePublisher }

plugin/mq/runforgeTaskMq.js

@@ -0,0 +1,26 @@
+const TASK_QUEUE = 'runforge_task_queue'
+
+/**
+ * 声明乐跑任务主队列（无 RabbitMQ 插件）
+ */
+async function assertRunforgeTaskIngress(channel, logger) {
+    await channel.assertQueue(TASK_QUEUE, { durable: true })
+    return { mode: 'direct', queue: TASK_QUEUE }
+}
+
+/**
+ * 投递乐跑任务 JSON 消息体（与 Worker 消费格式一致）
+ */
+function publishRunforgeTask(channel, messageObject) {
+    const body = Buffer.from(JSON.stringify(messageObject))
+    channel.sendToQueue(TASK_QUEUE, body, {
+        persistent: true,
+        contentType: 'application/json'
+    })
+}
+
+module.exports = {
+    TASK_QUEUE,
+    assertRunforgeTaskIngress,
+    publishRunforgeTask
+}