From 63f94ff23bb07d79e746eba842c0b74dcff87be1 Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 21 Sep 2026 09:18:20 +0000 Subject: [PATCH] feat(server): expose dedup hint, parse error and reparse endpoint Co-authored-by: Cursor --- internal/materialindex/pipeline.go | 52 ++++++++++- internal/server/asset_handlers.go | 143 ++++++++++++++++++++++++----- internal/server/server.go | 1 + 3 files changed, 167 insertions(+), 29 deletions(-) diff --git a/internal/materialindex/pipeline.go b/internal/materialindex/pipeline.go index bc9e1943..8cf895a9 100644 --- a/internal/materialindex/pipeline.go +++ b/internal/materialindex/pipeline.go @@ -150,14 +150,30 @@ func (p *Pipeline) GetParseState(ctx context.Context, blobSHA string) string { return rec.State } +// ParseInfo is the user-facing parse status for one blob. +type ParseInfo struct { + State string `json:"state"` + ErrorMessage string `json:"error_message,omitempty"` +} + // GetParseStates batch-loads parse states keyed by blob sha. func (p *Pipeline) GetParseStates(ctx context.Context, blobSHAs []string) map[string]string { - out := make(map[string]string, len(blobSHAs)) + infos := p.GetParseInfos(ctx, blobSHAs) + out := make(map[string]string, len(infos)) + for sha, info := range infos { + out[sha] = info.State + } + return out +} + +// GetParseInfos batch-loads parse state + error message keyed by blob sha. +func (p *Pipeline) GetParseInfos(ctx context.Context, blobSHAs []string) map[string]ParseInfo { + out := make(map[string]ParseInfo, len(blobSHAs)) if p == nil || p.pool == nil || len(blobSHAs) == 0 { return out } rows, err := p.pool.Query(ctx, ` - SELECT blob_sha256, state FROM asset_parses + SELECT blob_sha256, state, COALESCE(error_message, '') FROM asset_parses WHERE engine = $1 AND blob_sha256 = ANY($2) `, engineMinerU, blobSHAs) if err != nil { @@ -165,14 +181,40 @@ func (p *Pipeline) GetParseStates(ctx context.Context, blobSHAs []string) map[st } defer rows.Close() for rows.Next() { - var sha, state string - if rows.Scan(&sha, &state) == nil { - out[sha] = state + var sha, state, errMsg string + if rows.Scan(&sha, &state, &errMsg) == nil { + out[sha] = ParseInfo{State: state, ErrorMessage: errMsg} } } return out } +// ResetForReparse rolls a failed/exhausted parse back to pending so the next +// EnqueueAsync/EnsureParsed cycle retries MinerU. Returns false when the blob +// is unknown or still in progress (pending/running/done). +func (p *Pipeline) ResetForReparse(ctx context.Context, blobSHA string) bool { + if p == nil || p.pool == nil || blobSHA == "" { + return false + } + tag, err := p.pool.Exec(ctx, ` + UPDATE asset_parses SET state = $2, attempts = 0, error_message = '', updated_at = now() + WHERE blob_sha256 = $1 AND engine = $3 AND state NOT IN ($4, $5) + `, blobSHA, statePending, engineMinerU, stateRunning, stateDone) + if err != nil { + return false + } + if tag.RowsAffected() > 0 { + return true + } + // No row updated: either the record is done/running (must not reset) or it + // never existed. When it never existed, allow a fresh parse to start. + rec, err := p.getRecord(ctx, blobSHA) + if err != nil || rec != nil { + return false + } + return true +} + // EnsureParsed waits for or triggers MinerU parse. On miss/disable/fail returns // fallback text from squad.ExtractTextWithAnchors (may be empty). func (p *Pipeline) EnsureParsed(ctx context.Context, fileName string, blobSHA string, data []byte) (markdown string, fromMinerU bool, err error) { diff --git a/internal/server/asset_handlers.go b/internal/server/asset_handlers.go index 18d68155..106acb93 100644 --- a/internal/server/asset_handlers.go +++ b/internal/server/asset_handlers.go @@ -8,17 +8,21 @@ import ( "strings" "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/materialindex" + "chat-agent-system/internal/mineru" "chat-agent-system/internal/upload" ) type projectAssetResponse struct { - AssetID int64 `json:"asset_id"` - Filename string `json:"filename"` - SHA256 string `json:"sha256"` - MimeType string `json:"mime_type"` - SizeBytes int64 `json:"size_bytes"` - ProjectID int64 `json:"project_id"` - ParseState string `json:"parse_state,omitempty"` + AssetID int64 `json:"asset_id"` + Filename string `json:"filename"` + SHA256 string `json:"sha256"` + MimeType string `json:"mime_type"` + SizeBytes int64 `json:"size_bytes"` + ProjectID int64 `json:"project_id"` + ParseState string `json:"parse_state,omitempty"` + ParseError string `json:"parse_error,omitempty"` + Deduplicated bool `json:"deduplicated,omitempty"` } func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request) { @@ -80,21 +84,79 @@ func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request } parseState := "" - if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil { - matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID) - if matIdx.Enabled() { - parseState = "pending" + if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() { + if mineru.SupportsExt(ref.FileName) { + matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID) + // Report the real state: same-content uploads hit an existing parse + // (possibly already done) instead of faking "pending". + if state := matIdx.GetParseState(r.Context(), ref.BlobSHA256); state != "" { + parseState = state + } else { + parseState = "pending" + } + } else { + parseState = "skipped" } } writeJSONResponse(w, http.StatusOK, projectAssetResponse{ - AssetID: ref.ID, - Filename: ref.FileName, - SHA256: ref.BlobSHA256, - MimeType: mimeType, - SizeBytes: int64(len(data)), - ProjectID: projectID, - ParseState: parseState, + AssetID: ref.ID, + Filename: ref.FileName, + SHA256: ref.BlobSHA256, + MimeType: mimeType, + SizeBytes: int64(len(data)), + ProjectID: projectID, + ParseState: parseState, + Deduplicated: ref.IsDeduplicated, + }) +} + +// handleReparseProjectAsset resets a failed material parse and re-enqueues MinerU. +func (s *Server) handleReparseProjectAsset(w http.ResponseWriter, r *http.Request) { + projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || projectID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, "invalid project id") + return + } + assetID, err := strconv.ParseInt(r.PathValue("asset_id"), 10, 64) + if err != nil || assetID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, "invalid asset id") + return + } + if _, ok := s.requireProjectAccess(w, r, projectID); !ok { + return + } + cas := s.cas + if cas == nil { + writeSafeError(w, r, http.StatusServiceUnavailable, "storage unavailable") + return + } + data, ref, err := cas.ReadRefBytes(r.Context(), assetID) + if err != nil || ref == nil { + writeSafeError(w, r, http.StatusNotFound, "asset not found") + return + } + if ref.ProjectID != projectID { + writeSafeError(w, r, http.StatusNotFound, "asset not found") + return + } + matIdx := bizdev.Snapshot(s.svc).MaterialIndex + if matIdx == nil || !matIdx.Enabled() { + writeSafeError(w, r, http.StatusServiceUnavailable, "parse engine unavailable") + return + } + if !mineru.SupportsExt(ref.FileName) { + writeSafeError(w, r, http.StatusBadRequest, "file type not supported by parser") + return + } + if !matIdx.ResetForReparse(r.Context(), ref.BlobSHA256) { + writeSafeError(w, r, http.StatusConflict, "parse is in progress or already complete") + return + } + matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID) + writeJSONResponse(w, http.StatusOK, map[string]any{ + "asset_id": ref.ID, + "parse_state": "pending", }) } @@ -124,20 +186,53 @@ func (s *Server) handleListProjectAssets(w http.ResponseWriter, r *http.Request) shas = append(shas, ref.BlobSHA256) } } - states := map[string]string{} + infos := map[string]materialindex.ParseInfo{} if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil { - states = matIdx.GetParseStates(r.Context(), shas) + infos = matIdx.GetParseInfos(r.Context(), shas) } - out := make([]projectAssetResponse, 0, len(refs)) + seenSHA := make(map[string]bool, len(refs)) + shaCount := make(map[string]int, len(refs)) for _, ref := range refs { - out = append(out, projectAssetResponse{ + if ref != nil && ref.BlobSHA256 != "" { + shaCount[ref.BlobSHA256]++ + } + } + out := make([]projectAssetResponse, 0, len(refs)) + firstRefOfSHA := make(map[string]int64, len(refs)) + for _, ref := range refs { + if ref != nil && ref.BlobSHA256 != "" { + if _, ok := firstRefOfSHA[ref.BlobSHA256]; !ok { + firstRefOfSHA[ref.BlobSHA256] = ref.ID + } + } + } + for _, ref := range refs { + info := infos[ref.BlobSHA256] + item := projectAssetResponse{ AssetID: ref.ID, Filename: ref.FileName, SHA256: ref.BlobSHA256, ProjectID: ref.ProjectID, - ParseState: states[ref.BlobSHA256], - }) + ParseState: info.State, + ParseError: info.ErrorMessage, + } + // 同项目内相同内容多次引用：除首条外全部标注重复（物理只存一份）。 + if shaCount[ref.BlobSHA256] > 1 && firstRefOfSHA[ref.BlobSHA256] != ref.ID { + item.Deduplicated = true + } + if item.ParseState == "" && ref.BlobSHA256 != "" && !seenSHA[ref.BlobSHA256] { + // 内容已入库但从未触发解析（如早于 MinerU 接入的历史资产）： + // 列表读取时自动补触发，前端轮询即可看到 pending→running→done。 + seenSHA[ref.BlobSHA256] = true + if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() && mineru.SupportsExt(ref.FileName) { + if data, _, err := cas.ReadRefBytes(r.Context(), ref.ID); err == nil { + matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID) + item.ParseState = "pending" + } + } + } + out = append(out, item) } writeJSONResponse(w, http.StatusOK, out) } diff --git a/internal/server/server.go b/internal/server/server.go index 4df7a335..e49b3584 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -246,6 +246,7 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("POST /api/v1/projects/{id}/assets", s.handleUploadProjectAsset) mux.HandleFunc("GET /api/v1/projects/{id}/assets", s.handleListProjectAssets) mux.HandleFunc("GET /api/v1/projects/{id}/assets/{asset_id}/raw", s.handleGetProjectAssetRaw) + mux.HandleFunc("POST /api/v1/projects/{id}/assets/{asset_id}/reparse", s.handleReparseProjectAsset) // 6. 多级上下文与偏好记忆管理 mux.HandleFunc("GET /api/v1/context/user", s.handleGetUserContext)