Saas-voc-server/scripts/sync-jd-listing-reviews.ts

import 'dotenv/config';
import { createHash } from 'node:crypto';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { dirname, resolve } from 'node:path';
import { z } from 'zod';
import { loadConfig } from '../src/config/env.js';
import { ParseRestClient, parseDate, type ParseObject } from '../src/db/parse-rest.client.js';
import { VOC_PARSE_CLASSES } from '../src/db/parse-rest.schema.js';
import { makeReviewKey } from '../src/modules/domestic-voc/domain/identity.js';
import type { ListingSourceSnapshot } from '../src/modules/listing-ai/domain.js';

type JsonRecord = Record<string, unknown>;
interface StoredSource extends ParseObject { productId?: string; payload: ListingSourceSnapshot }
interface StoredReview extends ParseObject { naturalKey: string; productId: string; reviewDate?: unknown }
interface Checkpoint {
  workspaceId: string;
  cohort: string;
  nextPage: number;
  totalItems: number;
  totalPages: number;
  pagesProcessed: number;
  commentsScanned: number;
  commentsMatched: number;
  commentsCreated: number;
  commentsUpdated: number;
  commentsDuplicate: number;
  commentsUnmatched: number;
  matchedProducts: string[];
  startedAt: string;
  updatedAt: string;
  status: 'running' | 'completed';
  phase?: 'windowed';
  historyStart?: string;
  historyEnd?: string;
  windowStart?: string;
  windowEnd?: string;
  windowsProcessed?: number;
}

const ACTIVE_COHORT = 'listing-jd-v3-formal-625';
const METHOD = 'jingdong.pop.PopCommentJsfService.getVenderCommentsForJos';
const PAGE_SIZE = 50;
const DEFAULT_HISTORY_START = '2026-06-01 00:00:00';
const DEFAULT_HISTORY_END = '2026-08-29 00:00:00';
const args = new Map(process.argv.slice(2).map((arg) => {
  const [key, ...rest] = arg.split('=');
  return [key!, rest.join('=') || 'true'];
}));
const workspaceId = args.get('--workspace') ?? process.env.SAAS_DEFAULT_WORKSPACE_ID ?? 'demashi';
const checkpointPath = resolve(args.get('--checkpoint') ?? 'logs/jd-listing-review-sync-2026-06-01-to-2026-08-28.json');
const resume = args.get('--resume') === 'true';
const delayMs = Math.max(100, Number(args.get('--delay-ms') ?? 200));
const maxPages = Math.max(1, Number(args.get('--max-pages') ?? Number.POSITIVE_INFINITY));
const historyStartArg = args.get('--history-start') ?? DEFAULT_HISTORY_START;
const historyEndArg = args.get('--history-end') ?? DEFAULT_HISTORY_END;
const windowDays = Math.max(1, Number(args.get('--window-days') ?? 7));
const pruneOutsideRange = args.get('--prune-outside-range') !== 'false';
const sourceSchema = z.object({
  JD_SOURCE_PARSE_URL: z.url(),
  JD_SOURCE_PARSE_APP_ID: z.string().min(1),
  JD_SOURCE_PARSE_MASTER_KEY: z.string().min(1),
  JD_APP_KEY: z.string().min(1),
  JD_APP_SECRET: z.string().min(1),
});

