| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670 |
- /**
- * Amazon SP-API 数据采集调度器(新版)
- *
- * 相较旧版 sp-api-schedule.ts,本版将四个采集步骤拆分为独立的公共方法:
- * - collectListings 采集 Listing 数据
- * - collectOrders 采集订单数据
- * - collectReports 采集报表数据
- * - parseReports 解析/检查未完成报表
- *
- * 每个方法均接受可选的 ProgressCallback,适合通过 SSE 向前端推送实时进度。
- * 定时任务(每天凌晨 1 点)调用上述方法时不传 callback,仅打印日志。
- *
- * 使用方式:
- * import { newSpApiScheduler } from './new-sp-api-schedule.ts';
- * // 仅手动触发,不启动 cron(由旧版 spApiScheduler 维护 cron)
- * await newSpApiScheduler.collectListings(shopId, progressCallback);
- */
- import nodeCron from 'npm:node-cron';
- // ===== 调度器类 =====
- class NewSpApiScheduler {
- #cronTask = null;
- #reportCronTask = null;
- #lastRunTime = null;
- // ========== 私有工具 ==========
- #delay(ms) {
- return new Promise(resolve => setTimeout(resolve, ms));
- }
- #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 店铺列表
- * @param shopId 若传入则只返回该店铺
- */
- 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();
- }
- // ========== 单店铺内部实现 ==========
- /**
- * 执行单个店铺的 Listing 采集
- * 通过 /api/amazon/forward 调用 SP-API,内部 cleanListings 处理全部分页
- */
- async #doCollectListings(shopId, config, marketplaceId, progress) {
- 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();
- progress?.({ type: 'info', message: `增量采集,使用最新 lastUpdatedDate: ${lastUpdatedAfter}` });
- } else {
- const oneYearAgo = new Date(Date.now() - 365 * 24 * 60 * 60 * 1000);
- lastUpdatedAfter = oneYearAgo.toISOString();
- progress?.({ type: 'info', message: `无历史数据,从一年前开始全量采集: ${lastUpdatedAfter}` });
- }
- } catch (error) {
- const oneYearAgo = new Date(Date.now() - 365 * 24 * 60 * 60 * 1000);
- lastUpdatedAfter = oneYearAgo.toISOString();
- progress?.({ type: 'warn', message: `查询 Listing 历史失败,默认使用一年前: ${lastUpdatedAfter}` });
- }
- const path = `/listings/2021-08-01/items/${sellerId}?marketplaceIds=${marketplaceId}&includedData=summaries,attributes,issues&withStatus=BUYABLE,DISCOVERABLE&sortBy=lastUpdatedDate&sortOrder=ASC&pageSize=10&lastUpdatedAfter=${lastUpdatedAfter}`;
- progress?.({ type: 'info', message: '正在调用亚马逊 Listing API(cleanListings 内部处理分页,请耐心等待)...' });
- console.log(`[New SP-API Scheduler] 店铺 ${shopId} 请求 Listing API,path: ${path}`);
- const response = await fetch('http://localhost:3000/api/amazon/forward', {
- method: 'POST',
- headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
- body: JSON.stringify({ path, method: 'GET', functionName: 'cleanListings' })
- });
- const result = await response.json();
- if (!result.success) {
- throw new Error(result.message || 'Listing 采集失败');
- }
- const processed = result.data?.processed || 0;
- console.log(`[New SP-API Scheduler] 店铺 ${shopId} Listing 采集完成: ${processed} 条记录`);
- return { processed };
- }
- /**
- * 执行单个店铺的订单采集(增量)
- * 通过 /api/amazon/forward 调用 SP-API,内部 cleanOrders 处理全部分页
- */
- async #doCollectOrders(shopId, config, marketplaceId, progress) {
- 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();
- progress?.({ type: 'info', message: `增量采集,使用最新 orderDate: ${createdAfter}` });
- } else {
- const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
- createdAfter = ninetyDaysAgo.toISOString();
- progress?.({ type: 'info', message: `无历史数据,从90天前开始采集: ${createdAfter}` });
- }
- } catch (error) {
- const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
- createdAfter = ninetyDaysAgo.toISOString();
- progress?.({ type: 'warn', message: `查询 Order 历史失败,默认使用90天前: ${createdAfter}` });
- }
- progress?.({ type: 'info', message: '正在调用亚马逊 Orders API(cleanOrders 内部处理分页,请耐心等待)...' });
- console.log(`[New SP-API Scheduler] 店铺 ${shopId} 请求 Orders API,createdAfter: ${createdAfter}`);
- 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 || '订单采集失败');
- }
- const processed = result.data?.processed || 0;
- console.log(`[New SP-API Scheduler] 店铺 ${shopId} 订单采集完成: ${processed} 条记录`);
- return { processed };
- }
- /**
- * 执行单个店铺的报表创建与保存
- */
- async #doCollectReports(shopId, config, marketplaceId, progress) {
- const Parse = globalThis.Parse;
- const reportTypes = ['GET_FBA_FULFILLMENT_CUSTOMER_RETURNS_DATA'];
- for (const reportType of reportTypes) {
- try {
- // 1. 查询上次 dataEndTime 作为本次起点
- let dataStartTime;
- try {
- const q = new Parse.Query('Reports');
- q.equalTo('shop', shopId);
- q.equalTo('reportType', reportType);
- q.descending('dataEndTime');
- q.limit(1);
- const last = await q.first({ useMasterKey: true });
- dataStartTime = last?.get('dataEndTime')
- ? last.get('dataEndTime').toISOString()
- : 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();
- progress?.({ type: 'info', message: `创建报表 ${reportType},时间范围: ${dataStartTime} ~ ${dataEndTime}` });
- console.log(`[New 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) {
- progress?.({ type: 'error', message: `创建报表失败 ${reportType}: ${JSON.stringify(createResult)}` });
- console.error(`[New SP-API Scheduler] 创建报表失败 ${reportType}:`, JSON.stringify(createResult));
- continue;
- }
- progress?.({ type: 'info', message: `报表已创建 reportId: ${reportId},等待查询状态...` });
- console.log(`[New 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 });
- progress?.({ type: 'success', message: `报表 ${reportType} 已保存,processingStatus: ${reportInfo.processingStatus || 'IN_QUEUE'}` });
- console.log(`[New SP-API Scheduler] 报表已保存, status: ${reportInfo.processingStatus}`);
- await this.#delay(2000);
- } catch (error) {
- progress?.({ type: 'error', message: `报表 ${reportType} 处理失败: ${error.message}` });
- console.error(`[New SP-API Scheduler] 报表 ${reportType} 处理失败: ${error.message}`);
- }
- }
- }
- /**
- * 检查并更新 processingStatus 不为 DONE 的报表
- */
- async #doCheckPendingReports(shopId, progress) {
- const Parse = globalThis.Parse;
- 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) {
- progress?.({ type: 'info', message: '没有状态未完成的报表,跳过状态更新' });
- console.log('[New SP-API Scheduler] 没有未完成的报表');
- return;
- }
- progress?.({ type: 'info', message: `发现 ${pendingReports.length} 个未完成报表,逐一查询最新状态...` });
- console.log(`[New SP-API Scheduler] 发现 ${pendingReports.length} 个未完成报表`);
- for (const report of pendingReports) {
- try {
- const reportId = report.get('reportId');
- const sid = report.get('shop').id;
- if (!reportId || !sid) continue;
- const getRes = await fetch('http://localhost:3000/api/amazon/forward', {
- method: 'POST',
- headers: { 'Content-Type': 'application/json', 'shop-objectid': sid },
- body: JSON.stringify({ path: `/reports/2021-06-30/reports/${reportId}`, method: 'GET' })
- });
- const reportData = await getRes.json();
- const reportInfo = reportData?.data || {};
- 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 });
- progress?.({ type: 'info', message: `报表 ${reportId} 状态更新为: ${newStatus}` });
- console.log(`[New SP-API Scheduler] 报表 ${reportId} 状态更新为: ${newStatus}`);
- await this.#delay(10000);
- } catch (error) {
- progress?.({ type: 'error', message: `更新报表状态失败: ${error.message}` });
- console.error(`[New SP-API Scheduler] 更新报表状态失败: ${error.message}`);
- }
- }
- }
- /**
- * 解析 isParsed=false & processingStatus=DONE 的报表
- */
- async #doProcessUnparsedReports(shopId, progress) {
- 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) {
- progress?.({ type: 'info', message: '没有待解析的报表(状态 DONE 且未解析)' });
- console.log(`[New SP-API Scheduler] 店铺 ${shopId} 没有待解析的报表`);
- return { processed: 0 };
- }
- progress?.({ type: 'info', message: `发现 ${unparsedReports.length} 个待解析报表` });
- console.log(`[New SP-API Scheduler] 发现 ${unparsedReports.length} 个待解析报表`);
- let totalProcessed = 0;
- for (const report of unparsedReports) {
- try {
- const reportDocumentId = report.get('reportDocumentId');
- const reportType = report.get('reportType');
- if (!reportDocumentId) {
- progress?.({ type: 'warn', message: `报表 ${report.id} 缺少 reportDocumentId,跳过` });
- continue;
- }
- progress?.({ type: 'info', message: `正在解析报表 ${reportDocumentId} (${reportType})` });
- console.log(`[New SP-API Scheduler] 解析报表 ${reportDocumentId} (${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) {
- const parsed = result.data?.processResult?.processed || 0;
- report.set('isParsed', true);
- report.set('parsedAt', new Date());
- report.set('parsedRecords', parsed);
- await report.save(null, { useMasterKey: true });
- totalProcessed += parsed;
- progress?.({ type: 'success', message: `报表 ${reportDocumentId} 解析完成: ${parsed} 条记录`, processed: parsed });
- console.log(`[New SP-API Scheduler] 报表 ${reportDocumentId} 解析完成: ${parsed} 条记录`);
- } else {
- progress?.({ type: 'warn', message: `报表 ${reportDocumentId} 解析未成功: ${result.message}` });
- console.warn(`[New SP-API Scheduler] 报表 ${reportDocumentId} 解析未成功: ${result.message}`);
- }
- await this.#delay(10000);
- } catch (error) {
- progress?.({ type: 'error', message: `报表解析异常: ${error.message}` });
- console.error(`[New SP-API Scheduler] 报表解析异常: ${error.message}`);
- }
- }
- return { processed: totalProcessed };
- }
- // ========== 构造函数 ==========
- constructor() {
- console.log('[New SP-API Scheduler] 初始化 Amazon SP-API 数据采集调度器(新版)');
- }
- // ========== 公共采集方法 ==========
- /**
- * 采集所有活跃店铺(或指定店铺)的 Listing 数据
- * @param shopId 可选,不传则采集全部活跃店铺
- * @param progress 可选进度回调,适合 SSE 实时推送
- */
- async collectListings(shopId, progress) {
- const shops = await this.#getActiveAmazonShops(shopId);
- if (shops.length === 0) {
- const msg = shopId ? `未找到店铺 ${shopId} 或店铺未激活` : '没有找到活跃的 Amazon 店铺';
- progress?.({ type: 'warn', message: msg });
- console.warn(`[New SP-API 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 marketplaceId = shop.get('marketplaceId') || 'ATVPDKIKX0DER';
- if (!config || !config.SpApiConfig) {
- const msg = `店铺【${shopName}】(${sid}) 缺少 SpApiConfig 配置,跳过`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- continue;
- }
- progress?.({ type: 'info', message: `开始采集店铺【${shopName}】的 Listing 数据` });
- console.log(`[New SP-API Scheduler] 开始采集店铺 ${shopName} (${sid}) 的 Listing 数据`);
- try {
- const result = await this.#doCollectListings(sid, config, marketplaceId, progress);
- progress?.({ type: 'success', message: `店铺【${shopName}】Listing 采集完成`, processed: result.processed });
- totalProcessed += result.processed;
- await this.#delay(2000);
- } catch (error) {
- const msg = `店铺【${shopName}】Listing 采集失败: ${error.message}`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- }
- }
- const message = `Listing 采集完成: 成功 ${shops.length - errorCount}/${shops.length} 个店铺,共 ${totalProcessed} 条记录`;
- return { success: errorCount === 0, message, processed: totalProcessed, errorCount, errors };
- }
- /**
- * 采集所有活跃店铺(或指定店铺)的订单数据(增量)
- * @param shopId 可选
- * @param progress 可选进度回调
- */
- async collectOrders(shopId, progress) {
- const shops = await this.#getActiveAmazonShops(shopId);
- if (shops.length === 0) {
- const msg = shopId ? `未找到店铺 ${shopId} 或店铺未激活` : '没有找到活跃的 Amazon 店铺';
- progress?.({ type: 'warn', message: msg });
- console.warn(`[New SP-API 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 marketplaceId = shop.get('marketplaceId') || 'ATVPDKIKX0DER';
- if (!config || !config.SpApiConfig) {
- const msg = `店铺【${shopName}】(${sid}) 缺少 SpApiConfig 配置,跳过`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- continue;
- }
- progress?.({ type: 'info', message: `开始采集店铺【${shopName}】的订单数据` });
- console.log(`[New SP-API Scheduler] 开始采集店铺 ${shopName} (${sid}) 的订单数据`);
- try {
- const result = await this.#doCollectOrders(sid, config, marketplaceId, progress);
- progress?.({ type: 'success', message: `店铺【${shopName}】订单采集完成`, processed: result.processed });
- totalProcessed += result.processed;
- await this.#delay(2000);
- } catch (error) {
- const msg = `店铺【${shopName}】订单采集失败: ${error.message}`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- }
- }
- const message = `订单采集完成: 成功 ${shops.length - errorCount}/${shops.length} 个店铺,共 ${totalProcessed} 条记录`;
- return { success: errorCount === 0, message, processed: totalProcessed, errorCount, errors };
- }
- /**
- * 为所有活跃店铺(或指定店铺)创建并保存报表
- * @param shopId 可选
- * @param progress 可选进度回调
- */
- async collectReports(shopId, progress) {
- const shops = await this.#getActiveAmazonShops(shopId);
- if (shops.length === 0) {
- const msg = shopId ? `未找到店铺 ${shopId} 或店铺未激活` : '没有找到活跃的 Amazon 店铺';
- progress?.({ type: 'warn', message: msg });
- console.warn(`[New SP-API Scheduler] ${msg}`);
- return { success: true, message: msg, processed: 0, errorCount: 0, errors: [] };
- }
- let successCount = 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 marketplaceId = shop.get('marketplaceId') || 'ATVPDKIKX0DER';
- if (!config || !config.SpApiConfig) {
- const msg = `店铺【${shopName}】(${sid}) 缺少 SpApiConfig 配置,跳过`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- continue;
- }
- progress?.({ type: 'info', message: `开始采集店铺【${shopName}】的报表数据` });
- console.log(`[New SP-API Scheduler] 开始采集店铺 ${shopName} (${sid}) 的报表数据`);
- try {
- await this.#doCollectReports(sid, config, marketplaceId, progress);
- progress?.({ type: 'success', message: `店铺【${shopName}】报表采集完成` });
- successCount++;
- await this.#delay(2000);
- } catch (error) {
- const msg = `店铺【${shopName}】报表采集失败: ${error.message}`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- }
- }
- const message = `报表采集完成: 成功 ${successCount}/${shops.length} 个店铺`;
- return { success: errorCount === 0, message, processed: successCount, errorCount, errors };
- }
- /**
- * 检查未完成报表状态并解析已完成的报表
- * @param shopId 可选
- * @param progress 可选进度回调
- */
- async parseReports(shopId, progress) {
- const shops = await this.#getActiveAmazonShops(shopId);
- if (shops.length === 0) {
- const msg = shopId ? `未找到店铺 ${shopId} 或店铺未激活` : '没有找到活跃的 Amazon 店铺';
- progress?.({ type: 'warn', message: msg });
- console.warn(`[New SP-API Scheduler] ${msg}`);
- return { success: true, message: msg, processed: 0, errorCount: 0, errors: [] };
- }
- // 先全局检查并更新 pending 状态
- progress?.({ type: 'info', message: '检查所有未完成报表的最新状态...' });
- try {
- await this.#doCheckPendingReports(shopId, progress);
- } catch (error) {
- progress?.({ type: 'warn', message: `检查未完成报表状态时出错: ${error.message}` });
- console.warn(`[New SP-API Scheduler] 检查未完成报表状态出错: ${error.message}`);
- }
- let totalProcessed = 0;
- let errorCount = 0;
- const errors = [];
- for (const shop of shops) {
- const sid = shop.id;
- const shopName = shop.get('name');
- progress?.({ type: 'info', message: `开始解析店铺【${shopName}】的已完成报表` });
- console.log(`[New SP-API Scheduler] 开始解析店铺 ${shopName} (${sid}) 的已完成报表`);
- try {
- const result = await this.#doProcessUnparsedReports(sid, progress);
- progress?.({ type: 'success', message: `店铺【${shopName}】报表解析完成: ${result.processed} 条记录`, processed: result.processed });
- totalProcessed += result.processed;
- await this.#delay(2000);
- } catch (error) {
- const msg = `店铺【${shopName}】报表解析失败: ${error.message}`;
- progress?.({ type: 'error', message: msg });
- console.error(`[New SP-API Scheduler] ${msg}`);
- errors.push(msg);
- errorCount++;
- }
- }
- const message = `报表解析完成: 成功 ${shops.length - errorCount}/${shops.length} 个店铺,共解析 ${totalProcessed} 条记录`;
- return { success: errorCount === 0, message, processed: totalProcessed, errorCount, errors };
- }
- // ========== 定时任务管理 ==========
- /**
- * 启动定时任务
- * - 凌晨 1 点:采集 Listing / Order / Report
- * - 凌晨 2 点:解析已完成报表
- */
- async start() {
- if (this.#cronTask) {
- console.log('[New SP-API Scheduler] 定时任务已在运行中,跳过重复启动');
- return;
- }
- this.#cronTask = nodeCron.schedule('0 1 * * *', async () => {
- console.log('[New SP-API Scheduler] ===== 凌晨1点:开始每日数据采集 =====');
- this.#lastRunTime = new Date();
- try {
- await this.collectListings(undefined, undefined);
- await this.collectOrders(undefined, undefined);
- await this.collectReports(undefined, undefined);
- } catch (e) {
- console.error('[New SP-API Scheduler] 每日采集任务失败:', e.message);
- }
- console.log('[New SP-API Scheduler] ===== 每日数据采集完成 =====');
- }, { timezone: 'Asia/Shanghai' });
- this.#reportCronTask = nodeCron.schedule('0 2 * * *', async () => {
- console.log('[New SP-API Scheduler] ===== 凌晨2点:开始报表解析 =====');
- try {
- await this.parseReports(undefined, undefined);
- } catch (e) {
- console.error('[New SP-API Scheduler] 报表解析任务失败:', e.message);
- }
- console.log('[New SP-API Scheduler] ===== 报表解析完成 =====');
- }, { timezone: 'Asia/Shanghai' });
- console.log('[New SP-API Scheduler] 定时任务已启动(凌晨1点采集,凌晨2点解析)');
- }
- /**
- * 停止定时任务
- */
- stop() {
- if (this.#cronTask) {
- this.#cronTask.stop();
- this.#cronTask = null;
- console.log('[New SP-API Scheduler] 采集定时任务已停止');
- }
- if (this.#reportCronTask) {
- this.#reportCronTask.stop();
- this.#reportCronTask = null;
- console.log('[New SP-API Scheduler] 报表解析定时任务已停止');
- }
- }
- /**
- * 获取调度器状态
- */
- getStatus() {
- return {
- isRunning: !!(this.#cronTask || this.#reportCronTask),
- lastRunTime: this.#lastRunTime,
- nextRunTime: this.#calculateNextRunTime()
- };
- }
- }
- // 导出单例
- export const newSpApiScheduler = new NewSpApiScheduler();
|