/** * 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} 延迟 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} 活跃店铺列表 */ 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('[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} */ 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} */ 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} */ 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} */ 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} */ 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} */ 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} 执行结果对象,包含成功数、失败数、耗时等信息 */ 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} */ 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} 执行结果对象 */ async triggerManually(shopId) { console.log('[SP-API Scheduler] 手动触发数据采集任务'); return await this.#executeDataCollection(shopId); // 此时#executeDataCollection已声明 } } // 导出单例 export const spApiScheduler = new SpApiScheduler();