zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-22 16:36:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Feat: add built in DSL file API (#17003)

Browse Source 
### Summary

1. list and get by id API for builtin DSL
2. add DSL default component param values validation
3. remove all hard code keys for parser config
This commit is contained in:
Jack
2026-07-18 11:00:08 +08:00
committed by GitHub
parent 1c0432a816
commit 10c00a9614
32 changed files with 1824 additions and 673 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/admin/handler.go
View File

@@ -25,7 +25,7 @@ import (
"ragflow/internal/common"
"ragflow/internal/engine"
"ragflow/internal/engine/redis"
"ragflow/internal/httputil"
"ragflow/internal/handler"
"ragflow/internal/server"
"ragflow/internal/service"
"ragflow/internal/storage"
@@ -1039,7 +1039,7 @@ func (h *Handler) RemoveIngestionTasks(c *gin.Context) {
if req.Email == nil && req.Status == nil {
tasks, err := h.service.RemoveIngestionTasks(req.Tasks)
if err != nil {
common.ErrorWithCode(c, httputil.IngestionTaskErrorCode(err), err.Error())
common.ErrorWithCode(c, handler.IngestionTaskErrorCode(err), err.Error())
return
}
@@ -1047,7 +1047,7 @@ func (h *Handler) RemoveIngestionTasks(c *gin.Context) {
} else {
tasks, err := h.service.RemoveIngestionTasksByCondition(req.Tasks, req.Email, req.Status)
if err != nil {
common.ErrorWithCode(c, httputil.IngestionTaskErrorCode(err), err.Error())
common.ErrorWithCode(c, handler.IngestionTaskErrorCode(err), err.Error())
return
}
common.SuccessWithData(c, tasks, "Remove tasks successfully")
@@ -1070,7 +1070,7 @@ func (h *Handler) StopIngestionTasks(c *gin.Context) {
if req.Email == nil && req.Status == nil {
tasks, err := h.service.StopIngestionTasks(req.Tasks)
if err != nil {
common.ErrorWithCode(c, httputil.IngestionTaskErrorCode(err), err.Error())
common.ErrorWithCode(c, handler.IngestionTaskErrorCode(err), err.Error())
return
}
var result []map[string]string

56
internal/handler/document.go
View File

@@ -29,7 +29,6 @@ import (
"path/filepath"
"ragflow/internal/common"
"ragflow/internal/entity"
"ragflow/internal/httputil"
"ragflow/internal/utility"
"strconv"
"strings"
@@ -45,7 +44,6 @@ var IMG_BASE64_PREFIX = "data:image/png;base64,"
// documentServiceIface defines the DocumentService methods used by DocumentHandler.
type documentServiceIface interface {
CreateDocument(req *service.CreateDocumentRequest) (*entity.Document, error)
GetDocumentByID(id string) (*service.DocumentResponse, error)
UpdateDocument(id string, req *service.UpdateDocumentRequest) error
DeleteDocument(id string) error
@@ -98,41 +96,6 @@ func NewDocumentHandler(documentService documentServiceIface, datasetService *se
}
}
// CreateDocument create document
// @Summary Create Document
// @Description Create new document
// @Tags documents
// @Accept json
// @Produce json
// @Param request body service.CreateDocumentRequest true "document info"
// @Success 200 {object} map[string]interface{}
// @Router /api/v1/documents [post]
func (h *DocumentHandler) CreateDocument(c *gin.Context) {
_, errorCode, errorMessage := GetUser(c)
if errorCode != common.CodeSuccess {
common.ErrorWithCode(c, errorCode, errorMessage)
return
}
var req service.CreateDocumentRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{
"error": err.Error(),
})
return
}
document, err := h.documentService.CreateDocument(&req)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{
"error": err.Error(),
})
return
}
common.SuccessWithData(c, document, "created successfully")
}
// GetDocumentByID get document by ID
// @Summary Get Document Info
// @Description Get document details by ID
@@ -987,15 +950,15 @@ func (h *DocumentHandler) uploadWebDocument(c *gin.Context, kb *entity.Knowledge
// mapDocKeysWithRunStatus renames a freshly-created document's raw keys to the
// public response shape (chunk_num→chunk_count, token_num→token_count,
// kb_id→dataset_id, parser_id→chunk_method) and reports run as a label.
// kb_id→dataset_id) and reports run as a label.
// Mirrors Python map_doc_keys_with_run_status / map_doc_keys.
func mapDocKeysWithRunStatus(raw map[string]interface{}) map[string]interface{} {
out := map[string]interface{}{
"chunk_count": raw["chunk_num"],
"token_count": raw["token_num"],
"dataset_id": raw["kb_id"],
"chunk_method": raw["parser_id"],
"run": "UNSTART",
"chunk_count": raw["chunk_num"],
"token_count": raw["token_num"],
"dataset_id": raw["kb_id"],
"parser_id": raw["parser_id"],
"run": "UNSTART",
}
for _, k := range []string{"id", "name", "type", "size", "suffix", "source_type", "created_by", "parser_config", "location", "pipeline_id", "content_hash"} {
if v, ok := raw[k]; ok {
@@ -1058,7 +1021,6 @@ func mapDocumentListItem(doc *entity.DocumentListItem, metaFields map[string]int
"suffix": doc.Suffix,
"run": mapRunStatus(doc.Run),
"status": stringValue(doc.Status),
"chunk_method": doc.ParserID,
"parser_id": doc.ParserID,
"pipeline_id": stringValue(doc.PipelineID),
"pipeline_name": stringValue(doc.PipelineName),
@@ -1409,7 +1371,7 @@ func (h *DocumentHandler) ListIngestionTasks(c *gin.Context) {
parseResult, err = h.documentService.ListIngestionTasks(userID, req.DatasetID, 0, 0)
if err != nil {
common.ResponseWithCodeData(c, httputil.IngestionTaskErrorCode(err), nil, err.Error())
common.ResponseWithCodeData(c, IngestionTaskErrorCode(err), nil, err.Error())
return
}
@@ -1459,7 +1421,7 @@ func (h *DocumentHandler) StopIngestionTasks(c *gin.Context) {
parseResult, err := h.documentService.StopIngestionTasks(req.Tasks, userID)
if err != nil {
common.ResponseWithCodeData(c, httputil.IngestionTaskErrorCode(err), nil, err.Error())
common.ResponseWithCodeData(c, IngestionTaskErrorCode(err), nil, err.Error())
return
}
common.SuccessWithData(c, parseResult, "success")
@@ -1485,7 +1447,7 @@ func (h *DocumentHandler) RemoveIngestionTasks(c *gin.Context) {
deletedTasks, err := h.documentService.RemoveIngestionTasks(req.Tasks, userID)
if err != nil {
common.ResponseWithCodeData(c, httputil.IngestionTaskErrorCode(err), nil, err.Error())
common.ResponseWithCodeData(c, IngestionTaskErrorCode(err), nil, err.Error())
return
}
common.SuccessWithData(c, deletedTasks, "success")

3
internal/handler/document_test.go
View File

@@ -138,9 +138,6 @@ func (f *fakeDocumentService) DownloadDocument(datasetID, docID string) (*servic
FileName: "doc.pdf",
}, nil
}
func (f *fakeDocumentService) CreateDocument(req *service.CreateDocumentRequest) (*entity.Document, error) {
return nil, nil
}
func (f *fakeDocumentService) GetDocumentByID(id string) (*service.DocumentResponse, error) {
if f.docErr != nil {
return nil, f.docErr

19
internal/handler/error.go
View File

@@ -17,9 +17,11 @@
package handler
import (
"errors"
"net/http"
"ragflow/internal/common"
"ragflow/internal/service"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
@@ -66,3 +68,20 @@ func HandleNoRoute(c *gin.Context) {
"error": "Not Found",
})
}
// IngestionTaskErrorCode maps ingestion-task service errors to common.ErrorCode
// for HTTP responses.
func IngestionTaskErrorCode(err error) common.ErrorCode {
var transitionErr *service.InvalidTaskTransitionError
if errors.As(err, &transitionErr) {
return common.CodeConflict
}
var conflictErr *service.TaskStatusConflictError
if errors.As(err, &conflictErr) {
return common.CodeConflict
}
if errors.Is(err, common.ErrTaskNotFound) {
return common.CodeNotFound
}
return common.CodeExceptionError
}

2
internal/httputil/ingestion_task_error_test.go → internal/handler/error_test.go
View File

@@ -1,4 +1,4 @@
package httputil
package handler
import (
"testing"

33
internal/handler/pipeline.go
View File

@@ -24,11 +24,12 @@ import (
pipelinepkg "ragflow/internal/ingestion/pipeline"
)
// builtinPipelineLister is the registry surface PipelineHandler needs.
// builtinPipelineRegistry is the registry surface PipelineHandler needs.
// Defined as an interface so handler tests can inject a fake without
// touching the embedded template FS.
type builtinPipelineLister interface {
List() []*pipelinepkg.BuiltinPipelineMeta
type builtinPipelineRegistry interface {
List() *pipelinepkg.BuiltinPipelineListResponse
Get(ref string) (*pipelinepkg.BuiltinPipeline, bool)
}
// PipelineHandler handles pipeline endpoints.
@@ -36,7 +37,7 @@ type builtinPipelineLister interface {
// Built-in templates are shipped with the binary; user-defined templates
// (canvas DSL) may be added later.
type PipelineHandler struct {
registry builtinPipelineLister
registry builtinPipelineRegistry
}
// NewPipelineHandler builds a PipelineHandler backed by the embedded
@@ -59,7 +60,10 @@ func NewPipelineHandler() *PipelineHandler {
// This is public static data, so no auth is required.
func (h *PipelineHandler) ListPipelines(c *gin.Context) {
if h == nil || h.registry == nil {
common.SuccessWithData(c, []*pipelinepkg.BuiltinPipelineMeta{}, "success")
common.SuccessWithData(c, &pipelinepkg.BuiltinPipelineListResponse{
BuiltinPipelines: []*pipelinepkg.BuiltinPipelineMeta{},
Total: 0,
}, "success")
return
}
@@ -70,3 +74,22 @@ func (h *PipelineHandler) ListPipelines(c *gin.Context) {
common.SuccessWithData(c, h.registry.List(), "success")
}
// GetPipeline GET /api/v1/pipelines/:id
// Returns a single built-in pipeline template with its DSL.
// This is public static data, so no auth is required.
func (h *PipelineHandler) GetPipeline(c *gin.Context) {
if h == nil || h.registry == nil {
common.ErrorWithCode(c, common.CodeDataError, "pipeline not found")
return
}
id := c.Param("id")
tpl, ok := h.registry.Get(id)
if !ok || tpl == nil {
common.ErrorWithCode(c, common.CodeDataError, "pipeline not found")
return
}
common.SuccessWithData(c, map[string]any{"dsl": tpl.DSL}, "success")
}

138
internal/handler/pipeline_test.go
View File

@@ -27,12 +27,29 @@ import (
pipelinepkg "ragflow/internal/ingestion/pipeline"
)
// fakePipelineLister is a test double for builtinPipelineLister.
type fakePipelineLister struct {
// fakePipelineRegistry is a test double for builtinPipelineRegistry.
type fakePipelineRegistry struct {
items []*pipelinepkg.BuiltinPipelineMeta
}
func (f fakePipelineLister) List() []*pipelinepkg.BuiltinPipelineMeta { return f.items }
func (f fakePipelineRegistry) List() *pipelinepkg.BuiltinPipelineListResponse {
return &pipelinepkg.BuiltinPipelineListResponse{
BuiltinPipelines: f.items,
Total: int64(len(f.items)),
}
}
func (f fakePipelineRegistry) Get(ref string) (*pipelinepkg.BuiltinPipeline, bool) {
for _, item := range f.items {
if item.ParserID == ref {
return &pipelinepkg.BuiltinPipeline{
BuiltinPipelineMeta: *item,
DSL: map[string]any{"components": map[string]any{}},
}, true
}
}
return nil, false
}
func pipelineCtx() (*gin.Context, *httptest.ResponseRecorder) {
gin.SetMode(gin.TestMode)
@@ -49,16 +66,14 @@ func TestPipelineHandler_ListPipelines_TypeBuiltin(t *testing.T) {
Title: "General",
Description: "general desc",
Filename: "ingestion_pipeline_general.json",
DSL: map[string]any{"components": map[string]any{}},
},
{
ParserID: "book",
Title: "Book",
Filename: "ingestion_pipeline_book.json",
DSL: map[string]any{"components": map[string]any{}},
},
}
h := &PipelineHandler{registry: fakePipelineLister{items: items}}
h := &PipelineHandler{registry: fakePipelineRegistry{items: items}}
c, w := pipelineCtx()
h.ListPipelines(c)
@@ -68,10 +83,12 @@ func TestPipelineHandler_ListPipelines_TypeBuiltin(t *testing.T) {
}
var resp struct {
Code int `json:"code"`
Data []struct {
ParserID string `json:"parser_id"`
Title string `json:"title"`
DSL map[string]any `json:"dsl"`
Data struct {
Canvas []struct {
ID string `json:"id"`
Title string `json:"title"`
} `json:"canvas"`
Total int64 `json:"total"`
} `json:"data"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
@@ -80,11 +97,14 @@ func TestPipelineHandler_ListPipelines_TypeBuiltin(t *testing.T) {
if resp.Code != int(common.CodeSuccess) {
t.Fatalf("code = %d, want %d", resp.Code, int(common.CodeSuccess))
}
if len(resp.Data) != 2 {
t.Fatalf("len(data) = %d, want 2", len(resp.Data))
if resp.Data.Total != 2 {
t.Fatalf("total = %d, want 2", resp.Data.Total)
}
if resp.Data[0].ParserID != "general" || resp.Data[0].Title != "General" {
t.Errorf("data[0] = %+v", resp.Data[0])
if len(resp.Data.Canvas) != 2 {
t.Fatalf("len(canvas) = %d, want 2", len(resp.Data.Canvas))
}
if resp.Data.Canvas[0].ID != "general" || resp.Data.Canvas[0].Title != "General" {
t.Errorf("canvas[0] = %+v", resp.Data.Canvas[0])
}
}
@@ -92,7 +112,7 @@ func TestPipelineHandler_ListPipelines_NoTypeParam(t *testing.T) {
items := []*pipelinepkg.BuiltinPipelineMeta{
{ParserID: "general", Title: "General"},
}
h := &PipelineHandler{registry: fakePipelineLister{items: items}}
h := &PipelineHandler{registry: fakePipelineRegistry{items: items}}
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
@@ -105,15 +125,21 @@ func TestPipelineHandler_ListPipelines_NoTypeParam(t *testing.T) {
t.Fatalf("status = %d, want 200", w.Code)
}
var resp struct {
Data []struct {
ParserID string `json:"parser_id"`
Data struct {
Canvas []struct {
ID string `json:"id"`
} `json:"canvas"`
Total int64 `json:"total"`
} `json:"data"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if len(resp.Data) != 1 {
t.Fatalf("len(data) = %d, want 1 (no type param defaults to builtin)", len(resp.Data))
if resp.Data.Total != 1 {
t.Fatalf("total = %d, want 1", resp.Data.Total)
}
if len(resp.Data.Canvas) != 1 {
t.Fatalf("len(canvas) = %d, want 1 (no type param defaults to builtin)", len(resp.Data.Canvas))
}
}
@@ -135,19 +161,21 @@ func TestPipelineHandler_ListPipelines_RealRegistry(t *testing.T) {
h.ListPipelines(c)
var resp struct {
Data []struct {
ParserID string `json:"parser_id"`
Data struct {
Canvas []struct {
ID string `json:"id"`
} `json:"canvas"`
} `json:"data"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if len(resp.Data) == 0 {
if len(resp.Data.Canvas) == 0 {
t.Fatal("expected non-empty builtin pipeline list from real registry")
}
seen := map[string]bool{}
for _, item := range resp.Data {
seen[item.ParserID] = true
for _, item := range resp.Data.Canvas {
seen[item.ID] = true
}
if !seen["general"] {
t.Error("real registry listing missing general")
@@ -156,3 +184,65 @@ func TestPipelineHandler_ListPipelines_RealRegistry(t *testing.T) {
t.Error("alias naive must be hidden from listing")
}
}
func TestPipelineHandler_GetPipeline(t *testing.T) {
items := []*pipelinepkg.BuiltinPipelineMeta{
{ParserID: "general", Title: "General"},
}
h := &PipelineHandler{registry: fakePipelineRegistry{items: items}}
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
c.Request = httptest.NewRequest("GET", "/api/v1/pipelines/general", nil)
c.Params = gin.Params{{Key: "id", Value: "general"}}
h.GetPipeline(c)
if w.Code != 200 {
t.Fatalf("status = %d, want 200", w.Code)
}
var resp struct {
Code int `json:"code"`
Data struct {
DSL map[string]any `json:"dsl"`
} `json:"data"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if resp.Code != int(common.CodeSuccess) {
t.Fatalf("code = %d, want %d", resp.Code, int(common.CodeSuccess))
}
if resp.Data.DSL == nil {
t.Fatal("dsl should not be nil")
}
}
func TestPipelineHandler_GetPipeline_NotFound(t *testing.T) {
items := []*pipelinepkg.BuiltinPipelineMeta{
{ParserID: "general", Title: "General"},
}
h := &PipelineHandler{registry: fakePipelineRegistry{items: items}}
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
c.Request = httptest.NewRequest("GET", "/api/v1/pipelines/nonexistent", nil)
c.Params = gin.Params{{Key: "id", Value: "nonexistent"}}
h.GetPipeline(c)
if w.Code != 200 {
t.Fatalf("status = %d, want 200 (error is in body code)", w.Code)
}
var resp struct {
Code int `json:"code"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if resp.Code == int(common.CodeSuccess) {
t.Fatal("expected error code for nonexistent pipeline")
}
}

23
internal/httputil/ingestion_task_error.go
View File

@@ -1,23 +0,0 @@
package httputil
import (
"errors"
"ragflow/internal/common"
"ragflow/internal/service"
)
func IngestionTaskErrorCode(err error) common.ErrorCode {
var transitionErr *service.InvalidTaskTransitionError
if errors.As(err, &transitionErr) {
return common.CodeConflict
}
var conflictErr *service.TaskStatusConflictError
if errors.As(err, &conflictErr) {
return common.CodeConflict
}
if errors.Is(err, common.ErrTaskNotFound) {
return common.CodeNotFound
}
return common.CodeExceptionError
}

60
internal/ingestion/pipeline/builtin_registry.go
View File

@@ -8,6 +8,7 @@ import (
"io/fs"
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
@@ -21,12 +22,20 @@ var builtinTemplateFS embed.FS
// BuiltinPipelineMeta is the API-facing metadata for one built-in ingestion
// pipeline template. The ParserID field is the value stored in the dataset's
// parser_id column for built-in pipelines.
// JSON tag "id" aligns with AgentItem.id for format consistency with
// GET /api/v1/agents?canvas_category=dataflow_canvas.
type BuiltinPipelineMeta struct {
ParserID string `json:"parser_id"`
Title string `json:"title"`
Description string `json:"description,omitempty"`
Filename string `json:"filename"`
DSL map[string]any `json:"dsl"`
ParserID string `json:"id"`
Title string `json:"title"`
Description string `json:"description,omitempty"`
Filename string `json:"filename"`
}
// BuiltinPipelineListResponse wraps the builtin pipeline list for
// format consistency with ListAgentsResponse.
type BuiltinPipelineListResponse struct {
BuiltinPipelines []*BuiltinPipelineMeta `json:"canvas"`
Total int64 `json:"total"`
}
type BuiltinPipeline struct {
@@ -119,7 +128,7 @@ func (r *Registry) canonicalRef(ref string) string {
return ref
}
func (r *Registry) List() []*BuiltinPipelineMeta {
func (r *Registry) List() *BuiltinPipelineListResponse {
if r == nil {
return nil
}
@@ -131,7 +140,10 @@ func (r *Registry) List() []*BuiltinPipelineMeta {
cp := *item
out = append(out, &cp)
}
return out
return &BuiltinPipelineListResponse{
BuiltinPipelines: out,
Total: int64(len(out)),
}
}
func (r *Registry) Refs() []string {
@@ -176,6 +188,39 @@ func (r *Registry) IsValid(ref string) bool {
return ok
}
// fileTypeParserOverrides maps a file type value to the canonical parser_id that
// must be used for all files of that type. The type values are FileType constants
// defined in the utility package.
var fileTypeParserOverrides = map[string]string{
"visual": "picture",
"aural": "audio",
}
var (
presentationExtPattern = regexp.MustCompile(`(?i)\.(ppt|pptx|pages)$`)
emailExtPattern = regexp.MustCompile(`(?i)\.(msg|eml)$`)
)
// DefaultParserID returns the canonical parser_id for a document given its file
// type, filename and the dataset-level parser_id. File-type-based overrides
// (e.g. "visual" → "picture", "aural" → "audio") take precedence, followed by
// filename extension heuristics (presentation / email), then falling back to
// the dataset's parser_id.
func (r *Registry) DefaultParserID(fileType, filename, fallback string) string {
if override, ok := fileTypeParserOverrides[strings.ToLower(fileType)]; ok {
return override
}
base := filepath.Base(strings.TrimSpace(filename))
switch {
case presentationExtPattern.MatchString(base):
return "presentation"
case emailExtPattern.MatchString(base):
return "email"
default:
return fallback
}
}
func parseTemplate(filename string, raw []byte) (*BuiltinPipeline, error) {
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber()
@@ -197,7 +242,6 @@ func parseTemplate(filename string, raw []byte) (*BuiltinPipeline, error) {
Title: englishText(data["title"], ref),
Description: englishText(data["description"], ""),
Filename: filename,
DSL: dsl,
},
DSL: dsl,
}

41
internal/ingestion/pipeline/builtin_registry_test.go
View File

@@ -20,9 +20,12 @@ func TestRegistryLoadsSummaries(t *testing.T) {
if err != nil {
t.Fatalf("NewRegistryFromDir: %v", err)
}
if got := len(r.List()); got != 2 {
if got := len(r.List().BuiltinPipelines); got != 2 {
t.Fatalf("len(List()) = %d, want 2", got)
}
if got := r.List().Total; got != 2 {
t.Fatalf("Total = %d, want 2", got)
}
tpl, ok := r.Get("general")
if !ok {
t.Fatal("expected general template")
@@ -164,7 +167,7 @@ func TestRegistryAliasNaiveResolvesGeneral(t *testing.T) {
t.Error("Refs() contains alias naive; aliases must be hidden from listing")
}
}
for _, meta := range r.List() {
for _, meta := range r.List().BuiltinPipelines {
if meta.ParserID == "naive" {
t.Error("List() contains alias naive; aliases must be hidden from listing")
}
@@ -238,3 +241,37 @@ func TestLoadBuiltinDSL_UnknownFails(t *testing.T) {
t.Fatal("LoadBuiltinDSL(nonexistent) = nil, want error")
}
}
func TestDefaultRegistry_DefaultParserID(t *testing.T) {
r, err := DefaultRegistry()
if err != nil {
t.Fatalf("DefaultRegistry: %v", err)
}
tests := []struct {
name string
fileType string
filename string
fallback string
want string
}{
{name: "visual", fileType: "visual", filename: "img.png", fallback: "naive", want: "picture"},
{name: "aural", fileType: "aural", filename: "audio.mp3", fallback: "naive", want: "audio"},
{name: "presentation by ext", fileType: "doc", filename: "deck.pptx", fallback: "naive", want: "presentation"},
{name: "email by ext", fileType: "doc", filename: "mail.eml", fallback: "naive", want: "email"},
{name: "pages as presentation", fileType: "doc", filename: "slides.pages", fallback: "naive", want: "presentation"},
{name: "msg as email", fileType: "doc", filename: "message.msg", fallback: "naive", want: "email"},
{name: "fallback default", fileType: "doc", filename: "notes.txt", fallback: "manual", want: "manual"},
{name: "unknown filetype fallback", fileType: "unknown", filename: "file.bin", fallback: "general", want: "general"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := r.DefaultParserID(tt.fileType, tt.filename, tt.fallback)
if got != tt.want {
t.Fatalf("DefaultParserID(%q, %q, %q) = %q, want %q",
tt.fileType, tt.filename, tt.fallback, got, tt.want)
}
})
}
}

101
internal/ingestion/pipeline/component_params.go Normal file
View File

@@ -0,0 +1,101 @@
package pipeline
import (
"encoding/json"
"fmt"
)
// ComponentParamsSchema describes the full set of user-configurable parameters
// for one component in a DSL. The frontend uses this to render editable forms;
// the API uses it for validation.
type ComponentParamsSchema struct {
// CpnID is the component instance identifier in the DSL, e.g. "Parser:HipSignsRhyme".
CpnID string `json:"cpn_id"`
// ComponentName is the logical component type, e.g. "Parser", "Tokenizer".
ComponentName string `json:"component_name"`
// ParamsDefaults holds the params map from the DSL with "outputs" excluded.
// Keys are param names (e.g. "setups", "fields", "chunk_token_size");
// values are their DSL-baked defaults.
ParamsDefaults map[string]any `json:"params_defaults"`
}
// ExtractAllComponentParams extracts the params schema for every component
// present in the DSL JSON. It excludes the internal "outputs" key from each
// component's params — those are wire-format definitions, not user-configurable.
func ExtractAllComponentParams(dslJSON []byte) ([]ComponentParamsSchema, error) {
var dsl struct {
Components map[string]struct {
Obj struct {
ComponentName string `json:"component_name"`
Params map[string]any `json:"params"`
} `json:"obj"`
} `json:"components"`
}
if err := json.Unmarshal(dslJSON, &dsl); err != nil {
return nil, fmt.Errorf("ExtractAllComponentParams: decode DSL: %w", err)
}
if dsl.Components == nil {
return nil, fmt.Errorf("ExtractAllComponentParams: DSL has no components")
}
out := make([]ComponentParamsSchema, 0, len(dsl.Components))
for cpnID, comp := range dsl.Components {
params := make(map[string]any, len(comp.Obj.Params))
for k, v := range comp.Obj.Params {
if k == "outputs" {
continue
}
params[k] = deepCopyValue(v)
}
out = append(out, ComponentParamsSchema{
CpnID: cpnID,
ComponentName: comp.Obj.ComponentName,
ParamsDefaults: params,
})
}
return out, nil
}
// deepCopyValue returns a deep copy of a value that might be a nested map or slice.
func deepCopyValue(v any) any {
switch val := v.(type) {
case map[string]any:
cp := make(map[string]any, len(val))
for mk, mv := range val {
cp[mk] = deepCopyValue(mv)
}
return cp
case []any:
cp := make([]any, len(val))
for i, item := range val {
cp[i] = deepCopyValue(item)
}
return cp
default:
return v
}
}
// deepCopyMapStr deep-copies a map[string]any so the caller can safely mutate the result.
func deepCopyMapStr(src map[string]any) map[string]any {
cp := make(map[string]any, len(src))
for k, v := range src {
cp[k] = deepCopyValue(v)
}
return cp
}
// ComponentParamsDefaults extracts component params from DSL JSON and returns
// them in the component_params storage format {cpnID: {param: value}}.
// Values are deep-copied from the DSL so callers can safely mutate the result.
func ComponentParamsDefaults(dslJSON []byte) (map[string]map[string]any, error) {
schemas, err := ExtractAllComponentParams(dslJSON)
if err != nil {
return nil, err
}
out := make(map[string]map[string]any, len(schemas))
for _, s := range schemas {
out[s.CpnID] = deepCopyMapStr(s.ParamsDefaults)
}
return out, nil
}

239
internal/ingestion/pipeline/component_params_test.go Normal file
View File

@@ -0,0 +1,239 @@
package pipeline
import (
"encoding/json"
"io/fs"
"path/filepath"
"slices"
"strings"
"testing"
)
func TestExtractAllComponentParams(t *testing.T) {
// Minimal DSL with all component types (general template pattern).
dsl := map[string]any{
"components": map[string]any{
"File": map[string]any{
"obj": map[string]any{
"component_name": "File",
"params": map[string]any{},
},
},
"Parser:HipSignsRhyme": map[string]any{
"obj": map[string]any{
"component_name": "Parser",
"params": map[string]any{
"outputs": map[string]any{"html": map[string]any{"type": "string"}},
"setups": map[string]any{
"pdf": map[string]any{"parse_method": "deepdoc"},
"spreadsheet": map[string]any{"parse_method": "deepdoc"},
},
},
},
},
"Tokenizer:LegalReadersDecide": map[string]any{
"obj": map[string]any{
"component_name": "Tokenizer",
"params": map[string]any{
"fields": "text",
"filename_embd_weight": 0.1,
"search_method": []any{"embedding", "full_text"},
"outputs": map[string]any{},
},
},
},
},
}
dslJSON, err := json.Marshal(dsl)
if err != nil {
t.Fatalf("marshal fixture: %v", err)
}
schemas, err := ExtractAllComponentParams(dslJSON)
if err != nil {
t.Fatalf("ExtractAllComponentParams: %v", err)
}
// Build a lookup by cpnID for assertion.
byCPN := make(map[string]ComponentParamsSchema, len(schemas))
for _, s := range schemas {
if _, dup := byCPN[s.CpnID]; dup {
t.Errorf("duplicate cpnID %q in result", s.CpnID)
}
byCPN[s.CpnID] = s
}
// --- File ---
file, ok := byCPN["File"]
if !ok {
t.Fatal("missing component: File")
}
if file.ComponentName != "File" {
t.Errorf("File.ComponentName = %q, want %q", file.ComponentName, "File")
}
if len(file.ParamsDefaults) != 0 {
t.Errorf("File.ParamsDefaults should be empty, got %v", file.ParamsDefaults)
}
// --- Parser ---
parser, ok := byCPN["Parser:HipSignsRhyme"]
if !ok {
t.Fatal("missing component: Parser:HipSignsRhyme")
}
if parser.ComponentName != "Parser" {
t.Errorf("Parser.ComponentName = %q, want %q", parser.ComponentName, "Parser")
}
// ParamsDefaults must contain "setups", must NOT contain "outputs".
if _, ok := parser.ParamsDefaults["setups"]; !ok {
t.Errorf("Parser.ParamsDefaults missing key %q", "setups")
}
if _, ok := parser.ParamsDefaults["outputs"]; ok {
t.Errorf("Parser.ParamsDefaults should NOT contain %q (excluded)", "outputs")
}
if setups, ok := parser.ParamsDefaults["setups"].(map[string]any); !ok {
t.Errorf("Parser.ParamsDefaults[\"setups\"] is not a map: %T", parser.ParamsDefaults["setups"])
} else {
if _, ok := setups["pdf"]; !ok {
t.Errorf("Parser setups missing key %q", "pdf")
}
if _, ok := setups["spreadsheet"]; !ok {
t.Errorf("Parser setups missing key %q", "spreadsheet")
}
}
// --- Tokenizer ---
tokenizer, ok := byCPN["Tokenizer:LegalReadersDecide"]
if !ok {
t.Fatal("missing component: Tokenizer:LegalReadersDecide")
}
if tokenizer.ComponentName != "Tokenizer" {
t.Errorf("Tokenizer.ComponentName = %q, want %q", tokenizer.ComponentName, "Tokenizer")
}
// ParamsDefaults must contain "fields", "filename_embd_weight", "search_method".
if _, ok := tokenizer.ParamsDefaults["fields"]; !ok {
t.Errorf("Tokenizer.ParamsDefaults missing key %q", "fields")
}
if _, ok := tokenizer.ParamsDefaults["filename_embd_weight"]; !ok {
t.Errorf("Tokenizer.ParamsDefaults missing key %q", "filename_embd_weight")
}
if _, ok := tokenizer.ParamsDefaults["search_method"]; !ok {
t.Errorf("Tokenizer.ParamsDefaults missing key %q", "search_method")
}
if _, ok := tokenizer.ParamsDefaults["outputs"]; ok {
t.Errorf("Tokenizer.ParamsDefaults should NOT contain %q (excluded)", "outputs")
}
// --- No extra/leaked ---
if len(schemas) != 3 {
t.Errorf("expected 3 components, got %d: %v", len(schemas), componentIDs(schemas))
}
}
func TestExtractAllComponentParams_EmptyDSL(t *testing.T) {
dsl := map[string]any{
"components": map[string]any{},
}
dslJSON, _ := json.Marshal(dsl)
schemas, err := ExtractAllComponentParams(dslJSON)
if err != nil {
t.Fatalf("ExtractAllComponentParams: %v", err)
}
if len(schemas) != 0 {
t.Errorf("expected 0 components, got %d", len(schemas))
}
}
func TestExtractAllComponentParams_InvalidJSON(t *testing.T) {
_, err := ExtractAllComponentParams([]byte("not json"))
if err == nil {
t.Error("expected error for invalid JSON")
}
}
func TestExtractAllComponentParams_MissingComponents(t *testing.T) {
dsl := map[string]any{"globals": map[string]any{}}
dslJSON, _ := json.Marshal(dsl)
_, err := ExtractAllComponentParams(dslJSON)
if err == nil {
t.Error("expected error for DSL missing components")
}
}
func componentIDs(schemas []ComponentParamsSchema) []string {
ids := make([]string, len(schemas))
for i, s := range schemas {
ids[i] = s.CpnID
}
slices.Sort(ids)
return ids
}
func TestExtractAllComponentParams_AllBuiltinTemplates(t *testing.T) {
entries, err := fs.ReadDir(builtinTemplateFS, "template")
if err != nil {
t.Fatalf("read embedded template dir: %v", err)
}
var tested int
for _, entry := range entries {
if entry.IsDir() {
continue
}
name := entry.Name()
if !strings.HasPrefix(name, builtinTemplatePrefix) || !strings.HasSuffix(name, ".json") {
continue
}
tested++
raw, err := fs.ReadFile(builtinTemplateFS, filepath.Join("template", name))
if err != nil {
t.Errorf("%s: read: %v", name, err)
continue
}
// The template file wraps the DSL in {"dsl": {...}}.
var wrapper struct {
DSL json.RawMessage `json:"dsl"`
}
if err := json.Unmarshal(raw, &wrapper); err != nil {
t.Errorf("%s: unmarshal wrapper: %v", name, err)
continue
}
if len(wrapper.DSL) == 0 {
t.Errorf("%s: missing dsl key", name)
continue
}
schemas, err := ExtractAllComponentParams(wrapper.DSL)
if err != nil {
t.Errorf("%s: ExtractAllComponentParams: %v", name, err)
continue
}
if len(schemas) == 0 {
t.Errorf("%s: expected at least 1 component, got 0", name)
continue
}
// Every schema must have a non-empty CpnID and ComponentName.
for _, s := range schemas {
if s.CpnID == "" {
t.Errorf("%s: component %q has empty CpnID", name, s.ComponentName)
}
if s.ComponentName == "" {
t.Errorf("%s: cpnID %q has empty ComponentName", name, s.CpnID)
}
if s.ParamsDefaults == nil {
t.Errorf("%s: cpnID %q has nil ParamsDefaults", name, s.CpnID)
}
// "outputs" must never leak into ParamsDefaults.
if _, ok := s.ParamsDefaults["outputs"]; ok {
t.Errorf("%s: cpnID %q ParamsDefaults contains excluded key %q", name, s.CpnID, "outputs")
}
}
t.Logf("%s: %d components: %v", name, len(schemas), componentIDs(schemas))
}
if tested == 0 {
t.Fatal("no builtin templates found")
}
t.Logf("tested %d builtin templates", tested)
}

2
internal/ingestion/service/doc_state.go
View File

@@ -59,7 +59,7 @@ func (u *docStateUpdater) apply(r *taskpkg.PipelineResult) {
common.Warn(fmt.Sprintf("failed to update document metadata: %v", err))
}
}
if err := u.docSvc.IncrementChunkNum(r.DocID, r.KbID, r.ChunkCount, r.TokenConsumption, 0); err != nil {
if err := u.docSvc.IncrementChunkNum(r.DocID, r.KbID, r.ChunkCount, r.TokenConsumption, r.Duration); err != nil {
common.Warn(fmt.Sprintf("failed to increment chunk num: %v", err))
}
}

11
internal/ingestion/service/execute_task_ack_test.go
View File

@@ -170,13 +170,20 @@ func TestExecuteTask_HeartbeatsInProgressDuringLongTask(t *testing.T) {
go ingestor.executeTask(newAckTaskCtx(context.Background(), taskID, docID, handle))
<-started
time.Sleep(30 * time.Millisecond) // let the ticker fire a few times
close(proceed) // release the long task
// Poll for heartbeats with a generous deadline so the test is resilient
// to slow CI schedulers. The ticker fires every heartbeatInterval (5ms);
// the first tick may be delayed if the goroutine is not scheduled promptly.
heartbeatDeadline := time.Now().Add(2 * time.Second)
for handle.inProgress.Load() == 0 && time.Now().Before(heartbeatDeadline) {
time.Sleep(5 * time.Millisecond)
}
if handle.inProgress.Load() == 0 {
t.Fatal("expected InProgress heartbeats while runDocumentTask was blocked, got 0")
}
close(proceed) // release the long task — only after confirming heartbeats
// Poll for Ack completion (executeTask must finish MarkCompleted + Ack).
deadline := time.Now().Add(2 * time.Second)
for handle.acks.Load() == 0 && time.Now().Before(deadline) {

3
internal/ingestion/service/ingestion_service.go
View File

@@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"ragflow/internal/utility"
"runtime"
"strings"
"sync"
"sync/atomic"
@@ -83,7 +84,7 @@ type Ingestor struct {
func NewIngestor(name string, maxConcurrency int32, supportedTypes []string) *Ingestor {
if maxConcurrency <= 0 {
maxConcurrency = 1
maxConcurrency = int32(runtime.NumCPU())
}
ctx, cancel := context.WithCancel(context.Background())
id := utility.GenerateUUID()

42
internal/ingestion/task/pipeline_executor.go
View File

@@ -40,6 +40,7 @@ type PipelineResult struct {
Metadata map[string]any
ChunkCount int
TokenConsumption int
Duration float64 // pipeline wall-clock seconds
}
type PipelineExecutor struct {
@@ -52,7 +53,7 @@ type PipelineExecutor struct {
loadDSLFunc func(ctx context.Context, canvasID string) (string, string, error)
runPipelineFunc func(ctx context.Context, dsl string) (map[string]any, string, error)
progressSink pipelinepkg.ProgressSink
requireResume bool // when true, defaultRunPipeline passes WithRequireResume to the pipeline
requireResume bool // when true, the pipeline run passes WithRequireResume
}
func validateTaskContext(taskCtx *TaskContext) error {
@@ -148,6 +149,7 @@ func (s *PipelineExecutor) Doc() *entity.Document { return &s.taskCtx.Doc }
func (s *PipelineExecutor) Tenant() *entity.Tenant { return &s.taskCtx.Tenant }
func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error) {
start := time.Now()
if err := ctx.Err(); err != nil {
return nil, err
}
@@ -170,7 +172,7 @@ func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error)
return nil, nil
}
result, err := s.processOutput(ctx, pipelineOutput)
result, err := s.processOutput(ctx, pipelineOutput, start)
if err != nil {
return nil, err
}
@@ -181,7 +183,7 @@ func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error)
return result, nil
}
func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map[string]any) (*PipelineResult, error) {
func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map[string]any, start time.Time) (*PipelineResult, error) {
if pipelineOutput == nil {
return nil, nil
}
@@ -228,6 +230,7 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map
Metadata: metadata,
ChunkCount: len(chunks),
TokenConsumption: embeddingTokenConsumption,
Duration: time.Since(start).Seconds(),
}, nil
}
@@ -281,6 +284,33 @@ func (s *PipelineExecutor) loadDSLFromCanvas(ctx context.Context, canvasID strin
return string(raw), canvasID, nil
}
// warnUnknownComponentParams logs a warning for any component id in the
// parserConfig whose id is absent from the pipeline DSL. The runtime merge
// (component params -> PatchDSL / override_params) silently drops such
// entries, so we surface them here for operability. API-side validation
// already rejects unknown ids on write; this is purely a defensive guard
// for legacy/stale rows.
func warnUnknownComponentParams(dsl string, parserConfig map[string]any) {
if len(parserConfig) == 0 {
return
}
schemas, err := pipelinepkg.ExtractAllComponentParams([]byte(dsl))
if err != nil {
common.Warn(fmt.Sprintf("warnUnknownComponentParams: cannot parse DSL to validate component params: %v", err))
return
}
dslCPNs := make(map[string]struct{}, len(schemas))
for _, s := range schemas {
dslCPNs[s.CpnID] = struct{}{}
}
for cpnID := range parserConfig {
if _, ok := dslCPNs[cpnID]; !ok {
common.Warn(fmt.Sprintf(
"parser_config references cpnID %q not present in the pipeline DSL; it will be ignored at runtime", cpnID))
}
}
}
func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) (map[string]any, string, error) {
if s == nil || s.taskCtx == nil {
return nil, dsl, fmt.Errorf("pipeline executor: nil task context")
@@ -288,6 +318,12 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) (
parserConfig := map[string]interface{}(s.taskCtx.Doc.ParserConfig)
common.InjectExtractorLLMID(parserConfig, s.taskCtx.Tenant.LLMID)
// Surface component params whose cpnID is absent from the DSL. The
// runtime merge (PatchDSL / override_params) silently drops such entries;
// API-side validation already rejects unknown ids on write, so this is a
// defensive guard for legacy/stale rows.
warnUnknownComponentParams(dsl, parserConfig)
patchedDSL, err := pipelinepkg.PatchDSL(dsl, parserConfig)
if err != nil {
return nil, dsl, fmt.Errorf("patch dsl with parser_config: %w", err)

184
internal/ingestion/task/pipeline_executor_defaults_test.go Normal file
View File

File diff suppressed because one or more lines are too long

12
internal/ingestion/task/pipeline_executor_test.go
View File

@@ -261,7 +261,7 @@ func TestRecordPipelineLog_ValidJSONParsed(t *testing.T) {
func TestRunPipeline_NilOutput(t *testing.T) {
svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0)
_, err := svc.processOutput(context.Background(), nil)
_, err := svc.processOutput(context.Background(), nil, time.Now())
if err != nil {
t.Errorf("expected nil error for nil output, got %v", err)
}
@@ -271,7 +271,7 @@ func TestRunPipeline_EmptyOutput(t *testing.T) {
svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0).WithLogCreateFunc(
func(log *entity.PipelineOperationLog) error { return nil },
)
_, err := svc.processOutput(context.Background(), map[string]any{})
_, err := svc.processOutput(context.Background(), map[string]any{}, time.Now())
if err != nil {
t.Errorf("expected nil error for empty output, got %v", err)
}
@@ -281,7 +281,7 @@ func TestRunPipeline_NormalizedEmpty(t *testing.T) {
svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0).WithLogCreateFunc(
func(log *entity.PipelineOperationLog) error { return nil },
)
_, err := svc.processOutput(context.Background(), map[string]any{"markdown": ""})
_, err := svc.processOutput(context.Background(), map[string]any{"markdown": ""}, time.Now())
if err != nil {
t.Errorf("expected nil error for empty normalized output, got %v", err)
}
@@ -299,7 +299,7 @@ func TestRunPipeline_FullFlow(t *testing.T) {
{"text": "world"},
},
}
_, err := svc.processOutput(context.Background(), output)
_, err := svc.processOutput(context.Background(), output, time.Now())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@@ -317,7 +317,7 @@ func TestRunPipeline_AlreadyHasVectors(t *testing.T) {
{"text": "hello", "q_768_vec": []float64{0.1, 0.2}},
},
}
_, err := svc.processOutput(context.Background(), output)
_, err := svc.processOutput(context.Background(), output, time.Now())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@@ -330,7 +330,7 @@ func TestRunPipeline_ContextCanceled(t *testing.T) {
_, err := svc.processOutput(ctx, map[string]any{
"chunks": []map[string]any{{"text": "hello"}},
})
}, time.Now())
if err == nil {
t.Error("expected context canceled error")
}

7
internal/ingestion/task/pipeline_real_integration_test.go
View File

@@ -1,6 +1,3 @@
//go:build integration
// +build integration
package task
import (
@@ -386,7 +383,7 @@ func TestRunPipeline_RealPipelineOutput_ProducesIndexFields(t *testing.T) {
return nil, nil
})
if _, err := svc.processOutput(context.Background(), pipelineOut); err != nil {
if _, err := svc.processOutput(context.Background(), pipelineOut, time.Now()); err != nil {
t.Fatalf("RunPipeline: %v", err)
}
@@ -430,7 +427,7 @@ func taskRepoRoot(t *testing.T) string {
func mustLoadTaskRealIntegrationConfig(t *testing.T) *server.Config {
t.Helper()
if err := common.Init("info", common.FileOutput{}); err != nil {
if err := common.Init("info", common.FileOutput{}, ""); err != nil {
t.Fatalf("init common logger: %v", err)
}
server.SetLogger(zap.NewNop())

2
internal/parser/parser/pdf_parser_cgo.go
View File

@@ -6,6 +6,7 @@ import (
"context"
"errors"
"fmt"
"ragflow/internal/common"
deepdocpdf "ragflow/internal/deepdoc/parser/pdf"
deepdoctype "ragflow/internal/deepdoc/parser/type"
@@ -15,6 +16,7 @@ func (p *PDFParser) ParseWithResult(filename string, data []byte) ParseResult {
if err := p.validateParseMethod(); err != nil {
return ParseResult{Err: err}
}
common.Info(fmt.Sprintf("------------file: %s, parse_method: %s", filename, p.ParseMethod))
switch normalizePDFParseMethod(p.ParseMethod) {
case "plain_text":
return parsePDFWithPlainText(filename, data, p)

2
internal/router/router.go
View File

@@ -167,6 +167,7 @@ func (r *Router) Setup(engine *gin.Engine) {
// Query: ?type=builtin returns built-in templates (default).
if r.pipelineHandler != nil {
apiNoAuth.GET("/pipelines", r.pipelineHandler.ListPipelines)
apiNoAuth.GET("/pipelines/:id", r.pipelineHandler.GetPipeline)
}
// searchbots
@@ -318,7 +319,6 @@ func (r *Router) Setup(engine *gin.Engine) {
// Document routes
documents := v1.Group("/documents")
{
documents.POST("", r.documentHandler.CreateDocument)
documents.POST("/upload", r.documentHandler.UploadInfo)
documents.GET("", r.documentHandler.ListDocuments)
documents.GET("/artifact/:filename", r.documentHandler.GetDocumentArtifact)

5
internal/service/chunk/chunk.go
View File

@@ -631,7 +631,10 @@ func (s *ChunkService) StopParsing(userID, datasetID string, req service.StopPar
return nil, common.CodeDataError, fmt.Errorf("You don't own the document %s", docID)
}
task, _ := dao.NewIngestionTaskDAO().GetByDocumentID(docID)
task, err := dao.NewIngestionTaskDAO().GetByDocumentID(docID)
if err != nil {
return nil, common.CodeServerError, fmt.Errorf("get ingestion task for %s: %w", docID, err)
}
if task == nil || task.Status == common.COMPLETED ||
task.Status == common.STOPPED || task.Status == common.FAILED {
return &service.StopParsingResponse{

469
internal/service/dataset.go
View File

@@ -1548,21 +1548,6 @@ func (d *DatasetService) SearchDatasets(req *SearchDatasetsRequest, userID strin
}, nil
}
// AutoMetadataField mirrors the REST dataset auto metadata field schema.
type AutoMetadataField struct {
Name string `json:"name"`
Type string `json:"type"`
Description *string `json:"description,omitempty"`
Examples interface{} `json:"examples,omitempty"`
RestrictValues bool `json:"restrict_values,omitempty"`
}
// AutoMetadataConfig mirrors the REST dataset auto metadata schema.
type AutoMetadataConfig struct {
Enabled *bool `json:"enabled,omitempty"`
Fields []AutoMetadataField `json:"fields,omitempty"`
}
// MetadataConfigField mirrors one field in the dataset metadata config API.
type MetadataConfigField struct {
Key string `json:"key"`
@@ -1579,17 +1564,14 @@ type MetadataConfigRequest struct {
// CreateDatasetRequest represents the request for creating a dataset.
type CreateDatasetRequest struct {
Name string `json:"name" binding:"required"`
Avatar *string `json:"avatar,omitempty"`
Description *string `json:"description,omitempty"`
EmbeddingModel *string `json:"embedding_model,omitempty"`
Permission *string `json:"permission,omitempty"`
ChunkMethod *string `json:"chunk_method,omitempty"`
ParseType *int `json:"parse_type,omitempty"`
PipelineID *string `json:"pipeline_id,omitempty"`
ParserConfig map[string]interface{} `json:"parser_config,omitempty"`
AutoMetadataConfig *AutoMetadataConfig `json:"auto_metadata_config,omitempty"`
Ext map[string]interface{} `json:"ext,omitempty"`
Name string `json:"name" binding:"required"`
EmbeddingModel *string `json:"embedding_model,omitempty"`
Permission *string `json:"permission,omitempty"`
ParserID *string `json:"parser_id,omitempty"`
PipelineID *string `json:"pipeline_id,omitempty"`
// ParseType indicates pipeline selection mode: 1 = BuiltIn (parser_id),
// 2 = Pipeline (pipeline_id). nil means unspecified (backward compat).
ParseType *int `json:"parse_type,omitempty"`
}
// ListDatasets lists datasets with pagination and filtering.
@@ -1693,42 +1675,42 @@ func (d *DatasetService) CreateDataset(req *CreateDatasetRequest, tenantID strin
return nil, common.CodeDataError, errors.New("Tenant not found.")
}
// parse_type explicitly signals the pipeline mode (1 = BuiltIn, 2 = Pipeline).
// When absent (nil), infer intent from which field is present for backward compat.
isPipelineMode := req.ParseType != nil && *req.ParseType == 2
isBuiltinMode := req.ParseType != nil && *req.ParseType == 1
if isBuiltinMode && req.PipelineID != nil {
// BuiltIn mode: discard pipeline_id so only parser_id matters.
req.PipelineID = nil
}
if isPipelineMode && req.ParserID != nil {
// Pipeline mode: ignore parser_id.
req.ParserID = nil
}
if req.ParseType == nil && req.ParserID != nil && req.PipelineID != nil {
return nil, common.CodeDataError, errors.New("parser_id and pipeline_id are mutually exclusive")
}
parserID := ""
permission := "me"
embeddingModel := ""
parserConfig := req.ParserConfig
pipelineID := req.PipelineID
description := req.Description
avatar := req.Avatar
var language *string
if req.Description != nil && len(*req.Description) > 65535 {
return nil, common.CodeDataError, errors.New("String should have at most 65535 characters")
}
if req.Avatar != nil {
if len(*req.Avatar) > 65535 {
return nil, common.CodeDataError, errors.New("String should have at most 65535 characters")
}
if err := validateDatasetAvatar(*req.Avatar); err != nil {
return nil, common.CodeDataError, err
}
}
if req.Permission != nil {
permission = strings.TrimSpace(*req.Permission)
if permission != "me" && permission != "team" {
return nil, common.CodeDataError, errors.New("Input should be 'me' or 'team'")
}
}
if req.ChunkMethod != nil {
parserID = strings.TrimSpace(*req.ChunkMethod)
if req.ParserID != nil {
parserID = strings.TrimSpace(*req.ParserID)
if err := validateParserID(parserID); err != nil {
return nil, common.CodeDataError, err
}
pipelineID = nil
}
if req.ParseType != nil && (*req.ParseType < 0 || *req.ParseType > 64) {
return nil, common.CodeDataError, fmt.Errorf("Input should be between 0 and 64")
}
if req.PipelineID != nil {
normalizedPipelineID, err := normalizeDatasetPipelineID(*req.PipelineID)
if err != nil {
@@ -1742,151 +1724,27 @@ func (d *DatasetService) CreateDataset(req *CreateDatasetRequest, tenantID strin
return nil, common.CodeDataError, err
}
}
if err := validateDatasetParserConfigSize(parserConfig); err != nil {
return nil, common.CodeDataError, err
}
// ext mirrors the Python REST implementation and overrides known top-level fields.
for key, value := range req.Ext {
switch key {
case "name":
nameValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Dataset name must be string.")
}
nameValue = strings.TrimSpace(nameValue)
if nameValue == "" {
return nil, common.CodeDataError, errors.New("Dataset name can't be empty.")
}
if len(nameValue) > entity.DatasetNameLimit {
return nil, common.CodeDataError, fmt.Errorf("Dataset name length is %d which is large than %d", len(nameValue), entity.DatasetNameLimit)
}
name = nameValue
case "description":
descriptionValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Description must be string.")
}
if len(descriptionValue) > 65535 {
return nil, common.CodeDataError, errors.New("String should have at most 65535 characters")
}
description = &descriptionValue
case "avatar":
avatarValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Avatar must be string.")
}
if len(avatarValue) > 65535 {
return nil, common.CodeDataError, errors.New("String should have at most 65535 characters")
}
if err := validateDatasetAvatar(avatarValue); err != nil {
return nil, common.CodeDataError, err
}
avatar = &avatarValue
case "language":
languageValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Language must be string.")
}
languageValue = strings.TrimSpace(languageValue)
language = &languageValue
case "permission":
permissionValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Permission must be string.")
}
permissionValue = strings.TrimSpace(permissionValue)
if permissionValue != "me" && permissionValue != "team" {
return nil, common.CodeDataError, errors.New("Input should be 'me' or 'team'")
}
permission = permissionValue
case "embedding_model", "embd_id":
embeddingModelValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Embedding model identifier must follow <model_name>@<provider> format")
}
embeddingModelValue = strings.TrimSpace(embeddingModelValue)
if err := validateDatasetEmbeddingModel(embeddingModelValue); err != nil {
return nil, common.CodeDataError, err
}
embeddingModel = embeddingModelValue
case "chunk_method", "parser_id":
parserIDValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, parserIDError()
}
parserIDValue = strings.TrimSpace(parserIDValue)
if err := validateParserID(parserIDValue); err != nil {
return nil, common.CodeDataError, err
}
parserID = parserIDValue
pipelineID = nil
case "pipeline_id":
pipelineIDValue, ok := value.(string)
if !ok {
return nil, common.CodeDataError, errors.New("pipeline_id must be 32 hex characters")
}
normalizedPipelineID, err := normalizeDatasetPipelineID(pipelineIDValue)
if err != nil {
return nil, common.CodeDataError, err
}
pipelineID = normalizedPipelineID
case "parser_config":
parserConfigValue, ok := value.(map[string]interface{})
if !ok {
return nil, common.CodeDataError, errors.New("parser_config must be valid JSON")
}
if err := validateDatasetParserConfigSize(parserConfigValue); err != nil {
return nil, common.CodeDataError, err
}
parserConfig = parserConfigValue
// Reject references to a custom canvas the caller does not own or share.
// The handler passes the caller's user id in tenantID for CreateDataset.
if pipelineID != nil && strings.TrimSpace(*pipelineID) != "" {
if ok, err := canvasAccessibleForUser(tenantID, strings.TrimSpace(*pipelineID)); err != nil {
return nil, common.CodeServerError, err
} else if !ok {
return nil, common.CodeDataError, errors.New("canvas is not accessible")
}
}
// parser_id wins when it is present; otherwise parse_type and pipeline_id must arrive together.
if parserID == "" {
if req.ParseType == nil && pipelineID == nil {
parserID = "naive"
} else if req.ParseType == nil || pipelineID == nil {
missingFields := make([]string, 0, 2)
if req.ParseType == nil {
missingFields = append(missingFields, "parse_type")
}
if pipelineID == nil {
missingFields = append(missingFields, "pipeline_id")
}
return nil, common.CodeDataError, fmt.Errorf("parser_id omitted -> required fields missing: %s", strings.Join(missingFields, ", "))
}
// Resolve component params defaults from the DSL template. parser_config
// stores component params directly: {cpnID: {param: value}}.
parserConfig, cpErr := resolveComponentParamsDefaults(parserID, pipelineID)
if cpErr != nil {
common.Warn("failed to resolve component params defaults for dataset",
zap.String("parserID", parserID), zap.Error(cpErr))
parserConfig = entity.JSONMap{}
}
if req.AutoMetadataConfig != nil {
parserConfig = applyAutoMetadataConfig(parserConfig, req.AutoMetadataConfig)
}
// Build parser_config. When pipeline_id is set, seed from pipeline DSL defaults.
var parserConfigMap map[string]interface{}
if pipelineID != nil {
if parserConfig == nil {
parserConfig = make(map[string]interface{})
}
canvas, err := dao.NewUserCanvasDAO().GetByID(*pipelineID)
if err != nil || canvas == nil {
return nil, common.CodeDataError, fmt.Errorf("pipeline %s not found", *pipelineID)
}
pipelineDefaults := common.ExtractPipelineDefaults(canvas.DSL)
if pipelineDefaults != nil {
parserConfigMap = common.DeepMergeMaps(pipelineDefaults, parserConfig)
} else {
parserConfigMap = common.DeepMergeMaps(nil, parserConfig)
}
common.InjectExtractorLLMID(parserConfigMap, tenant.LLMID)
} else {
parserConfigMap = common.GetParserConfig(parserID, parserConfig)
parserConfigMap["llm_id"] = tenant.LLMID
}
if err := validateDatasetParserConfigSize(parserConfigMap); err != nil {
return nil, common.CodeDataError, err
}
var parserConfigMap map[string]interface{} = parserConfig
embdID := tenant.EmbdID
tenantEmbdID := ptrStringValue(tenant.TenantEmbdID)
@@ -1908,7 +1766,6 @@ func (d *DatasetService) CreateDataset(req *CreateDatasetRequest, tenantID strin
}
kbID := utility.GenerateToken()
status := string(entity.StatusValid)
// Deduplicate name within tenant
duplicateName, err := common.DuplicateName(func(n, tid string) bool {
@@ -1933,16 +1790,6 @@ func (d *DatasetService) CreateDataset(req *CreateDatasetRequest, tenantID strin
Status: &status,
}
if description != nil {
kb.Description = description
}
if avatar != nil {
kb.Avatar = avatar
}
if language != nil {
kb.Language = language
}
if err = d.kbDAO.Create(kb); err != nil {
return nil, common.CodeServerError, errors.New("Failed to save dataset")
}
@@ -2080,21 +1927,21 @@ type DatasetConnectorRequest struct {
}
type UpdateDatasetRequest struct {
Name *string `json:"name,omitempty"`
Avatar *string `json:"avatar,omitempty"`
Description *string `json:"description,omitempty"`
Language *string `json:"language,omitempty"`
Connectors *[]DatasetConnectorRequest `json:"connectors,omitempty"`
EmbdID *string `json:"embd_id,omitempty"`
EmbeddingModel *string `json:"embedding_model,omitempty"`
Permission *string `json:"permission,omitempty"`
ParserID *string `json:"parser_id,omitempty"`
ChunkMethod *string `json:"chunk_method,omitempty"`
Pagerank *int64 `json:"pagerank,omitempty"`
ParserConfig map[string]interface{} `json:"parser_config,omitempty"`
PipelineID *string `json:"pipeline_id,omitempty"`
AutoMetadataConfig *AutoMetadataConfig `json:"auto_metadata_config,omitempty"`
Ext map[string]interface{} `json:"ext,omitempty"`
Name *string `json:"name,omitempty"`
Avatar *string `json:"avatar,omitempty"`
Description *string `json:"description,omitempty"`
Language *string `json:"language,omitempty"`
Connectors *[]DatasetConnectorRequest `json:"connectors,omitempty"`
EmbdID *string `json:"embd_id,omitempty"`
EmbeddingModel *string `json:"embedding_model,omitempty"`
Permission *string `json:"permission,omitempty"`
ParserID *string `json:"parser_id,omitempty"`
Pagerank *int64 `json:"pagerank,omitempty"`
ParserConfig map[string]interface{} `json:"parser_config,omitempty"`
PipelineID *string `json:"pipeline_id,omitempty"`
// ParseType indicates pipeline selection mode: 1 = BuiltIn (parser_id),
// 2 = Pipeline (pipeline_id). nil means unspecified (backward compat).
ParseType *int `json:"parse_type,omitempty"`
}
// UpdateDataset Update a dataset
@@ -2118,7 +1965,6 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
}
updates := make(map[string]interface{})
extUpdates := normalizeDatasetUpdateExt(req.Ext)
if req.Name != nil {
name := strings.TrimSpace(*req.Name)
@@ -2159,6 +2005,19 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
}
updates["permission"] = permission
}
// parse_type explicitly signals the pipeline mode (1 = BuiltIn, 2 = Pipeline).
// When absent (nil), existing mutual-exclusivity behavior is preserved.
isPipelineMode := req.ParseType != nil && *req.ParseType == 2
isBuiltinMode := req.ParseType != nil && *req.ParseType == 1
if isBuiltinMode && req.PipelineID != nil {
req.PipelineID = nil
}
if isPipelineMode && req.ParserID != nil {
req.ParserID = nil
}
if req.PipelineID != nil {
pipelineID, err := normalizeDatasetPipelineID(*req.PipelineID)
if err != nil {
@@ -2169,52 +2028,23 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
}
}
for key, value := range extUpdates {
if _, exists := updates[key]; !exists {
updates[key] = value
}
}
parserID, parserIDProvided, err := datasetUpdateParserID(req)
if err != nil {
return nil, common.CodeDataError, err
}
if !parserIDProvided {
if extParserID, ok := updates["parser_id"]; ok {
parserIDValue, ok := extParserID.(string)
if !ok {
return nil, common.CodeDataError, parserIDError()
}
parserID = strings.TrimSpace(parserIDValue)
if err := validateParserID(parserID); err != nil {
return nil, common.CodeDataError, err
}
parserIDProvided = true
}
}
if parserIDProvided {
updates["parser_id"] = parserID
}
// When parse_type is absent, parser_id and pipeline_id are mutually exclusive.
if req.ParseType == nil && parserIDProvided && req.PipelineID != nil {
return nil, common.CodeDataError, errors.New("parser_id and pipeline_id are mutually exclusive")
}
embdID, embdIDProvided, err := datasetUpdateEmbeddingID(req)
if err != nil {
return nil, common.CodeDataError, err
}
if !embdIDProvided {
if extEmbdID, ok := updates["embd_id"]; ok {
embdIDValue, ok := extEmbdID.(string)
if !ok {
return nil, common.CodeDataError, errors.New("Embedding model identifier must follow <model_name>@<provider> format")
}
embdID = strings.TrimSpace(embdIDValue)
if embdID != "" {
if err := validateDatasetEmbeddingModel(embdID); err != nil {
return nil, common.CodeDataError, err
}
}
embdIDProvided = true
}
}
if embdIDProvided {
tenantEmbdID := ptrStringValue(kb.TenantEmbdID)
if embdID == "" {
@@ -2236,19 +2066,37 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
updates["tenant_embd_id"] = stringPtrIfNotEmpty(tenantEmbdID)
}
if req.AutoMetadataConfig != nil {
req.ParserConfig = applyAutoMetadataConfig(req.ParserConfig, req.AutoMetadataConfig)
}
if req.ParserConfig != nil {
if err := validateDatasetParserConfigSize(req.ParserConfig); err != nil {
return nil, common.CodeDataError, err
}
if len(req.ParserConfig) > 0 {
merged := common.DeepMergeMaps(kb.ParserConfig, req.ParserConfig)
if err := validateDatasetParserConfigSize(merged); err != nil {
return nil, common.CodeDataError, err
// Resolve the effective pipeline and load its DSL schema.
effectiveParserID := kb.ParserID
if parserIDProvided {
effectiveParserID = parserID
}
effectivePipelineID := kb.PipelineID
if req.PipelineID != nil {
if normalized, err := normalizeDatasetPipelineID(*req.PipelineID); err == nil {
effectivePipelineID = normalized
}
} else if parserIDProvided && kb.PipelineID != nil {
effectivePipelineID = nil
}
isCanvas := effectivePipelineID != nil && strings.TrimSpace(*effectivePipelineID) != ""
dslJSON, dslErr := loadPipelineDSL(isCanvas, effectiveParserID, effectivePipelineID)
if dslErr != nil {
common.Warn("failed to load pipeline DSL for building parser_config",
zap.String("parserID", effectiveParserID), zap.Error(dslErr))
}
if dslJSON != nil {
// Start from DSL defaults, overlay the cleaned incoming overrides.
// Unknown cpnIDs, unknown params, and legacy flat fields are
// dropped. Full replace — no merge with the stored config.
updates["parser_config"] = buildParserConfig(dslJSON, map[string]interface{}(req.ParserConfig))
}
updates["parser_config"] = entity.JSONMap(merged)
}
}
@@ -2273,12 +2121,17 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
if parserIDProvided && parserID != kb.ParserID {
if _, ok := updates["parser_config"]; !ok {
updates["parser_config"] = entity.JSONMap(common.GetParserConfig(parserID, nil))
if resolved, cpErr := resolveComponentParamsDefaults(parserID, nil); cpErr != nil {
common.Warn("failed to resolve component params defaults on parser_id switch",
zap.String("parserID", parserID), zap.Error(cpErr))
} else if resolved != nil {
updates["parser_config"] = resolved
}
}
}
if kb.PipelineID != nil && parserIDProvided {
if _, ok := updates["pipeline_id"]; !ok {
updates["pipeline_id"] = ""
updates["pipeline_id"] = nil // clear to NULL, not empty string
}
}
@@ -2290,28 +2143,17 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req UpdateDat
if parserIDProvided {
cfgParserID = parserID
}
parserConfigMap := common.GetParserConfig(cfgParserID, nil)
if upPipelineID, ok := updates["pipeline_id"].(string); ok && upPipelineID != "" {
canvas, err := dao.NewUserCanvasDAO().GetByID(upPipelineID)
if err != nil || canvas == nil {
return nil, common.CodeDataError, fmt.Errorf("pipeline %s not found", upPipelineID)
}
pipelineDefaults := common.ExtractPipelineDefaults(canvas.DSL)
if pipelineDefaults != nil {
parserConfigMap = common.DeepMergeMaps(pipelineDefaults, parserConfigMap)
}
cfgPipelineID, _ := updates["pipeline_id"].(string)
var cpPipelineID *string
if cfgPipelineID != "" {
cpPipelineID = &cfgPipelineID
}
if req.ParserConfig != nil && len(req.ParserConfig) > 0 {
parserConfigMap = common.DeepMergeMaps(parserConfigMap, req.ParserConfig)
if cpDefaults, cpErr := resolveComponentParamsDefaults(cfgParserID, cpPipelineID); cpErr != nil {
common.Warn("failed to resolve component params defaults on pipeline change",
zap.String("parserID", cfgParserID), zap.Error(cpErr))
} else if cpDefaults != nil {
updates["parser_config"] = cpDefaults
}
tenant, err := d.tenantDAO.GetByID(tenantID)
if err == nil && tenant != nil {
common.InjectExtractorLLMID(parserConfigMap, tenant.LLMID)
}
if err := validateDatasetParserConfigSize(parserConfigMap); err != nil {
return nil, common.CodeDataError, err
}
updates["parser_config"] = entity.JSONMap(parserConfigMap)
}
if nameValue, ok := updates["name"].(string); ok && strings.ToLower(nameValue) != strings.ToLower(kb.Name) {
@@ -2379,10 +2221,6 @@ func datasetUpdateParserID(req UpdateDatasetRequest) (string, bool, error) {
parserID = strings.TrimSpace(*req.ParserID)
provided = true
}
if req.ChunkMethod != nil {
parserID = strings.TrimSpace(*req.ChunkMethod)
provided = true
}
if !provided {
return "", false, nil
}
@@ -3114,30 +2952,6 @@ func (d *DatasetService) verifyEmbeddingAvailability(embdID string, tenantID str
return true, ""
}
func applyAutoMetadataConfig(parserConfig map[string]interface{}, config *AutoMetadataConfig) map[string]interface{} {
if parserConfig == nil {
parserConfig = make(map[string]interface{})
}
fields := make([]map[string]interface{}, 0, len(config.Fields))
for _, field := range config.Fields {
fields = append(fields, map[string]interface{}{
"name": field.Name,
"type": field.Type,
"description": field.Description,
"examples": field.Examples,
"restrict_values": field.RestrictValues,
})
}
parserConfig["metadata"] = fields
enableMetadata := true
if config.Enabled != nil {
enableMetadata = *config.Enabled
}
parserConfig["enable_metadata"] = enableMetadata
return parserConfig
}
func parserConfigValueOrEmptyList(parserConfig map[string]interface{}, key string) interface{} {
if parserConfig == nil {
return []interface{}{}
@@ -3191,7 +3005,7 @@ func datasetListItemToMap(kb *entity.KnowledgebaseListItem) map[string]interface
"document_count": kb.DocNum,
"token_num": kb.TokenNum,
"chunk_count": kb.ChunkNum,
"chunk_method": kb.ParserID,
"parser_id": kb.ParserID,
"embedding_model": kb.EmbdID,
"nickname": kb.Nickname,
}
@@ -3228,7 +3042,7 @@ func datasetToMap(kb *entity.Knowledgebase) map[string]interface{} {
"chunk_count": kb.ChunkNum,
"similarity_threshold": kb.SimilarityThreshold,
"vector_similarity_weight": kb.VectorSimilarityWeight,
"chunk_method": kb.ParserID,
"parser_id": kb.ParserID,
"parser_config": kb.ParserConfig,
"pagerank": kb.Pagerank,
"create_time": kb.CreateTime,
@@ -3329,3 +3143,48 @@ func (d *DatasetService) RenameTag(datasetID, userID, fromTag, toTag string) (ma
func (d *DatasetService) GetFieldMap(ids []string) (map[string]interface{}, error) {
return d.kbDAO.GetFieldMap(ids)
}
// resolveComponentParamsDefaults loads the DSL for the target pipeline and
// returns the component params defaults as an entity.JSONMap {cpnID: {param: value}}.
// For builtin templates the DSL is loaded from the embedded registry; for custom
// canvas pipelines it is loaded from the canvas row in the database.
func resolveComponentParamsDefaults(parserID string, pipelineID *string) (entity.JSONMap, error) {
isCanvas := pipelineID != nil && strings.TrimSpace(*pipelineID) != ""
var cp map[string]map[string]any
var err error
if isCanvas {
dslJSON, lerr := loadCanvasDSLJSON(strings.TrimSpace(*pipelineID))
if lerr != nil {
return nil, fmt.Errorf("load canvas DSL: %w", lerr)
}
cp, err = pipelinepkg.ComponentParamsDefaults(dslJSON)
} else {
registry, regErr := pipelinepkg.DefaultRegistry()
if regErr != nil {
return nil, fmt.Errorf("builtin registry: %w", regErr)
}
if !registry.IsValid(parserID) {
return nil, fmt.Errorf("unknown builtin parser_id: %q", parserID)
}
dslStr, dslErr := pipelinepkg.LoadBuiltinDSL(parserID)
if dslErr != nil {
return nil, fmt.Errorf("load builtin DSL: %w", dslErr)
}
cp, err = pipelinepkg.ComponentParamsDefaults([]byte(dslStr))
}
if err != nil {
return nil, err
}
out := make(entity.JSONMap, len(cp))
for k, v := range cp {
out[k] = v
}
return out, nil
}
// canvasAccessibleForUser reports whether the given canvas is owned by or
// team-shared with the caller.
func canvasAccessibleForUser(userID, canvasID string) (bool, error) {
tenantIDs, _ := dao.NewUserTenantDAO().GetTenantIDsByUserID(userID)
return dao.NewUserCanvasDAO().Accessible(canvasID, userID, tenantIDs), nil
}

207
internal/service/dataset_create_test.go Normal file
View File

@@ -0,0 +1,207 @@
//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package service
import (
"strings"
"testing"
"ragflow/internal/common"
"ragflow/internal/dao"
"ragflow/internal/entity"
)
// testDatasetCreateService builds a DatasetService for CreateDataset tests.
// CreateDataset resolves the tenant via d.tenantDAO, so it must be wired here
// (unlike the update path which does not touch it).
// insertCreateDatasetTenant seeds a tenant with status="1" (required by
// TenantDAO.GetByID, which CreateDataset calls) for the given tenant id.
func insertCreateDatasetTenant(t *testing.T, tenantID string) {
t.Helper()
var existing entity.Tenant
if err := dao.DB.Where("id = ?", tenantID).First(&existing).Error; err != nil {
tn := &entity.Tenant{
ID: tenantID,
LLMID: "llm-default",
EmbdID: "embd-default",
TenantEmbdID: sptr("embd-1"),
ASRID: "asr-default",
Status: sptr("1"),
}
if err := dao.DB.Create(tn).Error; err != nil {
t.Fatalf("insert test tenant: %v", err)
}
}
}
func testDatasetCreateService(t *testing.T) *DatasetService {
t.Helper()
return &DatasetService{
kbDAO: dao.NewKnowledgebaseDAO(),
documentDAO: dao.NewDocumentDAO(),
connectorDAO: dao.NewConnectorDAO(),
tenantDAO: dao.NewTenantDAO(),
}
}
func TestCreateDataset_NoComponentParams(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertCreateDatasetTenant(t, "tenant-1")
chunkMethod := "naive"
result, code, err := testDatasetCreateService(t).CreateDataset(&CreateDatasetRequest{
Name: "ds-no-cp",
ParserID: &chunkMethod,
}, "tenant-1")
if err != nil {
t.Fatalf("CreateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
if result["parser_id"] != strings.TrimSpace(chunkMethod) {
t.Fatalf("expected parser_id %q, got %#v", chunkMethod, result["parser_id"])
}
}
func TestCreateDataset_ComponentParamsPopulated(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertCreateDatasetTenant(t, "tenant-1")
chunkMethod := "naive"
result, code, err := testDatasetCreateService(t).CreateDataset(&CreateDatasetRequest{
Name: "ds-with-cp-defaults",
ParserID: &chunkMethod,
}, "tenant-1")
if err != nil {
t.Fatalf("CreateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
parserConfig, ok := result["parser_config"].(entity.JSONMap)
if !ok {
t.Fatalf("parser_config is not entity.JSONMap, got %T", result["parser_config"])
}
if len(parserConfig) == 0 {
t.Fatal("parser_config is empty, expected DSL component params defaults")
}
// Verify at least one component from the general/naive template has defaults.
// The general template has TokenChunker, Tokenizer, Parser, and File components.
found := false
for cpnID := range parserConfig {
if strings.Contains(cpnID, "TokenChunker") {
found = true
break
}
}
if !found {
t.Errorf("parser_config does not contain any TokenChunker: %v", parserConfig)
}
}
func TestCreateDataset_ParseTypeBuiltinClearsPipelineID(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertCreateDatasetTenant(t, "tenant-1")
// Seed a canvas so it exists, but parse_type=1 should ignore it.
seedDatasetUpdateCanvas(t, "abcdef0123456789abcdef0123456789", "tenant-1",
datasetUpdateCanvasDSL("Parser:HipSignsRhyme", "chunk_token_num"))
chunkMethod := "naive"
pipelineID := "ABCDEF0123456789ABCDEF0123456789"
parseType := 1
result, code, err := testDatasetCreateService(t).CreateDataset(&CreateDatasetRequest{
Name: "ds-builtin-clears-pipeline",
ParserID: &chunkMethod,
PipelineID: &pipelineID,
ParseType: &parseType,
}, "tenant-1")
if err != nil {
t.Fatalf("CreateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
// parse_type=1 clears pipeline_id → only parser_id should be persisted.
if result["parser_id"] != chunkMethod {
t.Fatalf("expected parser_id %q, got %#v", chunkMethod, result["parser_id"])
}
if pid, ok := result["pipeline_id"]; ok && pid != nil && pid != "" {
t.Fatalf("expected pipeline_id to be cleared for BuiltIn mode, got %#v", pid)
}
}
func TestCreateDataset_ParseTypePipelineIgnoresParserID(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertCreateDatasetTenant(t, "tenant-1")
seedDatasetUpdateCanvas(t, "abcdef0123456789abcdef0123456789", "tenant-1",
datasetUpdateCanvasDSL("Parser:CustomP", "chunk_token_num"))
chunkMethod := "book"
pipelineID := "ABCDEF0123456789ABCDEF0123456789"
parseType := 2
result, code, err := testDatasetCreateService(t).CreateDataset(&CreateDatasetRequest{
Name: "ds-pipeline-ignores-parser",
ParserID: &chunkMethod,
PipelineID: &pipelineID,
ParseType: &parseType,
}, "tenant-1")
if err != nil {
t.Fatalf("CreateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
// parse_type=2 ignores parser_id → pipeline_id should be persisted;
// parser_id should fall back to the default ("naive") since it wasn't set.
if result["pipeline_id"] != strings.ToLower(pipelineID) {
t.Fatalf("expected pipeline_id %q, got %#v", strings.ToLower(pipelineID), result["pipeline_id"])
}
}
func TestCreateDataset_RejectsBothWithoutParseType(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertCreateDatasetTenant(t, "tenant-1")
chunkMethod := "naive"
pipelineID := "abcdef0123456789abcdef0123456789"
_, code, err := testDatasetCreateService(t).CreateDataset(&CreateDatasetRequest{
Name: "ds-both-no-parse-type",
ParserID: &chunkMethod,
PipelineID: &pipelineID,
// ParseType deliberately nil
}, "tenant-1")
if err == nil {
t.Fatal("expected mutual-exclusivity error when both set without parse_type")
}
if code != common.CodeDataError {
t.Fatalf("expected data error code, got %d", code)
}
if !strings.Contains(err.Error(), "mutually exclusive") {
t.Fatalf("expected error to mention 'mutually exclusive', got: %v", err)
}
}

345
internal/service/dataset_update_test.go
View File

@@ -17,6 +17,7 @@
package service
import (
"encoding/json"
"strings"
"testing"
@@ -39,24 +40,14 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
permission := string(entity.TenantPermissionTeam)
chunkMethod := string(entity.ParserTypeBook)
embeddingModel := "BAAI/bge-large-zh-v1.5@Builtin"
pipelineID := "ABCDEF0123456789ABCDEF0123456789"
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
Name: &name,
Description: &description,
Language: &language,
Permission: &permission,
ChunkMethod: &chunkMethod,
ParserID: &chunkMethod,
EmbeddingModel: &embeddingModel,
PipelineID: &pipelineID,
ParserConfig: map[string]interface{}{
"parent_child": map[string]interface{}{
"use_parent_child": true,
},
"ext": map[string]interface{}{
"delimiter": "\n\n",
},
},
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
@@ -67,15 +58,12 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
if result["name"] != "Renamed Dataset" {
t.Fatalf("expected trimmed name in response, got %#v", result["name"])
}
if result["chunk_method"] != chunkMethod {
t.Fatalf("expected chunk method %q, got %#v", chunkMethod, result["chunk_method"])
if result["parser_id"] != chunkMethod {
t.Fatalf("expected parser_id %q, got %#v", chunkMethod, result["parser_id"])
}
if result["embedding_model"] != embeddingModel {
t.Fatalf("expected embedding model %q, got %#v", embeddingModel, result["embedding_model"])
}
if result["pipeline_id"] != strings.ToLower(pipelineID) {
t.Fatalf("expected normalized pipeline id, got %#v", result["pipeline_id"])
}
if connectors, ok := result["connectors"].([]*dao.ConnectorDatasetListItem); !ok || len(connectors) != 0 {
t.Fatalf("expected empty connector list, got %#v", result["connectors"])
}
@@ -99,14 +87,92 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
if persisted.EmbdID != embeddingModel {
t.Fatalf("expected embd id %q, got %q", embeddingModel, persisted.EmbdID)
}
if persisted.PipelineID == nil || *persisted.PipelineID != strings.ToLower(pipelineID) {
t.Fatalf("expected normalized pipeline id persisted, got %#v", persisted.PipelineID)
// parser_config stores DSL runtime component params directly.
if _, ok := persisted.ParserConfig["Parser:HipSignsRhyme"].(map[string]interface{}); !ok {
t.Fatalf("expected Parser:HipSignsRhyme in parser_config, got %#v", persisted.ParserConfig)
}
if pc, ok := persisted.ParserConfig["parent_child"].(map[string]interface{}); !ok || pc["use_parent_child"] != true {
t.Fatalf("expected parent_child preserved as nested, got %#v", persisted.ParserConfig)
}
func TestUpdateDataset_RejectsSimultaneousParserIDAndPipelineID(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
chunkMethod := "book"
pipelineID := "abcdef0123456789abcdef0123456789"
_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserID: &chunkMethod,
PipelineID: &pipelineID,
})
if err == nil {
t.Fatal("expected mutual-exclusivity error when both parser_id and pipeline_id are set")
}
if pc, ok := persisted.ParserConfig["ext"].(map[string]interface{}); !ok || pc["delimiter"] != "\n\n" {
t.Fatalf("expected ext preserved as nested, got %#v", persisted.ParserConfig)
if code != common.CodeDataError {
t.Fatalf("expected data error code, got %d", code)
}
if !strings.Contains(err.Error(), "mutually exclusive") {
t.Fatalf("expected error to mention 'mutually exclusive', got: %v", err)
}
}
func TestUpdateDataset_ParseTypeBuiltinClearsPipelineID(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
seedDatasetUpdateCanvas(t, "abcdef0123456789abcdef0123456789", "tenant-1",
datasetUpdateCanvasDSL("Parser:HipSignsRhyme", "chunk_token_num"))
chunkMethod := "book"
pipelineID := "ABCDEF0123456789ABCDEF0123456789"
parseType := 1
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserID: &chunkMethod,
PipelineID: &pipelineID,
ParseType: &parseType,
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
// parse_type=1 clears pipeline_id → only parser_id should be set.
if result["parser_id"] != chunkMethod {
t.Fatalf("expected parser_id %q, got %#v", chunkMethod, result["parser_id"])
}
if pid, ok := result["pipeline_id"]; ok && pid != nil && pid != "" {
t.Fatalf("expected pipeline_id to be cleared for BuiltIn mode, got %#v", pid)
}
}
func TestUpdateDataset_ParseTypePipelineIgnoresParserID(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
seedDatasetUpdateCanvas(t, "abcdef0123456789abcdef0123456789", "tenant-1",
datasetUpdateCanvasDSL("Parser:CustomP", "chunk_token_num"))
chunkMethod := "book"
pipelineID := "ABCDEF0123456789ABCDEF0123456789"
parseType := 2
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserID: &chunkMethod,
PipelineID: &pipelineID,
ParseType: &parseType,
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
// parse_type=2 ignores parser_id → pipeline_id should be set;
// parser_id should keep the original value.
if result["pipeline_id"] != strings.ToLower(pipelineID) {
t.Fatalf("expected pipeline_id %q, got %#v", strings.ToLower(pipelineID), result["pipeline_id"])
}
}
@@ -466,3 +532,238 @@ func insertDatasetUpdateTenantModel(t *testing.T, id, providerID, instanceID, mo
t.Fatalf("insert test tenant model: %v", err)
}
}
// seedDatasetUpdateCanvas migrates user_canvas on the active test DB and
// inserts a canvas row with the given DSL.
func seedDatasetUpdateCanvas(t *testing.T, id, userID string, dslJSON []byte) {
t.Helper()
if err := dao.DB.AutoMigrate(&entity.UserCanvas{}); err != nil {
t.Fatalf("migrate user_canvas: %v", err)
}
var dslMap map[string]any
if err := json.Unmarshal(dslJSON, &dslMap); err != nil {
t.Fatalf("unmarshal seed dsl: %v", err)
}
canvas := &entity.UserCanvas{
ID: id,
UserID: userID,
Tags: "",
Permission: "me",
CanvasCategory: "agent_canvas",
DSL: entity.JSONMap(dslMap),
}
if err := dao.DB.Create(canvas).Error; err != nil {
t.Fatalf("seed canvas: %v", err)
}
}
// datasetUpdateCanvasDSL builds a minimal canvas DSL declaring a Parser
// component with the given cpnID and param keys.
func datasetUpdateCanvasDSL(cpnID string, paramKeys ...string) []byte {
params := map[string]any{"outputs": map[string]any{}}
for _, k := range paramKeys {
params[k] = map[string]any{}
}
dsl := map[string]any{
"components": map[string]any{
cpnID: map[string]any{
"obj": map[string]any{"component_name": "Parser", "params": params},
},
},
}
raw, _ := json.Marshal(dsl)
return raw
}
// --- Step 3: component_params validation wired into UpdateDataset ---
// insertDatasetUpdateCanvasKB seeds a KB bound to a custom canvas pipeline
// (PipelineID set, ParserID empty) for the canvas validation tests.
func insertDatasetUpdateCanvasKB(t *testing.T, id, tenantID, name, pipelineID string) {
t.Helper()
pid := pipelineID
kb := &entity.Knowledgebase{
ID: id,
TenantID: tenantID,
Name: name,
EmbdID: "BAAI/bge-large-zh-v1.5@Builtin",
CreatedBy: tenantID,
Permission: string(entity.TenantPermissionMe),
ParserID: "",
PipelineID: &pid,
ParserConfig: entity.JSONMap{"chunk_token_num": float64(128)},
Status: sptr(string(entity.StatusValid)),
}
if err := dao.DB.Create(kb).Error; err != nil {
t.Fatalf("insert test canvas kb: %v", err)
}
}
func TestUpdateDataset_StripsUnknownParam_Builtin(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserConfig: map[string]interface{}{
"Parser:HipSignsRhyme": map[string]interface{}{
"no_such_param": 1,
},
},
})
if err != nil {
t.Fatalf("UpdateDataset should succeed (unknown params are stripped): %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
persisted, err := dao.NewKnowledgebaseDAO().GetByID("kb-1")
if err != nil {
t.Fatalf("get updated kb: %v", err)
}
cp := map[string]interface{}(persisted.ParserConfig)
rhyme, ok := cp["Parser:HipSignsRhyme"].(map[string]interface{})
if !ok {
t.Fatalf("expected Parser:HipSignsRhyme persisted, got %#v", cp)
}
if _, exists := rhyme["no_such_param"]; exists {
t.Fatal("expected no_such_param to be stripped")
}
if result["parser_id"] != string(entity.ParserTypeNaive) {
t.Fatalf("expected parser_id preserved, got %#v", result["parser_id"])
}
}
func TestUpdateDataset_AcceptsValidComponentParams_Builtin(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserConfig: map[string]interface{}{
"Parser:HipSignsRhyme": map[string]interface{}{
"pdf": map[string]interface{}{"parse_method": "deepdoc"},
},
},
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
if result["parser_id"] != string(entity.ParserTypeNaive) {
t.Fatalf("expected parser_id preserved, got %#v", result["parser_id"])
}
persisted, err := dao.NewKnowledgebaseDAO().GetByID("kb-1")
if err != nil {
t.Fatalf("get updated kb: %v", err)
}
cp := map[string]interface{}(persisted.ParserConfig)
if len(cp) == 0 {
t.Fatalf("expected component_params persisted, got %#v", persisted.ParserConfig)
}
rhyme, ok := cp["Parser:HipSignsRhyme"].(map[string]interface{})
if !ok {
t.Fatalf("expected Parser:HipSignsRhyme persisted, got %#v", cp)
}
pdf, ok := rhyme["pdf"].(map[string]interface{})
if !ok || pdf["parse_method"] != "deepdoc" {
t.Fatalf("expected pdf setup persisted, got %#v", rhyme)
}
}
func TestUpdateDataset_StripsCanvasUnknownParam(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
dsl := datasetUpdateCanvasDSL("Parser:CustomRhyme", "pdf")
seedDatasetUpdateCanvas(t, "canvas-1", "tenant-1", dsl)
insertDatasetUpdateCanvasKB(t, "kb-1", "tenant-1", "Original", "canvas-1")
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserConfig: map[string]interface{}{
"Parser:NoSuch": map[string]interface{}{
"pdf": map[string]interface{}{},
},
},
})
if err != nil {
t.Fatalf("UpdateDataset should succeed (unknown cpnID is stripped): %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
persisted, err := dao.NewKnowledgebaseDAO().GetByID("kb-1")
if err != nil {
t.Fatalf("get updated kb: %v", err)
}
cp := map[string]interface{}(persisted.ParserConfig)
if _, exists := cp["Parser:NoSuch"]; exists {
t.Fatal("expected unknown cpnID Parser:NoSuch to be stripped")
}
// The valid cpnID from the canvas DSL should still be present.
if _, exists := cp["Parser:CustomRhyme"]; !exists {
t.Fatal("expected Parser:CustomRhyme (from canvas DSL defaults) to be present")
}
if result["pipeline_id"] != "canvas-1" {
t.Fatalf("expected pipeline_id preserved, got %#v", result["pipeline_id"])
}
}
func TestUpdateDataset_AcceptsValidCanvasComponentParams(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
dsl := datasetUpdateCanvasDSL("Parser:CustomRhyme", "pdf")
seedDatasetUpdateCanvas(t, "canvas-1", "tenant-1", dsl)
insertDatasetUpdateCanvasKB(t, "kb-1", "tenant-1", "Original", "canvas-1")
result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserConfig: map[string]interface{}{
"Parser:CustomRhyme": map[string]interface{}{
"pdf": map[string]interface{}{},
},
},
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
if result["pipeline_id"] != "canvas-1" {
t.Fatalf("expected pipeline_id preserved, got %#v", result["pipeline_id"])
}
}
// TestUpdateDataset_SwitchCanvasToBuiltinValidatesAgainstBuiltin covers the
// effective-ref edge where a request parser_id clears the existing canvas
// pipeline: the override must be validated against the new builtin template,
// not the stale canvas.
func TestUpdateDataset_SwitchCanvasToBuiltinValidatesAgainstBuiltin(t *testing.T) {
db := setupDatasetUpdateTestDB(t)
pushServiceDB(t, db)
dsl := datasetUpdateCanvasDSL("Parser:CustomRhyme", "pdf")
seedDatasetUpdateCanvas(t, "canvas-1", "tenant-1", dsl)
insertDatasetUpdateCanvasKB(t, "kb-1", "tenant-1", "Original", "canvas-1")
chunkMethod := "naive"
_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", UpdateDatasetRequest{
ParserID: &chunkMethod,
ParserConfig: map[string]interface{}{
// Valid for the "general" builtin template, not the canvas.
"Parser:HipSignsRhyme": map[string]interface{}{
"pdf": map[string]interface{}{"parse_method": "deepdoc"},
},
},
})
if err != nil {
t.Fatalf("UpdateDataset failed: %v", err)
}
if code != common.CodeSuccess {
t.Fatalf("expected success code, got %d", code)
}
}

301
internal/service/document.go
View File

@@ -38,6 +38,7 @@ import (
"ragflow/internal/engine"
enginetypes "ragflow/internal/engine/types"
"ragflow/internal/entity"
pipelinepkg "ragflow/internal/ingestion/pipeline"
"ragflow/internal/storage"
"ragflow/internal/tokenizer"
"ragflow/internal/utility"
@@ -85,15 +86,6 @@ func NewDocumentService() *DocumentService {
}
// CreateDocumentRequest create document request
type CreateDocumentRequest struct {
Name string `json:"name" binding:"required"`
KbID string `json:"kb_id" binding:"required"`
ParserID string `json:"parser_id" binding:"required"`
CreatedBy string `json:"created_by" binding:"required"`
Type string `json:"type"`
Source string `json:"source"`
}
// UpdateDocumentRequest update document request
type UpdateDocumentRequest struct {
Name *string `json:"name"`
@@ -102,8 +94,6 @@ type UpdateDocumentRequest struct {
ChunkNum *int64 `json:"chunk_num"`
Progress *float64 `json:"progress"`
ProgressMsg *string `json:"progress_msg"`
// FIXME: need to confirm below field
ProcessDuration *float64 `json:"progress_duration"`
}
// DocumentResponse document response
@@ -147,7 +137,6 @@ type ArtifactResponse struct {
type UpdateDatasetDocumentRequest struct {
Name *string `json:"name"`
ChunkMethod *string `json:"chunk_method"`
ParserID *string `json:"parser_id"`
ChunkCount *int64 `json:"chunk_count"`
TokenCount *int64 `json:"token_count"`
@@ -163,7 +152,7 @@ type UpdateDatasetDocumentResponse struct {
ID string `json:"id"`
Thumbnail *string `json:"thumbnail,omitempty"`
DatasetID string `json:"dataset_id"`
ChunkMethod string `json:"chunk_method"`
ParserID string `json:"parser_id"`
PipelineID *string `json:"pipeline_id,omitempty"`
ParserConfig map[string]interface{} `json:"parser_config"`
SourceType string `json:"source_type"`
@@ -551,27 +540,6 @@ func (s *DocumentService) DownloadDocument(datasetID, docID string) (*DownloadDo
}
// CreateDocument create document
func (s *DocumentService) CreateDocument(req *CreateDocumentRequest) (*entity.Document, error) {
document := &entity.Document{
ID: utility.GenerateUUID(),
Name: &req.Name,
KbID: req.KbID,
ParserID: req.ParserID,
ParserConfig: entity.JSONMap{},
CreatedBy: req.CreatedBy,
Type: req.Type,
SourceType: req.Source,
Suffix: ".doc",
Status: func() *string { s := "1"; return &s }(),
}
if err := s.InsertDocument(document); err != nil {
return nil, fmt.Errorf("failed to create document: %w", err)
}
return document, nil
}
// GetDocumentByID get document by ID
func (s *DocumentService) GetDocumentByID(id string) (*DocumentResponse, error) {
document, err := s.documentDAO.GetByID(id)
@@ -1222,29 +1190,6 @@ func (s *DocumentService) StartParseDocuments(doc *entity.Document, kb *entity.K
}
}
if opts.ApplyKB {
if doc.ParserConfig == nil {
doc.ParserConfig = entity.JSONMap{}
}
config := map[string]interface{}{
"llm_id": kb.ParserConfig["llm_id"],
"enable_metadata": false,
"metadata": map[string]interface{}{},
}
if value, ok := kb.ParserConfig["enable_metadata"]; ok {
config["enable_metadata"] = value
}
if value, ok := kb.ParserConfig["metadata"]; ok {
config["metadata"] = value
}
if err := s.updateDocumentParserConfig(doc.ID, config); err != nil {
return err
}
for key, value := range config {
doc.ParserConfig[key] = value
}
}
if _, err := s.IngestDocuments(doc.KbID, userID, []string{doc.ID}); err != nil {
return err
}
@@ -2201,7 +2146,7 @@ func (s *DocumentService) UpdateDatasetDocument(userID, datasetID, documentID st
return nil, common.CodeServerError, err
}
if code, err := s.validateDatasetDocumentUpdate(doc, req, present); err != nil {
if code, err := s.validateDatasetDocumentUpdate(datasetID, documentID, userID, doc, req, present); err != nil {
return nil, code, err
}
@@ -2218,24 +2163,61 @@ func (s *DocumentService) UpdateDatasetDocument(userID, datasetID, documentID st
}
if present["parser_config"] && req.ParserConfig != nil {
if err := s.updateDocumentParserConfig(doc.ID, req.ParserConfig); err != nil {
return nil, common.CodeDataError, err
// Resolve effective pipeline to load the DSL for cleaning.
isCanvas := kb.PipelineID != nil && strings.TrimSpace(*kb.PipelineID) != ""
if req.PipelineID != nil {
isCanvas = strings.TrimSpace(*req.PipelineID) != ""
}
if req.ParserID != nil {
isCanvas = false
}
effParserID := kb.ParserID
if req.ParserID != nil {
effParserID = strings.TrimSpace(*req.ParserID)
}
effPipelineID := kb.PipelineID
if req.PipelineID != nil {
effPipelineID = req.PipelineID
}
if req.ParserID != nil && req.PipelineID == nil && kb.PipelineID != nil {
effPipelineID = nil
}
dslJSON, err := loadPipelineDSL(isCanvas, effParserID, effPipelineID)
if err != nil {
common.Warn("cleanAndUpdateDocumentParserConfig: failed to load DSL, falling back to merge",
zap.Error(err))
if err := s.updateDocumentParserConfig(doc.ID, req.ParserConfig); err != nil {
return nil, common.CodeDataError, err
}
} else {
cleaned := buildParserConfig(dslJSON, map[string]interface{}(req.ParserConfig))
if err := s.documentDAO.UpdateByID(doc.ID, map[string]interface{}{
"parser_config": cleaned,
}); err != nil {
return nil, common.CodeDataError, err
}
}
}
if req.PipelineID != nil && *req.PipelineID != "" {
if err := s.resetDocumentForReparse(doc, kb.TenantID, nil, req.PipelineID); err != nil {
return nil, common.CodeDataError, err
if present["pipeline_id"] {
if req.PipelineID != nil && strings.TrimSpace(*req.PipelineID) != "" {
if err := s.resetDocumentForReparse(doc, kb.TenantID, nil, req.PipelineID); err != nil {
return nil, common.CodeDataError, err
}
} else {
// Explicitly cleared: drop the custom canvas so the worker falls
// back to the built-in template, matching validation.
empty := ""
if err := s.resetDocumentForReparse(doc, kb.TenantID, nil, &empty); err != nil {
return nil, common.CodeDataError, err
}
}
} else if present["parser_id"] && req.ParserID != nil && strings.TrimSpace(*req.ParserID) != "" {
parserID := strings.TrimSpace(*req.ParserID)
if err := s.resetDocumentForReparse(doc, kb.TenantID, &parserID, nil); err != nil {
return nil, common.CodeDataError, err
}
} else if req.ChunkMethod != nil && *req.ChunkMethod != "" {
if err := s.updateChunkMethod(doc, kb.TenantID, *req.ChunkMethod, req.ParserConfig, present["parser_config"]); err != nil {
return nil, common.CodeDataError, err
}
}
if present["enabled"] && req.Enabled != nil {
@@ -2260,7 +2242,7 @@ func (s *DocumentService) UpdateDatasetDocument(userID, datasetID, documentID st
return s.toUpdateDatasetDocumentResponse(updatedDoc, metaFields), common.CodeSuccess, nil
}
func (s *DocumentService) validateDatasetDocumentUpdate(doc *entity.Document, req *UpdateDatasetDocumentRequest, present map[string]bool) (common.ErrorCode, error) {
func (s *DocumentService) validateDatasetDocumentUpdate(datasetID, documentID, userID string, doc *entity.Document, req *UpdateDatasetDocumentRequest, present map[string]bool) (common.ErrorCode, error) {
if req == nil {
return common.CodeDataError, errors.New("Invalid request payload")
}
@@ -2280,18 +2262,6 @@ func (s *DocumentService) validateDatasetDocumentUpdate(doc *entity.Document, re
}
}
if present["chunk_method"] {
if req.ChunkMethod == nil || strings.TrimSpace(*req.ChunkMethod) == "" {
return common.CodeDataError, errors.New("`chunk_method` (empty string) is not valid")
}
chunkMethod := strings.TrimSpace(*req.ChunkMethod)
if err := validateParserID(chunkMethod); err != nil {
return common.CodeDataError, fmt.Errorf("`chunk_method` %s doesn't exist", chunkMethod)
}
if doc.Type == "visual" || isPresentationFile(doc.Name) {
return common.CodeDataError, errors.New("Not supported yet!")
}
}
if present["parser_id"] && req.ParserID != nil {
parserID := strings.TrimSpace(*req.ParserID)
if (doc.Type == "visual" && parserID != "picture") || (isPresentationFile(doc.Name) && parserID != "presentation") {
@@ -2446,6 +2416,138 @@ func (s *DocumentService) updateDocumentNameOnly(doc *entity.Document, tenantID,
)
}
// loadCanvasDSLJSON returns the DSL JSON for a custom canvas pipeline. The
// canvas's dsl column holds the same component-graph structure that built-in
// templates use, so it can be validated by the same schema extractor. It is a
// package-level function so both document and knowledge-base updates reuse it.
func loadCanvasDSLJSON(canvasID string) ([]byte, error) {
if strings.TrimSpace(canvasID) == "" {
return nil, fmt.Errorf("empty canvas id")
}
canvas, err := dao.NewUserCanvasDAO().GetByID(canvasID)
if err != nil {
if errors.Is(err, dao.ErrUserCanvasNotFound) {
return nil, fmt.Errorf("canvas %s not found", canvasID)
}
return nil, fmt.Errorf("load canvas %s: %w", canvasID, err)
}
if len(canvas.DSL) == 0 {
return nil, fmt.Errorf("canvas %s has no DSL", canvasID)
}
raw, err := json.Marshal(canvas.DSL)
if err != nil {
return nil, fmt.Errorf("marshal canvas %s DSL: %w", canvasID, err)
}
return raw, nil
}
// loadPipelineDSL loads the DSL JSON for a pipeline identified by parserID
// (built-in) or pipelineID (custom canvas). When both are provided, isCanvas
// selects which one to use.
func loadPipelineDSL(isCanvas bool, parserID string, pipelineID *string) ([]byte, error) {
if isCanvas {
return loadCanvasDSLJSON(strings.TrimSpace(*pipelineID))
}
registry, err := pipelinepkg.DefaultRegistry()
if err != nil {
return nil, fmt.Errorf("builtin pipeline registry: %w", err)
}
if !registry.IsValid(parserID) {
return nil, fmt.Errorf("unknown builtin parser_id: %s", parserID)
}
dslStr, err := pipelinepkg.LoadBuiltinDSL(parserID)
if err != nil {
return nil, fmt.Errorf("load builtin DSL for %q: %w", parserID, err)
}
return []byte(dslStr), nil
}
// cleanComponentParams filters rawConfig against the DSL schema given by dslJSON.
// Keys containing ':' are treated as component IDs; they are kept only when both
// the cpnID AND the param name exist in the DSL schema. Keys without ':' (legacy
// flat fields such as chunk_token_num, image_context_size) are dropped with a
// warning — they do not belong in the new component-params world.
func cleanComponentParams(dslJSON []byte, rawConfig map[string]interface{}) map[string]interface{} {
schemas, err := pipelinepkg.ExtractAllComponentParams(dslJSON)
if err != nil {
common.Warn("cleanComponentParams: failed to extract DSL schema, returning input as-is",
zap.Error(err))
return rawConfig
}
validCPNs := make(map[string]map[string]struct{}, len(schemas))
for _, s := range schemas {
keys := make(map[string]struct{}, len(s.ParamsDefaults))
for k := range s.ParamsDefaults {
keys[k] = struct{}{}
}
validCPNs[s.CpnID] = keys
}
result := make(map[string]interface{}, len(rawConfig))
for key, val := range rawConfig {
if !strings.Contains(key, ":") {
common.Warn("cleanComponentParams: dropping legacy flat field",
zap.String("key", key))
continue
}
validKeys, ok := validCPNs[key]
if !ok {
common.Warn("cleanComponentParams: dropping unknown cpnID",
zap.String("cpnID", key))
continue
}
params, ok := val.(map[string]any)
if !ok {
continue
}
cleaned := make(map[string]any, len(params))
for pk, pv := range params {
if _, ok := validKeys[pk]; ok {
cleaned[pk] = pv
} else {
common.Warn("cleanComponentParams: dropping unknown param",
zap.String("cpnID", key), zap.String("param", pk))
}
}
if len(cleaned) > 0 {
result[key] = cleaned
}
}
return result
}
// buildParserConfig builds the final parser_config by starting from the DSL
// defaults for every component, then overlaying the cleaned incoming overrides.
// This ensures all components from the current pipeline are present while
// stripping stale params from other pipelines.
func buildParserConfig(dslJSON []byte, rawConfig map[string]interface{}) entity.JSONMap {
cleaned := cleanComponentParams(dslJSON, rawConfig)
defaults, err := pipelinepkg.ComponentParamsDefaults(dslJSON)
if err != nil {
common.Warn("buildParserConfig: failed to extract DSL defaults, using cleaned only",
zap.Error(err))
return entity.JSONMap(cleaned)
}
result := make(entity.JSONMap, len(defaults))
for cpnID, params := range defaults {
base := make(map[string]interface{}, len(params))
for k, v := range params {
base[k] = v
}
if over, ok := cleaned[cpnID]; ok {
if om, ok := over.(map[string]any); ok {
result[cpnID] = common.DeepMergeMaps(base, map[string]interface{}(om))
} else {
result[cpnID] = base
}
} else {
result[cpnID] = base
}
}
return result
}
func (s *DocumentService) updateDocumentParserConfig(documentID string, config map[string]any) error {
if len(config) == 0 {
return nil
@@ -2598,22 +2700,6 @@ func (s *DocumentService) decrementDocumentAndKBCountersForReparse(doc *entity.D
return decremented, err
}
func (s *DocumentService) updateChunkMethod(doc *entity.Document, tenantID string, chunkMethod string, parserConfig map[string]any, hasParserConfig bool) error {
chunkMethod = strings.TrimSpace(chunkMethod)
if !strings.EqualFold(doc.ParserID, chunkMethod) {
if err := s.resetDocumentForReparse(doc, tenantID, &chunkMethod, nil); err != nil {
return err
}
}
if !hasParserConfig {
defaultConfig := common.GetParserConfig(chunkMethod, nil)
if err := s.updateDocumentParserConfig(doc.ID, defaultConfig); err != nil {
return err
}
}
return nil
}
func (s *DocumentService) updateDocumentStatusOnly(doc *entity.Document, kb *entity.Knowledgebase, status int) error {
statusStr := strconv.Itoa(status)
if doc.Status != nil && *doc.Status == statusStr {
@@ -2646,7 +2732,7 @@ func (s *DocumentService) toUpdateDatasetDocumentResponse(doc *entity.Document,
ID: doc.ID,
Thumbnail: doc.Thumbnail,
DatasetID: doc.KbID,
ChunkMethod: doc.ParserID,
ParserID: doc.ParserID,
PipelineID: doc.PipelineID,
ParserConfig: map[string]interface{}(doc.ParserConfig),
SourceType: doc.SourceType,
@@ -2991,11 +3077,15 @@ func (s *DocumentService) newDatasetDocument(kb *entity.Knowledgebase, tenantID,
if i := strings.LastIndex(filename, "."); i >= 0 {
suffix = filename[i+1:]
}
parserID := selectUploadParser(utility.FileType(filetype), filename, kb.ParserID)
if kb.PipelineID != nil {
parserID = "" // canvas pipeline mode — parser_id not applicable
}
loc := location
doc := &entity.Document{
ID: docID,
KbID: kb.ID,
ParserID: selectUploadParser(utility.FileType(filetype), filename, kb.ParserID),
ParserID: parserID,
PipelineID: kb.PipelineID,
ParserConfig: parserConfig,
CreatedBy: tenantID,
@@ -3012,11 +3102,24 @@ func (s *DocumentService) newDatasetDocument(kb *entity.Knowledgebase, tenantID,
hash := contentHashHex(blob)
doc.ContentHash = &hash
}
// When the document's builtin parser_id differs from the KB's (e.g. visual→picture,
// aural→audio), re-resolve component_params defaults from the document's own DSL
// template so cpnIDs and param keys match the pipeline that will actually execute.
if kb.PipelineID == nil && parserID != kb.ParserID {
if cp, err := resolveComponentParamsDefaults(parserID, nil); err != nil {
common.Warn("newDatasetDocument: resolve component_params defaults",
zap.String("parserID", parserID), zap.Error(err))
} else if cp != nil {
doc.ParserConfig = cp
}
}
return doc
}
// docToRawMap serialises a freshly created Document into the raw key shape the
// handler remaps (chunk_num→chunk_count, kb_id→dataset_id, parser_id→chunk_method).
// handler remaps (chunk_num→chunk_count, kb_id→dataset_id).
func docToRawMap(doc *entity.Document) map[string]interface{} {
m := map[string]interface{}{
"id": doc.ID,

80
internal/service/document_test.go
View File

@@ -567,29 +567,6 @@ func insertTestFile(t *testing.T, id, parentID, name string, location *string) {
}
}
func TestCreateDocumentIncrementsKBDocNum(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
insertTestKB(t, "kb-create", "tenant-1", 0, 0, 0)
svc := testDocumentService(t)
doc, err := svc.CreateDocument(&CreateDocumentRequest{
Name: "created.txt",
KbID: "kb-create",
ParserID: "naive",
CreatedBy: "tenant-1",
Type: "doc",
Source: "local",
})
if err != nil {
t.Fatalf("CreateDocument failed: %v", err)
}
if doc == nil || doc.KbID != "kb-create" {
t.Fatalf("unexpected doc: %+v", doc)
}
assertKBDocNum(t, "kb-create", 1)
}
func TestDeleteDocumentFull_Basic(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
@@ -812,8 +789,8 @@ func TestUploadLocalDocuments_MirrorsPythonCoreFields(t *testing.T) {
if doc["location"] != "nested/path/deck(1).pptx" {
t.Fatalf("location=%v, want nested/path/deck(1).pptx", doc["location"])
}
if doc["parser_id"] != "presentation" {
t.Fatalf("parser_id=%v, want presentation", doc["parser_id"])
if doc["parser_id"] != "" {
t.Fatalf("parser_id=%v, want empty (canvas pipeline mode)", doc["parser_id"])
}
if doc["content_hash"] != "06b05ab6733a618578af5f94892f3950" {
t.Fatalf("content_hash=%v", doc["content_hash"])
@@ -860,7 +837,7 @@ func TestUploadEmptyDocument_CreatesVirtualDocumentAndFileLink(t *testing.T) {
if code != common.CodeSuccess {
t.Fatalf("code=%v, want success", code)
}
if got["type"] != "virtual" || got["parser_id"] != "manual" || got["size"] != int64(0) {
if got["type"] != "virtual" || got["parser_id"] != "" || got["size"] != int64(0) {
t.Fatalf("unexpected doc map: %v", got)
}
@@ -1723,7 +1700,7 @@ func TestUpdateDatasetDocumentRenameUpdatesDocumentAndFile(t *testing.T) {
}
}
func TestUpdateDatasetDocumentChunkMethodResetsForReparse(t *testing.T) {
func TestUpdateDatasetDocumentParserIDResetsForReparse(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
@@ -1732,12 +1709,12 @@ func TestUpdateDatasetDocumentChunkMethodResetsForReparse(t *testing.T) {
chunkMethod := "manual"
svc := testDocumentService(t)
resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
ChunkMethod: &chunkMethod,
}, map[string]bool{"chunk_method": true})
ParserID: &chunkMethod,
}, map[string]bool{"parser_id": true})
if err != nil {
t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
}
if resp.ChunkMethod != chunkMethod || resp.Run != "UNSTART" || resp.TokenCount != 0 || resp.ChunkCount != 0 {
if resp.ParserID != chunkMethod || resp.Run != "UNSTART" || resp.TokenCount != 0 || resp.ChunkCount != 0 {
t.Fatalf("response = %+v, want method=%s run=UNSTART counts=0", resp, chunkMethod)
}
@@ -1754,37 +1731,6 @@ func TestUpdateDatasetDocumentChunkMethodResetsForReparse(t *testing.T) {
}
}
func TestUpdateDatasetDocumentParserIDResetsForReparse(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)
parserID := "manual"
svc := testDocumentService(t)
resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
ParserID: &parserID,
}, map[string]bool{"parser_id": true})
if err != nil {
t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
}
if resp.ChunkMethod != parserID || resp.Run != "UNSTART" || resp.TokenCount != 0 || resp.ChunkCount != 0 {
t.Fatalf("response = %+v, want parser_id=%s run=UNSTART counts=0", resp, parserID)
}
doc, _ := dao.NewDocumentDAO().GetByID("doc-1")
if doc.ParserID != parserID {
t.Fatalf("parser_id = %q, want %q", doc.ParserID, parserID)
}
if doc.TokenNum != 0 || doc.ChunkNum != 0 {
t.Fatalf("doc counters = token:%d chunk:%d, want zero", doc.TokenNum, doc.ChunkNum)
}
kb, _ := dao.NewKnowledgebaseDAO().GetByID("kb-1")
if kb.TokenNum != 0 || kb.ChunkNum != 0 {
t.Fatalf("kb counters = token:%d chunk:%d, want zero", kb.TokenNum, kb.ChunkNum)
}
}
func TestResetDocumentForReparseSkipsSecondCounterDecrement(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
@@ -2326,7 +2272,7 @@ func TestMergeFieldValuesKeepsNumericValues(t *testing.T) {
}
}
func TestUpdateDatasetDocumentPipelineIDTakesPrecedenceOverChunkMethod(t *testing.T) {
func TestUpdateDatasetDocumentPipelineIDTakesPrecedenceOverParserID(t *testing.T) {
db := setupServiceTestDB(t)
pushServiceDB(t, db)
insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
@@ -2336,17 +2282,17 @@ func TestUpdateDatasetDocumentPipelineIDTakesPrecedenceOverChunkMethod(t *testin
chunkMethod := "manual"
svc := testDocumentService(t)
resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
PipelineID: &pipelineID,
ChunkMethod: &chunkMethod,
}, map[string]bool{"pipeline_id": true, "chunk_method": true})
PipelineID: &pipelineID,
ParserID: &chunkMethod,
}, map[string]bool{"pipeline_id": true, "parser_id": true})
if err != nil {
t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
}
if resp.PipelineID == nil || *resp.PipelineID != pipelineID {
t.Fatalf("pipeline_id = %v, want %q", resp.PipelineID, pipelineID)
}
if resp.ChunkMethod != "naive" {
t.Fatalf("chunk_method = %q, want original naive", resp.ChunkMethod)
if resp.ParserID != "naive" {
t.Fatalf("parser_id = %q, want original naive", resp.ParserID)
}
}

28
internal/service/document_upload_helpers.go
View File

@@ -2,37 +2,21 @@ package service
import (
"encoding/hex"
"path/filepath"
"regexp"
"strings"
pipelinepkg "ragflow/internal/ingestion/pipeline"
"ragflow/internal/utility"
"github.com/zeebo/xxh3"
)
var (
presentationUploadPattern = regexp.MustCompile(`(?i)\.(ppt|pptx|pages)$`)
emailUploadPattern = regexp.MustCompile(`(?i)\.(msg|eml)$`)
)
// selectUploadParser mirrors Python FileService.get_parser.
// selectUploadParser resolves the parser_id for a document by delegating to
// the builtin pipeline registry, which owns the file-type-to-parser mapping.
func selectUploadParser(docType utility.FileType, filename, defaultParser string) string {
switch docType {
case utility.FileTypeVISUAL:
return "picture"
case utility.FileTypeAURAL:
return "audio"
}
base := filepath.Base(strings.TrimSpace(filename))
switch {
case presentationUploadPattern.MatchString(base):
return "presentation"
case emailUploadPattern.MatchString(base):
return "email"
default:
registry, err := pipelinepkg.DefaultRegistry()
if err != nil || registry == nil {
return defaultParser
}
return registry.DefaultParserID(string(docType), filename, defaultParser)
}
// contentHashHex mirrors Python xxhash.xxh128(blob).hexdigest().

14
internal/service/file2document.go
View File

@@ -215,6 +215,20 @@ func (s *File2DocumentService) convertFiles(fileIDs, kbIDs []string, userID stri
}
if kb.PipelineID != nil {
doc.PipelineID = kb.PipelineID
doc.ParserID = "" // canvas pipeline mode — parser_id not applicable
}
// When the document's builtin parser_id differs from the KB's
// (e.g. visual→picture, aural→audio), re-resolve component_params
// defaults from the document's own DSL template so cpnIDs and
// param keys match the pipeline that will actually execute.
if kb.PipelineID == nil && parserID != kb.ParserID {
if cp, err := resolveComponentParamsDefaults(parserID, nil); err != nil {
common.Warn("convertFiles: resolve component_params defaults",
zap.String("parserID", parserID), zap.Error(err))
} else if cp != nil {
doc.ParserConfig = cp
}
}
// InsertDocument creates the row and increments KB doc_num in one

2
internal/tokenizer/tokenizer_concurrent_test.go
View File

@@ -30,7 +30,7 @@ import (
func init() {
// Initialize logger for tests
if err := common.Init("info", common.FileOutput{}); err != nil {
if err := common.Init("info", common.FileOutput{}, ""); err != nil {
fmt.Printf("Failed to initialize logger: %v\n", err)
}
}