SnowCat 雪猫底层全源码：NapCat 4.18.28 底座 + SnowLuma 能力移植 + V8 通道适配 9.9.37
26 包 monorepo（TS 全源码）： - snowcat-core: NT 内核 API 封装 + listener（含 9.9.37 六回调适配）+ Packet 发包 - snowcat-onebot: OneBot v11 全动作 + MCP Streamable HTTP 适配器 + 稳定线 - snowcat-webui-frontend/backend: 雪猫控制台（React+HeroUI） - snowcat-shell: 单文件 bundle 出口（vite inlineDynamicImports） - snowcat-native: napi2native / MoeHoo / ffmpeg / pty 运行时 - 56 项 SnowLuma 独门动作移植（qzone 全家桶/群相册/装扮/闪照/收藏） - 工具件: 解包器/启动器/便携制作器/doctor GitHub 关联已全量指向 DandanLLab/SnowCatQQ
This commit is contained in:
天天的鸟蛋蛋 committed
commit
b9655a5ab1
1048 files changed
+130851
No files matched your search
@@ -0,0 +1,102 @@
|
||||
import { OneBotAction, OneBotRequestToolkit } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
import fs from 'fs';
|
||||
import { FileNapCatOneBotUUID } from 'snowcat-common/src/file-uuid';
|
||||
import { calculateTimeout } from '@/snowcat-onebot/config/config';
|
||||
|
||||
export interface ResolvedFileInfo {
|
||||
downloadPath: string;
|
||||
fileName: string;
|
||||
fileSize: number;
|
||||
}
|
||||
|
||||
export interface DownloadResult {
|
||||
// 文件信息
|
||||
file_name?: string;
|
||||
file_size?: number;
|
||||
chunk_size?: number;
|
||||
|
||||
// 分片数据
|
||||
index?: number;
|
||||
data?: string;
|
||||
size?: number;
|
||||
progress?: number;
|
||||
base64_size?: number;
|
||||
|
||||
// 完成信息
|
||||
total_chunks?: number;
|
||||
total_bytes?: number;
|
||||
message?: string;
|
||||
data_type?: 'file_info' | 'file_chunk' | 'file_complete';
|
||||
|
||||
// 可选扩展字段
|
||||
width?: number;
|
||||
height?: number;
|
||||
out_format?: string;
|
||||
}
|
||||
|
||||
export abstract class BaseDownloadStream<PayloadType, ResultType> extends OneBotAction<PayloadType, StreamPacket<ResultType>> {
|
||||
protected async resolveDownload (file?: string): Promise<ResolvedFileInfo> {
|
||||
const target = file || '';
|
||||
let downloadPath = '';
|
||||
let fileName = '';
|
||||
let fileSize = 0;
|
||||
|
||||
const contextMsgFile = FileNapCatOneBotUUID.decode(target);
|
||||
if (contextMsgFile && contextMsgFile.msgId && contextMsgFile.elementId) {
|
||||
const { peer, msgId, elementId } = contextMsgFile;
|
||||
downloadPath = await this.core.apis.FileApi.downloadMedia(msgId, peer.chatType, peer.peerUid, elementId, '', '');
|
||||
const rawMessage = (await this.core.apis.MsgApi.getMsgsByMsgId(peer, [msgId]))?.msgList
|
||||
.find(msg => msg.msgId === msgId);
|
||||
const mixElement = rawMessage?.elements.find(e => e.elementId === elementId);
|
||||
const mixElementInner = mixElement?.videoElement ?? mixElement?.fileElement ?? mixElement?.pttElement ?? mixElement?.picElement;
|
||||
if (!mixElementInner) throw new Error('element not found');
|
||||
fileSize = parseInt(mixElementInner.fileSize?.toString() ?? '0');
|
||||
fileName = mixElementInner.fileName ?? '';
|
||||
return { downloadPath, fileName, fileSize };
|
||||
}
|
||||
|
||||
const contextModelIdFile = FileNapCatOneBotUUID.decodeModelId(target);
|
||||
if (contextModelIdFile && contextModelIdFile.modelId) {
|
||||
const { peer, modelId } = contextModelIdFile;
|
||||
downloadPath = await this.core.apis.FileApi.downloadFileForModelId(peer, modelId, '');
|
||||
return { downloadPath, fileName, fileSize };
|
||||
}
|
||||
|
||||
const searchResult = (await this.core.apis.FileApi.searchForFile([target]));
|
||||
if (searchResult) {
|
||||
const fileSizeNum = parseInt(searchResult.fileSize);
|
||||
const timeoutConfig = this.obContext.configLoader.configData.timeout;
|
||||
const estimatedTime = calculateTimeout(timeoutConfig, fileSizeNum, timeoutConfig.downloadSpeedKBps);
|
||||
downloadPath = await this.core.apis.FileApi.downloadFileById(searchResult.id, fileSizeNum, estimatedTime);
|
||||
fileSize = fileSizeNum;
|
||||
fileName = searchResult.fileName;
|
||||
return { downloadPath, fileName, fileSize };
|
||||
}
|
||||
|
||||
throw new Error('file not found');
|
||||
}
|
||||
|
||||
protected async streamFileChunks (req: OneBotRequestToolkit, streamPath: string, chunkSize: number, chunkDataType: string): Promise<{ totalChunks: number; totalBytes: number }> {
|
||||
const stats = await fs.promises.stat(streamPath);
|
||||
const totalSize = stats.size;
|
||||
const readStream = fs.createReadStream(streamPath, { highWaterMark: chunkSize });
|
||||
let chunkIndex = 0;
|
||||
let bytesRead = 0;
|
||||
for await (const chunk of readStream) {
|
||||
const base64Chunk = (chunk as Buffer).toString('base64');
|
||||
bytesRead += (chunk as Buffer).length;
|
||||
await req.send({
|
||||
type: StreamStatus.Stream,
|
||||
data_type: chunkDataType,
|
||||
index: chunkIndex,
|
||||
data: base64Chunk,
|
||||
size: (chunk as Buffer).length,
|
||||
progress: Math.round((bytesRead / totalSize) * 100),
|
||||
base64_size: base64Chunk.length,
|
||||
} as unknown as StreamPacket<any>);
|
||||
chunkIndex++;
|
||||
}
|
||||
return { totalChunks: chunkIndex, totalBytes: bytesRead };
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotAction } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { join } from 'node:path';
|
||||
import { readdir, unlink } from 'node:fs/promises';
|
||||
import { Type } from '@sinclair/typebox';
|
||||
|
||||
export class CleanStreamTempFile extends OneBotAction<void, void> {
|
||||
override actionName = ActionName.CleanStreamTempFile;
|
||||
override actionSummary = '清理流式传输临时文件';
|
||||
override actionTags = ['流式传输扩展'];
|
||||
override payloadExample = {};
|
||||
override returnExample = {
|
||||
message: 'success',
|
||||
};
|
||||
|
||||
override payloadSchema = Type.Object({});
|
||||
override returnSchema = Type.Null();
|
||||
|
||||
async _handle (_payload: void): Promise<void> {
|
||||
try {
|
||||
// 获取临时文件夹路径
|
||||
const tempPath = this.core.SnowCatTempPath;
|
||||
|
||||
// 读取文件夹中的所有文件
|
||||
const files = await readdir(tempPath);
|
||||
|
||||
// 删除每个文件
|
||||
const deletePromises = files.map(async (file) => {
|
||||
const filePath = join(tempPath, file);
|
||||
try {
|
||||
await unlink(filePath);
|
||||
this.core.context.logger.log(`已删除文件: ${filePath}`);
|
||||
} catch (err: unknown) {
|
||||
this.core.context.logger.log(`删除文件 ${filePath} 失败: ${(err as Error).message}`);
|
||||
}
|
||||
});
|
||||
await Promise.all(deletePromises);
|
||||
} catch (err: unknown) {
|
||||
this.core.context.logger.log(`清理流临时文件失败: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotRequestToolkit } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { Static, Type } from '@sinclair/typebox';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
import fs from 'fs';
|
||||
import { imageSizeFallBack } from 'snowcat-image-size';
|
||||
import { BaseDownloadStream, DownloadResult } from './BaseDownloadStream';
|
||||
|
||||
export const DownloadFileImageStreamPayloadSchema = Type.Object({
|
||||
file: Type.Optional(Type.String({ description: '文件路径或 URL' })),
|
||||
file_id: Type.Optional(Type.String({ description: '文件 ID' })),
|
||||
chunk_size: Type.Optional(Type.Number({ default: 64 * 1024, description: '分块大小 (字节)' })), // 默认64KB分块
|
||||
});
|
||||
|
||||
export type DownloadFileImageStreamPayload = Static<typeof DownloadFileImageStreamPayloadSchema>;
|
||||
|
||||
export class DownloadFileImageStream extends BaseDownloadStream<DownloadFileImageStreamPayload, DownloadResult> {
|
||||
override actionName = ActionName.DownloadFileImageStream;
|
||||
override actionSummary = '下载图片文件流';
|
||||
override actionTags = ['流式传输扩展'];
|
||||
override payloadExample = {
|
||||
file: 'image_file_id',
|
||||
};
|
||||
|
||||
override returnExample = {
|
||||
file: 'temp_image_path',
|
||||
};
|
||||
|
||||
override payloadSchema = DownloadFileImageStreamPayloadSchema;
|
||||
override returnSchema = Type.Any({ description: '下载结果 (流式)' });
|
||||
override useStream = true;
|
||||
|
||||
async _handle (payload: DownloadFileImageStreamPayload, _adaptername: string, _config: NetworkAdapterConfig, req: OneBotRequestToolkit): Promise<StreamPacket<DownloadResult>> {
|
||||
try {
|
||||
payload.file ||= payload.file_id || '';
|
||||
const chunkSize = payload.chunk_size || 64 * 1024;
|
||||
|
||||
const { downloadPath, fileName, fileSize } = await this.resolveDownload(payload.file);
|
||||
|
||||
const stats = await fs.promises.stat(downloadPath);
|
||||
const totalSize = fileSize || stats.size;
|
||||
const { width, height } = await imageSizeFallBack(downloadPath);
|
||||
|
||||
// 发送文件信息（与 DownloadFileStream 对齐，但包含宽高）
|
||||
await req.send({
|
||||
type: StreamStatus.Stream,
|
||||
data_type: 'file_info',
|
||||
file_name: fileName,
|
||||
file_size: totalSize,
|
||||
chunk_size: chunkSize,
|
||||
width,
|
||||
height,
|
||||
});
|
||||
|
||||
const { totalChunks, totalBytes } = await this.streamFileChunks(req, downloadPath, chunkSize, 'file_chunk');
|
||||
|
||||
// 返回完成状态（与 DownloadFileStream 对齐）
|
||||
return {
|
||||
type: StreamStatus.Response,
|
||||
data_type: 'file_complete',
|
||||
total_chunks: totalChunks,
|
||||
total_bytes: totalBytes,
|
||||
message: 'Download completed',
|
||||
};
|
||||
} catch (error) {
|
||||
throw new Error(`Download failed: ${(error as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotRequestToolkit } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { Static, Type } from '@sinclair/typebox';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
import fs from 'fs';
|
||||
import { FFmpegService } from '@/snowcat-core/helper/ffmpeg/ffmpeg';
|
||||
import { BaseDownloadStream, DownloadResult } from './BaseDownloadStream';
|
||||
|
||||
const out_format = ['mp3', 'amr', 'wma', 'm4a', 'spx', 'ogg', 'wav', 'flac'];
|
||||
|
||||
export const DownloadFileRecordStreamPayloadSchema = Type.Object({
|
||||
file: Type.Optional(Type.String({ description: '文件路径或 URL' })),
|
||||
file_id: Type.Optional(Type.String({ description: '文件 ID' })),
|
||||
chunk_size: Type.Optional(Type.Number({ default: 64 * 1024, description: '分块大小 (字节)' })), // 默认64KB分块
|
||||
out_format: Type.Optional(Type.String({ description: '输出格式' })),
|
||||
});
|
||||
|
||||
export type DownloadFileRecordStreamPayload = Static<typeof DownloadFileRecordStreamPayloadSchema>;
|
||||
|
||||
export class DownloadFileRecordStream extends BaseDownloadStream<DownloadFileRecordStreamPayload, DownloadResult> {
|
||||
override actionName = ActionName.DownloadFileRecordStream;
|
||||
override actionSummary = '下载语音文件流';
|
||||
override actionTags = ['流式传输扩展'];
|
||||
override payloadExample = {
|
||||
file: 'record_file_id',
|
||||
};
|
||||
|
||||
override returnExample = {
|
||||
file: 'temp_record_path',
|
||||
};
|
||||
|
||||
override payloadSchema = DownloadFileRecordStreamPayloadSchema;
|
||||
override returnSchema = Type.Any({ description: '下载结果 (流式)' });
|
||||
override useStream = true;
|
||||
|
||||
async _handle (payload: DownloadFileRecordStreamPayload, _adaptername: string, _config: NetworkAdapterConfig, req: OneBotRequestToolkit): Promise<StreamPacket<DownloadResult>> {
|
||||
try {
|
||||
payload.file ||= payload.file_id || '';
|
||||
const chunkSize = payload.chunk_size || 64 * 1024;
|
||||
|
||||
const { downloadPath, fileName, fileSize } = await this.resolveDownload(payload.file);
|
||||
|
||||
// 处理输出格式转换
|
||||
let streamPath = downloadPath;
|
||||
if (payload.out_format && typeof payload.out_format === 'string') {
|
||||
if (!out_format.includes(payload.out_format)) {
|
||||
throw new Error('转换失败 out_format 字段可能格式不正确');
|
||||
}
|
||||
|
||||
const outputFile = `${downloadPath}.${payload.out_format}`;
|
||||
|
||||
try {
|
||||
// 如果已存在目标文件则跳过转换
|
||||
await fs.promises.access(outputFile);
|
||||
streamPath = outputFile;
|
||||
} catch {
|
||||
// 尝试解码 amr 到 out format直接 ffmpeg 转换
|
||||
await FFmpegService.convertAudioFmt(downloadPath, outputFile, payload.out_format);
|
||||
streamPath = outputFile;
|
||||
}
|
||||
}
|
||||
|
||||
const stats = await fs.promises.stat(streamPath);
|
||||
const totalSize = fileSize || stats.size;
|
||||
|
||||
await req.send({
|
||||
type: StreamStatus.Stream,
|
||||
data_type: 'file_info',
|
||||
file_name: fileName,
|
||||
file_size: totalSize,
|
||||
chunk_size: chunkSize,
|
||||
out_format: payload.out_format,
|
||||
});
|
||||
|
||||
const { totalChunks, totalBytes } = await this.streamFileChunks(req, streamPath, chunkSize, 'file_chunk');
|
||||
|
||||
return {
|
||||
type: StreamStatus.Response,
|
||||
data_type: 'file_complete',
|
||||
total_chunks: totalChunks,
|
||||
total_bytes: totalBytes,
|
||||
message: 'Download completed',
|
||||
};
|
||||
} catch (error) {
|
||||
throw new Error(`Download failed: ${(error as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotRequestToolkit } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { Static, Type } from '@sinclair/typebox';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
import fs from 'fs';
|
||||
import { BaseDownloadStream, DownloadResult } from './BaseDownloadStream';
|
||||
const DownloadFileStreamPayloadSchema = Type.Object({
|
||||
file: Type.Optional(Type.String({ description: '文件路径或 URL' })),
|
||||
file_id: Type.Optional(Type.String({ description: '文件 ID' })),
|
||||
chunk_size: Type.Optional(Type.Number({ default: 64 * 1024, description: '分块大小 (字节)' })), // 默认64KB分块
|
||||
});
|
||||
|
||||
export type DownloadFileStreamPayload = Static<typeof DownloadFileStreamPayloadSchema>;
|
||||
|
||||
export class DownloadFileStream extends BaseDownloadStream<DownloadFileStreamPayload, DownloadResult> {
|
||||
override actionName = ActionName.DownloadFileStream;
|
||||
override payloadSchema = DownloadFileStreamPayloadSchema;
|
||||
override returnSchema = Type.Any({ description: '下载结果 (流式)' });
|
||||
override actionSummary = '下载文件流';
|
||||
override actionDescription = '以流式方式从网络或本地下载文件';
|
||||
override actionTags = ['流式接口'];
|
||||
override payloadExample = {
|
||||
file: 'http://example.com/file.png',
|
||||
};
|
||||
|
||||
override returnExample = {
|
||||
type: 'stream',
|
||||
data_type: 'file_info',
|
||||
file_name: 'file.png',
|
||||
file_size: 1024,
|
||||
};
|
||||
|
||||
override useStream = true;
|
||||
|
||||
async _handle (payload: DownloadFileStreamPayload, _adaptername: string, _config: NetworkAdapterConfig, req: OneBotRequestToolkit): Promise<StreamPacket<DownloadResult>> {
|
||||
try {
|
||||
payload.file ||= payload.file_id || '';
|
||||
const chunkSize = payload.chunk_size || 64 * 1024;
|
||||
|
||||
const { downloadPath, fileName, fileSize } = await this.resolveDownload(payload.file);
|
||||
|
||||
const stats = await fs.promises.stat(downloadPath);
|
||||
const totalSize = fileSize || stats.size;
|
||||
|
||||
await req.send({
|
||||
type: StreamStatus.Stream,
|
||||
data_type: 'file_info',
|
||||
file_name: fileName,
|
||||
file_size: totalSize,
|
||||
chunk_size: chunkSize,
|
||||
});
|
||||
|
||||
const { totalChunks, totalBytes } = await this.streamFileChunks(req, downloadPath, chunkSize, 'file_chunk');
|
||||
|
||||
return {
|
||||
type: StreamStatus.Response,
|
||||
data_type: 'file_complete',
|
||||
total_chunks: totalChunks,
|
||||
total_bytes: totalBytes,
|
||||
message: 'Download completed',
|
||||
};
|
||||
} catch (error) {
|
||||
throw new Error(`Download failed: ${(error as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
# Stream-Api
|
||||
|
||||
## 流式接口
|
||||
@@ -0,0 +1,16 @@
|
||||
import { OneBotAction, OneBotRequestToolkit } from '../OneBotAction';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
export type StreamPacketBasic = {
|
||||
type: StreamStatus;
|
||||
data_type?: string;
|
||||
};
|
||||
export type StreamPacket<T> = T & StreamPacketBasic;
|
||||
export enum StreamStatus {
|
||||
Stream = 'stream', // 分片流数据包
|
||||
Response = 'response', // 流最终响应
|
||||
Reset = 'reset', // 重置流
|
||||
Error = 'error', // 流错误
|
||||
}
|
||||
export abstract class BasicStream<T, R> extends OneBotAction<T, StreamPacket<R>> {
|
||||
abstract override _handle (_payload: T, _adaptername: string, _config: NetworkAdapterConfig, req: OneBotRequestToolkit): Promise<StreamPacket<R>>;
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotAction, OneBotRequestToolkit } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { Static, Type } from '@sinclair/typebox';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
|
||||
export const TestDownloadStreamPayloadSchema = Type.Object({
|
||||
error: Type.Optional(Type.Boolean({ default: false, description: '是否触发测试错误' })),
|
||||
});
|
||||
|
||||
export type TestDownloadStreamPayload = Static<typeof TestDownloadStreamPayloadSchema>;
|
||||
|
||||
export class TestDownloadStream extends OneBotAction<TestDownloadStreamPayload, StreamPacket<{ data: string; }>> {
|
||||
override actionName = ActionName.TestDownloadStream;
|
||||
override actionSummary = '测试下载流';
|
||||
override actionTags = ['流式传输扩展'];
|
||||
override payloadExample = {
|
||||
url: 'http://example.com/file',
|
||||
};
|
||||
|
||||
override returnExample = {
|
||||
success: true,
|
||||
};
|
||||
|
||||
override payloadSchema = TestDownloadStreamPayloadSchema;
|
||||
override returnSchema = Type.Any({ description: '测试流数据' });
|
||||
override useStream = true;
|
||||
|
||||
async _handle (_payload: TestDownloadStreamPayload, _adaptername: string, _config: NetworkAdapterConfig, req: OneBotRequestToolkit) {
|
||||
for (let i = 0; i < 10; i++) {
|
||||
await req.send({ type: StreamStatus.Stream, data: `Index-> ${i + 1}`, data_type: 'data_chunk' });
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
}
|
||||
if (_payload.error) {
|
||||
throw new Error('This is a test error');
|
||||
}
|
||||
return {
|
||||
type: StreamStatus.Response,
|
||||
data_type: 'data_complete',
|
||||
data: 'Stream transmission complete',
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,363 @@
|
||||
import { ActionName } from '@/snowcat-onebot/action/router';
|
||||
import { OneBotAction } from '@/snowcat-onebot/action/OneBotAction';
|
||||
import { Static, Type } from '@sinclair/typebox';
|
||||
import { NetworkAdapterConfig } from '@/snowcat-onebot/config/config';
|
||||
import { StreamPacket, StreamStatus } from './StreamBasic';
|
||||
import fs from 'fs';
|
||||
import { join as joinPath } from 'node:path';
|
||||
import { randomUUID, createHash } from 'crypto';
|
||||
|
||||
import { unlink } from 'node:fs';
|
||||
|
||||
// 简化配置
|
||||
const CONFIG = {
|
||||
TIMEOUT: 10 * 60 * 1000, // 10分钟超时
|
||||
MEMORY_THRESHOLD: 10 * 1024 * 1024, // 10MB，超过使用磁盘
|
||||
MEMORY_LIMIT: 100 * 1024 * 1024, // 100MB内存总限制
|
||||
} as const;
|
||||
|
||||
export const UploadFileStreamPayloadSchema = Type.Object({
|
||||
stream_id: Type.String({ description: '流 ID' }),
|
||||
chunk_data: Type.Optional(Type.String({ description: '分块数据 (Base64)' })),
|
||||
chunk_index: Type.Optional(Type.Number({ description: '分块索引' })),
|
||||
total_chunks: Type.Optional(Type.Number({ description: '总分块数' })),
|
||||
file_size: Type.Optional(Type.Number({ description: '文件总大小' })),
|
||||
expected_sha256: Type.Optional(Type.String({ description: '期望的 SHA256' })),
|
||||
is_complete: Type.Optional(Type.Boolean({ description: '是否完成' })),
|
||||
filename: Type.Optional(Type.String({ description: '文件名' })),
|
||||
reset: Type.Optional(Type.Boolean({ description: '是否重置' })),
|
||||
verify_only: Type.Optional(Type.Boolean({ description: '是否仅验证' })),
|
||||
file_retention: Type.Number({ default: 5 * 60 * 1000, description: '文件保留时间 (毫秒)' }), // 默认5分钟 回收 不设置或0为不回收
|
||||
});
|
||||
|
||||
export type UploadFileStreamPayload = Static<typeof UploadFileStreamPayloadSchema>;
|
||||
|
||||
// 简化流状态接口
|
||||
interface StreamState {
|
||||
id: string;
|
||||
filename: string;
|
||||
totalChunks: number;
|
||||
receivedChunks: number;
|
||||
missingChunks: Set<number>;
|
||||
|
||||
// 可选属性
|
||||
fileSize?: number;
|
||||
expectedSha256?: string;
|
||||
|
||||
// 存储策略
|
||||
useMemory: boolean;
|
||||
memoryChunks?: Map<number, Buffer>;
|
||||
tempDir?: string;
|
||||
finalPath?: string;
|
||||
fileRetention?: number;
|
||||
|
||||
// 管理
|
||||
createdAt: number;
|
||||
timeoutId: NodeJS.Timeout;
|
||||
}
|
||||
|
||||
interface StreamResult {
|
||||
stream_id: string;
|
||||
status: 'file_created' | 'chunk_received' | 'file_complete';
|
||||
received_chunks: number;
|
||||
total_chunks: number;
|
||||
file_path?: string;
|
||||
file_size?: number;
|
||||
sha256?: string;
|
||||
}
|
||||
|
||||
export class UploadFileStream extends OneBotAction<UploadFileStreamPayload, StreamPacket<StreamResult>> {
|
||||
override actionName = ActionName.UploadFileStream;
|
||||
override payloadSchema = UploadFileStreamPayloadSchema;
|
||||
override returnSchema = Type.Any({ description: '上传结果 (流式)' });
|
||||
override actionSummary = '上传文件流';
|
||||
override actionDescription = '以流式方式上传文件数据到机器人';
|
||||
override actionTags = ['流式接口'];
|
||||
override payloadExample = {
|
||||
stream_id: 'uuid-1234-5678',
|
||||
chunk_data: 'SGVsbG8gV29ybGQ=',
|
||||
chunk_index: 0,
|
||||
total_chunks: 1,
|
||||
file_size: 11,
|
||||
};
|
||||
|
||||
override returnExample = {
|
||||
type: 'stream',
|
||||
stream_id: 'uuid-1234-5678',
|
||||
status: 'chunk_received',
|
||||
received_chunks: 1,
|
||||
total_chunks: 1,
|
||||
};
|
||||
|
||||
override useStream = true;
|
||||
|
||||
private static streams = new Map<string, StreamState>();
|
||||
private static memoryUsage = 0;
|
||||
|
||||
async _handle (payload: UploadFileStreamPayload, _adaptername: string, _config: NetworkAdapterConfig): Promise<StreamPacket<StreamResult>> {
|
||||
const { stream_id, reset, verify_only } = payload;
|
||||
|
||||
if (reset) {
|
||||
this.cleanupStream(stream_id);
|
||||
throw new Error('Stream reset completed');
|
||||
}
|
||||
|
||||
if (verify_only) {
|
||||
const stream = UploadFileStream.streams.get(stream_id);
|
||||
if (!stream) throw new Error('Stream not found');
|
||||
return this.getStreamStatus(stream);
|
||||
}
|
||||
|
||||
const stream = this.getOrCreateStream(payload);
|
||||
|
||||
if (payload.chunk_data && payload.chunk_index !== undefined) {
|
||||
return await this.processChunk(stream, payload.chunk_data, payload.chunk_index);
|
||||
}
|
||||
|
||||
if (payload.is_complete || stream.receivedChunks === stream.totalChunks) {
|
||||
return await this.completeStream(stream);
|
||||
}
|
||||
|
||||
return this.getStreamStatus(stream);
|
||||
}
|
||||
|
||||
private getOrCreateStream (payload: UploadFileStreamPayload): StreamState {
|
||||
let stream = UploadFileStream.streams.get(payload.stream_id);
|
||||
|
||||
if (!stream) {
|
||||
if (!payload.total_chunks) {
|
||||
throw new Error('total_chunks required for new stream');
|
||||
}
|
||||
stream = this.createStream(payload);
|
||||
}
|
||||
|
||||
return stream;
|
||||
}
|
||||
|
||||
private createStream (payload: UploadFileStreamPayload): StreamState {
|
||||
const { stream_id, total_chunks, file_size, filename, expected_sha256 } = payload;
|
||||
|
||||
const useMemory = this.shouldUseMemory(file_size);
|
||||
if (useMemory && file_size && (UploadFileStream.memoryUsage + file_size) > CONFIG.MEMORY_LIMIT) {
|
||||
throw new Error('Memory limit exceeded');
|
||||
}
|
||||
|
||||
const stream: StreamState = {
|
||||
id: stream_id,
|
||||
filename: filename || `upload_${randomUUID()}`,
|
||||
totalChunks: total_chunks!,
|
||||
receivedChunks: 0,
|
||||
missingChunks: new Set(Array.from({ length: total_chunks! }, (_, i) => i)),
|
||||
fileSize: file_size,
|
||||
expectedSha256: expected_sha256,
|
||||
useMemory,
|
||||
createdAt: Date.now(),
|
||||
timeoutId: this.setupTimeout(stream_id),
|
||||
fileRetention: payload.file_retention,
|
||||
};
|
||||
try {
|
||||
if (useMemory) {
|
||||
stream.memoryChunks = new Map();
|
||||
if (file_size) UploadFileStream.memoryUsage += file_size;
|
||||
} else {
|
||||
this.setupDiskStorage(stream);
|
||||
}
|
||||
|
||||
UploadFileStream.streams.set(stream_id, stream);
|
||||
return stream;
|
||||
} catch (error) {
|
||||
// 如果设置存储失败，清理已创建的资源
|
||||
clearTimeout(stream.timeoutId);
|
||||
if (stream.tempDir && fs.existsSync(stream.tempDir)) {
|
||||
try {
|
||||
fs.rmSync(stream.tempDir, { recursive: true, force: true });
|
||||
} catch (cleanupError) {
|
||||
console.error('Failed to cleanup temp dir during creation error:', cleanupError);
|
||||
}
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private shouldUseMemory (fileSize?: number): boolean {
|
||||
return fileSize !== undefined && fileSize <= CONFIG.MEMORY_THRESHOLD;
|
||||
}
|
||||
|
||||
private setupDiskStorage (stream: StreamState): void {
|
||||
const tempDir = joinPath(this.core.SnowCatTempPath, `upload_${stream.id}`);
|
||||
const finalPath = joinPath(this.core.SnowCatTempPath, stream.filename);
|
||||
|
||||
fs.mkdirSync(tempDir, { recursive: true });
|
||||
|
||||
stream.tempDir = tempDir;
|
||||
stream.finalPath = finalPath;
|
||||
}
|
||||
|
||||
private setupTimeout (streamId: string): NodeJS.Timeout {
|
||||
return setTimeout(() => {
|
||||
console.log(`Stream ${streamId} timeout`);
|
||||
this.cleanupStream(streamId);
|
||||
}, CONFIG.TIMEOUT);
|
||||
}
|
||||
|
||||
private async processChunk (stream: StreamState, chunkData: string, chunkIndex: number): Promise<StreamPacket<StreamResult>> {
|
||||
// 验证索引
|
||||
if (chunkIndex < 0 || chunkIndex >= stream.totalChunks) {
|
||||
throw new Error(`Invalid chunk index: ${chunkIndex}`);
|
||||
}
|
||||
|
||||
// 检查重复
|
||||
if (!stream.missingChunks.has(chunkIndex)) {
|
||||
return this.getStreamStatus(stream);
|
||||
}
|
||||
|
||||
const buffer = Buffer.from(chunkData, 'base64');
|
||||
|
||||
// 存储分片
|
||||
if (stream.useMemory) {
|
||||
stream.memoryChunks!.set(chunkIndex, buffer);
|
||||
} else {
|
||||
const chunkPath = joinPath(stream.tempDir!, `${chunkIndex}.chunk`);
|
||||
await fs.promises.writeFile(chunkPath, buffer);
|
||||
}
|
||||
|
||||
// 更新状态
|
||||
stream.missingChunks.delete(chunkIndex);
|
||||
stream.receivedChunks++;
|
||||
this.refreshTimeout(stream);
|
||||
|
||||
return {
|
||||
type: StreamStatus.Stream,
|
||||
stream_id: stream.id,
|
||||
status: 'chunk_received',
|
||||
received_chunks: stream.receivedChunks,
|
||||
total_chunks: stream.totalChunks,
|
||||
};
|
||||
}
|
||||
|
||||
private refreshTimeout (stream: StreamState): void {
|
||||
clearTimeout(stream.timeoutId);
|
||||
stream.timeoutId = this.setupTimeout(stream.id);
|
||||
}
|
||||
|
||||
private getStreamStatus (stream: StreamState): StreamPacket<StreamResult> {
|
||||
return {
|
||||
type: StreamStatus.Stream,
|
||||
stream_id: stream.id,
|
||||
status: 'file_created',
|
||||
received_chunks: stream.receivedChunks,
|
||||
total_chunks: stream.totalChunks,
|
||||
};
|
||||
}
|
||||
|
||||
private async completeStream (stream: StreamState): Promise<StreamPacket<StreamResult>> {
|
||||
// 合并分片
|
||||
const finalBuffer = stream.useMemory
|
||||
? await this.mergeMemoryChunks(stream)
|
||||
: await this.mergeDiskChunks(stream);
|
||||
|
||||
// 验证SHA256
|
||||
const sha256 = this.validateSha256(stream, finalBuffer);
|
||||
|
||||
// 保存文件
|
||||
const finalPath = stream.finalPath || joinPath(this.core.SnowCatTempPath, stream.filename);
|
||||
await fs.promises.writeFile(finalPath, finalBuffer);
|
||||
|
||||
// 清理资源但保留文件
|
||||
this.cleanupStream(stream.id, false);
|
||||
if (stream.fileRetention && stream.fileRetention > 0) {
|
||||
setTimeout(() => {
|
||||
unlink(finalPath, err => {
|
||||
if (err) this.core.context.logger.logError(`Failed to delete retained file ${finalPath}:`, err);
|
||||
});
|
||||
}, stream.fileRetention);
|
||||
}
|
||||
return {
|
||||
type: StreamStatus.Response,
|
||||
stream_id: stream.id,
|
||||
status: 'file_complete',
|
||||
received_chunks: stream.receivedChunks,
|
||||
total_chunks: stream.totalChunks,
|
||||
file_path: finalPath,
|
||||
file_size: finalBuffer.length,
|
||||
sha256,
|
||||
};
|
||||
}
|
||||
|
||||
private async mergeMemoryChunks (stream: StreamState): Promise<Buffer> {
|
||||
const chunks: Buffer[] = [];
|
||||
for (let i = 0; i < stream.totalChunks; i++) {
|
||||
const chunk = stream.memoryChunks!.get(i);
|
||||
if (!chunk) throw new Error(`Missing memory chunk ${i}`);
|
||||
chunks.push(chunk);
|
||||
}
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
private async mergeDiskChunks (stream: StreamState): Promise<Buffer> {
|
||||
const chunks: Buffer[] = [];
|
||||
for (let i = 0; i < stream.totalChunks; i++) {
|
||||
const chunkPath = joinPath(stream.tempDir!, `${i}.chunk`);
|
||||
if (!fs.existsSync(chunkPath)) throw new Error(`Missing chunk file ${i}`);
|
||||
chunks.push(await fs.promises.readFile(chunkPath));
|
||||
}
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
private validateSha256 (stream: StreamState, buffer: Buffer): string | undefined {
|
||||
if (!stream.expectedSha256) return undefined;
|
||||
|
||||
const actualSha256 = createHash('sha256').update(buffer).digest('hex');
|
||||
if (actualSha256 !== stream.expectedSha256) {
|
||||
throw new Error(`SHA256 mismatch. Expected: ${stream.expectedSha256}, Got: ${actualSha256}`);
|
||||
}
|
||||
return actualSha256;
|
||||
}
|
||||
|
||||
private cleanupStream (streamId: string, deleteFinalFile = true): void {
|
||||
const stream = UploadFileStream.streams.get(streamId);
|
||||
if (!stream) return;
|
||||
|
||||
try {
|
||||
// 清理超时
|
||||
clearTimeout(stream.timeoutId);
|
||||
|
||||
// 清理内存
|
||||
if (stream.useMemory) {
|
||||
if (stream.fileSize) {
|
||||
UploadFileStream.memoryUsage = Math.max(0, UploadFileStream.memoryUsage - stream.fileSize);
|
||||
}
|
||||
stream.memoryChunks?.clear();
|
||||
}
|
||||
|
||||
// 清理临时文件夹及其所有内容
|
||||
if (stream.tempDir) {
|
||||
try {
|
||||
if (fs.existsSync(stream.tempDir)) {
|
||||
fs.rmSync(stream.tempDir, { recursive: true, force: true });
|
||||
console.log(`Cleaned up temp directory: ${stream.tempDir}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Failed to cleanup temp directory ${stream.tempDir}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
// 删除最终文件（如果需要）
|
||||
if (deleteFinalFile && stream.finalPath) {
|
||||
try {
|
||||
if (fs.existsSync(stream.finalPath)) {
|
||||
fs.unlinkSync(stream.finalPath);
|
||||
console.log(`Deleted final file: ${stream.finalPath}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Failed to delete final file ${stream.finalPath}:`, error);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Cleanup error for stream ${streamId}:`, error);
|
||||
} finally {
|
||||
UploadFileStream.streams.delete(streamId);
|
||||
console.log(`Stream ${streamId} cleaned up`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
NapCat OneBot WebSocket 文件流上传测试脚本
|
||||
用于测试 UploadFileStream 接口的一次性分片上传功能
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import base64
|
||||
import hashlib
|
||||
import os
|
||||
import uuid
|
||||
from typing import List, Optional
|
||||
import websockets
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
class OneBotUploadTester:
|
||||
def __init__(self, ws_url: str = "ws://localhost:3001", access_token: Optional[str] = None):
|
||||
self.ws_url = ws_url
|
||||
self.access_token = access_token
|
||||
self.websocket = None
|
||||
|
||||
async def connect(self):
|
||||
"""连接到 OneBot WebSocket"""
|
||||
headers = {}
|
||||
if self.access_token:
|
||||
headers["Authorization"] = f"Bearer {self.access_token}"
|
||||
|
||||
print(f"连接到 {self.ws_url}")
|
||||
self.websocket = await websockets.connect(self.ws_url, additional_headers=headers)
|
||||
print("WebSocket 连接成功")
|
||||
|
||||
async def disconnect(self):
|
||||
"""断开 WebSocket 连接"""
|
||||
if self.websocket:
|
||||
await self.websocket.close()
|
||||
print("WebSocket 连接已断开")
|
||||
|
||||
def calculate_file_chunks(self, file_path: str, chunk_size: int = 64) -> tuple[List[bytes], str, int]:
|
||||
"""
|
||||
计算文件分片和 SHA256
|
||||
|
||||
Args:
|
||||
file_path: 文件路径
|
||||
chunk_size: 分片大小（默认64KB）
|
||||
|
||||
Returns:
|
||||
(chunks, sha256_hash, total_size)
|
||||
"""
|
||||
chunks = []
|
||||
hasher = hashlib.sha256()
|
||||
total_size = 0
|
||||
|
||||
with open(file_path, 'rb') as f:
|
||||
while True:
|
||||
chunk = f.read(chunk_size)
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
hasher.update(chunk)
|
||||
total_size += len(chunk)
|
||||
|
||||
sha256_hash = hasher.hexdigest()
|
||||
print(f"文件分析完成:")
|
||||
print(f" - 文件大小: {total_size} 字节")
|
||||
print(f" - 分片数量: {len(chunks)}")
|
||||
print(f" - SHA256: {sha256_hash}")
|
||||
|
||||
return chunks, sha256_hash, total_size
|
||||
|
||||
async def send_action(self, action: str, params: dict, echo: str = None) -> dict:
|
||||
"""发送 OneBot 动作请求"""
|
||||
if not echo:
|
||||
echo = str(uuid.uuid4())
|
||||
|
||||
message = {
|
||||
"action": action,
|
||||
"params": params,
|
||||
"echo": echo
|
||||
}
|
||||
|
||||
print(f"发送请求: {action}")
|
||||
await self.websocket.send(json.dumps(message))
|
||||
|
||||
# 等待响应
|
||||
while True:
|
||||
response = await self.websocket.recv()
|
||||
data = json.loads(response)
|
||||
|
||||
# 检查是否是我们的响应
|
||||
if data.get("echo") == echo:
|
||||
return data
|
||||
else:
|
||||
# 可能是其他消息，继续等待
|
||||
print(f"收到其他消息: {data}")
|
||||
continue
|
||||
|
||||
async def upload_file_stream_batch(self, file_path: str, chunk_size: int = 64 ) -> str:
|
||||
"""
|
||||
一次性批量上传文件流
|
||||
|
||||
Args:
|
||||
file_path: 要上传的文件路径
|
||||
chunk_size: 分片大小
|
||||
|
||||
Returns:
|
||||
上传完成后的文件路径
|
||||
"""
|
||||
file_path = Path(file_path)
|
||||
if not file_path.exists():
|
||||
raise FileNotFoundError(f"文件不存在: {file_path}")
|
||||
|
||||
# 分析文件
|
||||
chunks, sha256_hash, total_size = self.calculate_file_chunks(str(file_path), chunk_size)
|
||||
stream_id = str(uuid.uuid4())
|
||||
|
||||
print(f"\n开始上传文件: {file_path.name}")
|
||||
print(f"流ID: {stream_id}")
|
||||
|
||||
# 一次性发送所有分片
|
||||
total_chunks = len(chunks)
|
||||
|
||||
for chunk_index, chunk_data in enumerate(chunks):
|
||||
# 将分片数据编码为 base64
|
||||
chunk_base64 = base64.b64encode(chunk_data).decode('utf-8')
|
||||
|
||||
# 构建参数
|
||||
params = {
|
||||
"stream_id": stream_id,
|
||||
"chunk_data": chunk_base64,
|
||||
"chunk_index": chunk_index,
|
||||
"total_chunks": total_chunks,
|
||||
"file_size": total_size,
|
||||
"expected_sha256": sha256_hash,
|
||||
"filename": file_path.name,
|
||||
"file_retention": 30 * 1000
|
||||
}
|
||||
|
||||
# 发送分片
|
||||
response = await self.send_action("upload_file_stream", params)
|
||||
|
||||
if response.get("status") != "ok":
|
||||
raise Exception(f"上传分片 {chunk_index} 失败: {response}")
|
||||
|
||||
# 解析流响应
|
||||
stream_data = response.get("data", {})
|
||||
print(f"分片 {chunk_index + 1}/{total_chunks} 上传成功 "
|
||||
f"(接收: {stream_data.get('received_chunks', 0)}/{stream_data.get('total_chunks', 0)})")
|
||||
|
||||
# 发送完成信号
|
||||
print(f"\n所有分片发送完成，请求文件合并...")
|
||||
complete_params = {
|
||||
"stream_id": stream_id,
|
||||
"is_complete": True
|
||||
}
|
||||
|
||||
response = await self.send_action("upload_file_stream", complete_params)
|
||||
|
||||
if response.get("status") != "ok":
|
||||
raise Exception(f"文件合并失败: {response}")
|
||||
|
||||
result = response.get("data", {})
|
||||
|
||||
if result.get("status") == "file_complete":
|
||||
print(f"✅ 文件上传成功!")
|
||||
print(f" - 文件路径: {result.get('file_path')}")
|
||||
print(f" - 文件大小: {result.get('file_size')} 字节")
|
||||
print(f" - SHA256: {result.get('sha256')}")
|
||||
return result.get('file_path')
|
||||
else:
|
||||
raise Exception(f"文件状态异常: {result}")
|
||||
|
||||
async def test_upload(self, file_path: str, chunk_size: int = 64 ):
|
||||
"""测试文件上传"""
|
||||
try:
|
||||
await self.connect()
|
||||
|
||||
# 执行上传
|
||||
uploaded_path = await self.upload_file_stream_batch(file_path, chunk_size)
|
||||
|
||||
print(f"\n🎉 测试完成! 上传后的文件路径: {uploaded_path}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 测试失败: {e}")
|
||||
raise
|
||||
finally:
|
||||
await self.disconnect()
|
||||
|
||||
def create_test_file(file_path: str, size_mb: float = 1):
|
||||
"""创建测试文件"""
|
||||
size_bytes = int(size_mb * 1024 * 1024)
|
||||
|
||||
with open(file_path, 'wb') as f:
|
||||
# 写入一些有意义的测试数据
|
||||
test_data = b"NapCat Upload Test Data - " * 100
|
||||
written = 0
|
||||
while written < size_bytes:
|
||||
write_size = min(len(test_data), size_bytes - written)
|
||||
f.write(test_data[:write_size])
|
||||
written += write_size
|
||||
|
||||
print(f"创建测试文件: {file_path} ({size_mb}MB)")
|
||||
|
||||
async def main():
|
||||
parser = argparse.ArgumentParser(description="NapCat OneBot 文件流上传测试")
|
||||
parser.add_argument("--url", default="ws://localhost:3001", help="WebSocket URL")
|
||||
parser.add_argument("--token", help="访问令牌")
|
||||
parser.add_argument("--file", help="要上传的文件路径")
|
||||
parser.add_argument("--chunk-size", type=int, default=64*1024, help="分片大小(字节)")
|
||||
parser.add_argument("--create-test", type=float, help="创建测试文件(MB)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# 创建测试文件
|
||||
if args.create_test:
|
||||
test_file = "test_upload_file.bin"
|
||||
create_test_file(test_file, args.create_test)
|
||||
if not args.file:
|
||||
args.file = test_file
|
||||
|
||||
if not args.file:
|
||||
print("请指定要上传的文件路径，或使用 --create-test 创建测试文件")
|
||||
return
|
||||
|
||||
# 创建测试器并运行
|
||||
tester = OneBotUploadTester(args.url, args.token)
|
||||
await tester.test_upload(args.file, args.chunk_size)
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 安装依赖提示
|
||||
try:
|
||||
import websockets
|
||||
except ImportError:
|
||||
print("请先安装依赖: pip install websockets")
|
||||
exit(1)
|
||||
|
||||
asyncio.run(main())
|
||||
Reference in new issue
Block a user