zc/autoAiWorkSys
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
张成
2025-11-26 15:00:14 +08:00
parent 7858459118
commit e27c0dc41a
9 changed files with 1046 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files

460
api/middleware/schedule/command.js
View File

@@ -11,52 +11,48 @@ const ScheduleConfig = require('./config');
*/
class CommandManager {
constructor() {
this.pendingCommands = new Map(); // 等待响应的指令 { commandId: { resolve, reject, timeout } }
this.pendingCommands = new Map(); // 等待响应的指令 { command_id: { resolve, reject, timeout } }
}
/**
* 执行指令序列
* @param {number} task_id - 任务ID
* @param {Array} commands - 指令数组
* @param {object} mqttClient - MQTT客户端
* @param {object} options - 执行选项
* @param {object} options - 执行选项（保留用于扩展）
* @returns {Promise<object>} 执行结果
*/
async executeCommands(taskId, commands, mqttClient, options = {}) {
// try {
async executeCommands(task_id, commands, mqttClient, options = {}) {
if (!commands || commands.length === 0) {
throw new Error('没有找到要执行的指令');
}
const {
maxRetries = 1, // 最大重试次数
retryDelay = 1000 // 重试延迟(毫秒)
} = options;
console.log(`[指令管理] 开始执行 ${commands.length} 个指令`);
const results = [];
const errors = [];
// 顺序执行指令，失败时停止
// 顺序执行指令，失败时继续执行后续指令
for (let i = 0; i < commands.length; i++) {
const command = commands[i];
let retryCount = 0;
let commandResult = null;
try {
console.log(`[指令管理] 执行指令 ${i + 1}/${commands.length}: ${command.command_name || command.name}`);
// 重试逻辑
while (retryCount <= maxRetries) {
console.log(`[指令管理] 执行指令 ${i + 1}/${commands.length}: ${command.command_name || command.name} (尝试 ${retryCount + 1}/${maxRetries + 1})`);
commandResult = await this.executeCommand(taskId, command, mqttClient);
results. push(commandResult);
break; // 成功执行，跳出重试循环
const commandResult = await this.executeCommand(task_id, command, mqttClient);
results.push(commandResult);
} catch (error) {
// 指令执行失败，记录错误但继续执行后续指令
errors.push({
command: command,
error: error
});
console.error(`[指令管理] 指令执行失败: ${command.command_name || command.name}, 错误: ${error.message}`);
// 失败时继续执行后续指令
}
}
const successCount = results.length;
@@ -72,203 +68,260 @@ class CommandManager {
successCount: successCount,
errorCount: errorCount
};
// } catch (error) {
// console.error(`[指令管理] 执行指令序列失败:`, error);
// throw error;
// }
}
/**
* 执行单个指令
* 执行单个指令（统一封装）
* 统一处理成功、失败、超时，统一记录数据库
* @param {number} task_id - 任务ID
* @param {object} command - 指令对象
* @param {object} mqttClient - MQTT客户端
* @returns {Promise<object>} 执行结果
*/
async executeCommand(taskId, command, mqttClient) {
const startTime = new Date();
let commandRecord = null;
async executeCommand(task_id, command, mqttClient) {
const start_time = new Date();
let command_id = null;
let command_record = null;
const task = await db.getModel('task_status').findByPk(taskId);
try {
// 1. 获取任务信息
const task = await db.getModel('task_status').findByPk(task_id);
if (!task) {
throw new Error(`任务不存在: ${task_id}`);
}
// 获取指令信息（支持两种格式）
const commandName = command.command_name;
const commandType = command.command_type;
const commandParams = command.command_params ? JSON.parse(command.command_params) : {};
// 2. 获取指令信息
const command_name = command.command_name || command.name || '未知指令';
const command_type = command.command_type || command.type;
const command_params = command.command_params ?
(typeof command.command_params === 'string' ? JSON.parse(command.command_params) : command.command_params) :
{};
// 创建指令记录
commandRecord = await db.getModel('task_commands').create({
task_id: taskId,
command_type: commandType,
command_name: commandName,
command_params: JSON.stringify(commandParams),
priority: command.priority || 1,
sequence: command.sequence || 1,
max_retries: command.maxRetries || command.max_retries || 3,
status: 'pending'
});
if (!command_type) {
throw new Error('指令类型不能为空');
}
let commandId = commandRecord.id;
console.log(`[指令管理] 创建指令记录: ${commandName} (ID: ${commandId})`);
// 3. 创建指令记录
command_record = await db.getModel('task_commands').create({
task_id: task_id,
command_type: command_type,
command_name: command_name,
command_params: JSON.stringify(command_params),
priority: command.priority || 1,
sequence: command.sequence || 1,
max_retries: command.maxRetries || command.max_retries || 3,
status: 'pending'
});
command_id = command_record.id;
console.log(`[指令管理] 创建指令记录: ${command_name} (ID: ${command_id})`);
// 更新指令状态为运行中
await this.updateCommandStatus(commandId, 'running');
// 4. 更新指令状态为运行中
await this._update_command_status(command_id, 'running', null, null, start_time);
// 5. 执行指令（统一封装）
const result = await this._execute_command_with_timeout(
command_id,
command_type,
command_name,
command_params,
task.sn_code,
mqttClient,
start_time
);
console.log(`[指令管理] 执行指令: ${commandName} (ID: ${commandId})`);
// 6. 记录成功结果
await this._record_command_result(command_id, 'completed', result, null, start_time);
const sn_code = task.sn_code;
return {
command_id: command_id,
command_name: command_name,
result: result,
duration: new Date() - start_time,
success: true
};
// 将驼峰命名转换为下划线命名（如：getOnlineResume -> get_online_resume）
const toSnakeCase = (str) => {
// 如果已经是下划线格式，直接返回
} catch (error) {
// 统一处理错误（失败或超时）
if (command_id) {
await this._record_command_result(
command_id,
'failed',
null,
error,
start_time
);
}
// 重新抛出错误，让调用方知道执行失败
throw error;
}
}
/**
* 执行指令（带超时保护）
* @private
*/
async _execute_command_with_timeout(command_id, command_type, command_name, command_params, sn_code, mqttClient, start_time) {
// 将驼峰命名转换为下划线命名
const to_snake_case = (str) => {
if (str.includes('_')) {
return str;
}
// 驼峰转下划线
return str.replace(/([A-Z])/g, '_$1').toLowerCase().replace(/^_/, '');
};
const methodName = toSnakeCase(commandType);
const method_name = to_snake_case(command_type);
// 获取指令超时时间（从配置中获取，默认5分钟）
const timeout = ScheduleConfig.taskTimeouts[commandType] || ScheduleConfig.taskTimeouts[methodName] || 5 * 60 * 1000;
const timeout = ScheduleConfig.taskTimeouts[command_type] ||
ScheduleConfig.taskTimeouts[method_name] ||
5 * 60 * 1000;
let result;
try {
// 使用超时机制包装指令执行
const commandPromise = (async () => {
if (commandType && jobManager[methodName]) {
return await jobManager[methodName](sn_code, mqttClient, commandParams);
} else {
// 如果转换后找不到，尝试直接使用原名称
if (jobManager[commandType]) {
return await jobManager[commandType](sn_code, mqttClient, commandParams);
} else {
throw new Error(`未知的指令类型: ${commandType} (尝试的方法名: ${methodName})`);
}
}
})();
// 构建指令执行 Promise
const command_promise = (async () => {
if (command_type && jobManager[method_name]) {
return await jobManager[method_name](sn_code, mqttClient, command_params);
} else if (jobManager[command_type]) {
return await jobManager[command_type](sn_code, mqttClient, command_params);
} else {
throw new Error(`未知的指令类型: ${command_type} (尝试的方法名: ${method_name})`);
}
})();
// 使用超时机制
result = await ScheduleUtils.withTimeout(
commandPromise,
// 使用超时机制包装
try {
const result = await ScheduleUtils.withTimeout(
command_promise,
timeout,
`指令执行超时: ${commandName} (超时时间: ${timeout / 1000}秒)`
`指令执行超时: ${command_name} (超时时间: ${timeout / 1000}秒)`
);
return result;
} catch (error) {
const endTime = new Date();
const duration = endTime - startTime;
// 如果是超时错误，更新指令状态为失败
const errorMessage = error.message || '指令执行失败';
await this.updateCommandStatus(commandId, 'failed', null, errorMessage);
// 判断是否为超时错误
if (error.message && error.message.includes('超时')) {
error.isTimeout = true;
}
throw error;
}
const endTime = new Date();
const duration = endTime - startTime;
// 更新指令状态为完成
await this.updateCommandStatus(commandId, 'completed', result);
return {
commandId: commandId,
commandName: commandName,
result: result,
duration: duration,
success: true
};
}
/**
* 记录指令执行结果（统一封装）
* 处理成功、失败、超时等情况，统一记录到数据库
* @private
*/
async _record_command_result(command_id, status, result, error, start_time) {
const end_time = new Date();
const duration = end_time.getTime() - start_time.getTime();
let error_message = null;
let error_stack = null;
let result_data = null;
// 处理错误信息
if (error) {
error_message = error.message || '指令执行失败';
error_stack = error.stack || '';
// 如果是超时错误，添加标识
if (error.isTimeout) {
error_message = `[超时] ${error_message}`;
}
}
// 处理结果数据
if (result && status === 'completed') {
result_data = result;
}
// 更新数据库
await this._update_command_status(
command_id,
status,
result_data,
error_message,
start_time,
end_time,
duration,
error_stack
);
// 记录日志
if (status === 'completed') {
console.log(`[指令管理] 指令执行成功: ${command_id} (耗时: ${duration}ms)`);
} else if (status === 'failed') {
const error_type = error && error.isTimeout ? '超时' : '失败';
console.error(`[指令管理] 指令执行${error_type}: ${command_id}, 错误: ${error_message}`, {
command_id: command_id,
duration: duration,
isTimeout: error && error.isTimeout
});
}
}
/**
* 更新指令状态
* @param {number} commandId - 指令ID
* @param {string} status - 状态
* @param {object} result - 结果
* @param {string} errorMessage - 错误信息
* 更新指令状态（统一封装）
* @private
*/
async updateCommandStatus(commandId, status, result = null, errorMessage = null) {
async _update_command_status(command_id, status, result, error_message, start_time, end_time, duration, error_stack) {
try {
const updateData = {
const update_data = {
status: status,
updated_at: new Date()
};
if (status === 'running') {
updateData.start_time = new Date();
} else if (status === 'completed' || status === 'failed') {
updateData.end_time = new Date();
if (result) {
// 将结果转换为JSON字符串，并限制长度（TEXT类型最大约65KB）
let resultStr = JSON.stringify(result);
const maxLength = 60000; // 限制为60KB，留一些余量
if (resultStr.length > maxLength) {
// 如果结果太长，尝试压缩或截断
try {
// 如果是对象，尝试只保存关键信息
if (typeof result === 'object' && result !== null) {
const summary = {
success: result.success !== undefined ? result.success : true,
message: result.message || '执行成功',
dataLength: resultStr.length,
truncated: true,
preview: resultStr.substring(0, 1000) // 保存前1000字符作为预览
};
resultStr = JSON.stringify(summary);
} else {
// 直接截断
resultStr = resultStr.substring(0, maxLength) + '...[数据已截断]';
}
} catch (e) {
// 如果处理失败，直接截断
resultStr = resultStr.substring(0, maxLength) + '...[数据已截断]';
}
}
updateData.result = resultStr;
updateData.progress = 100;
}
if (errorMessage) {
// 错误信息也限制长度
const maxErrorLength = 10000; // 错误信息限制10KB
updateData.error_message = errorMessage.length > maxErrorLength
? errorMessage.substring(0, maxErrorLength) + '...[错误信息已截断]'
: errorMessage;
}
// 计算执行时长
const command = await db.getModel('task_commands').findByPk(commandId);
if (command && command.start_time) {
const duration = new Date() - new Date(command.start_time);
updateData.duration = duration;
// 设置开始时间
if (status === 'running' && start_time) {
update_data.start_time = start_time;
}
// 设置结束时间和执行时长
if ((status === 'completed' || status === 'failed') && end_time) {
update_data.end_time = end_time;
if (duration !== undefined) {
update_data.duration = duration;
}
}
await db.getModel('task_commands').update(updateData, {
where: { id: commandId }
// 处理执行结果
if (result && status === 'completed') {
const result_str = this._format_result_for_storage(result);
update_data.result = result_str;
update_data.progress = 100;
}
// 处理错误信息
if (error_message) {
update_data.error_message = this._truncate_string(error_message, 10000);
}
if (error_stack) {
update_data.error_stack = this._truncate_string(error_stack, 50000);
}
// 更新数据库
await db.getModel('task_commands').update(update_data, {
where: { id: command_id }
});
} catch (error) {
logs.error(`[指令管理] 更新指令状态失败:`, error, {
commandId: commandId,
} catch (db_error) {
logs.error(`[指令管理] 更新指令状态失败:`, db_error, {
command_id: command_id,
status: status
});
// 如果是因为数据太长导致的错误，尝试只保存错误信息
if (error.message && error.message.includes('Data too long')) {
if (db_error.message && db_error.message.includes('Data too long')) {
try {
await db.getModel('task_commands').update({
status: status,
error_message: '结果数据过长，无法保存完整结果',
end_time: new Date(),
end_time: end_time || new Date(),
updated_at: new Date()
}, {
where: { id: commandId }
where: { id: command_id }
});
} catch (e) {
console.error(`[指令管理] 保存截断结果也失败:`, e);
@@ -277,6 +330,89 @@ class CommandManager {
}
}
/**
* 格式化结果数据用于存储
* @private
*/
_format_result_for_storage(result) {
try {
let result_str = JSON.stringify(result);
const max_length = 60000; // 限制为60KB
if (result_str.length > max_length) {
// 如果结果太长，尝试压缩或截断
if (typeof result === 'object' && result !== null) {
const summary = {
success: result.success !== undefined ? result.success : true,
message: result.message || '执行成功',
dataLength: result_str.length,
truncated: true,
preview: result_str.substring(0, 1000) // 保存前1000字符作为预览
};
result_str = JSON.stringify(summary);
} else {
result_str = result_str.substring(0, max_length) + '...[数据已截断]';
}
}
return result_str;
} catch (error) {
return JSON.stringify({ error: '结果序列化失败', message: error.message });
}
}
/**
* 截断字符串
* @private
*/
_truncate_string(str, max_length) {
if (!str || str.length <= max_length) {
return str;
}
return str.substring(0, max_length) + '...[已截断]';
}
/**
* 更新指令状态（兼容旧接口，内部调用统一方法）
* @param {number} command_id - 指令ID
* @param {string} status - 状态
* @param {object} result - 结果
* @param {string} error_message - 错误信息
* @deprecated 建议使用 _update_command_status 统一方法
*/
async updateCommandStatus(command_id, status, result = null, error_message = null) {
const start_time = status === 'running' ? new Date() : null;
const end_time = (status === 'completed' || status === 'failed') ? new Date() : null;
// 计算执行时长
let duration = null;
if (end_time && start_time) {
duration = end_time.getTime() - start_time.getTime();
} else if (end_time) {
// 如果没有开始时间，尝试从数据库获取
try {
const command = await db.getModel('task_commands').findByPk(command_id);
if (command && command.start_time) {
duration = end_time.getTime() - new Date(command.start_time).getTime();
}
} catch (e) {
// 忽略错误
}
}
await this._update_command_status(
command_id,
status,
result,
error_message,
start_time,
end_time,
duration,
null
);
}
/**
* 清理过期的指令记录