async function main(): Promise<void> {
  const sourceConfig = sourceSchema.parse(process.env);
  const config = loadConfig();
  const target = new ParseRestClient({
    serverUrl: config.parse.serverUrl,
    appId: config.parse.appId,
    masterKey: config.parse.masterKey,
    timeoutMs: config.parse.timeoutMs,
  });
  const authSource = new ParseRestClient({
    serverUrl: sourceConfig.JD_SOURCE_PARSE_URL,
    appId: sourceConfig.JD_SOURCE_PARSE_APP_ID,
    masterKey: sourceConfig.JD_SOURCE_PARSE_MASTER_KEY,
    timeoutMs: config.jdListing.timeoutMs,
  });
  const authRow = (await authSource.find<JsonRecord>('EcomAuth', {
    where: { platform: 'jd', type: 'access_token' }, order: '-createdAt', limit: 1,
  })).results[0];
  const authData = record(authRow?.['data']);
  const accessToken = text(authData['access_token']);
  if (!accessToken) throw new Error('jd_authorization_missing');

  const currentSources = await target.findAll<StoredSource>(VOC_PARSE_CLASSES.listingSourceSnapshot, {
    workspaceId, platform: 'jd', isCurrent: true, catalogIncluded: true, catalogCohort: ACTIVE_COHORT,
  });
  const cohortProductIds = new Set(currentSources.map((row) => row.payload.productId));
  if (cohortProductIds.size !== 625) throw new Error(`listing_review_cohort_count_mismatch:${cohortProductIds.size}:625`);

  const allSources = await target.findAll<StoredSource>(VOC_PARSE_CLASSES.listingSourceSnapshot, { workspaceId, platform: 'jd' });
  const skuToProduct = buildSkuMap(allSources, cohortProductIds);
  if (!skuToProduct.size) throw new Error('listing_review_sku_map_empty');
  const conflicts = findSkuConflicts(allSources, cohortProductIds);
  if (conflicts.length) throw new Error(`listing_review_sku_conflicts:${conflicts.slice(0, 5).join(',')}`);

  let existingReviews = await target.findAll<StoredReview>(VOC_PARSE_CLASSES.review, { workspaceId, platform: 'jd' });
  if (pruneOutsideRange) {
    const outsideRange = existingReviews.filter((review) => cohortProductIds.has(review.productId)
      && !isWithinJdRange(review.reviewDate, historyStartArg, historyEndArg));
    const deleteRequests = outsideRange.map((review) => ({
      method: 'DELETE' as const,
      path: `/classes/${VOC_PARSE_CLASSES.review}/${review.objectId}`,
    }));
    for (let index = 0; index < deleteRequests.length; index += 50) {
      await writeFmodeBatch(target, deleteRequests.slice(index, index + 50));
    }
    existingReviews = existingReviews.filter((review) => !outsideRange.includes(review));
    console.log(JSON.stringify({ event: 'listing_review_prune', deleted: outsideRange.length, historyStart: historyStartArg, historyEndExclusive: historyEndArg }));
  }
  const existingByNaturalKey = new Map(existingReviews.map((row) => [row.naturalKey, row]));
  const seen = new Set(existingByNaturalKey.keys());
  let checkpoint = resume ? await readCheckpoint(checkpointPath) : null;
  if (checkpoint?.status === 'completed') {
    console.log(JSON.stringify({ mode: 'already_completed', checkpointPath, ...checkpoint }, null, 2));
    return;
  }
  const now = new Date().toISOString();
  checkpoint ??= {
    workspaceId, cohort: ACTIVE_COHORT, nextPage: 1, totalItems: 0, totalPages: 0,
    pagesProcessed: 0, commentsScanned: 0, commentsMatched: 0, commentsCreated: 0,
    commentsUpdated: 0, commentsDuplicate: 0, commentsUnmatched: 0, matchedProducts: [],
    startedAt: now, updatedAt: now, status: 'running',
  };
  if (checkpoint.workspaceId !== workspaceId || checkpoint.cohort !== ACTIVE_COHORT) {
    throw new Error('listing_review_checkpoint_scope_mismatch');
  }
  if (checkpoint.phase === 'windowed'
    && (checkpoint.historyStart !== normalizeJdDateTime(historyStartArg)
      || checkpoint.historyEnd !== normalizeJdDateTime(historyEndArg))) {
    throw new Error(`listing_review_checkpoint_range_mismatch:${checkpoint.historyStart}:${checkpoint.historyEnd}`);
  }
  const matchedProducts = new Set(checkpoint.matchedProducts);
  const client = new JdJosReviewClient(sourceConfig.JD_APP_KEY, sourceConfig.JD_APP_SECRET, accessToken);
  if (!checkpoint.totalItems) {
    const metadata = await client.page(1, PAGE_SIZE);
    checkpoint.totalItems = metadata.totalItem;
    checkpoint.totalPages = Math.ceil(metadata.totalItem / PAGE_SIZE);
  }
  if (checkpoint.phase !== 'windowed') {
    const oldestStoredReview = existingReviews
      .filter((review) => cohortProductIds.has(review.productId))
      .map((review) => storedDateIso(review.reviewDate))
      .filter((value): value is string => Boolean(value))
      .sort()[0];
    const historyEnd = normalizeJdDateTime(historyEndArg || oldestStoredReview || jdTomorrow());
    const historyStart = normalizeJdDateTime(historyStartArg);
    checkpoint.phase = 'windowed';
    checkpoint.historyStart = historyStart;
    checkpoint.historyEnd = historyEnd;
    checkpoint.windowStart = historyStart;
    checkpoint.windowEnd = minJdDateTime(addJdDays(historyStart, windowDays), historyEnd);
    checkpoint.windowsProcessed = 0;
    checkpoint.nextPage = 1;
    checkpoint.updatedAt = new Date().toISOString();
    await persistCheckpoint(checkpointPath, checkpoint);
    console.log(JSON.stringify({ event: 'listing_review_window_migration', historyStart, historyEnd, retainedRecentPages: checkpoint.pagesProcessed }));
  }
  let pagesThisRun = 0;

  while (checkpoint.windowStart && checkpoint.windowEnd && checkpoint.historyEnd
    && jdEpoch(checkpoint.windowStart) < jdEpoch(checkpoint.historyEnd)
    && pagesThisRun < maxPages) {
    const page = checkpoint.nextPage;
    const response = await client.page(page, PAGE_SIZE, {
      beginTime: checkpoint.windowStart,
      endTime: checkpoint.windowEnd,
    });
    const windowPages = Math.ceil(response.totalItem / PAGE_SIZE);
    if (windowPages > 200) throw new Error(`listing_review_window_too_large:${checkpoint.windowStart}:${checkpoint.windowEnd}:${response.totalItem}`);
    if (!response.comments.length && page <= windowPages) throw new Error(`listing_review_empty_page:${page}/${windowPages}`);
    const requests: Array<{ method: 'POST' | 'PUT'; path: string; body: unknown }> = [];
    for (const comment of response.comments) {
      checkpoint.commentsScanned += 1;
      const skuId = text(comment['skuid'] ?? comment['skuId']);
      const productId = skuToProduct.get(skuId);
      if (!productId) { checkpoint.commentsUnmatched += 1; continue; }
      const content = text(comment['content']);
      if (!content) { checkpoint.commentsUnmatched += 1; continue; }
      const reviewId = text(comment['commentId']);
      const reviewDate = dateIso(comment['creationTime']);
      const reviewKey = makeReviewKey({ platform: 'jd', productId, reviewId, content, reviewDate });
      const naturalKey = [workspaceId, 'jd', reviewKey].map(encodeURIComponent).join('|');
      const existing = existingByNaturalKey.get(naturalKey);
      if (!existing && seen.has(naturalKey)) { checkpoint.commentsDuplicate += 1; continue; }
      const body = {
        naturalKey, workspaceId, platform: 'jd', productId, sourceReviewId: reviewId || null,
        reviewKey, rating: rating(comment['score']), content,
        reviewDate: reviewDate ? parseDate(reviewDate) : null,
        rawPayload: sanitizeComment(comment),
      };
      if (existing) {
        requests.push({ method: 'PUT', path: `/classes/${VOC_PARSE_CLASSES.review}/${existing.objectId}`, body });
        checkpoint.commentsUpdated += 1;
      } else {
        requests.push({ method: 'POST', path: `/classes/${VOC_PARSE_CLASSES.review}`, body });
        checkpoint.commentsCreated += 1;
      }
      seen.add(naturalKey);
      matchedProducts.add(productId);
      checkpoint.commentsMatched += 1;
    }
    for (let index = 0; index < requests.length; index += 50) {
      await writeFmodeBatch(target, requests.slice(index, index + 50));
    }
    checkpoint.pagesProcessed += 1;
    pagesThisRun += 1;
    const windowFinished = page >= Math.max(1, windowPages);
    if (windowFinished) {
      checkpoint.windowsProcessed = (checkpoint.windowsProcessed ?? 0) + 1;
      checkpoint.windowStart = checkpoint.windowEnd;
      checkpoint.windowEnd = minJdDateTime(addJdDays(checkpoint.windowStart, windowDays), checkpoint.historyEnd);
      checkpoint.nextPage = 1;
    } else {
      checkpoint.nextPage = page + 1;
    }
    checkpoint.matchedProducts = [...matchedProducts].sort();
    checkpoint.updatedAt = new Date().toISOString();
    await persistCheckpoint(checkpointPath, checkpoint);
    if (checkpoint.pagesProcessed % 25 === 0 || windowFinished && (checkpoint.windowsProcessed ?? 0) % 25 === 0) {
      console.log(JSON.stringify({ event: 'listing_review_progress', window: checkpoint.windowsProcessed, windowStart: checkpoint.windowStart, page, windowPages, scanned: checkpoint.commentsScanned, matched: checkpoint.commentsMatched, created: checkpoint.commentsCreated, updated: checkpoint.commentsUpdated, matchedProducts: matchedProducts.size }));
    }
    await wait(delayMs);
  }
  if (checkpoint.windowStart && checkpoint.historyEnd && jdEpoch(checkpoint.windowStart) >= jdEpoch(checkpoint.historyEnd)) checkpoint.status = 'completed';
  checkpoint.updatedAt = new Date().toISOString();
  checkpoint.matchedProducts = [...matchedProducts].sort();
  await persistCheckpoint(checkpointPath, checkpoint);
  console.log(JSON.stringify({
    mode: checkpoint.status, workspaceId, cohortProducts: cohortProductIds.size, mappedSkus: skuToProduct.size,
    totalItems: checkpoint.totalItems, totalPages: checkpoint.totalPages, pagesProcessed: checkpoint.pagesProcessed,
    commentsScanned: checkpoint.commentsScanned, commentsMatched: checkpoint.commentsMatched,
    commentsCreated: checkpoint.commentsCreated, commentsUpdated: checkpoint.commentsUpdated,
    commentsDuplicate: checkpoint.commentsDuplicate, commentsUnmatched: checkpoint.commentsUnmatched,
    matchedProducts: matchedProducts.size, checkpointPath,
  }, null, 2));
}

