package httpapi import ( "context" "database/sql" "encoding/json" "errors" "log" "strconv" "strings" "time" "git.118im.com/im_team/im_workbench/internal/ai" "git.118im.com/im_team/im_workbench/internal/events" "git.118im.com/im_team/im_workbench/internal/store" ) const ( aiWorkerCount = 2 aiWorkerPollInterval = time.Second aiTaskRunTimeout = 3 * time.Minute ) type aiDocumentTaskSnapshot struct { SourceText string `json:"source_text"` TitleHint string `json:"title_hint"` DocumentType string `json:"document_type"` Language string `json:"language"` } type aiTableTaskSnapshot struct { FileID string `json:"file_id"` FileName string `json:"file_name"` FileType string `json:"file_type"` FileSize int64 `json:"file_size"` Topic string `json:"topic"` SheetName string `json:"sheet_name"` Language string `json:"language"` RecordID string `json:"record_id"` } type aiMeetingSummaryTaskSnapshot struct { MeetingID string `json:"meeting_id"` SummaryID string `json:"summary_id"` FileIDs []string `json:"file_ids"` SummaryType string `json:"summary_type"` ForceRegenerate bool `json:"force_regenerate"` } func (a *API) StartAIWorkers(ctx context.Context) { if a == nil || a.store == nil || a.ai == nil { return } for i := 0; i < aiWorkerCount; i++ { go a.aiWorkerLoop(ctx, i+1) } } func (a *API) aiWorkerLoop(ctx context.Context, workerID int) { ticker := time.NewTicker(aiWorkerPollInterval) defer ticker.Stop() for { task, err := a.store.ClaimQueuedAITask(ctx) if errors.Is(err, store.ErrNotFound) { select { case <-ctx.Done(): return case <-ticker.C: continue } } if err != nil { log.Printf("ai worker claim failed: worker=%d err=%v", workerID, err) select { case <-ctx.Done(): return case <-ticker.C: continue } } a.runAITask(ctx, task) } } func (a *API) runAITask(parent context.Context, task *store.AITask) { ctx, cancel := context.WithTimeout(parent, aiTaskRunTimeout) defer cancel() st, err := a.store.ForTenant(ctx, task.TenantID, task.TenantCode) if err != nil { log.Printf("ai task tenant store failed: task_id=%s tenant_id=%d err=%v", task.TaskID, task.TenantID, err) return } switch task.Feature { case "smart_document": a.runAIDocumentTask(ctx, st, task) case "smart_table": a.runAITableTask(ctx, st, task) case "meeting_summary": a.runAIMeetingSummaryTask(ctx, st, task) default: a.finishAITaskFailure(ctx, st, task, "AI_TASK_UNSUPPORTED", "unsupported ai task feature") } } func (a *API) runAIDocumentTask(ctx context.Context, st *store.Store, task *store.AITask) { var input aiDocumentTaskSnapshot if !decodeTaskSnapshot(task, &input) { a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") return } result, err := a.ai.GenerateDocument(ctx, strings.TrimSpace(input.SourceText), strings.TrimSpace(input.TitleHint), strings.TrimSpace(input.DocumentType), strings.TrimSpace(input.Language)) if err != nil { a.finishAITaskAIError(ctx, st, task, err, hashText(input.SourceText)) return } taskResult := map[string]any{ "title": result.Title, "summary": result.Summary, "content_markdown": result.ContentMarkdown, "content_plain": result.ContentPlain, "sections": result.Sections, "source_excerpt": excerpt(input.SourceText, 200), "can_save": true, } ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, "", taskResult, "", "") if err != nil { log.Printf("ai document task finish failed: task_id=%s err=%v", task.TaskID, err) return } if !ok { return } userID := taskUserID(task) _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_document", TaskID: task.TaskID, Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.SourceText), OutputHash: hashText(result.ContentMarkdown), }) a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "smart_document", result.Title, result.Summary, map[string]any{ "task_id": task.TaskID, "feature": "smart_document", "status": "succeeded", "progress": 100, "result_type": "text", "result": compactAITextResult(result.Title, result.Summary, result.ContentMarkdown, result.ContentPlain), "updated_at": time.Now().UTC().Format(time.RFC3339Nano), }) } func (a *API) runAITableTask(ctx context.Context, st *store.Store, task *store.AITask) { var input aiTableTaskSnapshot if !decodeTaskSnapshot(task, &input) { a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") return } fileType := strings.ToLower(strings.TrimSpace(input.FileType)) fileMeta := st.AIFileMetadata(ctx, input.FileID) profile := a.tableProfileForAIContext(ctx, input.FileID, fileMeta, fileType) result, err := a.ai.AnalyzeTable(ctx, ai.TableAnalysisInput{ Topic: input.Topic, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize, Profile: profile, }) if err != nil { a.finishAITaskAIError(ctx, st, task, err, hashText(input.FileID+":"+input.Topic)) return } recordID := strings.TrimSpace(input.RecordID) if recordID == "" { recordID = "airec_tbl_" + randomToken(10) } if err := st.SaveAITableAnalysis(ctx, task.TenantCode, taskUserID(task), store.AIRecord{ RecordID: recordID, RecordType: "smart_table", Title: result.Title, Summary: excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160), Status: "succeeded", TaskID: task.TaskID, }, store.AITableAnalysis{ RecordID: recordID, FileID: input.FileID, FileName: input.FileName, FileType: fileType, FileSize: input.FileSize, Topic: input.Topic, MetricsJSON: mustJSON(result.Metrics, "[]"), PreviewJSON: mustJSON(result.Preview, "{}"), AlertsJSON: mustJSON(result.Alerts, "[]"), Interpretation: result.Interpretation, PromptVersion: "smart_table_v1", Model: result.Usage.Model, }); err != nil { a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) return } taskResult := map[string]any{"record_id": recordID, "title": result.Title, "metrics": result.Metrics, "alerts": result.Alerts, "interpretation": result.Interpretation} ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, recordID, taskResult, "", "") if err != nil { log.Printf("ai table task finish failed: task_id=%s err=%v", task.TaskID, err) return } if !ok { return } _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ TenantCode: task.TenantCode, UserID: task.UserID, Feature: "smart_table", TaskID: task.TaskID, Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.FileID + ":" + input.Topic), OutputHash: hashText(result.Interpretation), }) a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, "smart_table", result.Title, result.Summary, map[string]any{ "task_id": task.TaskID, "feature": "smart_table", "status": "succeeded", "progress": 100, "result_type": "record", "record_id": recordID, "result": map[string]any{"record_id": recordID, "title": result.Title, "summary": excerpt(firstNonEmpty(result.Summary, result.Interpretation), 160)}, "updated_at": time.Now().UTC().Format(time.RFC3339Nano), }) } func (a *API) runAIMeetingSummaryTask(ctx context.Context, st *store.Store, task *store.AITask) { var input aiMeetingSummaryTaskSnapshot if !decodeTaskSnapshot(task, &input) { a.finishAITaskFailure(ctx, st, task, reasonMalformedRequest, "invalid task input") return } userID := taskUserID(task) meeting, err := st.Meeting(ctx, userID, input.MeetingID) if err != nil { a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) return } files, err := st.MeetingNoteFiles(ctx, userID, input.MeetingID) if err != nil { a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) return } fileNames, fileIDs := meetingSummaryFiles(files, input.FileIDs) result, err := a.ai.SummarizeMeeting(ctx, ai.MeetingSummaryInput{ MeetingTitle: meeting.Title, StartAt: meeting.StartAt.UTC().Format(time.RFC3339Nano), EndAt: meeting.EndAt.UTC().Format(time.RFC3339Nano), Participants: meetingSummaryParticipants(*meeting), Files: fileNames, }) if err != nil { a.finishAITaskAIError(ctx, st, task, err, hashText(input.MeetingID+strings.Join(fileIDs, ","))) return } summaryID := strings.TrimSpace(input.SummaryID) if summaryID == "" { summaryID = "aisum_" + randomToken(10) } if err := st.UpsertAIMeetingSummary(ctx, task.TenantCode, userID, store.AIMeetingSummary{ SummaryID: summaryID, MeetingID: input.MeetingID, TaskID: task.TaskID, ConclusionsJSON: mustJSON(result.Conclusions, "[]"), DecisionsJSON: mustJSON(result.Decisions, "[]"), ActionItemsJSON: mustJSON(result.ActionItems, "[]"), SourceFileIDsJSON: mustJSON(fileIDs, "[]"), PromptVersion: "meeting_summary_v1", Model: result.Usage.Model, }); err != nil { a.finishAITaskFailure(ctx, st, task, reasonAITaskFailed, err.Error()) return } taskResult := map[string]any{ "summary_id": summaryID, "meeting_id": input.MeetingID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems, } ok, err := st.FinishAITaskResult(ctx, task.TaskID, "succeeded", 100, summaryID, taskResult, "", "") if err != nil { log.Printf("ai meeting summary task finish failed: task_id=%s err=%v", task.TaskID, err) return } if !ok { return } _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ TenantCode: task.TenantCode, UserID: task.UserID, Feature: "meeting_summary", TaskID: task.TaskID, Provider: "deepseek", Model: result.Usage.Model, ProviderRequestID: result.Usage.ProviderRequestID, PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, LatencyMS: int(result.Usage.LatencyMS), Status: "succeeded", InputHash: hashText(input.MeetingID + strings.Join(fileIDs, ",")), OutputHash: hashText(mustJSON(result.Conclusions, "[]")), }) a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, userID, task.TaskID, "meeting_summary", meeting.Title, "", map[string]any{ "task_id": task.TaskID, "feature": "meeting_summary", "status": "succeeded", "progress": 100, "result_type": "summary", "meeting_id": input.MeetingID, "summary_id": summaryID, "result": map[string]any{"meeting_id": input.MeetingID, "summary_id": summaryID, "conclusions": result.Conclusions, "decisions": result.Decisions, "action_items": result.ActionItems}, "updated_at": time.Now().UTC().Format(time.RFC3339Nano), }) } func (a *API) finishAITaskAIError(ctx context.Context, st *store.Store, task *store.AITask, err error, inputHash string) { reason := aiErrorReason(err) a.finishAITaskFailure(ctx, st, task, reason, err.Error()) _ = st.InsertAIUsageLog(ctx, store.AIUsageLog{ TenantCode: task.TenantCode, UserID: task.UserID, Feature: task.Feature, TaskID: task.TaskID, Provider: "deepseek", Status: "failed", ErrorReason: reason, InputHash: inputHash, }) } func (a *API) finishAITaskFailure(ctx context.Context, st *store.Store, task *store.AITask, reason, message string) { ok, err := st.FinishAITaskResult(ctx, task.TaskID, "failed", 100, "", nil, reason, message) if err != nil { log.Printf("ai task failure update failed: task_id=%s reason=%s err=%v", task.TaskID, reason, err) return } if !ok { return } a.publishAITaskEvent(ctx, task.TenantID, task.TenantCode, taskUserID(task), task.TaskID, task.Feature, "AI 任务失败", message, map[string]any{ "task_id": task.TaskID, "feature": task.Feature, "status": "failed", "progress": 100, "result_type": "none", "error": map[string]any{"reason": reason, "message": message}, "updated_at": time.Now().UTC().Format(time.RFC3339Nano), }) } func (a *API) publishAITaskEvent(ctx context.Context, tenantID int64, tenantCode string, userID int64, taskID, feature, title, summary string, data map[string]any) { if a == nil || a.events == nil || userID <= 0 || strings.TrimSpace(taskID) == "" { return } event := events.Event{ EventID: events.StableEventID(tenantID, "ai", "workbench.ai.task.updated", taskID+":"+stringValue(data["status"]), userID), TenantID: tenantID, TenantCode: tenantCode, ActorUserID: userID, Module: "ai", Action: "workbench.ai.task.updated", ResourceID: taskID, TargetUserIDs: []int64{userID}, Title: strings.TrimSpace(title), Summary: strings.TrimSpace(summary), OccurredAt: time.Now().UTC().Round(time.Millisecond), Data: data, } pubCtx, cancel := context.WithTimeout(ctx, workbenchEventPublishTimeout) defer cancel() if err := a.events.Publish(pubCtx, event); err != nil { log.Printf("ai task event publish failed: event_id=%s task_id=%s feature=%s err=%v", event.EventID, taskID, feature, err) } } func aiTaskSnapshot(value any) (sql.NullString, error) { data, err := json.Marshal(value) if err != nil { return sql.NullString{}, err } return sql.NullString{String: string(data), Valid: true}, nil } func decodeTaskSnapshot(task *store.AITask, dst any) bool { if task == nil || !task.InputSnapshot.Valid || strings.TrimSpace(task.InputSnapshot.String) == "" { return false } return json.Unmarshal([]byte(task.InputSnapshot.String), dst) == nil } func (a *API) tableProfileForAIContext(ctx context.Context, fileID string, meta store.AIFileMetadata, fileType string) ai.TableProfile { if path := resolveTableFilePath(fileID, meta.Location); path != "" { return tableProfileFromPath(path, fileType) } if strings.TrimSpace(meta.Location) == "" { return ai.TableProfile{Parsed: false, Message: "source file metadata is not available"} } path, cleanup, err := a.downloadOSSObjectToTemp(ctx, meta.Location, fileType) if err != nil { return ai.TableProfile{Parsed: false, Message: "source file is not available: " + err.Error()} } defer cleanup() return tableProfileFromPath(path, fileType) } func taskUserID(task *store.AITask) int64 { if task == nil { return 0 } userID, _ := strconv.ParseInt(strings.TrimSpace(task.UserID), 10, 64) return userID } func compactAITextResult(title, summary, markdown, plain string) map[string]any { result := map[string]any{"title": title, "summary": summary, "content_plain": plain} if len([]rune(markdown)) <= 8000 { result["content_markdown"] = markdown } return result } func stringValue(value any) string { text, _ := value.(string) return strings.TrimSpace(text) }