feat(wordcloud): WordCloudJob 模型/迁移 + 词云生成与轮询适配器

2026-08-12 19:08:57 +08:00
parent d47b62c337
commit 8925e83b8f
9 changed files with 1076 additions and 15 deletions
@@ -35,6 +35,7 @@
    "class-transformer": "^0.5.1",
    "class-validator": "^0.14.1",
    "dayjs": "^1.11.0",
+    "exceljs": "^4.4.0",
    "ioredis": "^5.4.0",
    "joi": "^17.13.0",
    "passport": "^0.7.0",
@@ -46,6 +47,7 @@
    "@nestjs/cli": "^11.0.0",
    "@nestjs/schematics": "^11.0.0",
    "@types/express": "^5.0.0",
+    "@types/multer": "^2.2.0",
    "@types/node": "^22.0.0",
    "@types/passport-jwt": "^4.0.1",
    "eslint": "^9.0.0",
@@ -0,0 +1,22 @@
+-- CreateEnum
+CREATE TYPE "WordCloudJobStatus" AS ENUM ('QUEUED', 'RUNNING', 'SUCCESS', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "WordCloudJob" (
+    "id" TEXT NOT NULL,
+    "userId" TEXT NOT NULL,
+    "status" "WordCloudJobStatus" NOT NULL DEFAULT 'QUEUED',
+    "progress" INTEGER NOT NULL DEFAULT 0,
+    "imageUrl" TEXT,
+    "error" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "WordCloudJob_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "WordCloudJob_userId_idx" ON "WordCloudJob"("userId");
+
+-- AddForeignKey
+ALTER TABLE "WordCloudJob" ADD CONSTRAINT "WordCloudJob_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "WordCloudJob" ADD COLUMN     "remoteJobId" TEXT;
@@ -29,6 +29,7 @@ model User {
  designLists    DesignList[]
  uploads        Upload[]
  customizations CustomizationTask[]
+  wordCloudJobs  WordCloudJob[]

  @@index([phone])
 }
@@ -221,3 +222,28 @@ enum CustomizationTaskStatus {
  SUCCESS
  FAILED
 }
+
+// ── 词云任务（R4：小程序交互生成 + 下单 WCD 派单共用）────────────────
+model WordCloudJob {
+  id           String             @id @default(cuid())
+  userId       String
+  // wordcloud 侧的外部 job_id（POST /api/jobs 返回）；空表示尚未在外部创建
+  remoteJobId  String?
+  status       WordCloudJobStatus @default(QUEUED)
+  progress     Int                @default(0)
+  imageUrl     String?
+  error        String?
+  user         User               @relation(fields: [userId], references: [id])
+
+  createdAt DateTime             @default(now())
+  updatedAt DateTime             @updatedAt
+
+  @@index([userId])
+}
+
+enum WordCloudJobStatus {
+  QUEUED
+  RUNNING
+  SUCCESS
+  FAILED
+}
@@ -16,6 +16,7 @@ import { AddressesModule } from './addresses/addresses.module';
 import { OrdersModule } from './orders/orders.module';
 import { PaymentsModule } from './payments/payments.module';
 import { UploadModule } from './upload/upload.module';
+import { WordCloudModule } from './wordcloud/wordcloud.module';
 import { QueueModule } from './queue/queue.module';
 import { HealthModule } from './health/health.module';

@@ -40,6 +41,7 @@ import { HealthModule } from './health/health.module';
    OrdersModule,
    PaymentsModule,
    UploadModule,
+    WordCloudModule,
    QueueModule,
    HealthModule,
  ],
@@ -0,0 +1,52 @@
+import {
+  Body,
+  Controller,
+  Get,
+  HttpException,
+  HttpStatus,
+  Param,
+  Post,
+  UploadedFile,
+  UseInterceptors,
+} from '@nestjs/common';
+import { FileInterceptor } from '@nestjs/platform-express';
+import { ApiBearerAuth, ApiConsumes, ApiOperation, ApiTags } from '@nestjs/swagger';
+import { CurrentUser, JwtPayload } from '../common/decorators/current-user.decorator';
+import { WordCloudService } from './wordcloud.service';
+
+@ApiTags('词云')
+@ApiBearerAuth()
+@Controller('wordcloud')
+export class WordCloudController {
+  constructor(private readonly wordCloudService: WordCloudService) {}
+
+  /** 创建词云任务：multipart(image) + form 字段 names/params */
+  @Post('generate')
+  @ApiOperation({ summary: '创建词云任务（底图 + 名单文本）' })
+  @ApiConsumes('multipart/form-data')
+  @UseInterceptors(FileInterceptor('image'))
+  async generate(
+    @CurrentUser() user: JwtPayload,
+    @UploadedFile() image: Express.Multer.File | undefined,
+    @Body('names') names: string,
+    @Body('params') params?: string,
+  ) {
+    if (!image) {
+      throw new HttpException('缺少底图', HttpStatus.BAD_REQUEST);
+    }
+    const jobId = await this.wordCloudService.createGenerateJob(
+      user.sub,
+      { buffer: image.buffer, originalname: image.originalname },
+      names ?? '',
+      params,
+    );
+    return { jobId };
+  }
+
+  /** 轮询词云任务（仅本人） */
+  @Get('jobs/:id')
+  @ApiOperation({ summary: '轮询词云任务状态（仅本人）' })
+  async job(@CurrentUser() user: JwtPayload, @Param('id') id: string) {
+    return this.wordCloudService.getUserJob(user.sub, id);
+  }
+}
@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { WordCloudController } from './wordcloud.controller';
+import { WordCloudService } from './wordcloud.service';
+
+@Module({
+  controllers: [WordCloudController],
+  providers: [WordCloudService],
+})
+export class WordCloudModule {}
@@ -0,0 +1,193 @@
+import { HttpException, HttpStatus, Injectable, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import axios from 'axios';
+import ExcelJS from 'exceljs';
+import { PrismaService } from '../prisma/prisma.service';
+
+type RemoteStatus = 'queued' | 'running' | 'success' | 'failed';
+
+/** Buffer → 独立的 ArrayBuffer（用于 Blob 构造，规避 Buffer<ArrayBufferLike> 类型不兼容） */
+function toArrayBuffer(buf: Buffer): ArrayBuffer {
+  return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength) as ArrayBuffer;
+}
+
+/**
+ * wordcloud 平台适配器。
+ * 契约：与 wordcloud 之间的接口见 docs/wordcloud-contract.md；对小程序暴露的接口见
+ * docs/api-contract-v1.md §8。
+ * 职责：把小程序的名字文本转成 .xlsx → 代理调 wordcloud POST /api/jobs → 落 WordCloudJob →
+ * 轮询 wordcloud GET /api/jobs/{id} → 回写状态/产物。
+ */
+@Injectable()
+export class WordCloudService {
+  private readonly logger = new Logger(WordCloudService.name);
+
+  constructor(
+    private readonly config: ConfigService,
+    private readonly prisma: PrismaService,
+  ) {}
+
+  /** 词云平台是否已配置（WORDCLOUD_API_URL 非空） */
+  get configured(): boolean {
+    return !!this.config.get<string>('wordcloud.apiUrl');
+  }
+
+  private get apiUrl(): string {
+    return (this.config.get<string>('wordcloud.apiUrl') ?? '').replace(/\/+$/, '');
+  }
+
+  private get timeoutMs(): number {
+    return this.config.get<number>('wordcloud.timeoutMs') ?? 30000;
+  }
+
+  /** 名单文本 → 单列 .xlsx（wordcloud 按 DATA_COL_INDEX 读名字列，A 列 index=0） */
+  private async buildNamesXlsx(names: string[]): Promise<Buffer> {
+    const workbook = new ExcelJS.Workbook();
+    const sheet = workbook.addWorksheet('names');
+    names.forEach((n) => sheet.addRow([{ text: n }]));
+    return Buffer.from(await workbook.xlsx.writeBuffer());
+  }
+
+  /** 代理创建 wordcloud 任务，返回外部 job_id */
+  private async createRemoteJob(
+    xlsx: Buffer,
+    image: { buffer: Buffer; originalname: string },
+    params: Record<string, unknown>,
+  ): Promise<string> {
+    const form = new FormData();
+    form.append('name_list', new Blob([toArrayBuffer(xlsx)]), 'names.xlsx');
+    form.append('mask_image', new Blob([toArrayBuffer(image.buffer)]), image.originalname || 'mask.png');
+    form.append('params', JSON.stringify(params));
+
+    let res;
+    try {
+      res = await axios.post(`${this.apiUrl}/api/jobs`, form, {
+        timeout: this.timeoutMs,
+        maxBodyLength: Infinity,
+        maxContentLength: Infinity,
+      });
+    } catch (e) {
+      this.logger.error(`创建词云任务失败: ${(e as Error).message}`);
+      throw new HttpException('词云平台暂不可用', HttpStatus.SERVICE_UNAVAILABLE);
+    }
+    const jobId = res.data?.job_id;
+    if (!jobId) {
+      throw new HttpException('词云平台未返回任务ID', HttpStatus.BAD_GATEWAY);
+    }
+    return jobId;
+  }
+
+  /**
+   * 交互式生成：小程序上传底图 + 名单文本 → 在 wordcloud 创建任务并落库，返回内部 jobId。
+   */
+  async createGenerateJob(
+    userId: string,
+    image: { buffer: Buffer; originalname: string },
+    namesText: string,
+    paramsJson = '{}',
+  ): Promise<string> {
+    if (!this.configured) {
+      throw new HttpException('词云平台未配置', HttpStatus.SERVICE_UNAVAILABLE);
+    }
+    if (!image?.buffer) {
+      throw new HttpException('缺少底图', HttpStatus.BAD_REQUEST);
+    }
+    const names = namesText
+      .split(/[\n,]|，/)
+      .map((s) => s.trim())
+      .filter(Boolean);
+    if (!names.length) {
+      throw new HttpException('名单不能为空', HttpStatus.BAD_REQUEST);
+    }
+
+    let userParams: Record<string, unknown> = {};
+    if (paramsJson) {
+      try {
+        userParams = JSON.parse(paramsJson);
+      } catch {
+        throw new HttpException('params 必须是合法 JSON', HttpStatus.BAD_REQUEST);
+      }
+    }
+
+    // 名单写在 xlsx A 列，DATA_COL_INDEX 强制对齐到 0
+    const params: Record<string, unknown> = {
+      MODE: 'IMAGE',
+      DATA_COL_INDEX: 0,
+      SEED: 42,
+      N_REPETITIONS: 20,
+      ENABLE_STROKE_WEIGHTS: false,
+      FONT_COLOR: '#000000',
+      ...userParams,
+    };
+    params.DATA_COL_INDEX = 0;
+
+    const xlsx = await this.buildNamesXlsx(names);
+    const remoteJobId = await this.createRemoteJob(xlsx, image, params);
+
+    const job = await this.prisma.wordCloudJob.create({
+      data: { userId, remoteJobId, status: 'QUEUED' },
+    });
+    return job.id;
+  }
+
+  private mapStatus(s?: string): RemoteStatus {
+    if (s === 'running') return 'running';
+    if (s === 'success') return 'success';
+    if (s === 'failed') return 'failed';
+    return 'queued';
+  }
+
+  /** 查询本人词云任务：归属校验 → 代理轮询 wordcloud → 回写 DB → 返回小程序侧模型 */
+  async getUserJob(userId: string, id: string) {
+    const job = await this.prisma.wordCloudJob.findFirst({ where: { id, userId } });
+    if (!job) {
+      throw new HttpException('任务不存在', HttpStatus.NOT_FOUND);
+    }
+    if (!job.remoteJobId) {
+      throw new HttpException('任务尚未创建', HttpStatus.CONFLICT);
+    }
+
+    let remote: Record<string, unknown>;
+    try {
+      const res = await axios.get(`${this.apiUrl}/api/jobs/${job.remoteJobId}`, {
+        timeout: this.timeoutMs,
+      });
+      remote = res.data;
+    } catch (e) {
+      this.logger.warn(`轮询词云任务失败: ${(e as Error).message}`);
+      throw new HttpException('词云平台暂不可用', HttpStatus.SERVICE_UNAVAILABLE);
+    }
+
+    const status = this.mapStatus(remote.status as string | undefined);
+    const progress =
+      typeof remote.progress_percent === 'number' ? remote.progress_percent : job.progress;
+
+    // success 时尝试解析结果图（当前 COS 写入未实现，P1 接入后返回公网 URL）
+    const imageUrl = status === 'success' ? await this.resolveResultImage(job.remoteJobId) : null;
+
+    const updated = await this.prisma.wordCloudJob.update({
+      where: { id: job.id },
+      data: {
+        status: status.toUpperCase() as 'QUEUED' | 'RUNNING' | 'SUCCESS' | 'FAILED',
+        progress,
+        imageUrl,
+        error: (remote.error as string | null) ?? null,
+      },
+    });
+
+    return {
+      id: updated.id,
+      status,
+      progress,
+      imageUrl: updated.imageUrl ?? undefined,
+      error: updated.error ?? undefined,
+    };
+  }
+
+  /** 下载 wordcloud 结果 PNG 并返回小程序可用地址（COS 转存前置，P1 实现） */
+  private async resolveResultImage(_remoteJobId: string): Promise<string | null> {
+    // TODO(P1 COS)：GET /api/jobs/{id}/result → 下载 png → 写入 COS → 返回公网 URL。
+    // 当前 COS 写入尚未实现，结果图地址留待 P1 接入。
+    return null;
+  }
+}