4l_capital_central_control_client/src/robot.js

const file = require("./utils/file");
const {getRustConfig} = require("./config");
const {spawn, exec, fork, execSync} = require("child_process");
const {logger} = require("./utils/logger");
const http = require("./utils/http");
const e = require("express");
/*******
 * web
 * **** */


const RobotStatus = Object.freeze({
    // 已停止的
    STOPPED: "STOPPED",
    // 正在停止
    STOP_PENDING: "STOP_PENDING",
    //正在运行
    RUNNING: "RUNNING",
    //启动中
    START_PENDING: "START_PENDING",
    //下载中
    DOWNLOADING: "DOWNLOADING",
    //错误
    ERROR: "ERROR"
});

let appMap = new Map();


function getApp(key) {
    let app = {
        id: -1,
        childProcess: undefined,
        nowBalance: 0,
        messlist: [],
        threadStatus: RobotStatus.STOPPED,
        errorMessage: "成功",
        threadStartTime: -1,
    }
    // logger.info(appMap, appMap.has(key));
    if (appMap.has(key)) {
        app = appMap.get(key)
    } else {
        appMap.set(key, app)
    }
    // logger.info(app);
    return app
}

function delApp(key) {
    appMap.delete(key)
}

async function run(param) {
    var key = param.id
    var appName = param.path
    var app = getApp(key)
    var config = getRustConfig()


    /****
     *** 第一步：防止 反复启动，先要kill一次
     ***/
    var waitForCondition = function (intervalDuration, conditionFunction) {
        return new Promise((resolve, reject) => {
            const intervalId = setInterval(() => {
                if (conditionFunction()) {
                    clearInterval(intervalId);
                    resolve();
                }
            }, intervalDuration);
        });
    }

    logger.info('当前app', app);
    if (app.childProcess !== undefined && app.threadStatus !== RobotStatus.STOPPED
        && app.threadStatus !== RobotStatus.ERROR && app.threadStatus !== RobotStatus.STOP_PENDING) {
        logger.info('防止重复启动，需要先杀死');
        app.threadStatus = RobotStatus.STOP_PENDING
        robotStatus(app)
        app.childProcess.kill()

        // 当这个条件变为true时，会结束等待
        await waitForCondition(1000, () => {
            if (app.threadStatus === RobotStatus.STOPPED) {
                logger.info('成功杀死可以重新开启');
                return true
            } else {
                logger.info('等待杀死');
                return false
            }
        });
    }
    /********************新杀进程*************************************/
    // await waitForCondition(1000, () => {
    //     if (app.threadStatus === RobotStatus.STOPPED) {
    //         logger.info('成功杀死可以重新开启');
    //         return true
    //     } else {
    //         logger.info('等待杀死');
    //         return false
    //     }
    // });
    // return;
    /********************新杀进程*************************************/

    // 初始化机器人状态
    app.threadStatus = RobotStatus.START_PENDING
    app.id = key
    robotStatus(app)

    /****
     *** 第二步：路径经组装
     *   注意：可能存在一台服务器多个机器人，通过机器人ID创建文件夹区分，需要组装好路径
     ***/
        //系统不同 做不同的路径处理
    const platform = process.platform;
    let exeName = appName;           //可执行程序
    // let configName = "config.toml";        //配置文件
    let configName = "config.json";        //配置文件
    let appPath = "";
    if (platform === 'win32') {//直接运行相对路径
        appPath = config.winPath + "/" + app.id
    } else {//打包为docker 镜像
        appPath = config.linuxPath + "/" + app.id
    }

    /****
     *** 第三步：rust 启动程序检查（下载更新）
     ***/

    var isDow = false
    var scheduleDow = 0
    var scheduleConfig = 0
    //1. 检查目录
    file.checkPathSync(appPath);
    //2、 检查执行程序
    var destination = appPath + "/" + exeName
    if (!file.checkFilePath(destination)) {
        app.threadStatus = RobotStatus.DOWNLOADING
        robotStatus(app)
        isDow = true
        var dowHeaders = {...config.headers};
        if (platform === 'win32') {
            downloadFileWithPowerShell(
                config.baseUrl + config.dowAppURL + '/?path=' + appName,
                appPath + "/" + appName,
                dowHeaders, (err, b) => {
                    if (err === null) {
                        scheduleDow = 1
                    } else {
                        app.threadStatus = RobotStatus.ERROR
                        app.errorMessage = '下载失败！'
                        scheduleDow = -1
                        robotStatus(app)
                    }
                });
        } else {
            downloadFileWithLinux(
                config.baseUrl + config.dowAppURL + '/?path=' + appName,
                appPath + "/" + appName,
                dowHeaders, (err, b) => {
                    if (err === null) {
                        scheduleDow = 1
                    } else {
                        app.threadStatus = RobotStatus.ERROR
                        app.errorMessage = '下载失败！'
                        scheduleDow = -1
                        robotStatus(app)
                    }
                });
        }
    } else {
        scheduleDow = 1
    }

    /****
     *** 第四步：rust 启动配置检查（下载更新）
     ***/
        //2 为防止启动指令不同，每次重新写入
    var destination2 = appPath + "/" + configName
    var urrrl = config.baseUrl + config.dowConfigURL + '/?robotId=' + app.id
    var configHeaders = {...config.headers};
    http.request_get(urrrl, {...config.headers})
        .then((data) => {
            logger.info('配置参数：', data);
            const json = JSON.parse(data)
            //处理成配置文件
            const map = json.data
            var json_obj = JSON.parse("{}")
            for (k in map) {
                //未完成 参数解析有问题
                if (k === "account") {
                    var account = map["account"];
                    delete map.account
                    json_obj["account_name"] = account["name"]
                    json_obj["access_key"] = account["accessKey"]
                    json_obj["secret_key"] = account["secretKey"]
                    json_obj["pass_key"] = account["pass"]
                } else if (k === "ref_pair") {
                    json_obj["ref_pair"] = [map["ref_pair"]]
                } else {
                    json_obj[k] = map[k]
                }
                logger.info(map[k] + "\t")
            }
            logger.info("参数组装完成！")
            te = JSON.stringify(json_obj)
            file.writeFile(destination2, te, (errer, b) => {
                if (errer === null && b === true) {
                    scheduleConfig = 1
                } else {
                    logger.info("配置参数写入配置失败！", errer)
                    app.threadStatus = RobotStatus.ERROR
                    app.errorMessage = '配置参数写入配置失败！'
                    scheduleConfig = -1
                    robotStatus(app)
                }
            })
        })
        .catch((e) => {
            logger.info("配置参数获取失败", e)
            app.threadStatus = RobotStatus.ERROR
            app.errorMessage = '配置参数获取失败！'
            scheduleConfig = -1
            robotStatus(app)
        })

    //监听下载只有下载完成了才能继续
    while (true) {
        await delay(5000);
        let info_t = ""
        let info_t2 = ""
        //是否开启下载,如果是新下载，下载完成需要授权
        if (isDow) {
            if (scheduleDow === 1) {
                info_t = "启动文件：下载完成!"
                //文件授权
                execSync(`chmod +x ${appPath + "/" + appName}`, (error, stdout, stderr) => {
                    if (error) {
                        logger.error(`启动文件：授权失败: ${error}`);
                    }
                    logger.info(`启动文件：授权完成！`);
                });
            } else if (scheduleDow === -1) {
                info_t = "启动文件：下载失败!"
            } else {
                info_t = "启动文件：还在下载..."
            }
        } else {
            info_t = "启动文件：完整无需下载"
        }

        if (scheduleConfig === 1) {
            info_t2 = "配置文件:读取成功!"
        } else if (scheduleConfig === -1) {
            info_t = "配置文件：读取失败!"
        } else {
            info_t2 = "配置文件:还在读取..."
        }
        logger.info(info_t, info_t2);
        if (scheduleDow === 1 && scheduleConfig === 1) {
            break
        } else if (scheduleConfig === -1 || scheduleDow === -1) {
            return;
        }
    }

    app.threadStatus = RobotStatus.START_PENDING
    robotStatus(app)
    logger.info("开始启动程序！");
    //3. spawn启动
    const exePath = appPath + "/" + exeName
    const configPath = appPath + "/" + configName
    logger.info(`文件地址：${exePath}-----${configPath}`);
    const command = exePath
    const args = ['--config=' + configPath]
    app.childProcess = spawn(command, args)


    app.threadStartTime = new Date().getTime()
    /**********监听*********/
    app.childProcess.stdout.on('data', (msg) => {
        // logger.info('stdout:' + msg.toString())
    })

    app.childProcess.on('message', (msg) => {
        // logger.info(`message: ${msg}`);
    });
    app.childProcess.on('exit', (code, signal) => {
        logger.info(`子进程退出-exit，退出码: ${code}, 信号: ${signal}`);
    });

    app.childProcess.on('close', (code) => {
        logger.info(`子进程退出-close，退出码 ${code}`);
        app.threadStatus = RobotStatus.STOPPED
        robotStatus(app)
    });
    app.threadStatus = RobotStatus.RUNNING
}

function delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
}

async function closeApp(param) {
    var key = param.id
    var app = getApp(key)
    logger.info(` 信号:  `, app.threadStatus);
    if (app.threadStatus === RobotStatus.RUNNING) {
        app.childProcess.kill('SIGTERM');
        app.threadStatus = RobotStatus.STOP_PENDING
        robotStatus(app)
    }
}


// 下载执行程序，使用的实行服务器指令的方式进行下载
function downloadFileWithPowerShell(url, destination, headers, funBreak) {
    const headersString = Object.entries(headers)
        .map(([key, value]) => `'${key}'='${value}'`) // 使用单引号来确保特殊字符不被解析
        .join('; '); // 使用分号和空格分隔每个键值对
    const command = `powershell -command "Invoke-WebRequest -Uri '${url}' -OutFile '${destination}' -Headers @{${headersString}}"`;
    exec(command, (error, stdout, stderr) => {
        if (error) {
            logger.error(`下载出错: ${error}`);
            return funBreak(error);
        }
        logger.info(`下载完成！`);
        funBreak(null, true); // 通知下载成功，修改了B为true
    });
}

function downloadFileWithLinux(url, destination, headers, funBreak) {
    const headersString = Object.entries(headers)
        .map(([key, value]) => `-H '${key}: ${value}'`) // 设置curl的HTTP头参数
        .join(' '); // 使用空格分隔每个头参数
    const command = `curl ${headersString} '${url}' -o '${destination}'`;
    exec(command, (error, stdout, stderr) => {
        if (error) {
            logger.error(`下载出错: ${error}`);
            return funBreak(error);
        }
        logger.info(`下载完成！`);
        funBreak(null, true); // 通知下载成功，修改了B为true
    });
}


