jianhen-voc/deploy/collect-sorftime-reviews-year.mjs

import crypto from 'node:crypto';
import fs from 'node:fs';

const configPath = process.env.VOC_CONFIG || '/etc/jianhen-voc/config.json';
const config = JSON.parse(fs.readFileSync(configPath, 'utf8'));
const parseUrl = process.env.PARSE_URL || 'http://127.0.0.1:3000/parse';
const relayUrl = process.env.SORFTIME_URL || 'http://127.0.0.1:3000/api/sorftime/forward';
const appId = config.parse.appId;
const masterKey = config.parse.masterKey;
const maxRowsPerResponse = 10;

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));
const shopPointer = objectId => ({ __type: 'Pointer', className: 'Shop', objectId });
const isoDate = date => date.toISOString().slice(0, 10);

function log(event, data = {}) {
  console.log(JSON.stringify({ at: new Date().toISOString(), event, ...data }));
}

function hash(value) {
  return crypto.createHash('sha256').update(value).digest('hex');
}

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, 400) || response.status}`);
  }
  return result;
}

async function allRows(className, keys = '') {
  const rows = [];
  let skip = 0;
  while (true) {
    const page = await parseRequest(`/classes/${className}?limit=1000&skip=${skip}${keys ? `&keys=${keys}` : ''}`);
    rows.push(...(page.results || []));
    const length = page.results?.length || 0;
    if (length < 1000) return rows;
    skip += length;
  }
}

function unwrap(result) {
  let data = result?.data ?? result?.Data ?? result;
  while (data && typeof data === 'object' && !Array.isArray(data)
    && (data.data !== undefined || data.Data !== undefined)) {
    data = data.data ?? data.Data;
  }
  return data;
}

async function queryReviews({ asin, domain, start, end, star = '1,2,3,4,5' }) {
  for (let attempt = 1; attempt <= 4; attempt++) {
    try {
      const response = await fetch(relayUrl, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          path: '/api/ProductReviewsQuery',
          method: 'POST',
          query: { domain },
          body: {
            ASIN: asin,
            PageIndex: 1,
            PageSize: 100,
            OnlyPurchase: 0,
            Star: star,
            QueryStartDt: start,
            QueryEndDt: end,
          },
        }),
        signal: AbortSignal.timeout(90000),
      });
      const result = await response.json();
      if (!response.ok || result?.success === false) {
        throw new Error(result?.message || result?.Message || `Relay HTTP ${response.status}`);
      }
      const data = unwrap(result);
      return data?.Reviews || data?.Items || data?.List || (Array.isArray(data) ? data : []);
    } catch (error) {
      if (attempt === 4) throw error;
      await sleep(attempt * 3000);
    }
  }
}

function reviewIdOf(review, target) {
  const id = String(review.ReviewId || review.Id || review.ID || review.id || '').trim();
  if (id) return id;
  const identity = [
    target.domain,
    target.asin,
    review.ReviewsDate || review.ReviewDate || review.Date || '',
    review.ProfileName || review.Author || '',
    review.Content || review.ReviewContent || review.Body || '',
  ].join('|');
  return `derived-${hash(identity)}`;
}

function reviewDateOf(review) {
  return String(review.ReviewsDate || review.ReviewDate || review.Date || '').trim();
}

function parseReviewDate(value) {
  if (!value) return undefined;
  const normalized = /^\d{8}$/.test(value)
    ? `${value.slice(0, 4)}-${value.slice(4, 6)}-${value.slice(6, 8)}`
    : value;
  const date = new Date(normalized);
  return Number.isNaN(date.getTime()) ? undefined : { __type: 'Date', iso: date.toISOString() };
}

async function collectStarBuckets(target, start, end, resultMap, stats) {
  for (const star of ['1', '2', '3', '4', '5']) {
    const starReviews = await queryReviews({ asin: target.asin, domain: target.domain, start, end, star });
    stats.requests++;
    for (const review of starReviews) resultMap.set(reviewIdOf(review, target), review);
    log('review_star_bucket', { asin: target.asin, domain: target.domain, start, end, star, rows: starReviews.length });
    if (starReviews.length >= maxRowsPerResponse) {
      stats.unresolvedSaturatedBuckets.push({ asin: target.asin, domain: target.domain, start, end, star });
    }
  }
}

async function saveBatch(requests) {
  for (let start = 0; start < requests.length; start += 25) {
    const result = await parseRequest('/batch', 'POST', { requests: requests.slice(start, start + 25) });
    const failure = result.find?.(item => item.error)?.error;
    if (failure) throw new Error(failure.error || failure.code || 'Parse batch failed');
  }
}

const [products, listings, shops, existingReviews] = await Promise.all([
  allRows('Product'),
  allRows('Listing'),
  allRows('Shop'),
  allRows('SorftimeReviews', 'objectId,recordKey,reviewId,asin,domain,shop'),
]);
const shopById = new Map(shops.map(shop => [shop.objectId, shop]));
const existingByReviewId = new Map();
for (const review of existingReviews) {
  if (review.reviewId && !existingByReviewId.has(review.reviewId)) {
    existingByReviewId.set(review.reviewId, review);
  }
}

const targets = new Map();
for (const row of [...products, ...listings]) {
  const asin = String(row.asin || row.Asin || '').trim().toUpperCase();
  const shop = shopById.get(row.shop?.objectId);
  if (!asin || !shop) continue;
  const domain = Number(row.domain || shop.domain || 1);
  const marketplaceId = String(row.marketplaceId || shop.marketplaceId || '');
  const key = `${domain}:${asin}`;
  if (!targets.has(key)) targets.set(key, { asin, domain, marketplaceId, shopId: shop.objectId, shopName: shop.name || '' });
}

const endDate = isoDate(new Date());
const start = new Date();
start.setUTCFullYear(start.getUTCFullYear() - 1);
const startDate = isoDate(start);
const runStats = {
  startedAt: new Date().toISOString(), startDate, endDate, targets: targets.size,
  requests: 0, fetchedUnique: 0, associations: 0, withContent: 0, inserted: 0, updated: 0,
  failures: [], unresolvedSaturatedBuckets: [], byTarget: [],
};
const globalReviews = new Map();
const associations = new Map();

log('review_collection_started', { targets: targets.size, startDate, endDate });

for (const target of targets.values()) {
  const collected = new Map();
  const targetStats = { requests: 0, unresolvedSaturatedBuckets: [] };
  try {
    await collectStarBuckets(target, startDate, endDate, collected, targetStats);
    let withContent = 0;
    for (const [reviewId, review] of collected) {
      const content = String(review.Content || review.ReviewContent || review.Body || '');
      if (content.trim()) withContent++;
      if (!globalReviews.has(reviewId)) {
        globalReviews.set(reviewId, {
          review,
          asins: new Set(),
          marketplaceIds: new Set(),
          domains: new Set(),
          shopIds: new Set(),
          targets: [],
        });
      }
      const global = globalReviews.get(reviewId);
      global.asins.add(target.asin);
      if (target.marketplaceId) global.marketplaceIds.add(target.marketplaceId);
      global.domains.add(target.domain);
      global.shopIds.add(target.shopId);
      global.targets.push(target);
      const associationKey = `${target.domain}:${target.asin}:${reviewId}`;
      associations.set(associationKey, {
        recordKey: associationKey,
        reviewId,
        asin: target.asin,
        marketplaceId: target.marketplaceId,
        domain: target.domain,
        source: 'sorftime-product-reviews-year',
        capturedAt: { __type: 'Date', iso: new Date().toISOString() },
        shop: shopPointer(target.shopId),
      });
    }
    runStats.requests += targetStats.requests;
    runStats.unresolvedSaturatedBuckets.push(...targetStats.unresolvedSaturatedBuckets);
    runStats.byTarget.push({ ...target, reviews: collected.size, withContent, requests: targetStats.requests });
    log('review_target_completed', { ...target, reviews: collected.size, withContent, requests: targetStats.requests });
  } catch (error) {
    runStats.failures.push({ ...target, error: error.message });
    log('review_target_failed', { ...target, error: error.message });
  }
  await sleep(500);
}

const capturedAt = { __type: 'Date', iso: new Date().toISOString() };
const reviewRequests = [];
for (const [reviewId, global] of globalReviews) {
  const existing = existingByReviewId.get(reviewId);
  const targetsForReview = global.targets.slice().sort((a, b) => `${a.domain}:${a.asin}`.localeCompare(`${b.domain}:${b.asin}`));
  const primary = targetsForReview.find(target => target.asin === existing?.asin) || targetsForReview[0];
  const review = global.review;
  const content = String(review.Content || review.ReviewContent || review.Body || '');
  reviewRequests.push({
    method: existing ? 'PUT' : 'POST',
    path: existing ? `/parse/classes/SorftimeReviews/${existing.objectId}` : '/parse/classes/SorftimeReviews',
    body: {
      recordKey: reviewId,
      reviewId,
      asin: primary.asin,
      asins: [...global.asins].sort(),
      marketplaceId: primary.marketplaceId,
      marketplaceIds: [...global.marketplaceIds].sort(),
      domain: primary.domain,
      domains: [...global.domains].sort((a, b) => a - b),
      shopIds: [...global.shopIds].sort(),
      source: 'sorftime-product-reviews-year',
      star: Number(review.Star || review.Rating || 0),
      rating: Number(review.Star || review.Rating || 0),
      title: review.Title || review.ReviewTitle || '',
      content,
      reviewDate: reviewDateOf(review),
      reviewPublishedAt: parseReviewDate(reviewDateOf(review)),
      author: review.ProfileName || review.Author || '',
      verified: Boolean(review.OnlyPurchase || review.VerifiedPurchase),
      capturedAt,
      shop: shopPointer(primary.shopId),
      rawData: review,
    },
  });
  if (existing) runStats.updated++; else runStats.inserted++;
  if (content.trim()) runStats.withContent++;
}
await saveBatch(reviewRequests);

const existingAssociations = await allRows('SorftimeReviewProduct', 'objectId,recordKey');
const associationByKey = new Map(existingAssociations.filter(row => row.recordKey).map(row => [row.recordKey, row.objectId]));
const associationRequests = [...associations.values()].map(body => {
  const objectId = associationByKey.get(body.recordKey);
  return {
    method: objectId ? 'PUT' : 'POST',
    path: objectId ? `/parse/classes/SorftimeReviewProduct/${objectId}` : '/parse/classes/SorftimeReviewProduct',
    body,
  };
});
await saveBatch(associationRequests);

runStats.fetchedUnique = globalReviews.size;
runStats.associations = associations.size;

runStats.finishedAt = new Date().toISOString();
console.log(JSON.stringify({ event: 'review_collection_finished', ...runStats }));