Saas-voc-server/scripts/sync-jd-listings.ts

import 'dotenv/config';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { dirname, resolve } from 'node:path';
import { z } from 'zod';
import { ParseRestClient } from '../src/db/parse-rest.client.js';
import { ensureListingParseSchemas, VOC_PARSE_CLASSES } from '../src/db/parse-rest.schema.js';
import { createDatabasePool } from '../src/db/pool.js';
import { loadConfig } from '../src/config/env.js';
import { JdProductClient } from '../src/modules/listing-ai/clients/jd-product.client.js';
import { JdSpClient } from '../src/modules/listing-ai/clients/jd-sp.client.js';
import { JdTokenProvider } from '../src/modules/listing-ai/clients/jd-token.provider.js';
import { normalizeJdListing } from '../src/modules/listing-ai/normalization/jd-listing.normalizer.js';
import { buildListingContextIndex, type ListingContextIndex } from '../src/modules/listing-ai/normalization/listing-context.enricher.js';
import { ParseRestListingAiRepository } from '../src/modules/listing-ai/repositories/parse-rest-listing-ai.repository.js';
import { PostgresListingAiRepository } from '../src/modules/listing-ai/repositories/postgres-listing-ai.repository.js';

const args=new Map(process.argv.slice(2).map((arg)=>{const [key,...rest]=arg.split('=');return[key!,rest.join('=')||'true'];}));
const maxProducts=Math.max(1,Number(args.get('--max-products')??Number.POSITIVE_INFINITY));
const workspaceId=args.get('--workspace')??process.env.SAAS_DEFAULT_WORKSPACE_ID??'demashi';
const checkpointPath=resolve(args.get('--checkpoint')??'logs/jd-listing-sync-checkpoint.json');
const resume=args.get('--resume')==='true';
const cohortOnly=args.get('--cohort-only')==='true';
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(){
  const source=sourceSchema.parse(process.env);const config=loadConfig();
  const authClient=new ParseRestClient({serverUrl:source.JD_SOURCE_PARSE_URL,appId:source.JD_SOURCE_PARSE_APP_ID,masterKey:source.JD_SOURCE_PARSE_MASTER_KEY,timeoutMs:config.jdListing.timeoutMs});
  const auth=await new JdTokenProvider(authClient).latest();
  const jd=new JdProductClient(new JdSpClient({baseUrl:'https://api-cn.jd.com/rest',appKey:source.JD_APP_KEY,appSecret:source.JD_APP_SECRET,timeoutMs:config.jdListing.timeoutMs,retries:config.jdListing.retries}),config.jdListing.pageSize);
  let close=async()=>{};let repository;let contextIndex:ListingContextIndex|undefined;
  if(config.storageDriver==='parse_rest'){const client=new ParseRestClient({serverUrl:config.parse.serverUrl,appId:config.parse.appId,masterKey:config.parse.masterKey,timeoutMs:config.parse.timeoutMs});await ensureListingParseSchemas(client);repository=new ParseRestListingAiRepository(client);const [products,relations,reviews]=await Promise.all([client.findAll<Record<string,unknown>>(VOC_PARSE_CLASSES.product,{workspaceId}),client.findAll<Record<string,unknown>>(VOC_PARSE_CLASSES.productRelation,{workspaceId}),client.findAll<Record<string,unknown>>(VOC_PARSE_CLASSES.review,{workspaceId})]);contextIndex=buildListingContextIndex(products,relations,reviews);}else{const pool=createDatabasePool(config);repository=new PostgresListingAiRepository(pool);close=async()=>pool.end();}
  const cohortProductIds=cohortOnly?new Set((await repository.listAllSources(workspaceId,'jd')).map((item)=>item.productId)):null;
  if(cohortOnly&&cohortProductIds?.size!==625)throw new Error(`listing_sync_cohort_count_mismatch:${cohortProductIds?.size??0}:625`);
  let completed=new Set<string>();
  if(resume){try{const checkpoint=JSON.parse(await readFile(checkpointPath,'utf8')) as {completed?:string[]};completed=new Set(checkpoint.completed??[]);}catch{} }
  let listed=0,synced=0,failed=0,total:number|null=null;const pending:Array<Promise<void>>=[];let checkpointWrites=Promise.resolve();
  const persist=()=>{checkpointWrites=checkpointWrites.then(async()=>{await mkdir(dirname(checkpointPath),{recursive:true});await writeFile(checkpointPath,JSON.stringify({workspaceId,total,listed,synced,failed,completed:[...completed],updatedAt:new Date().toISOString()},null,2));});return checkpointWrites;};
  const consume=async(row:Record<string,unknown>)=>{const productId=String(row['productId']??row['wareId']??row['id']??'');if(!productId||completed.has(productId)||(cohortProductIds&&!cohortProductIds.has(productId)))return;try{const detail=await jd.detail(auth.accessToken,productId);const normalized=normalizeJdListing({workspaceId,shopId:auth.shopId,row,detail});const snapshot=contextIndex?.enrich(normalized)??normalized;await repository.upsertSources([snapshot]);completed.add(productId);synced+=1;}catch(error){failed+=1;console.error(JSON.stringify({event:'jd_listing_failed',productId,error:error instanceof Error?error.message:'unknown'}));}await persist();};
  try{for await(const entry of jd.listAll(auth.accessToken,maxProducts)){listed+=1;total=entry.total;const task=consume(entry.row).finally(()=>pending.splice(pending.indexOf(task),1));pending.push(task);if(pending.length>=config.jdListing.detailConcurrency)await Promise.race(pending);}await Promise.all(pending);await persist();console.log(JSON.stringify({mode:'completed',workspaceId,total,listed,synced,failed,completed:completed.size,cohortOnly,cohortExpected:cohortProductIds?.size??null,checkpointPath},null,2));if(failed||(cohortProductIds&&completed.size!==cohortProductIds.size))process.exitCode=2;}finally{await close();}
}
main().catch((error)=>{console.error(`[sync-jd-listings] ${error instanceof Error?error.message:error}`);process.exitCode=1;});