305 lines
9.2 KiB
Go
305 lines
9.2 KiB
Go
package server
|
|
|
|
import (
|
|
"io"
|
|
"net/http"
|
|
"path/filepath"
|
|
"strconv"
|
|
"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"`
|
|
ParseError string `json:"parse_error,omitempty"`
|
|
Deduplicated bool `json:"deduplicated,omitempty"`
|
|
}
|
|
|
|
func (s *Server) handleUploadProjectAsset(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
|
|
}
|
|
user, ok := s.requireProjectAccess(w, r, projectID)
|
|
if !ok {
|
|
return
|
|
}
|
|
userID := user.ID
|
|
if err := r.ParseMultipartForm(defaultUploadBodyLimit); err != nil {
|
|
if isRequestBodyTooLarge(err) {
|
|
writeSafeError(w, r, http.StatusRequestEntityTooLarge, "upload too large")
|
|
return
|
|
}
|
|
writeDomainError(w, r, http.StatusBadRequest, err)
|
|
return
|
|
}
|
|
file, header, err := r.FormFile("file")
|
|
if err != nil {
|
|
writeSafeError(w, r, http.StatusBadRequest, "file required")
|
|
return
|
|
}
|
|
defer file.Close()
|
|
|
|
data, err := io.ReadAll(file)
|
|
if err != nil {
|
|
writeSafeError(w, r, http.StatusBadRequest, "read file failed")
|
|
return
|
|
}
|
|
if len(data) == 0 {
|
|
writeSafeError(w, r, http.StatusBadRequest, "cannot store empty file")
|
|
return
|
|
}
|
|
|
|
filename := header.Filename
|
|
if filename == "" {
|
|
filename = "upload.bin"
|
|
}
|
|
declaredMIME := header.Header.Get("Content-Type")
|
|
mimeType, verr := upload.ValidateProjectMaterial(filename, declaredMIME, data)
|
|
if verr != nil {
|
|
writeDomainError(w, r, http.StatusUnsupportedMediaType, verr)
|
|
return
|
|
}
|
|
|
|
cas := s.cas()
|
|
if cas == nil {
|
|
writeSafeError(w, r, http.StatusServiceUnavailable, "storage unavailable")
|
|
return
|
|
}
|
|
ref, err := cas.StoreFile(r.Context(), projectID, userID, filepath.Base(filename), "material", data, mimeType)
|
|
if err != nil {
|
|
writeDomainError(w, r, http.StatusInternalServerError, err)
|
|
return
|
|
}
|
|
|
|
parseState := ""
|
|
if matIdx := bizdev.Snapshot(s.svc()).MaterialIndex; matIdx != nil && matIdx.Enabled() && matIdx.HasStore() {
|
|
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,
|
|
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() || !matIdx.HasStore() {
|
|
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",
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleListProjectAssets(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
|
|
}
|
|
if _, ok := s.requireProjectAccess(w, r, projectID); !ok {
|
|
return
|
|
}
|
|
cas := s.cas()
|
|
if cas == nil {
|
|
writeJSONResponse(w, http.StatusOK, []projectAssetResponse{})
|
|
return
|
|
}
|
|
refs, err := cas.ListProjectRefs(r.Context(), projectID)
|
|
if err != nil {
|
|
writeDomainError(w, r, http.StatusInternalServerError, err)
|
|
return
|
|
}
|
|
|
|
shas := make([]string, 0, len(refs))
|
|
for _, ref := range refs {
|
|
if ref != nil && ref.BlobSHA256 != "" {
|
|
shas = append(shas, ref.BlobSHA256)
|
|
}
|
|
}
|
|
infos := map[string]materialindex.ParseInfo{}
|
|
if matIdx := bizdev.Snapshot(s.svc()).MaterialIndex; matIdx != nil {
|
|
infos = matIdx.GetParseInfos(r.Context(), shas)
|
|
}
|
|
|
|
seenSHA := make(map[string]bool, len(refs))
|
|
shaCount := make(map[string]int, len(refs))
|
|
for _, ref := range refs {
|
|
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: 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() && matIdx.HasStore() && 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)
|
|
}
|
|
|
|
func (s *Server) handleGetProjectAssetRaw(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
|
|
}
|
|
rc, ref, err := cas.OpenFileByRef(r.Context(), assetID)
|
|
if err != nil || ref == nil {
|
|
writeSafeError(w, r, http.StatusNotFound, "asset not found")
|
|
return
|
|
}
|
|
defer rc.Close()
|
|
if ref.ProjectID != projectID {
|
|
writeSafeError(w, r, http.StatusNotFound, "asset not found")
|
|
return
|
|
}
|
|
filename := ref.FileName
|
|
if filename == "" {
|
|
filename = "download.bin"
|
|
}
|
|
mimeType := mimeFromExt(filename)
|
|
w.Header().Set("Content-Type", mimeType)
|
|
w.Header().Set("Content-Disposition", "inline; filename=\""+filepath.Base(filename)+"\"")
|
|
w.Header().Set("Cache-Control", "private, max-age=300")
|
|
w.WriteHeader(http.StatusOK)
|
|
_, _ = io.Copy(w, rc)
|
|
}
|
|
|
|
func parseUserIDHeader(r *http.Request) int64 {
|
|
raw := strings.TrimSpace(r.Header.Get("X-User-ID"))
|
|
if raw == "" {
|
|
return 0
|
|
}
|
|
id, _ := strconv.ParseInt(raw, 10, 64)
|
|
return id
|
|
}
|
|
|
|
func mimeFromExt(name string) string {
|
|
ext := strings.ToLower(filepath.Ext(name))
|
|
switch ext {
|
|
case ".txt", ".md", ".csv", ".json", ".xml", ".yaml", ".yml", ".log":
|
|
return "text/plain"
|
|
case ".pdf":
|
|
return "application/pdf"
|
|
case ".docx":
|
|
return "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
case ".xlsx":
|
|
return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
default:
|
|
return "application/octet-stream"
|
|
}
|
|
}
|