package job import ( "context" "fmt" "strings" libbrave "haixun-backend/internal/library/brave" app "haixun-backend/internal/library/errors" "haixun-backend/internal/library/errors/code" libkg "haixun-backend/internal/library/knowledge" "haixun-backend/internal/library/placement" branddomain "haixun-backend/internal/model/brand/domain/usecase" jobdom "haixun-backend/internal/model/job/domain/usecase" kgusecase "haixun-backend/internal/model/knowledge_graph/domain/usecase" placementtopicdomain "haixun-backend/internal/model/placement_topic/domain/usecase" placementusecase "haixun-backend/internal/model/placement/usecase" scanpostusecase "haixun-backend/internal/model/scan_post/domain/usecase" threadsaccountdomain "haixun-backend/internal/model/threads_account/domain/usecase" ) type ScanPlacementDeps struct { Jobs jobdom.UseCase Brand branddomain.UseCase PlacementTopic placementtopicdomain.UseCase KnowledgeGraph kgusecase.UseCase ScanPost scanpostusecase.UseCase ThreadsAccount threadsaccountdomain.UseCase Placement placementusecase.UseCase } func RegisterScanPlacementHandler(runner *Runner, deps ScanPlacementDeps) { if runner == nil { return } runner.RegisterStepHandler("crawl", func(ctx context.Context, step StepContext) error { return runScanPlacement(ctx, step, deps) }) } func runScanPlacement(ctx context.Context, step StepContext, deps ScanPlacementDeps) error { payload := step.Run.Payload tenantID := stringField(payload, "tenant_id") ownerUID := stringField(payload, "owner_uid") brandID := brandIDFromPayload(payload) graphID := stringField(payload, "graph_id") if tenantID == "" || ownerUID == "" || brandID == "" { return fmt.Errorf("placement-scan payload missing tenant_id, owner_uid, or brand_id") } patrolMode := boolField(payload, "patrol_mode") brand, brandErr := deps.Brand.Get(ctx, tenantID, ownerUID, brandID) if brandErr != nil { return brandErr } if brand == nil { return fmt.Errorf("brand not found") } topicID := topicIDFromPayload(payload) if topicID == "" && step.Run != nil && strings.TrimSpace(step.Run.Scope) == "placement_topic" { topicID = strings.TrimSpace(step.Run.ScopeID) } var graphSummary *kgusecase.GraphSummary var graphErr error if topicID != "" { graphSummary, graphErr = deps.KnowledgeGraph.GetByTopic(ctx, tenantID, ownerUID, topicID) } else { graphSummary, graphErr = deps.KnowledgeGraph.Get(ctx, tenantID, ownerUID, brandID) } if graphErr != nil { if patrolMode && isKnowledgeGraphNotFound(graphErr) { graphSummary = nil if graphID == "" { if topicID != "" { graphID = topicID } else { graphID = brandID } } } else { return graphErr } } else if graphID == "" { graphID = graphSummary.ID } researchMap := brand.ResearchMap if topicID != "" && deps.PlacementTopic != nil { if topic, topicErr := deps.PlacementTopic.Get(ctx, tenantID, ownerUID, topicID); topicErr == nil && topic != nil { researchMap = topic.ResearchMap brand.ProductID = topic.ProductID } } patrolKeywords := []string{} if patrolMode { productBrief := strings.TrimSpace(brand.ProductBrief) if formatted := placement.ProductBriefFromContext(brand.ProductContext); formatted != "" { productBrief = formatted } patrolNodes := []libkg.Node{} if graphSummary != nil { patrolNodes = graphSummary.Nodes } patrolKeywords = libkg.ResolveScanPatrolKeywords( stringSliceField(payload, "patrol_keywords"), researchMap.PatrolKeywords, libkg.PatrolTagInputFromBrand(brand, productBrief), patrolNodes, ) if len(patrolKeywords) == 0 { return fmt.Errorf("請先在研究地圖填寫要回覆的海巡關鍵字") } } nodes := []libkg.Node{} if graphSummary != nil { nodes = graphSummary.Nodes } if !patrolMode { if graphSummary == nil { return fmt.Errorf("請先產生延伸知識圖譜,或改用手動海巡關鍵字") } if ids := stringSliceField(payload, "node_ids"); len(ids) > 0 { nodes = filterNodesByIDs(graphSummary.Nodes, ids) } else { nodes = selectedNodes(graphSummary.Nodes) } if len(nodes) == 0 { return fmt.Errorf("請先勾選要海巡的節點並儲存") } } research, err := deps.Placement.ResearchSettings(ctx, tenantID, ownerUID) if err != nil { return err } memberCtx, err := deps.ThreadsAccount.ResolveMemberPlacementContext(ctx, tenantID, ownerUID, research) if err != nil { return err } if !memberCtx.AllowsBrave && !memberCtx.AllowsThreadsAPI && !memberCtx.AllowsCrawler { return fmt.Errorf("目前連線模式無法海巡,請確認 Threads API、Brave 或 Chrome Session 設定") } if placement.MemberNeedsBraveKey(memberCtx) && strings.TrimSpace(memberCtx.BraveAPIKey) == "" { return fmt.Errorf("請在設定頁設定 Brave Search API key(跟隨此登入帳號)") } if memberCtx.DevMode && !memberCtx.BrowserConnected { return fmt.Errorf("開發模式需先同步 Chrome Session") } if !memberCtx.DevMode && memberCtx.AllowsThreadsAPI && !memberCtx.ApiConnected { return fmt.Errorf("正式模式需先完成 Threads API 連線") } updateProgress := func(summary string, percentage int) { _ = step.Heartbeat(ctx) _, _ = deps.Jobs.UpdateProgress(ctx, jobdom.UpdateProgressRequest{ JobID: step.JobID, WorkerID: step.WorkerID, Phase: "crawl", Summary: summary, Percentage: percentage, }) } exclusions := append([]string{}, researchMap.Exclusions...) if len(patrolKeywords) > 0 { updateProgress(fmt.Sprintf("依 %d 組海巡關鍵字準備雙軌搜尋…", len(patrolKeywords)), 5) } else { updateProgress("準備置入海巡…", 5) } if err := deps.ScanPost.ClearPlacementScan(ctx, tenantID, ownerUID, brandID, topicID); err != nil { return err } braveClient := libbrave.NewClient(memberCtx.BraveAPIKey) crawlerFn := placement.WrapPoliteCrawler(makeCrawlerSearchFn(deps, tenantID, ownerUID)) graphNodes := []libkg.Node{} if graphSummary != nil { graphNodes = graphSummary.Nodes } checkpointReq := scanpostusecase.CheckpointRequest{ TenantID: tenantID, OwnerUID: ownerUID, BrandID: brandID, TopicID: topicID, GraphID: graphID, ScanJobID: step.JobID, } candidates, err := placement.RunDualTrackDiscover(ctx, placement.DualTrackInput{ Nodes: graphNodes, PatrolKeywords: patrolKeywords, Exclusions: exclusions, Member: memberCtx, Client: braveClient, Crawler: crawlerFn, OnCheckpoint: func(batch []placement.ScanCandidate) error { if len(batch) == 0 { return nil } checkpointReq.Posts = batch saved, err := deps.ScanPost.UpsertScanCheckpoint(ctx, checkpointReq) if err != nil { return err } updateProgress(fmt.Sprintf("已儲存 %d 篇候選貼文(邊爬邊存)", saved), -1) return nil }, }, updateProgress) if err != nil { return err } scrapeReplies := memberCtx.ScrapeReplies if v, ok := payload["scrape_replies"].(bool); ok { scrapeReplies = v } else if memberCtx.ApiConnected && strings.TrimSpace(memberCtx.ThreadsAPIAccessToken) != "" { // Formal Threads API mode can fetch replies without browser session. scrapeReplies = true } if scrapeReplies { updateProgress("抓取高優先貼文留言…", 88) candidates = placement.AttachReplies(ctx, placement.ScrapeRepliesInput{ Posts: candidates, Member: memberCtx, RepliesPerPost: memberCtx.RepliesPerPost, }) } updateProgress(fmt.Sprintf("整理 %d 篇海巡結果…", len(candidates)), 92) count, err := deps.ScanPost.FinalizeScan(ctx, scanpostusecase.ReplaceRequest{ TenantID: tenantID, OwnerUID: ownerUID, BrandID: brandID, TopicID: topicID, GraphID: graphID, ScanJobID: step.JobID, Posts: candidates, }) if err != nil { return err } gold := 0 recent := 0 solved := 0 replyCount := 0 for _, item := range candidates { replyCount += len(item.Replies) if item.Priority == "gold" { gold++ } if item.Priority == "gold" || item.Priority == "recent" { recent++ } if item.SolvedByProduct { solved++ } } handoff := map[string]any{ "flow": "placement", "brand_id": brandID, "summary": fmt.Sprintf( "雙軌海巡完成:%d 篇(gold %d、近期軌 %d、產品可解 %d)", count, gold, recent, solved, ), "pain_breakdown": map[string]any{ "posts": count, "gold": gold, "recent_7d": recent, "solved_by_prod": solved, "replies": replyCount, }, "next_route": "/outreach?brand=" + brandID, "needs_supplemental_expand": false, "search_source_mode": string(memberCtx.SearchSourceMode), "dev_mode": memberCtx.DevMode, } _, err = deps.Jobs.CompleteRun(ctx, jobdom.CompleteRunRequest{ JobID: step.JobID, WorkerID: step.WorkerID, Result: map[string]any{ "post_count": count, "gold_count": gold, "recent_count": recent, "solved_count": solved, "reply_count": replyCount, "search_source_mode": string(memberCtx.SearchSourceMode), "handoff": handoff, }, }) return err } func selectedNodes(nodes []libkg.Node) []libkg.Node { out := make([]libkg.Node, 0, len(nodes)) for _, node := range nodes { if node.SelectedForScan { out = append(out, node) } } return out } func filterNodesByIDs(nodes []libkg.Node, ids []string) []libkg.Node { allowed := map[string]struct{}{} for _, id := range ids { id = strings.TrimSpace(id) if id != "" { allowed[id] = struct{}{} } } out := make([]libkg.Node, 0, len(ids)) for _, node := range nodes { if _, ok := allowed[node.ID]; ok { out = append(out, node) } } return out } func isKnowledgeGraphNotFound(err error) bool { if err == nil { return false } if e := app.FromError(err); e != nil && e.Category() == code.ResNotFound { return true } return strings.Contains(strings.ToLower(err.Error()), "knowledge graph not found") } func stringSliceField(payload map[string]any, key string) []string { if payload == nil { return nil } raw, ok := payload[key] if !ok || raw == nil { return nil } switch v := raw.(type) { case []string: return v case []any: out := make([]string, 0, len(v)) for _, item := range v { if s, ok := item.(string); ok { out = append(out, s) } } return out default: return nil } }