243 lines
7.6 KiB
Go
243 lines
7.6 KiB
Go
package usecase
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"apps/backend/internal/module/radar/domain"
|
||
)
|
||
|
||
// SweepRunResult is the outcome of one radar_sweep job execution.
|
||
type SweepRunResult struct {
|
||
Sweep *domain.RadarSweep
|
||
Created int
|
||
Rematched int
|
||
Judged int
|
||
Truncated int
|
||
FailedJudges int
|
||
FetchFailed bool
|
||
FailedReason string
|
||
}
|
||
|
||
// Notifier sends in-app alerts for sweep failures.
|
||
type SweepNotifier interface {
|
||
NotifySweepFailed(ctx context.Context, ownerUID int64, sweepID, watchID, reason string) error
|
||
}
|
||
|
||
/*
|
||
RunSweep is the full pipeline: create/reuse Sweep → fetch → judge → persist → finish.
|
||
|
||
Fetch failure: Sweep failed_reason set, job should fail, notify user — never empty success.
|
||
Partial judge failures: Sweep still succeeds with counters.
|
||
Resume: skip external_ids already on the Sweep record.
|
||
*/
|
||
func (s *Service) RunSweep(ctx context.Context, ownerUID int64, watchID, jobID string) (*SweepRunResult, error) {
|
||
if ownerUID <= 0 || watchID == "" {
|
||
return nil, fmt.Errorf("%w: owner_uid and watch_id required", domain.ErrValidation)
|
||
}
|
||
w, err := s.Repo.GetWatch(ctx, watchID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if w.OwnerUID != ownerUID {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
if w.Status != domain.WatchActive {
|
||
return nil, fmt.Errorf("%w: only active watches can be swept (status=%s)", domain.ErrValidation, w.Status)
|
||
}
|
||
|
||
var profile *domain.ServiceProfile
|
||
var productContext *ProductContextSnapshot
|
||
if w.ContextMode == domain.WatchContextProduct {
|
||
productContext, err = s.LoadProductContext(ctx, ownerUID, w.BrandID, w.ProductID)
|
||
if err != nil {
|
||
reason := "產品上下文不可用，已暫停巡邏：" + err.Error()
|
||
_, _ = s.PauseProductUnavailable(ctx, ownerUID, watchID, domain.PauseReasonProductUnavailable)
|
||
_ = s.notifySweepFailed(ctx, ownerUID, "", watchID, reason)
|
||
return nil, err
|
||
}
|
||
// Product watches do not require a service profile; it remains optional
|
||
// for regional/freshness context when present.
|
||
profile, _ = s.Repo.GetServiceProfile(ctx, ownerUID)
|
||
} else {
|
||
profile, _ = s.Repo.GetServiceProfile(ctx, ownerUID)
|
||
}
|
||
|
||
// Resume: if a sweep already exists for this job, reuse it.
|
||
var sw *domain.RadarSweep
|
||
if jobID != "" {
|
||
if existing, gerr := s.Repo.GetSweepByJobID(ctx, jobID); gerr == nil && existing != nil {
|
||
sw = existing
|
||
}
|
||
}
|
||
if sw == nil {
|
||
sw, err = s.BeginSweepRecord(ctx, ownerUID, watchID, jobID, domain.SweepPathAPI)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
already := map[string]bool{}
|
||
for _, id := range sw.JudgedExternalIDs {
|
||
already[id] = true
|
||
}
|
||
|
||
fetchWatch := mergeFetchWatch(w, nil)
|
||
if w.ContextMode == domain.WatchContextProduct {
|
||
if plan, perr := s.BuildProductQueryPlan(ctx, ownerUID, w.ProductID); perr == nil && plan != nil && len(plan.Groups) > 0 {
|
||
fetchWatch = mergeFetchWatch(w, plan)
|
||
}
|
||
}
|
||
cands, path, fetchCredits, ferr := s.FetchCandidates(ctx, ownerUID, fetchWatch, 40)
|
||
if ferr != nil {
|
||
reason := humanFetchError(ferr)
|
||
end := domain.NowNano()
|
||
_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
|
||
FailedReason: &reason,
|
||
EndedAt: end,
|
||
CreditsUsed: fetchCredits,
|
||
})
|
||
_ = s.notifySweepFailed(ctx, ownerUID, sw.ID, watchID, reason)
|
||
sw, _ = s.Repo.GetSweep(ctx, sw.ID)
|
||
return &SweepRunResult{Sweep: sw, FetchFailed: true, FailedReason: reason}, ferr
|
||
}
|
||
|
||
rawHitCount := len(cands)
|
||
filtered, prefilter := PrefilterCandidates(cands, w, productContext)
|
||
cands = filtered
|
||
_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
|
||
HitCount: rawHitCount,
|
||
CreditsUsed: fetchCredits,
|
||
CreditSearch: fetchCredits,
|
||
DedupedCount: prefilter.Deduped,
|
||
PrefilterPassCount: prefilter.Pass,
|
||
PrefilterReviewCount: prefilter.Review,
|
||
PrefilterRejectedCount: prefilter.Rejected,
|
||
})
|
||
// set path on record
|
||
if path != "" && sw.Path != path {
|
||
sw.Path = path
|
||
// path is not in SweepDelta; re-save via create is wrong — store via failed empty update isn't enough.
|
||
// Use UpdateSweep only for counters; path was set at Begin — recreate if needed.
|
||
if sw.Path == domain.SweepPathAPI && path == domain.SweepPathCrawler {
|
||
// best-effort: include in failed_reason empty path note not needed; set via full get/update memory
|
||
_ = s.setSweepPath(ctx, sw.ID, path)
|
||
}
|
||
}
|
||
|
||
created, rematched, judged, truncated, failed, judgeCredits, perr := s.ProcessCandidates(
|
||
ctx, ownerUID, w, profile, sw.ID, cands, already,
|
||
)
|
||
if perr != nil {
|
||
reason := "判定流程失敗：" + perr.Error()
|
||
end := domain.NowNano()
|
||
_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{FailedReason: &reason, EndedAt: end})
|
||
_ = s.notifySweepFailed(ctx, ownerUID, sw.ID, watchID, reason)
|
||
return nil, perr
|
||
}
|
||
|
||
end := domain.NowNano()
|
||
var failPtr *string
|
||
if failed > 0 && created == 0 && judged > 0 {
|
||
r := fmt.Sprintf("%d 筆判定失敗", failed)
|
||
failPtr = &r
|
||
}
|
||
sw, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
|
||
CreditsUsed: judgeCredits,
|
||
EndedAt: end,
|
||
FailedReason: failPtr,
|
||
})
|
||
_ = s.Repo.TouchWatchSweptAt(ctx, watchID, end)
|
||
|
||
return &SweepRunResult{
|
||
Sweep: sw,
|
||
Created: created,
|
||
Rematched: rematched,
|
||
Judged: judged,
|
||
Truncated: truncated,
|
||
FailedJudges: failed,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) setSweepPath(ctx context.Context, id, path string) error {
|
||
sw, err := s.Repo.GetSweep(ctx, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
sw.Path = path
|
||
// Memory/mongo lack ReplaceSweep — use UpdateSweep no-op + store path only on create.
|
||
// For mongo, UpdateOne $set path:
|
||
type pathSetter interface {
|
||
SetSweepPath(ctx context.Context, id, path string) error
|
||
}
|
||
if ps, ok := s.Repo.(pathSetter); ok {
|
||
return ps.SetSweepPath(ctx, id, path)
|
||
}
|
||
// memory: mutate via UpdateSweep zero + re-get won't change path; patch memory map if possible
|
||
if m, ok := s.Repo.(interface {
|
||
PatchSweepPath(ctx context.Context, id, path string) error
|
||
}); ok {
|
||
return m.PatchSweepPath(ctx, id, path)
|
||
}
|
||
_ = sw
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) notifySweepFailed(ctx context.Context, ownerUID int64, sweepID, watchID, reason string) error {
|
||
if s.Notifier == nil {
|
||
return nil
|
||
}
|
||
return s.Notifier.NotifySweepFailed(ctx, ownerUID, sweepID, watchID, reason)
|
||
}
|
||
|
||
// mergeFetchWatch keeps the subscriber's own keywords and adds demand-map
|
||
// queries. Replacing the watch terms with only plan groups made patrols
|
||
// miss the phrases the user actually typed.
|
||
func mergeFetchWatch(w *domain.RadarWatch, plan *domain.QueryPlan) *domain.RadarWatch {
|
||
if w == nil {
|
||
return nil
|
||
}
|
||
planned := *w
|
||
terms := append([]string{}, w.Terms...)
|
||
excludes := append([]string{}, w.ExcludeTerms...)
|
||
seen := map[string]bool{}
|
||
for _, t := range terms {
|
||
seen[strings.ToLower(strings.TrimSpace(t))] = true
|
||
}
|
||
if plan != nil {
|
||
for _, group := range plan.Groups {
|
||
q := strings.TrimSpace(group.Query)
|
||
if q != "" && !seen[strings.ToLower(q)] {
|
||
terms = append(terms, q)
|
||
seen[strings.ToLower(q)] = true
|
||
}
|
||
excludes = append(excludes, group.Exclude...)
|
||
}
|
||
}
|
||
planned.Terms = domain.ExpandSearchTerms(terms, domain.MaxWatchTerms)
|
||
planned.ExcludeTerms = excludes
|
||
return &planned
|
||
}
|
||
|
||
func humanFetchError(err error) string {
|
||
if err == nil {
|
||
return "抓取失敗"
|
||
}
|
||
msg := err.Error()
|
||
low := strings.ToLower(msg)
|
||
// never include token-like blobs
|
||
if strings.Contains(low, "bearer ") {
|
||
return "抓取路徑不可用"
|
||
}
|
||
if strings.Contains(low, "crawler session expired") ||
|
||
strings.Contains(low, "crawler session is invalid") ||
|
||
strings.Contains(low, "crawler session required") {
|
||
return "今天沒巡到：Chrome 登入已過期。請到設定重新同步已登入的 Threads 分頁，或先關掉開發模式改走 API 搜尋。"
|
||
}
|
||
if len(msg) > 200 {
|
||
msg = msg[:200]
|
||
}
|
||
return "今天沒巡到：" + msg
|
||
}
|