data2/JHM.git - Gitblit

#!/usr/bin/env node
 
const fs = require('fs');
const path = require('path');
const { TcpService } = require('./tcp-service');
const { MqttService } = require('./mqtt-service');
const { AliyunService } = require('./aliyun-service');
const { MetricAggregator } = require('./metric-aggregator');
const { getDefaultConfigPath } = require('./runtime-paths');
 
const LOG_LEVELS = {
  debug: 10,
  info: 20,
  warn: 30,
  error: 40,
};
 
// 统一生成本地日志时间，兼顾控制台和文件日志格式。
function formatLocalTimestamp(date = new Date()) {
  const year = date.getFullYear();
  const month = String(date.getMonth() + 1).padStart(2, '0');
  const day = String(date.getDate()).padStart(2, '0');
  const hour = String(date.getHours()).padStart(2, '0');
  const minute = String(date.getMinutes()).padStart(2, '0');
  const second = String(date.getSeconds()).padStart(2, '0');
  const millisecond = String(date.getMilliseconds()).padStart(3, '0');
 
  return `${year}-${month}-${day} ${hour}:${minute}:${second}.${millisecond}`;
}
 
function normalizeLogLevel(level) {
  const normalized = String(level || 'info').trim().toLowerCase();
  return LOG_LEVELS[normalized] ? normalized : 'info';
}
 
function createLogger(loggingConfig = {}) {
  const settings = {
    enabled: loggingConfig.enabled !== false,
    console: loggingConfig.console !== false,
    dir: loggingConfig.dir || path.join(process.cwd(), 'logs'),
    filePrefix: loggingConfig.filePrefix || 'jhm-service',
    level: normalizeLogLevel(loggingConfig.level),
  };
  let currentDate = '';
  let currentFilePath = '';
 
  function getDatePart(date = new Date()) {
    const year = date.getFullYear();
    const month = String(date.getMonth() + 1).padStart(2, '0');
    const day = String(date.getDate()).padStart(2, '0');
    return `${year}${month}${day}`;
  }
 
  function ensureLogFilePath() {
    if (!settings.enabled) {
      return null;
    }
 
    const today = getDatePart();
 
    if (currentFilePath && currentDate === today) {
      return currentFilePath;
    }
 
    fs.mkdirSync(settings.dir, { recursive: true });
    currentDate = today;
    currentFilePath = path.join(settings.dir, `${settings.filePrefix}-${today}.log`);
 
    return currentFilePath;
  }
 
  function shouldWrite(level) {
    return LOG_LEVELS[level] >= LOG_LEVELS[settings.level];
  }
 
  function format(level, message) {
    return `[${formatLocalTimestamp()}] [${level.toUpperCase()}] ${message}`;
  }
 
  function write(level, message) {
    const normalizedLevel = normalizeLogLevel(level);
 
    if (!shouldWrite(normalizedLevel)) {
      return;
    }
 
    const line = format(normalizedLevel, message);
 
    if (settings.console) {
      if (normalizedLevel === 'error') {
        console.error(line);
      } else if (normalizedLevel === 'warn') {
        console.warn(line);
      } else {
        console.log(line);
      }
    }
 
    const logFilePath = ensureLogFilePath();
 
    if (logFilePath) {
      try {
        fs.appendFileSync(logFilePath, `${line}\n`, 'utf8');
      } catch (error) {
        console.error(`[LOGGER] 写入日志文件失败: ${error.message}`);
      }
    }
  }
 
  return {
    debug(message) {
      write('debug', message);
    },
    info(message) {
      write('info', message);
    },
    warn(message) {
      write('warn', message);
    },
    error(message) {
      write('error', message);
    },
    getLogFilePath() {
      ensureLogFilePath();
      return currentFilePath;
    },
    async close() {
      currentDate = '';
    },
  };
}
 
function parseArgs(argv) {
  const options = {
    configPath: getDefaultConfigPath(),
  };
 
  for (let index = 2; index < argv.length; index += 1) {
    const arg = argv[index];
    const value = argv[index + 1];
 
    if (arg === '--config' && value) {
      options.configPath = path.resolve(value);
      index += 1;
    } else if (arg === '--help' || arg === '-h') {
      options.help = true;
    }
  }
 
  return options;
}
 
function printHelp() {
  console.log(`JHM TCP Socket 网关服务
 
用法:
  node app.js
  node app.js --config ./config.json
  jhm-service.exe --config .\\runtime\\config.json
  ./jhm-service --config ./runtime/config.json
 
参数:
  --config <path>    指定配置文件路径
  --help, -h         显示帮助
`);
}
 
function loadConfig(configPath) {
  if (!fs.existsSync(configPath)) {
    throw new Error(`未找到配置文件: ${configPath}`);
  }
 
  const content = fs.readFileSync(configPath, 'utf8');
  const config = JSON.parse(content);
 
  validateConfig(config);
 
  return config;
}
 