class JdJosReviewClient {
  constructor(private readonly appKey: string, private readonly appSecret: string, private readonly accessToken: string) {}
  async page(page: number, pageSize: number, filters: JsonRecord = {}): Promise<{ totalItem: number; comments: JsonRecord[] }> {
    for (let attempt = 0; attempt < 6; attempt += 1) {
      try {
        const params: Record<string, string> = {
          method: METHOD, access_token: this.accessToken, app_key: this.appKey,
          timestamp: jdTime(), v: '2.0', sign_method: 'md5',
          '360buy_param_json': JSON.stringify({ page, pageSize, ...filters }),
        };
        const plain = Object.keys(params).sort().map((key) => `${key}${params[key] ?? ''}`).join('');
        params['sign'] = createHash('md5').update(`${this.appSecret}${plain}${this.appSecret}`).digest('hex').toUpperCase();
        const response = await fetch(`https://api.jd.com/routerjson?${new URLSearchParams(params)}`, { signal: AbortSignal.timeout(30_000) });
        const body = await response.json() as JsonRecord;
        if (!response.ok) throw new Error(`jd_review_http_${response.status}`);
        const root = record(body[Object.keys(body)[0] ?? '']);
        if (text(root['code']) !== '0' || text(root['resultCode']) !== '200') {
          const detail = text(root['resultMessage'] ?? root['resultMsg'] ?? root['message'] ?? root['msg'] ?? root['errorMessage']);
          const diagnostic = JSON.stringify(Object.fromEntries(Object.entries(root).filter(([key]) => key !== 'comments'))).slice(0, 1_000);
          throw new Error(`jd_review_api_${text(root['code'])}_${text(root['resultCode'])}${detail ? `:${detail}` : ''}:${diagnostic}`);
        }
        return { totalItem: Math.max(0, Number(root['totalItem']) || 0), comments: list(root['comments']).map(record) };
      } catch (error) {
        if (attempt >= 5) throw error;
        await wait(Math.min(8_000, 500 * 2 ** attempt));
      }
    }
    throw new Error('jd_review_retry_exhausted');
  }
}

