611 lines
16 KiB
Go
611 lines
16 KiB
Go
package placement
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
libkg "haixun-backend/internal/library/knowledge"
|
||
"haixun-backend/internal/library/websearch"
|
||
)
|
||
|
||
const (
|
||
relevanceLimitPerTag = 12
|
||
recencyLimitPerTag = 8
|
||
)
|
||
|
||
type ScanCandidate struct {
|
||
Permalink string
|
||
ExternalID string
|
||
Author string
|
||
AuthorID string
|
||
AuthorAvatar string
|
||
Text string
|
||
SearchTag string
|
||
QueryDimension QueryDimension
|
||
RecencyDays int
|
||
GraphNodeID string
|
||
ProductFitScore int
|
||
Source DiscoverChannel
|
||
HasRelevance bool
|
||
HasRecency bool
|
||
Priority string
|
||
AuthorVerified bool
|
||
FollowerCount int
|
||
AuthorFollowers int
|
||
LikeCount int
|
||
ReplyCount int
|
||
EngagementScore int
|
||
PlacementScore int
|
||
SolvedByProduct bool
|
||
PostedAt string
|
||
Replies []ReplyCandidate
|
||
Embedding []float32
|
||
SemanticScore int
|
||
EngagementPredicted int
|
||
AudienceQualityScore int
|
||
}
|
||
|
||
type DualTrackInput struct {
|
||
Nodes []libkg.Node
|
||
PatrolKeywords []string
|
||
Exclusions []string
|
||
PatrolContext PostScanContext
|
||
Member MemberContext
|
||
WebSearch websearch.Client
|
||
Crawler CrawlerSearchFn
|
||
Limit int // max queries budget; 0 = default
|
||
OnCheckpoint func(candidates []ScanCandidate) error
|
||
}
|
||
|
||
type DualTrackProgress func(message string, pct int)
|
||
|
||
// CollectTagQueries builds crawl jobs from selected graph nodes.
|
||
func CollectTagQueries(nodes []libkg.Node, provider websearch.Provider) []TagQuery {
|
||
out := make([]TagQuery, 0, len(nodes)*4)
|
||
for _, node := range nodes {
|
||
if !node.SelectedForScan {
|
||
continue
|
||
}
|
||
fit := node.ProductFitScore
|
||
derived := node.DerivedTags
|
||
if len(derived.Relevance) == 0 && len(derived.Recency) == 0 {
|
||
derived = libkg.DerivePatrolTagsForNode(node, libkg.PatrolTagInput{})
|
||
}
|
||
for _, tag := range derived.Relevance {
|
||
tag = strings.TrimSpace(tag)
|
||
if tag == "" {
|
||
continue
|
||
}
|
||
q := BuildRelevanceQuery(provider, tag)
|
||
if q == "" {
|
||
continue
|
||
}
|
||
out = append(out, TagQuery{
|
||
Tag: tag,
|
||
Query: q,
|
||
Dimension: QueryRelevance,
|
||
GraphNodeID: node.ID,
|
||
ProductFitScore: fit,
|
||
})
|
||
}
|
||
for _, tag := range derived.Recency {
|
||
tag = strings.TrimSpace(tag)
|
||
if tag == "" {
|
||
continue
|
||
}
|
||
q7 := BuildRecencyQuery(provider, tag, IdealMaxPostAgeDays)
|
||
if q7 != "" {
|
||
out = append(out, TagQuery{
|
||
Tag: tag,
|
||
Query: q7,
|
||
Dimension: QueryRecency,
|
||
GraphNodeID: node.ID,
|
||
ProductFitScore: fit,
|
||
RecencyDays: IdealMaxPostAgeDays,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// RunDualTrackDiscover executes relevance + recency queries and merges by permalink.
|
||
func RunDualTrackDiscover(ctx context.Context, input DualTrackInput, onProgress DualTrackProgress) ([]ScanCandidate, error) {
|
||
queries := ResolveTagQueries(
|
||
input.Nodes,
|
||
input.PatrolKeywords,
|
||
input.Member.WebSearchProviderEnum(),
|
||
PatrolTagInputFromScanContext(input.PatrolContext),
|
||
)
|
||
if len(queries) == 0 {
|
||
if len(input.PatrolKeywords) > 0 {
|
||
return nil, fmt.Errorf("海巡關鍵字格式無效，請改用 2～8 字的真人搜尋短句")
|
||
}
|
||
selected := 0
|
||
for _, node := range input.Nodes {
|
||
if node.SelectedForScan {
|
||
selected++
|
||
}
|
||
}
|
||
if selected > 0 {
|
||
return nil, fmt.Errorf("已勾選節點但沒有可用的海巡 tag，請重新擴展圖譜或手動編輯 tag")
|
||
}
|
||
return nil, fmt.Errorf("請先勾選要海巡的節點並儲存")
|
||
}
|
||
|
||
merged := map[string]*ScanCandidate{}
|
||
order := make([]string, 0, 64)
|
||
|
||
runQuery := func(tq TagQuery, limit int) error {
|
||
posts, channel, err := discoverForQuery(ctx, input, tq, limit)
|
||
if err != nil {
|
||
if onProgress != nil {
|
||
onProgress(fmt.Sprintf("略過「%s」：%s", tq.Tag, err.Error()), -1)
|
||
}
|
||
return nil
|
||
}
|
||
if len(posts) == 0 {
|
||
return nil
|
||
}
|
||
for _, post := range posts {
|
||
if MatchesExclusion(post.Text, input.Exclusions) {
|
||
continue
|
||
}
|
||
if LooksLikeCasualChat(post.Text) {
|
||
continue
|
||
}
|
||
postedAt := strings.TrimSpace(post.PostedAt)
|
||
if !PassesDiscoverFilters(post.Text, tq.Tag, postedAt, tq.Dimension, tq.ProductFitScore, tq.RecencyDays, input.PatrolContext) {
|
||
continue
|
||
}
|
||
bodyFit := ScorePostBodyProductFit(post.Text, input.PatrolContext)
|
||
semanticScore := LocalIntentScore(post.Text, input.PatrolContext)
|
||
effectiveFit := bodyFit
|
||
if tq.ProductFitScore > 0 && bodyFit > 0 {
|
||
effectiveFit = (tq.ProductFitScore + bodyFit*2) / 3
|
||
}
|
||
key := post.Permalink
|
||
if key == "" {
|
||
continue
|
||
}
|
||
existing, ok := merged[key]
|
||
if !ok {
|
||
priority := "relevant"
|
||
if tq.Dimension == QueryRecency {
|
||
priority = "recent"
|
||
}
|
||
extID := post.ExternalID
|
||
if extID == "" {
|
||
if parsed, ok := ParseThreadsPostFromWebResult(post.Text, "", post.Permalink); ok {
|
||
extID = parsed.ExternalID
|
||
}
|
||
}
|
||
semP := &semanticScore
|
||
merged[key] = &ScanCandidate{
|
||
Permalink: post.Permalink,
|
||
ExternalID: extID,
|
||
Author: post.Author,
|
||
AuthorVerified: post.AuthorVerified,
|
||
FollowerCount: post.FollowerCount,
|
||
Text: post.Text,
|
||
SearchTag: tq.Tag,
|
||
QueryDimension: tq.Dimension,
|
||
RecencyDays: recencyDaysForCandidate(tq),
|
||
GraphNodeID: tq.GraphNodeID,
|
||
ProductFitScore: effectiveFit,
|
||
Source: channel,
|
||
HasRelevance: tq.Dimension == QueryRelevance,
|
||
HasRecency: tq.Dimension == QueryRecency,
|
||
Priority: priority,
|
||
LikeCount: post.LikeCount,
|
||
ReplyCount: post.ReplyCount,
|
||
SemanticScore: semanticScore,
|
||
PlacementScore: computePlacementScore(post.Text, effectiveFit, tq.Dimension == QueryRecency, nil, semP, nil),
|
||
SolvedByProduct: PostSolvedByProduct(post.Text, effectiveFit, input.PatrolContext),
|
||
PostedAt: postedAt,
|
||
}
|
||
order = append(order, key)
|
||
continue
|
||
}
|
||
if tq.Dimension == QueryRelevance {
|
||
existing.HasRelevance = true
|
||
}
|
||
if tq.Dimension == QueryRecency {
|
||
existing.HasRecency = true
|
||
existing.RecencyDays = mergeRecencyDays(existing.RecencyDays, tq.RecencyDays)
|
||
}
|
||
if effectiveFit > existing.ProductFitScore || semanticScore > existing.SemanticScore {
|
||
if effectiveFit > existing.ProductFitScore {
|
||
existing.ProductFitScore = effectiveFit
|
||
}
|
||
if semanticScore > existing.SemanticScore {
|
||
existing.SemanticScore = semanticScore
|
||
}
|
||
existing.SolvedByProduct = PostSolvedByProduct(existing.Text, existing.ProductFitScore, input.PatrolContext)
|
||
semP := existing.SemanticScore
|
||
semPtr := &semP
|
||
if semP <= 0 {
|
||
semPtr = nil
|
||
}
|
||
existing.PlacementScore = computePlacementScore(existing.Text, existing.ProductFitScore, existing.HasRecency, nil, semPtr, nil)
|
||
}
|
||
if strings.TrimSpace(existing.PostedAt) == "" && strings.TrimSpace(post.PostedAt) != "" {
|
||
existing.PostedAt = strings.TrimSpace(post.PostedAt)
|
||
}
|
||
existing.ExternalID = PreferReplyExternalID(existing.ExternalID, post.ExternalID)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
total := len(queries)
|
||
for i, tq := range queries {
|
||
if onProgress != nil {
|
||
pct := 10 + ((i + 1) * 75 / max(total, 1))
|
||
onProgress(fmt.Sprintf("雙軌海巡 %d/%d：%s", i+1, total, tq.Tag), pct)
|
||
}
|
||
limit := perTagDiscoverLimit(total, tq.Dimension)
|
||
if err := runQuery(tq, limit); err != nil {
|
||
return nil, err
|
||
}
|
||
if input.OnCheckpoint != nil {
|
||
snapshot := snapshotMergedCandidates(merged, order, false, input.PatrolContext)
|
||
if err := input.OnCheckpoint(snapshot); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
if input.Member.AllowsCrawler && input.Member.BrowserConnected && i < total-1 {
|
||
if err := politeDiscoverPause(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
}
|
||
|
||
cascadeExtra := 0
|
||
for _, st := range buildTagCascadeStates(queries) {
|
||
for countCandidatesForTag(merged, st.Tag) < MinRecencyCandidatesPerTag {
|
||
if cascadeExtra >= MaxRecencyCascadeQueries {
|
||
break
|
||
}
|
||
days := nextRecencyCascadeWindow(st.RanWindows)
|
||
if days == 0 {
|
||
break
|
||
}
|
||
tq, ok := buildRecencyCascadeQuery(st, days, input.Member.WebSearchProviderEnum())
|
||
if !ok {
|
||
st.RanWindows[days] = true
|
||
continue
|
||
}
|
||
if onProgress != nil {
|
||
onProgress(fmt.Sprintf("近 %d 天結果不足，改搜近 %d 天：%s", previousRecencyWindow(days), days, st.Tag), -1)
|
||
}
|
||
limit := perTagDiscoverLimit(total, QueryRecency)
|
||
if err := runQuery(tq, limit); err != nil {
|
||
return nil, err
|
||
}
|
||
st.RanWindows[days] = true
|
||
cascadeExtra++
|
||
if input.OnCheckpoint != nil {
|
||
snapshot := snapshotMergedCandidates(merged, order, false, input.PatrolContext)
|
||
if err := input.OnCheckpoint(snapshot); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
if input.Member.AllowsCrawler && input.Member.BrowserConnected {
|
||
if err := politeDiscoverPause(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
}
|
||
if cascadeExtra >= MaxRecencyCascadeQueries {
|
||
break
|
||
}
|
||
}
|
||
|
||
out := snapshotMergedCandidates(merged, order, true, input.PatrolContext)
|
||
if onProgress != nil {
|
||
onProgress(fmt.Sprintf("合併完成，共 %d 篇候選貼文", len(out)), 90)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func discoverForQuery(ctx context.Context, input DualTrackInput, tq TagQuery, limit int) ([]DiscoverPost, DiscoverChannel, error) {
|
||
apiKeyword := tq.Tag
|
||
if shaped := libkg.ThreadsAPIKeyword(tq.Tag); shaped != "" {
|
||
apiKeyword = shaped
|
||
}
|
||
req := DiscoverRequest{
|
||
Query: tq.Query,
|
||
Keyword: apiKeyword,
|
||
Recency: tq.Dimension == QueryRecency,
|
||
Limit: limit,
|
||
Member: input.Member,
|
||
Crawler: input.Crawler,
|
||
}
|
||
merged := map[string]DiscoverPost{}
|
||
channel := DiscoverChannel("")
|
||
posts, primaryChannel, err := Discover(ctx, req)
|
||
for _, post := range posts {
|
||
key := strings.TrimSpace(post.Permalink)
|
||
if key == "" {
|
||
continue
|
||
}
|
||
if existing, ok := merged[key]; ok {
|
||
post.ExternalID = PreferReplyExternalID(existing.ExternalID, post.ExternalID)
|
||
}
|
||
merged[key] = post
|
||
}
|
||
if primaryChannel != "" {
|
||
channel = primaryChannel
|
||
}
|
||
|
||
webEnabled := input.WebSearch != nil && input.WebSearch.Enabled()
|
||
if webEnabled && input.Member.AllowsBrave {
|
||
webPosts, werr := discoverViaWebSearch(ctx, input.WebSearch, input.Member, tq, limit)
|
||
if werr != nil && len(merged) == 0 && err != nil {
|
||
return nil, "", err
|
||
}
|
||
for _, post := range webPosts {
|
||
key := strings.TrimSpace(post.Permalink)
|
||
if key == "" {
|
||
continue
|
||
}
|
||
merged[key] = post
|
||
}
|
||
if len(merged) > 0 && channel == "" {
|
||
channel = input.Member.WebSearchDiscoverChannel()
|
||
}
|
||
} else if len(merged) == 0 {
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
if !webEnabled {
|
||
return nil, "", fmt.Errorf("%s 未設定且 Threads API 無結果", input.Member.WebSearchProviderLabel())
|
||
}
|
||
}
|
||
|
||
out := make([]DiscoverPost, 0, len(merged))
|
||
for _, post := range merged {
|
||
out = append(out, post)
|
||
}
|
||
if len(out) > limit {
|
||
out = out[:limit]
|
||
}
|
||
if channel == "" && len(out) > 0 {
|
||
channel = DiscoverThreadsAPI
|
||
}
|
||
return out, channel, nil
|
||
}
|
||
|
||
func discoverViaWebSearch(ctx context.Context, client websearch.Client, member MemberContext, tq TagQuery, limit int) ([]DiscoverPost, error) {
|
||
res, err := client.Search(ctx, websearch.SearchOptions{
|
||
Query: tq.Query,
|
||
Limit: limit,
|
||
Mode: websearch.ModeThreadsDiscover,
|
||
Country: member.BraveCountry,
|
||
SearchLang: member.BraveSearchLang,
|
||
UserLocation: member.ExaUserLocation,
|
||
StartPublishedDate: PublishedAfterForRecency(member.WebSearchProviderEnum(), tq.RecencyDays),
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if res.Status != "success" || len(res.Results) == 0 {
|
||
return nil, nil
|
||
}
|
||
source := member.WebSearchDiscoverChannel()
|
||
out := make([]DiscoverPost, 0, len(res.Results))
|
||
for _, item := range res.Results {
|
||
parsed, ok := ParseThreadsPostFromWebResult(item.Title, item.Snippet, item.URL)
|
||
if !ok {
|
||
continue
|
||
}
|
||
out = append(out, DiscoverPost{
|
||
Text: parsed.Text,
|
||
Permalink: parsed.Permalink,
|
||
ExternalID: parsed.ExternalID,
|
||
Author: parsed.Author,
|
||
Source: source,
|
||
})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func snapshotMergedCandidates(merged map[string]*ScanCandidate, order []string, applyFinalFilter bool, ctx PostScanContext) []ScanCandidate {
|
||
out := make([]ScanCandidate, 0, len(order))
|
||
for _, key := range order {
|
||
item := merged[key]
|
||
finalizeScanCandidate(item, ctx)
|
||
if applyFinalFilter && !passesFinalStrictFilter(item) {
|
||
continue
|
||
}
|
||
out = append(out, *item)
|
||
}
|
||
if applyFinalFilter && len(out) == 0 && len(order) > 0 {
|
||
for _, key := range order {
|
||
item := merged[key]
|
||
finalizeScanCandidate(item, ctx)
|
||
if !PassesRelaxedFinalFilters(item.Text, item.ProductFitScore, ctx) {
|
||
continue
|
||
}
|
||
out = append(out, *item)
|
||
}
|
||
}
|
||
if applyFinalFilter && len(out) == 0 && len(order) > 0 {
|
||
for _, key := range order {
|
||
item := merged[key]
|
||
finalizeScanCandidate(item, ctx)
|
||
if !PassesIngestFallbackFilters(item.Text, item.ProductFitScore, ctx) {
|
||
continue
|
||
}
|
||
out = append(out, *item)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func passesFinalStrictFilter(item *ScanCandidate) bool {
|
||
if item == nil {
|
||
return false
|
||
}
|
||
if item.ProductFitScore < minPostProductFitPatrol && item.Priority != "gold" {
|
||
return false
|
||
}
|
||
return item.SolvedByProduct
|
||
}
|
||
|
||
func finalizeScanCandidate(item *ScanCandidate, ctx PostScanContext) {
|
||
if item == nil {
|
||
return
|
||
}
|
||
if item.HasRelevance && item.HasRecency && item.ProductFitScore >= 45 {
|
||
item.Priority = "gold"
|
||
} else if item.HasRecency {
|
||
item.Priority = "recent"
|
||
} else {
|
||
item.Priority = "relevant"
|
||
}
|
||
var engP, semP, qualP *int
|
||
if item.EngagementPredicted > 0 {
|
||
v := item.EngagementPredicted
|
||
engP = &v
|
||
}
|
||
if item.SemanticScore > 0 {
|
||
v := item.SemanticScore
|
||
semP = &v
|
||
}
|
||
if item.AudienceQualityScore > 0 {
|
||
v := item.AudienceQualityScore
|
||
qualP = &v
|
||
}
|
||
item.PlacementScore = computePlacementScore(item.Text, item.ProductFitScore, item.HasRecency, engP, semP, qualP)
|
||
item.SolvedByProduct = PostSolvedByProduct(item.Text, item.ProductFitScore, ctx)
|
||
}
|
||
|
||
func computePlacementScore(text string, productFit int, recent bool, predictedEngagement *int, semanticScore *int, audienceQuality *int) int {
|
||
score := 30 + productFit/4
|
||
if HasPlacementIntent(text) {
|
||
score += 20
|
||
}
|
||
if LooksLikeRecommendationPost(text) {
|
||
score += 12
|
||
}
|
||
if recent {
|
||
score += 15
|
||
}
|
||
if productFit >= 60 {
|
||
score += 8
|
||
}
|
||
if predictedEngagement != nil && *predictedEngagement > 60 {
|
||
score += 10
|
||
}
|
||
if semanticScore != nil && *semanticScore > 50 {
|
||
score += 10
|
||
}
|
||
if audienceQuality != nil && *audienceQuality > 60 {
|
||
score += 5
|
||
}
|
||
if score > 100 {
|
||
return 100
|
||
}
|
||
return score
|
||
}
|
||
|
||
func computeAudienceQuality(followerCount int, authorVerified bool, likeCount int, replyCount int) int {
|
||
score := 30
|
||
if authorVerified {
|
||
score += 20
|
||
}
|
||
if followerCount > 10000 {
|
||
score += 15
|
||
} else if followerCount > 1000 {
|
||
score += 10
|
||
} else if followerCount > 100 {
|
||
score += 5
|
||
}
|
||
totalEngagement := likeCount + replyCount*3
|
||
if followerCount > 0 && totalEngagement > 0 {
|
||
engagementRate := (totalEngagement * 100) / followerCount
|
||
if engagementRate > 5 {
|
||
score += 15
|
||
} else if engagementRate > 2 {
|
||
score += 10
|
||
} else if engagementRate > 1 {
|
||
score += 5
|
||
}
|
||
}
|
||
if score > 100 {
|
||
return 100
|
||
}
|
||
return score
|
||
}
|
||
|
||
func max(a, b int) int {
|
||
if a > b {
|
||
return a
|
||
}
|
||
return b
|
||
}
|
||
|
||
func recencyDaysForCandidate(tq TagQuery) int {
|
||
if tq.Dimension != QueryRecency {
|
||
return 0
|
||
}
|
||
if tq.RecencyDays > 0 {
|
||
return tq.RecencyDays
|
||
}
|
||
return IdealMaxPostAgeDays
|
||
}
|
||
|
||
func mergeRecencyDays(existing, incoming int) int {
|
||
if incoming <= 0 {
|
||
return existing
|
||
}
|
||
if existing <= 0 {
|
||
return incoming
|
||
}
|
||
if incoming < existing {
|
||
return incoming
|
||
}
|
||
return existing
|
||
}
|
||
|
||
func previousRecencyWindow(days int) int {
|
||
prev := IdealMaxPostAgeDays
|
||
for _, window := range RecencyCascadeDays() {
|
||
if window == days {
|
||
return prev
|
||
}
|
||
prev = window
|
||
}
|
||
return IdealMaxPostAgeDays
|
||
}
|
||
|
||
func perTagDiscoverLimit(totalQueries int, dimension QueryDimension) int {
|
||
limit := relevanceLimitPerTag
|
||
if dimension == QueryRecency {
|
||
limit = recencyLimitPerTag
|
||
}
|
||
switch {
|
||
case totalQueries > 18:
|
||
return max(7, limit-3)
|
||
case totalQueries > 14:
|
||
return max(8, limit-2)
|
||
default:
|
||
return limit
|
||
}
|
||
}
|
||
|
||
func politeDiscoverPause(ctx context.Context) error {
|
||
wait := 2*time.Second + jitterDuration(2*time.Second)
|
||
timer := time.NewTimer(wait)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-timer.C:
|
||
return nil
|
||
}
|
||
}
|