function getSendChannels(config) {
  const channels = Array.isArray(config.send && config.send.channels)
    ? config.send.channels
    : ['mqtt'];
 
  return Array.from(new Set(channels
    .map((item) => String(item || '').trim().toLowerCase())
    .filter(Boolean)));
}
 
function getSendOptions(config) {
  const send = config.send || {};
  const mode = ['immediate', 'batch'].includes(String(send.mode || '').trim().toLowerCase())
    ? String(send.mode).trim().toLowerCase()
    : 'batch';
 
  return {
    mode,
    flushIntervalMs: Number(send.flushIntervalMs) > 0 ? Number(send.flushIntervalMs) : 60000,
    alignToMinute: send.alignToMinute !== false,
    includeDeviceIdField: send.includeDeviceIdField !== false,
    deviceIdField: send.deviceIdField || 'n',
    publishOnShutdown: send.publishOnShutdown !== false,
  };
}
 
function getBloodPressureOptions(config) {
  const bloodPressure = config.protocol && config.protocol.bloodPressure;
 
  return {
    publishTime: bloodPressure ? bloodPressure.publishTime !== false : true,
    flushImmediately: bloodPressure ? bloodPressure.flushImmediately !== false : true,
  };
}
 
function validateConfig(config) {
  if (!config.tcp || !Array.isArray(config.devices)) {
    throw new Error('config.json 必须包含 tcp 和 devices');
  }
 
  if (!config.tcp.host || !config.tcp.port) {
    throw new Error('config.json 必须包含 tcp.host 和 tcp.port');
  }
 
  const channels = getSendChannels(config);
 
  if (channels.length === 0) {
    throw new Error('config.json 的 send.channels 至少要启用一个通道');
  }
 
  if (channels.includes('mqtt')) {
    if (!config.mqtt) {
      throw new Error('config.json 已启用 mqtt，但缺少 mqtt 配置');
    }
 
    const hasBrokerUrl = Boolean(config.mqtt.brokerUrl);
    const hasHostMode = Boolean(config.mqtt.protocol && config.mqtt.host && config.mqtt.port);
 
    if (!hasBrokerUrl && !hasHostMode) {
      throw new Error('config.json 的 mqtt 必须配置 brokerUrl 或 protocol/host/port');
    }
 
    if (!config.mqtt.topicTemplate && !config.mqtt.defaultTopicPrefix) {
      throw new Error('config.json 的 mqtt 必须配置 topicTemplate 或 defaultTopicPrefix');
    }
  }
 
  if (channels.includes('aliyun')) {
    if (!config.aliyun) {
      throw new Error('config.json 已启用 aliyun，但缺少 aliyun 配置');
    }
 
    if (!config.aliyun.tupleApiBaseUrl && !config.aliyun.tupleApiUrl) {
      throw new Error('config.json 的 aliyun 必须配置 tupleApiBaseUrl 或 tupleApiUrl');
    }
  }
 
  if (!config.protocol || !config.protocol.alModelPath) {
    throw new Error('config.json 必须配置 protocol.alModelPath');
  }
 
  const sendOptions = getSendOptions(config);
 
  if (sendOptions.flushIntervalMs <= 0) {
    throw new Error('config.json 的 send.flushIntervalMs 必须大于 0');
  }
 
  if (config.logging && config.logging.level) {
    const supportedLevels = ['debug', 'info', 'warn', 'error'];
 
    if (!supportedLevels.includes(String(config.logging.level).toLowerCase())) {
      throw new Error(`config.json 的 logging.level 仅支持 ${supportedLevels.join('、')}`);
    }
  }
}
 
function resolveConfigPath(configFilePath, targetPath) {
  if (path.isAbsolute(targetPath)) {
    return targetPath;
  }
 
  return path.join(path.dirname(configFilePath), targetPath);
}
 
function createOnMetricHandler({ logger, aggregator, sendOptions, bloodPressureOptions }) {
  return function onMetric(device, metric, result) {
    logger.info(`[APP] 指标已缓存 deviceId=${device.deviceId} mode=${sendOptions.mode} metric=${JSON.stringify(metric)}`);
    return aggregator.ingest(device, metric)
      .then(() => {
        if (
          sendOptions.mode === 'batch'
          && bloodPressureOptions.flushImmediately
          && result
          && result.protocol === 'blood-pressure'
        ) {
          logger.info(`[APP] 血压报文触发整包立即发送 deviceId=${device.deviceId}`);
          return aggregator.flush({ reason: 'blood-pressure', deviceId: device.deviceId });
        }
 
        return null;
      })
      .catch((error) => {
        logger.error(`[APP] 指标处理失败 deviceId=${device.deviceId}: ${error.message}`);
      });
  };
}
 
