diff --git a/.doc/ops/short-term/2026-09-21-mineru-doc-parsing.md b/.doc/ops/short-term/2026-09-21-mineru-doc-parsing.md index 14a6daf2..6a95677f 100644 --- a/.doc/ops/short-term/2026-09-21-mineru-doc-parsing.md +++ b/.doc/ops/short-term/2026-09-21-mineru-doc-parsing.md @@ -19,7 +19,7 @@ write_scope: - docker-compose.*.yml - .doc/ops/ - .doc/test-reports/ -head_sha: 7f150831bba4482598689107c5bc6b07bf9e98df +head_sha: 5c3261c6d8a773fe31f331fc48ab0ccd64e091ac deployment: passed cleanup: pending mutex: storage+bizdev+kb diff --git a/internal/bizdev/materials.go b/internal/bizdev/materials.go index 9be62527..3283c54f 100644 --- a/internal/bizdev/materials.go +++ b/internal/bizdev/materials.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "errors" "fmt" + "log/slog" "path/filepath" "strings" @@ -50,6 +51,8 @@ func (s *Service) ResolveMaterials(ctx context.Context, req TaskRunRequest) ([]s for _, ref := range refs { data, loadedRef, err := s.cas.ReadRefBytes(ctx, ref.ID) if err != nil { + slog.Warn("resolve_materials skip unreadable asset", + "asset_id", ref.ID, "filename", ref.FileName, "error", err) continue } items, meta, err := s.materialToItems(ctx, loadedRef.FileName, loadedRef.BlobSHA256, data, loadedRef.ID, loadedRef.UserID, req.ProjectID, ragQuery) diff --git a/internal/materialindex/pipeline.go b/internal/materialindex/pipeline.go index 8ed71555..3f06639c 100644 --- a/internal/materialindex/pipeline.go +++ b/internal/materialindex/pipeline.go @@ -272,9 +272,10 @@ func (p *Pipeline) EnsureParsed(ctx context.Context, fileName string, blobSHA st } // EnqueueAsync starts background parse without blocking the caller. -func (p *Pipeline) EnqueueAsync(fileName string, blobSHA string, data []byte, projectID int64) { +// Returns true when a parse job was actually enqueued. +func (p *Pipeline) EnqueueAsync(fileName string, blobSHA string, data []byte, projectID int64) bool { if p == nil || !p.Enabled() || !p.HasStore() || !mineru.SupportsExt(fileName) || blobSHA == "" || len(data) == 0 { - return + return false } go func() { ctx, cancel := context.WithTimeout(context.Background(), p.waitLimit+2*time.Minute) @@ -289,6 +290,7 @@ func (p *Pipeline) EnqueueAsync(fileName string, blobSHA string, data []byte, pr } } }() + return true } func (p *Pipeline) triggerParse(ctx context.Context, fileName, blobSHA string, data []byte) error { diff --git a/internal/mineru/agent.go b/internal/mineru/agent.go index 0e0d1818..ac53b7ee 100644 --- a/internal/mineru/agent.go +++ b/internal/mineru/agent.go @@ -8,16 +8,115 @@ import ( "io" "net/http" "path/filepath" + "regexp" "strings" "time" ) +var chunkAnchorRe = regexp.MustCompile(`(?m)^\s*$`) + +// bumpPageAnchors rewrites chunk-local `` anchors to global +// numbering by adding offset (start-1) to each page number. +func BumpPageAnchors(md string, offset int) string { + if offset <= 0 { + return md + } + return chunkAnchorRe.ReplaceAllStringFunc(md, func(s string) string { + m := chunkAnchorRe.FindStringSubmatch(s) + if len(m) == 2 { + var n int + if _, err := fmt.Sscanf(m[1], "%d", &n); err == nil { + return fmt.Sprintf("", n+offset) + } + } + return s + }) +} + +// agentMaxPagesPerRequest is the documented Agent lightweight API page cap. +const agentMaxPagesPerRequest = 20 + +// agentMaxChunks bounds chunked parsing (20 pages × 40 = 800 pages max). +const agentMaxChunks = 40 + // parseAgentFile uses MinerU Agent lightweight API (no token; IP rate-limited). -// Limits: ≤10MB, ≤20 pages. Output is markdown only (CDN URL). +// Limits: ≤10MB, ≤20 pages per request. Output is markdown only (CDN URL). +// When the whole file exceeds the page cap (-30003), fall back to page_range +// chunking so documents without a valid precision token still parse. func (c *Client) parseAgentFile(ctx context.Context, fileName string, data []byte) (*ParseResult, error) { if len(data) > 10<<20 { return nil, &PermanentError{Code: "-30001", Msg: "file exceeds agent lightweight 10MB limit"} } + md, pages, taskID, err := c.parseAgentRange(ctx, fileName, data, "") + if err == nil { + if pages == 0 { + // Agent markdown lacks page markers; keep as single logical page. + md = "\n" + md + pages = 1 + } + return &ParseResult{BatchID: taskID, Markdown: md, TotalPages: pages, EngineMode: "agent"}, nil + } + if !IsPageLimitError(err) { + return nil, err + } + + // Whole file exceeded the 20-page cap: parse in page_range windows. + var sb strings.Builder + totalPages := 0 + firstTaskID := "" + for i := 0; i < agentMaxChunks; i++ { + start := 1 + i*agentMaxPagesPerRequest + end := start + agentMaxPagesPerRequest - 1 + rng := fmt.Sprintf("%d-%d", start, end) + md, _, tid, cerr := c.parseAgentRange(ctx, fileName, data, rng) + if cerr != nil { + if IsPageLimitError(cerr) || i > 0 { + // Beyond document end (after at least one success) or re-limit: stop. + break + } + return nil, cerr + } + if firstTaskID == "" { + firstTaskID = tid + } + if countPageAnchors(md) > 0 { + // Chunk-local anchors: renumber to global pages. + sb.WriteString("\n") + sb.WriteString(BumpPageAnchors(md, start-1)) + sb.WriteString("\n") + } else { + sb.WriteString(fmt.Sprintf("\n\n", start)) + sb.WriteString(md) + sb.WriteString("\n") + } + totalPages = end + } + out := strings.TrimSpace(sb.String()) + if out == "" { + return nil, fmt.Errorf("mineru agent: chunked parse produced no markdown") + } + return &ParseResult{ + BatchID: firstTaskID, + Markdown: out, + TotalPages: totalPages, + EngineMode: "agent_chunked", + }, nil +} + +// isPageLimitError reports the Agent API "-30003 file page count exceeds +// API limit" condition (both submit-time and poll-time shapes). +func IsPageLimitError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "-30003") || + strings.Contains(msg, "page count exceeds") +} + +// parseAgentRange parses fileName/data via Agent API with optional page_range +// ("" = whole file). Returns markdown, anchor count and task id. +func (c *Client) parseAgentRange(ctx context.Context, fileName string, data []byte, pageRange string) (md string, pages int, taskID string, err error) { name := filepath.Base(fileName) if name == "" || name == "." { name = "upload.bin" @@ -30,53 +129,47 @@ func (c *Client) parseAgentFile(ctx context.Context, fileName string, data []byt "enable_formula": true, "is_ocr": false, } + if pageRange != "" { + body["page_range"] = pageRange + } raw, err := c.postJSONNoAuth(ctx, c.Endpoint+"/api/v1/agent/parse/file", body) if err != nil { - return nil, err + return "", 0, "", err } var env apiEnvelope if err := json.Unmarshal(raw, &env); err != nil { - return nil, fmt.Errorf("mineru agent: decode apply: %w", err) + return "", 0, "", fmt.Errorf("mineru agent: decode apply: %w", err) } if !codeOK(env.Code) { - return nil, classifyAPIError(env.Code, env.Msg) + return "", 0, "", classifyAPIError(env.Code, env.Msg) } var dataOut struct { TaskID string `json:"task_id"` FileURL string `json:"file_url"` } if err := json.Unmarshal(env.Data, &dataOut); err != nil { - return nil, fmt.Errorf("mineru agent: decode apply data: %w", err) + return "", 0, "", fmt.Errorf("mineru agent: decode apply data: %w", err) } if dataOut.TaskID == "" || dataOut.FileURL == "" { - return nil, fmt.Errorf("mineru agent: empty task_id or file_url") + return "", 0, "", fmt.Errorf("mineru agent: empty task_id or file_url") } if err := c.putFile(ctx, dataOut.FileURL, data); err != nil { - return nil, err + return "", 0, dataOut.TaskID, err } mdURL, err := c.pollAgent(ctx, dataOut.TaskID) if err != nil { - return nil, err + return "", 0, dataOut.TaskID, err } mdBytes, err := c.download(ctx, mdURL) if err != nil { - return nil, err + return "", 0, dataOut.TaskID, err } - md := EnsureUTF8(strings.TrimSpace(string(mdBytes))) + md = EnsureUTF8(strings.TrimSpace(string(mdBytes))) if md == "" { - return nil, fmt.Errorf("mineru agent: empty markdown") + return "", 0, dataOut.TaskID, fmt.Errorf("mineru agent: empty markdown") } - pages := countPageAnchors(md) - if pages == 0 { - // Agent markdown often lacks page markers; keep as single logical page. - md = "\n" + md - pages = 1 - } - return &ParseResult{ - BatchID: dataOut.TaskID, - Markdown: md, - TotalPages: pages, - }, nil + pages = countPageAnchors(md) + return md, pages, dataOut.TaskID, nil } func (c *Client) pollAgent(ctx context.Context, taskID string) (markdownURL string, err error) { diff --git a/internal/mineru/agent_test.go b/internal/mineru/agent_test.go new file mode 100644 index 00000000..74f5cfa2 --- /dev/null +++ b/internal/mineru/agent_test.go @@ -0,0 +1,29 @@ +package mineru_test + +import ( + "strings" + "testing" + + "chat-agent-system/internal/mineru" +) + +func TestBumpPageAnchors(t *testing.T) { + in := "\nA\n\nB\n" + out := mineru.BumpPageAnchors(in, 20) + if !strings.Contains(out, "") || !strings.Contains(out, "") { + t.Fatalf("out=%s", out) + } +} + +func TestIsPageLimitError(t *testing.T) { + if !mineru.IsPageLimitError(errString("-30003 file page count exceeds API limit (20 pages)")) { + t.Fatal("should detect page limit") + } + if mineru.IsPageLimitError(errString("A0211 token expired")) { + t.Fatal("must not treat token error as page limit") + } +} + +type errString string + +func (e errString) Error() string { return string(e) } diff --git a/internal/mineru/client.go b/internal/mineru/client.go index dc72e3c1..171250fb 100644 --- a/internal/mineru/client.go +++ b/internal/mineru/client.go @@ -123,7 +123,9 @@ func (c *Client) ParseFile(ctx context.Context, fileName string, data []byte) (* if err != nil { return nil, err } - res.EngineMode = "agent" + if res.EngineMode == "" { + res.EngineMode = "agent" + } return res, nil }