// 上报-状态
function robotStatus(app) {
    var config = getRustConfig()

    var msg = (app.threadStatus !== RobotStatus.ERROR ? "完成" : app.errorMessage)
    http.request_post(`${config.baseUrl}/report/statusReport`, {
        "robotId": app.id,
        "status": app.threadStatus,
        "msg": msg
    }, {...config.headers}).then((data) => {
        // logger.info('??', data);
        logger.info('汇报--状态：', '机器人id：', app.id, '状态：', app.threadStatus);
    }).catch((error) => {
        logger.error(`请求遇到问题1: ${error.message}`); // 处理可能发生的错误
    });
}

// 上报-余额
function robotAmount(app) {
    //拿到策略余额
    try {
        var config = getRustConfig()
        http.request_get(`${config.rustUrl}/account`, {...config.headers})
            .then((data) => {
                var d = JSON.parse(data)
                //余额有变动上报一次
                if ((app.nowBalance + "") !== (d.now_balance + "")) {
                    http.request_post(`${config.baseUrl}/report/amountReport`, {
                        "robotId": app.id,
                        "amount": d.now_balance
                    }, {...config.headers})
                        .then((data2) => {
                            // logger.info('上报响应', data2);
                            logger.info('汇报--余额：pid:', app.childProcess.pid, '机器人id：', app.id, '机器人本地余额：', app.nowBalance, '实际余额：', d.now_balance);
                        }).catch((error) => {
                        logger.error(`请求遇到问题:2 ${error.message}`); // 处理可能发生的错误
                    });
                    app.nowBalance = d.now_balance
                }
            }).catch((error) => {
            logger.error(`请求遇到问题2: ${error.message}`); // 处理可能发生的错误
        });
    } catch (e) {
        logger.error('请求失败！:', e)
    }
}

// 定时清理停止状态的 机器人
function delRobot(app) {
    const platform = process.platform;
    if (platform !== 'win32' && app.threadStatus === RobotStatus.STOPPED) {
        var pid = app.childProcess.pid
        delApp(app.id)
    }
}


module.exports = {
    run, closeApp, delRobot, robotStatus, robotAmount, appMap, RobotStatus
};