chore: 清理归档文件和文档模板
删除不再需要的归档文件和过时的文档模板，包括多个README、安全策略、前端集成蓝图等文件，同时移除了未使用的业务文档和项目结构文件。 优化项目结构，移除冗余文件，保持代码库整洁。主要删除archive/handover目录下的多个文件及doc目录下的部分文档模板。
This commit is contained in:
@@ -1,115 +0,0 @@
|
||||
import { Job } from 'bullmq';
|
||||
import { WorkerHub } from './WorkerHub';
|
||||
import { CrawlerService } from '../services/CrawlerService';
|
||||
import { AIService } from '../services/AIService';
|
||||
import { FingerprintEngine } from '../core/ai/FingerprintEngine';
|
||||
import { ProductService } from '../services/ProductService';
|
||||
import { AuditService } from '../services/AuditService';
|
||||
import { logger } from '../utils/logger';
|
||||
|
||||
/**
|
||||
* [CORE_WORK_01] 采集 Worker (Crawler Worker)
|
||||
* @description 异步执行产品抓取、多模态解析、指纹生成并入库，支持任务追踪与审计
|
||||
*/
|
||||
export class CrawlerWorker {
|
||||
private static QUEUE_NAME = 'crawler-tasks';
|
||||
|
||||
/**
|
||||
* 初始化并注册 Worker
|
||||
*/
|
||||
static init() {
|
||||
WorkerHub.registerWorker(this.QUEUE_NAME, async (job: Job) => {
|
||||
const { url, sandbox, traceContext } = job.data;
|
||||
const { tenantId, shopId, taskId, traceId, userId } = traceContext;
|
||||
|
||||
logger.info(`[CrawlerWorker] Starting task ${job.id} for URL: ${url}`);
|
||||
|
||||
try {
|
||||
// 1. 抓取
|
||||
let productData = await CrawlerService.crawlProduct(url, { useSandbox: sandbox });
|
||||
|
||||
// 2. 多模态优化
|
||||
const optimized = await AIService.analyzeMultiModalProduct({
|
||||
title: productData.title || '',
|
||||
description: productData.description,
|
||||
attributes: productData.attributes || {},
|
||||
imageUrls: productData.images || []
|
||||
});
|
||||
|
||||
productData.title = optimized.optimizedTitle;
|
||||
productData.description = optimized.optimizedDescription;
|
||||
productData.attributes = { ...productData.attributes, ...optimized.validatedAttributes };
|
||||
|
||||
// 3. 指纹生成
|
||||
const fingerprint = await FingerprintEngine.generateCompositeFingerprint({
|
||||
title: productData.title,
|
||||
description: productData.description,
|
||||
mainImage: productData.mainImage || ''
|
||||
});
|
||||
|
||||
// 4. 入库
|
||||
const id = await ProductService.create({
|
||||
...productData,
|
||||
phash: fingerprint.phash,
|
||||
semanticHash: fingerprint.semanticHash,
|
||||
vectorEmbedding: JSON.stringify(fingerprint.vectorEmbedding),
|
||||
status: 'draft'
|
||||
});
|
||||
|
||||
// 5. 审计日志
|
||||
await AuditService.log({
|
||||
tenantId,
|
||||
shopId,
|
||||
taskId,
|
||||
traceId,
|
||||
userId,
|
||||
module: 'SYNC',
|
||||
action: 'CRAWLER_ASYNC_COMPLETE',
|
||||
resourceType: 'product',
|
||||
resourceId: String(id),
|
||||
afterSnapshot: { url, id },
|
||||
result: 'success',
|
||||
source: 'node'
|
||||
});
|
||||
|
||||
return { id, url, status: 'completed' };
|
||||
} catch (err: any) {
|
||||
logger.error(`[CrawlerWorker] Task ${job.id} failed: ${err.message}`);
|
||||
|
||||
// 错误审计
|
||||
await AuditService.log({
|
||||
tenantId,
|
||||
shopId,
|
||||
taskId,
|
||||
traceId,
|
||||
userId,
|
||||
module: 'SYNC',
|
||||
action: 'CRAWLER_ASYNC_FAILED',
|
||||
resourceType: 'product',
|
||||
resourceId: url,
|
||||
result: 'failed',
|
||||
errorCode: 'CRAWLER_WORKER_ERROR',
|
||||
errorMessage: err.message,
|
||||
source: 'node'
|
||||
});
|
||||
|
||||
throw err;
|
||||
}
|
||||
}, 10); // 并发数限制为 10
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交采集任务到队列
|
||||
*/
|
||||
static async submit(data: {
|
||||
url: string;
|
||||
sandbox?: boolean;
|
||||
traceContext: any;
|
||||
}) {
|
||||
const queue = WorkerHub.getQueue(this.QUEUE_NAME);
|
||||
return await queue.add(`crawl-${Date.now()}`, data, {
|
||||
attempts: 3,
|
||||
backoff: { type: 'exponential', delay: 1000 }
|
||||
});
|
||||
}
|
||||
}
|
||||
310
server/src/workers/PlatformSyncWorker.ts
Normal file
310
server/src/workers/PlatformSyncWorker.ts
Normal file
@@ -0,0 +1,310 @@
|
||||
import { Job } from 'bullmq';
|
||||
import { WorkerHub } from './WorkerHub';
|
||||
import { PlatformApiService, PlatformApiConfig, SyncOptions } from '../services/PlatformApiService';
|
||||
import { ProductService } from '../services/ProductService';
|
||||
import { OrderService } from '../services/OrderService';
|
||||
import { AuditService } from '../services/AuditService';
|
||||
import { logger } from '../utils/logger';
|
||||
|
||||
/**
|
||||
* PlatformSyncWorker - 平台数据同步Worker
|
||||
*
|
||||
* 功能定位：
|
||||
* - 异步执行有API平台的数据同步（Amazon, eBay, Shopee等）
|
||||
* - 支持商品同步、订单同步、库存更新
|
||||
* - 无API平台的采集由浏览器插件处理，不经过此Worker
|
||||
*
|
||||
* 安全约束：
|
||||
* - 并发数限制 ≤ 10（符合资源保护要求）
|
||||
* - 所有操作携带五元组追踪信息
|
||||
* - 支持限流和错误重试
|
||||
*
|
||||
* @author AI-Backend-1
|
||||
* @taskId BE-P008, BE-O007
|
||||
*/
|
||||
|
||||
interface SyncJobData {
|
||||
syncType: 'PRODUCT' | 'ORDER' | 'INVENTORY';
|
||||
platformConfig: PlatformApiConfig;
|
||||
syncOptions: SyncOptions;
|
||||
retryCount?: number;
|
||||
}
|
||||
|
||||
export class PlatformSyncWorker {
|
||||
private static readonly QUEUE_NAME = 'platform-sync-tasks';
|
||||
private static readonly MAX_RETRIES = 3;
|
||||
private static readonly CONCURRENCY = 10; // 符合资源限制
|
||||
|
||||
/**
|
||||
* 初始化并注册Worker
|
||||
*/
|
||||
static init() {
|
||||
WorkerHub.registerWorker(
|
||||
this.QUEUE_NAME,
|
||||
async (job: Job<SyncJobData>) => {
|
||||
const { syncType, platformConfig, syncOptions, retryCount = 0 } = job.data;
|
||||
const { tenantId, shopId, taskId, traceId, businessType } = syncOptions;
|
||||
|
||||
logger.info(`[PlatformSyncWorker] Starting ${syncType} sync task`, {
|
||||
jobId: job.id,
|
||||
platform: platformConfig.platform,
|
||||
tenantId,
|
||||
shopId,
|
||||
taskId,
|
||||
traceId,
|
||||
businessType,
|
||||
retryCount,
|
||||
});
|
||||
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
let result: any;
|
||||
|
||||
switch (syncType) {
|
||||
case 'PRODUCT':
|
||||
result = await this.syncProducts(platformConfig, syncOptions);
|
||||
break;
|
||||
case 'ORDER':
|
||||
result = await this.syncOrders(platformConfig, syncOptions);
|
||||
break;
|
||||
case 'INVENTORY':
|
||||
result = await this.syncInventory(platformConfig, syncOptions);
|
||||
break;
|
||||
default:
|
||||
throw new Error(`Unknown sync type: ${syncType}`);
|
||||
}
|
||||
|
||||
const duration = Date.now() - startTime;
|
||||
|
||||
// 审计日志 - 成功
|
||||
await AuditService.log({
|
||||
tenantId,
|
||||
shopId,
|
||||
taskId,
|
||||
traceId,
|
||||
businessType,
|
||||
module: 'PLATFORM_SYNC',
|
||||
action: `${syncType}_SYNC_SUCCESS`,
|
||||
resourceType: 'sync_job',
|
||||
resourceId: String(job.id),
|
||||
afterSnapshot: {
|
||||
platform: platformConfig.platform,
|
||||
syncType,
|
||||
duration,
|
||||
resultCount: result?.length || 0,
|
||||
},
|
||||
result: 'success',
|
||||
source: 'node',
|
||||
});
|
||||
|
||||
logger.info(`[PlatformSyncWorker] ${syncType} sync completed`, {
|
||||
jobId: job.id,
|
||||
duration,
|
||||
resultCount: result?.length || 0,
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
syncType,
|
||||
platform: platformConfig.platform,
|
||||
duration,
|
||||
resultCount: result?.length || 0,
|
||||
data: result,
|
||||
};
|
||||
} catch (error: any) {
|
||||
const duration = Date.now() - startTime;
|
||||
|
||||
logger.error(`[PlatformSyncWorker] ${syncType} sync failed`, {
|
||||
jobId: job.id,
|
||||
platform: platformConfig.platform,
|
||||
error: error.message,
|
||||
duration,
|
||||
retryCount,
|
||||
});
|
||||
|
||||
// 审计日志 - 失败
|
||||
await AuditService.log({
|
||||
tenantId,
|
||||
shopId,
|
||||
taskId,
|
||||
traceId,
|
||||
businessType,
|
||||
module: 'PLATFORM_SYNC',
|
||||
action: `${syncType}_SYNC_FAILED`,
|
||||
resourceType: 'sync_job',
|
||||
resourceId: String(job.id),
|
||||
result: 'failed',
|
||||
errorCode: 'PLATFORM_SYNC_ERROR',
|
||||
errorMessage: error.message,
|
||||
source: 'node',
|
||||
});
|
||||
|
||||
// 重试逻辑
|
||||
if (retryCount < this.MAX_RETRIES) {
|
||||
logger.info(`[PlatformSyncWorker] Retrying task ${job.id}`, {
|
||||
retryCount: retryCount + 1,
|
||||
});
|
||||
throw error; // 抛出错误触发BullMQ重试
|
||||
}
|
||||
|
||||
// 超过重试次数，返回失败结果
|
||||
return {
|
||||
success: false,
|
||||
syncType,
|
||||
platform: platformConfig.platform,
|
||||
duration,
|
||||
error: error.message,
|
||||
retryCount,
|
||||
};
|
||||
}
|
||||
},
|
||||
this.CONCURRENCY
|
||||
);
|
||||
|
||||
logger.info('[PlatformSyncWorker] Worker registered successfully');
|
||||
}
|
||||
|
||||
// ==================== Private Methods ====================
|
||||
|
||||
/**
|
||||
* 同步商品数据
|
||||
*/
|
||||
private static async syncProducts(
|
||||
config: PlatformApiConfig,
|
||||
options: SyncOptions
|
||||
): Promise<any[]> {
|
||||
const products = await PlatformApiService.syncProducts(config, options);
|
||||
|
||||
// 保存到数据库
|
||||
const savedProducts = [];
|
||||
for (const product of products) {
|
||||
try {
|
||||
const productId = await ProductService.create({
|
||||
tenantId: options.tenantId,
|
||||
shopId: options.shopId,
|
||||
platform: config.platform,
|
||||
platformProductId: product.platformProductId,
|
||||
title: product.title,
|
||||
price: product.price,
|
||||
currency: product.currency,
|
||||
status: product.status,
|
||||
traceId: options.traceId,
|
||||
businessType: options.businessType,
|
||||
syncedAt: product.syncedAt,
|
||||
});
|
||||
savedProducts.push({ ...product, internalId: productId });
|
||||
} catch (error: any) {
|
||||
logger.error('[PlatformSyncWorker] Failed to save product', {
|
||||
platformProductId: product.platformProductId,
|
||||
error: error.message,
|
||||
});
|
||||
// 继续处理其他商品
|
||||
}
|
||||
}
|
||||
|
||||
return savedProducts;
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步订单数据
|
||||
*/
|
||||
private static async syncOrders(
|
||||
config: PlatformApiConfig,
|
||||
options: SyncOptions
|
||||
): Promise<any[]> {
|
||||
const orders = await PlatformApiService.syncOrders(config, options);
|
||||
|
||||
// 保存到数据库
|
||||
const savedOrders = [];
|
||||
for (const order of orders) {
|
||||
try {
|
||||
const orderId = await OrderService.create({
|
||||
tenantId: options.tenantId,
|
||||
shopId: options.shopId,
|
||||
platform: config.platform,
|
||||
platformOrderId: order.platformOrderId,
|
||||
status: order.status,
|
||||
totalAmount: order.totalAmount,
|
||||
currency: order.currency,
|
||||
items: order.items,
|
||||
traceId: options.traceId,
|
||||
taskId: options.taskId,
|
||||
businessType: options.businessType,
|
||||
syncedAt: order.syncedAt,
|
||||
});
|
||||
savedOrders.push({ ...order, internalId: orderId });
|
||||
} catch (error: any) {
|
||||
logger.error('[PlatformSyncWorker] Failed to save order', {
|
||||
platformOrderId: order.platformOrderId,
|
||||
error: error.message,
|
||||
});
|
||||
// 继续处理其他订单
|
||||
}
|
||||
}
|
||||
|
||||
return savedOrders;
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步库存数据
|
||||
*/
|
||||
private static async syncInventory(
|
||||
config: PlatformApiConfig,
|
||||
options: SyncOptions
|
||||
): Promise<any[]> {
|
||||
// 先同步商品获取最新库存
|
||||
const products = await PlatformApiService.syncProducts(config, options);
|
||||
|
||||
// 更新库存
|
||||
const updatedInventory = [];
|
||||
for (const product of products) {
|
||||
try {
|
||||
// TODO: 调用库存服务更新库存
|
||||
// await InventoryService.updateStock(...);
|
||||
updatedInventory.push({
|
||||
platformProductId: product.platformProductId,
|
||||
stock: product.stock,
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
} catch (error: any) {
|
||||
logger.error('[PlatformSyncWorker] Failed to update inventory', {
|
||||
platformProductId: product.platformProductId,
|
||||
error: error.message,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return updatedInventory;
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交同步任务
|
||||
*/
|
||||
static async submitSyncTask(
|
||||
syncType: 'PRODUCT' | 'ORDER' | 'INVENTORY',
|
||||
platformConfig: PlatformApiConfig,
|
||||
syncOptions: SyncOptions
|
||||
): Promise<string> {
|
||||
const job = await WorkerHub.addJob(this.QUEUE_NAME, {
|
||||
syncType,
|
||||
platformConfig,
|
||||
syncOptions,
|
||||
retryCount: 0,
|
||||
}, {
|
||||
attempts: this.MAX_RETRIES,
|
||||
backoff: {
|
||||
type: 'exponential',
|
||||
delay: 5000, // 5秒初始延迟
|
||||
},
|
||||
});
|
||||
|
||||
logger.info(`[PlatformSyncWorker] Sync task submitted`, {
|
||||
jobId: job.id,
|
||||
syncType,
|
||||
platform: platformConfig.platform,
|
||||
});
|
||||
|
||||
return job.id as string;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user