diff --git a/internal/ingestion/component/chunker/crop_title_test.go b/internal/ingestion/component/chunker/crop_title_test.go new file mode 100644 index 0000000000..a64acd4a86 --- /dev/null +++ b/internal/ingestion/component/chunker/crop_title_test.go @@ -0,0 +1,65 @@ +//go:build cgo + +// 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. +// +// 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 chunker + +import ( + "context" + "strings" + "testing" +) + +// TestCropTitleChunks_CropsPositionedChunk pins residual A: chunks +// produced by the Title/Group/Hierarchy chunkers must be cropped on demand +// (mirroring the TokenChunker JSON path). The chunk shape mirrors the real +// Group/Hierarchy output: it carries doc_type_kwd + positions but NO ck_type +// (buildChunksFromRecordGroups never sets ck_type). cropTitleChunks must +// derive ck_type from doc_type_kwd so needsCrop (pdfcrop_cgo.go:151) fires. +// A text chunk with positions gets a rendered preview; a chunk without +// positions is left untouched; and the derived ck_type must not leak into +// the returned chunk shape. +func TestCropTitleChunks_CropsPositionedChunk(t *testing.T) { + ctx := context.Background() + pos := jsonPositions(t, []float64{1, 10, 100, 10, 100}) + in := []map[string]any{ + {"text": "page text", "doc_type_kwd": "text", "positions": pos}, + {"text": "no coords", "doc_type_kwd": "text"}, + } + + got := cropTitleChunks(ctx, mockCropEngine{}, in) + if len(got) != 2 { + t.Fatalf("len = %d, want 2", len(got)) + } + img, _ := got[0]["image"].(string) + if !strings.HasPrefix(img, "data:image/png;base64,") { + t.Errorf("positioned text chunk: image = %q, want data:image/png;base64, prefix", img) + } + if got[1]["image"] != nil { + t.Errorf("chunk without positions should not be cropped, got %v", got[1]["image"]) + } + if _, ok := got[0]["ck_type"]; ok { + t.Errorf("derived ck_type must not leak into output chunk: %v", got[0]) + } +} + +// TestCropTitleChunks_NilEnginePassthrough asserts the helper is a no-op +// when no PDF engine is available (best-effort contract). +func TestCropTitleChunks_NilEnginePassthrough(t *testing.T) { + in := []map[string]any{{"text": "hello"}} + got := cropTitleChunks(context.Background(), nil, in) + if len(got) != 1 || got[0]["text"] != "hello" { + t.Fatalf("nil engine should pass through unchanged: %v", got) + } +} diff --git a/internal/ingestion/component/chunker/group.go b/internal/ingestion/component/chunker/group.go index 992d4d9825..fe263c6119 100644 --- a/internal/ingestion/component/chunker/group.go +++ b/internal/ingestion/component/chunker/group.go @@ -36,6 +36,7 @@ import ( "context" "encoding/json" "fmt" + "log/slog" "regexp" "sort" "strings" @@ -108,7 +109,7 @@ func buildSectionIDs(levels []int, targetLevel int) []int { // supplied inputs. Detected headings + adjacent merges happen in two // goroutines (heading detection sequential, then a fan-out over // record-buckets for the merge pass). -func invokeGroup(_ context.Context, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) { +func invokeGroup(parentCtx context.Context, db *gorm.DB, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) { records := extractLineRecords(inputs) common.Debug("chunker stage", zap.String("component", "Chunker"), @@ -161,6 +162,21 @@ func invokeGroup(_ context.Context, inputs map[string]any, p *titleChunkerParam) zap.Int("chunks", len(chunks)), zap.Bool("plain_text", isPlainTextFormat(inputs)), ) + + // On-demand PDF preview cropping for image/table/text chunks, + // mirroring the TokenChunker JSON path (token.go:513). Best-effort: + // a missing or unreadable PDF simply skips cropping. + if upstream, uErr := decodeChunkerFromUpstream(inputs); uErr == nil { + engine, eErr := newPDFEngineFromUpstream(parentCtx, db, upstream) + if eErr != nil { + slog.Warn("GroupTitleChunker: could not open PDF for on-demand cropping", "err", eErr) + } + if engine != nil { + defer engine.Close() + chunks = cropTitleChunks(parentCtx, engine, chunks) + } + } + if len(chunks) == 0 { return emptyOutputs(), nil } @@ -484,7 +500,7 @@ func (c *GroupTitleChunkerComponent) Invoke(ctx context.Context, db *gorm.DB, in "_ERROR": "GroupTitleChunker: missing required upstream field \"name\"", }, nil } - return invokeGroup(ctx, withName(inputs, name), &c.param) + return invokeGroup(ctx, db, withName(inputs, name), &c.param) } // init registers GroupTitleChunker under CategoryIngestion. diff --git a/internal/ingestion/component/chunker/hierarchy.go b/internal/ingestion/component/chunker/hierarchy.go index 107e3b1233..c8dfd0d35c 100644 --- a/internal/ingestion/component/chunker/hierarchy.go +++ b/internal/ingestion/component/chunker/hierarchy.go @@ -41,6 +41,7 @@ package chunker import ( "context" "fmt" + "log/slog" "go.uber.org/zap" "gorm.io/gorm" @@ -139,7 +140,7 @@ func (n *chunkNode) getPaths(paths *[][]int, titles []int, depth int, includeHea } // invokeHierarchy runs the HierarchyTitleChunker strategy. -func invokeHierarchy(_ context.Context, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) { +func invokeHierarchy(parentCtx context.Context, db *gorm.DB, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) { records := extractLineRecords(inputs) common.Debug("chunker stage", zap.String("component", "Chunker"), @@ -246,6 +247,21 @@ func invokeHierarchy(_ context.Context, inputs map[string]any, p *titleChunkerPa zap.Int("chunks", len(chunks)), zap.Bool("plain_text", isPlainTextFormat(inputs)), ) + + // On-demand PDF preview cropping for image/table/text chunks, + // mirroring the TokenChunker JSON path (token.go:513). Best-effort: + // a missing or unreadable PDF simply skips cropping. + if upstream, uErr := decodeChunkerFromUpstream(inputs); uErr == nil { + engine, eErr := newPDFEngineFromUpstream(parentCtx, db, upstream) + if eErr != nil { + slog.Warn("HierarchyTitleChunker: could not open PDF for on-demand cropping", "err", eErr) + } + if engine != nil { + defer engine.Close() + chunks = cropTitleChunks(parentCtx, engine, chunks) + } + } + if len(chunks) == 0 { return emptyOutputs(), nil } @@ -302,7 +318,7 @@ func (c *HierarchyTitleChunkerComponent) Invoke(ctx context.Context, db *gorm.DB "_ERROR": "HierarchyTitleChunker: missing required upstream field \"name\"", }, nil } - return invokeHierarchy(ctx, withName(inputs, name), &c.param) + return invokeHierarchy(ctx, db, withName(inputs, name), &c.param) } // init registers HierarchyTitleChunker under CategoryIngestion. diff --git a/internal/ingestion/component/chunker/one.go b/internal/ingestion/component/chunker/one.go index daff2a4c1d..cbdb094017 100644 --- a/internal/ingestion/component/chunker/one.go +++ b/internal/ingestion/component/chunker/one.go @@ -141,10 +141,12 @@ func emitOneFromItems(items, chunks []schema.ChunkDoc) map[string]any { return emptyOutputs() } out := schema.ChunkDoc{ - Text: text, - DocType: docType, - CKType: docType, - Image: it.Image, + Text: text, + DocType: docType, + CKType: docType, + Image: it.Image, + Positions: it.Positions, + PDFPositions: it.PDFPositions, } return chunkOutputs([]schema.ChunkDoc{out}) } @@ -164,6 +166,11 @@ func emitOneFromItems(items, chunks []schema.ChunkDoc) map[string]any { return emptyOutputs() } out := schema.ChunkDoc{Text: merged, DocType: "text", CKType: "text"} + // Multi-item merge produces a single text-only chunk mirroring Python + // one.py:166-168. Per-item Positions/PDFPositions are intentionally + // not carried — merging coordinates from different source items would + // produce meaningless composite geometry, and the downstream + // processChunkPositions would map them to incorrect pages. if img != "" { out.Image = img } diff --git a/internal/ingestion/component/chunker/one_test.go b/internal/ingestion/component/chunker/one_test.go index 031774dfaf..5fda780c96 100644 --- a/internal/ingestion/component/chunker/one_test.go +++ b/internal/ingestion/component/chunker/one_test.go @@ -98,3 +98,46 @@ func TestOneChunker_JSONMultipleMerges(t *testing.T) { t.Errorf("image = %q, want first available media context", got) } } + +// TestOneChunker_PreservesPositions verifies that when a single upstream +// item carries PDF coordinates, the OneChunker preserves both Positions +// and PDFPositions (with their coordinate values) on the output chunk. +func TestOneChunker_PreservesPositions(t *testing.T) { + chunks := oneChunksOf(t, map[string]any{ + "name": "page.pdf", + "output_format": "json", + "json": []map[string]any{ + { + "text": "page text", + "positions": []any{[]any{10.0, 20.0, 30.0, 40.0}}, + "_pdf_positions": []any{[]any{1.0, 2.0, 3.0, 4.0, 5.0}}, + }, + }, + }) + if len(chunks) != 1 { + t.Fatalf("want 1 chunk, got %d", len(chunks)) + } + assertCoordTuple(t, "positions", chunks[0]["positions"], []float64{10.0, 20.0, 30.0, 40.0}) + assertCoordTuple(t, "_pdf_positions", chunks[0]["_pdf_positions"], []float64{1.0, 2.0, 3.0, 4.0, 5.0}) +} + +// assertCoordTuple verifies a positions/_pdf_positions field round-tripped +// as a [][]float64 with the expected single-row coordinate tuple. +func assertCoordTuple(t *testing.T, key string, got any, want []float64) { + t.Helper() + rows, ok := got.([][]float64) + if !ok { + t.Fatalf("%s = %v, want [][]float64 (got %T)", key, got, got) + } + if len(rows) != 1 { + t.Fatalf("%s has %d rows, want 1", key, len(rows)) + } + if len(rows[0]) != len(want) { + t.Fatalf("%s[0] = %v, want %v (len %d vs %d)", key, rows[0], want, len(rows[0]), len(want)) + } + for i, w := range want { + if rows[0][i] != w { + t.Errorf("%s[0][%d] = %v, want %v", key, i, rows[0][i], w) + } + } +} diff --git a/internal/ingestion/component/chunker/qa.go b/internal/ingestion/component/chunker/qa.go index 33c4359d99..9ec591ab38 100644 --- a/internal/ingestion/component/chunker/qa.go +++ b/internal/ingestion/component/chunker/qa.go @@ -101,7 +101,7 @@ func (c *QAChunkerComponent) invoke(_ context.Context, inputs map[string]any) (m qPrefix, aPrefix := "问题：", "回答：" // Python qa.py defaults to Chinese when no language is supplied; only - // an explicit "english" switches to English prefixes (diff Chunker-2.13). + // an explicit "english" switches to English prefixes eng := strings.EqualFold(c.param.Lang, "english") if eng { qPrefix, aPrefix = "Question: ", "Answer: " @@ -137,7 +137,7 @@ func (c *QAChunkerComponent) invoke(_ context.Context, inputs map[string]any) (m ContentLtks: contentLTKS, ContentSmLtks: contentSMLTKS, } - // Restore metadata lost before diff Chunker-1.8: top_int (row + // // index), image id + coordinates carried from the source item. if pair.RowNum >= 0 { chunk.TopInt = []int{pair.RowNum} @@ -172,14 +172,14 @@ type qaPair struct { RowNum int // Image and positions are carried from the upstream item so the QA // chunk preserves metadata that Python sets via beAdocPdf/beAdocDocx - // (diff Chunker-1.8). + // Image string PDFPositions json.RawMessage Positions json.RawMessage } // rmQAPrefixRe mirrors Python qa.py:241 `[\t:： ]+` — one-or-more separator -// chars, so "Q:: answer" is fully stripped (diff Chunker-2.12). +// chars, so "Q:: answer" is fully stripped var rmQAPrefixRe = regexp.MustCompile(`(?i)^(问题|答案|回答|user|assistant|Q|A|Question|Answer|问|答)[\t:： ]+`) func rmQAPrefix(txt string) string { @@ -424,7 +424,7 @@ func extractQAJSON(items []schema.ChunkDoc) []qaPair { } tmp := extractQAText(txt) // Preserve the source item's image id and coordinates on each - // extracted pair (diff Chunker-1.8). + // extracted pair for _, p := range tmp { p.Image = item.Image p.PDFPositions = item.PDFPositions diff --git a/internal/ingestion/component/chunker/qa_batch2_test.go b/internal/ingestion/component/chunker/qa_batch2_test.go index 84b334ac94..1bfb72bbbd 100644 --- a/internal/ingestion/component/chunker/qa_batch2_test.go +++ b/internal/ingestion/component/chunker/qa_batch2_test.go @@ -41,7 +41,7 @@ func qaInvoke(t *testing.T, inputs map[string]any) []map[string]any { return chunks } -// TestQAChunker_DefaultLangIsChinese exercises migration diff Chunker-2.13: +// TestQAChunker_DefaultLangIsChinese exercises migration : // when no language is supplied, Python defaults to Chinese prefixes // ("问题："/"回答："); the legacy Go code defaulted to English. func TestQAChunker_DefaultLangIsChinese(t *testing.T) { @@ -79,7 +79,7 @@ func TestRmQAPrefixStripsMultipleSeparators(t *testing.T) { } } -// TestQAChunker_SetsTopInt exercises migration diff Chunker-1.8 (top_int): +// TestQAChunker_SetsTopInt exercises migration (top_int): // each QA chunk must carry the source row index in `top_int`, matching // Python beAdoc(..., row_num=i). func TestQAChunker_SetsTopInt(t *testing.T) { diff --git a/internal/ingestion/component/chunker/qa_test.go b/internal/ingestion/component/chunker/qa_test.go index ad85e13486..e4ecefdf5c 100644 --- a/internal/ingestion/component/chunker/qa_test.go +++ b/internal/ingestion/component/chunker/qa_test.go @@ -213,7 +213,7 @@ func TestQAChunker_PrefixSpaceSeparatorStrips(t *testing.T) { } cww, _ := chunks[0]["content_with_weight"].(string) // Python qa.py:241 uses `[\t:： ]+`, so a space is a valid separator: - // a leading "A"/"Q" followed by a space is stripped (diff Chunker-2.12). + // a leading "A"/"Q" followed by a space is stripped. . if cww != "Question: language model is useful\tAnswer: How does it work" { t.Fatalf("space-separator prefix not stripped: %q", cww) } diff --git a/internal/ingestion/component/chunker/tag.go b/internal/ingestion/component/chunker/tag.go index e4a1e8496f..d1f341b987 100644 --- a/internal/ingestion/component/chunker/tag.go +++ b/internal/ingestion/component/chunker/tag.go @@ -125,6 +125,7 @@ func (c *TagChunkerComponent) invoke(_ context.Context, inputs map[string]any) ( ContentLtks: contentLTKS, ContentSmLtks: contentSMLTKS, TagKwd: splitTagKwd(pair.Tags), + TopInt: []int{pair.RowNum}, } chunks = append(chunks, chunk) } @@ -133,9 +134,13 @@ func (c *TagChunkerComponent) invoke(_ context.Context, inputs map[string]any) ( } // tagPair is a (content, tags) row extracted from the upstream payload. +// RowNum is the 0-based record START line (the first physical source +// line of the record), used by every extractor consistently and mapped +// to Python's top_int in beAdoc(tag.py:33). type tagPair struct { Content string Tags string + RowNum int } // extractTagText ports tag.py:60-89 (txt) and tag.py:91-113 (csv). @@ -157,18 +162,23 @@ func extractTagText(text string) []tagPair { func extractTagTextTab(lines []string) []tagPair { var pairs []tagPair content := "" - for _, line := range lines { + contentStart := -1 + for i, line := range lines { if strings.TrimSpace(line) == "" { continue } + if contentStart < 0 { + contentStart = i + } parts := strings.Split(line, "\t") if len(parts) != 2 { content += "\n" + line continue } content += "\n" + parts[0] - pairs = append(pairs, tagPair{Content: content, Tags: parts[1]}) + pairs = append(pairs, tagPair{Content: content, Tags: parts[1], RowNum: contentStart}) content = "" + contentStart = -1 } return pairs } @@ -202,6 +212,7 @@ func extractTagTextCSV(text string, lines []string) []tagPair { for curLine < len(lineStarts) && lineStarts[curLine] < endOff { curLine++ } + startLine := prevLine raw := strings.Join(lines[prevLine:curLine], "\n") prevLine = curLine @@ -210,7 +221,10 @@ func extractTagTextCSV(text string, lines []string) []tagPair { continue } content += "\n" + record[0] - pairs = append(pairs, tagPair{Content: content, Tags: record[1]}) + // RowNum is the 0-based record START line, kept consistent with + // extractTagTextTab (line i) and extractTagTable ( i) so every + // tag-pair source uses the same row-index convention. + pairs = append(pairs, tagPair{Content: content, Tags: record[1], RowNum: startLine}) content = "" } return pairs @@ -226,7 +240,7 @@ func extractTagTable(htmlStr string) []tagPair { } rows := htmlTR.FindAllStringSubmatch(htmlStr, -1) pairs := make([]tagPair, 0, len(rows)) - for _, row := range rows { + for i, row := range rows { cells := htmlTD.FindAllStringSubmatch(row[1], -1) var texts []string for _, cell := range cells { @@ -237,7 +251,7 @@ func extractTagTable(htmlStr string) []tagPair { } } if len(texts) >= 2 { - pairs = append(pairs, tagPair{Content: texts[0], Tags: texts[1]}) + pairs = append(pairs, tagPair{Content: texts[0], Tags: texts[1], RowNum: i}) } } return pairs diff --git a/internal/ingestion/component/chunker/tag_test.go b/internal/ingestion/component/chunker/tag_test.go index 7fb503f4ac..515e0b6979 100644 --- a/internal/ingestion/component/chunker/tag_test.go +++ b/internal/ingestion/component/chunker/tag_test.go @@ -133,3 +133,70 @@ func TestTagChunker_HTMLTable(t *testing.T) { t.Errorf("chunk0 tag_kwd = %v, want [a1 b1]", tags0) } } + +// TestTagChunk_TopInt verifies each tag-pair row carries the 0-based +// source row index as top_int, matching Python tag.py:33 beAdoc(row_num=i). +func TestTagChunk_TopInt(t *testing.T) { + chunks := tagChunksOf(t, map[string]any{ + "name": "tags.txt", + "output_format": "text", + "text": "tag1\tcontent1\ntag2\tcontent2", + }) + if len(chunks) != 2 { + t.Fatalf("want 2 chunks, got %d", len(chunks)) + } + if ti, ok := chunks[0]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(0) { + t.Errorf("chunk 0: top_int = %v, want [0]", chunks[0]["top_int"]) + } + if ti, ok := chunks[1]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(1) { + t.Errorf("chunk 1: top_int = %v, want [1]", chunks[1]["top_int"]) + } +} + +// TestTagChunk_CSVTopIntStartLine locks the RowNum convention from +// review finding #4: a multi-line quoted CSV record must report the +// record's START line, not the line after it ends. The first record +// spans physical lines 0-1, so its top_int is [0]; the second record +// is on line 2, so its top_int is [2]. +func TestTagChunk_CSVTopIntStartLine(t *testing.T) { + chunks := tagChunksOf(t, map[string]any{ + "name": "tags.csv", + "output_format": "text", + "text": "\"multi\nline content\",TAG\nsecond,OTHER", + }) + if len(chunks) != 2 { + t.Fatalf("want 2 chunks, got %d", len(chunks)) + } + if ti, ok := chunks[0]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(0) { + t.Errorf("chunk 0: top_int = %v, want [0]", chunks[0]["top_int"]) + } + if ti, ok := chunks[1]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(2) { + t.Errorf("chunk 1: top_int = %v, want [2]", chunks[1]["top_int"]) + } +} + +// TestTagChunk_TSVMultiLineRowNumStart pins the RowNum fix for TSV +// multi-line records: when a record spans multiple physical lines +// (because the first line does not contain a tab delimiter), the +// resulting chunk's top_int must report the first non-empty CONTENT +// start line, not the line where the tab-delimiter eventually +// terminates the record (PR #17419 review #1). +func TestTagChunk_TSVMultiLineRowNumStart(t *testing.T) { + chunks := tagChunksOf(t, map[string]any{ + "name": "tags.txt", + "output_format": "text", + // Line 0 is empty, line 1 has no tab (accumulated as prefix + // of the first record), line 2 is valid TSV. + // The prefix starts at line 1 → RowNum must be 1, not 2. + "text": "\nprefix\nrecord1\tTAG1\nmulti\nline-prefix\nrecord2\tTAG2", + }) + if len(chunks) != 2 { + t.Fatalf("want 2 chunks, got %d", len(chunks)) + } + if ti, ok := chunks[0]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(1) { + t.Errorf("chunk 0: top_int = %v, want [1] (content start)", chunks[0]["top_int"]) + } + if ti, ok := chunks[1]["top_int"].([]any); !ok || len(ti) != 1 || ti[0] != float64(3) { + t.Errorf("chunk 1: top_int = %v, want [3] (content start after blank line)", chunks[1]["top_int"]) + } +} diff --git a/internal/ingestion/component/chunker/title.go b/internal/ingestion/component/chunker/title.go index d0ae0d9993..40662c341f 100644 --- a/internal/ingestion/component/chunker/title.go +++ b/internal/ingestion/component/chunker/title.go @@ -764,9 +764,9 @@ func (c *TitleChunkerComponent) Invoke(ctx context.Context, db *gorm.DB, inputs } switch c.param.Method { case "hierarchy": - return invokeHierarchy(ctx, inputs, &c.param) + return invokeHierarchy(ctx, db, inputs, &c.param) case "group": - return invokeGroup(ctx, inputs, &c.param) + return invokeGroup(ctx, db, inputs, &c.param) default: return map[string]any{ "output_format": "chunks", diff --git a/internal/ingestion/component/chunker/token.go b/internal/ingestion/component/chunker/token.go index aa12e31f03..ca445867dd 100644 --- a/internal/ingestion/component/chunker/token.go +++ b/internal/ingestion/component/chunker/token.go @@ -252,6 +252,45 @@ func stripChunkerRuntimeTimestamps(inputs map[string]any) map[string]any { return out } +// cropTitleChunks crops image/table/text previews for chunks produced by +// the Title/Group/Hierarchy chunkers, mirroring the TokenChunker JSON path +// (cropImageChunks at token.go:513). A nil engine — or an +// empty chunk list — leaves chunks unchanged (best-effort, matching the +// on-demand PDF crop contract used by the TokenChunker path). +func cropTitleChunks(ctx context.Context, engine deepdoctype.PDFEngine, chunks []map[string]any) []map[string]any { + if engine == nil || len(chunks) == 0 { + return chunks + } + docs, _, err := schema.ChunkDocsFromAny(chunks) + if err != nil || len(docs) == 0 { + return chunks + } + // The Title/Group/Hierarchy chunkers emit doc_type_kwd but not the + // ck_type field that cropImageChunks' needsCrop consults + // (pdfcrop_cgo.go:151). Derive ck_type from doc_type_kwd so the crop + // decision matches the TokenChunker path. The derived ck_type is + // stripped from the returned maps so the downstream chunk shape is + // unchanged (setting ck_type in the real output would also change + // how a downstream TokenChunker merges these chunks — a separate + // concern, out of scope here). + for i := range docs { + if docs[i].CKType == "" { + switch docs[i].DocType { + case "image", "table": + docs[i].CKType = docs[i].DocType + default: + docs[i].CKType = "text" + } + } + } + cropped := cropImageChunks(ctx, engine, docs) + out := schema.ChunkDocsToMaps(cropped) + for _, m := range out { + delete(m, "ck_type") + } + return out +} + // invokeTextPayload handles plain-text input (output_format in // {markdown,text,html} on the python side). func (c *TokenChunkerComponent) invokeTextPayload(_ context.Context, text string, delimPattern, childrenPattern *regexp.Regexp) map[string]any { @@ -290,18 +329,21 @@ func (c *TokenChunkerComponent) invokeTextPayload(_ context.Context, text string } // sentenceDelimiter is the sentence/clause-boundary regex used to split -// oversized sections. It mirrors Python's default delimiter "\n。；！？" -// plus the English ". " fallback, and also breaks on ASCII "!" and "?" -// (diff Chunker-2.1). -var sentenceDelimiter = regexp.MustCompile(`(\n|[!?。；！？]|\.\s)`) +// oversized sections. It mirrors the delimiter Python's chunker actually +// uses in production: rag/app/naive.py:1285 passes "\n!?。；！？" to +// naive_merge, which includes ASCII "!" and "?" as well as the CJK +// punctuation "。；！？". It deliberately does NOT include an English +// ". " fallback: Python's production delimiter has no "\.\s", so adding +// it would diverge from Python's chunk boundaries. +var sentenceDelimiter = regexp.MustCompile(`(\n|[!?。；！？])`) // mergeByTokenSize implements exact token-based chunk merging that mirrors // Python's naive_merge (rag/nlp/__init__.py:1156). It uses // tokenizeStr (= tokenizer.NumTokensFromString, cl100k_base BPE) for -// precise token counting, splits input into paragraph sections, further -// subdivides oversized sections on sentence delimiters, and greedily -// merges into chunks of approximately chunk_token_size tokens with -// optional overlap from the previous chunk. +// precise token counting, treats the payload as a single section, splits +// oversized sections on sentence delimiters (dropping the delimiter, as +// Python does), and greedily merges into chunks of approximately +// chunk_token_size tokens with optional overlap from the previous chunk. func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *regexp.Regexp) map[string]any { target := c.param.ChunkTokenSize overlapPct := c.param.OverlappedPercent @@ -316,16 +358,29 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r overlapPct = 100 } - // Split into paragraph-aligned sections. - sections := splitIntoSections(text) + // Normalize line endings to LF before any splitting. Python's + // naive_merge (rag/nlp/__init__.py:1166) runs + // text = text.replace("\r\n", "\n").replace("\r", "\n") + // so CRLF/CR input must segment and split exactly like LF input. + // Without this, stray "\r" would survive inside chunks, diverging + // from Python. + text = strings.ReplaceAll(strings.ReplaceAll(text, "\r\n", "\n"), "\r", "\n") + + // Treat the whole payload as a single section, mirroring Python's + // naive_merge (rag/nlp/__init__.py:1157) which wraps the input string + // as a one-element list. naive_merge does NOT pre-split on blank + // lines, and because "\n" is itself a delimiter it is dropped (blank + // lines collapse), exactly as Python does. CRLF/CR normalization + // already happened above. + sections := []string{text} if len(sections) == 0 { return emptyOutputs() } // Sentence/clause-boundary regex for splitting oversized sections. - // Matches Python's default delimiter "\n。；！？" plus English ". " - // fallback. The ASCII "!" and "?" are included to match Python's - // full delimiter set (diff Chunker-2.1). + // Mirrors Python's production delimiter (rag/app/naive.py:1285 passes + // "\n!?。；！？") — ASCII "!" and "?" plus the CJK punctuation, with no + // English ". " fallback. sentenceDelim := sentenceDelimiter var cks []string // chunk texts @@ -342,7 +397,7 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r segTokens := tokens if overlapPct > 0 && len(cks) > 0 { // Strip parser tags before computing the overlap suffix, - // matching Python nlp/__init__.py:1181 (diff Chunker-2.2). + // matching Python nlp/__init__.py:1181 prev := removeTag(cks[len(cks)-1]) // Take the last overlapped_percent of the previous chunk // (in runes, matching Python's len(overlapped) * ratio). @@ -387,32 +442,36 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r continue } - // Oversized section: split on sentence delimiters, then merge. + // Oversized section: split on sentence delimiters. Python's + // naive_merge (rag/nlp/__init__.py:1216-1225) splits with a + // capturing-group re.split but then SKIPS any segment that is a + // pure delimiter (re.fullmatch(dels, sub_sec)), so the delimiter + // character (\n / 。 / ！ / ？) is DROPPED from the chunk text + // rather than retained. We mirror that by using regexp.Split + // (which discards the delimiter) and prepending a single "\n" to + // each segment, matching Python's add_chunk("\n"+sub_sec). parts := sentenceDelim.Split(sec, -1) for _, part := range parts { part = strings.TrimSpace(part) if part == "" { continue } + // Route every segment — including tiny <8-token fragments — + // through mergeOrNew so it honours the token threshold, + // mirroring Python's add_chunk. The old shortcut appended + // unconditionally, merging fragments into an already-overfull + // chunk (review #2). p := "\n" + part - pn := tokenizeStr(p) - if pn < 8 { - if len(cks) > 0 { - cks[len(cks)-1] += p - tkns[len(tkns)-1] += pn - } else { - cks = append(cks, p) - tkns = append(tkns, pn) - } - continue - } - mergeOrNew(p, pn) + mergeOrNew(p, tokenizeStr(p)) } } docs := make([]schema.ChunkDoc, 0, len(cks)) for _, ch := range cks { - ch = strings.TrimSpace(ch) + // Strip parser position tags from the final text: + // the merge paths may carry @@...## markers that must not leak into + // indexed/embedded chunk text. + ch = removeTag(strings.TrimSpace(ch)) if ch == "" { continue } @@ -422,25 +481,6 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r return chunkOutputs(final) } -// splitIntoSections partitions text into paragraph-level sections by -// splitting on blank-line boundaries. It uses \n\s*\n (not \n{2,}) so a -// CRLF paragraph break (\r\n\r\n) still matches — the Go chunker performs -// no \r\n normalization of its own, so a stricter \n{2,} would silently -// lose CRLF boundaries. NOTE: the precise 2.7/2.10 section-splitting -// semantics are still OPEN in docs/migration_python_go_diff.md (alignment -// target flow-vs-app Python undecided); this keeps the pre-existing, -// CRLF-safe behaviour until that is resolved in a dedicated change. -func splitIntoSections(text string) []string { - if text == "" { - return nil - } - // Split on a blank line: a newline, optional whitespace (absorbs a - // trailing \r on CRLF input), then another newline. - re := regexp.MustCompile(`\n\s*\n`) - parts := re.Split(text, -1) - return parts -} - // invokeJSONPayload handles structured upstream input. Items fan // across 4 goroutines; merge is by input index. func (c *TokenChunkerComponent) invokeJSONPayload(ctx context.Context, items []schema.ChunkDoc, delimPattern, childrenPattern *regexp.Regexp, engine deepdoctype.PDFEngine) map[string]any { @@ -502,7 +542,12 @@ func (c *TokenChunkerComponent) invokeJSONPayload(ctx context.Context, items []s out := make([]schema.ChunkDoc, 0, len(flat)) for _, m := range flat { - if strings.TrimSpace(m.Text) == "" { + // Strip parser position tags from the final text: + // the merge paths may carry @@...## markers that must not leak into + // indexed/embedded chunk text. Crop above reads positions, not text, + // so the ordering is safe. + m.Text = removeTag(strings.TrimSpace(m.Text)) + if m.Text == "" { continue } out = append(out, m) @@ -687,7 +732,7 @@ func collectContext(chunks []schema.ChunkDoc, i, ctxTokens int, above bool) stri } // takeFromEnd returns the smallest tail of text whose token count is >= -// tokens, counted exactly via tokenizeStr (diff Chunker-2.4). The previous +// tokens, counted exactly via tokenizeStr The previous // 4-bytes-per-token heuristic over-counted for CJK text. func takeFromEnd(text string, tokens int) string { runes := []rune(text) @@ -703,7 +748,7 @@ func takeFromEnd(text string, tokens int) string { } // takeFromStart returns the smallest prefix of text whose token count is >= -// tokens, counted exactly via tokenizeStr (diff Chunker-2.4). +// tokens, counted exactly via tokenizeStr func takeFromStart(text string, tokens int) string { runes := []rune(text) best := text @@ -758,7 +803,7 @@ func mergeByTokenSizeFromJSON(perItem [][]schema.ChunkDoc, chunkTokens int, over if len(merged) > 0 && merged[len(merged)-1].CKType == "text" && overlappedPct > 0 { // Strip parser tags before computing the overlap // suffix, matching Python nlp/__init__.py:1181 - // (diff Chunker-2.2). + // if prevText := removeTag(merged[len(merged)-1].Text); prevText != "" { runes := []rune(prevText) cut := int(float64(len(runes)) * (100 - overlappedPct) / 100.0) @@ -775,7 +820,7 @@ func mergeByTokenSizeFromJSON(perItem [][]schema.ChunkDoc, chunkTokens int, over prev := &merged[len(merged)-1] // Mirror Python token_chunker.py:236-239: when the accumulated // chunk has empty text, assign the incoming text directly instead - // of skipping it (diff Chunker-2.11). + // of skipping it if prev.Text == "" { prev.Text = ck.Text } else { diff --git a/internal/ingestion/component/chunker/token_batch1_test.go b/internal/ingestion/component/chunker/token_batch1_test.go index 0c79927f9c..7e869e1ce4 100644 --- a/internal/ingestion/component/chunker/token_batch1_test.go +++ b/internal/ingestion/component/chunker/token_batch1_test.go @@ -69,7 +69,7 @@ func TestMergeByTokenSizeFromJSON_OverlapStripsTags(t *testing.T) { } // The overlap prefix is prepended to the SECOND chunk. The original // first chunk legitimately keeps its own parser tag; only the overlap - // region (merged[1]) must be tag-free (diff Chunker-2.2). + // region (merged[1]) must be tag-free. . if strings.Contains(merged[1].Text, "@@") || strings.Contains(merged[1].Text, "##") { t.Errorf("overlap prefix leaked parser tag into chunk 1: %q", merged[1].Text) } @@ -155,7 +155,7 @@ func TestMergeByTokenSizeFromJSON_EmptyPrevKeepsChunk(t *testing.T) { } // TestTakeFromEndRespectsTokenCount and TestTakeFromStartRespectsTokenCount -// exercise migration diff Chunker-2.4: takeFromEnd/takeFromStart used a +// covers takeFromEnd/takeFromStart used a // fixed 4-bytes-per-token heuristic which badly over-counts for CJK text // (≈3 bytes/char, 1-2 tokens/char). They must now count tokens exactly via // tokenizeStr so the returned slice is close to the requested token budget. diff --git a/internal/ingestion/component/chunker/token_tag_test.go b/internal/ingestion/component/chunker/token_tag_test.go new file mode 100644 index 0000000000..97ba086a21 --- /dev/null +++ b/internal/ingestion/component/chunker/token_tag_test.go @@ -0,0 +1,87 @@ +// 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. +// +// 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 chunker + +import ( + "context" + "strings" + "testing" +) + +// TestTokenChunker_TextPath_StripsParserTags pins residual B: the +// TokenChunker text path must not leak parser position tags (@@...##) +// into the final chunk text. The input mimics a PDF parser payload where +// the body text still carries coordinate markers. +func TestTokenChunker_TextPath_StripsParserTags(t *testing.T) { + c, err := NewTokenChunker(map[string]any{ + "delimiter_mode": "token_size", + "chunk_token_size": 1000, + "delimiters": []string{"\n"}, + }) + if err != nil { + t.Fatalf("NewTokenChunker: %v", err) + } + out, err := c.Invoke(context.Background(), nil, map[string]any{ + "name": "doc.txt", + "output_format": "text", + "text": "First line with tag@@123## inside.\nSecond line with tag@@456## inside.", + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + chunks, _ := out["chunks"].([]map[string]any) + if len(chunks) == 0 { + t.Fatal("chunks: want >=1, got 0") + } + for i, ck := range chunks { + txt, _ := ck["text"].(string) + if strings.Contains(txt, "@@") || strings.Contains(txt, "##") { + t.Errorf("chunk %d text leaks parser tag: %q", i, txt) + } + } +} + +// TestTokenChunker_JSONPath_StripsParserTags pins residual B for the +// structured (output_format == "json") path. +func TestTokenChunker_JSONPath_StripsParserTags(t *testing.T) { + c, err := NewTokenChunker(map[string]any{ + "delimiter_mode": "delimiter", + "delimiters": []string{"\n"}, + }) + if err != nil { + t.Fatalf("NewTokenChunker: %v", err) + } + items := []map[string]any{ + {"text": "Alpha@@1## text\nBeta@@2## text", "doc_type_kwd": "text"}, + {"text": "Gamma@@3## text\nDelta@@4## text", "doc_type_kwd": "text"}, + } + out, err := c.Invoke(context.Background(), nil, map[string]any{ + "name": "doc.md", + "output_format": "json", + "json": items, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + chunks, _ := out["chunks"].([]map[string]any) + if len(chunks) == 0 { + t.Fatal("chunks: want >=1, got 0") + } + for i, ck := range chunks { + txt, _ := ck["text"].(string) + if strings.Contains(txt, "@@") || strings.Contains(txt, "##") { + t.Errorf("chunk %d text leaks parser tag: %q", i, txt) + } + } +} diff --git a/internal/ingestion/component/chunker/token_test.go b/internal/ingestion/component/chunker/token_test.go index 32ac5d9b44..bb0c432ff2 100644 --- a/internal/ingestion/component/chunker/token_test.go +++ b/internal/ingestion/component/chunker/token_test.go @@ -19,6 +19,8 @@ package chunker import ( "context" "math" + "reflect" + "strings" "testing" "ragflow/internal/agent/runtime" @@ -382,7 +384,7 @@ func TestTokenChunker_NewAcceptsPythonOverlappedRange(t *testing.T) { } // TestNormalizeOverlappedPercent is the Go port of Python -// common/float_utils.py:50-58 normalize_overlapped_percent (diff Chunker-2.6). +// common/float_utils.py:50-58 normalize_overlapped_percent. . // Python's user-facing input is a [0,1) fraction; the helper converts it to // the [0,90] integer-percentage scale the merge math expects. func TestNormalizeOverlappedPercent(t *testing.T) { @@ -400,6 +402,10 @@ func TestNormalizeOverlappedPercent(t *testing.T) { {"clamp 95 -> 90", 95, 90}, {"clamp -5 -> 0", -5, 0}, {"negative fraction -0.1 -> 0", -0.1, 0}, + {`numeric string "10" -> 10`, "10", 10}, + {`numeric string fraction "0.1" -> 10`, "0.1", 10}, + {"bad string -> 0", "abc", 0}, + {"nil -> 0", nil, 0}, // Huge out-of-range values must clamp to 90 (not 0). Go's // float->int is implementation-defined past int's range, so the // clamp must run before truncation (review finding #4). Python's @@ -407,10 +413,6 @@ func TestNormalizeOverlappedPercent(t *testing.T) { {"huge 1e300 -> 90", 1e300, 90}, {"huge -1e300 -> 0", -1e300, 0}, {"huge math.MaxFloat64 -> 90", math.MaxFloat64, 90}, - {`numeric string "10" -> 10`, "10", 10}, - {`numeric string fraction "0.1" -> 10`, "0.1", 10}, - {"bad string -> 0", "abc", 0}, - {"nil -> 0", nil, 0}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -421,8 +423,32 @@ func TestNormalizeOverlappedPercent(t *testing.T) { } } +// TestTokenChunkerParam_UpdatePreservesOverlappedPercent covers review +// finding #3: tokenChunkerParam.Update must not reset OverlappedPercent to 0 +// when the incoming config omits the key. All other fields use a presence +// guard, and a partial Update (e.g. changing only chunk_token_size) must +// preserve the previously configured overlap instead of clobbering it. +func TestTokenChunkerParam_UpdatePreservesOverlappedPercent(t *testing.T) { + p := defaultsToken(tokenChunkerParam{}) + p.TokenChunkerParam.OverlappedPercent = 30 // pre-existing config + + // Partial update: only chunk_token_size changes. + p.Update(map[string]any{"chunk_token_size": 100}) + if p.TokenChunkerParam.OverlappedPercent != 30 { + t.Errorf("after partial Update: overlapped_percent=%v, want 30 (preserved)", + p.TokenChunkerParam.OverlappedPercent) + } + + // Explicit key still wins and normalizes. + p.Update(map[string]any{"overlapped_percent": 0.5}) + if p.TokenChunkerParam.OverlappedPercent != 50 { + t.Errorf("after explicit Update: overlapped_percent=%v, want 50 (normalized)", + p.TokenChunkerParam.OverlappedPercent) + } +} + // TestTokenChunker_NormalizesOverlappedPercent asserts the stored value after -// construction matches Python's normalized [0,90] scale (diff Chunker-2.6). +// construction matches Python's normalized [0,90] scale. . func TestTokenChunker_NormalizesOverlappedPercent(t *testing.T) { cases := []struct { name string @@ -451,37 +477,6 @@ func TestTokenChunker_NormalizesOverlappedPercent(t *testing.T) { } } -// TestSplitIntoSections locks the CRLF-safe blank-line splitting used by -// mergeByTokenSize. It does NOT claim Chunker-2.7/2.10 coverage: that -// section-splitting semantics is still OPEN in -// docs/migration_python_go_diff.md (flow-vs-app Python alignment target -// undecided), so this only guards the existing behaviour. -func TestSplitIntoSections(t *testing.T) { - tests := []struct { - name string - text string - want int - }{ - {"two paragraphs", "a\n\nb", 2}, - {"three paragraphs with excess blank lines", "a\n\n\nb\n\nc", 3}, - {"single paragraph", "hello world", 1}, - // A blank line containing only whitespace is still a boundary - // under \n\s*\n (this is what reverted the stricter \n{2,}). - {"blank line with space is a boundary", "a\n \nb", 2}, - {"leading blank lines produce empty prefix (caller filters)", "\n\ntext", 2}, - // CRLF regression guard: \r\n\r\n must split the same as \n\n. - {"CRLF paragraph break splits", "a\r\n\r\nb", 2}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := splitIntoSections(tt.text) - if len(got) != tt.want { - t.Errorf("splitIntoSections(%q) = %d sections, want %d: %v", tt.text, len(got), tt.want, got) - } - }) - } -} - // TestTokenChunkerParam_ValidateOverlappedRange covers the strict overlap // handling in TokenChunkerParam.Validate (review findings #3 + #4): // a directly-constructed struct with a [0,1) fraction is scaled to its @@ -526,27 +521,134 @@ func TestTokenChunkerParam_ValidateOverlappedRange(t *testing.T) { } } -// TestTokenChunkerParam_UpdatePreservesOverlappedPercent locks review -// finding #3: tokenChunkerParam.Update must not reset OverlappedPercent to 0 -// when the incoming config omits the key. All other fields use a presence -// guard, and a partial Update (e.g. changing only chunk_token_size) must -// preserve the previously configured overlap instead of clobbering it. -func TestTokenChunkerParam_UpdatePreservesOverlappedPercent(t *testing.T) { - p := defaultsToken(tokenChunkerParam{}) - p.TokenChunkerParam.OverlappedPercent = 30 // pre-existing config - - // Partial update: only chunk_token_size changes. - p.Update(map[string]any{"chunk_token_size": 100}) - - if p.TokenChunkerParam.OverlappedPercent != 30 { - t.Errorf("after partial Update: overlapped_percent=%v, want 30 (preserved)", - p.TokenChunkerParam.OverlappedPercent) +// TestMergeByTokenSize_CRLFNormalization verifies that, like Python +// naive_merge, line endings are normalised +// (replace("\r\n","\n").replace("\r","\n")) before splitting, so CRLF/CR +// input must segment and split exactly like the equivalent LF input, and +// no carriage return may survive into a produced chunk. +func TestMergeByTokenSize_CRLFNormalization(t *testing.T) { + chunkTexts := func(t *testing.T, text string) []string { + t.Helper() + c := &TokenChunkerComponent{} + c.param.ChunkTokenSize = 128 + c.param.OverlappedPercent = 0 + out := c.mergeByTokenSize(text, nil) + raw, ok := out["chunks"].([]map[string]any) + if !ok { + t.Fatalf("mergeByTokenSize output missing chunks: %v", out) + } + texts := make([]string, 0, len(raw)) + for _, m := range raw { + if s, ok := m["text"].(string); ok { + texts = append(texts, s) + } + } + return texts } - // Explicit key still wins and normalizes. - p.Update(map[string]any{"overlapped_percent": 0.5}) - if p.TokenChunkerParam.OverlappedPercent != 50 { - t.Errorf("after explicit Update: overlapped_percent=%v, want 50 (normalized)", - p.TokenChunkerParam.OverlappedPercent) + t.Run("no carriage return survives", func(t *testing.T) { + texts := chunkTexts(t, "Para A\r\nPara B\r\nPara C") + if len(texts) == 0 { + t.Fatalf("expected chunks, got none") + } + for _, s := range texts { + if strings.Contains(s, "\r") { + t.Errorf("chunk text contains carriage return: %q", s) + } + } + }) + + t.Run("CRLF equals LF", func(t *testing.T) { + // With no blank-line pre-splitting, CRLF is + // normalised to LF and the blank-line run is preserved (not + // collapsed), so equal newline counts must yield equal chunks. + crlf := chunkTexts(t, "Para A\r\nPara B\r\nPara C") + lf := chunkTexts(t, "Para A\nPara B\nPara C") + if !reflect.DeepEqual(crlf, lf) { + t.Errorf("CRLF/LF divergence:\n crlf=%v\n lf =%v", crlf, lf) + } + }) +} + +// TestMergeByTokenSize_PreservesBlankLines verifies that an original +// blank-line run survives in the produced chunk: naive_merge treats the +// whole payload as a single section and does NOT split on blank lines. +func TestMergeByTokenSize_PreservesBlankLines(t *testing.T) { + c := &TokenChunkerComponent{} + c.param.ChunkTokenSize = 128 + c.param.OverlappedPercent = 0 + out := c.mergeByTokenSize("A\n\n\nB", nil) + raw, ok := out["chunks"].([]map[string]any) + if !ok { + t.Fatalf("mergeByTokenSize output missing chunks: %v", out) + } + var joined strings.Builder + for _, m := range raw { + if s, ok := m["text"].(string); ok { + joined.WriteString(s) + } + } + if got := joined.String(); !strings.Contains(got, "A\n\n\nB") { + t.Errorf("blank-line run not preserved: got chunk text %q, want it to contain %q", got, "A\n\n\nB") + } +} + +// TestMergeByTokenSize_OversizeDropsDelimiters verifies that when a +// section exceeds chunk_token_size, naive_merge splits on sentence +// delimiters with a capturing-group re.split but then SKIPS any segment +// that is a pure delimiter (re.fullmatch(dels, sub_sec)), so the +// delimiter character ("。") is DROPPED from the produced chunk text +// rather than retained (rag/nlp/__init__.py:1216-1225). The Go port uses +// regexp.Split (which discards the delimiter) and prepends a single "\n", +// so the merged chunk text must NOT contain the original "。". +func TestMergeByTokenSize_OversizeDropsDelimiters(t *testing.T) { + c := &TokenChunkerComponent{} + c.param.ChunkTokenSize = 5 + c.param.OverlappedPercent = 0 + text := "第一句。第二句。第三句。第四句。第五句。" + out := c.mergeByTokenSize(text, nil) + raw, ok := out["chunks"].([]map[string]any) + if !ok { + t.Fatalf("mergeByTokenSize output missing chunks: %v", out) + } + var joined strings.Builder + for _, m := range raw { + if s, ok := m["text"].(string); ok { + joined.WriteString(s) + } + } + if got := joined.String(); strings.Contains(got, "。") { + t.Errorf("sentence delimiter retained (Python drops it): got chunk text %q, want it to NOT contain %q", got, "。") + } +} + +// TestMergeByTokenSize_OversizeDropsBlankLines covers review #1: in the +// oversize sentence-split path, a blank line whose "\n" delimiter segments +// are dropped (matching Python naive_merge, rag/nlp/__init__.py:1216-1225) +// must NOT survive as a blank line. The lone "\n" delimiters produced by +// "\n\n" are dropped, so the merged chunk text must not contain "\n\n". +func TestMergeByTokenSize_OversizeDropsBlankLines(t *testing.T) { + c := &TokenChunkerComponent{} + c.param.ChunkTokenSize = 100 + c.param.OverlappedPercent = 0 + // Two ~70-char blocks (no sentence punctuation, so the only + // delimiters are the two blank-line newlines) separated by "\n\n". + // Total exceeds 100 tokens → oversize path; the "\n" delimiters are + // dropped, mirroring Python, so the blank line must not survive. + block := strings.Repeat("知识库检索增强生成技术", 7) // 70 chars + text := block + "\n\n" + block + out := c.mergeByTokenSize(text, nil) + raw, ok := out["chunks"].([]map[string]any) + if !ok { + t.Fatalf("mergeByTokenSize output missing chunks: %v", out) + } + var joined strings.Builder + for _, m := range raw { + if s, ok := m["text"].(string); ok { + joined.WriteString(s) + } + } + if got := joined.String(); strings.Contains(got, "\n\n") { + t.Errorf("blank line survived in oversize path (Python drops it): got chunk text %q, want no blank line (\\n\\n)", got) } } diff --git a/internal/ingestion/component/media_dispatch.go b/internal/ingestion/component/media_dispatch.go index 8d184bbb27..b359776feb 100644 --- a/internal/ingestion/component/media_dispatch.go +++ b/internal/ingestion/component/media_dispatch.go @@ -65,58 +65,22 @@ func maybeDispatchVideo( if fileType != utility.FileTypeVIDEO { return parserDispatchResult{}, false, nil } - setup, ok := setups["video"] - if !ok { + if _, ok := setups["video"]; !ok { return parserDispatchResult{}, false, nil } - tenantID := getStringOr(inputs, "tenant_id", "") - if tenantID == "" { - return parserDispatchResult{}, true, - fmt.Errorf("parser: video requires tenant_id") - } - // Resolve the tenant's IMAGE2TEXT model. - driver, modelName, apiConfig, _, err := resolveTenantModelByType(ctx, db, tenantID, entity.ModelTypeImage2Text) - if err != nil { - return parserDispatchResult{}, true, - fmt.Errorf("parser: video image2text model: %w", err) - } - - videoPrompt, _ := setup["prompt"].(string) - videoB64 := base64.StdEncoding.EncodeToString(binary) - - // Build a multimodal message with the video payload. - // Python uses cv_mdl.async_chat(video_bytes=blob, ...); - // Go ChatWithMessages is synchronous and uses a data URI. - mimeType := videoMIME(filename) - dataURI := "data:" + mimeType + ";base64," + videoB64 - messages := []modelModule.Message{{ - Role: "user", - Content: []interface{}{ - map[string]any{"type": "text", "text": videoPrompt}, - map[string]any{"type": "video_url", "video_url": map[string]any{"url": dataURI}}, - }, - }} - vision := true - resp, err := driver.ChatWithMessages(ctx, modelName, messages, apiConfig, &modelModule.ChatConfig{Vision: &vision}, nil) - if err != nil { - return parserDispatchResult{}, true, - fmt.Errorf("parser: video describe: %w", err) - } - txt := "" - if resp != nil && resp.Answer != nil { - txt = strings.TrimSpace(*resp.Answer) - } - - outputFormat, _ := setup["output_format"].(string) - if outputFormat == "" { - outputFormat = "text" - } - return parserDispatchResult{ - OutputFormat: outputFormat, - DocType: "video", - Text: txt, - }, true, nil + // Video parsing is intentionally not implemented yet: the underlying + // video-analysis capability is pending. The previously-shipped path sent a + // video_url data URI, but no model driver honors it — OpenAI-compatible + // drivers only accept image_url (the block is ignored), and Gemini's + // googleMessageParts only accepts text/image_url and silently drops + // video_url. Returning an explicit error is safer than silently producing + // a description from the prompt text alone. The real implementation must + // be provider-specific (OpenAI-compatible: frame extraction -> image_url; + // Gemini: raw-bytes inline_data; Qwen: file://) — see + // docs/migration_python_go_diff.md 2.7. + return parserDispatchResult{}, true, + fmt.Errorf("Parser: video parsing is not yet supported; underlying video analysis capability is pending") } // Image dispatch: OCR + IMAGE2TEXT vision describe --- @@ -321,7 +285,7 @@ func maybeDispatchAudio( outputFormat, _ := setup["output_format"].(string) if outputFormat == "" { - outputFormat = "text" + outputFormat = "json" } // Diff 2.11: when output_format is "json" the transcription must be // carried as a JSON item. Returning it only in Text made the Invoke @@ -415,7 +379,10 @@ func imageMIME(filename string) string { } // videoMIME maps common video filename extensions to MIME types -// for constructing base64 data URIs. +// for constructing base64 data URIs. Retained as a reference for the +// future real video-parsing implementation (provider-specific frame +// extraction / inline_data / file://); not currently used because +// maybeDispatchVideo returns an explicit unsupported error. func videoMIME(filename string) string { dot := strings.LastIndex(filename, ".") if dot == -1 { diff --git a/internal/ingestion/component/media_dispatch_test.go b/internal/ingestion/component/media_dispatch_test.go index 010dc0888b..61dfdd4bf2 100644 --- a/internal/ingestion/component/media_dispatch_test.go +++ b/internal/ingestion/component/media_dispatch_test.go @@ -324,6 +324,39 @@ func TestMaybeDispatchAudio_TextCarriesTranscription(t *testing.T) { } } +// TestMaybeDispatchAudio_DefaultOutputFormatJson covers Parser 2.11: +// the default audio output_format must be "json" (matching Python +// parser.py:232 and AllowedOutputFormat["audio"]={"json"}). +func TestMaybeDispatchAudio_DefaultOutputFormatJson(t *testing.T) { + const want = "hello world" + drv := &audioTranscribeDriver{transcription: want} + orig := resolveTenantModelByType + defer func() { resolveTenantModelByType = orig }() + resolveTenantModelByType = func(ctx context.Context, db *gorm.DB, tenantID string, modelType entity.ModelType) (modelModule.ModelDriver, string, *modelModule.APIConfig, int, error) { + return drv, "asr-model", &modelModule.APIConfig{}, 0, nil + } + setups := defaultSetups() + // Do NOT set output_format — exercise the default path. + res, dispatched, err := maybeDispatchAudio( + context.Background(), + nil, + utility.FileTypeAURAL, + "test.mp3", + []byte("fake-audio"), + map[string]any{"tenant_id": "t1"}, + setups, + ) + if err != nil { + t.Fatalf("maybeDispatchAudio: %v", err) + } + if !dispatched { + t.Fatal("expected dispatched=true for AURAL file") + } + if res.OutputFormat != "json" { + t.Fatalf("default OutputFormat = %q, want json", res.OutputFormat) + } +} + // TestMaybeDispatchMarkdownVision_EnhancesTables pins diff 2.5: markdown // vision enhancement must also process items whose doc_type_kwd is "table" // (Python checks {"image","table"} in parser/utils.py:181), not only "image". diff --git a/internal/ingestion/component/parser.go b/internal/ingestion/component/parser.go index 8659b9c092..dacfc6713d 100644 --- a/internal/ingestion/component/parser.go +++ b/internal/ingestion/component/parser.go @@ -311,7 +311,7 @@ func defaultSetups() map[string]schema.ParserSetup { "aiff", "au", "midi", "wma", "realaudio", "vqf", "oggvorbis", "ape", }, - "output_format": "text", + "output_format": "json", }, "video": { "suffix": []string{"mp4", "avi", "mkv"}, @@ -490,6 +490,13 @@ func (c *ParserComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[st // referenced images (![alt](url)). Mirrors Python's // enhance_media_sections_with_vision in _Markdown. dispatched, _, _ = maybeDispatchMarkdownVision(ctx, db, fileTypeExt, dispatched, inputs) + + // PDF vision figure enhancement: enrich parsed PDF JSON + // items with vision-model descriptions of embedded + // images/tables (doc_type_kwd "image"/"table" with non-empty + // image field). Mirrors Python's enhance_media_sections_with_vision + // in parser.py:_pdf + dispatched, _, _ = maybeDispatchPDFVisionEnhancement(ctx, db, fileTypeExt, dispatched, inputs) } // Known/supported families must fail loudly when dispatch or // parsing breaks. Only unknown families keep the raw-text fallback. diff --git a/internal/ingestion/component/pdf_vision_dispatch.go b/internal/ingestion/component/pdf_vision_dispatch.go index f98b6bb47c..ec9a67e5db 100644 --- a/internal/ingestion/component/pdf_vision_dispatch.go +++ b/internal/ingestion/component/pdf_vision_dispatch.go @@ -558,3 +558,96 @@ func isMinerUDriver(driver modelModule.ModelDriver) bool { } return false } + +// maybeDispatchPDFVisionEnhancement mirrors Python's +// enhance_media_sections_with_vision for PDF +// After the normal PDF parser produces JSON items, this function +// enriches image/table items by calling the tenant's IMAGE2TEXT +// model and appending vision descriptions to each item's text field. +// The markdown/text output paths are not enhanced (Python does the same). +// +// This follows the exact same convention as maybeDispatchDOCXVision and +// maybeDispatchMarkdownVision: it is not gated by a separate setup flag, +// it simply resolves the tenant's IMAGE2TEXT model and skips silently +// when none is configured — matching Python's try/except pass behaviour. +func maybeDispatchPDFVisionEnhancement( + ctx context.Context, + db *gorm.DB, + fileType utility.FileType, + dispatched parserDispatchResult, + inputs map[string]any, +) (parserDispatchResult, bool, error) { + if fileType != utility.FileTypePDF { + return dispatched, false, nil + } + if dispatched.Err != nil || dispatched.OutputFormat != "json" || len(dispatched.JSON) == 0 { + return dispatched, false, nil + } + tenantID := getStringOr(inputs, "tenant_id", "") + if tenantID == "" { + return dispatched, false, nil + } + driver, modelName, apiConfig, _, err := resolveTenantModelByType(ctx, db, tenantID, entity.ModelTypeImage2Text) + if err != nil { + return dispatched, false, nil + } + type target struct{ idx int } + var targets []target + for i, item := range dispatched.JSON { + kd, _ := item["doc_type_kwd"].(string) + if kd != "image" && kd != "table" { + continue + } + img, _ := item["image"].(string) + if img == "" { + continue + } + targets = append(targets, target{idx: i}) + } + if len(targets) == 0 { + return dispatched, false, nil + } + descriptions := make([]string, len(targets)) + var wg sync.WaitGroup + sem := make(chan struct{}, pdfVisionEnhanceConcurrency) + for slot, tg := range targets { + sem <- struct{}{} + wg.Add(1) + go func(slot int, itemIdx int) { + defer wg.Done() + defer func() { <-sem }() + img, _ := dispatched.JSON[itemIdx]["image"].(string) + if img == "" { + return + } + prompt, perr := docxVisionPromptBuilder("", "") + if perr != nil { + return + } + messages := buildVisionMessages(prompt, img) + resp, ierr := visionChatInvoker(ctx, driver, modelName, messages, apiConfig) + if ierr != nil { + return + } + descriptions[slot] = extractDOCXVisionAnswer(resp) + }(slot, tg.idx) + } + wg.Wait() + modified := false + for slot, tg := range targets { + desc := strings.TrimSpace(descriptions[slot]) + if desc == "" { + continue + } + existing, _ := dispatched.JSON[tg.idx]["text"].(string) + if existing != "" { + dispatched.JSON[tg.idx]["text"] = existing + "\n" + desc + } else { + dispatched.JSON[tg.idx]["text"] = desc + } + modified = true + } + return dispatched, modified, nil +} + +var pdfVisionEnhanceConcurrency = 10 diff --git a/internal/ingestion/component/schema/chunker.go b/internal/ingestion/component/schema/chunker.go index d8e7ddc952..6e9453dea8 100644 --- a/internal/ingestion/component/schema/chunker.go +++ b/internal/ingestion/component/schema/chunker.go @@ -188,7 +188,7 @@ type TokenChunkerParam struct { // OverlappedPercent is the overlap percentage in [0, 90]. Mirrors // Python common/float_utils.py:50-58 — an integer-like float in the // same range so DSL templates written for the Python pipeline work - // out of the box (diff Chunker-2.6). A [0,1) fraction input is also + // out of the box A [0,1) fraction input is also // accepted and normalized to this scale by tokenChunkerParam.Update // (via normalizeOverlappedPercent). OverlappedPercent float64 `json:"overlapped_percent"` diff --git a/internal/ingestion/component/tokenizer.go b/internal/ingestion/component/tokenizer.go index 524801a2e9..cf7c5644cc 100644 --- a/internal/ingestion/component/tokenizer.go +++ b/internal/ingestion/component/tokenizer.go @@ -35,18 +35,6 @@ // `internal/tokenizer/tokenizer.go:Tokenize` (Infinity engine // returns input unchanged; otherwise the C++ binding is used). // -// - CJK CAVEAT (plan §8 Q2): The `NumTokensFromString` helper in -// `internal/tokenizer` falls back to `len([]byte(s))` on a -// tiktoken-init failure (over-counts CJK). The Python equivalent -// returns 0. The Go port KEEPS the Go behaviour — the tokenizer -// package is the single source of truth for token counting and -// must not be re-implemented here. Test -// `TestTokenizerComponent_Invoke_Unicode` asserts only that the -// count is finite and non-negative, matching the test -// convention in plan §6 (coverage target: -// "Tokenizer returns finite token counts for empty / unicode / -// mixed-script text"). -// // - EMBEDDING MODEL RESOLUTION: mirrored. Python uses // `LLMBundle(tenant_id, embd_id).encode([...])` from // `rag/flow/tokenizer/tokenizer.py:54-66`; the Go port goes @@ -73,8 +61,8 @@ // Drivers that need to chunk internally can do so — the wire // call is one round-trip. // -// - TRACKING: WithTimeout (60s, matches python `@timeout(60)` on -// `batch_encode`), TrackProgress, TrackElapsed. See +// - TRACKING: TrackProgress, TrackElapsed. See +// `internal/agent/runtime/helpers.go` (plan §1 Phase 1). // `internal/agent/runtime/helpers.go` (plan §1 Phase 1). // // - WHAT IS NOT PORTED: @@ -99,7 +87,6 @@ import ( "slices" "strconv" "strings" - "time" "go.uber.org/zap" "gorm.io/gorm" @@ -114,25 +101,19 @@ import ( const ComponentNameTokenizer = "Tokenizer" -// tokenizerTimeout returns the per-batch timeout for embedding API calls. -// Reads COMPONENT_EXEC_TIMEOUT_TOKENIZER env var (seconds); defaults to 600s -// (10 min) to match the canvas-level component timeout default. -// Invalid / non-positive values fall back to the default. -func tokenizerTimeout() time.Duration { - if v := os.Getenv("COMPONENT_EXEC_TIMEOUT_TOKENIZER"); v != "" { - if secs, err := strconv.Atoi(v); err == nil && secs > 0 { - return time.Duration(secs) * time.Second +// embeddingBatchSize returns the embedding batch size, matching Python's +// settings.EMBEDDING_BATCH_SIZE. Reads TOKENIZER_EMBEDDING_BATCH_SIZE env +// var; defaults to 16. Invalid / non-positive values fall back to the +// default (diff Tokenizer Omission-3). +func embeddingBatchSize() int { + if v := os.Getenv("TOKENIZER_EMBEDDING_BATCH_SIZE"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n } } - return defaultTokenizerTimeout + return 16 } -var defaultTokenizerTimeout = 600 * time.Second - -// tokenizerEmbeddingBatchSize mirrors Python's -// settings.EMBEDDING_BATCH_SIZE default. -var tokenizerEmbeddingBatchSize = 16 - // titleExtRE strips a trailing file-extension (e.g. ".pdf") from the // upstream document name before tokenizing it. Mirrors the python // `re.sub(r"\.[a-zA-Z]+$", "", name)` in tokenizer.py:137. @@ -357,6 +338,15 @@ func (c *TokenizerComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map normalizeChunkTextFallback(chunks) + // Diff Tokenizer-8: chunk_order_int must be set on all paths + // (Python sets it unconditionally). tokenizeChunks also sets it + // for the full_text path; this covers the embedding-only path. + for i := range chunks { + if chunks[i].ChunkOrderInt == nil { + chunks[i].ChunkOrderInt = intPtr(i) + } + } + language := globals.GlobalOrInput(ctx, inputs, "lang", "English") if contains(c.param.SearchMethod, "full_text") { @@ -439,7 +429,7 @@ func (c *TokenizerComponent) embedChunks(ctx context.Context, tenantID, kbID, em // tokenizer.py:95 which passes name verbatim to embedding. The // empty-name guard above still uses TrimSpace, matching Python's // `.strip()==""` check at tokenizer.py:200. - titleResults, err := encodeWithTimeout(ctx, embedder, []string{name}) + titleResults, err := embedder.Encode(ctx, []string{name}) if err != nil { return nil, 0, fmt.Errorf("Tokenizer: encode title: %w", err) } @@ -452,12 +442,12 @@ func (c *TokenizerComponent) embedChunks(ctx context.Context, tenantID, kbID, em } contentResults := make([]EmbeddingResult, 0, len(texts)) - for start := 0; start < len(texts); start += tokenizerEmbeddingBatchSize { - end := start + tokenizerEmbeddingBatchSize + for start := 0; start < len(texts); start += embeddingBatchSize() { + end := start + embeddingBatchSize() if end > len(texts) { end = len(texts) } - batchResults, err := encodeWithTimeout(ctx, embedder, texts[start:end]) + batchResults, err := embedder.Encode(ctx, texts[start:end]) if err != nil { return nil, 0, fmt.Errorf("Tokenizer: encode: %w", err) } @@ -486,24 +476,13 @@ func (c *TokenizerComponent) embedChunks(ctx context.Context, tenantID, kbID, em return chunks, tokenCount, nil } -func encodeWithTimeout(ctx context.Context, embedder Embedder, texts []string) ([]EmbeddingResult, error) { - var ( - results []EmbeddingResult - encErr error - ) - timeoutErr := runtime.WithTimeout(ctx, tokenizerTimeout(), func(timeoutCtx context.Context) error { - results, encErr = embedder.Encode(ctx, texts) - return encErr - }) - if timeoutErr != nil { - return nil, timeoutErr - } - return results, nil -} - +// truncateForEmbedding truncates text to fit within maxTokens, +// reserving 10 tokens for special/model overhead. Mirrors Python +// tokenizer.py truncation: when maxTokens <= 10, the text is +// truncated to empty func truncateForEmbedding(text string, maxTokens int) string { if maxTokens <= 10 { - return text + return "" } return tokenizer.TrimContentToTokenLimit(text, maxTokens-10) } @@ -552,18 +531,39 @@ func stripRuntimeTimestamps(inputs map[string]any) map[string]any { } func chunksFromTokenizerUpstream(in schema.TokenizerFromUpstream) []schema.ChunkDoc { + var raw []schema.ChunkDoc switch in.OutputFormat { case schema.PayloadFormatChunks: - return cloneChunkDocs(in.Chunks) + raw = cloneChunkDocs(in.Chunks) case schema.PayloadFormatMarkdown: - return textPayloadToChunks(in.MarkdownResult) + raw = textPayloadToChunks(in.MarkdownResult) case schema.PayloadFormatText: - return textPayloadToChunks(in.TextResult) + raw = textPayloadToChunks(in.TextResult) case schema.PayloadFormatHTML: - return textPayloadToChunks(in.HTMLResult) + raw = textPayloadToChunks(in.HTMLResult) default: - return cloneChunkDocs(in.JSONResult) + raw = cloneChunkDocs(in.JSONResult) } + // Discard zero-value ChunkDocs (no text, no content_with_weight, no + // image, no summary) so they don't produce phantom embeddings + // downstream. Python's pipeline filters none-chunks before the + // tokenizer. + filtered := raw[:0] + for _, ck := range raw { + if isPhantomChunk(ck) { + continue + } + filtered = append(filtered, ck) + } + return filtered +} + +// isPhantomChunk returns true when a ChunkDoc has no usable content for +// downstream tokenization or embedding. A chunk carrying only a Summary +// (no Text/Image/ContentWithWeight) is kept — tokenizeChunks tokenizes the +// Summary in that case. Mirrors Python's none-chunk filtering. +func isPhantomChunk(ck schema.ChunkDoc) bool { + return ck.Text == "" && ck.Image == "" && ck.ContentWithWeight == "" && ck.Summary == "" } func textPayloadToChunks(payload *string) []schema.ChunkDoc { @@ -683,7 +683,10 @@ func tokenizeChunks(chunks []schema.ChunkDoc, titleStem string, language string) return fmt.Errorf("Tokenizer: keyword tokens marshal: %w", err) } } - if s := ck.Summary; strings.TrimSpace(s) != "" { + // Keep Go: skip whitespace-only summaries so they don't shadow + // the real Text. Python's truthy check (tokenizer.py:155) treats + // " " as present and blanks out content_ltks; Go is more sensible. + if s := strings.TrimSpace(ck.Summary); s != "" { st, err := tok.Tokenize(s) if err != nil { return fmt.Errorf("Tokenizer: summary tokenize: %w", err) diff --git a/internal/ingestion/component/tokenizer_test.go b/internal/ingestion/component/tokenizer_test.go index afdebcb587..5dd986e6d2 100644 --- a/internal/ingestion/component/tokenizer_test.go +++ b/internal/ingestion/component/tokenizer_test.go @@ -365,7 +365,7 @@ func TestTokenizerComponent_Invoke_EncoderCountMismatch(t *testing.T) { // Inject an embedder that returns the wrong number of vectors // regardless of input. wrong := &countMismatchedEmbedder{want: 1} - cIntf, err := NewTokenizerComponentWithResolver(nil, func(_, _, _ string) (Embedder, error) { return wrong, nil }) + cIntf, err := NewTokenizerComponentWithResolver(nil, func(_ context.Context, _, _, _ string) (Embedder, error) { return wrong, nil }) if err != nil { t.Fatalf("NewTokenizerComponentWithResolver: %v", err) } @@ -385,7 +385,7 @@ func TestTokenizerComponent_Invoke_EncoderCountMismatch(t *testing.T) { type countMismatchedEmbedder struct{ want int } -func (c *countMismatchedEmbedder) MaxTokens() int { return 0 } +func (c *countMismatchedEmbedder) MaxTokens() int { return 2048 } func (c *countMismatchedEmbedder) Encode(ctx context.Context, texts []string) ([]EmbeddingResult, error) { out := make([]EmbeddingResult, c.want) @@ -395,31 +395,6 @@ func (c *countMismatchedEmbedder) Encode(ctx context.Context, texts []string) ([ return out, nil } -// TestTokenizerComponent_Invoke_HonorsTimeout installs an -// embedder that blocks past a (test-shrunk) tokenizerTimeout and -// asserts the component returns context.DeadlineExceeded. -func TestTokenizerComponent_Invoke_HonorsTimeout(t *testing.T) { - requireTokenizerPool(t) - t.Setenv("COMPONENT_EXEC_TIMEOUT_TOKENIZER", "1") - - c, stub := withStubEmbedder(t, 4) - stub.delay = 2 * time.Second - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - _, err := c.Invoke(ctx, nil, map[string]any{ - "output_format": "chunks", - "chunks": []map[string]any{{"text": "alpha"}}, - }) - if err == nil { - t.Fatal("expected timeout error, got nil") - } - if !errors.Is(err, context.DeadlineExceeded) { - t.Errorf("expected context.DeadlineExceeded, got %v", err) - } -} - // TestTokenizerComponent_Smoke_EndToEnd is the BLOCKER smoke test // (plan §8 R3). Drives 1 chunk of ~1000 tokens through the real // tokenizer and a stub embedder with no artificial latency, then @@ -435,7 +410,7 @@ func TestTokenizerComponent_Invoke_HonorsTimeout(t *testing.T) { // production path against a real embedding API was not exercised // in this CI sandbox; the helper `withStubEmbedder` deliberately // avoids the network round-trip while still exercising the full -// wiring (TrackElapsed, WithTimeout, batched Encode, vector +// wiring (TrackElapsed, batched Encode, vector // stamping). func TestTokenizerComponent_Smoke_EndToEnd(t *testing.T) { @@ -525,7 +500,7 @@ func TestTokenizerComponent_Embedding_UsesFilenameWeight(t *testing.T) { requireTokenizerPool(t) cIntf, err := NewTokenizerComponentWithResolver(map[string]any{ "filename_embd_weight": 0.25, - }, func(_, _, _ string) (Embedder, error) { + }, func(_ context.Context, _, _, _ string) (Embedder, error) { stub := newStubEmbedder(2) stub.resultsByCall = []embeddingCallResult{ {vectors: [][]float64{{8, 8}}, tokenCount: 3}, @@ -672,9 +647,7 @@ func TestTokenizerComponent_Embedding_SetsTokenConsumptionIncludingTitleCall(t * requireTokenizerPool(t) c, stub := withStubEmbedder(t, 2) stub.callTokens = []int{3, 5, 7} - prevBatchSize := tokenizerEmbeddingBatchSize - tokenizerEmbeddingBatchSize = 1 - t.Cleanup(func() { tokenizerEmbeddingBatchSize = prevBatchSize }) + t.Setenv("TOKENIZER_EMBEDDING_BATCH_SIZE", "1") out, err := c.Invoke(context.Background(), nil, map[string]any{ "name": "doc.pdf", @@ -692,9 +665,7 @@ func TestTokenizerComponent_Embedding_SetsTokenConsumptionIncludingTitleCall(t * func TestTokenizerComponent_Embedding_BatchesByConfiguredBatchSize(t *testing.T) { requireTokenizerPool(t) c, stub := withStubEmbedder(t, 2) - prevBatchSize := tokenizerEmbeddingBatchSize - tokenizerEmbeddingBatchSize = 2 - t.Cleanup(func() { tokenizerEmbeddingBatchSize = prevBatchSize }) + t.Setenv("TOKENIZER_EMBEDDING_BATCH_SIZE", "2") if _, err := c.Invoke(context.Background(), nil, map[string]any{ "name": "doc.pdf", @@ -732,7 +703,7 @@ func floatSliceClose(got, want []float64) bool { func TestTokenizerComponent_InstanceResolversDoNotLeakAcrossComponents(t *testing.T) { requireTokenizerPool(t) - compAIntf, err := NewTokenizerComponentWithResolver(nil, func(_, _, _ string) (Embedder, error) { + compAIntf, err := NewTokenizerComponentWithResolver(nil, func(_ context.Context, _, _, _ string) (Embedder, error) { stub := newStubEmbedder(2) stub.resultsByCall = []embeddingCallResult{{vectors: [][]float64{{10, 10}}, tokenCount: 1}, {vectors: [][]float64{{1, 1}}, tokenCount: 1}} return stub, nil @@ -740,7 +711,7 @@ func TestTokenizerComponent_InstanceResolversDoNotLeakAcrossComponents(t *testin if err != nil { t.Fatalf("NewTokenizerComponentWithResolver(A): %v", err) } - compBIntf, err := NewTokenizerComponentWithResolver(nil, func(_, _, _ string) (Embedder, error) { + compBIntf, err := NewTokenizerComponentWithResolver(nil, func(_ context.Context, _, _, _ string) (Embedder, error) { stub := newStubEmbedder(2) stub.resultsByCall = []embeddingCallResult{{vectors: [][]float64{{20, 20}}, tokenCount: 1}, {vectors: [][]float64{{2, 2}}, tokenCount: 1}} return stub, nil @@ -751,11 +722,11 @@ func TestTokenizerComponent_InstanceResolversDoNotLeakAcrossComponents(t *testin compA := compAIntf.(*TokenizerComponent) compB := compBIntf.(*TokenizerComponent) - outA, err := compA.Invoke(context.Background(), map[string]any{"name": "docA", "output_format": "chunks", "chunks": []map[string]any{{"text": "alpha"}}}) + outA, err := compA.Invoke(context.Background(), nil, map[string]any{"name": "docA", "output_format": "chunks", "chunks": []map[string]any{{"text": "alpha"}}}) if err != nil { t.Fatalf("Invoke A: %v", err) } - outB, err := compB.Invoke(context.Background(), map[string]any{"name": "docB", "output_format": "chunks", "chunks": []map[string]any{{"text": "beta"}}}) + outB, err := compB.Invoke(context.Background(), nil, map[string]any{"name": "docB", "output_format": "chunks", "chunks": []map[string]any{{"text": "beta"}}}) if err != nil { t.Fatalf("Invoke B: %v", err) } diff --git a/internal/ingestion/component/tokenizer_unit_test.go b/internal/ingestion/component/tokenizer_unit_test.go index b50ac430c4..9e3bdcd65b 100644 --- a/internal/ingestion/component/tokenizer_unit_test.go +++ b/internal/ingestion/component/tokenizer_unit_test.go @@ -23,6 +23,7 @@ package component import ( "context" "encoding/json" + "os" "strings" "sync/atomic" "testing" @@ -90,8 +91,10 @@ func (s *stubEmbedder) Encode(ctx context.Context, texts []string) ([]EmbeddingR } // newStubEmbedder returns a stub embedder for instance-level resolver injection. +// maxTokens defaults to 2048 so truncateForEmbedding (Diff 14: maxTokens <= 10 -> "") +// does not empty the content text; tests that exercise truncation set maxTokens explicitly. func newStubEmbedder(dim int) *stubEmbedder { - return &stubEmbedder{dim: dim} + return &stubEmbedder{dim: dim, maxTokens: 2048} } // withStubEmbedder constructs a TokenizerComponent with an instance-scoped @@ -348,3 +351,147 @@ func TestChunkDocsToMaps_PreservesPDFPositions(t *testing.T) { t.Errorf("positions page already converted; want raw 1-indexed page 1, got %v", got[0][0]) } } + +// TestIsPhantomChunk verifies that zero-value ChunkDocs (no Text, no +// Image, no ContentWithWeight, no Summary) are identified as phantom, +// while any one of those fields being present keeps the chunk. +func TestIsPhantomChunk(t *testing.T) { + if !isPhantomChunk(schema.ChunkDoc{}) { + t.Error("empty ChunkDoc must be phantom") + } + if !isPhantomChunk(schema.ChunkDoc{Text: ""}) { + t.Error("ChunkDoc with empty Text only must be phantom") + } + if isPhantomChunk(schema.ChunkDoc{Text: "hello"}) { + t.Error("ChunkDoc with Text must not be phantom") + } + if isPhantomChunk(schema.ChunkDoc{Image: "data:image/png;base64,abc"}) { + t.Error("ChunkDoc with Image must not be phantom") + } + if isPhantomChunk(schema.ChunkDoc{ContentWithWeight: "weight"}) { + t.Error("ChunkDoc with ContentWithWeight must not be phantom") + } + if isPhantomChunk(schema.ChunkDoc{Summary: "a summary"}) { + t.Error("ChunkDoc with Summary must not be phantom") + } +} + +// TestTruncateForEmbedding_SmallMaxTokens covers Tokenizer Diff-14: when +// maxTokens <= 10, Go must truncate to empty, matching Python's +// truncation-to-empty behaviour (common/token_utils.py:183-185). +func TestTruncateForEmbedding_SmallMaxTokens(t *testing.T) { + long := strings.Repeat("a", 100) + if got := truncateForEmbedding(long, 5); got != "" { + t.Errorf("truncateForEmbedding(maxTokens=5) = %q, want %q", got, "") + } + if got := truncateForEmbedding(long, 10); got != "" { + t.Errorf("truncateForEmbedding(maxTokens=10) = %q, want %q", got, "") + } + // Normal path: maxTokens > 10 should truncate (not return empty). + if got := truncateForEmbedding(long, 50); got == "" { + t.Error("truncateForEmbedding(maxTokens=50) returned empty, want truncated text") + } +} + +// TestEmbeddingBatchSizeEnvVar covers Tokenizer Omission-3: the batch size +// must be configurable via TOKENIZER_EMBEDDING_BATCH_SIZE env var, matching +// Python's configurable settings.EMBEDDING_BATCH_SIZE. +func TestEmbeddingBatchSizeEnvVar(t *testing.T) { + if got := embeddingBatchSize(); got != 16 { + t.Errorf("embeddingBatchSize() default = %d, want 16", got) + } + os.Setenv("TOKENIZER_EMBEDDING_BATCH_SIZE", "32") + t.Cleanup(func() { os.Unsetenv("TOKENIZER_EMBEDDING_BATCH_SIZE") }) + if got := embeddingBatchSize(); got != 32 { + t.Errorf("embeddingBatchSize() after env = %d, want 32", got) + } + // Invalid value falls back to default. + os.Setenv("TOKENIZER_EMBEDDING_BATCH_SIZE", "bad") + if got := embeddingBatchSize(); got != 16 { + t.Errorf("embeddingBatchSize() invalid env = %d, want 16", got) + } +} + +// TestChunkOrderInt_EmbeddingOnly covers Tokenizer Diff-8: chunk_order_int +// must be set even when search_method does not include "full_text" (i.e. +// embedding-only path). tokenizeChunks previously only set it for the +// full_text branch. +func TestChunkOrderInt_EmbeddingOnly(t *testing.T) { + stub := newStubEmbedder(3) + comp, err := NewTokenizerComponentWithResolver( + map[string]any{"search_method": []string{"embedding"}, "fields": []string{"text"}}, + func(ctx context.Context, _, _, _ string) (Embedder, error) { return stub, nil }, + ) + if err != nil { + t.Fatalf("NewTokenizerComponentWithResolver: %v", err) + } + inputs := map[string]any{ + "name": "doc.pdf", + "output_format": "json", + "json": []map[string]any{ + {"text": "first chunk", "doc_type_kwd": "text"}, + {"text": "second chunk", "doc_type_kwd": "text"}, + }, + } + out, err := comp.Invoke(context.Background(), nil, inputs) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + chunks := out["chunks"].([]map[string]any) + if len(chunks) != 2 { + t.Fatalf("want 2 chunks, got %d", len(chunks)) + } + for i, ck := range chunks { + coi, ok := ck["chunk_order_int"] + if !ok { + t.Errorf("chunk %d: chunk_order_int missing (embedding-only path must set it)", i) + } + if coi == nil { + t.Errorf("chunk %d: chunk_order_int is nil", i) + } + } +} + +// TestChunksFromTokenizerUpstream_FiltersPhantomChunks covers Tokenizer +// Omission-2 at the pipeline level: when upstream input contains a +// zero-value ChunkDoc (no Text, no Image, no ContentWithWeight), it must +// be silently dropped from the output before tokenization and embedding. +// This mirrors Python's `if not text and not d.get("image"): continue` +// in tokenizer.py:80-82. +func TestChunksFromTokenizerUpstream_FiltersPhantomChunks(t *testing.T) { + // JSON path: three items, the middle one is a phantom. + items := []map[string]any{ + {"text": "valid chunk", "doc_type_kwd": "text"}, + {}, // phantom: no text, no image, no content_with_weight + {"text": "another valid", "doc_type_kwd": "text"}, + } + // Use embedding-only mode to avoid CGo tokenizer dependency. + stub := newStubEmbedder(3) + comp, err := NewTokenizerComponentWithResolver( + map[string]any{"search_method": []string{"embedding"}, "fields": []string{"text"}}, + func(ctx context.Context, _, _, _ string) (Embedder, error) { return stub, nil }, + ) + if err != nil { + t.Fatalf("NewTokenizerComponentWithResolver: %v", err) + } + out, err := comp.Invoke(context.Background(), nil, map[string]any{ + "name": "doc.pdf", + "output_format": "json", + "json": items, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + chunks := out["chunks"].([]map[string]any) + // Must drop the phantom — only 2 valid chunks remain. + if len(chunks) != 2 { + t.Fatalf("want 2 chunks (phantom filtered), got %d", len(chunks)) + } + // Verify the surviving chunks are the valid ones. + if chunks[0]["text"] != "valid chunk" { + t.Errorf("chunk 0 text = %q, want %q", chunks[0]["text"], "valid chunk") + } + if chunks[1]["text"] != "another valid" { + t.Errorf("chunk 1 text = %q, want %q", chunks[1]["text"], "another valid") + } +} diff --git a/internal/ingestion/pipeline/template_integration_test.go b/internal/ingestion/pipeline/template_integration_test.go index 8d22b96ff4..59ead62fe8 100644 --- a/internal/ingestion/pipeline/template_integration_test.go +++ b/internal/ingestion/pipeline/template_integration_test.go @@ -44,7 +44,7 @@ import ( type fixedEmbedder struct{} -func (fixedEmbedder) MaxTokens() int { return 0 } +func (fixedEmbedder) MaxTokens() int { return 2048 } func (fixedEmbedder) Encode(ctx context.Context, texts []string) ([]componentpkg.EmbeddingResult, error) { out := make([]componentpkg.EmbeddingResult, 0, len(texts)) diff --git a/internal/ingestion/service/real_consumer_pipeline_test.go b/internal/ingestion/service/real_consumer_pipeline_test.go index 9f8e4c7716..23d81f4527 100644 --- a/internal/ingestion/service/real_consumer_pipeline_test.go +++ b/internal/ingestion/service/real_consumer_pipeline_test.go @@ -79,11 +79,11 @@ func TestRealConsumer_PipelineMessageRoutesToExecuteTask(t *testing.T) { } ingestionTaskDAO := dao.NewIngestionTaskDAO() - _, err = ingestionTaskDAO.UpdateStatusIfCurrent(taskMsg.TaskID, common.CREATED, common.RUNNING) + _, err = ingestionTaskDAO.UpdateStatusIfCurrent(context.Background(), db, taskMsg.TaskID, common.CREATED, common.RUNNING) if err != nil { t.Fatalf("UpdateStatusIfCurrent: %v", err) } - task, err := ingestionTaskDAO.GetByID(taskMsg.TaskID) + task, err := ingestionTaskDAO.GetByID(context.Background(), db, taskMsg.TaskID) if err != nil || task == nil { t.Fatalf("task not found after publish: %s", taskMsg.TaskID) } @@ -102,7 +102,7 @@ func TestRealConsumer_PipelineMessageRoutesToExecuteTask(t *testing.T) { return nil } - ingestor.executeTask(taskCtx) + ingestor.executeTask(context.Background(), taskCtx) if !routedToPipeline { t.Fatal("expected executeTask to route queue-consumed pipeline task to runDocumentTask") @@ -111,7 +111,7 @@ func TestRealConsumer_PipelineMessageRoutesToExecuteTask(t *testing.T) { t.Fatalf("Ack: %v", err) } - finalTask, err := ingestionTaskDAO.GetByID(task.ID) + finalTask, err := ingestionTaskDAO.GetByID(context.Background(), db, task.ID) if err != nil { t.Fatalf("GetByID: %v", err) } diff --git a/internal/ingestion/task/chunk_process_test.go b/internal/ingestion/task/chunk_process_test.go index df251d8df4..1f1e409777 100644 --- a/internal/ingestion/task/chunk_process_test.go +++ b/internal/ingestion/task/chunk_process_test.go @@ -296,7 +296,8 @@ func TestProcessChunksForPipeline_PreservesContentWithWeight(t *testing.T) { func TestProcessChunkPositions_FlatFloat64(t *testing.T) { chunk := map[string]any{ - "positions": []float64{0, 100, 50, 200, 150}, + // positions is 1-indexed (parser normalized before we see it) + "positions": []float64{1, 100, 50, 200, 150}, } processChunkPositions(chunk) @@ -312,8 +313,8 @@ func TestProcessChunkPositions_FlatFloat64(t *testing.T) { func TestProcessChunkPositions_2DFloat64(t *testing.T) { chunk := map[string]any{ "positions": [][]float64{ - {0, 100, 50, 200, 150}, - {1, 200, 60, 300, 250}, + {1, 100, 50, 200, 150}, + {2, 200, 60, 300, 250}, }, } processChunkPositions(chunk) diff --git a/internal/ingestion/task/position.go b/internal/ingestion/task/position.go index cc9b1dc766..e9f53bf9ad 100644 --- a/internal/ingestion/task/position.go +++ b/internal/ingestion/task/position.go @@ -18,14 +18,14 @@ package task // AddPositions adds position fields to a chunk map. // Input positions is a flat []float64 grouped as [pn, left, right, top, bottom] -// every 5 elements. pn is 0-indexed; output is 1-indexed. +// every 5 elements. pn is ALREADY 1-indexed — the 0→1 conversion happens +// once, at the parser boundary (normalizePDFPageNumber for the DeepDoc PDF +// path; TCADP writes 1-indexed directly). This function is a passthrough: it +// must NOT add +1, otherwise the PDF path (which already normalized) would +// double-increment page numbers. // -// Mirrors Python: rag.nlp.add_positions() -// -// for pn, left, right, top, bottom in poss: -// page_num_int.append(int(pn + 1)) -// top_int.append(int(top)) -// position_int.append((int(pn + 1), int(left), int(right), int(top), int(bottom))) +// Mirrors Python: rag.nlp.add_positions() (Python adds +1 because its +// callers feed 0-indexed values; the Go pipeline normalizes earlier). func AddPositions(chunk map[string]any, positions []float64) { if len(positions) == 0 || len(positions)%5 != 0 { return @@ -36,7 +36,7 @@ func AddPositions(chunk map[string]any, positions []float64) { positionInt := make([][]int, 0, n) for i := 0; i < len(positions); i += 5 { - pn := int(positions[i]) + 1 // 0-indexed → 1-indexed + pn := int(positions[i]) // already 1-indexed left := int(positions[i+1]) right := int(positions[i+2]) top := int(positions[i+3]) diff --git a/internal/ingestion/task/position_test.go b/internal/ingestion/task/position_test.go index 6e31c6a1fc..2cb79159dc 100644 --- a/internal/ingestion/task/position_test.go +++ b/internal/ingestion/task/position_test.go @@ -7,13 +7,17 @@ import ( // ============================================================================= // AddPositions // Canonical format: [pageNum, left, right, top, bottom] × N -// Mirrors Python: rag.nlp.add_positions() +// Contract: input page numbers are ALREADY 1-indexed (the 0→1 conversion +// happens once, at the parser boundary in normalizePDFPageNumber). This +// function is a passthrough — it must NOT add +1, otherwise callers that +// already feed 1-indexed values (the PDF path after normalization) get a +// double-incremented page number. // ============================================================================= func TestAddPositions_Basic(t *testing.T) { chunk := map[string]any{} - // [pn=0, left=100, right=50, top=200, bottom=150] - positions := []float64{0, 100, 50, 200, 150} + // [pn=1 (first page, 1-indexed), left=100, right=50, top=200, bottom=150] + positions := []float64{1, 100, 50, 200, 150} AddPositions(chunk, positions) pageNum, ok := chunk["page_num_int"].([]int) @@ -36,8 +40,8 @@ func TestAddPositions_Basic(t *testing.T) { func TestAddPositions_MultiplePositions(t *testing.T) { chunk := map[string]any{} positions := []float64{ - 0, 100, 50, 200, 150, // pn=0, left=100, right=50, top=200, bottom=150 - 1, 200, 60, 300, 250, // pn=1, left=200, right=60, top=300, bottom=250 + 1, 100, 50, 200, 150, // pn=1, left=100, right=50, top=200, bottom=150 + 2, 200, 60, 300, 250, // pn=2, left=200, right=60, top=300, bottom=250 } AddPositions(chunk, positions) @@ -73,20 +77,25 @@ func TestAddPositions_EmptyPositions(t *testing.T) { func TestAddPositions_PartialPositions(t *testing.T) { chunk := map[string]any{} - positions := []float64{0, 100} // only 2 elements, not a complete position + positions := []float64{1, 100} // only 2 elements, not a complete position AddPositions(chunk, positions) if _, exists := chunk["page_num_int"]; exists { t.Error("page_num_int should not be set for partial positions") } } -func TestAddPositions_PageNumOffset(t *testing.T) { +func TestAddPositions_PassthroughNoOffset(t *testing.T) { + // page numbers are 1-indexed on entry; AddPositions must not add +1. chunk := map[string]any{} - positions := []float64{5, 100, 50, 200, 150} // pn=5 → 5+1=6 + positions := []float64{6, 100, 50, 200, 150} // pn=6, 1-indexed AddPositions(chunk, positions) pageNum := chunk["page_num_int"].([]int) if pageNum[0] != 6 { - t.Errorf("page_num_int = %d, want 6 (5+1)", pageNum[0]) + t.Errorf("page_num_int = %d, want 6 (passthrough, no +1)", pageNum[0]) + } + position := chunk["position_int"].([][]int) + if position[0][0] != 6 { + t.Errorf("position_int[0][0] = %d, want 6 (passthrough, no +1)", position[0][0]) } } diff --git a/internal/ingestion/task/real_consumer_test.go b/internal/ingestion/task/real_consumer_test.go index 24185f9049..3be0c70ac7 100644 --- a/internal/ingestion/task/real_consumer_test.go +++ b/internal/ingestion/task/real_consumer_test.go @@ -67,7 +67,7 @@ func TestRealProducerConsumer(t *testing.T) { DatasetID: "kb1", Status: common.CREATED, } - created, err := dao.NewIngestionTaskDAO().Create(ingestionTask) + created, err := dao.NewIngestionTaskDAO().Create(context.Background(), db, ingestionTask) if err != nil { t.Fatalf("Create: %v", err) } @@ -103,11 +103,11 @@ func TestRealProducerConsumer(t *testing.T) { // Mirrors Start():142-143 — UpdateStatusIfCurrent ingestionTaskDAO := dao.NewIngestionTaskDAO() - _, err = ingestionTaskDAO.UpdateStatusIfCurrent(taskMsg.TaskID, common.CREATED, common.RUNNING) + _, err = ingestionTaskDAO.UpdateStatusIfCurrent(context.Background(), db, taskMsg.TaskID, common.CREATED, common.RUNNING) if err != nil { t.Fatalf("UpdateStatusIfCurrent: %v", err) } - task, err := ingestionTaskDAO.GetByID(taskMsg.TaskID) + task, err := ingestionTaskDAO.GetByID(context.Background(), db, taskMsg.TaskID) if err != nil { t.Fatalf("GetByID: %v", err) } @@ -161,7 +161,7 @@ func TestRealProducerConsumer(t *testing.T) { t.Log("Consumer: PipelineExecutor.Execute() - OK") // Mirrors executeTask — mark as completed - if _, err := ingestionTaskDAO.UpdateStatusIfCurrent(task.ID, common.RUNNING, common.COMPLETED); err != nil { + if _, err := ingestionTaskDAO.UpdateStatusIfCurrent(context.Background(), db, task.ID, common.RUNNING, common.COMPLETED); err != nil { t.Fatalf("UpdateStatus: %v", err) } @@ -169,7 +169,7 @@ func TestRealProducerConsumer(t *testing.T) { taskHandle.Ack() // ── 6. Verify ── - final, _ := ingestionTaskDAO.GetByID(task.ID) + final, _ := ingestionTaskDAO.GetByID(context.Background(), db, task.ID) if final.Status != common.COMPLETED { t.Errorf("final status = %s, want %s", final.Status, common.COMPLETED) } diff --git a/internal/parser/parser/office_parsers_nocgo_test.go b/internal/parser/parser/office_parsers_nocgo_test.go index 19cb033221..f868264bb2 100644 --- a/internal/parser/parser/office_parsers_nocgo_test.go +++ b/internal/parser/parser/office_parsers_nocgo_test.go @@ -3,7 +3,6 @@ package parser import ( - "context" "errors" "testing" ) diff --git a/internal/parser/parser/pdf_parser_common.go b/internal/parser/parser/pdf_parser_common.go index 431bc376fe..7de46ede7b 100644 --- a/internal/parser/parser/pdf_parser_common.go +++ b/internal/parser/parser/pdf_parser_common.go @@ -599,13 +599,18 @@ func normalizePDFPositions(raw any) [][]any { return normalized } +// normalizePDFPageNumber converts a DeepDoc 0-indexed page number to the +// 1-indexed form stored in _pdf_positions / positions. It is the SINGLE +// 0→1 conversion point: DeepDoc (pdf_oxide/pdfium) emits 0-indexed pages, +// and every downstream consumer (AddPositions for ES storage, +// PositionsFromMatrix for the PDFium render path) expects 1-indexed input. +// Adding +1 unconditionally — instead of only for v<=0 — keeps all pages +// consistent; the old heuristic left page>=1 unconverted, which AddPositions +// then double-incremented and PositionsFromMatrix mis-decremented. func normalizePDFPageNumber(raw any) (int, bool) { switch v := raw.(type) { case int: - if v <= 0 { - return v + 1, true - } - return v, true + return v + 1, true case int64: return normalizePDFPageNumber(int(v)) case float64: diff --git a/internal/parser/parser/pdf_parser_common_test.go b/internal/parser/parser/pdf_parser_common_test.go index 757fce8f35..cf63a8f4c4 100644 --- a/internal/parser/parser/pdf_parser_common_test.go +++ b/internal/parser/parser/pdf_parser_common_test.go @@ -85,21 +85,57 @@ func TestPDFParseResultToJSON_NormalizesCoreFields(t *testing.T) { if got, want := res.JSON[1]["doc_type_kwd"], "image"; got != want { t.Fatalf("JSON[1].doc_type_kwd = %v, want %v", got, want) } - if got, want := res.JSON[1]["page_number"], 1; got != want { + if got, want := res.JSON[1]["page_number"], 2; got != want { t.Fatalf("JSON[1].page_number = %v, want %v", got, want) } secondPDFPositions, ok := res.JSON[1]["_pdf_positions"].([][]any) if !ok { t.Fatalf("JSON[1]._pdf_positions type = %T, want [][]any", res.JSON[1]["_pdf_positions"]) } - if len(secondPDFPositions) != 1 || secondPDFPositions[0][0] != 1 { - t.Fatalf("JSON[1]._pdf_positions = %+v, want canonical 1-based positions", secondPDFPositions) + if len(secondPDFPositions) != 1 || secondPDFPositions[0][0] != 2 { + t.Fatalf("JSON[1]._pdf_positions = %+v, want canonical 1-based positions (DeepDoc page 1 → 2)", secondPDFPositions) } if got, want := res.JSON[1]["image"], "data:image/png;base64,aGVsbG8="; got != want { t.Fatalf("JSON[1].image = %v, want %v", got, want) } } +// TestNormalizePDFPageNumber_UnconditionalIncrement pins the contract that +// DeepDoc emits 0-indexed page numbers and normalizePDFPageNumber is the +// SINGLE conversion point to 1-indexed. It must add +1 unconditionally — +// not just for v<=0 — so that downstream AddPositions (a passthrough) and +// PositionsFromMatrix (which subtracts 1 for the 0-indexed PDFium engine) +// each see a consistent 1-indexed value. +func TestNormalizePDFPageNumber_UnconditionalIncrement(t *testing.T) { + cases := []struct { + name string + in any + want int + ok bool + }{ + {"zero (first page, 0-indexed)", 0, 1, true}, + {"one (second page, 0-indexed)", 1, 2, true}, + {"five", 5, 6, true}, + {"int64", int64(2), 3, true}, + {"float64", float64(3), 4, true}, + {"page list takes last element", []any{float64(0), float64(1)}, 2, true}, + {"int list takes last element", []int{0, 1, 2}, 3, true}, + {"empty list", []any{}, 0, false}, + {"non-numeric", "x", 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := normalizePDFPageNumber(tc.in) + if ok != tc.ok { + t.Fatalf("ok = %v, want %v", ok, tc.ok) + } + if ok && got != tc.want { + t.Errorf("got = %d, want %d (unconditional +1)", got, tc.want) + } + }) + } +} + func TestPDFParseResultToJSON_PreservesPositivePageNumbers(t *testing.T) { parsed := &deepdoctype.ParseResult{ Sections: []deepdoctype.Section{ @@ -120,8 +156,10 @@ func TestPDFParseResultToJSON_PreservesPositivePageNumbers(t *testing.T) { } res := pdfParseResultToJSON("one-based.pdf", parsed) - if got, want := res.JSON[0]["page_number"], 3; got != want { - t.Fatalf("JSON[1].page_number = %v, want %v", got, want) + // DeepDoc page 3 is 0-indexed (the 4th page); normalizePDFPageNumber + // converts it to 1-indexed page 4. + if got, want := res.JSON[0]["page_number"], 4; got != want { + t.Fatalf("JSON[0].page_number = %v, want %v", got, want) } if got, want := res.JSON[0]["doc_type_kwd"], "table"; got != want { t.Fatalf("JSON[0].doc_type_kwd = %v, want %v", got, want) @@ -167,12 +205,18 @@ func TestPDFParseResultToJSON_DefaultKeepsHeaderFooterLikePython(t *testing.T) { if len(res.JSON) != 3 { t.Fatalf("JSON len = %d, want 3", len(res.JSON)) } + // Sections are now sorted by (page, top, left). Header and Footer have + // no position data (page=0, top=0), Body has top=30, so the sorted order + // is Header/Footer (tied top=0, stable) then Body (top=30). if got, want := res.JSON[0]["text"], "Header"; got != want { t.Fatalf("JSON[0].text = %v, want %v", got, want) } - if got, want := res.JSON[1]["text"], "Body"; got != want { + if got, want := res.JSON[1]["text"], "Footer"; got != want { t.Fatalf("JSON[1].text = %v, want %v", got, want) } + if got, want := res.JSON[2]["text"], "Body"; got != want { + t.Fatalf("JSON[2].text = %v, want %v", got, want) + } } func TestPDFParseResultToJSONWithOptions_FiltersHeaderFooterWhenEnabled(t *testing.T) { diff --git a/internal/parser/parser/pdf_parser_nocgo_test.go b/internal/parser/parser/pdf_parser_nocgo_test.go index 59b8d88b2f..9e0350849f 100644 --- a/internal/parser/parser/pdf_parser_nocgo_test.go +++ b/internal/parser/parser/pdf_parser_nocgo_test.go @@ -3,7 +3,6 @@ package parser import ( - "context" "errors" "testing" ) diff --git a/internal/parser/parser/pdf_parser_tcadp.go b/internal/parser/parser/pdf_parser_tcadp.go index f32584dc6d..126d9cd99e 100644 --- a/internal/parser/parser/pdf_parser_tcadp.go +++ b/internal/parser/parser/pdf_parser_tcadp.go @@ -77,6 +77,9 @@ func parsePDFWithTCADP(filename string, data []byte, parser *PDFParser) ParseRes if err != nil { return ParseResult{Err: fmt.Errorf("parser: TCADP read zip: %w", err)} } + if downloadResp.StatusCode >= 300 { + return ParseResult{Err: fmt.Errorf("parser: TCADP download HTTP %d: %s", downloadResp.StatusCode, string(zipBytes))} + } items, pageCount, err := tcadpItemsFromZip(zipBytes) if err != nil { return ParseResult{Err: err} @@ -143,6 +146,18 @@ func tcadpAnyToItems(raw any) []map[string]any { case map[string]any: text := strings.TrimSpace(stringValue(v["content"])) contentType := strings.ToLower(strings.TrimSpace(stringValue(v["type"]))) + page := extractTCADPPage(v) + emit := func(text, docType, layout string) []map[string]any { + m := map[string]any{"text": text, "doc_type_kwd": docType, "layout": layout} + if page > 0 { + // 1-indexed 5-tuple. AddPositions is a passthrough so + // the final position_int / page_num_int carry the same + // 1-indexed page number the caller passes. Mirrors + // Python presentation.py:148-149. + m["positions"] = []float64{float64(page), 0, 0, 0, 0} + } + return []map[string]any{m} + } switch contentType { case "table": if text == "" { @@ -151,28 +166,42 @@ func tcadpAnyToItems(raw any) []map[string]any { if text == "" { return nil } - return []map[string]any{{"text": text, "doc_type_kwd": "table", "layout": "table"}} + return emit(text, "table", "table") case "image": caption := strings.TrimSpace(stringValue(v["caption"])) if caption == "" { caption = "[Image]" } - return []map[string]any{{"text": caption, "doc_type_kwd": "image", "layout": "figure"}} + return emit(caption, "image", "figure") case "equation": if text == "" { return nil } - return []map[string]any{{"text": "$$" + text + "$$", "doc_type_kwd": "text", "layout": "equation"}} + return emit("$$"+text+"$$", "text", "equation") default: if text == "" { return nil } - return []map[string]any{{"text": text, "doc_type_kwd": "text", "layout": "text"}} + return emit(text, "text", "text") } } return nil } +// extractTCADPPage returns the 1-indexed page number carried by a raw TCADP +// element, using the same key set collectPDFPageNumbers walks +// (pdf_parser_remote_common.go). It returns 0 when the element has no page +// information (e.g. spreadsheet TCADP), so callers can skip position emission +// and remain parity-correct with Python (table.py sets no page either). +func extractTCADPPage(v map[string]any) int { + for _, key := range []string{"page_number", "page_num", "page_no", "page_index", "page_idx", "page"} { + if page := int(numberValue(v[key])); page > 0 { + return page + } + } + return 0 +} + func tcadpTableRowsText(raw any) string { table, ok := raw.(map[string]any) if !ok { diff --git a/internal/parser/parser/pdf_parser_tcadp_test.go b/internal/parser/parser/pdf_parser_tcadp_test.go index 838539f43d..fae95fbe97 100644 --- a/internal/parser/parser/pdf_parser_tcadp_test.go +++ b/internal/parser/parser/pdf_parser_tcadp_test.go @@ -86,6 +86,102 @@ func TestPDFParser_ParseWithResult_TCADPRequiresAPIServer(t *testing.T) { } } +// TestTCADPAnyToItems_PropagatesPageNumber verifies the fix for migration +// : per-slide page numbers present in the raw TCADP response +// must be attached to each chunk item as "positions" (a 1-indexed 5-tuple +// [page, 0, 0, 0, 0]). The shared ingestion pipeline +// (processChunkPositions -> AddPositions, a passthrough) derives +// top_int=[0], position_int=[[page,0,0,0,0]] and page_num_int=[page] from +// this field. Elements without a page number (e.g. spreadsheet TCADP) +// must NOT receive a positions field. +func TestTCADPAnyToItems_PropagatesPageNumber(t *testing.T) { + cases := []struct { + name string + raw any + wantPos []float64 // nil => positions must be absent + }{ + { + name: "text element with page_number", + raw: map[string]any{"content": "slide text", "type": "text", "page_number": 3}, + wantPos: []float64{3, 0, 0, 0, 0}, // 1-indexed + }, + { + name: "image element with page_number", + raw: map[string]any{"caption": "fig", "type": "image", "page_number": 5}, + wantPos: []float64{5, 0, 0, 0, 0}, + }, + { + name: "table element with page_number", + raw: map[string]any{"type": "table", "table_data": map[string]any{"rows": []any{[]any{"a", "b"}}}, "page_number": 2}, + wantPos: []float64{2, 0, 0, 0, 0}, + }, + { + name: "element without page number gets no positions", + raw: map[string]any{"content": "no page", "type": "text"}, + wantPos: nil, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + items := tcadpAnyToItems(tc.raw) + if len(items) == 0 { + t.Fatalf("tcadpAnyToItems returned no items") + } + item := items[0] + got, ok := item["positions"].([]float64) + if tc.wantPos == nil { + if ok { + t.Errorf("positions present = %v, want absent (element without page number)", got) + } + return + } + if !ok { + t.Fatalf("positions missing, want %v", tc.wantPos) + } + if len(got) != len(tc.wantPos) { + t.Fatalf("positions len = %d, want %d (%v)", len(got), len(tc.wantPos), tc.wantPos) + } + for i := range tc.wantPos { + if got[i] != tc.wantPos[i] { + t.Errorf("positions[%d] = %v, want %v (full: %v)", i, got[i], tc.wantPos[i], got) + } + } + }) + } +} + +// TestTCADPAnyToItems_NestedArrayKeepsEachPage verifies that a nested +// array of TCADP elements produces one item per element, each carrying +// its own page-derived positions — not just the first element. +func TestTCADPAnyToItems_NestedArrayKeepsEachPage(t *testing.T) { + raw := []any{ + map[string]any{"content": "p1", "type": "text", "page_number": 1}, + map[string]any{"content": "p7", "type": "text", "page_number": 7}, + } + items := tcadpAnyToItems(raw) + if len(items) != 2 { + t.Fatalf("items = %d, want 2", len(items)) + } + want := [][]float64{{1, 0, 0, 0, 0}, {7, 0, 0, 0, 0}} // page 1 -> [1,...], page 7 -> [7,...] (1-indexed) + for i, w := range want { + got, ok := items[i]["positions"].([]float64) + if !ok { + t.Errorf("items[%d].positions missing, want %v", i, w) + continue + } + if len(got) != len(w) { + t.Errorf("items[%d].positions = %v, want len %d", i, got, len(w)) + continue + } + for j := range w { + if got[j] != w[j] { + t.Errorf("items[%d].positions[%d] = %v, want %v", i, j, got[j], w[j]) + } + } + } +} + func tcadpZipFixture(t *testing.T) []byte { t.Helper() var buf bytes.Buffer diff --git a/internal/parser/parser/pdf_postprocess.go b/internal/parser/parser/pdf_postprocess.go index e116a924e8..969324d21b 100644 --- a/internal/parser/parser/pdf_postprocess.go +++ b/internal/parser/parser/pdf_postprocess.go @@ -26,6 +26,7 @@ func applyPDFPostProcess(result *deepdoctype.ParseResult, opts pdfPostProcessOpt if result == nil { return } + sortSectionsByPosition(result) if opts.enableMultiColumn && opts.pageWidth > 0 { reorderPDFMultiColumn(result, opts.pageWidth, opts.zoom) } @@ -83,6 +84,28 @@ func assignPDFDocTypeKeywords(result *deepdoctype.ParseResult, flatten bool) { } } +// sortSectionsByPosition reorders sections into reading order: page number, +// then vertical position (top), then horizontal position (left). The DeepDoc +// layout engine does not guarantee reading order in its output, so this sort +// ensures the downstream chunker receives items in document order regardless +// of the engine's internal extraction sequence. +func sortSectionsByPosition(result *deepdoctype.ParseResult) { + if result == nil || len(result.Sections) < 2 { + return + } + sort.SliceStable(result.Sections, func(i, j int) bool { + pi, pj := firstSectionPage(result.Sections[i]), firstSectionPage(result.Sections[j]) + if pi != pj { + return pi < pj + } + ti, tj := firstSectionTop(result.Sections[i]), firstSectionTop(result.Sections[j]) + if math.Abs(ti-tj) > 1e-6 { + return ti < tj + } + return firstSectionLeft(result.Sections[i]) < firstSectionLeft(result.Sections[j]) + }) +} + // applyRemoveTOC mirrors Python parser.py:663-681 three-way dispatch: // - No outlines → pattern-based remove_toc on all sections // - First outline on page 1 → outline-based remove_toc_pdf diff --git a/internal/parser/parser/ppt_parser.go b/internal/parser/parser/ppt_parser.go index b57322225e..d7e4b12a42 100644 --- a/internal/parser/parser/ppt_parser.go +++ b/internal/parser/parser/ppt_parser.go @@ -20,21 +20,32 @@ package parser import "context" -type PPTParser struct{} +// PPTParser delegates to PPTXParser which handles both OLE binary +// (.ppt) and OOXML (.pptx) containers uniformly +type PPTParser struct { + pptx PPTXParser +} func NewPPTParser() *PPTParser { - return &PPTParser{} + return &PPTParser{pptx: PPTXParser{format: "ppt"}} } func (p *PPTParser) String() string { return "PPTParser" } +// ConfigureFromSetup forwards the slides-family setup map to the +// underlying PPTXParser so both .ppt and .pptx containers share +// the same TCADP configuration +func (p *PPTParser) ConfigureFromSetup(setup map[string]any) { + p.pptx.ConfigureFromSetup(setup) +} + // ParseWithResult delegates to PPTXParser's structured output // for the legacy PPT format using the "ppt" container format // hint (OLE binary). The two file families differ only in the // binary container; the python parser.py:slides branch treats // them uniformly. func (p *PPTParser) ParseWithResult(ctx context.Context, filename string, data []byte) ParseResult { - return (&PPTXParser{format: "ppt"}).ParseWithResult(ctx, filename, data) + return p.pptx.ParseWithResult(ctx, filename, data) } diff --git a/internal/parser/parser/ppt_parser_cgo_test.go b/internal/parser/parser/ppt_parser_cgo_test.go index 2cb7928604..81591d0142 100644 --- a/internal/parser/parser/ppt_parser_cgo_test.go +++ b/internal/parser/parser/ppt_parser_cgo_test.go @@ -3,6 +3,12 @@ package parser import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" "testing" officeOxide "github.com/yfedoseev/office_oxide/go" @@ -63,6 +69,155 @@ func TestPPTParser_ParseWithResult_CGO(t *testing.T) { } } +// TestPPTParser_TCADPFileType verifies that when a .ppt file is routed +// through PPTParser with parse_method="tcadp", the underlying +// PPTXParser must pass "PPT" as the file_type to TCADP, not "PPTX". +// This test verifies format propagation from PPTParser through +// ConfigureFromSetup to the embedded PPTXParser. +func TestPPTParser_TCADPFileType(t *testing.T) { + // PPTParser delegates to PPTXParser{format:"ppt"}. + p := NewPPTParser() + setup := map[string]any{ + "parse_method": "tcadp", + "output_format": "json", + } + p.ConfigureFromSetup(setup) + + // Verify the embedded PPTXParser received the config and keeps + // format="ppt" (which maps to "PPT" in TCADP fileType). + if p.pptx.format != "ppt" { + t.Errorf("PPTParser.pptx.format = %q, want %q", p.pptx.format, "ppt") + } + if p.pptx.ParseMethod != "tcadp" { + t.Errorf("PPTParser.pptx.ParseMethod = %q, want %q", p.pptx.ParseMethod, "tcadp") + } + if p.pptx.OutputFormat != "json" { + t.Errorf("PPTParser.pptx.OutputFormat = %q, want %q", p.pptx.OutputFormat, "json") + } +} + +// TestPPTXParser_TCADPFileType verifies that a PPTXParser +// with format="pptx" must derive fileType "PPTX" for TCADP calls. +func TestPPTXParser_TCADPFileType(t *testing.T) { + p := NewPPTXParser() + if p.format != "pptx" { + t.Fatalf("NewPPTXParser().format = %q, want pptx", p.format) + } + setup := map[string]any{ + "parse_method": "tcadp", + "output_format": "json", + } + p.ConfigureFromSetup(setup) + if p.ParseMethod != "tcadp" { + t.Errorf("ParseMethod = %q, want tcadp", p.ParseMethod) + } +} + +// TestPPTParser_TCADPIntegration drives the end-to-end TCADP path for a +// .ppt file: PPTParser must POST file_type="PPT" to the reconstruct +// endpoint and then process the returned ZIP artifact into JSON items. +func TestPPTParser_TCADPIntegration(t *testing.T) { + testPresentationTCADPIntegration(t, NewPPTParser(), "PPT", "presentation.ppt") +} + +// TestPPTXParser_TCADPIntegration drives the end-to-end TCADP path for a +// .pptx file: PPTXParser must POST file_type="PPTX" to the reconstruct +// endpoint and then process the returned ZIP artifact into JSON items. +func TestPPTXParser_TCADPIntegration(t *testing.T) { + testPresentationTCADPIntegration(t, NewPPTXParser(), "PPTX", "presentation.pptx") +} + +// tcadpPresentationParser is the shared contract of PPTParser and +// PPTXParser used by the TCADP integration helper below. +type tcadpPresentationParser interface { + ConfigureFromSetup(setup map[string]any) + ParseWithResult(ctx context.Context, filename string, data []byte) ParseResult +} + +func testPresentationTCADPIntegration(t *testing.T, p tcadpPresentationParser, wantFileType, filename string) { + t.Helper() + zipPayload := tcadpZipFixture(t) + var gotFileType string + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/reconstruct_document": + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read reconstruct request: %v", err) + } else { + var req struct { + FileType string `json:"file_type"` + } + if err := json.Unmarshal(body, &req); err != nil { + t.Errorf("decode reconstruct request: %v", err) + } + gotFileType = req.FileType + } + _, _ = w.Write([]byte(`{"DocumentRecognizeResultUrl":"` + server.URL + `/download.zip"}`)) + case "/download.zip": + w.Header().Set("Content-Type", "application/zip") + _, _ = w.Write(zipPayload) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + p.ConfigureFromSetup(map[string]any{ + "parse_method": "tcadp", + "output_format": "json", + "tcadp_apiserver": server.URL, + }) + ctx := t.Context() + res := p.ParseWithResult(ctx, filename, []byte("dummy-presentation-bytes")) + if res.Err != nil { + t.Fatalf("ParseWithResult: %v", res.Err) + } + if gotFileType != wantFileType { + t.Errorf("TCADP request file_type = %q, want %q", gotFileType, wantFileType) + } + if res.OutputFormat != "json" { + t.Errorf("OutputFormat = %q, want json", res.OutputFormat) + } + if len(res.JSON) == 0 { + t.Fatalf("JSON items = 0, want the fixture's parsed content") + } +} + +// TestPPTXParser_TCADPDownloadHTTPError verifies that a non-2xx response +// from the TCADP download endpoint is surfaced as an explicit error rather +// than parsed as a (malformed) ZIP artifact. +func TestPPTXParser_TCADPDownloadHTTPError(t *testing.T) { + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/reconstruct_document": + _, _ = w.Write([]byte(`{"DocumentRecognizeResultUrl":"` + server.URL + `/download.zip"}`)) + case "/download.zip": + http.Error(w, "upstream failure", http.StatusInternalServerError) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + p := NewPPTXParser() + p.ConfigureFromSetup(map[string]any{ + "parse_method": "tcadp", + "output_format": "json", + "tcadp_apiserver": server.URL, + }) + ctx := t.Context() + res := p.ParseWithResult(ctx, "a.pptx", []byte("dummy")) + if res.Err == nil { + t.Fatal("ParseWithResult: want error for non-2xx download, got nil") + } + if !strings.Contains(res.Err.Error(), "download HTTP") { + t.Errorf("err = %v, want error mentioning 'download HTTP'", res.Err) + } +} + // buildPPTX creates a minimal valid PPTX document with one slide // containing the given text, using office_oxide's PptxWriter. func buildPPTX(t *testing.T, text string) []byte { diff --git a/internal/parser/parser/pptx_parser.go b/internal/parser/parser/pptx_parser.go index 3386d48f5d..a8d50ab005 100644 --- a/internal/parser/parser/pptx_parser.go +++ b/internal/parser/parser/pptx_parser.go @@ -33,6 +33,14 @@ import ( // OLE format. type PPTXParser struct { format string + + // TCADP cloud-parsing configuration + ParseMethod string + TCADPAPIServer string + TCADPAPIKey string + TCADPTableResultType string + TCADPMarkdownImageResponseType string + OutputFormat string } func NewPPTXParser() *PPTXParser { @@ -43,10 +51,62 @@ func (p *PPTXParser) String() string { return "PPTXParser" } +// ConfigureFromSetup reads the slides-family setup map. Mirrors the +// XLSXParser ConfigureFromSetup pattern +func (p *PPTXParser) ConfigureFromSetup(setup map[string]any) { + if p == nil || setup == nil { + return + } + if v, ok := setup["parse_method"].(string); ok { + p.ParseMethod = v + } + if v, ok := setup["tcadp_apiserver"].(string); ok { + p.TCADPAPIServer = v + } + if v, ok := setup["tcadp_api_key"].(string); ok { + p.TCADPAPIKey = v + } + if v, ok := setup["table_result_type"].(string); ok { + p.TCADPTableResultType = v + } + if v, ok := setup["markdown_image_response_type"].(string); ok { + p.TCADPMarkdownImageResponseType = v + } + if v, ok := setup["output_format"].(string); ok { + p.OutputFormat = v + } + if p.OutputFormat == "" { + p.OutputFormat = "json" + } +} + // ParseWithResult emits one JSON item per slide with the slide's // plain text. Mirrors the python parser.py:slides branch which // forces output_format="json" for the slide family. func (p *PPTXParser) ParseWithResult(ctx context.Context, filename string, data []byte) ParseResult { + // p == nil guard: the struct is embedded by value in PPTParser and + // always created via NewPPTXParser or the "ppt"-format constructor in + // PPTParser, so this branch is unreachable from normal call paths. + // Kept as defensive guard — a nil dereference here would obscure the + // root cause behind a nil-pointer panic. + if p == nil { + return ParseResult{Err: fmt.Errorf("PPTXParser is nil")} + } + method := strings.ToLower(strings.TrimSpace(p.ParseMethod)) + switch method { + case "tcadp": + return parsePresentationWithTCADP(ctx, + filename, data, strings.ToUpper(p.format), + p.TCADPAPIServer, p.TCADPAPIKey, + p.TCADPTableResultType, p.TCADPMarkdownImageResponseType, + p.OutputFormat, + ) + case "", "deepdoc": + // Continue with the local office_oxide parser. + default: + // PDF-specific methods like "paddleocr" / "mineru" are + // meaningless for PPTX; treat as default path. + } doc, err := officeOxide.OpenFromBytes(data, p.format) if err != nil { return ParseResult{Err: fmt.Errorf("presentation open: %w", err)} diff --git a/internal/parser/parser/pptx_tcadp.go b/internal/parser/parser/pptx_tcadp.go new file mode 100644 index 0000000000..47620ffa72 --- /dev/null +++ b/internal/parser/parser/pptx_tcadp.go @@ -0,0 +1,95 @@ +package parser + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + + models "ragflow/internal/entity/models" +) + +// parsePresentationWithTCADP sends binary presentation (PPTX/PPT) data +// to the TCADP cloud reconstruction service and returns the structured +// parse result. Mirrors the spreadsheet-family parseSpreadsheetWithTCADP +// in xls_tcadp.go +func parsePresentationWithTCADP(ctx context.Context, filename string, data []byte, fileType string, + tcadpAPIServer, tcadpAPIKey, tableResultType, markdownImageResponseType string, + outputFormat string, +) ParseResult { + if len(data) == 0 { + return emptyPDFResult(filename) + } + baseURL := strings.TrimSpace(tcadpAPIServer) + if baseURL == "" { + baseURL = strings.TrimSpace(os.Getenv("TCADP_APISERVER")) + } + if baseURL == "" { + return ParseResult{Err: fmt.Errorf("parser: TCADP requires tcadp_apiserver or TCADP_APISERVER")} + } + apiKey := strings.TrimSpace(tcadpAPIKey) + if apiKey == "" { + apiKey = strings.TrimSpace(os.Getenv("TCADP_API_KEY")) + } + requestBody := map[string]any{ + "file_type": fileType, + "file_base64": base64.StdEncoding.EncodeToString(data), + "file_start_page_number": 1, + "file_end_page_number": 1000, + "config": map[string]any{ + "TableResultType": tableResultType, + "MarkdownImageResponseType": markdownImageResponseType, + }, + } + resp, err := models.PostJSONRequest(ctx, models.NewDriverHTTPClient(), + strings.TrimRight(baseURL, "/")+"/reconstruct_document", bearer(apiKey), requestBody) + if err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP submit: %w", err)} + } + defer resp.Body.Close() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP read submit: %w", err)} + } + if resp.StatusCode >= 300 { + return ParseResult{Err: fmt.Errorf("parser: TCADP HTTP %d: %s", resp.StatusCode, string(raw))} + } + var payload struct { + DocumentRecognizeResultURL string `json:"DocumentRecognizeResultUrl"` + } + if err := json.Unmarshal(raw, &payload); err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP decode submit: %w", err)} + } + if payload.DocumentRecognizeResultURL == "" { + return ParseResult{Err: fmt.Errorf("parser: TCADP returned no DocumentRecognizeResultUrl")} + } + downloadReq, err := http.NewRequestWithContext(ctx, http.MethodGet, + payload.DocumentRecognizeResultURL, nil) + if err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP download request: %w", err)} + } + if auth := bearer(apiKey); auth != "" { + downloadReq.Header.Set("Authorization", auth) + } + downloadResp, err := models.NewDriverHTTPClient().Do(downloadReq) + if err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP download: %w", err)} + } + defer downloadResp.Body.Close() + zipBytes, err := io.ReadAll(downloadResp.Body) + if err != nil { + return ParseResult{Err: fmt.Errorf("parser: TCADP read zip: %w", err)} + } + if downloadResp.StatusCode >= 300 { + return ParseResult{Err: fmt.Errorf("parser: TCADP download HTTP %d: %s", downloadResp.StatusCode, string(zipBytes))} + } + items, pageCount, err := tcadpItemsFromZip(zipBytes) + if err != nil { + return ParseResult{Err: err} + } + return pdfItemsToResult(filename, items, outputFormat, pageCount) +} diff --git a/internal/parser/parser/xls_tcadp.go b/internal/parser/parser/xls_tcadp.go index 43321dfc28..e0c31e0972 100644 --- a/internal/parser/parser/xls_tcadp.go +++ b/internal/parser/parser/xls_tcadp.go @@ -75,6 +75,9 @@ func parseSpreadsheetWithTCADP(filename string, data []byte, fileType string, tc if err != nil { return ParseResult{Err: fmt.Errorf("parser: TCADP read zip: %w", err)} } + if downloadResp.StatusCode >= 300 { + return ParseResult{Err: fmt.Errorf("parser: TCADP download HTTP %d: %s", downloadResp.StatusCode, string(zipBytes))} + } items, pageCount, err := tcadpItemsFromZip(zipBytes) if err != nil { return ParseResult{Err: err}