function buildSkuMap(rows: StoredSource[], cohortProductIds: Set<string>): Map<string, string> {
  const output = new Map<string, string>();
  for (const row of rows) {
    const source = row.payload;
    if (!cohortProductIds.has(source.productId)) continue;
    output.set(source.productId, source.productId);
    for (const sku of source.skus ?? []) if (sku.skuId) output.set(String(sku.skuId), source.productId);
  }
  return output;
}

function findSkuConflicts(rows: StoredSource[], cohortProductIds: Set<string>): string[] {
  const owners = new Map<string, string>(); const conflicts = new Set<string>();
  for (const row of rows) {
    const source = row.payload; if (!cohortProductIds.has(source.productId)) continue;
    for (const sku of source.skus ?? []) {
      const skuId = String(sku.skuId || ''); if (!skuId) continue;
      const previous = owners.get(skuId); if (previous && previous !== source.productId) conflicts.add(skuId); else owners.set(skuId, source.productId);
    }
  }
  return [...conflicts];
}

function sanitizeComment(comment: JsonRecord): JsonRecord {
  const keys = ['commentId', 'creationTime', 'content', 'skuName', 'score', 'skuid', 'images', 'isVenderReply', 'replyCount', 'usefulCount', 'skuImage', 'status', 'videos'];
  return Object.fromEntries(keys.filter((key) => comment[key] !== undefined).map((key) => [key, comment[key]]));
}
async function writeFmodeBatch(
  client: ParseRestClient,
  requests: Array<{ method: 'POST' | 'PUT' | 'DELETE'; path: string; body?: unknown }>,
): Promise<void> {
  // Fmode exposes Parse at /backend/{appId}/data, but its batch router expects
  // nested request paths to start at /data rather than repeat the full mount.
  const results = await client.request<Array<{ error?: { code?: number; error?: string } }>>('/batch', {
    method: 'POST',
    body: {
      requests: requests.map((request) => ({ ...request, path: `/data${request.path}` })),
    },
  });
  const failure = results.find((result) => result.error)?.error;
  if (failure) throw new Error(`listing_review_batch_${failure.code ?? 'unknown'}:${failure.error ?? 'failed'}`);
}
function storedDateIso(value: unknown): string | null {
  if (typeof value === 'string') return dateIso(value);
  const iso = text(record(value)['iso']);
  return iso ? dateIso(iso) : null;
}
function isWithinJdRange(value: unknown, start: string, endExclusive: string): boolean {
  const iso = storedDateIso(value);
  if (!iso) return false;
  const epoch = new Date(iso).valueOf();
  return epoch >= jdEpoch(normalizeJdDateTime(start)) && epoch < jdEpoch(normalizeJdDateTime(endExclusive));
}
function normalizeJdDateTime(value: string): string {
  if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(value)) return value;
  const date = new Date(value);
  if (Number.isNaN(date.valueOf())) throw new Error(`listing_review_invalid_date:${value}`);
  return formatJdDateTime(date);
}
function jdEpoch(value: string): number {
  const epoch = new Date(`${value.replace(' ', 'T')}+08:00`).valueOf();
  if (Number.isNaN(epoch)) throw new Error(`listing_review_invalid_jd_date:${value}`);
  return epoch;
}
function addJdDays(value: string, days: number): string { return formatJdDateTime(new Date(jdEpoch(value) + days * 86_400_000)); }
function minJdDateTime(left: string, right: string): string { return jdEpoch(left) <= jdEpoch(right) ? left : right; }
function jdTomorrow(): string { return `${formatJdDateTime(new Date(Date.now() + 86_400_000)).slice(0, 10)} 00:00:00`; }
function rating(value: unknown): number { const parsed = Number(value); return Number.isFinite(parsed) && parsed >= 0 && parsed <= 5 ? parsed : 0; }
function dateIso(value: unknown): string | null { const parsed = Number(value); const date = Number.isFinite(parsed) ? new Date(parsed < 10_000_000_000 ? parsed * 1_000 : parsed) : new Date(String(value ?? '')); return Number.isNaN(date.valueOf()) ? null : date.toISOString(); }
function record(value: unknown): JsonRecord { return value && typeof value === 'object' && !Array.isArray(value) ? value as JsonRecord : {}; }
function list(value: unknown): unknown[] { return Array.isArray(value) ? value : []; }
function text(value: unknown): string { return value === null || value === undefined ? '' : String(value).trim(); }
function wait(milliseconds: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, milliseconds)); }
function formatJdDateTime(date: Date): string { const parts = new Intl.DateTimeFormat('sv-SE', { timeZone: 'Asia/Shanghai', year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', second: '2-digit', hourCycle: 'h23' }).formatToParts(date); const get = (type: Intl.DateTimeFormatPartTypes) => parts.find((part) => part.type === type)?.value ?? ''; return `${get('year')}-${get('month')}-${get('day')} ${get('hour')}:${get('minute')}:${get('second')}`; }
function jdTime(): string { return formatJdDateTime(new Date()); }
async function readCheckpoint(path: string): Promise<Checkpoint | null> { try { return JSON.parse(await readFile(path, 'utf8')) as Checkpoint; } catch { return null; } }
async function persistCheckpoint(path: string, checkpoint: Checkpoint): Promise<void> { await mkdir(dirname(path), { recursive: true }); await writeFile(path, JSON.stringify(checkpoint, null, 2)); }

main().catch((error) => { console.error(`[sync-jd-listing-reviews] ${error instanceof Error ? error.message : error}`); process.exitCode = 1; });