| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175 |
- const parseUrl = process.env.PARSE_URL || 'http://127.0.0.1:3000/parse';
- const apiUrl = process.env.API_URL || 'http://127.0.0.1:3000/api/amazon';
- const appId = process.env.PARSE_APP_ID;
- const masterKey = process.env.PARSE_MASTER_KEY;
- const rounds = Number(process.env.REPORT_POLL_ROUNDS || 12);
- const roundDelayMs = Number(process.env.REPORT_POLL_DELAY_MS || 30000);
- if (!appId || !masterKey) throw new Error('Missing PARSE_APP_ID or PARSE_MASTER_KEY');
- const headers = {
- 'Content-Type': 'application/json',
- 'X-Parse-Application-Id': appId,
- 'X-Parse-Master-Key': masterKey,
- };
- const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
- function log(event, data = {}) {
- console.log(JSON.stringify({ at: new Date().toISOString(), event, ...data }));
- }
- async function parseRequest(path, method = 'GET', body) {
- const response = await fetch(`${parseUrl}${path}`, {
- method,
- headers,
- body: body === undefined ? undefined : JSON.stringify(body),
- });
- const text = await response.text();
- let result = {};
- try { result = JSON.parse(text); } catch {}
- if (!response.ok || result.error) {
- throw new Error(`${method} ${path}: ${result.error?.message || result.error || text.slice(0, 300) || response.status}`);
- }
- return result;
- }
- async function amazonRequest(shopId, path, body) {
- for (let attempt = 1; attempt <= 5; attempt++) {
- try {
- const response = await fetch(`${apiUrl}${path}`, {
- method: 'POST',
- headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
- body: JSON.stringify(body),
- signal: AbortSignal.timeout(180000),
- });
- const text = await response.text();
- let result = {};
- try { result = JSON.parse(text); } catch {}
- if (!response.ok || result.success === false) {
- throw new Error(result.message || result.errors?.[0]?.message || text.slice(0, 300) || `HTTP ${response.status}`);
- }
- return result.data || result;
- } catch (error) {
- if (attempt === 5) throw error;
- const delay = attempt * 15000;
- log('amazon_retry', { shopId, path, attempt, delay, error: error.message });
- await sleep(delay);
- }
- }
- }
- async function getReports() {
- const reports = [];
- let skip = 0;
- while (true) {
- const page = await parseRequest(`/classes/Reports?limit=1000&skip=${skip}&order=createdAt`);
- reports.push(...(page.results || []));
- const length = page.results?.length || 0;
- if (length < 1000) return reports;
- skip += length;
- }
- }
- async function updateReport(reportId, body) {
- return parseRequest(`/classes/Reports/${reportId}`, 'PUT', body);
- }
- const finalStates = new Set(['DONE', 'CANCELLED', 'FATAL']);
- const processedReports = new Set();
- let totalParsedRecords = 0;
- for (let round = 1; round <= rounds; round++) {
- const reports = (await getReports()).filter(report => report.isParsed !== true);
- let pending = 0;
- let processedThisRound = 0;
- const statusCounts = {};
- log('report_round_started', { round, reports: reports.length });
- for (const report of reports) {
- const shopId = report.shop?.objectId;
- const amazonReportId = report.reportId;
- if (!shopId || !amazonReportId) {
- log('report_skipped', { objectId: report.objectId, reason: 'missing shop pointer or reportId' });
- continue;
- }
- let status = report.processingStatus || 'IN_QUEUE';
- let documentId = report.reportDocumentId || '';
- if (!finalStates.has(status)) {
- try {
- const result = await amazonRequest(shopId, '/forward', {
- path: `/reports/2021-06-30/reports/${amazonReportId}`,
- method: 'GET',
- });
- status = result.processingStatus || status;
- documentId = result.reportDocumentId || documentId;
- await updateReport(report.objectId, {
- processingStatus: status,
- ...(documentId ? { reportDocumentId: documentId } : {}),
- ...(result.processingStartTime ? { processingStartTime: { __type: 'Date', iso: result.processingStartTime } } : {}),
- ...(result.processingEndTime ? { processingEndTime: { __type: 'Date', iso: result.processingEndTime } } : {}),
- });
- log('report_status', { objectId: report.objectId, shopId, reportId: amazonReportId, status, hasDocument: Boolean(documentId) });
- } catch (error) {
- pending++;
- log('report_status_failed', { objectId: report.objectId, shopId, reportId: amazonReportId, error: error.message });
- continue;
- }
- }
- statusCounts[status] = (statusCounts[status] || 0) + 1;
- if (status !== 'DONE') {
- if (!finalStates.has(status)) pending++;
- continue;
- }
- if (!documentId) {
- pending++;
- log('report_waiting_document', { objectId: report.objectId, shopId, reportId: amazonReportId });
- continue;
- }
- if (processedReports.has(report.objectId)) continue;
- try {
- const result = await amazonRequest(shopId, '/processReport', {
- reportDocumentId: documentId,
- reportType: report.reportType,
- });
- const processResult = result.processResult || result.data?.processResult || {};
- if (processResult.success === false) throw new Error(processResult.message || 'report parser returned failure');
- const parsed = Number(processResult.processed || 0);
- await updateReport(report.objectId, {
- isParsed: true,
- parsedAt: { __type: 'Date', iso: new Date().toISOString() },
- parsedRecords: parsed,
- });
- processedReports.add(report.objectId);
- totalParsedRecords += parsed;
- processedThisRound++;
- log('report_parsed', { objectId: report.objectId, shopId, reportId: amazonReportId, reportType: report.reportType, parsed });
- } catch (error) {
- pending++;
- log('report_parse_failed', { objectId: report.objectId, shopId, reportId: amazonReportId, error: error.message });
- }
- await sleep(1000);
- }
- log('report_round_finished', { round, pending, processedThisRound, statusCounts, totalParsedRecords });
- if (pending === 0) break;
- if (round < rounds) await sleep(roundDelayMs);
- }
- const remaining = (await getReports()).filter(report => report.isParsed !== true && !['CANCELLED', 'FATAL'].includes(report.processingStatus));
- log('report_processing_finished', {
- processedReports: processedReports.size,
- totalParsedRecords,
- remaining: remaining.length,
- remainingStatuses: remaining.reduce((counts, report) => {
- const status = report.processingStatus || 'UNKNOWN';
- counts[status] = (counts[status] || 0) + 1;
- return counts;
- }, {}),
- });
|