diff --git a/applications/ai-model/src/tasks/AISummarize.ts b/applications/ai-model/src/tasks/AISummarize.ts index aed0bdf0..9199aced 100644 --- a/applications/ai-model/src/tasks/AISummarize.ts +++ b/applications/ai-model/src/tasks/AISummarize.ts @@ -1,4 +1,6 @@ import "reflect-metadata"; +import type { SessionDigestCoverage } from "@root/common/services/database/AgcDbAccessService"; + import { injectable, inject } from "tsyringe"; import { agendaInstance } from "@root/common/scheduler/agenda"; import { TaskHandlerTypes, TaskParameters } from "@root/common/scheduler/@types/Tasks"; @@ -19,6 +21,15 @@ import { PooledTaskResult } from "../services/generators/text/PooledTextGeneratorService"; +const MIN_SUMMARY_MESSAGE_COUNT = 10; + +interface TaskContext { + groupId: string; + sessionId: string; + latestMessageTimestamp: number; + messageCount: number; +} + /** * AI 摘要任务处理器 * 负责对群聊消息进行 AI 摘要生成 @@ -65,14 +76,9 @@ export class AISummarizeTaskHandler { await ctxBuilder.init(); - // 任务上下文类型定义 - interface TaskContext { - groupId: string; - sessionId: string; - } - // 收集所有需要处理的任务 const allTasks: PooledTask[] = []; + const activeSessionGraceMs = config.preprocessors.TimeoutSplitter.timeoutInMinutes * 60 * 1000; for (const groupId of attrs.groupIds) { /* 1. 获取指定时间范围内的消息 */ @@ -102,44 +108,104 @@ export class AISummarizeTaskHandler { for (const msg of msgs) { const { sessionId } = msg; - // 如果 sessionId 已经被生成过摘要，跳过 - if (!(await this.agcDbAccessService.isSessionIdSummarized(sessionId))) { - if (!sessions[sessionId]) { - sessions[sessionId] = []; - } - sessions[sessionId].push(msg); + if (!sessions[sessionId]) { + sessions[sessionId] = []; } + sessions[sessionId].push(msg); } if (Object.keys(sessions).length === 0) { this.LOGGER.info(`群 ${groupId} 在指定时间范围内无消息，跳过`); continue; } - // 考虑到最后一个session可能正在发生，还没有闭合，因此需要删掉 + + // 最新 session 在静默时间不足时先跳过，避免把仍在发生的对话截断。 + // 网络恢复补跑需要尽快追上断网期间遗漏的消息，因此允许调用方显式跳过这层保护。 const newestSessionId = msgs[msgs.length - 1].sessionId; + const newestSessionMessages = sessions[newestSessionId]; - delete sessions[newestSessionId]; - this.LOGGER.debug(`删掉了最后一个sessionId为 ${newestSessionId} 的session`); - this.LOGGER.info(`分组完成，共 ${Object.keys(sessions).length} 个需要处理的session`); + if (newestSessionMessages) { + const newestSessionLatestTimestamp = + newestSessionMessages[newestSessionMessages.length - 1].timestamp; + const idleTime = attrs.endTimeStamp - newestSessionLatestTimestamp; - // 3. 删掉消息量不够的session + if (attrs.ignoreActiveSessionGrace) { + this.LOGGER.info( + `已启用网络恢复补跑模式，最新 session ${newestSessionId} 即使静默时间不足也纳入摘要` + ); + } else if (idleTime < activeSessionGraceMs) { + delete sessions[newestSessionId]; + this.LOGGER.debug( + `最新 session ${newestSessionId} 静默时间不足 ${Math.ceil(activeSessionGraceMs / 60000)} 分钟，暂不摘要` + ); + } else { + this.LOGGER.debug(`最新 session ${newestSessionId} 已静默足够久，纳入摘要`); + } + } + this.LOGGER.info(`分组完成，共 ${Object.keys(sessions).length} 个候选 session`); + + const sessionMessagesToSummarize: Record = {}; + const sessionDigestMetadata: Record< + string, + { + latestMessageTimestamp: number; + messageCount: number; + } + > = {}; + + // 3. 过滤掉摘要已经覆盖的 session，并只保留新增消息片段 for (const sessionId in sessions) { - if (sessions[sessionId].length <= 10) { + const fullSessionMessages = + await this.imDbAccessService.getProcessedChatMessagesBySessionId(sessionId); + + if (fullSessionMessages.length === 0) { + this.LOGGER.warning(`session ${sessionId} 未找到完整消息，跳过`); + continue; + } + + const latestMessageTimestamp = + fullSessionMessages[fullSessionMessages.length - 1].timestamp; + const messageCount = fullSessionMessages.length; + + if ( + await this.agcDbAccessService.isSessionDigestFresh( + sessionId, + latestMessageTimestamp, + messageCount + ) + ) { + this.LOGGER.info(`session ${sessionId} 已经摘要到最新消息，跳过`); + continue; + } + + const coverage = await this.agcDbAccessService.getSessionDigestCoverage(sessionId); + const messagesToSummarize = this._getMessagesToSummarizeByCoverage( + fullSessionMessages, + coverage + ); + + if (messagesToSummarize.length <= MIN_SUMMARY_MESSAGE_COUNT) { this.LOGGER.warning( - `session ${sessionId} 消息数量不足，消息数量为${sessions[sessionId].length}，跳过` + `session ${sessionId} 新增可摘要消息数量不足，消息数量为 ${messagesToSummarize.length}，跳过` ); - delete sessions[sessionId]; + continue; } + + sessionMessagesToSummarize[sessionId] = messagesToSummarize; + sessionDigestMetadata[sessionId] = { + latestMessageTimestamp, + messageCount + }; } /* 4. 构建任务列表 */ - for (const sessionId in sessions) { + for (const sessionId in sessionMessagesToSummarize) { this.LOGGER.info( - `准备处理session ${sessionId} ，该session内共 ${sessions[sessionId].length} 条消息` + `准备处理 session ${sessionId}，本次新增可摘要消息共 ${sessionMessagesToSummarize[sessionId].length} 条` ); // 构建上下文 const ctx = await ctxBuilder.buildCtx( - sessions[sessionId], + sessionMessagesToSummarize[sessionId], config.groupConfigs[groupId].groupIntroduction ); @@ -148,13 +214,29 @@ export class AISummarizeTaskHandler { allTasks.push({ input: ctx, modelNames: config.groupConfigs[groupId].aiModels, - context: { groupId, sessionId }, + context: { + groupId, + sessionId, + latestMessageTimestamp: sessionDigestMetadata[sessionId].latestMessageTimestamp, + messageCount: sessionDigestMetadata[sessionId].messageCount + }, checkJsonFormat: true }); } } - this.LOGGER.info(`共收集到 ${allTasks.length} 个任务，开始并行处理（并行度=5）`); + this.LOGGER.info( + `共收集到 ${allTasks.length} 个任务，开始并行处理（并行度=${config.ai.maxConcurrentRequests}）` + ); + + if (allTasks.length === 0) { + this.LOGGER.info("没有需要生成摘要的 session，任务完成"); + pooledTextGeneratorService.dispose(); + ctxBuilder.dispose(); + this.LOGGER.success(`🥳任务完成: ${job.attrs.name}`); + + return; + } // 并行处理所有任务，每个任务完成时回调 let completedCount = 0; @@ -164,7 +246,7 @@ export class AISummarizeTaskHandler { async (result: PooledTaskResult) => { await job.touch(); // 保证任务存活 completedCount++; - const { sessionId } = result.context; + const { sessionId, latestMessageTimestamp, messageCount } = result.context; if (!result.isSuccess) { this.LOGGER.error( @@ -189,7 +271,12 @@ export class AISummarizeTaskHandler { this.LOGGER.warning( `session ${sessionId} 生成摘要长度过短，长度为 ${resultStr.length}，跳过` ); - console.log(resultStr); + + return; + } + + if (results.length === 0) { + this.LOGGER.warning(`session ${sessionId} 生成摘要为空，跳过`); return; } @@ -203,8 +290,15 @@ export class AISummarizeTaskHandler { Object.assign(resultItem, { updateTime: Date.now() }); } - // 存储摘要结果 - await this.agcDbAccessService.storeAIDigestResults(results as AIDigestResult[]); + // 存储摘要结果，并记录本次摘要覆盖到的消息范围 + await this.agcDbAccessService.storeAIDigestResultsWithSessionMetadata( + sessionId, + results as AIDigestResult[], + { + summarizedUntil: latestMessageTimestamp, + summarizedMessageCount: messageCount + } + ); this.LOGGER.success(`session ${sessionId} 存储摘要成功！`); } catch (error) { this.LOGGER.error( @@ -226,4 +320,31 @@ export class AISummarizeTaskHandler { } ); } + + /** + * 根据摘要覆盖范围切出本次需要摘要的消息 + * @param messages 完整 session 消息 + * @param coverage 摘要覆盖范围 + * @returns 本次需要摘要的消息 + */ + private _getMessagesToSummarizeByCoverage( + messages: ProcessedChatMessageWithRawMessage[], + coverage: SessionDigestCoverage | null + ): ProcessedChatMessageWithRawMessage[] { + if (!coverage) { + return messages; + } + + const messagesAfterTimestamp = messages.filter(msg => msg.timestamp > coverage.summarizedUntil); + + if (messagesAfterTimestamp.length > 0) { + return messagesAfterTimestamp; + } + + if (coverage.summarizedMessageCount !== null && coverage.summarizedMessageCount < messages.length) { + return messages.slice(coverage.summarizedMessageCount); + } + + return []; + } } diff --git a/applications/ai-model/src/tasks/InterestScore.ts b/applications/ai-model/src/tasks/InterestScore.ts index 85df0849..e89d472c 100644 --- a/applications/ai-model/src/tasks/InterestScore.ts +++ b/applications/ai-model/src/tasks/InterestScore.ts @@ -118,20 +118,36 @@ export class InterestScoreTaskHandler { }) ); + const batchSize = config.ai.embedding.batchSize; + const totalBatchCount = Math.ceil(filteredDigestResults.length / batchSize); + // 构建所有话题详情文本 const topics = filteredDigestResults.map( digestResult => `话题：${digestResult.topic} 正文内容：${digestResult.detail}` ); - // 批量获取所有话题的分数 - await job.touch(); // 保证任务存活 - const scores = await rater.scoreTopics(argArr, topics); + for (let i = 0; i < filteredDigestResults.length; i += batchSize) { + const currentBatchDigestResults = filteredDigestResults.slice(i, i + batchSize); + const currentBatchTopics = topics.slice(i, i + batchSize); + const currentBatchIndex = Math.floor(i / batchSize) + 1; + + this.LOGGER.info( + `处理兴趣度评分批次 ${currentBatchIndex}/${totalBatchCount}，当前批次共 ${currentBatchTopics.length} 条` + ); + + await job.touch(); + const scores = await rater.scoreTopics(argArr, currentBatchTopics); + + for (let j = 0; j < currentBatchDigestResults.length; j++) { + await this.interestScoreDbAccessService.storeInterestScoreResult( + currentBatchDigestResults[j].topicId, + scores[j] + ); + } - // 存储所有分数结果 - for (let i = 0; i < filteredDigestResults.length; i++) { - await this.interestScoreDbAccessService.storeInterestScoreResult( - filteredDigestResults[i].topicId, - scores[i] + await job.touch(); + this.LOGGER.info( + `兴趣度评分批次 ${currentBatchIndex}/${totalBatchCount} 已写入 ${currentBatchDigestResults.length} 条结果` ); } diff --git a/applications/data-provider/README.md b/applications/data-provider/README.md index a453df16..0eb9d57c 100644 --- a/applications/data-provider/README.md +++ b/applications/data-provider/README.md @@ -8,6 +8,7 @@ - **消息解析**：解析protobuf格式的消息内容 - **消息格式化**：将原始消息转换为结构化数据 - **数据提供**：通过统一的接口向其他服务提供消息数据 +- **群文件获取**：可选接入 OneBot/NapCat HTTP API，列出群文件并下载到本地目录 ## 技术栈 @@ -21,6 +22,7 @@ src/ ├── di/ # 依赖注入容器 ├── providers/ # 数据提供者实现 │ ├── contracts/ # 提供者接口定义 +│ ├── OneBotProvider/ # OneBot/NapCat 群文件提供者 │ └── QQProvider/ # QQ消息提供者 │ ├── parsers/ # protobuf消息解析器 │ ├── @types/ # 类型定义 @@ -36,6 +38,31 @@ QQ Provider负责从QQ数据库中读取消息并进行解析： - 处理消息正文、图片、语音等多种消息类型 - 使用protobuf解析消息内容 +## OneBot 文件 Provider 说明 + +OneBot 文件 Provider 通过 OneBot/NapCat HTTP API 获取群文件能力，适合后续构建群文件知识库： + +- 支持 `get_group_root_files`、`get_group_file_list`、`get_group_files` 多种文件列表接口 +- 支持 `get_group_file_url`、`get_file_url` 获取临时下载链接 +- 下载文件时按群号保存到本地目录，并清洗危险文件名字符 +- 该 Provider 不会自动接入当前聊天消息 Pipeline，需要由后续群文件同步任务显式调用 + +配置示例： + +```json +{ + "dataProviders": { + "OneBot": { + "enabled": true, + "baseURL": "http://127.0.0.1:3000", + "accessToken": "", + "downloadDirectory": "/path/to/group-files", + "requestTimeoutMs": 30000 + } + } +} +``` + ## 开发命令 ```bash diff --git a/applications/data-provider/src/di/container.ts b/applications/data-provider/src/di/container.ts index 486df440..d7e56400 100644 --- a/applications/data-provider/src/di/container.ts +++ b/applications/data-provider/src/di/container.ts @@ -7,6 +7,7 @@ import { container } from "tsyringe"; import { ProvideDataTaskHandler } from "../tasks/ProvideDataTask"; import { QQProvider } from "../providers/QQProvider/QQProvider"; +import { OneBotFileProvider } from "../providers/OneBotProvider/OneBotFileProvider"; import { DATA_PROVIDER_TOKENS } from "./tokens"; @@ -26,6 +27,22 @@ export function getQQProvider(): QQProvider { return container.resolve(DATA_PROVIDER_TOKENS.QQProvider); } +/** + * 注册 OneBotFileProvider 到 DI 容器 + */ +export function registerOneBotFileProvider(): void { + container.register(DATA_PROVIDER_TOKENS.OneBotFileProvider, { useClass: OneBotFileProvider }); +} + +/** + * 从 DI 容器获取 OneBotFileProvider 实例 + * 每次调用返回新实例（非单例） + * @returns OneBotFileProvider 实例 + */ +export function getOneBotFileProvider(): OneBotFileProvider { + return container.resolve(DATA_PROVIDER_TOKENS.OneBotFileProvider); +} + /** * 注册任务处理器到 DI 容器 */ diff --git a/applications/data-provider/src/di/tokens.ts b/applications/data-provider/src/di/tokens.ts index ddde191c..91cf92c2 100644 --- a/applications/data-provider/src/di/tokens.ts +++ b/applications/data-provider/src/di/tokens.ts @@ -12,5 +12,7 @@ export const DATA_PROVIDER_TOKENS = { /** 数据提供任务处理器 */ ProvideDataTaskHandler: Symbol.for("ProvideDataTaskHandler"), /** QQ 消息提供者 */ - QQProvider: Symbol.for("QQProvider") + QQProvider: Symbol.for("QQProvider"), + /** OneBot/NapCat 群文件提供者 */ + OneBotFileProvider: Symbol.for("OneBotFileProvider") } as const; diff --git a/applications/data-provider/src/index.ts b/applications/data-provider/src/index.ts index e66bbfea..7c48db0d 100644 --- a/applications/data-provider/src/index.ts +++ b/applications/data-provider/src/index.ts @@ -9,7 +9,12 @@ import { } from "@root/common/di/container"; import { bootstrap, bootstrapAll } from "@root/common/util/lifecycle/bootstrap"; -import { registerTaskHandlers, getProvideDataTaskHandler, registerQQProvider } from "./di/container"; +import { + registerTaskHandlers, + getProvideDataTaskHandler, + registerQQProvider, + registerOneBotFileProvider +} from "./di/container"; const LOGGER = Logger.withTag("🌏 data-provider-root-script"); @@ -38,6 +43,7 @@ class DataProviderApplication { // 4. 注册 QQProvider registerQQProvider(); + registerOneBotFileProvider(); // 5. 注册任务处理器 registerTaskHandlers(); diff --git a/applications/data-provider/src/providers/OneBotProvider/OneBotFileProvider.ts b/applications/data-provider/src/providers/OneBotProvider/OneBotFileProvider.ts new file mode 100644 index 00000000..ad0965c3 --- /dev/null +++ b/applications/data-provider/src/providers/OneBotProvider/OneBotFileProvider.ts @@ -0,0 +1,412 @@ +import "reflect-metadata"; +import { mkdir, writeFile } from "fs/promises"; +import path from "path"; + +import { injectable, inject } from "tsyringe"; +import { ConfigManagerService } from "@root/common/services/config/ConfigManagerService"; +import { DownloadedGroupFile, GroupFileInfo } from "@root/common/contracts/data-provider/index"; +import Logger from "@root/common/util/Logger"; +import { Disposable } from "@root/common/util/lifecycle/Disposable"; +import { mustInitBeforeUse } from "@root/common/util/lifecycle/mustInitBeforeUse"; + +import { COMMON_TOKENS } from "../../di/tokens"; +import { IGroupFileProvider } from "../contracts/IGroupFileProvider"; + +interface OneBotConfig { + enabled: boolean; + baseURL: string; + accessToken: string; + downloadDirectory: string; + requestTimeoutMs: number; +} + +interface OneBotResponse { + status?: string; + retcode?: number; + data?: T; + message?: string; + wording?: string; +} + +type OneBotFileRaw = Record; + +/** + * OneBot/NapCat 群文件 Provider + * 负责通过 OneBot HTTP API 列出群文件、获取下载链接并下载到本地。 + */ +@injectable() +@mustInitBeforeUse +export class OneBotFileProvider extends Disposable implements IGroupFileProvider { + private LOGGER = Logger.withTag("OneBotFileProvider"); + private config: OneBotConfig | null = null; + + /** + * 构造函数 + * @param configManagerService 配置管理服务 + */ + public constructor( + @inject(COMMON_TOKENS.ConfigManagerService) private configManagerService: ConfigManagerService + ) { + super(); + } + + /** + * 初始化 OneBot 文件 Provider + */ + public async init(): Promise { + const config = (await this.configManagerService.getCurrentConfig()).dataProviders.OneBot; + + if (!config || !config.enabled) { + throw new Error("OneBot/NapCat 文件 Provider 未启用"); + } + + this.config = { + enabled: config.enabled, + baseURL: this._trimTrailingSlashes(config.baseURL), + accessToken: config.accessToken, + downloadDirectory: config.downloadDirectory, + requestTimeoutMs: config.requestTimeoutMs + }; + + await mkdir(this.config.downloadDirectory, { recursive: true }); + this.LOGGER.success("初始化完成！"); + } + + /** + * 列出指定群的根目录文件 + * @param groupId 群号 + * @returns 群文件列表 + */ + public async listGroupFiles(groupId: string): Promise { + const actions = ["get_group_root_files", "get_group_file_list", "get_group_files"]; + + for (const action of actions) { + try { + const result = await this._callAction(action, { + group_id: this._toOneBotGroupId(groupId) + }); + const files = this._extractFiles(result).map(file => this._normalizeGroupFile(groupId, file)); + + this.LOGGER.info(`群 ${groupId} 通过 ${action} 获取到 ${files.length} 个文件`); + + return files; + } catch (error) { + this.LOGGER.warning(`调用 ${action} 获取群 ${groupId} 文件列表失败: ${error}`); + } + } + + throw new Error(`无法获取群 ${groupId} 的文件列表`); + } + + /** + * 获取群文件下载链接 + * @param groupId 群号 + * @param fileId 文件 ID + * @param busid 文件业务 ID + * @returns 临时下载链接 + */ + public async getGroupFileDownloadUrl(groupId: string, fileId: string, busid?: number): Promise { + const actions = ["get_group_file_url", "get_file_url"]; + + for (const action of actions) { + try { + const result = await this._callAction>(action, { + group_id: this._toOneBotGroupId(groupId), + group: groupId, + file_id: fileId, + file: fileId, + busid + }); + const url = this._pickString(result, [ + "url", + "download_url", + "downloadUrl", + "file_url", + "fileUrl" + ]); + + if (url) { + return url; + } + } catch (error) { + this.LOGGER.warning(`调用 ${action} 获取文件 ${fileId} 下载链接失败: ${error}`); + } + } + + throw new Error(`无法获取群 ${groupId} 文件 ${fileId} 的下载链接`); + } + + /** + * 下载群文件到配置目录 + * @param groupId 群号 + * @param fileId 文件 ID + * @param fileName 文件名；不传时会用 fileId 作为文件名 + * @param busid 文件业务 ID + * @returns 下载结果 + */ + public async downloadGroupFile( + groupId: string, + fileId: string, + fileName?: string, + busid?: number + ): Promise { + if (!this.config) { + throw new Error("OneBot/NapCat 文件 Provider 未初始化"); + } + + const url = await this.getGroupFileDownloadUrl(groupId, fileId, busid); + const response = await this._fetch(url, { + method: "GET" + }); + + if (!response.ok) { + throw new Error(`下载文件失败，HTTP 状态码: ${response.status}`); + } + + const buffer = Buffer.from(await response.arrayBuffer()); + const safeFileName = this._sanitizeFileName(fileName || fileId); + const groupDirectory = path.join(this.config.downloadDirectory, groupId); + const localPath = path.join(groupDirectory, safeFileName); + + await mkdir(groupDirectory, { recursive: true }); + await writeFile(localPath, buffer); + + return { + fileInfo: { + groupId, + fileId, + fileName: safeFileName, + fileSize: buffer.length, + busid + }, + localPath, + sizeBytes: buffer.length + }; + } + + private async _callAction(action: string, params: Record): Promise { + if (!this.config) { + throw new Error("OneBot/NapCat 文件 Provider 未初始化"); + } + + const response = await this._fetch(`${this.config.baseURL}/${action}`, { + method: "POST", + headers: { + "content-type": "application/json", + ...this._buildAuthHeaders() + }, + body: JSON.stringify(this._dropUndefinedValues(params)) + }); + + if (!response.ok) { + throw new Error(`OneBot API ${action} HTTP 状态码异常: ${response.status}`); + } + + const body = (await response.json()) as OneBotResponse | T; + + if (this._isWrappedOneBotResponse(body)) { + if (body.status && body.status !== "ok") { + throw new Error(body.wording || body.message || `OneBot API ${action} 调用失败`); + } + + if (typeof body.retcode === "number" && body.retcode !== 0) { + throw new Error( + body.wording || body.message || `OneBot API ${action} 返回 retcode=${body.retcode}` + ); + } + + return body.data as T; + } + + return body as T; + } + + private async _fetch(url: string, init: RequestInit): Promise { + if (!this.config) { + throw new Error("OneBot/NapCat 文件 Provider 未初始化"); + } + + const abortController = new AbortController(); + const timeout = setTimeout(() => abortController.abort(), this.config.requestTimeoutMs); + + try { + return await fetch(url, { + ...init, + signal: abortController.signal + }); + } finally { + clearTimeout(timeout); + } + } + + private _buildAuthHeaders(): Record { + if (!this.config || !this.config.accessToken) { + return {}; + } + + return { + authorization: `Bearer ${this.config.accessToken}` + }; + } + + private _extractFiles(result: unknown): OneBotFileRaw[] { + if (Array.isArray(result)) { + return result.filter(item => typeof item === "object" && item !== null) as OneBotFileRaw[]; + } + + if (typeof result !== "object" || result === null) { + return []; + } + + const record = result as Record; + const candidates = [record.files, record.file_list, record.fileList, record.items]; + + for (const candidate of candidates) { + if (Array.isArray(candidate)) { + return candidate.filter(item => typeof item === "object" && item !== null) as OneBotFileRaw[]; + } + } + + return []; + } + + private _normalizeGroupFile(groupId: string, raw: OneBotFileRaw): GroupFileInfo { + const fileId = this._pickString(raw, ["file_id", "fileId", "id"]); + const fileName = this._pickString(raw, ["file_name", "fileName", "name"]); + const folderId = this._pickString(raw, ["folder_id", "folderId", "parent_folder_id", "parentFolderId"]); + const fileSize = this._pickNumber(raw, ["file_size", "fileSize", "size"]); + const uploadTime = this._pickOptionalTime(raw, ["upload_time", "uploadTime", "uploaded_at", "uploadedAt"]); + const busid = this._pickOptionalNumber(raw, ["busid", "bus_id", "busId"]); + + if (!fileId) { + throw new Error(`群 ${groupId} 文件缺少 file_id`); + } + + if (!fileName) { + throw new Error(`群 ${groupId} 文件 ${fileId} 缺少文件名`); + } + + return { + groupId, + fileId, + fileName, + fileSize, + uploadTime, + busid, + folderId + }; + } + + private _pickString(record: Record, keys: string[]): string { + for (const key of keys) { + const value = record[key]; + + if (typeof value === "string" && value.length > 0) { + return value; + } + + if (typeof value === "number") { + return String(value); + } + } + + return ""; + } + + private _pickNumber(record: Record, keys: string[]): number { + const result = this._pickOptionalNumber(record, keys); + + return result ?? 0; + } + + private _pickOptionalNumber(record: Record, keys: string[]): number | undefined { + for (const key of keys) { + const value = record[key]; + + if (typeof value === "number") { + return value; + } + + if (typeof value === "string" && value.length > 0) { + const parsed = Number(value); + + if (!Number.isNaN(parsed)) { + return parsed; + } + } + } + + return undefined; + } + + private _pickOptionalTime(record: Record, keys: string[]): number | undefined { + const value = this._pickOptionalNumber(record, keys); + + if (value === undefined) { + return undefined; + } + + if (value < 10_000_000_000) { + return value * 1000; + } + + return value; + } + + private _toOneBotGroupId(groupId: string): string | number { + const parsed = Number(groupId); + + if (Number.isSafeInteger(parsed)) { + return parsed; + } + + return groupId; + } + + private _sanitizeFileName(fileName: string): string { + const forbiddenChars = new Set(["/", "\\", ":", "*", "?", '"', "<", ">", "|"]); + let result = ""; + + for (const char of fileName) { + if (forbiddenChars.has(char)) { + result += "_"; + } else { + result += char; + } + } + + const trimmed = result.trim(); + + return trimmed || "unnamed-file"; + } + + private _dropUndefinedValues(params: Record): Record { + const result: Record = {}; + + for (const key of Object.keys(params)) { + const value = params[key]; + + if (value !== undefined) { + result[key] = value; + } + } + + return result; + } + + private _trimTrailingSlashes(value: string): string { + let endIndex = value.length; + + while (endIndex > 0 && value[endIndex - 1] === "/") { + endIndex--; + } + + return value.slice(0, endIndex); + } + + private _isWrappedOneBotResponse(body: OneBotResponse | T): body is OneBotResponse { + return ( + typeof body === "object" && body !== null && ("status" in body || "retcode" in body || "data" in body) + ); + } +} diff --git a/applications/data-provider/src/providers/QQProvider/QQProvider.ts b/applications/data-provider/src/providers/QQProvider/QQProvider.ts index 74a17d0a..a75262f5 100644 --- a/applications/data-provider/src/providers/QQProvider/QQProvider.ts +++ b/applications/data-provider/src/providers/QQProvider/QQProvider.ts @@ -66,7 +66,7 @@ export class QQProvider extends Disposable implements IIMProvider { this.db = this._registerDisposable(db); // 加密相关配置 - this.LOGGER.info(`当前的dbKey: ${config.dbKey}`); + this.LOGGER.info("已读取 QQ 数据库密钥配置"); await db.exec(` PRAGMA key = '${config.dbKey}'; PRAGMA cipher_page_size = 4096; @@ -245,9 +245,22 @@ export class QQProvider extends Disposable implements IIMProvider { } // 获取消息正文：解析40800中的所有element（或者叫做fragment） - processedMsg.messageContent = await this._parseMessageContent( - this.messagePBParser.parseMessageSegment(result[GMC.msgContent]).messages - ); + try { + processedMsg.messageContent = await this._parseMessageContent( + this.messagePBParser.parseMessageSegment(result[GMC.msgContent]).messages + ); + } catch (error) { + if (error === ErrorReasons.PROTOBUF_ERROR) { + this.LOGGER.warning( + `msgId: ${result[GMC.msgId]} 的消息正文解析失败，已跳过该条异常消息。群号: ${ + result[GMC.groupUin] + }，发送者: ${result[GMC.sendMemberName] ?? result[GMC.sendNickName]}` + ); + continue; + } + + throw error; + } if (processedMsg.messageContent === "") { this.LOGGER.debug( `msgId: ${result[GMC.msgId]}的消息内容为空，忽略该消息。 diff --git a/applications/data-provider/src/providers/contracts/IGroupFileProvider.ts b/applications/data-provider/src/providers/contracts/IGroupFileProvider.ts new file mode 100644 index 00000000..68747a72 --- /dev/null +++ b/applications/data-provider/src/providers/contracts/IGroupFileProvider.ts @@ -0,0 +1,14 @@ +import { DownloadedGroupFile, GroupFileInfo } from "@root/common/contracts/data-provider"; +import { Disposable } from "@root/common/util/lifecycle/Disposable"; + +export interface IGroupFileProvider extends Disposable { + init(): Promise; + listGroupFiles(groupId: string): Promise; + getGroupFileDownloadUrl(groupId: string, fileId: string, busid?: number): Promise; + downloadGroupFile( + groupId: string, + fileId: string, + fileName?: string, + busid?: number + ): Promise; +} diff --git a/applications/data-provider/src/test/OneBotFileProvider.unit.test.ts b/applications/data-provider/src/test/OneBotFileProvider.unit.test.ts new file mode 100644 index 00000000..9758767f --- /dev/null +++ b/applications/data-provider/src/test/OneBotFileProvider.unit.test.ts @@ -0,0 +1,163 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { mockMkdir, mockWriteFile, mockLogger } = vi.hoisted(() => ({ + mockMkdir: vi.fn().mockResolvedValue(undefined), + mockWriteFile: vi.fn().mockResolvedValue(undefined), + mockLogger: { + info: vi.fn(), + success: vi.fn(), + warning: vi.fn(), + error: vi.fn(), + debug: vi.fn() + } +})); + +vi.mock("fs/promises", () => ({ + mkdir: mockMkdir, + writeFile: mockWriteFile +})); + +vi.mock("@root/common/util/Logger", () => ({ + default: { + withTag: () => mockLogger + } +})); + +vi.mock("@root/common/util/lifecycle/mustInitBeforeUse", () => ({ + mustInitBeforeUse: any>(constructor: T) => constructor +})); + +vi.mock("@root/common/util/lifecycle/Disposable", () => ({ + Disposable: class MockDisposable { + protected _registerDisposable(disposable: T): T { + return disposable; + } + protected _registerDisposableFunction(_func: () => Promise | void): void {} + async dispose(): Promise {} + get isDisposed(): boolean { + return false; + } + } +})); + +import { OneBotFileProvider } from "../providers/OneBotProvider/OneBotFileProvider"; + +const createConfigManagerService = (overrides = {}) => + ({ + getCurrentConfig: vi.fn().mockResolvedValue({ + dataProviders: { + OneBot: { + enabled: true, + baseURL: "http://127.0.0.1:3000/", + accessToken: "test-token", + downloadDirectory: "/tmp/synthos-onebot-files", + requestTimeoutMs: 5000, + ...overrides + } + } + }) + }) as any; + +describe("OneBotFileProvider", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.stubGlobal("fetch", vi.fn()); + }); + + it("应通过 OneBot 接口列出群文件并归一化字段", async () => { + vi.mocked(fetch).mockResolvedValueOnce( + new Response( + JSON.stringify({ + status: "ok", + retcode: 0, + data: { + files: [ + { + file_id: "file-1", + file_name: "资料.pdf", + file_size: "123", + upload_time: 1700000000, + busid: 102 + } + ] + } + }) + ) + ); + + const provider = new OneBotFileProvider(createConfigManagerService()); + + await provider.init(); + const files = await provider.listGroupFiles("123456"); + + expect(files).toEqual([ + { + groupId: "123456", + fileId: "file-1", + fileName: "资料.pdf", + fileSize: 123, + uploadTime: 1700000000000, + busid: 102, + folderId: "" + } + ]); + expect(fetch).toHaveBeenCalledWith( + "http://127.0.0.1:3000/get_group_root_files", + expect.objectContaining({ + method: "POST", + headers: expect.objectContaining({ + authorization: "Bearer test-token" + }) + }) + ); + }); + + it("应获取群文件下载链接", async () => { + vi.mocked(fetch).mockResolvedValueOnce( + new Response( + JSON.stringify({ + status: "ok", + retcode: 0, + data: { + url: "https://example.com/file.pdf" + } + }) + ) + ); + + const provider = new OneBotFileProvider(createConfigManagerService()); + + await provider.init(); + const url = await provider.getGroupFileDownloadUrl("123456", "file-1", 102); + + expect(url).toBe("https://example.com/file.pdf"); + }); + + it("应下载群文件到配置目录并清洗文件名", async () => { + vi.mocked(fetch) + .mockResolvedValueOnce( + new Response( + JSON.stringify({ + status: "ok", + retcode: 0, + data: { + url: "https://example.com/file.pdf" + } + }) + ) + ) + .mockResolvedValueOnce(new Response("hello")); + + const provider = new OneBotFileProvider(createConfigManagerService()); + + await provider.init(); + const result = await provider.downloadGroupFile("123456", "file-1", "a/b?.pdf", 102); + + expect(result.localPath).toBe("/tmp/synthos-onebot-files/123456/a_b_.pdf"); + expect(result.sizeBytes).toBe(5); + expect(mockWriteFile).toHaveBeenCalledWith( + "/tmp/synthos-onebot-files/123456/a_b_.pdf", + Buffer.from("hello") + ); + }); +}); diff --git a/applications/data-provider/src/test/QQProvider.unit.test.ts b/applications/data-provider/src/test/QQProvider.unit.test.ts index ca566175..563739a7 100644 --- a/applications/data-provider/src/test/QQProvider.unit.test.ts +++ b/applications/data-provider/src/test/QQProvider.unit.test.ts @@ -1,4 +1,5 @@ import { describe, it, expect, beforeEach, afterEach, vi, Mock } from "vitest"; +import ErrorReasons from "@root/common/contracts/ErrorReasons"; import { MsgElementType } from "../providers/QQProvider/@types/mappers/MsgElementType"; import { GroupMsgColumn as GMC } from "../providers/QQProvider/@types/mappers/GroupMsgColumn"; @@ -354,6 +355,41 @@ describe("QQProvider", () => { expect(result).toHaveLength(0); }); + it("主消息 protobuf 解析失败时应跳过单条异常消息", async () => { + const brokenRow = createMockDbRow({ + [GMC.msgId]: "1111111111111111111", + [GMC.msgContent]: Buffer.from("broken content") + }); + const validRow = createMockDbRow({ + [GMC.msgId]: "2222222222222222222", + [GMC.msgContent]: Buffer.from("valid content") + }); + + mockDbMethods.all.mockResolvedValue([brokenRow, validRow]); + mockParserMethods.parseMessageSegment + .mockImplementationOnce(() => { + throw ErrorReasons.PROTOBUF_ERROR; + }) + .mockReturnValueOnce({ + messages: [ + { + messageId: "elem_1", + elementType: MsgElementType.TEXT, + messageText: "正常消息" + } + ] + }); + + const result = await qqProvider.getMsgByTimeRange(mockTimestamp - 1000, mockTimestamp + 1000); + + expect(result).toHaveLength(1); + expect(result[0]).toMatchObject({ + msgId: "2222222222222222222", + messageContent: "正常消息" + }); + expect(mockLogger.warning).toHaveBeenCalledWith(expect.stringContaining("已跳过该条异常消息")); + }); + it("指定群号时应在 SQL 中包含群号条件", async () => { mockDbMethods.all.mockResolvedValue([]); mockParserMethods.parseMessageSegment.mockReturnValue({ messages: [] }); @@ -403,7 +439,7 @@ describe("QQProvider", () => { [GMC.msgTime]: Math.floor(mockTimestamp / 1000), [GMC.groupUin]: mockGroupId, [GMC.senderUin]: "987654321", - [GMC.replyMsgSeq]: null, + [GMC.replyMsgSeq]: 12345, [GMC.msgContent]: Buffer.from("mock"), [GMC.msgType]: MsgType.REPLY, [GMC.extraData]: Buffer.from("mock extra data"), diff --git a/applications/db-cli/src/applications/MigrateDB.ts b/applications/db-cli/src/applications/MigrateDB.ts index bdab1511..549c6c35 100644 --- a/applications/db-cli/src/applications/MigrateDB.ts +++ b/applications/db-cli/src/applications/MigrateDB.ts @@ -68,6 +68,13 @@ export class MigrateDB extends Disposable implements IApplication { modelName TEXT, updateTime INTEGER );`; + const createAGCMetadataTableSQL = ` + CREATE TABLE IF NOT EXISTS ai_digest_session_metadata ( + sessionId TEXT NOT NULL PRIMARY KEY, + summarizedUntil INTEGER NOT NULL, + summarizedMessageCount INTEGER NOT NULL, + updatedAt INTEGER NOT NULL + );`; const createInterestScoreTableSQL = ` CREATE TABLE IF NOT EXISTS interset_score_results ( topicId TEXT NOT NULL PRIMARY KEY, @@ -80,6 +87,7 @@ export class MigrateDB extends Disposable implements IApplication { await newDB.run(createIMDBTableSQL); await newDB.run(createAGCTableSQL); + await newDB.run(createAGCMetadataTableSQL); await newDB.run(createInterestScoreTableSQL); this.LOGGER.info("创建表结构成功"); diff --git a/applications/orchestrator/src/index.ts b/applications/orchestrator/src/index.ts index 23bbc467..66215ad2 100644 --- a/applications/orchestrator/src/index.ts +++ b/applications/orchestrator/src/index.ts @@ -11,6 +11,7 @@ import { sleep } from "@root/common/util/promisify/sleep"; import { bootstrap, bootstrapAll } from "@root/common/util/lifecycle/bootstrap"; import { setupReportScheduler } from "./schedulers/reportScheduler"; +import { NetworkRecoveryPipelineScheduler } from "./schedulers/networkRecoveryPipelineScheduler"; /** * Pipeline 执行顺序（严格串行）: @@ -25,6 +26,7 @@ import { setupReportScheduler } from "./schedulers/reportScheduler"; // 注意：日报生成任务由 reportScheduler 负责，独立于主 Pipeline const LOGGER = Logger.withTag("🎭 orchestrator-root-script"); +const networkRecoveryPipelineScheduler = new NetworkRecoveryPipelineScheduler(); @bootstrap // eslint-disable-next-line @typescript-eslint/no-unused-vars @@ -56,6 +58,14 @@ class OrchestratorApplication { TaskHandlerTypes.RunPipeline, async job => { LOGGER.info(`🚀 开始执行 Pipeline 任务: ${job.attrs.name}`); + const attrs = job.attrs.data || {}; + + if (await networkRecoveryPipelineScheduler.shouldSkipPipelineDueToNetwork()) { + await job.touch(); + + return; + } + config = await ConfigManagerService.getCurrentConfig(); // 刷新配置 const startTimeStamp = getHoursAgoTimestamp(config.orchestrator.dataSeekTimeWindowInHours); // 如果是负数则代表自动获取时间范围 const endTimeStamp = Date.now(); @@ -113,12 +123,13 @@ class OrchestratorApplication { // ==================== 步骤 3: AISummarize ==================== LOGGER.info("🤖 [3/5] 开始执行 AISummarize 任务..."); - const aiSummarizeSuccess = await scheduleAndWaitForJob( + const aiSummarizeSuccess = await scheduleAndWaitForJob( TaskHandlerTypes.AISummarize, { groupIds, startTimeStamp, - endTimeStamp + endTimeStamp, + ignoreActiveSessionGrace: attrs.ignoreActiveSessionGrace }, POLL_INTERVAL, TASK_TIMEOUT @@ -192,6 +203,7 @@ class OrchestratorApplication { } LOGGER.success(`🎉 Pipeline 任务全部完成！`); + networkRecoveryPipelineScheduler.markPipelineRan(); }, { concurrency: 1, @@ -203,7 +215,7 @@ class OrchestratorApplication { await sleep(10 * 1000); // 等其他apps启动后再开始流水线 TODO: 换成更优雅的方式 // 读取配置，设置定时执行 Pipeline - const pipelineIntervalMinutes = config.orchestrator?.pipelineIntervalInMinutes; + const pipelineIntervalMinutes = config.orchestrator.pipelineIntervalInMinutes; LOGGER.debug(`Pipeline 任务将每隔 ${pipelineIntervalMinutes} 分钟执行一次`); await agendaInstance.every(pipelineIntervalMinutes + " minutes", TaskHandlerTypes.RunPipeline); @@ -211,6 +223,7 @@ class OrchestratorApplication { LOGGER.success("✅ Orchestrator 准备就绪，启动 Agenda 调度器"); await agendaInstance.start(); + networkRecoveryPipelineScheduler.start(); // 设置日报定时任务 await setupReportScheduler(); diff --git a/applications/orchestrator/src/schedulers/networkRecoveryPipelineScheduler.ts b/applications/orchestrator/src/schedulers/networkRecoveryPipelineScheduler.ts new file mode 100644 index 00000000..5ce0ee85 --- /dev/null +++ b/applications/orchestrator/src/schedulers/networkRecoveryPipelineScheduler.ts @@ -0,0 +1,143 @@ +import Logger from "@root/common/util/Logger"; +import { checkConnectivity } from "@root/common/util/network/checkConnectivity"; +import { agendaInstance } from "@root/common/scheduler/agenda"; +import { TaskHandlerTypes } from "@root/common/scheduler/@types/Tasks"; + +const NETWORK_RECOVERY_LOGGER = Logger.withTag("🌐 [orchestrator] [NetworkRecoveryPipelineScheduler]"); + +const CONNECTIVITY_CHECK_INTERVAL_MS = 30 * 1000; +const CONNECTIVITY_CHECK_TIMEOUT_MS = 3000; + +/** + * 监听网络恢复，并在网络问题导致 Pipeline 暂缓后立即补跑。 + */ +export class NetworkRecoveryPipelineScheduler { + private LOGGER = NETWORK_RECOVERY_LOGGER; + private lastConnectivityState: boolean | null = null; + private hasPipelineMissedDueToNetwork: boolean = false; + private timer: ReturnType | null = null; + private isCheckingConnectivity: boolean = false; + + /** + * 启动网络恢复监听。 + */ + public start(): void { + if (this.timer) { + return; + } + + this.LOGGER.info(`启动网络恢复监听，检查间隔: ${CONNECTIVITY_CHECK_INTERVAL_MS}ms`); + void this._checkAndHandleConnectivity(); + this.timer = setInterval(() => { + void this._checkAndHandleConnectivity(); + }, CONNECTIVITY_CHECK_INTERVAL_MS); + } + + /** + * Pipeline 执行前检查网络；离线时记录待补跑并跳过本次执行。 + * @returns 当前是否应因网络离线跳过 Pipeline + */ + public async shouldSkipPipelineDueToNetwork(): Promise { + const online = await checkConnectivity(CONNECTIVITY_CHECK_TIMEOUT_MS); + + this.lastConnectivityState = online; + + if (!online) { + this.hasPipelineMissedDueToNetwork = true; + this.LOGGER.warning("当前网络不可用，本次 Pipeline 暂缓；网络恢复后将立即补跑"); + + return true; + } + + return false; + } + + /** + * 标记已经完成一次网络补跑需求。 + */ + public markPipelineRan(): void { + this.hasPipelineMissedDueToNetwork = false; + } + + /** + * 停止网络恢复监听。 + */ + public stop(): void { + if (!this.timer) { + return; + } + + clearInterval(this.timer); + this.timer = null; + } + + private async _checkAndHandleConnectivity(): Promise { + if (this.isCheckingConnectivity) { + return; + } + + this.isCheckingConnectivity = true; + + try { + const online = await checkConnectivity(CONNECTIVITY_CHECK_TIMEOUT_MS); + const previousState = this.lastConnectivityState; + + this.lastConnectivityState = online; + + if (!online) { + if (previousState !== false) { + this.LOGGER.warning("检测到网络不可用，后续恢复时将立即补跑 Pipeline"); + } + this.hasPipelineMissedDueToNetwork = true; + + return; + } + + if (previousState === false && this.hasPipelineMissedDueToNetwork) { + this.LOGGER.success("检测到网络已恢复，准备立即补跑 Pipeline"); + await this._scheduleRecoveryPipeline(); + } + } catch (error) { + this.LOGGER.error(`网络恢复监听失败: ${error}`); + } finally { + this.isCheckingConnectivity = false; + } + } + + private async _scheduleRecoveryPipeline(): Promise { + if (await this._hasActivePipelineJob()) { + this.LOGGER.info("已有 Pipeline 正在运行或等待执行，本次网络恢复不重复调度"); + + return; + } + + await agendaInstance.now(TaskHandlerTypes.RunPipeline, { + ignoreActiveSessionGrace: true + }); + this.hasPipelineMissedDueToNetwork = false; + this.LOGGER.success("已提交网络恢复后的 Pipeline 补跑任务，本次将允许摘要最新活跃 session"); + } + + private async _hasActivePipelineJob(): Promise { + const jobs = await agendaInstance.jobs({ name: TaskHandlerTypes.RunPipeline }); + const now = Date.now(); + + return jobs.some(job => { + const attrs = job.attrs; + + if (attrs.lockedAt) { + return true; + } + + if (attrs.repeatInterval) { + return false; + } + + if (!attrs.nextRunAt) { + return false; + } + + return attrs.nextRunAt.getTime() <= now + CONNECTIVITY_CHECK_INTERVAL_MS; + }); + } +} diff --git a/applications/webui-frontend/src/pages/reports/reports.tsx b/applications/webui-frontend/src/pages/reports/reports.tsx index 028e0472..cbc01265 100644 --- a/applications/webui-frontend/src/pages/reports/reports.tsx +++ b/applications/webui-frontend/src/pages/reports/reports.tsx @@ -417,6 +417,28 @@ export default function ReportsPage() { const handleGenerateReport = async () => { setGenerating(true); try { + const configResponse = await getCurrentConfig(); + + if (!configResponse.success || !configResponse.data) { + Notification.error({ + title: "获取配置失败", + description: configResponse.message || "无法确认日报功能状态" + }); + + return; + } + + const config = configResponse.data as { report?: { enabled?: boolean } }; + + if (config.report?.enabled === false) { + Notification.error({ + title: "日报功能未开启", + description: "请在控制页面打开report功能" + }); + + return; + } + const response = await triggerReportGenerate(generateType); if (response.success && response.data.success) { diff --git a/common/contracts/data-provider/index.ts b/common/contracts/data-provider/index.ts index fc9e942f..ff763137 100644 --- a/common/contracts/data-provider/index.ts +++ b/common/contracts/data-provider/index.ts @@ -19,6 +19,22 @@ export interface ProcessedChatMessage { export type ProcessedChatMessageWithRawMessage = RawChatMessage & ProcessedChatMessage; +export interface GroupFileInfo { + groupId: string; + fileId: string; + fileName: string; + fileSize: number; + uploadTime?: number; + busid?: number; + folderId?: string; +} + +export interface DownloadedGroupFile { + fileInfo: GroupFileInfo; + localPath: string; + sizeBytes: number; +} + // IM类型 export enum IMTypes { QQ = "QQ", diff --git a/common/scheduler/@types/Tasks.ts b/common/scheduler/@types/Tasks.ts index 158e1c5c..c916f2fc 100644 --- a/common/scheduler/@types/Tasks.ts +++ b/common/scheduler/@types/Tasks.ts @@ -33,6 +33,7 @@ export interface TaskParamsMap { groupIds: string[]; startTimeStamp: number; endTimeStamp: number; + ignoreActiveSessionGrace?: boolean; }; [TaskHandlerTypes.InterestScore]: { startTimeStamp: number; @@ -47,7 +48,9 @@ export interface TaskParamsMap { endTimeStamp: number; }; // Pipeline 任务参数 - [TaskHandlerTypes.RunPipeline]: {}; + [TaskHandlerTypes.RunPipeline]: { + ignoreActiveSessionGrace?: boolean; + }; // 日报生成任务参数 [TaskHandlerTypes.GenerateReport]: { reportType: ReportType; diff --git a/common/services/config/schemas/GlobalConfig.ts b/common/services/config/schemas/GlobalConfig.ts index 5b28ecfb..23ee66e3 100644 --- a/common/services/config/schemas/GlobalConfig.ts +++ b/common/services/config/schemas/GlobalConfig.ts @@ -99,7 +99,17 @@ export const GlobalConfigSchema = z.object({ }) .describe("数据库补丁配置") }) - .describe("QQ 数据源配置") + .describe("QQ 数据源配置"), + OneBot: z + .object({ + enabled: z.boolean().describe("是否启用 OneBot/NapCat 文件 Provider"), + baseURL: z.string().url().describe("OneBot/NapCat HTTP API 基础地址"), + accessToken: z.string().describe("OneBot 访问令牌，未配置鉴权时为空字符串"), + downloadDirectory: z.string().describe("群文件下载保存目录"), + requestTimeoutMs: z.number().positive().int().describe("OneBot HTTP 请求超时时间（毫秒）") + }) + .optional() + .describe("OneBot/NapCat 文件 Provider 配置") }) .describe("dataProviders配置"), diff --git a/common/services/database/AgcDbAccessService.ts b/common/services/database/AgcDbAccessService.ts index 53dea6f7..24103cdd 100644 --- a/common/services/database/AgcDbAccessService.ts +++ b/common/services/database/AgcDbAccessService.ts @@ -10,6 +10,18 @@ import { COMMON_TOKENS } from "../../di/tokens"; import { CommonDBService } from "./infra/CommonDBService"; import { createAGCTableSQL } from "./constants/InitialSQL"; +export interface SessionDigestMetadata { + sessionId: string; + summarizedUntil: number; + summarizedMessageCount: number; + updatedAt: number; +} + +export interface SessionDigestCoverage { + summarizedUntil: number; + summarizedMessageCount: number | null; +} + /** * AI 生成内容数据库访问服务 * 负责 AI 摘要结果的存储和查询 @@ -34,7 +46,6 @@ export class AgcDbAccessService extends Disposable { * @param result 摘要结果 */ public async storeAIDigestResult(result: AIDigestResult) { - // to fix await this.db.run( `INSERT INTO ai_digest_results (topicId, sessionId, topic, contributors, detail, modelName, updateTime) VALUES (?,?,?,?,?,?,?) ON CONFLICT(topicId) DO UPDATE SET @@ -68,23 +79,39 @@ export class AgcDbAccessService extends Disposable { } /** - * 更新一个sessionId的摘要结果。会先删除该sessionId对应的所有摘要结果（topics），然后再插入新的摘要结果 + * 追加存储一个 session 的摘要结果，并更新该 session 已摘要的消息范围 * @param sessionId 会话id * @param results 摘要结果 + * @param metadata 摘要覆盖范围 */ - public async updateAIDigestResultBySessionId(sessionId: string, results: AIDigestResult[]): Promise { - // 1. 防御性检查：results中所有result的sessionId都必须是指定的sessionId + public async storeAIDigestResultsWithSessionMetadata( + sessionId: string, + results: AIDigestResult[], + metadata: Omit + ): Promise { + if (results.length === 0) { + throw new Error(`session ${sessionId} 的摘要结果不能为空`); + } + for (const result of results) { if (result.sessionId !== sessionId) { throw new Error(`result的sessionId必须是${sessionId}，但实际为${result.sessionId}`); } } - // 2. 删除该sessionId对应的所有摘要结果（topics） - await this.db.run(`DELETE FROM ai_digest_results WHERE sessionId = ?`, [sessionId]); - - // 3. 插入新的摘要结果 - await this.storeAIDigestResults(results); + await this.db.run(`BEGIN IMMEDIATE TRANSACTION`); + try { + await this.storeAIDigestResults(results); + await this.upsertSessionDigestMetadata( + sessionId, + metadata.summarizedUntil, + metadata.summarizedMessageCount + ); + await this.db.run(`COMMIT`); + } catch (error) { + await this.db.run(`ROLLBACK`); + throw error; + } } /** @@ -128,6 +155,99 @@ export class AgcDbAccessService extends Disposable { return result[Object.keys(result)[0]] === 1; } + /** + * 获取 session 摘要覆盖范围元数据 + * @param sessionId 会话id + * @returns 摘要覆盖范围元数据 + */ + public async getSessionDigestMetadata(sessionId: string): Promise { + const result = await this.db.get( + `SELECT * FROM ai_digest_session_metadata WHERE sessionId = ?`, + [sessionId] + ); + + return result || null; + } + + /** + * 获取 session 摘要覆盖范围。老数据没有覆盖范围元数据时，用摘要生成时间兼容判断。 + * @param sessionId 会话id + * @returns 摘要覆盖范围 + */ + public async getSessionDigestCoverage(sessionId: string): Promise { + const metadata = await this.getSessionDigestMetadata(sessionId); + + if (metadata) { + return { + summarizedUntil: metadata.summarizedUntil, + summarizedMessageCount: metadata.summarizedMessageCount + }; + } + + const latestDigest = await this.db.get<{ latestUpdateTime: number | null }>( + `SELECT MAX(updateTime) AS latestUpdateTime FROM ai_digest_results WHERE sessionId = ?`, + [sessionId] + ); + + if (!latestDigest || latestDigest.latestUpdateTime === null) { + return null; + } + + return { + summarizedUntil: latestDigest.latestUpdateTime, + summarizedMessageCount: null + }; + } + + /** + * 保存 session 摘要覆盖范围元数据 + * @param sessionId 会话id + * @param summarizedUntil 摘要覆盖到的最新消息时间戳 + * @param summarizedMessageCount 摘要覆盖的消息数量 + */ + public async upsertSessionDigestMetadata( + sessionId: string, + summarizedUntil: number, + summarizedMessageCount: number + ): Promise { + await this.db.run( + `INSERT INTO ai_digest_session_metadata (sessionId, summarizedUntil, summarizedMessageCount, updatedAt) + VALUES (?, ?, ?, ?) + ON CONFLICT(sessionId) DO UPDATE SET + summarizedUntil = max(ai_digest_session_metadata.summarizedUntil, excluded.summarizedUntil), + summarizedMessageCount = max(ai_digest_session_metadata.summarizedMessageCount, excluded.summarizedMessageCount), + updatedAt = excluded.updatedAt`, + [sessionId, summarizedUntil, summarizedMessageCount, Date.now()] + ); + } + + /** + * 判断当前 session 摘要是否覆盖了给定消息范围 + * @param sessionId 会话id + * @param latestMessageTimestamp 当前 session 最新消息时间戳 + * @param messageCount 当前 session 消息数量 + * @returns 当前摘要是否仍然有效 + */ + public async isSessionDigestFresh( + sessionId: string, + latestMessageTimestamp: number, + messageCount: number + ): Promise { + const coverage = await this.getSessionDigestCoverage(sessionId); + + if (!coverage) { + return false; + } + + if (coverage.summarizedMessageCount === null) { + return coverage.summarizedUntil >= latestMessageTimestamp; + } + + return ( + coverage.summarizedUntil >= latestMessageTimestamp && coverage.summarizedMessageCount >= messageCount + ); + } + // 获取数据消息，用于数据库迁移、导出、备份等操作 public async selectAll(): Promise { return this.db.all(`SELECT * FROM ai_digest_results`); diff --git a/common/services/database/ImDbAccessService.ts b/common/services/database/ImDbAccessService.ts index 2cb26ede..05d82b72 100644 --- a/common/services/database/ImDbAccessService.ts +++ b/common/services/database/ImDbAccessService.ts @@ -164,6 +164,24 @@ export class ImDbAccessService extends Disposable { return results; } + /** + * 获取指定 session 的所有消息，包含预处理后的消息 + * @param sessionId 会话ID + * @returns 消息列表，已经按照时间从早到晚排序 + */ + public async getProcessedChatMessagesBySessionId( + sessionId: string + ): Promise { + const results = await this.db.all( + `SELECT * FROM chat_messages WHERE sessionId = ?`, + [sessionId] + ); + + results.sort((a, b) => a.timestamp - b.timestamp); + + return results; + } + public async getSessionIdsByGroupIdAndTimeRange( groupId: string, timeStart: number, diff --git a/common/services/database/constants/InitialSQL.ts b/common/services/database/constants/InitialSQL.ts index 24ec3d87..8b364fc0 100644 --- a/common/services/database/constants/InitialSQL.ts +++ b/common/services/database/constants/InitialSQL.ts @@ -22,6 +22,13 @@ export const createAGCTableSQL = ` detail TEXT, modelName TEXT, updateTime INTEGER + ); + + CREATE TABLE IF NOT EXISTS ai_digest_session_metadata ( + sessionId TEXT NOT NULL PRIMARY KEY, + summarizedUntil INTEGER NOT NULL, + summarizedMessageCount INTEGER NOT NULL, + updatedAt INTEGER NOT NULL );`; export const createInterestScoreTableSQL = ` diff --git a/common/util/network/checkConnectivity.ts b/common/util/network/checkConnectivity.ts index 42abd246..eb7fd2e4 100644 --- a/common/util/network/checkConnectivity.ts +++ b/common/util/network/checkConnectivity.ts @@ -1,77 +1,83 @@ -import * as http from "http"; - -// 通过向 Google 的网络连通性检测端点发起 HTTP 请求进行判断。 -// 重庆大学校园网环境下可能无法正常工作，此时应使用 checkConnectivityV2 方法。 -async function checkConnectivityV1(timeout: number = 5000): Promise { - return new Promise(resolve => { - const url = "http://connectivitycheck.gstatic.com/generate_204"; - - const req = http.get(url, res => { - // Google 的该端点应返回 204 No Content - resolve(res.statusCode === 204); - }); - - // 设置请求超时 - req.setTimeout(timeout, () => { - req.destroy(); - resolve(false); - }); - - // 捕获请求错误（如 DNS 失败、连接拒绝、无网络等） - req.on("error", () => { - resolve(false); - }); - }); -} - -async function checkConnectivityV2(timeout: number = 5000): Promise { - return new Promise(resolve => { - const url = "http://www.msftconnecttest.com/connecttest.txt"; - const req = http.get(url, res => { - // Microsoft 的该端点应返回"Microsoft Connect Test" - if (res.statusCode === 200) { - let data = ""; - - res.on("data", chunk => { - data += chunk; - }); - res.on("end", () => { - resolve(data.includes("Microsoft Connect Test")); - }); - } - }); - - req.setTimeout(timeout, () => { - req.destroy(); - resolve(false); - }); - - req.on("error", () => { - resolve(false); - }); - }); -} - -/** - * 检测当前设备是否能够访问互联网。 - * @param timeout - 请求超时时间（毫秒），默认为 5000ms - * @returns Promise - 若成功收到 204 响应则返回 true，否则返回 false - */ -export async function checkConnectivity(timeout: number = 5000): Promise { - // v1和v2只要有一个成功即可 - try { - return (await checkConnectivityV1(timeout)) || (await checkConnectivityV2(timeout)); - } catch (error) { - console.error("网络连通性检查失败:", error); - - return false; - } -} - -// 示例用法 -// import { checkConnectivity } from "./network-utils"; - -// (async () => { -// const online = await checkConnectivity(); -// console.log("当前是否联网:", online); -// })(); +import * as http from "http"; + +// 通过向 Google 的网络连通性检测端点发起 HTTP 请求进行判断。 +// 重庆大学校园网环境下可能无法正常工作，此时应使用 checkConnectivityV2 方法。 +async function checkConnectivityV1(timeout: number = 5000): Promise { + return new Promise(resolve => { + const url = "http://connectivitycheck.gstatic.com/generate_204"; + + const req = http.get(url, res => { + // Google 的该端点应返回 204 No Content + res.resume(); + resolve(res.statusCode === 204); + }); + + // 设置请求超时 + req.setTimeout(timeout, () => { + req.destroy(); + resolve(false); + }); + + // 捕获请求错误（如 DNS 失败、连接拒绝、无网络等） + req.on("error", () => { + resolve(false); + }); + }); +} + +async function checkConnectivityV2(timeout: number = 5000): Promise { + return new Promise(resolve => { + const url = "http://www.msftconnecttest.com/connecttest.txt"; + const req = http.get(url, res => { + // Microsoft 的该端点应返回"Microsoft Connect Test" + if (res.statusCode !== 200) { + res.resume(); + resolve(false); + + return; + } + + let data = ""; + + res.on("data", chunk => { + data += chunk; + }); + res.on("end", () => { + resolve(data.includes("Microsoft Connect Test")); + }); + }); + + req.setTimeout(timeout, () => { + req.destroy(); + resolve(false); + }); + + req.on("error", () => { + resolve(false); + }); + }); +} + +/** + * 检测当前设备是否能够访问互联网。 + * @param timeout - 请求超时时间（毫秒），默认为 5000ms + * @returns Promise - 若成功收到 204 响应则返回 true，否则返回 false + */ +export async function checkConnectivity(timeout: number = 5000): Promise { + // v1和v2只要有一个成功即可 + try { + return (await checkConnectivityV1(timeout)) || (await checkConnectivityV2(timeout)); + } catch (error) { + console.error("网络连通性检查失败:", error); + + return false; + } +} + +// 示例用法 +// import { checkConnectivity } from "./network-utils"; + +// (async () => { +// const online = await checkConnectivity(); +// console.log("当前是否联网:", online); +// })();