package api import ( "crypto/rand" "encoding/hex" "fmt" "io" "log" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "time" "github.com/gin-gonic/gin" "eai_agentplatform/backend/internal/dal" "eai_agentplatform/backend/internal/middleware" "eai_agentplatform/backend/internal/model" "eai_agentplatform/backend/internal/web" ) // mediaDAO 素材仓库(便于测试时覆写),包内共享。 var mediaDAO dal.MediaFileDAO func init() { mediaDAO = dal.MediaFileDAO{} } var blockedUploadExt = map[string]bool{ "exe": true, "bin": true, "dll": true, "so": true, "dylib": true, "msi": true, "apk": true, "ipa": true, "deb": true, "rpm": true, "pkg": true, "appimage": true, "bat": true, "cmd": true, "com": true, "scr": true, "sys": true, "drv": true, "ps1": true, "psm1": true, "vbs": true, "vbe": true, "js": true, "jse": true, "wsf": true, "wsh": true, "reg": true, "lnk": true, "iso": true, "img": true, "dmg": true, } var videoExt = map[string]bool{ "mp4": true, "mov": true, "avi": true, "mkv": true, "webm": true, "m4v": true, "wmv": true, "flv": true, } var validBindType = map[string]bool{"company": true, "product": true, "course": true, "none": true} const defaultChunkSize = 5 * 1024 * 1024 // 5MB // randomID 生成随机十六进制标识(文件名/上传会话) func randomID() string { b := make([]byte, 16) if _, err := rand.Read(b); err != nil { return strconv.FormatInt(time.Now().UnixNano(), 16) } return hex.EncodeToString(b) } func extOf(filename string) string { return strings.ToLower(strings.TrimPrefix(filepath.Ext(filename), ".")) } func isUploadableExt(ext string) bool { normalized := strings.ToLower(strings.TrimSpace(ext)) if normalized == "" { return false } return !blockedUploadExt[normalized] } func isVideoExt(ext string) bool { return videoExt[strings.ToLower(strings.TrimSpace(ext))] } func sizeLimitFor(ext string) int64 { if isVideoExt(ext) { return Cfg.FileMaxVideo } return Cfg.FileMaxDoc } // 审批前置:未审批文件存 pending/、驳回文件存 rejected/ 子目录(均不对外公开), // 审批通过后移动到 approved/ 子目录(静态公开,仅 approved 可访问)。 func approvedDir() string { return filepath.Join(Cfg.KBDataDir, "approved") } func pendingDir() string { return filepath.Join(Cfg.KBDataDir, "pending") } func rejectedDir() string { return filepath.Join(Cfg.KBDataDir, "rejected") } // mediaPathFor 按状态返回素材物理路径 func mediaPathFor(m model.MediaFile) string { switch m.Status { case "approved": return filepath.Join(approvedDir(), m.StoredName) case "rejected": return filepath.Join(rejectedDir(), m.StoredName) default: return filepath.Join(pendingDir(), m.StoredName) } } func parseBind(c *gin.Context) (string, *uint) { bt := c.PostForm("bind_type") if bt == "" { bt = "none" } if !validBindType[bt] { bt = "none" } var bid *uint if s := c.PostForm("bind_id"); s != "" { if n, err := strconv.ParseUint(s, 10, 64); err == nil && n > 0 { v := uint(n) bid = &v } } return bt, bid } func parseKnowledgeSpaceKey(c *gin.Context) string { key := sanitizeSpaceKey(c.PostForm("knowledge_space_key")) if key == "" { return "general" } return ensureKnowledgeSpaceKeyOrDefault(key) } func sourceOf(c *gin.Context) string { u := middleware.CurrentUser(c) if u != nil && u.Role == "admin" { return "admin" } return "employee" } // ============ 直传 ============ // Upload POST /api/media/upload —— 文档/小视频直传 func Upload(c *gin.Context) { u := middleware.CurrentUser(c) file, header, err := c.Request.FormFile("file") if err != nil { web.Fail(c, web.NewBadRequest("缺少文件字段 file")) return } defer file.Close() ext := extOf(header.Filename) if !isUploadableExt(ext) { web.Fail(c, web.NewBadRequest("不支持的文件类型")) return } if isVideoExt(ext) && header.Size > Cfg.ChunkThreshold { web.Fail(c, web.NewBadRequest("视频超过 100MB 请使用分片上传")) return } if header.Size > sizeLimitFor(ext) { web.Fail(c, web.NewBadRequest("文件超过大小限制")) return } bindType, bindID := parseBind(c) knowledgeSpaceKey := parseKnowledgeSpaceKey(c) source := sourceOf(c) status := "pending" targetDir := pendingDir() if source == "admin" { status = "approved" // 管理员上传自动通过 targetDir = approvedDir() } storedName := randomID() + "." + ext dst := filepath.Join(targetDir, storedName) if err := os.MkdirAll(targetDir, 0o755); err != nil { web.Fail(c, web.NewBadRequest("创建存储目录失败")) return } out, err := os.Create(dst) if err != nil { web.Fail(c, web.NewBadRequest("保存文件失败")) return } if _, err := io.Copy(out, file); err != nil { out.Close() web.Fail(c, web.NewBadRequest("写入文件失败")) return } out.Close() m := model.MediaFile{ Filename: header.Filename, StoredName: storedName, StoredPath: storedName, FileExt: ext, FileSize: header.Size, Status: status, Source: source, SubmitterID: u.ID, BindType: bindType, BindID: bindID, KnowledgeSpaceKey: knowledgeSpaceKey, } if !mediaDAO.Insert(&m) { web.Fail(c, web.NewBadRequest("创建素材记录失败")) return } if status == "approved" { go runExtractPipeline(m.ID) } web.OK(c, gin.H{"media_id": m.ID, "status": m.Status, "filename": m.Filename}) } // ============ 分片上传 ============ type uploadSession struct { Filename string FileSize int64 Ext string BindType string BindID *uint ChunkSize int64 ChunkCount int Chunks map[int]bool KnowledgeSpaceKey string } var uploadSessions = struct { sync.RWMutex m map[string]*uploadSession }{m: map[string]*uploadSession{}} // UploadInit POST /api/media/upload-init —— 初始化分片上传 func UploadInit(c *gin.Context) { var req struct { Filename string `json:"filename"` FileSize int64 `json:"file_size"` BindType string `json:"bind_type"` BindID *uint `json:"bind_id"` KnowledgeSpaceKey string `json:"knowledge_space_key"` } if err := c.ShouldBindJSON(&req); err != nil || req.Filename == "" || req.FileSize <= 0 { web.Fail(c, web.NewBadRequest("filename/file_size 必填")) return } ext := extOf(req.Filename) if !isUploadableExt(ext) { web.Fail(c, web.NewBadRequest("不支持的文件类型")) return } if req.FileSize > sizeLimitFor(ext) { web.Fail(c, web.NewBadRequest("文件超过大小限制")) return } if req.BindType == "" || !validBindType[req.BindType] { req.BindType = "none" } id := randomID() chunkCount := int((req.FileSize + defaultChunkSize - 1) / defaultChunkSize) sess := &uploadSession{ Filename: req.Filename, FileSize: req.FileSize, Ext: ext, BindType: req.BindType, BindID: req.BindID, ChunkSize: defaultChunkSize, ChunkCount: chunkCount, Chunks: map[int]bool{}, } sess.KnowledgeSpaceKey = ensureKnowledgeSpaceKeyOrDefault(sanitizeSpaceKey(req.KnowledgeSpaceKey)) uploadSessions.Lock() uploadSessions.m[id] = sess uploadSessions.Unlock() web.OK(c, gin.H{"upload_id": id, "chunk_size": defaultChunkSize, "chunk_count": chunkCount}) } // UploadChunk POST /api/media/upload-chunk —— 上传分片 func UploadChunk(c *gin.Context) { id := c.PostForm("upload_id") idx, err := strconv.Atoi(c.PostForm("chunk_index")) if err != nil || id == "" { web.Fail(c, web.NewBadRequest("upload_id/chunk_index 必填")) return } uploadSessions.RLock() sess := uploadSessions.m[id] uploadSessions.RUnlock() if sess == nil { web.Fail(c, web.NewNotFoundError("上传会话不存在")) return } if idx < 0 || idx >= sess.ChunkCount { web.Fail(c, web.NewBadRequest("chunk_index 越界")) return } file, _, err := c.Request.FormFile("file") if err != nil { web.Fail(c, web.NewBadRequest("缺少分片文件")) return } defer file.Close() tmpDir := filepath.Join(pendingDir(), "tmp", id) if err := os.MkdirAll(tmpDir, 0o755); err != nil { web.Fail(c, web.NewBadRequest("创建临时目录失败")) return } dst := filepath.Join(tmpDir, fmt.Sprintf("%06d", idx)) out, err := os.Create(dst) if err != nil { web.Fail(c, web.NewBadRequest("保存分片失败")) return } if _, err := io.Copy(out, file); err != nil { out.Close() web.Fail(c, web.NewBadRequest("写入分片失败")) return } out.Close() uploadSessions.Lock() sess.Chunks[idx] = true uploadSessions.Unlock() web.OK(c, gin.H{"upload_id": id, "chunk_index": idx, "received": len(sess.Chunks)}) } // UploadComplete POST /api/media/upload-complete —— 合并分片 func UploadComplete(c *gin.Context) { u := middleware.CurrentUser(c) var req struct { UploadID string `json:"upload_id"` } if err := c.ShouldBindJSON(&req); err != nil || req.UploadID == "" { web.Fail(c, web.NewBadRequest("upload_id 必填")) return } uploadSessions.RLock() sess := uploadSessions.m[req.UploadID] uploadSessions.RUnlock() if sess == nil { web.Fail(c, web.NewNotFoundError("上传会话不存在")) return } if len(sess.Chunks) != sess.ChunkCount { web.Fail(c, web.NewBadRequest(fmt.Sprintf("分片不完整:%d/%d", len(sess.Chunks), sess.ChunkCount))) return } source := sourceOf(c) status := "pending" targetDir := pendingDir() if source == "admin" { status = "approved" // 管理员上传自动通过 targetDir = approvedDir() } storedName := randomID() + "." + sess.Ext if err := os.MkdirAll(targetDir, 0o755); err != nil { web.Fail(c, web.NewBadRequest("创建存储目录失败")) return } final := filepath.Join(targetDir, storedName) out, err := os.Create(final) if err != nil { web.Fail(c, web.NewBadRequest("创建最终文件失败")) return } tmpDir := filepath.Join(pendingDir(), "tmp", req.UploadID) for i := 0; i < sess.ChunkCount; i++ { part := filepath.Join(tmpDir, fmt.Sprintf("%06d", i)) f, err := os.Open(part) if err != nil { out.Close() web.Fail(c, web.NewBadRequest("读取分片失败")) return } if _, err := io.Copy(out, f); err != nil { f.Close() out.Close() web.Fail(c, web.NewBadRequest("合并分片失败")) return } f.Close() } out.Close() os.RemoveAll(tmpDir) uploadSessions.Lock() delete(uploadSessions.m, req.UploadID) uploadSessions.Unlock() m := model.MediaFile{ Filename: sess.Filename, StoredName: storedName, StoredPath: storedName, FileExt: sess.Ext, FileSize: sess.FileSize, Status: status, Source: source, SubmitterID: u.ID, BindType: sess.BindType, BindID: sess.BindID, KnowledgeSpaceKey: sess.KnowledgeSpaceKey, } if !mediaDAO.Insert(&m) { web.Fail(c, web.NewBadRequest("创建素材记录失败")) return } if status == "approved" { go runExtractPipeline(m.ID) } web.OK(c, gin.H{"media_id": m.ID, "status": m.Status, "filename": m.Filename}) } // ============ 预览 / 状态 ============ // Preview GET /api/media/preview/{media_id} —— 仅 approved 可预览 func Preview(c *gin.Context) { id, ok := parseID(c, "media_id") if !ok { return } m, found := mediaDAO.GetByID(id) if !found { web.Fail(c, web.NewNotFoundError("素材不存在")) return } if m.Status != "approved" { web.Fail(c, web.NewForbiddenError("素材未通过审批,不可预览")) return } web.OK(c, gin.H{ "preview_url": "/media/" + m.StoredName, "file_ext": m.FileExt, "can_preview": true, }) } // MediaStatus GET /api/media/status/{media_id} func MediaStatus(c *gin.Context) { id, ok := parseID(c, "media_id") if !ok { return } m, found := mediaDAO.GetByID(id) if !found { web.Fail(c, web.NewNotFoundError("素材不存在")) return } web.OK(c, gin.H{ "status": m.Status, "extracted": m.Extracted, "chunk_count": chunkDAO.CountByMediaFile(m.ID), "knowledge_space_key": m.KnowledgeSpaceKey, }) } // ============ 审批 ============ // AuditList GET /api/media/audit-list?status=&page=&size= func AuditList(c *gin.Context) { page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) size, _ := strconv.Atoi(c.DefaultQuery("size", "20")) if page < 1 { page = 1 } if size < 1 || size > 100 { size = 20 } total, items := mediaDAO.ListForAudit( c.Query("status"), sanitizeSpaceKey(c.Query("knowledge_space_key")), page, size) web.OK(c, gin.H{"total": total, "items": items}) } // AuditMedia POST /api/media/audit/{media_id} —— 审批(通过→触发异步提取) func AuditMedia(c *gin.Context) { u := middleware.CurrentUser(c) id, ok := parseID(c, "media_id") if !ok { return } m, found := mediaDAO.GetByID(id) if !found { web.Fail(c, web.NewNotFoundError("素材不存在")) return } if m.Status != "pending" { web.Fail(c, web.NewConflictError("该素材已审批,不可重复操作")) return } var req struct { Action string `json:"action"` RejectReason string `json:"reject_reason"` } if err := c.ShouldBindJSON(&req); err != nil { web.Fail(c, web.NewBadRequest("请求参数错误")) return } now := time.Now() auditBy := u.ID switch req.Action { case "approve": // 审批前置:通过后才把文件从 pending/ 移动到 approved/(公开) src := filepath.Join(pendingDir(), m.StoredName) dst := filepath.Join(approvedDir(), m.StoredName) if err := os.MkdirAll(approvedDir(), 0o755); err != nil { web.Fail(c, web.NewBadRequest("创建公开目录失败")) return } if err := os.Rename(src, dst); err != nil { web.Fail(c, web.NewBadRequest("移动文件失败:"+err.Error())) return } m.Status = "approved" m.AuditBy = &auditBy m.AuditAt = &now m.RejectReason = "" case "reject": if strings.TrimSpace(req.RejectReason) == "" { web.Fail(c, web.NewBadRequest("驳回理由必填")) return } // 驳回:把文件从 pending/ 移到 rejected/,落实「审批前置」物理隔离 src := filepath.Join(pendingDir(), m.StoredName) dst := filepath.Join(rejectedDir(), m.StoredName) if err := os.MkdirAll(rejectedDir(), 0o755); err != nil { web.Fail(c, web.NewBadRequest("创建驳回目录失败")) return } if err := os.Rename(src, dst); err != nil { web.Fail(c, web.NewBadRequest("移动文件失败:"+err.Error())) return } m.Status = "rejected" m.RejectReason = req.RejectReason m.AuditBy = &auditBy m.AuditAt = &now default: web.Fail(c, web.NewBadRequest("action 必须为 approve 或 reject")) return } if !mediaDAO.Update(&m) { web.Fail(c, web.NewBadRequest("审批失败")) return } if m.Status == "approved" { go runExtractPipeline(m.ID) } web.OK(c, gin.H{"id": m.ID, "status": m.Status}) } // ============ 异步文档提取管线 ============ // runExtractPipeline 审批通过后:文档转 PDF → pdftotext → 切片入库 func runExtractPipeline(mediaID uint) { m, found := mediaDAO.GetByID(mediaID) if !found { return } docExts := map[string]bool{"ppt": true, "pptx": true, "doc": true, "docx": true} var text string var err error src := mediaPathFor(m) switch { case docExts[m.FileExt]: text, err = libreofficeExtract(src) case m.FileExt == "pdf": text, err = pdftotextExtract(src) default: // 视频/图片:仅标记,不提取 mediaDAO.MarkExtracted(m.ID) return } if err != nil { log.Printf("[提取失败] media_id=%d: %v", mediaID, err) return } chunks := splitIntoChunks(text, 1000) spaceKey := resolveMediaKnowledgeSpaceKey(m) sourceID := strconv.FormatUint(uint64(m.ID), 10) items := make([]model.KnowledgeChunk, 0, len(chunks)) for i, chunk := range chunks { if strings.TrimSpace(chunk) == "" { continue } items = append(items, model.KnowledgeChunk{ MediaFileID: &m.ID, SourceType: m.FileExt, SourceID: sourceID, KnowledgeSpaceKey: spaceKey, ChunkIndex: i, Content: chunk, }) } chunkDAO.BulkInsert(items) mediaDAO.MarkExtracted(m.ID) triggerKnowledgeIndexRebuild() log.Printf("[提取完成] media_id=%d chunks=%d", mediaID, len(chunks)) } func libreofficeExtract(src string) (string, error) { tmp, err := os.MkdirTemp("", "lo-convert") if err != nil { return "", err } defer os.RemoveAll(tmp) // 独立 UserInstallation:systemd 加固(ProtectHome=true)下 HOME 不可写, // 避免 LibreOffice profile 初始化失败。 profileURI := "file://" + filepath.Join(tmp, "profile") cmd := exec.Command(Cfg.LibreOfficeBin, "-env:UserInstallation="+profileURI, "--headless", "--convert-to", "pdf", "--outdir", tmp, src) if out, err := cmd.CombinedOutput(); err != nil { return "", fmt.Errorf("libreoffice 转换失败: %v %s", err, out) } base := strings.TrimSuffix(filepath.Base(src), filepath.Ext(src)) + ".pdf" return pdftotextExtract(filepath.Join(tmp, base)) } func pdftotextExtract(pdf string) (string, error) { cmd := exec.Command(Cfg.PdftotextBin, "-enc", "UTF-8", pdf, "-") out, err := cmd.Output() if err != nil { return "", fmt.Errorf("pdftotext 提取失败: %v", err) } return string(out), nil } func splitIntoChunks(text string, maxChars int) []string { paragraphs := strings.Split(text, "\n\n") var chunks []string current := "" for _, p := range paragraphs { p = strings.TrimSpace(p) if p == "" { continue } if len(current)+len(p) > maxChars { if current != "" { chunks = append(chunks, current) } current = p } else if current == "" { current = p } else { current += "\n\n" + p } } if current != "" { chunks = append(chunks, current) } return chunks }