XXL-JOB部署及用任务驱动文档处理流程

### 调度中心JDBC链接
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=root_pwd
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
### 报警邮箱
spring.mail.host=smtp.qq.com
spring.mail.port=25
spring.mail.username=xxx@qq.com
spring.mail.password=xxx
spring.mail.properties.mail.smtp.auth=true
spring.mail.properties.mail.smtp.starttls.enable=true
spring.mail.properties.mail.smtp.starttls.required=true
spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory
### 调度中心通讯TOKEN [选填]：非空时启用；
xxl.job.accessToken=
### 调度中心国际化配置 [必填]： 默认为 "zh_CN"/中文简体, 可选范围为 "zh_CN"/中文简体, "zh_TC"/中文繁体 and "en"/英文；
xxl.job.i18n=zh_CN
## 调度线程池最大线程配置【必填】
xxl.job.triggerpool.fast.max=200
xxl.job.triggerpool.slow.max=100
### 调度中心日志表数据保存天数 [必填]：过期日志自动清理；限制大于等于7时生效，否则, 如-1，关闭自动清理功能；
xxl.job.logretentiondays=30

/**
 * 文档分段补偿任务
 * 扫描 CONVERTED 状态超过阈值的文档，重新触发分段
 */
@XxlJob("documentSplitCompensation")
public void documentSplitCompensation() {
    log.info("========== 开始执行文档分段补偿任务 ==========");
    int successCount = 0;
    int failCount = 0;

    try {
        // 查询 CONVERTED 状态的文档
        // todo 注：实际项目中应该在实体和数据库中添加 updateTime 和 retryCount 字段
        // 这里简化处理，查询所有 CONVERTED 状态的文档
        LambdaQueryWrapper<KnowledgeDocument> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(KnowledgeDocument::getStatus, DocumentStatus.CONVERTED);
        queryWrapper.isNotNull(KnowledgeDocument::getConvertedDocUrl);

        List<KnowledgeDocument> documents = knowledgeDocumentService.list(queryWrapper);
        log.info("发现 {} 个待补偿的 CONVERTED 状态文档", documents.size());

        for (KnowledgeDocument document : documents) {
            try {
                // 检查重试次数（从 extension 字段解析，或使用默认值）
                int retryCount = getRetryCount(document);
                if (retryCount >= MAX_RETRY_COUNT) {
                    log.warn("文档 {} 已达最大重试次数 {}，跳过补偿", document.getDocId(), retryCount);
                    continue;
                }

                log.info("补偿处理文档分段，documentId: {}, retryCount: {}", document.getDocId(), retryCount);

                // 执行分段
                //todo 多余查询?
                int segmentCount = documentProcessService.splitDocument(document.getDocId());

                // 更新重试次数
                updateRetryCount(document.getDocId(), retryCount + 1);

                log.info("文档分段补偿成功，documentId: {}, segmentCount: {}", document.getDocId(), segmentCount);
                successCount++;
            } catch (Exception e) {
                log.error("文档分段补偿失败，documentId: {}", document.getDocId(), e);
                failCount++;
            }
        }
    } catch (Exception e) {
        log.error("文档分段补偿任务执行异常", e);
    }

    log.info("========== 文档分段补偿任务完成，成功: {}，失败: {} ==========", successCount, failCount);
}

/**
 * 向量化补偿任务
 * 扫描 CHUNKED 状态但存在未向量化的 segment，重新触发向量化
 */
@XxlJob("documentEmbeddingCompensation")
public void documentEmbeddingCompensation() {
    log.info("========== 开始执行向量化补偿任务 ==========");
    int successCount = 0;
    int failCount = 0;

    try {
        // 查询 CHUNKED 状态的文档
        //todo 扫表注意索引问题
        LambdaQueryWrapper<KnowledgeDocument> docQueryWrapper = new LambdaQueryWrapper<>();
        docQueryWrapper.eq(KnowledgeDocument::getStatus, DocumentStatus.CHUNKED);

        List<KnowledgeDocument> documents = knowledgeDocumentService.list(docQueryWrapper);
        log.info("发现 {} 个 CHUNKED 状态的文档", documents.size());

        for (KnowledgeDocument document : documents) {
            try {
                // 检查是否存在未向量化的片段
                LambdaQueryWrapper<KnowledgeSegment> segmentQueryWrapper = new LambdaQueryWrapper<>();
                segmentQueryWrapper.eq(KnowledgeSegment::getDocumentId, document.getDocId());
                segmentQueryWrapper.eq(KnowledgeSegment::getStatus, SegmentStatus.INIT);
                segmentQueryWrapper.eq(KnowledgeSegment::getSkipEmbedding, 0);
                segmentQueryWrapper.isNull(KnowledgeSegment::getEmbeddingId);

                //todo 这段逻辑可以内置到embeddingAndStore
                long unembeddedCount = knowledgeSegmentService.count(segmentQueryWrapper);
                if (unembeddedCount == 0) {
                    // 没有未向量化的片段，检查是否所有片段都已处理
                    // 如果所有片段都已向量化，更新文档状态
                    checkAndUpdateDocumentStatus(document.getDocId());
                    continue;
                }

                // 检查重试次数
                int retryCount = getRetryCount(document);
                if (retryCount >= MAX_RETRY_COUNT) {
                    log.warn("文档 {} 已达最大重试次数 {}，跳过补偿", document.getDocId(), retryCount);
                    continue;
                }

                log.info("补偿处理向量化，documentId: {}, unembeddedCount: {}, retryCount: {}",
                        document.getDocId(), unembeddedCount, retryCount);

                // 执行向量化
                boolean success = documentProcessService.embeddingAndStore(document.getDocId());

                if (success) {
                    // 更新重试次数
                    updateRetryCount(document.getDocId(), retryCount + 1);
                    log.info("向量化补偿成功，documentId: {}", document.getDocId());
                    successCount++;
                } else {
                    log.warn("向量化补偿失败，documentId: {}", document.getDocId());
                    failCount++;
                }
            } catch (Exception e) {
                log.error("向量化补偿失败，documentId: {}", document.getDocId(), e);
                failCount++;
            }
        }
    } catch (Exception e) {
        log.error("向量化补偿任务执行异常", e);
    }

    log.info("========== 向量化补偿任务完成，成功: {}，失败: {} ==========", successCount, failCount);
}