diff --git a/docker/Dockerfile.app b/docker/Dockerfile.app index 1d9ce32b..083820f9 100644 --- a/docker/Dockerfile.app +++ b/docker/Dockerfile.app @@ -67,6 +67,7 @@ RUN if [ -n "$APK_MIRROR_ARG" ]; then \ python3 python3-pip python3-dev libffi-dev libssl-dev \ nodejs npm \ gosu && \ + libsqlite3-0 ffmpeg \ python3 -m pip install --break-system-packages --upgrade pip setuptools wheel && \ mkdir -p /home/appuser/.local/bin && \ curl -LsSf https://astral.sh/uv/install.sh | CARGO_HOME=/home/appuser/.cargo UV_INSTALL_DIR=/home/appuser/.local/bin sh && \ diff --git a/frontend/src/i18n/locales/en-US.ts b/frontend/src/i18n/locales/en-US.ts index 733fd223..5f99a453 100755 --- a/frontend/src/i18n/locales/en-US.ts +++ b/frontend/src/i18n/locales/en-US.ts @@ -103,15 +103,17 @@ export default { noValidFilesSelected: 'All selected files are unsupported', hiddenFilesFiltered: 'Filtered {count} hidden files', imagesFilteredNoVLM: 'Filtered {count} image files (VLM not enabled)', + videosFilteredNoVLM: 'Filtered {count} video files (VLM not enabled)', + audiosFilteredNoASR: 'Filtered {count} audio files (ASR not enabled)', invalidFilesFiltered: 'Filtered {count} unsupported files', unsupportedFileType: 'Unsupported file type', unsupportedTypesHint: 'Some document types ({types}) have no available parser engine and cannot be processed', goToParserSettings: 'Configure', - failedFilesList: 'Failed files:', + failedFilesList: 'Failed files:', andMoreFiles: '...and {count} more files', duplicateFilesSkipped: '{count} duplicate files skipped', uploadFile: 'Upload File', - uploadFileDesc: 'Supports PDF, Word, TXT, etc.', + uploadFileDesc: 'Supports PDF, Word, TXT, images, videos, audio, etc.', importURL: 'Import from URL', addDocument: 'Add Document', importURLDesc: 'Import via URL link', diff --git a/frontend/src/i18n/locales/ko-KR.ts b/frontend/src/i18n/locales/ko-KR.ts index b6154b37..596c7222 100755 --- a/frontend/src/i18n/locales/ko-KR.ts +++ b/frontend/src/i18n/locales/ko-KR.ts @@ -102,6 +102,8 @@ export default { noValidFilesSelected: "선택한 파일 모두 지원되지 않음", hiddenFilesFiltered: "{count}개 숨김 파일 필터링됨", imagesFilteredNoVLM: "{count}개 이미지 파일 필터링됨(VLM 미활성화)", + videosFilteredNoVLM: "{count}개 동영상 파일 필터링됨(VLM 미활성화)", + audiosFilteredNoASR: "{count}개 오디오 파일 필터링됨(ASR 미활성화)", invalidFilesFiltered: "{count}개 지원되지 않는 파일 필터링됨", unsupportedFileType: "지원되지 않는 파일 형식", unsupportedTypesHint: "일부 문서 유형({types})에 사용 가능한 파서 엔진이 없어 처리할 수 없습니다", @@ -110,7 +112,7 @@ export default { andMoreFiles: "...외 {count}개 파일", duplicateFilesSkipped: "{count}개 중복 파일 무시됨", uploadFile: "파일 업로드", - uploadFileDesc: "PDF, Word, TXT 등 지원", + uploadFileDesc: "PDF, Word, TXT, 이미지, 동영상, 오디오 등 지원", importURL: "웹페이지 가져오기", addDocument: "문서 추가", importURLDesc: "URL 링크로 가져오기", diff --git a/frontend/src/i18n/locales/ru-RU.ts b/frontend/src/i18n/locales/ru-RU.ts index 9d4b3788..9410e14f 100755 --- a/frontend/src/i18n/locales/ru-RU.ts +++ b/frontend/src/i18n/locales/ru-RU.ts @@ -85,6 +85,8 @@ export default { noValidFilesSelected: 'Все выбранные файлы не поддерживаются', hiddenFilesFiltered: 'Отфильтровано {count} скрытых файлов', imagesFilteredNoVLM: 'Отфильтровано {count} изображений (VLM не включен)', + videosFilteredNoVLM: 'Отфильтровано {count} видеофайлов (VLM не включен)', + audiosFilteredNoASR: 'Отфильтровано {count} аудиофайлов (ASR не включен)', invalidFilesFiltered: 'Отфильтровано {count} неподдерживаемых файлов', unsupportedFileType: 'Неподдерживаемый тип файла', unsupportedTypesHint: 'Некоторые типы документов ({types}) не имеют доступного парсера и не могут быть обработаны', @@ -93,7 +95,7 @@ export default { andMoreFiles: '...и ещё {count} файлов', duplicateFilesSkipped: 'Пропущено {count} повторяющихся файлов', uploadFile: 'Загрузить файл', - uploadFileDesc: 'Поддерживает PDF, Word, TXT и т.д.', + uploadFileDesc: 'Поддерживает PDF, Word, TXT, изображения, видео, аудио и т.д.', importURL: 'Импорт из URL', importURLDesc: 'Импорт по ссылке URL', importURLTitle: 'Импорт из URL', diff --git a/frontend/src/i18n/locales/zh-CN.ts b/frontend/src/i18n/locales/zh-CN.ts index edd54e31..08130813 100755 --- a/frontend/src/i18n/locales/zh-CN.ts +++ b/frontend/src/i18n/locales/zh-CN.ts @@ -102,6 +102,8 @@ export default { noValidFilesSelected: "选中的文件均不支持", hiddenFilesFiltered: "已过滤 {count} 个隐藏文件", imagesFilteredNoVLM: "已过滤 {count} 个图片文件(未启用VLM)", + videosFilteredNoVLM: "已过滤 {count} 个视频文件(未启用VLM)", + audiosFilteredNoASR: "已过滤 {count} 个音频文件(未启用ASR)", invalidFilesFiltered: "已过滤 {count} 个不支持的文件", unsupportedFileType: "不支持的文件格式", unsupportedTypesHint: "部分文档类型（{types}）暂无可用解析引擎，上传后将无法解析", @@ -110,7 +112,7 @@ export default { andMoreFiles: "...及其他 {count} 个文件", duplicateFilesSkipped: "已忽略 {count} 个重复文件", uploadFile: "上传文件", - uploadFileDesc: "支持 PDF、Word、TXT 等", + uploadFileDesc: "支持 PDF、Word、TXT、图片、视频、音频等", importURL: "导入网页", addDocument: "添加文档", importURLDesc: "通过URL链接导入", diff --git a/frontend/src/utils/index.ts b/frontend/src/utils/index.ts index d7aacd9d..ccc3f83c 100644 --- a/frontend/src/utils/index.ts +++ b/frontend/src/utils/index.ts @@ -40,7 +40,7 @@ export function formatStringDate(date: any) { year + "-" + month + "-" + day + " " + hour + ":" + minute + ":" + second ); } -const DEFAULT_VALID_TYPES = new Set(["pdf", "txt", "md", "docx", "doc", "pptx", "ppt", "jpg", "jpeg", "png", "csv", "xlsx", "xls"]); +const DEFAULT_VALID_TYPES = new Set(["pdf", "txt", "md", "docx", "doc", "pptx", "ppt", "jpg", "jpeg", "png", "csv", "xlsx", "xls", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv", "mp3", "wav", "m4a", "flac", "ogg"]); /** * Returns true when the file should be **rejected**. diff --git a/frontend/src/views/knowledge/KnowledgeBase.vue b/frontend/src/views/knowledge/KnowledgeBase.vue index a9d324c3..e12171fa 100644 --- a/frontend/src/views/knowledge/KnowledgeBase.vue +++ b/frontend/src/views/knowledge/KnowledgeBase.vue @@ -210,6 +210,18 @@ const fileTypeOptions = computed(() => [ { content: 'MD', value: 'md' }, { content: 'URL', value: 'url' }, { content: t('knowledgeBase.typeManual'), value: 'manual' }, + { content: 'MP4', value: 'mp4' }, + { content: 'MOV', value: 'mov' }, + { content: 'AVI', value: 'avi' }, + { content: 'MKV', value: 'mkv' }, + { content: 'WEBM', value: 'webm' }, + { content: 'WMV', value: 'wmv' }, + { content: 'FLV', value: 'flv' }, + { content: 'MP3', value: 'mp3' }, + { content: 'WAV', value: 'wav' }, + { content: 'M4A', value: 'm4a' }, + { content: 'FLAC', value: 'flac' }, + { content: 'OGG', value: 'ogg' }, ]); type TagInputInstance = ComponentPublicInstance<{ focus: () => void; select: () => void }>; const tagDropdownOptions = computed(() => @@ -1115,11 +1127,14 @@ const handleFolderUpload = async (event: Event) => { } const vlmEnabled = kbInfo.value?.vlm_config?.enabled || false; + const asrEnabled = kbInfo.value?.asr_config?.enabled || false; const dynamicTypes = supportedFileTypes.value.size > 0 ? supportedFileTypes.value : undefined const validFiles: File[] = []; let hiddenFileCount = 0; let imageFilteredCount = 0; + let videoFilteredCount = 0; + let audioFilteredCount = 0; for (let i = 0; i < files.length; i++) { const file = files[i]; @@ -1132,13 +1147,25 @@ const handleFolderUpload = async (event: Event) => { continue; } + const fileExt = file.name.substring(file.name.lastIndexOf('.') + 1).toLowerCase(); + const imageTypes = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp']; + const videoTypes = ['mp4', 'mov', 'avi', 'mkv', 'webm', 'wmv', 'flv']; + const audioTypes = ['mp3', 'wav', 'm4a', 'flac', 'ogg']; + if (!vlmEnabled) { - const fileExt = file.name.substring(file.name.lastIndexOf('.') + 1).toLowerCase(); - const imageTypes = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp']; if (imageTypes.includes(fileExt)) { imageFilteredCount++; continue; } + if (videoTypes.includes(fileExt)) { + videoFilteredCount++; + continue; + } + } + + if (!asrEnabled && audioTypes.includes(fileExt)) { + audioFilteredCount++; + continue; } if (!kbFileTypeVerification(file, true, dynamicTypes)) { @@ -1151,7 +1178,6 @@ const handleFolderUpload = async (event: Event) => { if (input) input.value = ''; return; } - MessagePlugin.info(t('knowledgeBase.uploadingFolder', { total: validFiles.length })); // 批量上传 @@ -1514,7 +1540,7 @@ async function createNewSession(value: string): Promise { ref="uploadInputRef" type="file" class="document-upload-input" - :accept="acceptFileTypes || '.pdf,.docx,.doc,.txt,.md,.json,.jpg,.jpeg,.png,.csv,.xlsx,.xls,.pptx,.ppt,.mp3,.wav,.m4a,.flac,.ogg'" + :accept="acceptFileTypes || '.pdf,.docx,.doc,.txt,.md,.json,.jpg,.jpeg,.png,.csv,.xlsx,.xls,.pptx,.ppt,.mp3,.wav,.m4a,.flac,.ogg,.mp4,.mov,.avi,.mkv,.webm,.wmv,.flv'" multiple @change="handleDocumentUpload" /> diff --git a/internal/application/service/knowledge.go b/internal/application/service/knowledge.go index 18f91c52..2eef0660 100644 --- a/internal/application/service/knowledge.go +++ b/internal/application/service/knowledge.go @@ -1479,6 +1479,8 @@ type ProcessChunksOptions struct { QuestionCount int EnableMultimodel bool StoredImages []docparser.StoredImage + IsVideo bool // true when knowledge is a standalone video file + VideoFilePath string // storage path to video file (provider:// URL) // ParentChunks holds parent chunk data when parent-child chunking is enabled. // When set, the chunks passed to processChunks are child chunks, and each // child's ParentIndex references an entry in this slice. @@ -1864,13 +1866,15 @@ func (s *knowledgeService) processChunks(ctx context.Context, return } - // Skip summary/question generation for image-type knowledge — the text chunk - // is just a markdown image reference, so LLM summary would be useless. - // The multimodal task will provide a caption as the description instead. + // Skip summary/question generation for image-type and video-type knowledge — the text chunk + // is just a markdown reference, so LLM summary would be useless. + // The multimodal task will provide a caption/summary as the description instead. isImage := IsImageType(knowledge.FileType) - pendingMultimodal := isImage && options.EnableMultimodel && len(options.StoredImages) > 0 + isVideo := IsVideoType(knowledge.FileType) + pendingMultimodal := (isImage && options.EnableMultimodel && len(options.StoredImages) > 0) || + (isVideo && options.EnableMultimodel && options.IsVideo && options.VideoFilePath != "") - // For image files with pending multimodal processing, keep "processing" status + // For image/video files with pending multimodal processing, keep "processing" status // so the frontend waits until the description is ready before showing "completed". if pendingMultimodal { knowledge.ParseStatus = types.ParseStatusProcessing @@ -1884,7 +1888,7 @@ func (s *knowledgeService) processChunks(ctx context.Context, knowledge.UpdatedAt = now // Set summary status based on whether summary generation will be triggered - if len(textChunks) > 0 && !isImage { + if len(textChunks) > 0 && !isImage && !isVideo { knowledge.SummaryStatus = types.SummaryStatusPending } else { knowledge.SummaryStatus = types.SummaryStatusNone @@ -1895,7 +1899,7 @@ func (s *knowledgeService) processChunks(ctx context.Context, } // Enqueue question generation task if enabled (async, non-blocking) - if options.EnableQuestionGeneration && len(textChunks) > 0 && !isImage { + if options.EnableQuestionGeneration && len(textChunks) > 0 && !isImage && !isVideo { questionCount := options.QuestionCount if questionCount <= 0 { questionCount = 3 @@ -1907,7 +1911,7 @@ func (s *knowledgeService) processChunks(ctx context.Context, } // Enqueue summary generation task (async, non-blocking) - if len(textChunks) > 0 && !isImage { + if len(textChunks) > 0 && !isImage && !isVideo { s.enqueueSummaryGenerationTask(ctx, knowledge.KnowledgeBaseID, knowledge.ID) } @@ -1916,6 +1920,11 @@ func (s *knowledgeService) processChunks(ctx context.Context, s.enqueueImageMultimodalTasks(ctx, knowledge, kb, options.StoredImages, chunks) } + // Enqueue multimodal tasks for videos (async, non-blocking) + if options.EnableMultimodel && options.IsVideo && options.VideoFilePath != "" { + s.enqueueVideoMultimodalTasks(ctx, knowledge, kb, options.VideoFilePath, chunks) + } + // Update tenant's storage usage tenantInfo.StorageUsed += totalStorageSize if err := s.tenantRepo.AdjustStorageUsed(ctx, tenantInfo.ID, totalStorageSize); err != nil { @@ -3002,7 +3011,8 @@ func (s *knowledgeService) ReparseKnowledge(ctx context.Context, knowledgeID str func isValidFileType(filename string) bool { switch strings.ToLower(getFileType(filename)) { case "pdf", "txt", "docx", "doc", "md", "markdown", "png", "jpg", "jpeg", "gif", "csv", "xlsx", "xls", "pptx", "ppt", "json", - "mp3", "wav", "m4a", "flac", "ogg": + "mp3", "wav", "m4a", "flac", "ogg", + "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv": return true default: return false @@ -7424,6 +7434,16 @@ func IsAudioType(fileType string) bool { } } +// IsVideoType checks if a file type is a video format +func IsVideoType(fileType string) bool { + switch strings.ToLower(fileType) { + case "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv": + return true + default: + return false + } +} + // downloadFileFromURL downloads a remote file to a temp file and returns its binary content. // payloadFileName and payloadFileType are in/out pointers: if they point to an empty string, // the function resolves the value from Content-Disposition / URL path and writes it back. @@ -7899,6 +7919,8 @@ func (s *knowledgeService) ProcessDocument(ctx context.Context, t *asynq.Task) e QuestionCount: payload.QuestionCount, EnableMultimodel: payload.EnableMultimodel, StoredImages: storedImages, + IsVideo: convertResult != nil && convertResult.IsVideo, + VideoFilePath: payload.FilePath, } if kb.ChunkingConfig.EnableParentChild { @@ -8111,6 +8133,50 @@ func (s *knowledgeService) enqueueImageMultimodalTasks( } } +// enqueueVideoMultimodalTasks enqueues asynq tasks for multimodal video processing. +func (s *knowledgeService) enqueueVideoMultimodalTasks( + ctx context.Context, + knowledge *types.Knowledge, + kb *types.KnowledgeBase, + videoFilePath string, + chunks []types.ParsedChunk, +) { + if s.task == nil || videoFilePath == "" { + return + } + + // Get the first chunk's ID as the parent chunk + chunkID := "" + if len(chunks) > 0 { + chunkID = chunks[0].ChunkID + } + + lang, _ := types.LanguageFromContext(ctx) + payload := types.VideoMultimodalPayload{ + TenantID: knowledge.TenantID, + KnowledgeID: knowledge.ID, + KnowledgeBaseID: kb.ID, + ChunkID: chunkID, + VideoURL: videoFilePath, + EnableVLM: true, + EnableASR: true, + Language: lang, + } + + payloadBytes, err := json.Marshal(payload) + if err != nil { + logger.Warnf(ctx, "Failed to marshal video multimodal payload: %v", err) + return + } + + task := asynq.NewTask(types.TypeVideoMultimodal, payloadBytes) + if _, err := s.task.Enqueue(task); err != nil { + logger.Warnf(ctx, "Failed to enqueue video multimodal task for %s: %v", videoFilePath, err) + } else { + logger.Infof(ctx, "Enqueued video:multimodal task for %s", videoFilePath) + } +} + // ProcessFAQImport handles Asynq FAQ import tasks (including dry run mode) func (s *knowledgeService) ProcessFAQImport(ctx context.Context, t *asynq.Task) error { var payload types.FAQImportPayload diff --git a/internal/application/service/video_multimodal.go b/internal/application/service/video_multimodal.go new file mode 100644 index 00000000..a349099f --- /dev/null +++ b/internal/application/service/video_multimodal.go @@ -0,0 +1,596 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "strings" + "time" + + filesvc "github.com/Tencent/WeKnora/internal/application/service/file" + "github.com/Tencent/WeKnora/internal/application/service/retriever" + "github.com/Tencent/WeKnora/internal/logger" + "github.com/Tencent/WeKnora/internal/models/asr" + "github.com/Tencent/WeKnora/internal/models/utils/ollama" + "github.com/Tencent/WeKnora/internal/models/video" + "github.com/Tencent/WeKnora/internal/models/vlm" + "github.com/Tencent/WeKnora/internal/types" + "github.com/Tencent/WeKnora/internal/types/interfaces" + secutils "github.com/Tencent/WeKnora/internal/utils" + "github.com/google/uuid" + "github.com/hibiken/asynq" +) + +const ( + vlmFramePrompt = "Analyze this video frame and provide a detailed description in Chinese. Requirements:\n" + + "1. Describe visible objects, people, and their actions\n" + + "2. Describe the scene and environment\n" + + "3. Extract and describe any text on the screen\n" + + "4. Note any notable events or changes\n" + + "5. Be concise but comprehensive" + vlmSummaryPrompt = "Summarize the following video frame descriptions into a coherent video summary in Chinese. Requirements:\n" + + "1. The descriptions are ordered chronologically by timestamp\n" + + "2. Focus on the main storyline and key events\n" + + "3. Maintain the temporal flow of the video\n" + + "4. Be concise but comprehensive" +) + +// VideoMultimodalService handles video:multimodal asynq tasks. +// It reads videos from storage (via FileService for provider:// URLs), +// extracts frames using FFmpeg, performs VLM analysis and ASR transcription, +// and creates child chunks. +type VideoMultimodalService struct { + chunkService interfaces.ChunkService + modelService interfaces.ModelService + kbService interfaces.KnowledgeBaseService + knowledgeRepo interfaces.KnowledgeRepository + tenantRepo interfaces.TenantRepository + retrieveEngine interfaces.RetrieveEngineRegistry + ollamaService *ollama.OllamaService + taskEnqueuer interfaces.TaskEnqueuer +} + +func NewVideoMultimodalService( + chunkService interfaces.ChunkService, + modelService interfaces.ModelService, + kbService interfaces.KnowledgeBaseService, + knowledgeRepo interfaces.KnowledgeRepository, + tenantRepo interfaces.TenantRepository, + retrieveEngine interfaces.RetrieveEngineRegistry, + ollamaService *ollama.OllamaService, + taskEnqueuer interfaces.TaskEnqueuer, +) interfaces.TaskHandler { + return &VideoMultimodalService{ + chunkService: chunkService, + modelService: modelService, + kbService: kbService, + knowledgeRepo: knowledgeRepo, + tenantRepo: tenantRepo, + retrieveEngine: retrieveEngine, + ollamaService: ollamaService, + taskEnqueuer: taskEnqueuer, + } +} + +// Handle implements asynq handler for TypeVideoMultimodal. +func (s *VideoMultimodalService) Handle(ctx context.Context, task *asynq.Task) error { + var payload types.VideoMultimodalPayload + if err := json.Unmarshal(task.Payload(), &payload); err != nil { + return fmt.Errorf("unmarshal video multimodal payload: %w", err) + } + + logger.Infof(ctx, "[VideoMultimodal] Processing video: chunk=%s, url=%s, vlm=%v, asr=%v", + payload.ChunkID, payload.VideoURL, payload.EnableVLM, payload.EnableASR) + + ctx = context.WithValue(ctx, types.TenantIDContextKey, payload.TenantID) + if payload.Language != "" { + ctx = context.WithValue(ctx, types.LanguageContextKey, payload.Language) + } + + var videoBytes []byte + videoBytes, err := s.readVideoBytes(ctx, payload) + if err != nil { + return fmt.Errorf("read video bytes: %w", err) + } + + if !payload.EnableVLM && !payload.EnableASR { + logger.Infof(ctx, "[VideoMultimodal] Both VLM and ASR are disabled, nothing to do") + return nil + } + + var frames []*video.Frame + var extractor video.Extractor + if payload.EnableVLM { + var extErr error + extractor, extErr = video.NewExtractor(nil) + if extErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] FFmpeg not found or not executable, frame extraction disabled: %v", extErr) + logger.Warnf(ctx, "[VideoMultimodal] Note: Please install FFmpeg to enable VLM frame analysis") + frames = []*video.Frame{} + } else { + options := video.DefaultExtractOptions() + if payload.MaxFrames > 0 { + options.MaxFrames = payload.MaxFrames + } + var extractErr error + frames, extractErr = extractor.ExtractFrames(ctx, videoBytes, options) + if extractErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] Frame extraction failed: %v", extractErr) + frames = []*video.Frame{} + } + } + } else { + logger.Infof(ctx, "[VideoMultimodal] VLM disabled, skipping frame extraction") + frames = []*video.Frame{} + } + + frameDescriptions := make([]string, 0, len(frames)) + + if payload.EnableVLM { + vlmMdl, vlmErr := s.resolveVLM(ctx, payload.KnowledgeBaseID) + if vlmErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to resolve VLM: %v", vlmErr) + } else { + for i, frame := range frames { + desc, descErr := vlmMdl.Predict(ctx, frame.ImageData, vlmFramePrompt) + if descErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] VLM analysis failed for frame %d at %.2fs: %v", + i, frame.Timestamp, descErr) + continue + } + frameDesc := fmt.Sprintf("[%.2fs] %s", frame.Timestamp, desc) + frameDescriptions = append(frameDescriptions, frameDesc) + logger.Infof(ctx, "[VideoMultimodal] Frame %d analyzed at %.2fs, desc len=%d", + i, frame.Timestamp, len(desc)) + } + } + } + + var asrText string + if payload.EnableASR { + asrText, err = s.transcribeAudio(ctx, videoBytes, payload.KnowledgeBaseID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] ASR transcription failed: %v", err) + } else if asrText != "" { + logger.Infof(ctx, "[VideoMultimodal] ASR transcription completed, len=%d", len(asrText)) + } + } + + var videoSummary string + if len(frameDescriptions) > 0 && payload.EnableVLM { + vlmMdl, _ := s.resolveVLM(ctx, payload.KnowledgeBaseID) + if vlmMdl != nil { + combinedInput := strings.Join(frameDescriptions, "\n") + summary, sumErr := vlmMdl.Predict(ctx, []byte(combinedInput), vlmSummaryPrompt) + if sumErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] Video summarization failed: %v", sumErr) + videoSummary = combinedInput + } else { + videoSummary = summary + } + } else { + videoSummary = strings.Join(frameDescriptions, "\n") + } + } + + videoInfo := types.VideoInfo{ + URL: payload.VideoURL, + FrameCount: len(frameDescriptions), + HasVLMAnalysis: len(frameDescriptions) > 0, + HasASR: asrText != "", + VideoSummary: videoSummary, + ASRText: asrText, + FrameDescriptions: frameDescriptions, + } + videoInfoJSON, _ := json.Marshal([]types.VideoInfo{videoInfo}) + + newChunks := s.buildVideoChunks(payload, frameDescriptions, videoSummary, asrText, string(videoInfoJSON)) + + if len(newChunks) == 0 { + // Even if VLM/ASR both failed, mark knowledge as completed + s.finalizeVideoKnowledge(ctx, payload, "") + return nil + } + + if err := s.chunkService.CreateChunks(ctx, newChunks); err != nil { + return fmt.Errorf("create video multimodal chunks: %w", err) + } + for _, c := range newChunks { + logger.Infof(ctx, "[VideoMultimodal] Created %s chunk %s for video %s, len=%d", + c.ChunkType, c.ID, payload.VideoURL, len(c.Content)) + } + + // Index chunks so they can be retrieved + s.indexChunks(ctx, payload, newChunks) + + // Update the parent text chunk's VideoInfo (mirrors old docreader behaviour) + s.updateParentChunkVideoInfo(ctx, payload, videoInfo) + + // For standalone video files, use summary as the knowledge description + // and mark the knowledge as completed (it was kept in "processing" until now). + s.finalizeVideoKnowledge(ctx, payload, videoSummary) + + // Enqueue question generation for the frame descriptions/summary/ASR content if KB has it enabled. + // During initial processChunks, question generation is skipped for video-type + // knowledge because the text chunk is just a markdown reference. Now that we + // have real textual content (frame descriptions, summary, ASR), we can generate questions. + s.enqueueQuestionGenerationIfEnabled(ctx, payload) + + return nil +} + +func (s *VideoMultimodalService) readVideoBytes(ctx context.Context, payload types.VideoMultimodalPayload) ([]byte, error) { + var err error + var videoBytes []byte + if types.ParseProviderScheme(payload.VideoURL) != "" { + fileSvc := s.resolveFileServiceForPayload(ctx, payload) + if fileSvc == nil { + logger.Warnf(ctx, "[VideoMultimodal] Resolve tenant file service failed, fallback to URL/local: tenant=%d kb=%s", + payload.TenantID, payload.KnowledgeBaseID) + } else { + // provider:// scheme — read via FileService + reader, getErr := fileSvc.GetFile(ctx, payload.VideoURL) + if getErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] FileService.GetFile(%s) failed: %v", payload.VideoURL, getErr) + } else { + videoBytes, err = io.ReadAll(reader) + reader.Close() + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Read provider file %s failed: %v", payload.VideoURL, err) + videoBytes = nil + } + } + } + } + if videoBytes == nil && payload.VideoLocalPath != "" { + videoBytes, err = os.ReadFile(payload.VideoLocalPath) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Local file %s not available (%v), trying URL", payload.VideoLocalPath, err) + videoBytes = nil + } + } + if videoBytes == nil { + videoBytes, err = downloadVideoFromURL(payload.VideoURL) + if err != nil { + logger.Errorf(ctx, "[VideoMultimodal] Failed to download video from URL %s: %v", payload.VideoURL, err) + return nil, fmt.Errorf("read video from URL %s failed: %w", payload.VideoURL, err) + } + logger.Infof(ctx, "[VideoMultimodal] Video downloaded from URL, len=%d", len(videoBytes)) + } + return videoBytes, nil +} + +// downloadVideoFromURL downloads video bytes from an HTTP(S) URL. +func downloadVideoFromURL(videoURL string) ([]byte, error) { + return secutils.DownloadBytes(videoURL) +} + +// transcribeAudio extracts and transcribes audio from a video using ASR. +func (s *VideoMultimodalService) transcribeAudio(ctx context.Context, videoBytes []byte, kbID string) (string, error) { + asrMdl, err := s.resolveASR(ctx, kbID) + if err != nil { + return "", fmt.Errorf("resolve ASR: %w", err) + } + return asrMdl.Transcribe(ctx, videoBytes, "video.mp4") +} + +// buildVideoChunks creates child chunks for video multimodal analysis results: +// - Frame descriptions (one chunk per frame) +// - Video summary (single chunk) +// - ASR transcription (single chunk) +func (s *VideoMultimodalService) buildVideoChunks( + payload types.VideoMultimodalPayload, + frameDescriptions []string, + videoSummary string, + asrText string, + videoInfoJSON string, +) []*types.Chunk { + var newChunks []*types.Chunk + + for _, desc := range frameDescriptions { + chunk := &types.Chunk{ + ID: uuid.New().String(), + TenantID: payload.TenantID, + KnowledgeID: payload.KnowledgeID, + KnowledgeBaseID: payload.KnowledgeBaseID, + Content: desc, + ChunkType: types.ChunkTypeVideoFrame, + ParentChunkID: payload.ChunkID, + IsEnabled: true, + Flags: types.ChunkFlagRecommended, + VideoInfo: videoInfoJSON, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + } + newChunks = append(newChunks, chunk) + } + + if videoSummary != "" { + newChunks = append(newChunks, &types.Chunk{ + ID: uuid.New().String(), + TenantID: payload.TenantID, + KnowledgeID: payload.KnowledgeID, + KnowledgeBaseID: payload.KnowledgeBaseID, + Content: videoSummary, + ChunkType: types.ChunkTypeVideoCaption, + ParentChunkID: payload.ChunkID, + IsEnabled: true, + Flags: types.ChunkFlagRecommended, + VideoInfo: videoInfoJSON, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }) + } + + if asrText != "" { + newChunks = append(newChunks, &types.Chunk{ + ID: uuid.New().String(), + TenantID: payload.TenantID, + KnowledgeID: payload.KnowledgeID, + KnowledgeBaseID: payload.KnowledgeBaseID, + Content: asrText, + ChunkType: types.ChunkTypeVideoASR, + ParentChunkID: payload.ChunkID, + IsEnabled: true, + Flags: types.ChunkFlagRecommended, + VideoInfo: videoInfoJSON, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }) + } + + return newChunks +} + +// finalizeVideoKnowledge updates the knowledge after multimodal processing: +// - For standalone video files: sets Description from summary and marks ParseStatus as completed. +// - For videos extracted from documents: no-op (description comes from summary generation). +func (s *VideoMultimodalService) finalizeVideoKnowledge(ctx context.Context, payload types.VideoMultimodalPayload, summary string) { + knowledge, err := s.knowledgeRepo.GetKnowledgeByIDOnly(ctx, payload.KnowledgeID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to get knowledge %s: %v", payload.KnowledgeID, err) + return + } + if knowledge == nil { + return + } + if !IsVideoType(knowledge.FileType) { + return + } + + if summary != "" { + knowledge.Description = summary + } + knowledge.ParseStatus = types.ParseStatusCompleted + knowledge.UpdatedAt = time.Now() + if err := s.knowledgeRepo.UpdateKnowledge(ctx, knowledge); err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to finalize knowledge: %v", err) + } else { + logger.Infof(ctx, "[VideoMultimodal] Finalized video knowledge %s (status=completed, description=%d chars)", + payload.KnowledgeID, len(knowledge.Description)) + } +} + +// indexChunks indexes the newly created video chunks into the retrieval engine +// so they can participate in semantic search. +func (s *VideoMultimodalService) indexChunks(ctx context.Context, payload types.VideoMultimodalPayload, chunks []*types.Chunk) { + kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID) + if err != nil || kb == nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to get KB for indexing: %v", err) + return + } + + embeddingModel, err := s.modelService.GetEmbeddingModel(ctx, kb.EmbeddingModelID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to get embedding model for indexing: %v", err) + return + } + + tenantInfo, err := s.tenantRepo.GetTenantByID(ctx, payload.TenantID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to get tenant for indexing: %v", err) + return + } + + engine, err := retriever.NewCompositeRetrieveEngine(s.retrieveEngine, tenantInfo.GetEffectiveEngines()) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to init retrieve engine: %v", err) + return + } + + indexInfoList := make([]*types.IndexInfo, 0, len(chunks)) + for _, chunk := range chunks { + indexInfoList = append(indexInfoList, &types.IndexInfo{ + Content: chunk.Content, + SourceID: chunk.ID, + SourceType: types.ChunkSourceType, + ChunkID: chunk.ID, + KnowledgeID: chunk.KnowledgeID, + KnowledgeBaseID: chunk.KnowledgeBaseID, + }) + } + + if err := engine.BatchIndex(ctx, embeddingModel, indexInfoList); err != nil { + logger.Errorf(ctx, "[VideoMultimodal] Failed to index video chunks: %v", err) + return + } + + // Mark chunks as indexed. + // Must re-fetch from DB because the in-memory objects lack auto-generated fields + // (e.g. seq_id), and GORM Save would overwrite them with zero values. + for _, chunk := range chunks { + dbChunk, err := s.chunkService.GetChunkByIDOnly(ctx, chunk.ID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to fetch chunk %s for status update: %v", chunk.ID, err) + continue + } + dbChunk.Status = int(types.ChunkStatusIndexed) + if err := s.chunkService.UpdateChunk(ctx, dbChunk); err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to update chunk %s status to indexed: %v", chunk.ID, err) + } + } + + logger.Infof(ctx, "[VideoMultimodal] Indexed %d video chunks for video %s", len(chunks), payload.VideoURL) +} + +// updateParentChunkVideoInfo updates the parent text chunk's VideoInfo field, +// replicating the behaviour of the old docreader flow where the parent chunk +// carried the full video metadata (URL, frame descriptions, summary, ASR). +func (s *VideoMultimodalService) updateParentChunkVideoInfo( + ctx context.Context, + payload types.VideoMultimodalPayload, + videoInfo types.VideoInfo, +) { + if payload.ChunkID == "" { + return + } + + chunk, err := s.chunkService.GetChunkByIDOnly(ctx, payload.ChunkID) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to get parent chunk %s: %v", payload.ChunkID, err) + return + } + + var existingInfos []types.VideoInfo + if chunk.VideoInfo != "" { + _ = json.Unmarshal([]byte(chunk.VideoInfo), &existingInfos) + } + + found := false + for i, info := range existingInfos { + if info.URL == videoInfo.URL { + existingInfos[i] = videoInfo + found = true + break + } + } + if !found { + existingInfos = append(existingInfos, videoInfo) + } + + videoInfoJSON, _ := json.Marshal(existingInfos) + chunk.VideoInfo = string(videoInfoJSON) + chunk.UpdatedAt = time.Now() + if err := s.chunkService.UpdateChunk(ctx, chunk); err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to update parent chunk %s VideoInfo: %v", chunk.ID, err) + } else { + logger.Infof(ctx, "[VideoMultimodal] Updated parent chunk %s VideoInfo for video", chunk.ID) + } +} + +// resolveVLM creates a vlm.VLM instance for the given knowledge base, +// supporting both new-style (ModelID) and legacy (inline BaseURL) configs. +func (s *VideoMultimodalService) resolveVLM(ctx context.Context, kbID string) (vlm.VLM, error) { + kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, kbID) + if err != nil { + return nil, fmt.Errorf("get knowledge base %s: %w", kbID, err) + } + if kb == nil { + return nil, fmt.Errorf("knowledge base %s not found", kbID) + } + + vlmCfg := kb.VLMConfig + if !vlmCfg.IsEnabled() { + return nil, fmt.Errorf("VLM is not enabled for knowledge base %s", kbID) + } + + // New-style: resolve model through ModelService + if vlmCfg.ModelID != "" { + return s.modelService.GetVLMModel(ctx, vlmCfg.ModelID) + } + + // Legacy: create VLM from inline config + return vlm.NewVLMFromLegacyConfig(vlmCfg, s.ollamaService) +} + +// resolveASR creates an asr.ASR instance for the given knowledge base. +func (s *VideoMultimodalService) resolveASR(ctx context.Context, kbID string) (asr.ASR, error) { + kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, kbID) + if err != nil { + return nil, fmt.Errorf("get knowledge base %s: %w", kbID, err) + } + if kb == nil { + return nil, fmt.Errorf("knowledge base %s not found", kbID) + } + + asrCfg := kb.ASRConfig + if !asrCfg.IsASREnabled() { + return nil, fmt.Errorf("ASR is not enabled for knowledge base %s", kbID) + } + + return s.modelService.GetASRModel(ctx, asrCfg.ModelID) +} + +// resolveFileServiceForPayload resolves tenant/KB scoped file service for reading provider:// URLs. +func (s *VideoMultimodalService) resolveFileServiceForPayload(ctx context.Context, payload types.VideoMultimodalPayload) interfaces.FileService { + tenant, err := s.tenantRepo.GetTenantByID(ctx, payload.TenantID) + if err != nil || tenant == nil { + logger.Warnf(ctx, "[VideoMultimodal] GetTenantByID failed: tenant=%d err=%v", payload.TenantID, err) + return nil + } + + provider := types.ParseProviderScheme(payload.VideoURL) + if provider == "" { + kb, kbErr := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID) + if kbErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] GetKnowledgeBaseByIDOnly failed: kb=%s err=%v", payload.KnowledgeBaseID, kbErr) + } else if kb != nil { + provider = strings.ToLower(strings.TrimSpace(kb.GetStorageProvider())) + } + } + + baseDir := strings.TrimSpace(os.Getenv("LOCAL_STORAGE_BASE_DIR")) + fileSvc, _, svcErr := filesvc.NewFileServiceFromStorageConfig(provider, tenant.StorageEngineConfig, baseDir) + if svcErr != nil { + logger.Warnf(ctx, "[VideoMultimodal] resolve file service failed: tenant=%d provider=%s err=%v", payload.TenantID, provider, svcErr) + return nil + } + return fileSvc +} + +// enqueueQuestionGenerationIfEnabled checks if the knowledge base has question +// generation enabled and, if so, enqueues a task for the video knowledge. +func (s *VideoMultimodalService) enqueueQuestionGenerationIfEnabled(ctx context.Context, payload types.VideoMultimodalPayload) { + if s.taskEnqueuer == nil { + return + } + + kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID) + if err != nil || kb == nil { + return + } + if kb.QuestionGenerationConfig == nil || !kb.QuestionGenerationConfig.Enabled { + return + } + + questionCount := kb.QuestionGenerationConfig.QuestionCount + if questionCount <= 0 { + questionCount = 3 + } + if questionCount > 10 { + questionCount = 10 + } + + taskPayload := types.QuestionGenerationPayload{ + TenantID: payload.TenantID, + KnowledgeBaseID: payload.KnowledgeBaseID, + KnowledgeID: payload.KnowledgeID, + QuestionCount: questionCount, + Language: payload.Language, + } + payloadBytes, err := json.Marshal(taskPayload) + if err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to marshal question generation payload: %v", err) + return + } + + task := asynq.NewTask(types.TypeQuestionGeneration, payloadBytes, asynq.Queue("low"), asynq.MaxRetry(3)) + if _, err := s.taskEnqueuer.Enqueue(task); err != nil { + logger.Warnf(ctx, "[VideoMultimodal] Failed to enqueue question generation for %s: %v", payload.KnowledgeID, err) + } else { + logger.Infof(ctx, "[VideoMultimodal] Enqueued question generation task for video knowledge %s (count=%d)", + payload.KnowledgeID, questionCount) + } +} diff --git a/internal/container/container.go b/internal/container/container.go index 0999e9f5..8726c629 100644 --- a/internal/container/container.go +++ b/internal/container/container.go @@ -47,7 +47,6 @@ import ( "github.com/Tencent/WeKnora/internal/application/service/llmcontext" memoryService "github.com/Tencent/WeKnora/internal/application/service/memory" "github.com/Tencent/WeKnora/internal/application/service/retriever" - infra_web_search "github.com/Tencent/WeKnora/internal/infrastructure/web_search" "github.com/Tencent/WeKnora/internal/config" "github.com/Tencent/WeKnora/internal/database" "github.com/Tencent/WeKnora/internal/datasource" @@ -64,6 +63,7 @@ import ( "github.com/Tencent/WeKnora/internal/im/telegram" "github.com/Tencent/WeKnora/internal/im/wecom" "github.com/Tencent/WeKnora/internal/infrastructure/docparser" + infra_web_search "github.com/Tencent/WeKnora/internal/infrastructure/web_search" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/mcp" "github.com/Tencent/WeKnora/internal/models/embedding" @@ -173,6 +173,7 @@ func BuildContainer(container *dig.Container) *dig.Container { must(container.Provide(service.NewChunkExtractService, dig.Name("chunkExtractor"))) must(container.Provide(service.NewDataTableSummaryService, dig.Name("dataTableSummary"))) must(container.Provide(service.NewImageMultimodalService, dig.Name("imageMultimodal"))) + must(container.Provide(service.NewVideoMultimodalService, dig.Name("videoMultimodal"))) must(container.Provide(service.NewMessageService)) must(container.Provide(service.NewMCPServiceService)) diff --git a/internal/infrastructure/docparser/builtin_converter.go b/internal/infrastructure/docparser/builtin_converter.go index 3779cfa7..ec6ac077 100644 --- a/internal/infrastructure/docparser/builtin_converter.go +++ b/internal/infrastructure/docparser/builtin_converter.go @@ -28,6 +28,10 @@ var audioFormats = map[string]bool{ "mp3": true, "wav": true, "m4a": true, "flac": true, "ogg": true, } +var videoFormats = map[string]bool{ + "mp4": true, "mov": true, "avi": true, "mkv": true, "webm": true, "wmv": true, "flv": true, +} + func init() { for k := range imageFormats { simpleFormats[k] = true @@ -35,6 +39,9 @@ func init() { for k := range audioFormats { simpleFormats[k] = true } + for k := range videoFormats { + simpleFormats[k] = true + } } // IsSimpleFormat returns true if the file type can be handled by the Go SimpleFormatReader. @@ -74,6 +81,8 @@ func (b *SimpleFormatReader) Read(_ context.Context, req *types.ReadRequest) (*t return imageToResult(req.FileName, req.FileContent), nil case audioFormats[ft]: return audioToResult(req.FileName, req.FileContent), nil + case videoFormats[ft]: + return videoToResult(req.FileName, req.FileContent), nil default: return nil, fmt.Errorf("unsupported simple format: %s", ft) } @@ -113,6 +122,11 @@ func IsAudioFormat(fileType string) bool { return audioFormats[strings.ToLower(strings.TrimPrefix(fileType, "."))] } +// IsVideoFormat returns true if the file type is a recognized video format. +func IsVideoFormat(fileType string) bool { + return videoFormats[strings.ToLower(strings.TrimPrefix(fileType, "."))] +} + // audioToResult wraps a standalone audio file. The actual transcription is // handled by the ASR model in the knowledge service pipeline. Here we just // return a placeholder markdown with the raw bytes preserved for upstream @@ -130,6 +144,22 @@ func audioToResult(fileName string, data []byte) *types.ReadResult { } } +// videoToResult wraps a standalone video file. The actual frame extraction, +// VLM analysis and ASR transcription is handled by the VideoMultimodalService +// in the knowledge service pipeline. Here we just return a placeholder markdown +// with the raw bytes preserved for upstream processing. +func videoToResult(fileName string, data []byte) *types.ReadResult { + if fileName == "" { + fileName = "video.mp4" + } + // Return a placeholder; the knowledge service will handle the actual processing. + return &types.ReadResult{ + MarkdownContent: fmt.Sprintf("[Video file: %s]", fileName), + IsVideo: true, + VideoData: data, + } +} + // ensureOriginalImageRef checks whether the input file is an image and, if the // returned markdown does not already contain a markdown image reference for it, // prepends one and appends the raw bytes to imageRefs. This guarantees that diff --git a/internal/infrastructure/docparser/engine_registry.go b/internal/infrastructure/docparser/engine_registry.go index 8469aff1..294ccdd6 100644 --- a/internal/infrastructure/docparser/engine_registry.go +++ b/internal/infrastructure/docparser/engine_registry.go @@ -42,7 +42,7 @@ func (e *builtinEngine) Description() string { return "DocReader built-in parser engine" } func (e *builtinEngine) FileTypes(_ bool) []string { - return []string{"docx", "doc", "pdf", "md", "markdown", "xlsx", "xls", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp"} + return []string{"docx", "doc", "pdf", "md", "markdown", "xlsx", "xls", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv"} } func (e *builtinEngine) CheckAvailable(docreaderConnected bool, _ map[string]string) (bool, string) { if docreaderConnected { @@ -67,7 +67,7 @@ func (e *simpleEngine) Description() string { return "Simple format & image parsing (no external service required)" } func (e *simpleEngine) FileTypes(_ bool) []string { - return []string{"md", "markdown", "txt", "csv", "json", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg"} + return []string{"md", "markdown", "txt", "csv", "json", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv"} } func (e *simpleEngine) CheckAvailable(_ bool, _ map[string]string) (bool, string) { return true, "" diff --git a/internal/models/video/extractor.go b/internal/models/video/extractor.go new file mode 100644 index 00000000..610799a3 --- /dev/null +++ b/internal/models/video/extractor.go @@ -0,0 +1,95 @@ +package video + +import ( + "context" + "io" +) + +// Extractor defines the interface for video frame extraction operations. +type Extractor interface { + // ExtractFrames extracts frames from a video file. + // It returns a slice of Frame objects representing the extracted frames. + ExtractFrames(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]*Frame, error) + + // ExtractFramesToReader extracts frames from a video and returns an io.Reader + // for each frame's image data. This is useful for streaming processing. + ExtractFramesToReaders(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]io.Reader, error) + + // GetVideoInfo retrieves basic information about a video file. + // It returns a map containing duration, width, height, etc. + GetVideoInfo(ctx context.Context, videoBytes []byte) (map[string]interface{}, error) +} + +// Frame represents a single extracted frame from a video. +type Frame struct { + // Index is the sequential index of the frame (0-based). + Index int + + // Timestamp is the time in seconds from the start of the video. + Timestamp float64 + + // ImageData contains the raw image bytes of the frame. + ImageData []byte + + // Format specifies the image format (e.g., "jpeg", "png"). + Format string + + // Width is the width of the frame in pixels. + Width int + + // Height is the height of the frame in pixels. + Height int +} + +// ExtractOptions configures how frames are extracted from a video. +type ExtractOptions struct { + // FrameInterval specifies how often to extract frames (in seconds). + // For example, 1.0 means extract one frame per second. + FrameInterval float64 + + // MaxFrames specifies the maximum number of frames to extract. + // If 0, extract all frames according to FrameInterval. + MaxFrames int + + // Format specifies the output image format for extracted frames. + // Supported formats: "jpeg" (default), "png". + Format string + + // Quality specifies the image quality for JPEG output (1-100, default 80). + Quality int + + // Width specifies the target width for resizing frames (0 = no resize). + Width int + + // Height specifies the target height for resizing frames (0 = no resize). + // If only one dimension is specified, the aspect ratio is preserved. + Height int +} + +// DefaultExtractOptions returns the default frame extraction options. +func DefaultExtractOptions() *ExtractOptions { + return &ExtractOptions{ + FrameInterval: 1.0, + MaxFrames: 0, + Format: "jpeg", + Quality: 80, + Width: 0, + Height: 0, + } +} + +// Config holds the configuration needed to create a video extractor. +type Config struct { + // FFMpegPath is the path to the ffmpeg executable. + // If empty, it will try to find ffmpeg in the system PATH. + FFMpegPath string + + // TempDir is the directory to use for temporary files. + // If empty, the system default temporary directory is used. + TempDir string +} + +// NewExtractor creates a new video frame extractor based on the provided configuration. +func NewExtractor(config *Config) (Extractor, error) { + return NewFFMpegExtractor(config) +} diff --git a/internal/models/video/ffmpeg_extractor.go b/internal/models/video/ffmpeg_extractor.go new file mode 100644 index 00000000..3ba8de2a --- /dev/null +++ b/internal/models/video/ffmpeg_extractor.go @@ -0,0 +1,298 @@ +package video + +import ( + "bytes" + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + + "github.com/Tencent/WeKnora/internal/logger" +) + +// FFMpegExtractor implements the Extractor interface using FFmpeg. +type FFMpegExtractor struct { + ffmpegPath string + tempDir string +} + +// NewFFMpegExtractor creates a new FFmpeg-based video extractor. +func NewFFMpegExtractor(config *Config) (Extractor, error) { + ffmpegPath := "ffmpeg" + tempDir := os.TempDir() + + if config != nil { + if config.FFMpegPath != "" { + ffmpegPath = config.FFMpegPath + } + if config.TempDir != "" { + tempDir = config.TempDir + } + } + + extractor := &FFMpegExtractor{ffmpegPath: ffmpegPath, tempDir: tempDir} + if err := extractor.checkFFmpeg(); err != nil { + return nil, err + } + return extractor, nil +} + +// checkFFmpeg verifies that FFmpeg is available. +func (e *FFMpegExtractor) checkFFmpeg() error { + cmd := exec.Command(e.ffmpegPath, "-version") + _, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("ffmpeg not found or not executable: %w", err) + } + return nil +} + +// ExtractFrames extracts frames from a video using FFmpeg. +func (e *FFMpegExtractor) ExtractFrames(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]*Frame, error) { + if options == nil { + options = DefaultExtractOptions() + } + + tempVideoPath, err := e.writeTempVideo(videoBytes) + if err != nil { + return nil, err + } + defer os.Remove(tempVideoPath) + + tempFrameDir, err := os.MkdirTemp(e.tempDir, "video-frames-") + if err != nil { + return nil, fmt.Errorf("failed to create temp frame directory: %w", err) + } + defer os.RemoveAll(tempFrameDir) + + if _, err := e.getVideoDuration(ctx, tempVideoPath); err != nil { + logger.Warnf(ctx, "[VideoFFMpeg] Failed to get video duration: %v", err) + } + + if err := e.extractFramesWithFFmpeg(ctx, tempVideoPath, tempFrameDir, options); err != nil { + return nil, err + } + + return e.readExtractedFrames(ctx, tempFrameDir, options) +} + +// ExtractFramesToReaders extracts frames and returns them as io.Readers. +func (e *FFMpegExtractor) ExtractFramesToReaders(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]io.Reader, error) { + frames, err := e.ExtractFrames(ctx, videoBytes, options) + if err != nil { + return nil, err + } + + readers := make([]io.Reader, 0, len(frames)) + for _, frame := range frames { + readers = append(readers, bytes.NewReader(frame.ImageData)) + } + + return readers, nil +} + +// writeTempVideo writes the video bytes to a temporary file. +func (e *FFMpegExtractor) writeTempVideo(videoBytes []byte) (string, error) { + tempFile, err := os.CreateTemp(e.tempDir, "video-input-*.mp4") + if err != nil { + return "", fmt.Errorf("failed to create temp video file: %w", err) + } + defer tempFile.Close() + + if _, err := tempFile.Write(videoBytes); err != nil { + os.Remove(tempFile.Name()) + return "", fmt.Errorf("failed to write video bytes: %w", err) + } + + return tempFile.Name(), nil +} + +// getVideoDuration gets the duration of a video using FFmpeg. +func (e *FFMpegExtractor) getVideoDuration(ctx context.Context, videoPath string) (float64, error) { + cmd := exec.CommandContext(ctx, e.ffmpegPath, + "-i", videoPath, + "-f", "null", + "-") + + output, err := cmd.CombinedOutput() + if err != nil { + return 0, err + } + + outputStr := string(output) + durationStr := "" + lines := strings.Split(outputStr, "\n") + for _, line := range lines { + if strings.Contains(line, "Duration:") { + parts := strings.Split(line, "Duration:") + if len(parts) > 1 { + durationPart := strings.TrimSpace(parts[1]) + durationStr = strings.Split(durationPart, ",")[0] + break + } + } + } + + if durationStr == "" { + return 0, fmt.Errorf("could not parse duration") + } + + timeParts := strings.Split(durationStr, ":") + if len(timeParts) != 3 { + return 0, fmt.Errorf("invalid duration format: %s", durationStr) + } + + hours, _ := strconv.Atoi(timeParts[0]) + minutes, _ := strconv.Atoi(timeParts[1]) + seconds, _ := strconv.ParseFloat(timeParts[2], 64) + + totalSeconds := float64(hours*3600) + float64(minutes*60) + seconds + return totalSeconds, nil +} + +// extractFramesWithFFmpeg uses FFmpeg to extract frames from a video. +func (e *FFMpegExtractor) extractFramesWithFFmpeg(ctx context.Context, videoPath, outputDir string, options *ExtractOptions) error { + filters := []string{fmt.Sprintf("fps=1/%f", options.FrameInterval)} + + if options.Width > 0 || options.Height > 0 { + scaleFilter := "scale=" + if options.Width > 0 { + scaleFilter += strconv.Itoa(options.Width) + } else { + scaleFilter += "-1" + } + scaleFilter += ":" + if options.Height > 0 { + scaleFilter += strconv.Itoa(options.Height) + } else { + scaleFilter += "-1" + } + filters = append(filters, scaleFilter) + } + + args := []string{ + "-i", videoPath, + "-vf", strings.Join(filters, ","), + } + + quality := options.Quality + if quality < 1 { + quality = 1 + } else if quality > 100 { + quality = 100 + } + if options.Format == "jpeg" { + args = append(args, "-q:v", strconv.Itoa(quality)) + } + + outputPattern := filepath.Join(outputDir, "frame-%05d."+options.Format) + args = append(args, outputPattern) + + cmd := exec.CommandContext(ctx, e.ffmpegPath, args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + + if err := cmd.Run(); err != nil { + return fmt.Errorf("ffmpeg extraction failed: %w, stderr: %s", err, stderr.String()) + } + + logger.Infof(ctx, "[VideoFFMpeg] Extracted frames to: %s", outputDir) + return nil +} + +// readExtractedFrames reads the extracted frames from the temporary directory. +func (e *FFMpegExtractor) readExtractedFrames(ctx context.Context, frameDir string, options *ExtractOptions) ([]*Frame, error) { + files, err := filepath.Glob(filepath.Join(frameDir, "frame-*."+options.Format)) + if err != nil { + return nil, fmt.Errorf("failed to list frame files: %w", err) + } + + if len(files) == 0 { + return nil, fmt.Errorf("no frames were extracted") + } + + frames := make([]*Frame, 0, len(files)) + for i, filePath := range files { + imgData, err := os.ReadFile(filePath) + if err != nil { + logger.Warnf(ctx, "[VideoFFMpeg] Failed to read frame %d: %v", i, err) + continue + } + + frame := &Frame{ + Index: i, + Timestamp: float64(i) * options.FrameInterval, + ImageData: imgData, + Format: options.Format, + Width: options.Width, + Height: options.Height, + } + + frames = append(frames, frame) + + if options.MaxFrames > 0 && len(frames) >= options.MaxFrames { + break + } + } + + if len(frames) == 0 { + return nil, fmt.Errorf("no frames could be read") + } + + logger.Infof(ctx, "[VideoFFMpeg] Successfully read %d frames", len(frames)) + return frames, nil +} + +// GetVideoInfo returns basic information about a video. +func (e *FFMpegExtractor) GetVideoInfo(ctx context.Context, videoBytes []byte) (map[string]interface{}, error) { + tempVideoPath, err := e.writeTempVideo(videoBytes) + if err != nil { + return nil, err + } + defer os.Remove(tempVideoPath) + + duration, err := e.getVideoDuration(ctx, tempVideoPath) + if err != nil { + return nil, err + } + + cmd := exec.CommandContext(ctx, e.ffmpegPath, "-i", tempVideoPath, "-f", "null", "-") + output, _ := cmd.CombinedOutput() + outputStr := string(output) + + width, height := 0, 0 + lines := strings.Split(outputStr, "\n") + for _, line := range lines { + if strings.Contains(line, "Video:") { + parts := strings.Split(line, ",") + for _, part := range parts { + part = strings.TrimSpace(part) + if strings.Contains(part, "x") && !strings.Contains(part, "SAR") { + resParts := strings.Split(part, "x") + if len(resParts) == 2 { + width, _ = strconv.Atoi(strings.TrimSpace(resParts[0])) + heightStr := strings.TrimSpace(resParts[1]) + heightParts := strings.Split(heightStr, " ") + if len(heightParts) > 0 { + height, _ = strconv.Atoi(heightParts[0]) + } + } + break + } + } + break + } + } + + videoInfo := map[string]interface{}{ + "duration": duration, + "width": width, + "height": height, + } + + return videoInfo, nil +} diff --git a/internal/router/task.go b/internal/router/task.go index ff4e99fe..b804028c 100644 --- a/internal/router/task.go +++ b/internal/router/task.go @@ -23,6 +23,7 @@ type AsynqTaskParams struct { ChunkExtractor interfaces.TaskHandler `name:"chunkExtractor"` DataTableSummary interfaces.TaskHandler `name:"dataTableSummary"` ImageMultimodal interfaces.TaskHandler `name:"imageMultimodal"` + VideoMultimodal interfaces.TaskHandler `name:"videoMultimodal"` } func getAsynqRedisClientOpt() *asynq.RedisClientOpt { @@ -109,6 +110,9 @@ func RunAsynqServer(params AsynqTaskParams) *asynq.ServeMux { // Register image multimodal handler mux.HandleFunc(types.TypeImageMultimodal, params.ImageMultimodal.Handle) + // Register video multimodal handler + mux.HandleFunc(types.TypeVideoMultimodal, params.VideoMultimodal.Handle) + // Register data source sync handler mux.HandleFunc(types.TypeDataSourceSync, params.DataSourceService.ProcessSync) diff --git a/internal/types/chunk.go b/internal/types/chunk.go index f283cc86..1ea28919 100644 --- a/internal/types/chunk.go +++ b/internal/types/chunk.go @@ -20,6 +20,12 @@ const ( ChunkTypeImageOCR ChunkType = "image_ocr" // ChunkTypeImageCaption 表示图片描述的 Chunk ChunkTypeImageCaption ChunkType = "image_caption" + // ChunkTypeVideoFrame 表示视频帧分析内容的 Chunk + ChunkTypeVideoFrame ChunkType = "video_frame" + // ChunkTypeVideoCaption 表示视频整体描述的 Chunk + ChunkTypeVideoCaption ChunkType = "video_caption" + // ChunkTypeVideoASR 表示视频音频转录的 Chunk + ChunkTypeVideoASR ChunkType = "video_asr" // ChunkTypeSummary 表示摘要类型的 Chunk ChunkTypeSummary = "summary" // ChunkTypeEntity 表示实体类型的 Chunk @@ -95,6 +101,24 @@ type ImageInfo struct { OCRText string `json:"ocr_text"` } +// VideoInfo 表示与 Chunk 关联的视频信息 +type VideoInfo struct { + // 视频URL + URL string `json:"url" gorm:"type:text"` + // 帧数量 + FrameCount int `json:"frame_count"` + // 是否有VLM分析 + HasVLMAnalysis bool `json:"has_vlm_analysis"` + // 是否有ASR转录 + HasASR bool `json:"has_asr"` + // 视频摘要 + VideoSummary string `json:"video_summary"` + // ASR转录文本 + ASRText string `json:"asr_text"` + // 帧描述列表 + FrameDescriptions []string `json:"frame_descriptions"` +} + // Chunk represents a document chunk // Chunks are meaningful text segments extracted from original documents // and are the basic units of knowledge base retrieval @@ -147,6 +171,8 @@ type Chunk struct { ContentHash string `json:"content_hash" gorm:"type:varchar(64);index"` // 图片信息，存储为 JSON ImageInfo string `json:"image_info" gorm:"type:text"` + // 视频信息，存储为 JSON + VideoInfo string `json:"video_info" gorm:"type:text"` // Chunk creation time CreatedAt time.Time `json:"created_at"` // Chunk last update time diff --git a/internal/types/docparser.go b/internal/types/docparser.go index 861a8af4..5ef46947 100644 --- a/internal/types/docparser.go +++ b/internal/types/docparser.go @@ -22,6 +22,8 @@ type ReadResult struct { Error string IsAudio bool // true when the result contains raw audio data needing ASR transcription AudioData []byte // raw audio bytes for ASR processing + IsVideo bool // true when the result contains raw video data needing multimodal processing + VideoData []byte // raw video bytes for video multimodal processing } // ImageRef represents an image reference extracted from the document. diff --git a/internal/types/extract_graph.go b/internal/types/extract_graph.go index 52c0c6c5..47da6a05 100644 --- a/internal/types/extract_graph.go +++ b/internal/types/extract_graph.go @@ -13,6 +13,7 @@ const ( TypeKnowledgeMove = "knowledge:move" // 知识移动任务 TypeDataTableSummary = "datatable:summary" // 表格摘要任务 TypeImageMultimodal = "image:multimodal" // 图片多模态处理任务（OCR + VLM Caption） + TypeVideoMultimodal = "video:multimodal" // 视频多模态处理任务（抽帧 + VLM + ASR） TypeManualProcess = "manual:process" // 手工知识更新任务（cleanup + 重新索引） TypeDataSourceSync = "datasource:sync" // 数据源同步任务 ) @@ -121,7 +122,7 @@ type KnowledgeMoveProgress struct { SourceKBID string `json:"source_kb_id"` TargetKBID string `json:"target_kb_id"` Status KBCloneTaskStatus `json:"status"` - Progress int `json:"progress"` // 0-100 + Progress int `json:"progress"` // 0-100 Total int `json:"total"` // 总知识数 Processed int `json:"processed"` // 已处理数 Failed int `json:"failed"` // 失败数 @@ -138,8 +139,8 @@ type ManualProcessPayload struct { TenantID uint64 `json:"tenant_id"` KnowledgeID string `json:"knowledge_id"` KnowledgeBaseID string `json:"knowledge_base_id"` - Content string `json:"content"` // cleaned markdown content - NeedCleanup bool `json:"need_cleanup"` // true for update, false for create + Content string `json:"content"` // cleaned markdown content + NeedCleanup bool `json:"need_cleanup"` // true for update, false for create } // ImageMultimodalPayload represents the image multimodal processing task payload. @@ -147,14 +148,28 @@ type ImageMultimodalPayload struct { TenantID uint64 `json:"tenant_id"` KnowledgeID string `json:"knowledge_id"` KnowledgeBaseID string `json:"knowledge_base_id"` - ChunkID string `json:"chunk_id"` // parent text chunk - ImageURL string `json:"image_url"` // provider:// URL (e.g. local://..., minio://...) - ImageLocalPath string `json:"image_local_path"` // deprecated: kept for backward compat with in-flight tasks + ChunkID string `json:"chunk_id"` // parent text chunk + ImageURL string `json:"image_url"` // provider:// URL (e.g. local://..., minio://...) + ImageLocalPath string `json:"image_local_path"` // deprecated: kept for backward compat with in-flight tasks EnableOCR bool `json:"enable_ocr"` EnableCaption bool `json:"enable_caption"` Language string `json:"language,omitempty"` // Request locale for {{language}} in prompt templates } +// VideoMultimodalPayload represents the video multimodal processing task payload. +type VideoMultimodalPayload struct { + TenantID uint64 `json:"tenant_id"` + KnowledgeID string `json:"knowledge_id"` + KnowledgeBaseID string `json:"knowledge_base_id"` + ChunkID string `json:"chunk_id"` // parent text chunk + VideoURL string `json:"video_url"` // provider:// URL (e.g. local://..., minio://...) + VideoLocalPath string `json:"video_local_path"` // deprecated: kept for backward compat with in-flight tasks + EnableVLM bool `json:"enable_vlm"` // enable VLM frame analysis + EnableASR bool `json:"enable_asr"` // enable ASR audio transcription + Language string `json:"language,omitempty"` // Request locale for {{language}} in prompt templates + MaxFrames int `json:"max_frames,omitempty"` // optional: max frames to extract, 0 = no limit +} + // KBCloneTaskStatus represents the status of a knowledge base clone task type KBCloneTaskStatus string diff --git a/migrations/versioned/000032_add_video_info_to_chunks.down.sql b/migrations/versioned/000032_add_video_info_to_chunks.down.sql new file mode 100644 index 00000000..cd9439bb --- /dev/null +++ b/migrations/versioned/000032_add_video_info_to_chunks.down.sql @@ -0,0 +1,7 @@ +-- Migration: 000032_add_video_info_to_chunks (down) +-- Description: Remove video_info column from chunks table. +DO $$ BEGIN RAISE NOTICE '[Migration 000032 down] Removing video_info column from chunks'; END $$; + +ALTER TABLE chunks DROP COLUMN IF EXISTS video_info; + +DO $$ BEGIN RAISE NOTICE '[Migration 000032 down] video_info column removed successfully'; END $$; diff --git a/migrations/versioned/000032_add_video_info_to_chunks.up.sql b/migrations/versioned/000032_add_video_info_to_chunks.up.sql new file mode 100644 index 00000000..e2e355ff --- /dev/null +++ b/migrations/versioned/000032_add_video_info_to_chunks.up.sql @@ -0,0 +1,10 @@ +-- Migration: 000032_add_video_info_to_chunks +-- Description: Add video_info column to chunks table for video multimodal processing. +-- This column stores JSON-serialized VideoInfo struct containing video analysis results. +DO $$ BEGIN RAISE NOTICE '[Migration 000032] Adding video_info column to chunks'; END $$; + +ALTER TABLE chunks ADD COLUMN IF NOT EXISTS video_info TEXT; + +COMMENT ON COLUMN chunks.video_info IS 'Video information in JSON format: {"url": string, "frame_count": int, "has_vlm_analysis": bool, "has_asr": bool, "video_summary": string, "asr_text": string, "frame_descriptions": string[]}'; + +DO $$ BEGIN RAISE NOTICE '[Migration 000032] video_info column added successfully'; END $$;