async function main() {
  const options = parseArgs(process.argv);
 
  if (options.help) {
    printHelp();
    return;
  }
 
  const bootstrapLogger = createLogger({
    enabled: false,
    console: true,
  });
  bootstrapLogger.info(`[APP] 启动参数解析完成 config=${options.configPath}`);
 
  const config = loadConfig(options.configPath);
  const logDir = resolveConfigPath(options.configPath, (config.logging && config.logging.dir) || './logs');
  const logger = createLogger({
    ...(config.logging || {}),
    dir: logDir,
  });
  const sendChannels = getSendChannels(config);
  const sendOptions = getSendOptions(config);
  const bloodPressureOptions = getBloodPressureOptions(config);
  const alModelPath = resolveConfigPath(options.configPath, config.protocol.alModelPath);
 
  if (config.logging && config.logging.enabled === false) {
    logger.info('[APP] 文件日志已禁用，仅输出到控制台');
  } else {
    logger.info(`[APP] 本地日志已启用 dir=${logDir} file=${logger.getLogFilePath()} level=${normalizeLogLevel(config.logging && config.logging.level)}`);
  }
 
  logger.info(`[APP] 配置加载完成 tcp=${config.tcp.host}:${config.tcp.port} devices=${config.devices.length} channels=${sendChannels.join(',')} sendMode=${sendOptions.mode} flushIntervalMs=${sendOptions.flushIntervalMs} alModel=${alModelPath}`);
 
  const mqttService = sendChannels.includes('mqtt') ? new MqttService(config.mqtt, logger) : null;
  const aliyunService = sendChannels.includes('aliyun') ? new AliyunService(config.aliyun, logger) : null;
 
  // 聚合器负责把多次单点指标合并成一次完整物模型上报。
  const aggregator = new MetricAggregator({
    logger,
    ...sendOptions,
    onFlush: async (device, payload, meta) => {
      logger.info(`[APP] 开始批量分发 deviceId=${device.deviceId} channels=${sendChannels.join(',')} reason=${meta.reason} payload=${JSON.stringify(payload)}`);
 
      let ok = true;
 
      if (mqttService) {
        try {
          mqttService.publish(device, payload);
        } catch (error) {
          ok = false;
          logger.error(`[APP] MQTT 批量分发失败 deviceId=${device.deviceId}: ${error.message}`);
        }
      }
 
      if (aliyunService) {
        try {
          const result = await aliyunService.publish(device, payload);
 
          if (result && result.skipped) {
            ok = false;
            logger.warn(`[APP] 阿里云批量分发已跳过 deviceId=${device.deviceId} reason=${result.reason}`);
          } else if (result && result.ok === false) {
            ok = false;
            logger.error(`[APP] 阿里云批量分发失败 deviceId=${device.deviceId}: ${result.reason}`);
          }
        } catch (error) {
          ok = false;
          logger.error(`[APP] 阿里云批量分发异常 deviceId=${device.deviceId}: ${error.message}`);
        }
      }
 
      return ok;
    },
  });
 
  const tcpService = new TcpService({
    tcpConfig: config.tcp,
    devices: config.devices,
    alModelPath,
    publishBloodPressureTime: bloodPressureOptions.publishTime,
    logger,
    onMetric: createOnMetricHandler({
      logger,
      aggregator,
      sendOptions,
      bloodPressureOptions,
    }),
  });
 
  if (mqttService) {
    logger.info('[APP] 正在启动 MQTT 通道');
    mqttService.start();
  }
 
  if (aliyunService) {
    logger.info('[APP] 正在启动阿里云通道');
    aliyunService.start();
  }
 
  aggregator.start();
  await tcpService.start();
  logger.info('[APP] 所有服务已启动');
 
  let shuttingDown = false;
 
  async function shutdown(signal) {
    if (shuttingDown) {
      return;
    }
 
    shuttingDown = true;
    logger.warn(`[APP] 收到 ${signal}，开始关闭服务`);
 
    await tcpService.stop();
    await aggregator.stop();
 
    if (mqttService) {
      await mqttService.stop();
    }
 
    if (aliyunService) {
      await aliyunService.stop();
    }
 
    logger.info('[APP] 服务已完成关闭');
    await logger.close();
    process.exit(0);
  }
 
  process.on('SIGINT', () => {
    shutdown('SIGINT');
  });
 
  process.on('SIGTERM', () => {
    shutdown('SIGTERM');
  });
 
  process.on('uncaughtException', (error) => {
    logger.error(`[APP] 未捕获异常: ${error.stack || error.message}`);
  });
 
  process.on('unhandledRejection', (reason) => {
    logger.error(`[APP] 未处理的 Promise 拒绝: ${reason}`);
  });
}
 
if (require.main === module) {
  main().catch((error) => {
    console.error(error.message);
    process.exit(1);
  });
}
 
module.exports = {
  createLogger,
  formatLocalTimestamp,
  getBloodPressureOptions,
  getSendChannels,
  getSendOptions,
  loadConfig,
  main,
  createOnMetricHandler,
  normalizeLogLevel,
  parseArgs,
  printHelp,
  resolveConfigPath,
  validateConfig,
};