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; interface StoredSource extends ParseObject { productId?: string; payload: ListingSourceSnapshot } interface StoredReview extends ParseObject { naturalKey: string; productId: string; sourceReviewId?: 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 scope = args.get('--scope') === 'own' ? 'own' : 'listing-cohort'; const scopeName = scope === 'own' ? 'voc-own-products' : ACTIVE_COHORT; 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 { 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('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'); let allSources: StoredSource[] = []; let cohortProductIds: Set; let skuToProduct: Map; if (scope === 'own') { const ownProducts = await target.findAll(VOC_PARSE_CLASSES.product, { workspaceId, platform: 'jd', role: 'own', }); cohortProductIds = new Set(ownProducts.map((row) => text(row.productId)).filter(Boolean)); skuToProduct = buildOwnSkuMap(cohortProductIds); if (!cohortProductIds.size) throw new Error('listing_review_own_product_scope_empty'); } else { allSources = await target.findAll(VOC_PARSE_CLASSES.listingSourceSnapshot, { workspaceId, platform: 'jd' }); const currentSources = await target.findAll(VOC_PARSE_CLASSES.listingSourceSnapshot, { workspaceId, platform: 'jd', isCurrent: true, catalogIncluded: true, catalogCohort: ACTIVE_COHORT, }); cohortProductIds = new Set(currentSources.map((row) => row.payload.productId)); if (cohortProductIds.size !== 625) throw new Error(`listing_review_cohort_count_mismatch:${cohortProductIds.size}:625`); skuToProduct = buildSkuMap(allSources, cohortProductIds); } if (!skuToProduct.size) throw new Error('listing_review_sku_map_empty'); const conflicts = scope === 'own' ? [] : findSkuConflicts(allSources, cohortProductIds); if (conflicts.length) throw new Error(`listing_review_sku_conflicts:${conflicts.slice(0, 5).join(',')}`); let existingReviews = await target.findAll(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 existingBySourceReviewId = new Map(existingReviews .filter((row) => text(row.sourceReviewId)) .map((row) => [text(row.sourceReviewId), 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: scopeName, 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 !== scopeName) { 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) ?? existingBySourceReviewId.get(reviewId); 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); if (reviewId && existing) existingBySourceReviewId.set(reviewId, existing); 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, scope, 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 = { 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): Map { const output = new Map(); 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 buildOwnSkuMap(ownProductIds: Set): Map { const output = new Map(); for (const productId of ownProductIds) output.set(productId, productId); return output; } function findSkuConflicts(rows: StoredSource[], cohortProductIds: Set): string[] { const owners = new Map(); const conflicts = new Set(); 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 { // 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>('/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 { 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 { try { return JSON.parse(await readFile(path, 'utf8')) as Checkpoint; } catch { return null; } } async function persistCheckpoint(path: string, checkpoint: Checkpoint): Promise { 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; });