/** * 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} 延迟 Promise */ #delay(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } /** * 获取所有叶子节点类目 * @private * @returns {Promise} 叶子节点类目列表 */ 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} 类目列表 */ 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} 店铺列表 */ 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} */ 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} */ 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} */ 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} */ 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} 执行结果对象 */ 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} 执行结果对象,包含成功数、失败数、耗时等信息 */ 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} 执行结果对象,包含成功数、失败数、耗时等信息 */ 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} 执行结果对象 */ 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} 执行结果对象 */ 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} */ 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} 执行结果对象 */ async triggerCategoryProducts(shopId, nodeIds, progress) { console.log('[Sorftime Scheduler] 手动触发类目热销产品采集任务'); return await this.#executeCategoryProductsCollection(shopId, nodeIds, progress); } /** * 手动触发类目关键词采集 * @returns {Promise} 执行结果对象 */ async triggerCategoryKeywords(shopId, nodeIds, progress) { console.log('[Sorftime Scheduler] 手动触发类目关键词采集任务'); return await this.#executeCategoryKeywordsCollection(shopId, nodeIds, progress); } /** * 手动触发类目市场趋势采集 * @returns {Promise} 执行结果对象 */ async triggerCategoryTrends(shopId, nodeIds, progress) { console.log('[Sorftime Scheduler] 手动触发类目市场趋势采集任务'); return await this.#executeCategoryMarketTrendCollection(shopId, nodeIds, progress); } /** * 手动触发产品评论采集 * @returns {Promise} 执行结果对象 */ async triggerProductReviews(progress) { console.log('[Sorftime Scheduler] 手动触发产品评论采集任务'); return await this.#executeProductReviewsCollection(progress); } /** * 按需触发单个类目的全量数据采集(热销产品 + 关键词 + 市场趋势) * 前端选中某个类目但数据库无数据时调用 * @param {string} nodeId - 类目节点ID * @param {number} domain - 站点域名代码(默认1=美国) * @returns {Promise} 执行结果对象 */ 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} 执行结果 */ 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();