All files / modules/book-generator book-queue.processor.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
/**
 * 书籍生成 - 数据库驱动（替代队列）
 *
 * 之前使用 Bull/Redis + 内存队列，但队列在开发环境下不可靠。
 * 改为完全靠数据库驱动：
 *   1. genStage 字段直接追踪进度
 *   2. 服务启动时自动恢复中断的任务
 *   3. 生成异步执行，不阻塞请求
 */
 
import { bookStore } from './book-generator.store';
import { prisma } from '../../models';
import { memoryQueue } from '../../services/memory-queue';
 
/**
 * 服务启动时恢复中断的生成任务
 *
 * 原则：
 *  1. 已有内容产出的书籍不重新生成（保护已有成果）
 *  2. 只恢复真正中断在早期阶段的书籍（无大纲/无内容）
 *  3. 音频遗漏由 AudioScanner 兜底，不需要重跑整个生成流程
 */
export async function resumeInterruptedTasks() {
  console.log('[DBQueue] 扫描中断的生成任务...');
 
  try {
    // 只恢复早期阶段：outlining（无大纲）、outline_completed（有大纲但无内容开始）
    // 跳过 content_completed / audio_generating / audio_completed / video_generating
    // 这些阶段的书籍已有产出，音频遗漏由 AudioScanner 兜底
    const earlyStages = ['outlining', 'outline_completed'];
 
    const interruptedBooks = await prisma.book.findMany({
      where: { genStage: { in: earlyStages } },
      select: { id: true, title: true, description: true, bookScale: true, genStage: true },
    });
 
    if (interruptedBooks.length === 0) {
      console.log('[DBQueue] 没有需要恢复的早期中断任务');
      return;
    }
 
    // 进一步过滤：如果书籍已有章节（说明之前生成过部分内容），也跳过
    const booksToResume: typeof interruptedBooks = [];
    for (const book of interruptedBooks) {
      const chapterCount = await prisma.bookChapter.count({ where: { bookId: book.id } });
      if (chapterCount === 0) {
        booksToResume.push(book);
        console.log(`[DBQueue] 需恢复: 《${book.title}》 (bookId=${book.id}, stage=${book.genStage}, 章节数=0)`);
      } else {
        console.log(`[DBQueue] ⏭️ 跳过: 《${book.title}》 (bookId=${book.id}, stage=${book.genStage}, 已有${chapterCount}个章节，不破坏已有成果)`);
      }
    }
 
    if (booksToResume.length === 0) {
      console.log('[DBQueue] 所有早期中断书籍都已有章节产出，跳过恢复（音频由 AudioScanner 兜底）');
      return;
    }
 
    console.log(`[DBQueue] 恢复 ${booksToResume.length} 个真正中断的任务...`);
 
    const { langGraphGenerator } = await import('./index.js');
    for (const book of booksToResume) {
      console.log(`[DBQueue] 恢复: 《${book.title}》 (bookId=${book.id})`);
      langGraphGenerator.generate(
        book.id.toString(),
        book.description || book.title,
        book.bookScale || '1000'
      ).then(() => {
        console.log(`[DBQueue] 恢复完成: bookId=${book.id}`);
      }).catch(err => {
        console.error(`[DBQueue] 恢复失败: bookId=${book.id}`, err);
      });
    }
  } catch (error) {
    console.error('[DBQueue] 扫描恢复失败:', error);
  }
}
 
/**
 * 初始化（保持兼容，不再需要队列处理器注册）
 */
export function initBookGenerationQueue() {
  console.log('[DBQueue] 已启用数据库驱动模式（跳过队列）');
}