From 455ed57231f493478969855cea4bbc6867ab7b67 Mon Sep 17 00:00:00 2001 From: Max Date: Tue, 2 Sep 2025 10:05:51 +0800 Subject: [PATCH] Refactor job handling and enhance document processing features - Renamed the Start method to Push for better semantic clarity in job execution. - Introduced OnceAndSave, CronAndSave, and DaemonAndSave methods for immediate job creation and saving. - Added new test cases for OnceAndSave, CronAndSave, and DaemonAndSave methods to ensure proper functionality. - Updated AddFileProcess, AddTextProcess, and AddURLProcess to utilize new job handling methods, improving consistency across document operations. - Enhanced error handling and logging in document processing functions for better traceability. --- job/job.go | 49 ++++- job/job_test.go | 143 ++++++++++++++- openapi/kb/addfile.go | 405 +++++++++++++++++++++++++++++++++++++---- openapi/kb/addtext.go | 391 +++++++++++++++++++++++++++++++++++---- openapi/kb/addurl.go | 391 +++++++++++++++++++++++++++++++++++---- openapi/kb/document.go | 21 --- openapi/kb/graph.go | 18 +- openapi/kb/kb.go | 13 ++ openapi/kb/segment.go | 38 +--- 9 files changed, 1297 insertions(+), 172 deletions(-) diff --git a/job/job.go b/job/job.go index 8fe48407..f1bff2da 100644 --- a/job/job.go +++ b/job/job.go @@ -23,6 +23,21 @@ func Once(mode ModeType, data map[string]interface{}) (*Job, error) { return makeJob(raw) } +// OnceAndSave create a new job and save it immediately +func OnceAndSave(mode ModeType, data map[string]interface{}) (*Job, error) { + job, err := Once(mode, data) + if err != nil { + return nil, err + } + + err = SaveJob(job) + if err != nil { + return nil, fmt.Errorf("failed to save job: %w", err) + } + + return job, nil +} + // Cron create a new job func Cron(mode ModeType, data map[string]interface{}, expression string) (*Job, error) { data["mode"] = mode @@ -35,6 +50,21 @@ func Cron(mode ModeType, data map[string]interface{}, expression string) (*Job, return makeJob(raw) } +// CronAndSave create a new cron job and save it immediately +func CronAndSave(mode ModeType, data map[string]interface{}, expression string) (*Job, error) { + job, err := Cron(mode, data, expression) + if err != nil { + return nil, err + } + + err = SaveJob(job) + if err != nil { + return nil, fmt.Errorf("failed to save job: %w", err) + } + + return job, nil +} + // Daemon create a new job func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) { data["mode"] = mode @@ -46,8 +76,23 @@ func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) { return makeJob(raw) } -// Start start the job -func (j *Job) Start() error { +// DaemonAndSave create a new daemon job and save it immediately +func DaemonAndSave(mode ModeType, data map[string]interface{}) (*Job, error) { + job, err := Daemon(mode, data) + if err != nil { + return nil, err + } + + err = SaveJob(job) + if err != nil { + return nil, fmt.Errorf("failed to save job: %w", err) + } + + return job, nil +} + +// Push pushes the job to execution queue (renamed from Start for better semantics) +func (j *Job) Push() error { // Get executions for this job executions, err := j.GetExecutions() if err != nil { diff --git a/job/job_test.go b/job/job_test.go index 703c8c21..486f572b 100644 --- a/job/job_test.go +++ b/job/job_test.go @@ -164,7 +164,7 @@ func TestOnceGoroutine(t *testing.T) { t.Fatal(err) } - err = testJob.Start() + err = testJob.Push() if err != nil { t.Fatal(err) } @@ -199,7 +199,7 @@ func TestOnceProcess(t *testing.T) { t.Fatal(err) } - err = testJob.Start() + err = testJob.Push() if err != nil { t.Fatal(err) } @@ -361,7 +361,7 @@ func TestCommand(t *testing.T) { t.Fatal(err) } - err = testJob.Start() + err = testJob.Push() if err != nil { t.Fatal(err) } @@ -486,7 +486,7 @@ func TestJobExecution(t *testing.T) { } // Start the job - err = testJob.Start() + err = testJob.Push() if err != nil { t.Fatalf("Failed to start job: %v", err) } @@ -543,3 +543,138 @@ func TestJobExecution(t *testing.T) { t.Log("No log items found, this may be expected if logging is async") } } + +// TestOnceAndSave test OnceAndSave method +func TestOnceAndSave(t *testing.T) { + // Setup + test.Prepare(&testing.T{}, config.Conf) + defer test.Clean() + + // Register test processes + registerTestProcesses() + + // Test OnceAndSave - should create and save job in one step + testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "Test OnceAndSave Job", + "description": "Job created and saved with OnceAndSave method", + }) + if err != nil { + t.Fatalf("Failed to create and save job: %v", err) + } + + // Job should have a valid JobID after OnceAndSave + if testJob.JobID == "" { + t.Error("Expected job to have JobID after OnceAndSave") + } + + // Verify job was saved to database + retrievedJob, err := job.GetJob(testJob.JobID) + if err != nil { + t.Fatalf("Failed to retrieve saved job: %v", err) + } + + if retrievedJob.Name != "Test OnceAndSave Job" { + t.Errorf("Expected job name 'Test OnceAndSave Job', got '%s'", retrievedJob.Name) + } + + // Add execution and push + err = testJob.Add(&job.ExecutionOptions{ + Priority: 1, + SharedData: map[string]interface{}{ + "test_data": "OnceAndSave test", + }, + }, "test.job.echo", "OnceAndSave test") + if err != nil { + t.Fatal(err) + } + + err = testJob.Push() + if err != nil { + t.Fatal(err) + } + + // Give some time for execution + time.Sleep(2 * time.Second) + + t.Log("OnceAndSave job completed successfully") +} + +// TestCronAndSave test CronAndSave method +func TestCronAndSave(t *testing.T) { + // Setup + test.Prepare(&testing.T{}, config.Conf) + defer test.Clean() + + // Register test processes + registerTestProcesses() + + // Test CronAndSave - should create and save cron job in one step + testJob, err := job.CronAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "Test CronAndSave Job", + "description": "Cron job created and saved with CronAndSave method", + }, "0 0 * * *") + if err != nil { + t.Fatalf("Failed to create and save cron job: %v", err) + } + + // Job should have a valid JobID after CronAndSave + if testJob.JobID == "" { + t.Error("Expected cron job to have JobID after CronAndSave") + } + + // Verify cron job was saved to database + retrievedJob, err := job.GetJob(testJob.JobID) + if err != nil { + t.Fatalf("Failed to retrieve saved cron job: %v", err) + } + + if retrievedJob.ScheduleType != string(job.ScheduleTypeCron) { + t.Errorf("Expected schedule type cron, got %s", retrievedJob.ScheduleType) + } + + if retrievedJob.Name != "Test CronAndSave Job" { + t.Errorf("Expected job name 'Test CronAndSave Job', got '%s'", retrievedJob.Name) + } + + t.Log("CronAndSave job created and saved successfully") +} + +// TestDaemonAndSave test DaemonAndSave method +func TestDaemonAndSave(t *testing.T) { + // Setup + test.Prepare(&testing.T{}, config.Conf) + defer test.Clean() + + // Register test processes + registerTestProcesses() + + // Test DaemonAndSave - should create and save daemon job in one step + testJob, err := job.DaemonAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "Test DaemonAndSave Job", + "description": "Daemon job created and saved with DaemonAndSave method", + }) + if err != nil { + t.Fatalf("Failed to create and save daemon job: %v", err) + } + + // Job should have a valid JobID after DaemonAndSave + if testJob.JobID == "" { + t.Error("Expected daemon job to have JobID after DaemonAndSave") + } + + // Verify daemon job was saved to database + retrievedJob, err := job.GetJob(testJob.JobID) + if err != nil { + t.Fatalf("Failed to retrieve saved daemon job: %v", err) + } + + if retrievedJob.ScheduleType != string(job.ScheduleTypeDaemon) { + t.Errorf("Expected schedule type daemon, got %s", retrievedJob.ScheduleType) + } + + if retrievedJob.Name != "Test DaemonAndSave Job" { + t.Errorf("Expected job name 'Test DaemonAndSave Job', got '%s'", retrievedJob.Name) + } + + t.Log("DaemonAndSave job created and saved successfully") +} diff --git a/openapi/kb/addfile.go b/openapi/kb/addfile.go index 44f56c1b..ccd56bf3 100644 --- a/openapi/kb/addfile.go +++ b/openapi/kb/addfile.go @@ -6,26 +6,25 @@ import ( "github.com/gin-gonic/gin" "github.com/yaoapp/gou/graphrag/utils" + "github.com/yaoapp/gou/model" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/kun/exception" "github.com/yaoapp/kun/log" "github.com/yaoapp/kun/maps" "github.com/yaoapp/yao/attachment" + "github.com/yaoapp/yao/job" "github.com/yaoapp/yao/kb" "github.com/yaoapp/yao/openapi/response" ) -// AddFileProcess processes a file addition request with business logic only -// This function is Gin-agnostic and can be used for both sync and async operations -func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) error { +// CreateDocumentRecord creates a document record in the database immediately +// This is called synchronously when the API request comes in +func CreateDocumentRecord(ctx context.Context, req *AddFileRequest, jobID string) error { // Check if kb.Instance is available if kb.Instance == nil { return fmt.Errorf("knowledge base not initialized") } - // Validate request - if err := req.Validate(); err != nil { - return err - } - // Get file manager m, ok := attachment.Managers[req.Uploader] if !ok { @@ -49,11 +48,6 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e return fmt.Errorf("failed to get file info: %w", err) } - // DocID should be generated by the caller before calling this function - if req.DocID == "" { - return fmt.Errorf("document ID is required") - } - // Get KB config config, err := kb.GetConfig() if err != nil { @@ -73,29 +67,52 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e "file_path": path, "file_mime_type": contentType, "size": int64(fileInfo.Bytes), - } - - // Add job_id if provided (for async operations) - if len(jobID) > 0 && jobID[0] != "" { - documentData["job_id"] = jobID[0] + "job_id": jobID, } // Add base request fields req.BaseUpsertRequest.AddBaseFields(documentData) - // First create database record + // Create database record _, err = config.CreateDocument(maps.MapStrAny(documentData)) if err != nil { return fmt.Errorf("failed to save document metadata: %w", err) } + return nil +} + +// HandleFileContent processes the actual file content and updates the knowledge base +// This is called asynchronously by the job system +func HandleFileContent(ctx context.Context, req *AddFileRequest) error { + // Check if kb.Instance is available + if kb.Instance == nil { + return fmt.Errorf("knowledge base not initialized") + } + + // Get file manager + m, ok := attachment.Managers[req.Uploader] + if !ok { + return fmt.Errorf("invalid uploader: %s not found", req.Uploader) + } + + // Get file info and path + path, contentType, err := m.LocalPath(ctx, req.FileID) + if err != nil { + return fmt.Errorf("failed to get local path: %w", err) + } + + // Get KB config + config, err := kb.GetConfig() + if err != nil { + return fmt.Errorf("failed to get KB config: %w", err) + } + // Convert request to UpsertOptions upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions(path, contentType) if err != nil { - // Rollback: remove the database record - if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil { - log.Error("Failed to rollback document database record: %v", rollbackErr) - } + // Update status to error + config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()}) return fmt.Errorf("failed to convert request to upsert options: %w", err) } @@ -134,10 +151,38 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e return nil } +// AddFileHandler processes a file addition request with business logic only +// This function combines both document creation and content processing for sync operations +func AddFileHandler(ctx context.Context, req *AddFileRequest, jobID ...string) error { + // Validate request + if err := req.Validate(); err != nil { + return err + } + + // DocID should be generated by the caller before calling this function + if req.DocID == "" { + return fmt.Errorf("document ID is required") + } + + // For sync operations, create document record and process content immediately + var jid string + if len(jobID) > 0 { + jid = jobID[0] + } + + // Create document record + if err := CreateDocumentRecord(ctx, req, jid); err != nil { + return err + } + + // Process file content + return HandleFileContent(ctx, req) +} + // addFileWithRequest processes a file addition with pre-parsed request using Gin context func addFileWithRequest(c *gin.Context, req *AddFileRequest) { // Use the business logic function - err := AddFileProcess(c.Request.Context(), req) + err := AddFileHandler(c.Request.Context(), req) if err != nil { errorResp := &response.ErrorResponse{ Code: response.ErrServerError.Code, @@ -200,13 +245,17 @@ func AddFile(c *gin.Context) { func AddFileAsync(c *gin.Context) { var req AddFileRequest + log.Info("AddFileAsync: Starting async file addition") + // Check if kb.Instance is available if !checkKBInstance(c) { + log.Error("AddFileAsync: KB instance check failed") return } // Parse and bind JSON request if err := c.ShouldBindJSON(&req); err != nil { + log.Error("AddFileAsync: JSON binding failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: "Invalid request format: " + err.Error(), @@ -215,8 +264,11 @@ func AddFileAsync(c *gin.Context) { return } + log.Info("AddFileAsync: Request parsed successfully: %+v", req) + // Validate request if err := req.Validate(); err != nil { + log.Error("AddFileAsync: Request validation failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: err.Error(), @@ -225,35 +277,324 @@ func AddFileAsync(c *gin.Context) { return } + log.Info("AddFileAsync: Request validation passed") + // Validate file and get path _, _, err := validateFileAndGetPath(c, &req) if err != nil { + log.Error("AddFileAsync: File validation failed: %v", err) return } + log.Info("AddFileAsync: File validation passed") + // Convert request to UpsertOptions (just for validation) _, err = getUpsertOptions(c, &req.BaseUpsertRequest) if err != nil { + log.Error("AddFileAsync: UpsertOptions validation failed: %v", err) return } + log.Info("AddFileAsync: UpsertOptions validation passed") + // Generate document ID if not provided if req.DocID == "" { req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) } - // Create and run job - job := NewJob() - jobID := job.Run(func() { - err := AddFileProcess(context.Background(), &req, job.ID) - if err != nil { - log.Error("Async file processing failed: %v", err) - } + log.Info("AddFileAsync: Generated doc_id: %s", req.DocID) + + // Step 1: Create and save Job in one step to get JobID + j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "KB Add File", + "description": "Add file to knowledge base collection", }) + if err != nil { + log.Error("AddFileAsync: Job creation and save failed: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create and save job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddFileAsync: Job created and saved with ID: %s", j.JobID) + + // Step 2: Create document record immediately with job_id + err = CreateDocumentRecord(c.Request.Context(), &req, j.JobID) + if err != nil { + log.Error("AddFileAsync: Failed to create document record: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create document record: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddFileAsync: Document record created successfully") + + // Step 4: Add execution to job + jobData := map[string]interface{}{ + "collection_id": req.CollectionID, + "file_id": req.FileID, + "uploader": req.Uploader, + "locale": req.Locale, + "doc_id": req.DocID, + "metadata": req.Metadata, + "chunking": req.Chunking, + "embedding": req.Embedding, + "extraction": req.Extraction, + "fetcher": req.Fetcher, + "converter": req.Converter, + } + + err = j.Add(&job.ExecutionOptions{ + Priority: 1, + }, "kb.documents.processfile", jobData) + if err != nil { + log.Error("AddFileAsync: Failed to add job execution: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to add job execution: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + // Step 5: Push the job to execution queue + err = j.Push() + if err != nil { + log.Error("AddFileAsync: Failed to push job: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to push job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddFileAsync: Job pushed successfully") // Return job_id and doc_id response.RespondWithSuccess(c, response.StatusCreated, gin.H{ - "job_id": jobID, + "job_id": j.JobID, "doc_id": req.DocID, }) } + +// ProcessAddFile documents.addfile Knowledge Base add file processor +// Args[0] map: Request parameters {"collection_id": "collection", "file_id": "file123", "uploader": "local", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessAddFile(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddFileRequest structure + req := parseAddFileRequest(reqMap) + + // Get KB config to check if document exists + config, err := kb.GetConfig() + if err != nil { + exception.New("failed to get KB config: %s", 500, err.Error()).Throw() + } + + // Check if document already exists + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + + existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{}) + if err != nil || existingDoc == nil { + // Document doesn't exist, create it first (sync scenario) + log.Info("ProcessAddFile: Document %s not found, creating new record", req.DocID) + + // Get job_id from request if provided (for async scenario) + var jobID string + if jid, ok := reqMap["job_id"].(string); ok { + jobID = jid + } + + err = CreateDocumentRecord(ctx, req, jobID) + if err != nil { + exception.New("failed to create document record: %s", 500, err.Error()).Throw() + } + } else { + log.Info("ProcessAddFile: Document %s already exists, processing content only", req.DocID) + } + + // Process file content + err = HandleFileContent(ctx, req) + if err != nil { + exception.New("failed to process file: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// ProcessProcessFile documents.processfile Knowledge Base process file content processor (async version) +// Args[0] map: Request parameters {"collection_id": "collection", "file_id": "file123", "uploader": "local", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessProcessFile(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddFileRequest structure + req := parseAddFileRequest(reqMap) + + // This is async version - document record should already exist + // Just process the file content + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + err := HandleFileContent(ctx, req) + if err != nil { + exception.New("failed to process file content: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// parseAddFileRequest parses request map into AddFileRequest structure +func parseAddFileRequest(reqMap map[string]interface{}) *AddFileRequest { + req := &AddFileRequest{} + + // Required fields + if collectionID, ok := reqMap["collection_id"].(string); ok { + req.CollectionID = collectionID + } else { + exception.New("collection_id is required", 400).Throw() + } + + if fileID, ok := reqMap["file_id"].(string); ok { + req.FileID = fileID + } else { + exception.New("file_id is required", 400).Throw() + } + + // Optional fields + if uploader, ok := reqMap["uploader"].(string); ok { + req.Uploader = uploader + } else { + req.Uploader = "local" // Default to local uploader + } + + if locale, ok := reqMap["locale"].(string); ok { + req.Locale = locale + } + + if docID, ok := reqMap["doc_id"].(string); ok { + req.DocID = docID + } + + // Generate doc_id if not provided + if req.DocID == "" { + req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) + } + + // Handle metadata + if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok { + req.Metadata = metadata + } + + // Handle chunking configuration + if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok { + chunking := &ProviderConfig{} + if providerID, ok := chunkingMap["provider_id"].(string); ok { + chunking.ProviderID = providerID + } else { + exception.New("chunking.provider_id is required", 400).Throw() + } + if optionID, ok := chunkingMap["option_id"].(string); ok { + chunking.OptionID = optionID + } + req.Chunking = chunking + } else { + exception.New("chunking configuration is required", 400).Throw() + } + + // Handle embedding configuration + if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok { + embedding := &ProviderConfig{} + if providerID, ok := embeddingMap["provider_id"].(string); ok { + embedding.ProviderID = providerID + } else { + exception.New("embedding.provider_id is required", 400).Throw() + } + if optionID, ok := embeddingMap["option_id"].(string); ok { + embedding.OptionID = optionID + } + req.Embedding = embedding + } else { + exception.New("embedding configuration is required", 400).Throw() + } + + // Handle optional extraction configuration + if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok { + extraction := &ProviderConfig{} + if providerID, ok := extractionMap["provider_id"].(string); ok { + extraction.ProviderID = providerID + } + if optionID, ok := extractionMap["option_id"].(string); ok { + extraction.OptionID = optionID + } + req.Extraction = extraction + } + + // Handle optional fetcher configuration + if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok { + fetcher := &ProviderConfig{} + if providerID, ok := fetcherMap["provider_id"].(string); ok { + fetcher.ProviderID = providerID + } + if optionID, ok := fetcherMap["option_id"].(string); ok { + fetcher.OptionID = optionID + } + req.Fetcher = fetcher + } + + // Handle optional converter configuration + if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok { + converter := &ProviderConfig{} + if providerID, ok := converterMap["provider_id"].(string); ok { + converter.ProviderID = providerID + } + if optionID, ok := converterMap["option_id"].(string); ok { + converter.OptionID = optionID + } + req.Converter = converter + } + + return req +} diff --git a/openapi/kb/addtext.go b/openapi/kb/addtext.go index 8f268e4b..2edf0347 100644 --- a/openapi/kb/addtext.go +++ b/openapi/kb/addtext.go @@ -6,30 +6,24 @@ import ( "github.com/gin-gonic/gin" "github.com/yaoapp/gou/graphrag/utils" + "github.com/yaoapp/gou/model" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/kun/exception" "github.com/yaoapp/kun/log" "github.com/yaoapp/kun/maps" + "github.com/yaoapp/yao/job" "github.com/yaoapp/yao/kb" "github.com/yaoapp/yao/openapi/response" ) -// AddTextProcess processes a text addition request with business logic only -// This function is Gin-agnostic and can be used for both sync and async operations -func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) error { +// CreateTextDocumentRecord creates a text document record in the database immediately +// This is called synchronously when the API request comes in +func CreateTextDocumentRecord(ctx context.Context, req *AddTextRequest, jobID string) error { // Check if kb.Instance is available if kb.Instance == nil { return fmt.Errorf("knowledge base not initialized") } - // Validate request - if err := req.Validate(); err != nil { - return err - } - - // DocID should be generated by the caller before calling this function - if req.DocID == "" { - return fmt.Errorf("document ID is required") - } - // Get KB config config, err := kb.GetConfig() if err != nil { @@ -45,11 +39,7 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e "status": "pending", "text_content": req.Text, "size": int64(len(req.Text)), - } - - // Add job_id if provided (for async operations) - if len(jobID) > 0 && jobID[0] != "" { - documentData["job_id"] = jobID[0] + "job_id": jobID, } // Use title from metadata if available @@ -62,19 +52,34 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e // Add base request fields req.BaseUpsertRequest.AddBaseFields(documentData) - // First create database record + // Create database record _, err = config.CreateDocument(maps.MapStrAny(documentData)) if err != nil { return fmt.Errorf("failed to save document metadata: %w", err) } + return nil +} + +// HandleTextContent processes the actual text content and updates the knowledge base +// This is called asynchronously by the job system +func HandleTextContent(ctx context.Context, req *AddTextRequest) error { + // Check if kb.Instance is available + if kb.Instance == nil { + return fmt.Errorf("knowledge base not initialized") + } + + // Get KB config + config, err := kb.GetConfig() + if err != nil { + return fmt.Errorf("failed to get KB config: %w", err) + } + // Convert request to UpsertOptions upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions() if err != nil { - // Rollback: remove the database record - if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil { - log.Error("Failed to rollback document database record: %v", rollbackErr) - } + // Update status to error + config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()}) return fmt.Errorf("failed to convert request to upsert options: %w", err) } @@ -113,10 +118,38 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e return nil } +// AddTextHandler processes a text addition request with business logic only +// This function combines both document creation and content processing for sync operations +func AddTextHandler(ctx context.Context, req *AddTextRequest, jobID ...string) error { + // Validate request + if err := req.Validate(); err != nil { + return err + } + + // DocID should be generated by the caller before calling this function + if req.DocID == "" { + return fmt.Errorf("document ID is required") + } + + // For sync operations, create document record and process content immediately + var jid string + if len(jobID) > 0 { + jid = jobID[0] + } + + // Create document record + if err := CreateTextDocumentRecord(ctx, req, jid); err != nil { + return err + } + + // Process text content + return HandleTextContent(ctx, req) +} + // addTextWithRequest processes a text addition with pre-parsed request using Gin context func addTextWithRequest(c *gin.Context, req *AddTextRequest) { // Use the business logic function - err := AddTextProcess(c.Request.Context(), req) + err := AddTextHandler(c.Request.Context(), req) if err != nil { errorResp := &response.ErrorResponse{ Code: response.ErrServerError.Code, @@ -178,8 +211,17 @@ func AddText(c *gin.Context) { func AddTextAsync(c *gin.Context) { var req AddTextRequest + log.Info("AddTextAsync: Starting async text addition") + + // Check if kb.Instance is available + if !checkKBInstance(c) { + log.Error("AddTextAsync: KB instance check failed") + return + } + // Parse and bind JSON request if err := c.ShouldBindJSON(&req); err != nil { + log.Error("AddTextAsync: JSON binding failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: "Invalid request format: " + err.Error(), @@ -188,8 +230,11 @@ func AddTextAsync(c *gin.Context) { return } + log.Info("AddTextAsync: Request parsed successfully: %+v", req) + // Validate request if err := req.Validate(); err != nil { + log.Error("AddTextAsync: Request validation failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: err.Error(), @@ -198,34 +243,308 @@ func AddTextAsync(c *gin.Context) { return } - // Check if kb.Instance is available - if !checkKBInstance(c) { - return - } + log.Info("AddTextAsync: Request validation passed") // Convert request to UpsertOptions (just for validation) _, err := getUpsertOptions(c, &req.BaseUpsertRequest) if err != nil { + log.Error("AddTextAsync: UpsertOptions validation failed: %v", err) return } + log.Info("AddTextAsync: UpsertOptions validation passed") + // Generate document ID if not provided if req.DocID == "" { req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) } - // Create and run job - job := NewJob() - jobID := job.Run(func() { - err := AddTextProcess(context.Background(), &req, job.ID) - if err != nil { - log.Error("Async text processing failed: %v", err) - } + log.Info("AddTextAsync: Generated doc_id: %s", req.DocID) + + // Step 1: Create and save Job in one step to get JobID + j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "KB Add Text", + "description": "Add text to knowledge base collection", }) + if err != nil { + log.Error("AddTextAsync: Job creation and save failed: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create and save job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddTextAsync: Job created and saved with ID: %s", j.JobID) + + // Step 2: Create document record immediately with job_id + err = CreateTextDocumentRecord(c.Request.Context(), &req, j.JobID) + if err != nil { + log.Error("AddTextAsync: Failed to create document record: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create document record: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddTextAsync: Document record created successfully") + + // Step 3: Add execution to job + jobData := map[string]interface{}{ + "collection_id": req.CollectionID, + "text": req.Text, + "locale": req.Locale, + "doc_id": req.DocID, + "metadata": req.Metadata, + "chunking": req.Chunking, + "embedding": req.Embedding, + "extraction": req.Extraction, + "fetcher": req.Fetcher, + "converter": req.Converter, + } + + err = j.Add(&job.ExecutionOptions{ + Priority: 1, + }, "kb.documents.processtext", jobData) + if err != nil { + log.Error("AddTextAsync: Failed to add job execution: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to add job execution: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + // Step 4: Push the job to execution queue + err = j.Push() + if err != nil { + log.Error("AddTextAsync: Failed to push job: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to push job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddTextAsync: Job pushed successfully") // Return job_id and doc_id response.RespondWithSuccess(c, response.StatusCreated, gin.H{ - "job_id": jobID, + "job_id": j.JobID, "doc_id": req.DocID, }) } + +// ProcessAddText documents.addtext Knowledge Base add text processor (sync version) +// Args[0] map: Request parameters {"collection_id": "collection", "text": "content", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessAddText(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddTextRequest structure + req := parseAddTextRequest(reqMap) + + // Get KB config to check if document exists + config, err := kb.GetConfig() + if err != nil { + exception.New("failed to get KB config: %s", 500, err.Error()).Throw() + } + + // Check if document already exists + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + + existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{}) + if err != nil || existingDoc == nil { + // Document doesn't exist, create it first (sync scenario) + log.Info("ProcessAddText: Document %s not found, creating new record", req.DocID) + + // Get job_id from request if provided (for async scenario) + var jobID string + if jid, ok := reqMap["job_id"].(string); ok { + jobID = jid + } + + err = CreateTextDocumentRecord(ctx, req, jobID) + if err != nil { + exception.New("failed to create document record: %s", 500, err.Error()).Throw() + } + } else { + log.Info("ProcessAddText: Document %s already exists, processing content only", req.DocID) + } + + // Process text content + err = HandleTextContent(ctx, req) + if err != nil { + exception.New("failed to process text: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// ProcessProcessText documents.processtext Knowledge Base process text content processor (async version) +// Args[0] map: Request parameters {"collection_id": "collection", "text": "content", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessProcessText(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddTextRequest structure + req := parseAddTextRequest(reqMap) + + // This is async version - document record should already exist + // Just process the text content + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + err := HandleTextContent(ctx, req) + if err != nil { + exception.New("failed to process text content: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// parseAddTextRequest parses request map into AddTextRequest structure +func parseAddTextRequest(reqMap map[string]interface{}) *AddTextRequest { + req := &AddTextRequest{} + + // Required fields + if collectionID, ok := reqMap["collection_id"].(string); ok { + req.CollectionID = collectionID + } else { + exception.New("collection_id is required", 400).Throw() + } + + if text, ok := reqMap["text"].(string); ok { + req.Text = text + } else { + exception.New("text is required", 400).Throw() + } + + // Optional fields + if locale, ok := reqMap["locale"].(string); ok { + req.Locale = locale + } + + if docID, ok := reqMap["doc_id"].(string); ok { + req.DocID = docID + } + + // Generate doc_id if not provided + if req.DocID == "" { + req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) + } + + // Handle metadata + if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok { + req.Metadata = metadata + } + + // Handle chunking configuration + if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok { + chunking := &ProviderConfig{} + if providerID, ok := chunkingMap["provider_id"].(string); ok { + chunking.ProviderID = providerID + } else { + exception.New("chunking.provider_id is required", 400).Throw() + } + if optionID, ok := chunkingMap["option_id"].(string); ok { + chunking.OptionID = optionID + } + req.Chunking = chunking + } else { + exception.New("chunking configuration is required", 400).Throw() + } + + // Handle embedding configuration + if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok { + embedding := &ProviderConfig{} + if providerID, ok := embeddingMap["provider_id"].(string); ok { + embedding.ProviderID = providerID + } else { + exception.New("embedding.provider_id is required", 400).Throw() + } + if optionID, ok := embeddingMap["option_id"].(string); ok { + embedding.OptionID = optionID + } + req.Embedding = embedding + } else { + exception.New("embedding configuration is required", 400).Throw() + } + + // Handle optional extraction configuration + if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok { + extraction := &ProviderConfig{} + if providerID, ok := extractionMap["provider_id"].(string); ok { + extraction.ProviderID = providerID + } + if optionID, ok := extractionMap["option_id"].(string); ok { + extraction.OptionID = optionID + } + req.Extraction = extraction + } + + // Handle optional fetcher configuration + if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok { + fetcher := &ProviderConfig{} + if providerID, ok := fetcherMap["provider_id"].(string); ok { + fetcher.ProviderID = providerID + } + if optionID, ok := fetcherMap["option_id"].(string); ok { + fetcher.OptionID = optionID + } + req.Fetcher = fetcher + } + + // Handle optional converter configuration + if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok { + converter := &ProviderConfig{} + if providerID, ok := converterMap["provider_id"].(string); ok { + converter.ProviderID = providerID + } + if optionID, ok := converterMap["option_id"].(string); ok { + converter.OptionID = optionID + } + req.Converter = converter + } + + return req +} diff --git a/openapi/kb/addurl.go b/openapi/kb/addurl.go index f9a44f04..01817a91 100644 --- a/openapi/kb/addurl.go +++ b/openapi/kb/addurl.go @@ -6,30 +6,24 @@ import ( "github.com/gin-gonic/gin" "github.com/yaoapp/gou/graphrag/utils" + "github.com/yaoapp/gou/model" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/kun/exception" "github.com/yaoapp/kun/log" "github.com/yaoapp/kun/maps" + "github.com/yaoapp/yao/job" "github.com/yaoapp/yao/kb" "github.com/yaoapp/yao/openapi/response" ) -// AddURLProcess processes a URL addition request with business logic only -// This function is Gin-agnostic and can be used for both sync and async operations -func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) error { +// CreateURLDocumentRecord creates a URL document record in the database immediately +// This is called synchronously when the API request comes in +func CreateURLDocumentRecord(ctx context.Context, req *AddURLRequest, jobID string) error { // Check if kb.Instance is available if kb.Instance == nil { return fmt.Errorf("knowledge base not initialized") } - // Validate request - if err := req.Validate(); err != nil { - return err - } - - // DocID should be generated by the caller before calling this function - if req.DocID == "" { - return fmt.Errorf("document ID is required") - } - // Get KB config config, err := kb.GetConfig() if err != nil { @@ -44,11 +38,7 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err "type": "url", "status": "pending", "url": req.URL, - } - - // Add job_id if provided (for async operations) - if len(jobID) > 0 && jobID[0] != "" { - documentData["job_id"] = jobID[0] + "job_id": jobID, } // Use title from metadata if available @@ -61,19 +51,34 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err // Add base request fields req.BaseUpsertRequest.AddBaseFields(documentData) - // First create database record + // Create database record _, err = config.CreateDocument(maps.MapStrAny(documentData)) if err != nil { return fmt.Errorf("failed to save document metadata: %w", err) } + return nil +} + +// HandleURLContent processes the actual URL content and updates the knowledge base +// This is called asynchronously by the job system +func HandleURLContent(ctx context.Context, req *AddURLRequest) error { + // Check if kb.Instance is available + if kb.Instance == nil { + return fmt.Errorf("knowledge base not initialized") + } + + // Get KB config + config, err := kb.GetConfig() + if err != nil { + return fmt.Errorf("failed to get KB config: %w", err) + } + // Convert request to UpsertOptions upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions() if err != nil { - // Rollback: remove the database record - if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil { - log.Error("Failed to rollback document database record: %v", rollbackErr) - } + // Update status to error + config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()}) return fmt.Errorf("failed to convert request to upsert options: %w", err) } @@ -112,10 +117,38 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err return nil } +// AddURLHandler processes a URL addition request with business logic only +// This function combines both document creation and content processing for sync operations +func AddURLHandler(ctx context.Context, req *AddURLRequest, jobID ...string) error { + // Validate request + if err := req.Validate(); err != nil { + return err + } + + // DocID should be generated by the caller before calling this function + if req.DocID == "" { + return fmt.Errorf("document ID is required") + } + + // For sync operations, create document record and process content immediately + var jid string + if len(jobID) > 0 { + jid = jobID[0] + } + + // Create document record + if err := CreateURLDocumentRecord(ctx, req, jid); err != nil { + return err + } + + // Process URL content + return HandleURLContent(ctx, req) +} + // addURLWithRequest processes a URL addition with pre-parsed request using Gin context func addURLWithRequest(c *gin.Context, req *AddURLRequest) { // Use the business logic function - err := AddURLProcess(c.Request.Context(), req) + err := AddURLHandler(c.Request.Context(), req) if err != nil { errorResp := &response.ErrorResponse{ Code: response.ErrServerError.Code, @@ -178,8 +211,17 @@ func AddURL(c *gin.Context) { func AddURLAsync(c *gin.Context) { var req AddURLRequest + log.Info("AddURLAsync: Starting async URL addition") + + // Check if kb.Instance is available + if !checkKBInstance(c) { + log.Error("AddURLAsync: KB instance check failed") + return + } + // Parse and bind JSON request if err := c.ShouldBindJSON(&req); err != nil { + log.Error("AddURLAsync: JSON binding failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: "Invalid request format: " + err.Error(), @@ -188,8 +230,11 @@ func AddURLAsync(c *gin.Context) { return } + log.Info("AddURLAsync: Request parsed successfully: %+v", req) + // Validate request if err := req.Validate(); err != nil { + log.Error("AddURLAsync: Request validation failed: %v", err) errorResp := &response.ErrorResponse{ Code: response.ErrInvalidRequest.Code, ErrorDescription: err.Error(), @@ -198,34 +243,308 @@ func AddURLAsync(c *gin.Context) { return } - // Check if kb.Instance is available - if !checkKBInstance(c) { - return - } + log.Info("AddURLAsync: Request validation passed") // Convert request to UpsertOptions (just for validation) _, err := getUpsertOptions(c, &req.BaseUpsertRequest) if err != nil { + log.Error("AddURLAsync: UpsertOptions validation failed: %v", err) return } + log.Info("AddURLAsync: UpsertOptions validation passed") + // Generate document ID if not provided if req.DocID == "" { req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) } - // Create and run job - job := NewJob() - jobID := job.Run(func() { - err := AddURLProcess(context.Background(), &req, job.ID) - if err != nil { - log.Error("Async URL processing failed: %v", err) - } + log.Info("AddURLAsync: Generated doc_id: %s", req.DocID) + + // Step 1: Create and save Job in one step to get JobID + j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{ + "name": "KB Add URL", + "description": "Add URL to knowledge base collection", }) + if err != nil { + log.Error("AddURLAsync: Job creation and save failed: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create and save job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddURLAsync: Job created and saved with ID: %s", j.JobID) + + // Step 2: Create document record immediately with job_id + err = CreateURLDocumentRecord(c.Request.Context(), &req, j.JobID) + if err != nil { + log.Error("AddURLAsync: Failed to create document record: %v", err) + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to create document record: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddURLAsync: Document record created successfully") + + // Step 3: Add execution to job + jobData := map[string]interface{}{ + "collection_id": req.CollectionID, + "url": req.URL, + "locale": req.Locale, + "doc_id": req.DocID, + "metadata": req.Metadata, + "chunking": req.Chunking, + "embedding": req.Embedding, + "extraction": req.Extraction, + "fetcher": req.Fetcher, + "converter": req.Converter, + } + + err = j.Add(&job.ExecutionOptions{ + Priority: 1, + }, "kb.documents.processurl", jobData) + if err != nil { + log.Error("AddURLAsync: Failed to add job execution: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to add job execution: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + // Step 4: Push the job to execution queue + err = j.Push() + if err != nil { + log.Error("AddURLAsync: Failed to push job: %v", err) + // Rollback: remove document record + if config, err := kb.GetConfig(); err == nil { + config.RemoveDocument(req.DocID) + } + errorResp := &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: "Failed to push job: " + err.Error(), + } + response.RespondWithError(c, response.StatusInternalServerError, errorResp) + return + } + + log.Info("AddURLAsync: Job pushed successfully") // Return job_id and doc_id response.RespondWithSuccess(c, response.StatusCreated, gin.H{ - "job_id": jobID, + "job_id": j.JobID, "doc_id": req.DocID, }) } + +// ProcessAddURL documents.addurl Knowledge Base add URL processor (sync version) +// Args[0] map: Request parameters {"collection_id": "collection", "url": "https://example.com", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessAddURL(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddURLRequest structure + req := parseAddURLRequest(reqMap) + + // Get KB config to check if document exists + config, err := kb.GetConfig() + if err != nil { + exception.New("failed to get KB config: %s", 500, err.Error()).Throw() + } + + // Check if document already exists + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + + existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{}) + if err != nil || existingDoc == nil { + // Document doesn't exist, create it first (sync scenario) + log.Info("ProcessAddURL: Document %s not found, creating new record", req.DocID) + + // Get job_id from request if provided (for async scenario) + var jobID string + if jid, ok := reqMap["job_id"].(string); ok { + jobID = jid + } + + err = CreateURLDocumentRecord(ctx, req, jobID) + if err != nil { + exception.New("failed to create document record: %s", 500, err.Error()).Throw() + } + } else { + log.Info("ProcessAddURL: Document %s already exists, processing content only", req.DocID) + } + + // Process URL content + err = HandleURLContent(ctx, req) + if err != nil { + exception.New("failed to process URL: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// ProcessProcessURL documents.processurl Knowledge Base process URL content processor (async version) +// Args[0] map: Request parameters {"collection_id": "collection", "url": "https://example.com", ...} +// Return: map: Response data {"doc_id": "document_id"} +func ProcessProcessURL(process *process.Process) interface{} { + process.ValidateArgNums(1) + + // Get parameters + reqMap := process.ArgsMap(0) + + // Check knowledge base instance + if kb.Instance == nil { + exception.New("knowledge base not initialized", 500).Throw() + } + + // Convert parameters to AddURLRequest structure + req := parseAddURLRequest(reqMap) + + // This is async version - document record should already exist + // Just process the URL content + ctx := process.Context + if ctx == nil { + ctx = context.Background() + } + err := HandleURLContent(ctx, req) + if err != nil { + exception.New("failed to process URL content: %s", 500, err.Error()).Throw() + } + + // Return result + return maps.MapStrAny{ + "doc_id": req.DocID, + } +} + +// parseAddURLRequest parses request map into AddURLRequest structure +func parseAddURLRequest(reqMap map[string]interface{}) *AddURLRequest { + req := &AddURLRequest{} + + // Required fields + if collectionID, ok := reqMap["collection_id"].(string); ok { + req.CollectionID = collectionID + } else { + exception.New("collection_id is required", 400).Throw() + } + + if url, ok := reqMap["url"].(string); ok { + req.URL = url + } else { + exception.New("url is required", 400).Throw() + } + + // Optional fields + if locale, ok := reqMap["locale"].(string); ok { + req.Locale = locale + } + + if docID, ok := reqMap["doc_id"].(string); ok { + req.DocID = docID + } + + // Generate doc_id if not provided + if req.DocID == "" { + req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) + } + + // Handle metadata + if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok { + req.Metadata = metadata + } + + // Handle chunking configuration + if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok { + chunking := &ProviderConfig{} + if providerID, ok := chunkingMap["provider_id"].(string); ok { + chunking.ProviderID = providerID + } else { + exception.New("chunking.provider_id is required", 400).Throw() + } + if optionID, ok := chunkingMap["option_id"].(string); ok { + chunking.OptionID = optionID + } + req.Chunking = chunking + } else { + exception.New("chunking configuration is required", 400).Throw() + } + + // Handle embedding configuration + if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok { + embedding := &ProviderConfig{} + if providerID, ok := embeddingMap["provider_id"].(string); ok { + embedding.ProviderID = providerID + } else { + exception.New("embedding.provider_id is required", 400).Throw() + } + if optionID, ok := embeddingMap["option_id"].(string); ok { + embedding.OptionID = optionID + } + req.Embedding = embedding + } else { + exception.New("embedding configuration is required", 400).Throw() + } + + // Handle optional extraction configuration + if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok { + extraction := &ProviderConfig{} + if providerID, ok := extractionMap["provider_id"].(string); ok { + extraction.ProviderID = providerID + } + if optionID, ok := extractionMap["option_id"].(string); ok { + extraction.OptionID = optionID + } + req.Extraction = extraction + } + + // Handle optional fetcher configuration + if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok { + fetcher := &ProviderConfig{} + if providerID, ok := fetcherMap["provider_id"].(string); ok { + fetcher.ProviderID = providerID + } + if optionID, ok := fetcherMap["option_id"].(string); ok { + fetcher.OptionID = optionID + } + req.Fetcher = fetcher + } + + // Handle optional converter configuration + if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok { + converter := &ProviderConfig{} + if providerID, ok := converterMap["provider_id"].(string); ok { + converter.ProviderID = providerID + } + if optionID, ok := converterMap["option_id"].(string); ok { + converter.OptionID = optionID + } + req.Converter = converter + } + + return req +} diff --git a/openapi/kb/document.go b/openapi/kb/document.go index 14358645..0676e244 100644 --- a/openapi/kb/document.go +++ b/openapi/kb/document.go @@ -6,7 +6,6 @@ import ( "strings" "github.com/gin-gonic/gin" - "github.com/google/uuid" "github.com/yaoapp/gou/graphrag/types" "github.com/yaoapp/gou/model" "github.com/yaoapp/yao/attachment" @@ -52,26 +51,6 @@ var ( } ) -// SimpleJob represents a simple job for async operations -// TODO: replace with proper job system later -type SimpleJob struct { - ID string -} - -// NewJob creates a new simple job -func NewJob() *SimpleJob { - return &SimpleJob{ - ID: uuid.New().String(), - } -} - -// Run executes the job function asynchronously and returns job ID -func (j *SimpleJob) Run(fn func()) string { - // temporary solution to handle async operations ( TODO: use job queue ) - go fn() - return j.ID -} - // Document Management Handlers // ListDocuments lists documents with pagination diff --git a/openapi/kb/graph.go b/openapi/kb/graph.go index 5a571d12..d9f041e5 100644 --- a/openapi/kb/graph.go +++ b/openapi/kb/graph.go @@ -471,21 +471,13 @@ func ExtractSegmentGraphAsync(c *gin.Context) { // TODO: Implement document permission validation for docID - // Create and run job - job := NewJob() - jobID := job.Run(func() { - // TODO: Implement async extract segment graph logic - // err := ExtractSegmentGraphProcess(context.Background(), segmentID, extractOptions, job.ID) - // For now, just simulate async processing - // if err != nil { - // log.Error("Async graph extraction failed: %v", err) - // } - }) + // TODO: Implement async extract segment graph logic using Job system + // err := ExtractSegmentGraphProcess(context.Background(), segmentID, extractOptions, job.ID) - // Return job ID for status tracking + // Temporary response until async implementation is completed result := gin.H{ - "job_id": jobID, - "message": "Graph extraction started", + "message": "Async graph extraction not yet implemented", + "status": "pending_implementation", "doc_id": docID, "segment_id": segmentID, } diff --git a/openapi/kb/kb.go b/openapi/kb/kb.go index c43ccfe5..f0882f61 100644 --- a/openapi/kb/kb.go +++ b/openapi/kb/kb.go @@ -2,11 +2,24 @@ package kb import ( "github.com/gin-gonic/gin" + "github.com/yaoapp/gou/process" "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/kb" "github.com/yaoapp/yao/openapi/oauth/types" ) +func init() { + // Register kb process handlers + process.RegisterGroup("kb", map[string]process.Handler{ + "documents.addfile": ProcessAddFile, + "documents.addtext": ProcessAddText, + "documents.addurl": ProcessAddURL, + "documents.processfile": ProcessProcessFile, + "documents.processtext": ProcessProcessText, + "documents.processurl": ProcessProcessURL, + }) +} + // Attach attaches the Knowledge Base API to the router func Attach(group *gin.RouterGroup, oauth types.OAuth) { diff --git a/openapi/kb/segment.go b/openapi/kb/segment.go index b99b3180..43faa50e 100644 --- a/openapi/kb/segment.go +++ b/openapi/kb/segment.go @@ -671,22 +671,13 @@ func AddSegmentsAsync(c *gin.Context) { // TODO: Implement document permission validation for docID - // Create and run job - job := NewJob() - jobID := job.Run(func() { - // TODO: Implement async add segments logic - // This should call the same logic as AddSegments but in background - // err := AddSegmentsProcess(context.Background(), &req, job.ID) - // For now, just simulate async processing - // if err != nil { - // log.Error("Async segments addition failed: %v", err) - // } - }) + // TODO: Implement async add segments logic using Job system + // This should call the same logic as AddSegments but in background - // Return job ID for status tracking + // Temporary response until async implementation is completed result := gin.H{ - "job_id": jobID, - "message": "Segments addition started", + "message": "Async segments addition not yet implemented", + "status": "pending_implementation", "doc_id": docID, } @@ -730,22 +721,13 @@ func UpdateSegmentsAsync(c *gin.Context) { // TODO: Validate request body // TODO: Implement document permission validation for docID - // Create and run job - job := NewJob() - jobID := job.Run(func() { - // TODO: Implement async update segments logic - // This should call the same logic as UpdateSegments but in background - // err := UpdateSegmentsProcess(context.Background(), docID, requestBody, job.ID) - // For now, just simulate async processing - // if err != nil { - // log.Error("Async segments update failed: %v", err) - // } - }) + // TODO: Implement async update segments logic using Job system + // This should call the same logic as UpdateSegments but in background - // Return job ID for status tracking + // Temporary response until async implementation is completed result := gin.H{ - "job_id": jobID, - "message": "Segments update started", + "message": "Async segments update not yet implemented", + "status": "pending_implementation", "doc_id": docID, }