jianhen-voc/backend/api/module/schedule/sp-api-schedule.ts

/**
 * Amazon SP-API 数据采集定时任务
 * 
 * 功能：
 * - 每天凌晨1点自动执行数据采集
 * - 支持多店铺数据同步
 * - 采集 Listing、订单、报表数据
 * - 自动解析未处理的报表
 * 
 * 使用方式：
 * import { spApiScheduler } from './modules/sp-api-schedule.ts';
 * await spApiScheduler.start();
 */

import nodeCron from 'npm:node-cron';

class SpApiScheduler {
  // 私有属性（先声明）
  #isRunning = false;
  #lastRunTime = null;
  #cronTask= null;
  #reportCronTask = null;
  #cronEnabled = true;

  // ========== 私有工具方法（最优先声明，避免调用时未定义） ==========
  /**
   * 延迟函数
   * @private
   * @param {number} ms - 延迟毫秒数
   * @returns {Promise<void>} 延迟 Promise
   */
  #delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  /**
   * 计算下次执行时间（凌晨1点）
   * @private
   * @returns {Date} 下次执行时间
   */
  #calculateNextRunTime() {
    const now = new Date();
    const next = new Date(now);
    next.setHours(1, 0, 0, 0);
    if (next <= now) {
      next.setDate(next.getDate() + 1);
    }
    return next;
  }

  /**
   * 获取所有活跃的 Amazon 店铺
   * @private
   * @returns {Promise<any[]>} 活跃店铺列表
   */
  async #getActiveAmazonShops(shopId) {
    const Parse = globalThis.Parse
    const query = new Parse.Query('Shop');
    query.equalTo('platform', 'amazon');
    query.equalTo('status', 'active');
    if(shopId) {
      query.equalTo('objectId', shopId);
    }
    query.limit(1000);
    return await query.find();
  }

  /**
   * 记录执行日志
   * @private
   * @param {any} logData - 日志数据对象
   * @returns {Promise<void>}
   */
  async #logExecution(logData) {
    try {
      const Parse = globalThis.Parse
      const TaskLog = Parse.Object.extend('TaskExecutionLog');
      const log = new TaskLog();
      log.set('taskName', logData.taskName);
      log.set('startTime', logData.startTime);
      log.set('endTime', logData.endTime);
      log.set('duration', logData.duration);
      log.set('successCount', logData.successCount);
      log.set('errorCount', logData.errorCount);
      log.set('errors', logData.errors);
      log.set('status', logData.status);
      await log.save(null, { useMasterKey: true });
      console.log('[SP-API Scheduler] 执行日志已保存');
    } catch (error) {
      console.error('[SP-API Scheduler] 保存执行日志失败:', error.message);
    }
  }

  /**
   * 采集 Listing 数据
   * @private
   * @param {string} shopId - 店铺ID
   * @param {any} config - 店铺配置
   * @param {string} marketplaceId - 市场ID
   * @returns {Promise<void>}
   */
  async #collectListings(shopId, config, marketplaceId) {
    const sellerId = config.SpApiConfig.sellerID;
    marketplaceId = marketplaceId || 'ATVPDKIKX0DER';
    const Parse = globalThis.Parse;
    let lastUpdatedAfter;
    try {
      const shopPointer = Parse.Object.extend('Shop').createWithoutData(shopId);
      const query = new Parse.Query('Listing');
      query.equalTo('shop', shopPointer);
      query.descending('lastUpdatedDate');
      query.limit(1);
      const latestListing = await query.first({ useMasterKey: true });

      if (latestListing && latestListing.get('lastUpdatedDate')) {
        lastUpdatedAfter = latestListing.get('lastUpdatedDate').toISOString();
        console.log(`[SP-API Scheduler] 店铺 ${shopId} 使用最新的 lastUpdatedDate: ${lastUpdatedAfter}`);
      } else {
        const oneYearAgo = new Date(Date.now() - 365 * 24 * 60 * 60 * 1000);
        lastUpdatedAfter = oneYearAgo.toISOString();
        console.log(`[SP-API Scheduler] 店铺 ${shopId} 没有Listing数据,使用一年前的时间: ${lastUpdatedAfter}`);
      }
    } catch (error) {
      const oneYearAgo = new Date(Date.now() - 365 * 24 * 60 * 60 * 1000);
      lastUpdatedAfter = oneYearAgo.toISOString();
      console.log(`[SP-API Scheduler] 店铺 ${shopId} 查询Listing失败,使用一年前的时间: ${lastUpdatedAfter}, 错误: ${error.message}`);
    }
    let path = `/listings/2021-08-01/items/${sellerId}?marketplaceIds=${marketplaceId}&includedData=summaries,attributes,issues&withStatus=BUYABLE,DISCOVERABLE&sortBy=lastUpdatedDate&sortOrder=ASC&pageSize=10&lastUpdatedAfter=${lastUpdatedAfter}`
    const response = await fetch('http://localhost:3000/api/amazon/forward', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
      body: JSON.stringify({
        path: path,
        method: 'GET',
        functionName: "cleanListings"
      })
    });
    const result = await response.json();
    if (!result.success) {
      throw new Error(result.message || 'Listing 采集失败');
    }
    console.log(`[SP-API Scheduler] 店铺 ${shopId} Listing 数据采集完成: ${result.processed || 0} 条记录`);
  }

  /**
   * 采集订单数据（增量同步）
   * @private
   * @param {string} shopId - 店铺ID
   * @param {any} config - 店铺配置
   * @param {string[]} marketplaceIds - 市场ID列表
   * @param {any} shop - 店铺对象
   * @returns {Promise<void>}
   */
  async #collectOrders(shopId, config, marketplaceId) {
    marketplaceId = marketplaceId || 'ATVPDKIKX0DER';
    const Parse = globalThis.Parse;
    let createdAfter;
    try {
      const shopPointer = Parse.Object.extend('Shop').createWithoutData(shopId);
      const query = new Parse.Query('Order');
      query.equalTo('shop', shopPointer);
      query.descending('orderDate');
      query.limit(1);
      const latestOrder = await query.first({ useMasterKey: true });
      if (latestOrder && latestOrder.get('orderDate')) {
        createdAfter = latestOrder.get('orderDate').toISOString();
        console.log(`[SP-API Scheduler] 店铺 ${shopId} 使用最新的 orderDate: ${createdAfter}`);
      } else {
        const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
        createdAfter = ninetyDaysAgo.toISOString();
        console.log(`[SP-API Scheduler] 店铺 ${shopId} 没有Order数据,使用90天前的时间: ${createdAfter}`);
      }
    } catch (error) {
      const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
      createdAfter = ninetyDaysAgo.toISOString();
      console.log(`[SP-API Scheduler] 店铺 ${shopId} 查询Order失败,使用90天前的时间: ${createdAfter}, 错误: ${error.message}`);
    }
    const response = await fetch('http://localhost:3000/api/amazon/forward', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', "shop-objectid": shopId },
      body: JSON.stringify({
        path: `/orders/v0/orders?MarketplaceIds=${marketplaceId}&CreatedAfter=${createdAfter}`,
        method: 'GET',
        functionName: 'cleanOrders'
      })
    });
    const result = await response.json();
    if (!result.success) {
      throw new Error(result.message || '订单采集失败');
    }
    console.log(`[SP-API Scheduler] 店铺 ${shopId} 订单数据采集完成: ${result.processed || 0} 条记录`);
  }

  /**
   * 采集报表数据
   * @private
   * @param {string} shopId - 店铺ID
   * @param {any} config - 店铺配置
   * @param {string[]} marketplaceId - 市场ID列表
   * @returns {Promise<void>}
   */
  async #collectReports(shopId, config, marketplaceId) {
    const Parse = globalThis.Parse;
    const reportTypes = ['GET_FBA_FULFILLMENT_CUSTOMER_RETURNS_DATA'];

    for (const reportType of reportTypes) {
      try {
        // 1. 查询上次 dataEndTime 作为本次 dataStartTime
        let dataStartTime;
        try {
          const q = new Parse.Query('Reports');
          q.equalTo('shop', shopId);
          q.equalTo('reportType', reportType);
          // q.equalTo('marketplaceIds', [marketplaceId);
          q.descending('dataEndTime');
          q.limit(1);
          const last = await q.first({ useMasterKey: true });
          if (last && last.get('dataEndTime')) {
            dataStartTime = last.get('dataEndTime').toISOString();
          } else {
            dataStartTime = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000).toISOString();
          }
        } catch (e) {
          dataStartTime = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000).toISOString();
        }
        const dataEndTime = new Date().toISOString();
        console.log(`[SP-API Scheduler] 店铺 ${shopId} 创建报表 ${reportType}, range: ${dataStartTime} ~ ${dataEndTime}`);

        // 2. 创建报表
        const createRes = await fetch('http://localhost:3000/api/amazon/forward', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
          body: JSON.stringify({
            path: '/reports/2021-06-30/reports',
            method: 'POST',
            body: { reportType, marketplaceIds: [marketplaceId], dataStartTime, dataEndTime }
          })
        });
        const createResult = await createRes.json();
        const reportId = createResult?.data?.reportId;
        if (!reportId) {
          console.error(`[SP-API Scheduler] 创建报表失败 ${reportType}:`, JSON.stringify(createResult));
          continue;
        }
        console.log(`[SP-API Scheduler] 报表已创建, reportId: ${reportId}`);
        await this.#delay(3000);

        // 3. 获取报表状态
        const getRes = await fetch('http://localhost:3000/api/amazon/forward', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
          body: JSON.stringify({
            path: `/reports/2021-06-30/reports/${reportId}`,
            method: 'GET'
          })
        });
        const reportData = await getRes.json();
        const reportInfo = reportData?.data || {};

        // 4. 存入 Reports 表
        const ReportObj = Parse.Object.extend('Reports');
        const rpt = new ReportObj();
        rpt.set('shop', {
          __type: 'Pointer',
          className: 'Shop',
          objectId: shopId
        });
        rpt.set('reportId', reportId);
        rpt.set('reportType', reportType);
        rpt.set('marketplaceIds', reportInfo.marketplaceIds || []);
        rpt.set('dataStartTime', reportInfo.dataStartTime ? new Date(reportInfo.dataStartTime) : null);
        rpt.set('dataEndTime', reportInfo.dataEndTime ? new Date(reportInfo.dataEndTime) : null);
        rpt.set('createdTime', reportInfo.createdTime ? new Date(reportInfo.createdTime) : null);
        rpt.set('processingStartTime', reportInfo.processingStartTime ? new Date(reportInfo.processingStartTime) : null);
        rpt.set('processingEndTime', reportInfo.processingEndTime ? new Date(reportInfo.processingEndTime) : null);
        rpt.set('processingStatus', reportInfo.processingStatus || 'IN_QUEUE');
        rpt.set('reportDocumentId', reportInfo.reportDocumentId || null);
        rpt.set('isParsed', false);
        await rpt.save(null, { useMasterKey: true });
        console.log(`[SP-API Scheduler] 报表已保存到 Reports 表, status: ${reportInfo.processingStatus}`);
        await this.#delay(2000);
      } catch (error) {
        console.error(`[SP-API Scheduler] 报表 ${reportType} 处理失败: ${error.message}`);
      }
    }
  }

  /**
   * 处理未解析的报表
   * @private
   * @param {string} shopId - 店铺ID
   * @returns {Promise<void>}
   */
  async #processUnparsedReports(shopId) {
    const Parse = globalThis.Parse
    const query = new Parse.Query('Reports');
    query.equalTo('shop', shopId);
    query.equalTo('isParsed', false);
    query.equalTo('processingStatus', 'DONE');
    query.limit(1000);

    const unparsedReports = await query.find();
    if (unparsedReports.length === 0) {
      console.log(`[SP-API Scheduler] 店铺 ${shopId} 没有未解析的报表`);
      return;
    }
    console.log(`[SP-API Scheduler] 发现 ${unparsedReports.length} 个未解析报表`);
    for (const report of unparsedReports) {
      try {
        const reportDocumentId = report.get('reportDocumentId');
        const reportType = report.get('reportType');

        const response = await fetch('http://localhost:3000/api/amazon/processReport', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json', "shop-objectid": shopId },
          body: JSON.stringify({
            reportDocumentId,
            reportType,
            shopId
          })
        });
        const result = await response.json();
        if (result.success) {
          report.set('isParsed', true);
          report.set('parsedAt', new Date());
          report.set('parsedRecords', result.processed || 0);
          await report.save(null, { useMasterKey: true });
          console.log(`[SP-API Scheduler] 报表 ${reportDocumentId} 解析完成: ${result.processed} 条记录`);
        }
        await this.#delay(10000);
      } catch (error) {
        console.error(`[SP-API Scheduler] 报表解析失败: ${error.message}`);
      }
    }
  }

  /**
   * 检查并更新未完成的报表状态
   * 每天凌晨2点执行：查询 Reports 表中 status 不为 DONE 的报表，
   * 通过 reportId 获取最新状态并更新，然后解析已完成的报表
   * @private
   * @returns {Promise<void>}
   */
  async #checkPendingReports(shopid) {
    const Parse = globalThis.Parse;
    console.log('[SP-API Scheduler] ========================================');
    console.log('[SP-API Scheduler] 开始检查未完成的报表状态');
    console.log('[SP-API Scheduler] 执行时间:', new Date().toISOString());
    console.log('[SP-API Scheduler] ========================================');

    try {
      // 1. 查询所有 processingStatus 不为 DONE 的报表
      const query = new Parse.Query('Reports');
      query.notEqualTo('processingStatus', 'DONE');
      if(shopid) {
        query.equalTo('shop', shopid);
      }
      query.limit(1000);
      const pendingReports = await query.find({ useMasterKey: true });
      if (pendingReports.length === 0) {
        console.log('[SP-API Scheduler] 没有未完成的报表');
        return;
      }

      console.log(`[SP-API Scheduler] 发现 ${pendingReports.length} 个未完成报表`);

      // 2. 逐个通过 reportId 获取最新状态
      const shopIdsToProcess = new Set();
      for (const report of pendingReports) {
        try {
          const reportId = report.get('reportId');
          const shopId = report.get('shop').id;
          if (!reportId || !shopId) continue;

          console.log(`[SP-API Scheduler] 检查报表 ${reportId} 状态...`);
          const getRes = await fetch('http://localhost:3000/api/amazon/forward', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
            body: JSON.stringify({
              path: `/reports/2021-06-30/reports/${reportId}`,
              method: 'GET'
            })
          });
          const reportData = await getRes.json();
          const reportInfo = reportData?.data || {};

          // 3. 更新 Reports 表
          const newStatus = reportInfo.processingStatus || report.get('processingStatus');
          report.set('processingStatus', newStatus);
          if (reportInfo.reportDocumentId) {
            report.set('reportDocumentId', reportInfo.reportDocumentId);
          }
          await report.save(null, { useMasterKey: true });
          console.log(`[SP-API Scheduler] 报表 ${reportId} 状态更新为: ${newStatus}`);

          if (newStatus === 'DONE') {
            shopIdsToProcess.add(shopId);
          }

          await this.#delay(10000);
        } catch (error) {
          console.error(`[SP-API Scheduler] 更新报表状态失败: ${error.message}`);
        }
      }

      // 4. 对有新完成报表的店铺执行 processUnparsedReports
      for (const shopId of shopIdsToProcess) {
        try {
          console.log(`[SP-API Scheduler] 解析店铺 ${shopId} 的已完成报表`);
          await this.#processUnparsedReports(shopId);
        } catch (error) {
          console.error(`[SP-API Scheduler] 店铺 ${shopId} 报表解析失败: ${error.message}`);
        }
      }

      console.log('[SP-API Scheduler] 未完成报表检查完毕');
    } catch (error) {
      console.error(`[SP-API Scheduler] 检查未完成报表失败: ${error.message}`);
    }
  }

  /**
   * 处理单个店铺的数据采集
   * @private
   * @param {any} shop - 店铺对象
   * @returns {Promise<void>}
   */
  async #processShop(shop){
    const shopId = shop.id;
    const shopName = shop.get('name');
    const config = shop.get('config');
    const marketplaceId = shop.get('marketplaceId');

    if (!config || !config.SpApiConfig) {
      throw new Error('店铺配置缺失');
    }
    // 1. 采集 Listing 数据
    try {
      console.log(`[SP-API Scheduler] 采集店铺 ${shopName} 的 Listing 数据`);
      await this.#collectListings(shopId, config, marketplaceId);
      await this.#delay(3000);
    } catch (error) {
      console.error(`[SP-API Scheduler] Listing 采集失败: ${error.message}`);
    }

    // 2. 采集订单数据（增量同步）
    try {
      console.log(`[SP-API Scheduler] 采集店铺 ${shopName} 的订单数据`);
      await this.#collectOrders(shopId, config, marketplaceId);
      await this.#delay(3000);
    } catch (error) {
      console.error(`[SP-API Scheduler] 订单采集失败: ${error.message}`);
    }

    // 3. 采集报表数据
    try {
      console.log(`[SP-API Scheduler] 采集店铺 ${shopName} 的报表数据`);
      await this.#collectReports(shopId, config, marketplaceId);
      await this.#delay(3000);
    } catch (error) {
      console.error(`[SP-API Scheduler] 报表采集失败: ${error.message}`);
    }

    // 4. 处理未解析的报表
    try {
      console.log(`[SP-API Scheduler] 处理店铺 ${shopName} 的未解析报表`);
      await this.#checkPendingReports(shopId);
    } catch (error) {
      console.error(`[SP-API Scheduler] 报表处理失败: ${error.message}`);
    }

    // 5. 更新店铺同步状态
    shop.set('lastSyncTime', new Date());
    shop.set('syncStatus', 'completed');
    await shop.save(null, { useMasterKey: true });
  }

  /**
   * 执行数据采集任务（核心方法：移到调用方之前）
   * @private
   * @returns {Promise<any>} 执行结果对象，包含成功数、失败数、耗时等信息
   */
  async #executeDataCollection(shopId) {
    if (this.#isRunning) {
      console.log('[SP-API Scheduler] 任务正在执行中，跳过本次调度');
      return {
        success: false,
        message: '任务正在执行中',
        successCount: 0,
        errorCount: 0,
        duration: 0,
        errors: []
      };
    }
    this.#isRunning = true;
    const startTime = Date.now();
    const errors = [];
    let successCount = 0;
    let errorCount = 0;
    console.log('[SP-API Scheduler] ========================================');
    console.log('[SP-API Scheduler] 开始执行每日数据采集任务');
    console.log('[SP-API Scheduler] 执行时间:', new Date().toISOString());
    console.log('[SP-API Scheduler] ========================================');

    try {
      // 1. 获取所有 Amazon 平台的活跃店铺
      const shops = await this.#getActiveAmazonShops(shopId);

      if (shops.length === 0) {
        console.log('[SP-API Scheduler] 没有找到活跃的 Amazon 店铺');
        return {
          success: true,
          message: '没有活跃店铺需要处理',
          successCount: 0,
          errorCount: 0,
          duration: Date.now() - startTime,
          errors: []
        };
      }

      console.log(`[SP-API Scheduler] 发现 ${shops.length} 个 Amazon 店铺`);

      // 2. 循环处理每个店铺
      for (const shop of shops) {
        try {
          console.log(`\n[SP-API Scheduler] ----------------------------------------`);
          console.log(`[SP-API Scheduler] 开始处理店铺: ${shop.get('name')} (${shop.id})`);

          await this.#processShop(shop);
          successCount++;
          console.log(`[SP-API Scheduler] 店铺 ${shop.get('name')} 处理完成`);
          await this.#delay(2000);

        } catch (error) {
          errorCount++;
          const errorMsg = `店铺 ${shop.get('name')} 处理失败: ${error.message}`;
          errors.push(errorMsg);
          console.error(`[SP-API Scheduler] ${errorMsg}`);
        }
      }
      // 3. 记录执行日志
      const duration = Date.now() - startTime;
      await this.#logExecution({
        taskName: 'sp-api-daily-sync',
        startTime: new Date(startTime),
        endTime: new Date(),
        duration,
        successCount,
        errorCount,
        errors,
        status: errorCount === 0 ? 'success' : (successCount > 0 ? 'partial_success' : 'failed')
      });

      this.#lastRunTime = new Date();
      const message = `数据采集完成: 成功 ${successCount}/${shops.length} 个店铺，耗时 ${Math.round(duration / 1000)}秒`;
      
      console.log(`\n[SP-API Scheduler] ========================================`);
      console.log(`[SP-API Scheduler] ${message}`);
      console.log(`[SP-API Scheduler] ========================================\n`);

      return {
        success: errorCount === 0,
        message,
        successCount,
        errorCount,
        duration,
        errors
      };

    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `数据采集任务执行失败: ${error.message}`;
      console.error(`[SP-API Scheduler] ${errorMsg}`);

      return {
        success: false,
        message: errorMsg,
        successCount,
        errorCount: errorCount + 1,
        duration,
        errors: [...errors, errorMsg]
      };
    } finally {
      this.#isRunning = false;
    }
  }

  // ========== 公共方法（后声明，因为依赖前面的私有方法） ==========
  /**
   * 构造函数
   * 初始化 Amazon SP-API 数据采集调度器
   */
  constructor() {
    console.log('[SP-API Scheduler] 初始化 Amazon SP-API 数据采集调度器');
  }

  /**
   * 启动定时任务
   * 每天凌晨1点自动执行数据采集
   * @returns {Promise<void>}
   */
  async start() {
    if (this.#cronTask) {
      console.log('[SP-API Scheduler] 定时任务已在运行中');
      return;
    }
    // 每天凌晨1点执行 (Asia/Shanghai 时区)
    this.#cronTask = nodeCron.schedule('0 1 * * *', async () => {
      await this.#executeDataCollection();
    }, {
      timezone: "Asia/Shanghai"
    });

    console.log('[SP-API Scheduler] 定时任务已启动，将在每天凌晨1点执行');

    // 每天凌晨2点检查未完成的报表状态并解析
    this.#reportCronTask = nodeCron.schedule('0 2 * * *', async () => {
      await this.#checkPendingReports();
    }, {
      timezone: "Asia/Shanghai"
    });

    console.log('[SP-API Scheduler] 报表状态检查任务已启动，将在每天凌晨2点执行');

    // 测试模式：每5分钟执行一次（开发时可取消注释）
    // this.#cronTask = nodeCron.schedule('*/5 * * * *', async () => {
    //   await this.#executeDataCollection();
    // });
    // console.log('[SP-API Scheduler] 测试模式：每5分钟执行一次');
  }

  /**
   * 停止定时任务
   * @returns {void}
   */
  stop(){
    if (this.#cronTask) {
      this.#cronTask.stop();
      this.#cronTask = null;
      console.log('[SP-API Scheduler] 数据采集定时任务已停止');
    }
    if (this.#reportCronTask) {
      this.#reportCronTask.stop();
      this.#reportCronTask = null;
      console.log('[SP-API Scheduler] 报表状态检查定时任务已停止');
    }
  }

  /**
   * 获取调度器状态
   * @returns {object} 调度器状态对象
   */
  getStatus(){
    const nextRunTime = this.#calculateNextRunTime();
    return {
      isRunning: this.#isRunning,
      lastRunTime: this.#lastRunTime,
      nextRunTime,
      cronEnabled: this.#cronEnabled
    };
  }

  /**
   * 手动触发数据采集
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerManually(shopId) {
    console.log('[SP-API Scheduler] 手动触发数据采集任务');
    return await this.#executeDataCollection(shopId); // 此时#executeDataCollection已声明
  }
}

// 导出单例
export const spApiScheduler = new SpApiScheduler();