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

/**
 * Sorftime API 数据采集定时任务
 * 
 * 功能：
 * - 每月1号凌晨获取类目热销产品BRS 400
 * - 每周一凌晨1点通过类目反查关键词
 * 
 * 使用方式：
 * import { sorftimeScheduler } from './modules/sorftime-api-schedule.ts';
 * await sorftimeScheduler.start();
 */

// node-cron 将在需要时动态加载，避免编译时依赖问题
import nodeCron from 'npm:node-cron';



class SorftimeScheduler {
  // 私有属性（先声明）
  #isRunning = false;
  #lastRunTime = null;
  #categoryProductsCronTask = null;
  #keywordsCronTask = null;
  #marketTrendCronTask = null;
  #shopProductsCronTask = null;
  #productDetailCronTask = null;
  #cloudFunctionCronTask = null;
  #reviewsCronTask = null;
  #cronEnabled = true;

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

  /**
   * 获取所有叶子节点类目
   * @private
   * @returns {Promise<any[]>} 叶子节点类目列表
   */
  async #getLeafCategories() {
    const Parse = globalThis.Parse;
    const query = new Parse.Query('SelfCategory');
    query.equalTo('isLeaf', true);
    
    query.limit(10000);
    return await query.find({ useMasterKey: true });
  }

  /**
   * 根据 shopId 或 nodeIds 筛选类目
   * - nodeIds 优先：直接按 nodeId 数组过滤 SelfCategory
   * - shopId：读取店铺的 nodeIds 字段后过滤 SelfCategory
   * - 两者均未传：返回全部叶子节点类目
   * @private
   * @param {string} shopId - 可选
   * @param {string[]} nodeIds - 可选
   * @returns {Promise<any[]>} 类目列表
   */
  async #getLeafCategoriesByFilter(shopId, nodeIds) {
    const Parse = globalThis.Parse;
    if (nodeIds && nodeIds.length > 0) {
      const query = new Parse.Query('SelfCategory');
      query.containedIn('nodeId', nodeIds);
      query.limit(10000);
      return await query.find({ useMasterKey: true });
    }
    if (shopId) {
      const shopQuery = new Parse.Query('Shop');
      const shop = await shopQuery.get(shopId, { useMasterKey: true });
      const shopNodeIds = shop.get('nodeIds') || [];
      if (shopNodeIds.length > 0) {
        const catQuery = new Parse.Query('SelfCategory');
        catQuery.containedIn('nodeId', shopNodeIds);
        catQuery.limit(10000);
        return await catQuery.find({ useMasterKey: true });
      }
    }
    return await this.#getLeafCategories();
  }

  /**
   * 获取活跃的 Amazon 店铺列表
   * @private
   * @param {string} shopId - 可选，传入则只返回该店铺
   * @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('[Sorftime Scheduler] 执行日志已保存');
    } catch (error) {
      console.error('[Sorftime Scheduler] 保存执行日志失败:', error.message);
    }
  }

  /**
   * 采集类目热销产品（每月1号凌晨执行）
   * @private
   * @param {string} nodeId - 类目节点ID
   * @param {number} domain - 站点域名代码
   * @returns {Promise<void>}
   */
  async #collectCategoryProducts(nodeId, domain, progress) {
    console.log(`[Sorftime Scheduler] 开始采集类目 ${nodeId} 的热销产品 (domain: ${domain})`);
    
    // 固定的云函数ID
    const functionId = 'ZmYNPsoX9X';
    
    // 每页100条，共需要请求4次获取400条数据
    for (let page = 1; page <= 4; page++) {
      try {
        progress?.({ type: 'info', message: `类目 ${nodeId} 第 ${page}/4 页热销产品采集中...` });
        const requestBody = {
          path: "/api/CategoryProducts",
          method: "POST",
          query: { domain },
          body: {
            NodeId: nodeId,
            Page: page,
            Range: 400
          },
          functionId
        };
        
        const response = await fetch('http://localhost:3000/api/sorftime/forward', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify(requestBody)
        });

        const result = await response.json();
        
        if (result.Code !== 0) {
          throw new Error(result.message || `第${page}页数据采集失败`);
        }

        console.log(`[Sorftime Scheduler] 类目 ${nodeId} 第${page}页数据采集完成`);  
        // 请求间隔，避免频率过高
        await this.#delay(2000);
      } catch (error) {
        console.error(`[Sorftime Scheduler] 类目 ${nodeId} 第${page}页采集失败:`, error.message);
        throw error;
      }
    }
  }

  /**
   * 采集类目关键词（每周一凌晨1点执行）
   * @private
   * @param {string} nodeId - 类目节点ID
   * @param {number} domain - 站点域名代码
   * @returns {Promise<void>}
   */
  async #collectCategoryKeywords(nodeId, domain, progress) {
    console.log(`[Sorftime Scheduler] 开始采集类目 ${nodeId} 的关键词 (domain: ${domain})`);
    
    // 固定的云函数ID
    const functionId = 'KNN4L19aoi';
    
    try {
      const requestBody = {
        path: "/api/CategoryRequestKeyword",
        method: "POST",
        query: { domain },
        body: {
          Nodeid: nodeId,
          PageIndex: 1,
          PageSize: 100
        },
        functionId
      };

      const response = await fetch('http://localhost:3000/api/sorftime/forward', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(requestBody)
      });

      const result = await response.json();
      
      if (result.Code !== 0) {
        throw new Error(result.message || '关键词采集失败');
      }

      console.log(`[Sorftime Scheduler] 类目 ${nodeId} 关键词采集完成`);
    } catch (error) {
      console.error(`[Sorftime Scheduler] 类目 ${nodeId} 关键词采集失败:`, error.message);
      throw error;
    }
  }

  /**
   * 通过 SellerId 分页采集单个店铺的产品数据
   * @private
   * @param {string} shopId - 店铺 Parse objectId
   * @param {string} sellerId - 亚马逊 SellerId
   * @param {number} domain - 站点代码（1=US）
   * @param {Function} progress - 可选进度回调
   * @returns {Promise<{processed: number, pages: number}>}
   */
  async #doCollectShopProducts(shopId, sellerId, domain, progress) {
    const functionId = 'GSjAsvw9FK';
    let page = 1;
    let hasMore = true;
    let totalProcessed = 0;

    progress?.({ type: 'info', message: `SellerId: ${sellerId}，开始分页采集产品数据...` });
    console.log(`[Sorftime Scheduler] 店铺 ${shopId} SellerId: ${sellerId} 开始采集`);

    while (hasMore) {
      progress?.({ type: 'info', message: `正在采集第 ${page} 页...` });
      console.log(`[Sorftime Scheduler] 店铺 ${shopId} 采集第 ${page} 页`);

      const response = await fetch('http://localhost:3000/api/sorftime/forward', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          path: '/api/ProductQuery',
          method: 'POST',
          query: { domain },
          body: {
            Query: 1,
            QueryType: 5,
            Pattern: sellerId,
            Page: page
          },
          shop: shopId,
          functionId
        })
      });

      const result = await response.json();

      if (result.Code !== 0) {
        throw new Error(`第 ${page} 页采集失败: ${result.Message || JSON.stringify(result)}`);
      }

      const items = result.Data?.Items || result.Data?.Products || (Array.isArray(result.Data) ? result.Data : []);
      const count = Array.isArray(items) ? items.length : 0;
      totalProcessed += count;

      progress?.({ type: 'info', message: `第 ${page} 页获取 ${count} 条产品`, processed: totalProcessed });
      console.log(`[Sorftime Scheduler] 店铺 ${shopId} 第 ${page} 页获取 ${count} 条产品，累计 ${totalProcessed} 条`);

      if (count === 0 || page >= 100) {
        hasMore = false;
      } else {
        page++;
        await this.#delay(2000);
      }
    }

    return { processed: totalProcessed, pages: page };
  }

  /**
   * 采集类目市场趋势（每月1号凌晨1点执行）
   * @private
   * @param {string} nodeId - 类目节点ID
   * @param {number} domain - 站点域名代码
   * @returns {Promise<void>}
   */
  async #collectCategoryMarketTrend(nodeId, domain, progress) {
    console.log(`[Sorftime Scheduler] 开始采集类目 ${nodeId} 的市场趋势 (domain: ${domain})`);
    const functionId = 'OCOKi93GjS';
    const trendIndexes = [0, 1, 2, 3, 4, 5];

    for (const TrendIndex of trendIndexes) {
      try {
        progress?.({ type: 'info', message: `类目 ${nodeId} 趋势类型 ${TrendIndex}/5 采集中...` });
        const requestBody = {
          path: "/api/CategoryTrend",
          method: "POST",
          query: { domain },
          body: {
            NodeId: nodeId,
            TrendIndex
          },
          functionId
        };

        const response = await fetch('http://localhost:3000/api/sorftime/forward', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify(requestBody)
        });

        const result = await response.json();
        if (result.Code !== 0) {
          throw new Error(result.message || `市场趋势类型 ${TrendIndex} 采集失败`);
        }

        console.log(`[Sorftime Scheduler] 类目 ${nodeId} 市场趋势类型 ${TrendIndex} 采集完成`);
        await this.#delay(1000);
      } catch (error) {
        console.error(`[Sorftime Scheduler] 类目 ${nodeId} 市场趋势类型 ${TrendIndex} 采集失败:`, error.message);
        throw error;
      }
    }
  }

  /**
   * 执行类目市场趋势采集任务（每月1号凌晨1点）
   * @private
   * @returns {Promise<any>} 执行结果对象
   */
  async #executeCategoryMarketTrendCollection(shopId, nodeIds, progress) {
    if (this.#isRunning) {
      console.log('[Sorftime 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('[Sorftime Scheduler] ========================================');
    console.log('[Sorftime Scheduler] 开始执行类目市场趋势采集任务');
    console.log('[Sorftime Scheduler] 执行时间:', new Date().toISOString());
    console.log('[Sorftime Scheduler] ========================================');

    try {
      const categories = await this.#getLeafCategoriesByFilter(shopId, nodeIds);

      if (categories.length === 0) {
        console.log('[Sorftime Scheduler] 没有找到叶子节点类目数据');
        return {
          success: true,
          message: '没有叶子节点类目需要处理',
          successCount: 0,
          errorCount: 0,
          duration: Date.now() - startTime,
          errors: []
        };
      }

      console.log(`[Sorftime Scheduler] 发现 ${categories.length} 个叶子节点类目`);

      for (const category of categories) {
        try {
          const nodeId = category.get('nodeId');
          const categoryName = category.get('name');
          const domain = category.get('domain') || 1;

          console.log(`\n[Sorftime Scheduler] ----------------------------------------`);
          console.log(`[Sorftime Scheduler] 开始处理类目市场趋势: ${categoryName} (${nodeId}, domain: ${domain})`);
          progress?.({ type: 'info', message: `开始采集类目【${categoryName}】市场趋势` });

          await this.#collectCategoryMarketTrend(nodeId, domain, progress);
          successCount++;

          progress?.({ type: 'success', message: `类目【${categoryName}】市场趋势采集完成`, processed: successCount });
          console.log(`[Sorftime Scheduler] 类目 ${categoryName} 市场趋势采集完成`);
          await this.#delay(2000);
        } catch (error) {
          errorCount++;
          const errorMsg = `类目 ${category.get('name')} 市场趋势采集失败: ${error.message}`;
          errors.push(errorMsg);
          progress?.({ type: 'error', message: errorMsg });
          console.error(`[Sorftime Scheduler] ${errorMsg}`);
        }
      }

      const duration = Date.now() - startTime;
      await this.#logExecution({
        taskName: 'sorftime-category-trend-monthly',
        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}/${categories.length} 个类目，耗时 ${Math.round(duration / 1000)}秒`;

      console.log(`\n[Sorftime Scheduler] ========================================`);
      console.log(`[Sorftime Scheduler] ${message}`);
      console.log(`[Sorftime Scheduler] ========================================\n`);

      return {
        success: errorCount === 0,
        message,
        successCount,
        errorCount,
        duration,
        errors
      };
    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `类目市场趋势采集任务执行失败: ${error.message}`;
      console.error(`[Sorftime Scheduler] ${errorMsg}`);
      return {
        success: false,
        message: errorMsg,
        successCount,
        errorCount: errorCount + 1,
        duration,
        errors: [...errors, errorMsg]
      };
    } finally {
      this.#isRunning = false;
    }
  }

  /**
   * 执行类目热销产品采集任务（每月1号）
   * @private
   * @returns {Promise<any>} 执行结果对象，包含成功数、失败数、耗时等信息
   */
  async #executeCategoryProductsCollection(shopId, nodeIds, progress) {
    if (this.#isRunning) {
      console.log('[Sorftime 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('[Sorftime Scheduler] ========================================');
    console.log('[Sorftime Scheduler] 开始执行类目热销产品采集任务');
    console.log('[Sorftime Scheduler] 执行时间:', new Date().toISOString());
    console.log('[Sorftime Scheduler] ========================================');

    try {
      // 获取所有类目
      const categories = await this.#getLeafCategoriesByFilter(shopId, nodeIds);
      if (categories.length === 0) {
        console.log('[Sorftime Scheduler] 没有找到类目数据');
        return {
          success: true,
          message: '没有类目需要处理',
          successCount: 0,
          errorCount: 0,
          duration: Date.now() - startTime,
          errors: []
        };
      }

      console.log(`[Sorftime Scheduler] 发现 ${categories.length} 个类目`);
      // 循环处理每个类目
      for (const category of categories) {
        try {
          const nodeId = category.get('nodeId');
          const categoryName = category.get('name');
          const domain = category.get('domain') || 1; // 从类目中获取domain，默认为1（美国站）
          console.log(`\n[Sorftime Scheduler] ----------------------------------------`);
          console.log(`[Sorftime Scheduler] 开始处理类目: ${categoryName} (${nodeId}, domain: ${domain})`);
          progress?.({ type: 'info', message: `开始采集类目【${categoryName}】热销产品` });
          await this.#collectCategoryProducts(nodeId, domain, progress);
          successCount++;
          progress?.({ type: 'success', message: `类目【${categoryName}】热销产品采集完成`, processed: successCount });
          console.log(`[Sorftime Scheduler] 类目 ${categoryName} 处理完成`);
          await this.#delay(3000);
        } catch (error) {
          errorCount++;
          const errorMsg = `类目 ${category.get('name')} 处理失败: ${error.message}`;
          errors.push(errorMsg);
          progress?.({ type: 'error', message: errorMsg });
          console.error(`[Sorftime Scheduler] ${errorMsg}`);
        }
      }

      // 记录执行日志
      const duration = Date.now() - startTime;
      await this.#logExecution({
        taskName: 'sorftime-category-products-monthly',
        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}/${categories.length} 个类目，耗时 ${Math.round(duration / 1000)}秒`;
      
      console.log(`\n[Sorftime Scheduler] ========================================`);
      console.log(`[Sorftime Scheduler] ${message}`);
      console.log(`[Sorftime Scheduler] ========================================\n`);

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

    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `类目热销产品采集任务执行失败: ${error.message}`;
      console.error(`[Sorftime Scheduler] ${errorMsg}`);

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

  /**
   * 执行类目关键词采集任务（每周一）
   * @private
   * @returns {Promise<any>} 执行结果对象，包含成功数、失败数、耗时等信息
   */
  async #executeCategoryKeywordsCollection(shopId, nodeIds, progress) {
    if (this.#isRunning) {
      console.log('[Sorftime 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('[Sorftime Scheduler] ========================================');
    console.log('[Sorftime Scheduler] 开始执行类目关键词采集任务');
    console.log('[Sorftime Scheduler] 执行时间:', new Date().toISOString());
    console.log('[Sorftime Scheduler] ========================================');

    try {
      // 获取所有叶子节点类目
      const categories = await this.#getLeafCategoriesByFilter(shopId, nodeIds);

      if (categories.length === 0) {
        console.log('[Sorftime Scheduler] 没有找到叶子节点类目数据');
        return {
          success: true,
          message: '没有叶子节点类目需要处理',
          successCount: 0,
          errorCount: 0,
          duration: Date.now() - startTime,
          errors: []
        };
      }

      console.log(`[Sorftime Scheduler] 发现 ${categories.length} 个叶子节点类目`);

      // 循环处理每个类目
      for (const category of categories) {
        try {
          const nodeId = category.get('nodeId');
          const categoryName = category.get('name');
          const domain = category.get('domain') || 1; // 从类目中获取domain，默认为1（美国站）
          
          console.log(`\n[Sorftime Scheduler] ----------------------------------------`);
          console.log(`[Sorftime Scheduler] 开始处理类目: ${categoryName} (${nodeId}, domain: ${domain})`);
          progress?.({ type: 'info', message: `开始采集类目【${categoryName}】关键词` });

          await this.#collectCategoryKeywords(nodeId, domain, progress);
          successCount++;
          
          progress?.({ type: 'success', message: `类目【${categoryName}】关键词采集完成`, processed: successCount });
          console.log(`[Sorftime Scheduler] 类目 ${categoryName} 关键词采集完成`);
          await this.#delay(2000);

        } catch (error) {
          errorCount++;
          const errorMsg = `类目 ${category.get('name')} 关键词采集失败: ${error.message}`;
          errors.push(errorMsg);
          progress?.({ type: 'error', message: errorMsg });
          console.error(`[Sorftime Scheduler] ${errorMsg}`);
        }
      }

      // 记录执行日志
      const duration = Date.now() - startTime;
      await this.#logExecution({
        taskName: 'sorftime-category-keywords-weekly',
        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}/${categories.length} 个类目，耗时 ${Math.round(duration / 1000)}秒`;
      
      console.log(`\n[Sorftime Scheduler] ========================================`);
      console.log(`[Sorftime Scheduler] ${message}`);
      console.log(`[Sorftime Scheduler] ========================================\n`);

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

    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `类目关键词采集任务执行失败: ${error.message}`;
      console.error(`[Sorftime Scheduler] ${errorMsg}`);

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

  /**
   * 执行云函数调用任务（每两小时）
   * 调用指定的云函数进行数据处理
   * @private
   * @returns {Promise<any>} 执行结果对象
   */
  async #executeCloudFunctionInvocation() {
    console.log('[Sorftime Scheduler] ========================================');
    console.log('[Sorftime Scheduler] 开始执行云函数调用任务');
    console.log('[Sorftime Scheduler] 执行时间:', new Date().toISOString());
    console.log('[Sorftime Scheduler] ========================================');

    const startTime = Date.now();

    try {
      const response = await fetch('http://localhost:3000/api/functions', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json'
        },
        body: JSON.stringify({
          id: "Z4z6SB4o9e"
        })
      });

      const duration = Date.now() - startTime;

      if (!response.ok) {
        const errorMsg = `云函数调用失败，状态码: ${response.status}`;
        console.error(`[Sorftime Scheduler] ${errorMsg}`);
        return {
          success: false,
          message: errorMsg,
          duration,
          errors: [errorMsg]
        };
      }

      const result = await response.json();
      const message = `云函数调用完成，耗时 ${Math.round(duration / 1000)}秒`;

      console.log(`\n[Sorftime Scheduler] ========================================`);
      console.log(`[Sorftime Scheduler] ${message}`);
      console.log(`[Sorftime Scheduler] ========================================\n`);

      return {
        success: true,
        message,
        duration,
        data: result,
        errors: []
      };

    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `云函数调用任务执行失败: ${error.message}`;
      console.error(`[Sorftime Scheduler] ${errorMsg}`);

      return {
        success: false,
        message: errorMsg,
        duration,
        errors: [errorMsg]
      };
    }
  }

  /**
   * 执行产品评论采集任务（每天凌晨1点）
   * 根据 Product 表中的 ASIN，采集对应的评论信息
   * @private
   * @returns {Promise<any>} 执行结果对象
   */
  async #executeProductReviewsCollection(progress) {
    console.log('[Sorftime Scheduler] ========================================');
    console.log('[Sorftime Scheduler] 开始执行产品评论采集任务');
    console.log('[Sorftime Scheduler] 执行时间:', new Date().toISOString());
    console.log('[Sorftime Scheduler] ========================================');

    const startTime = Date.now();
    let successCount = 0;
    let errorCount = 0;
    const errors = [];

    progress?.({ type: 'info', message: '开始执行产品评论采集任务' });

    try {
      const Parse = globalThis.Parse;

      // 查询所有 Product 记录
      const productQuery = new Parse.Query('Product');
      productQuery.limit(10000);
      const products = await productQuery.find({ useMasterKey: true });

      if (products.length === 0) {
        console.log('[Sorftime Scheduler] 没有找到 Product 数据');
        progress?.({ type: 'info', message: '没有 Product 数据需要处理' });
        return {
          success: true,
          message: '没有 Product 数据需要处理',
          successCount: 0,
          errorCount: 0,
          duration: Date.now() - startTime,
          errors: []
        };
      }

      console.log(`[Sorftime Scheduler] 发现 ${products.length} 个 Product 记录`);
      progress?.({ type: 'info', message: `共发现 ${products.length} 个产品需要检查评论` });

      // 遍历每个 Product
      for (const product of products) {
        try {
          const asin = product.get('asin');
          if (!asin) {
            console.log('[Sorftime Scheduler] 跳过无 ASIN 的 Product');
            continue;
          }

          progress?.({ type: 'info', message: `开始处理 ASIN: ${asin}` });

          // 获取 Product 关联的 Shop 信息
          const shopRelation = product.get('shop');
          if (!shopRelation) {
            console.log(`[Sorftime Scheduler] Product 无关联 Shop，ASIN: ${asin}`);
            continue;
          }

          const shop = await shopRelation.fetch({ useMasterKey: true });
          const shopDomain = shop.get('domain') || 1;

          // 查询 SorftimeReviews 中该 ASIN 最新的 updateAt 时间
          let queryStartDt = '2025-01-01';
          const reviewQuery = new Parse.Query('SorftimeReviews');
          reviewQuery.equalTo('asin', asin);
          reviewQuery.descending('updatedAt');
          reviewQuery.limit(1);
          const latestReview = await reviewQuery.first({ useMasterKey: true });

          if (latestReview) {
            const updateAt = latestReview.get('updatedAt');
            if (updateAt) {
              // 格式化为 YYYY-MM-DD
              const date = new Date(updateAt);
              queryStartDt = date.toISOString().split('T')[0];
            }
          }

          console.log(`[Sorftime Scheduler] 开始采集评论，ASIN: ${asin}, 查询起始日期: ${queryStartDt}`);

          // 调用 Sorftime 接口获取评论信息
          const response = await fetch('http://localhost:3000/api/sorftime/forward', {
            method: 'POST',
            headers: {
              'Content-Type': 'application/json'
            },
            body: JSON.stringify({
              path: '/api/ProductReviewsQuery',
              method: 'POST',
              query: {
                domain: shopDomain,
                shopId: shop.id
              },
              body: {
                ASIN: asin,
                PageIndex: 1,
                OnlyPurchase: 1,
                Star: '1,2,3,4,5',
                Querystartdt: queryStartDt
              },
              functionId: 'gExKf7T2HD'
            })
          });

          if (!response.ok) {
            const errorMsg = `请求评论数据失败，ASIN: ${asin}, 状态码: ${response.status}`;
            errors.push(errorMsg);
            errorCount++;
            console.error(`[Sorftime Scheduler] ${errorMsg}`);
            continue;
          }

          const result = await response.json();
          if (result.success && result.data) {
            successCount++;
            console.log(`[Sorftime Scheduler] 评论数据采集成功，ASIN: ${asin}`);
            progress?.({ type: 'success', message: `评论数据采集成功，ASIN: ${asin}`, processed: successCount });
          } else {
            const errorMsg = `评论数据获取失败，ASIN: ${asin}`;
            errors.push(errorMsg);
            errorCount++;
            console.error(`[Sorftime Scheduler] ${errorMsg}`);
            progress?.({ type: 'error', message: errorMsg });
          }

          // 延迟 1 秒，避免请求过于频繁
          await this.#delay(500);

        } catch (error) {
          errorCount++;
          const errorMsg = `处理 Product 失败: ${error.message}`;
          errors.push(errorMsg);
          console.error(`[Sorftime Scheduler] ${errorMsg}`);
          progress?.({ type: 'error', message: errorMsg });
        }
      }

      const duration = Date.now() - startTime;
      const message = `产品评论采集完成: 成功 ${successCount}/${products.length} 个产品，耗时 ${Math.round(duration / 1000)}秒`;

      console.log(`\n[Sorftime Scheduler] ========================================`);
      console.log(`[Sorftime Scheduler] ${message}`);
      console.log(`[Sorftime Scheduler] ========================================\n`);

      progress?.({ type: errorCount === 0 ? 'success' : 'warning', message });

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

    } catch (error) {
      const duration = Date.now() - startTime;
      const errorMsg = `产品评论采集任务执行失败: ${error.message}`;
      console.error(`[Sorftime Scheduler] ${errorMsg}`);
      progress?.({ type: 'error', message: errorMsg });

      return {
        success: false,
        message: errorMsg,
        successCount: 0,
        errorCount: 1,
        duration,
        errors: [errorMsg]
      };
    }
  }

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

  /**
   * 启动定时任务
   * @returns {Promise<void>}
   */
  async start() {
    if (this.#categoryProductsCronTask || this.#keywordsCronTask || this.#marketTrendCronTask || this.#shopProductsCronTask || this.#productDetailCronTask || this.#cloudFunctionCronTask || this.#reviewsCronTask) {
      console.log('[Sorftime Scheduler] 定时任务已在运行中');
      return;
    }

    // 每月1号凌晨执行类目热销产品采集
    this.#categoryProductsCronTask = nodeCron.schedule('0 0 1 * *', async () => {
      await this.#executeCategoryProductsCollection(undefined, undefined, undefined);
    }, {
      timezone: "Asia/Shanghai"
    });

    console.log('[Sorftime Scheduler] 类目热销产品定时任务已启动，将在每月1号凌晨执行');

    // 每月1号凌晨1点执行类目市场趋势采集
    this.#marketTrendCronTask = nodeCron.schedule('0 1 1 * *', async () => {
      await this.#executeCategoryMarketTrendCollection(undefined, undefined, undefined);
    }, {
      timezone: "Asia/Shanghai"
    });

    console.log('[Sorftime Scheduler] 类目市场趋势定时任务已启动，将在每月1号凌晨1点执行');

    // 每周一凌晨1点执行类目关键词采集
    this.#keywordsCronTask = nodeCron.schedule('0 1 * * 1', async () => {
      await this.#executeCategoryKeywordsCollection(undefined, undefined, undefined);
    }, {
      timezone: "Asia/Shanghai"
    });

    console.log('[Sorftime Scheduler] 类目关键词定时任务已启动，将在每周一凌晨1点执行');

    // 每天凌晨1点采集产品评论数据
    this.#reviewsCronTask = nodeCron.schedule('0 1 * * *', async () => {
      console.log('[Sorftime Scheduler] ===== 凌晨1点：开始产品评论采集 =====');
      try {
        await this.#executeProductReviewsCollection(undefined);
      } catch (e) {
        console.error('[Sorftime Scheduler] 产品评论采集失败:', e.message);
      }
      console.log('[Sorftime Scheduler] ===== 产品评论采集完成 =====');
    }, { timezone: 'Asia/Shanghai' });

    console.log('[Sorftime Scheduler] 产品评论采集定时任务已启动，将在每天凌晨1点执行');

    // 每天凌晨2点采集店铺产品数据
    this.#shopProductsCronTask = nodeCron.schedule('0 2 * * *', async () => {
      console.log('[Sorftime Scheduler] ===== 凌晨2点：开始店铺产品采集 =====');
      try {
        await this.collectShopProducts(undefined, undefined);
      } catch (e) {
        console.error('[Sorftime Scheduler] 店铺产品采集失败:', e.message);
      }
      console.log('[Sorftime Scheduler] ===== 店铺产品采集完成 =====');
    }, { timezone: 'Asia/Shanghai' });

    console.log('[Sorftime Scheduler] 店铺产品采集定时任务已启动，将在每天凌晨2点执行');

    // 每两小时执行一次云函数调用
    this.#cloudFunctionCronTask = nodeCron.schedule('0 */2 * * *', async () => {
      console.log('[Sorftime Scheduler] ===== 每两小时：开始云函数调用 =====');
      try {
        await this.#executeCloudFunctionInvocation();
      } catch (e) {
        console.error('[Sorftime Scheduler] 云函数调用失败:', e.message);
      }
      console.log('[Sorftime Scheduler] ===== 云函数调用完成 =====');
    }, { timezone: 'Asia/Shanghai' });

    console.log('[Sorftime Scheduler] 云函数调用定时任务已启动，将每两小时执行一次');
  }



  /**
   * 停止定时任务
   * @returns {void}
   */
  stop() {
    if (this.#categoryProductsCronTask) {
      this.#categoryProductsCronTask.stop();
      this.#categoryProductsCronTask = null;
      console.log('[Sorftime Scheduler] 类目热销产品定时任务已停止');
    }
    if (this.#marketTrendCronTask) {
      this.#marketTrendCronTask.stop();
      this.#marketTrendCronTask = null;
      console.log('[Sorftime Scheduler] 类目市场趋势定时任务已停止');
    }
    if (this.#keywordsCronTask) {
      this.#keywordsCronTask.stop();
      this.#keywordsCronTask = null;
      console.log('[Sorftime Scheduler] 类目关键词定时任务已停止');
    }
    if (this.#shopProductsCronTask) {
      this.#shopProductsCronTask.stop();
      this.#shopProductsCronTask = null;
      console.log('[Sorftime Scheduler] 店铺产品采集定时任务已停止');
    }
    if (this.#productDetailCronTask) {
      this.#productDetailCronTask.stop();
      this.#productDetailCronTask = null;
      console.log('[Sorftime Scheduler] 产品详情同步定时任务已停止');
    }
    if (this.#cloudFunctionCronTask) {
      this.#cloudFunctionCronTask.stop();
      this.#cloudFunctionCronTask = null;
      console.log('[Sorftime Scheduler] 云函数调用定时任务已停止');
    }
    if (this.#reviewsCronTask) {
      this.#reviewsCronTask.stop();
      this.#reviewsCronTask = null;
      console.log('[Sorftime Scheduler] 产品评论采集定时任务已停止');
    }
  }

  /**
   * 获取调度器状态
   * @returns {object} 调度器状态对象
   */
  getStatus() {
    return {
      isRunning: this.#isRunning,
      lastRunTime: this.#lastRunTime,
      cronEnabled: this.#cronEnabled,
      tasks: {
        categoryProducts: {
          enabled: !!this.#categoryProductsCronTask,
          schedule: '每月1号凌晨'
        },
        marketTrend: {
          enabled: !!this.#marketTrendCronTask,
          schedule: '每月1号凌晨1点'
        },
        keywords: {
          enabled: !!this.#keywordsCronTask,
          schedule: '每周一凌晨1点'
        },
        shopProducts: {
          enabled: !!this.#shopProductsCronTask,
          schedule: '每天凌晨2点'
        },
        productDetails: {
          enabled: !!this.#productDetailCronTask,
          schedule: '每小时'
        },
        cloudFunction: {
          enabled: !!this.#cloudFunctionCronTask,
          schedule: '每两小时'
        },
        productReviews: {
          enabled: !!this.#reviewsCronTask,
          schedule: '每天凌晨1点'
        }
      }
    };
  }

  /**
   * 手动触发类目热销产品采集
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerCategoryProducts(shopId, nodeIds, progress) {
    console.log('[Sorftime Scheduler] 手动触发类目热销产品采集任务');
    return await this.#executeCategoryProductsCollection(shopId, nodeIds, progress);
  }

  /**
   * 手动触发类目关键词采集
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerCategoryKeywords(shopId, nodeIds, progress) {
    console.log('[Sorftime Scheduler] 手动触发类目关键词采集任务');
    return await this.#executeCategoryKeywordsCollection(shopId, nodeIds, progress);
  }

  /**
   * 手动触发类目市场趋势采集
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerCategoryTrends(shopId, nodeIds, progress) {
    console.log('[Sorftime Scheduler] 手动触发类目市场趋势采集任务');
    return await this.#executeCategoryMarketTrendCollection(shopId, nodeIds, progress);
  }

  /**
   * 手动触发产品评论采集
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerProductReviews(progress) {
    console.log('[Sorftime Scheduler] 手动触发产品评论采集任务');
    return await this.#executeProductReviewsCollection(progress);
  }

  /**
   * 按需触发单个类目的全量数据采集（热销产品 + 关键词 + 市场趋势）
   * 前端选中某个类目但数据库无数据时调用
   * @param {string} nodeId - 类目节点ID
   * @param {number} domain - 站点域名代码（默认1=美国）
   * @returns {Promise<any>} 执行结果对象
   */
  async triggerSingleCategory(nodeId, domain = 1, progress) {
    const startTime = Date.now();
    const errors = [];
    let tasks = { products: false, keywords: false, trends: false };

    console.log(`[Sorftime Scheduler] 按需触发单类目采集: nodeId=${nodeId}, domain=${domain}`);

    try {
      // 1. 热销产品
      try {
        progress?.({ type: 'info', message: `开始采集类目 ${nodeId} 热销产品` });
        await this.#collectCategoryProducts(nodeId, domain, progress);
        tasks.products = true;
        progress?.({ type: 'success', message: `类目 ${nodeId} 热销产品采集完成` });
        console.log(`[Sorftime Scheduler] 单类目 ${nodeId} 热销产品采集完成`);
      } catch (error) {
        const errMsg = error?.message || String(error);
        progress?.({ type: 'error', message: `热销产品采集失败: ${errMsg}` });
        errors.push(`热销产品采集失败: ${errMsg}`);
        console.error(`[Sorftime Scheduler] 热销产品采集失败:`, error);
      }

      // 延迟 2 秒
      await new Promise(resolve => setTimeout(resolve, 500));

      // 2. 关键词
      try {
        progress?.({ type: 'info', message: `开始采集类目 ${nodeId} 关键词` });
        await this.#collectCategoryKeywords(nodeId, domain, progress);
        tasks.keywords = true;
        progress?.({ type: 'success', message: `类目 ${nodeId} 关键词采集完成` });
        console.log(`[Sorftime Scheduler] 单类目 ${nodeId} 关键词采集完成`);
      } catch (error) {
        const errMsg = error?.message || String(error);
        progress?.({ type: 'error', message: `关键词采集失败: ${errMsg}` });
        errors.push(`关键词采集失败: ${errMsg}`);
        console.error(`[Sorftime Scheduler] 关键词采集失败:`, error);
      }

      // 延迟 2 秒
      await new Promise(resolve => setTimeout(resolve, 500));

      // 3. 市场趋势
      try {
        progress?.({ type: 'info', message: `开始采集类目 ${nodeId} 市场趋势` });
        await this.#collectCategoryMarketTrend(nodeId, domain, progress);
        tasks.trends = true;
        progress?.({ type: 'success', message: `类目 ${nodeId} 市场趋势采集完成` });
        console.log(`[Sorftime Scheduler] 单类目 ${nodeId} 市场趋势采集完成`);
      } catch (error) {
        const errMsg = error?.message || String(error);
        progress?.({ type: 'error', message: `市场趋势采集失败: ${errMsg}` });
        errors.push(`市场趋势采集失败: ${errMsg}`);
        console.error(`[Sorftime Scheduler] 市场趋势采集失败:`, error);
      }

      const duration = Date.now() - startTime;
      const allSuccess = errors.length === 0;
      const successCount = Object.values(tasks).filter(Boolean).length;

      // 异步记录执行日志，不阻塞返回
      if (this.#logExecution) {
        this.#logExecution({
          taskName: `trigger-single-category-${nodeId}`,
          startTime: new Date(startTime),
          endTime: new Date(),
          duration,
          successCount,
          errorCount: errors.length,
          errors,
          status: allSuccess ? 'success' : 'partial_success'
        }).catch(err => console.error('[Sorftime Scheduler] Log execution error:', err));
      }

      const result = {
        success: allSuccess,
        message: allSuccess
          ? `类目 ${nodeId} 全量采集完成，耗时 ${Math.round(duration / 1000)}秒`
          : `类目 ${nodeId} 部分采集失败（成功 ${successCount}/3 个任务）`,
        nodeId,
        domain,
        tasks,
        duration,
        errors: errors.length > 0 ? errors : undefined
      };

      console.log(`[Sorftime Scheduler] 采集结果:`, result);
      return result;
    } catch (err) {
      const duration = Date.now() - startTime;
      const errMsg = err?.message || String(err);
      console.error(`[Sorftime Scheduler] 采集异常:`, err);
      return {
        success: false,
        message: `类目 ${nodeId} 采集异常: ${errMsg}`,
        nodeId,
        domain,
        tasks,
        duration,
        errors: [errMsg]
      };
    }
  }


  /**
   * 采集所有活跃 Amazon 店铺（或指定店铺）的产品数据
   * 通过 SellerId (QueryType=5) 分页请求 /api/ProductQuery
   * @param {string} shopId - 可选，不传则采集全部活跃 Amazon 店铺
   * @param {Function} progress - 可选进度回调，适合 SSE 实时推送
   * @returns {Promise<any>} 执行结果
   */
  async collectShopProducts(shopId, progress) {
    const shops = await this.#getActiveAmazonShops(shopId);

    if (shops.length === 0) {
      const msg = shopId ? `未找到店铺 ${shopId} 或店铺未激活` : '没有找到活跃的 Amazon 店铺';
      progress?.({ type: 'warn', message: msg });
      console.warn(`[Sorftime Scheduler] ${msg}`);
      return { success: true, message: msg, processed: 0, errorCount: 0, errors: [] };
    }

    let totalProcessed = 0;
    let errorCount = 0;
    const errors = [];

    for (const shop of shops) {
      const sid = shop.id;
      const shopName = shop.get('name');
      const config = shop.get('config');
      const domain = 1;

      if (!config || !config.SpApiConfig || !config.SpApiConfig.sellerID) {
        const msg = `店铺【${shopName}】(${sid}) 缺少 SpApiConfig.sellerID 配置，跳过`;
        progress?.({ type: 'error', message: msg });
        console.error(`[Sorftime Scheduler] ${msg}`);
        errors.push(msg);
        errorCount++;
        continue;
      }

      const sellerId = config.SpApiConfig.sellerID;
      progress?.({ type: 'info', message: `开始采集店铺【${shopName}】的产品数据 (SellerId: ${sellerId})` });
      console.log(`[Sorftime Scheduler] 开始采集店铺 ${shopName} (${sid}) 产品数据`);

      try {
        const result = await this.#doCollectShopProducts(sid, sellerId, domain, progress);
        progress?.({ type: 'success', message: `店铺【${shopName}】产品采集完成: 共 ${result.pages} 页 ${result.processed} 条`, processed: result.processed });
        console.log(`[Sorftime Scheduler] 店铺 ${shopName} 产品采集完成: ${result.processed} 条`);
        totalProcessed += result.processed;
        await this.#delay(2000);
      } catch (error) {
        const msg = `店铺【${shopName}】产品采集失败: ${error.message}`;
        progress?.({ type: 'error', message: msg });
        console.error(`[Sorftime Scheduler] ${msg}`);
        errors.push(msg);
        errorCount++;
      }
    }

    const message = `店铺产品采集完成: 成功 ${shops.length - errorCount}/${shops.length} 个店铺，共 ${totalProcessed} 条记录`;
    return { success: errorCount === 0, message, processed: totalProcessed, errorCount, errors };
  }
}

// 导出单例
export const sorftimeScheduler = new SorftimeScheduler();