jianhen-voc/deploy/reparse-sales-traffic-reports.mjs

import zlib from 'node:zlib';
import {
  activeAmazonShops, allRows, dateValue, forward, hash, log,
  shopPointer, sleep, updateObject, upsertMany,
} from './sp-api-collector-lib.mjs';

function parseDocument(content) {
  const trimmed = content.replace(/^\uFEFF/, '').trim();
  if (!trimmed) return [];
  const parsed = JSON.parse(trimmed);
  const rows = [];
  for (const [section, value] of Object.entries(parsed)) {
    if (!Array.isArray(value)) continue;
    for (const item of value) {
      rows.push({ _section: section, ...(item && typeof item === 'object' ? item : { value: item }) });
    }
  }
  return rows.length ? rows : [parsed];
}

const shops = await activeAmazonShops();
const reports = (await allRows('Reports')).filter(report =>
  report.recordKey?.startsWith('MP:')
  && report.reportType === 'GET_SALES_AND_TRAFFIC_REPORT'
  && report.isParsed !== true
  && report.processingStatus === 'DONE'
  && report.reportDocumentId
);

for (const report of reports) {
  const shop = shops.find(item => item.objectId === report.shop?.objectId);
  if (!shop) {
    log('sales_report_skipped', { reportId: report.reportId, reason: 'shop not found' });
    continue;
  }
  try {
    const document = await forward(shop.objectId,
      `/reports/2021-06-30/documents/${encodeURIComponent(report.reportDocumentId)}`,
      'GET', undefined, 5);
    if (!document.url) throw new Error('Report document returned no URL');
    const response = await fetch(document.url, { signal: AbortSignal.timeout(300000) });
    if (!response.ok) throw new Error(`Report download failed: ${response.status}`);
    let bytes = Buffer.from(await response.arrayBuffer());
    if (document.compressionAlgorithm === 'GZIP') bytes = zlib.gunzipSync(bytes);
    const rows = parseDocument(new TextDecoder('utf-8').decode(bytes));
    const marketplaceKey = [...(report.marketplaceIds || [])].sort().join(',') || shop.marketplaceId;
    const objects = rows.map(row => {
      const rowHash = hash(row);
      return {
        recordKey: `${report.region || 'UNKNOWN'}:${marketplaceKey}:${report.reportType}:${rowHash}`,
        rowHash,
        reportType: report.reportType,
        reportId: report.reportId,
        region: report.region || '',
        marketplaceIds: report.marketplaceIds || [],
        capturedAt: dateValue(new Date()),
        rowData: row,
        rawData: row,
        shop: shopPointer(shop.objectId),
      };
    });
    const saved = await upsertMany('AmazonReportRow', objects);
    await updateObject('Reports', report.objectId, {
      isParsed: true,
      parsedAt: dateValue(new Date()),
      parsedRecords: objects.length,
    });
    log('sales_report_reparsed', {
      marketplaceIds: report.marketplaceIds,
      reportId: report.reportId,
      rows: objects.length,
      ...saved,
    });
  } catch (error) {
    log('sales_report_reparse_failed', { reportId: report.reportId, error: error.message });
  }
  await sleep(1000);
}

log('sales_report_reparse_finished', { reports: reports.length });