306 lines
10 KiB
Go
306 lines
10 KiB
Go
package creator
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func encodeRuleLists(input LeadRuleInput) (string, string, error) {
|
|
include, err := jsonText(input.IncludeKeywords)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
exclude, err := jsonText(input.ExcludeKeywords)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
return include, exclude, nil
|
|
}
|
|
|
|
func scanRule(scanner interface{ Scan(...any) error }) (LeadRule, error) {
|
|
var result LeadRule
|
|
var include, exclude []byte
|
|
if err := scanner.Scan(&result.ID, &result.Name, &result.Enabled, &result.SourceType, &result.Topic, &include, &exclude, &result.AIRequirement, &result.UpdatedAt, &result.CreatedAt); err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
var err error
|
|
result.IncludeKeywords, err = decodeStringList(include)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
result.ExcludeKeywords, err = decodeStringList(exclude)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
const ruleSelect = `SELECT id, name, enabled, source_type, topic, include_keywords, exclude_keywords, ai_requirement, updated_at, created_at FROM creator_lead_rule`
|
|
|
|
func (s *Store) CreateRule(ctx context.Context, input LeadRuleInput) (LeadRule, error) {
|
|
input, err := ValidateRuleInput(input)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
include, exclude, err := encodeRuleLists(input)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
id := newID("rule")
|
|
if _, err := s.db.ExecContext(ctx, `INSERT INTO creator_lead_rule (id,name,enabled,source_type,topic,include_keywords,exclude_keywords,ai_requirement) VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7::jsonb,$8)`, id, input.Name, input.Enabled, input.SourceType, input.Topic, include, exclude, input.AIRequirement); err != nil {
|
|
return LeadRule{}, databaseError(err)
|
|
}
|
|
return s.GetRule(ctx, id)
|
|
}
|
|
|
|
func (s *Store) GetRule(ctx context.Context, id string) (LeadRule, error) {
|
|
result, err := scanRule(s.db.QueryRowContext(ctx, ruleSelect+` WHERE id = $1`, id))
|
|
return result, rowError(err)
|
|
}
|
|
|
|
func (s *Store) ListRules(ctx context.Context, enabledOnly bool) ([]LeadRule, error) {
|
|
query := ruleSelect
|
|
if enabledOnly {
|
|
query += ` WHERE enabled`
|
|
}
|
|
query += ` ORDER BY created_at DESC,id`
|
|
rows, err := s.db.QueryContext(ctx, query)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]LeadRule, 0)
|
|
for rows.Next() {
|
|
item, err := scanRule(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UpdateRule(ctx context.Context, id string, input LeadRuleInput) (LeadRule, error) {
|
|
input, err := ValidateRuleInput(input)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
include, exclude, err := encodeRuleLists(input)
|
|
if err != nil {
|
|
return LeadRule{}, err
|
|
}
|
|
_, err = s.db.ExecContext(ctx, `UPDATE creator_lead_rule SET name=$2,enabled=$3,source_type=$4,topic=$5,include_keywords=$6::jsonb,exclude_keywords=$7::jsonb,ai_requirement=$8,updated_at=now() WHERE id=$1`, id, input.Name, input.Enabled, input.SourceType, input.Topic, include, exclude, input.AIRequirement)
|
|
if err != nil {
|
|
return LeadRule{}, databaseError(err)
|
|
}
|
|
return s.GetRule(ctx, id)
|
|
}
|
|
|
|
func (s *Store) SetRuleEnabled(ctx context.Context, id string, enabled bool) (LeadRule, error) {
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE creator_lead_rule SET enabled=$2,updated_at=now() WHERE id=$1`, id, enabled); err != nil {
|
|
return LeadRule{}, databaseError(err)
|
|
}
|
|
return s.GetRule(ctx, id)
|
|
}
|
|
|
|
func (s *Store) upsertRuleResult(ctx context.Context, commentID string, rule LeadRule, status, reason string, matched []string, analysedAt *time.Time) (RuleResult, error) {
|
|
snapshot, err := jsonText(rule)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
matchedJSON, err := jsonText(matched)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
_, err = s.db.ExecContext(ctx, `INSERT INTO creator_comment_rule_result (comment_id,rule_id,status,reason,matched_keywords,rule_snapshot,analysed_at) VALUES ($1,$2,$3,$4,$5::jsonb,$6::jsonb,$7) ON CONFLICT (comment_id,rule_id) DO UPDATE SET status=EXCLUDED.status,reason=EXCLUDED.reason,matched_keywords=EXCLUDED.matched_keywords,rule_snapshot=EXCLUDED.rule_snapshot,analysed_at=EXCLUDED.analysed_at`, commentID, rule.ID, status, reason, matchedJSON, snapshot, analysedAt)
|
|
if err != nil {
|
|
return RuleResult{}, databaseError(err)
|
|
}
|
|
return RuleResult{CommentID: commentID, RuleID: rule.ID, Status: status, Reason: reason, MatchedKeywords: matched, RuleSnapshot: rule, AnalysedAt: analysedAt}, nil
|
|
}
|
|
|
|
func (s *Store) AnalyzeComments(ctx context.Context, commentIDs []string, ruleID string, analyzer ThemeAnalyzer) ([]RuleAnalysisItem, error) {
|
|
if len(commentIDs) == 0 || len(commentIDs) > 100 || strings.TrimSpace(ruleID) == "" {
|
|
return nil, ErrInvalid
|
|
}
|
|
seen := make(map[string]struct{}, len(commentIDs))
|
|
items := make([]RuleAnalysisItem, 0, len(commentIDs))
|
|
var batchErr error
|
|
for _, commentID := range commentIDs {
|
|
commentID = strings.TrimSpace(commentID)
|
|
if commentID == "" {
|
|
return nil, ErrInvalid
|
|
}
|
|
if _, ok := seen[commentID]; ok {
|
|
continue
|
|
}
|
|
seen[commentID] = struct{}{}
|
|
result, err := s.AnalyzeComment(ctx, commentID, ruleID, analyzer)
|
|
item := RuleAnalysisItem{CommentID: commentID, Result: &result}
|
|
if err != nil {
|
|
item.Result = nil
|
|
item.Error = err.Error()
|
|
batchErr = errors.Join(batchErr, err)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, batchErr
|
|
}
|
|
|
|
func (s *Store) AnalyzeComment(ctx context.Context, commentID, ruleID string, analyzer ThemeAnalyzer) (RuleResult, error) {
|
|
comment, err := s.GetComment(ctx, commentID)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
rule, err := s.GetRule(ctx, ruleID)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
if !rule.Enabled {
|
|
return RuleResult{}, ErrConflict
|
|
}
|
|
work, err := s.GetWork(ctx, comment.WorkID)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
if rule.SourceType != "all" {
|
|
matchedSource := false
|
|
for _, source := range work.Sources {
|
|
if source.SourceType == rule.SourceType {
|
|
matchedSource = true
|
|
break
|
|
}
|
|
}
|
|
if !matchedSource && work.SourceType != rule.SourceType {
|
|
return RuleResult{}, ErrConflict
|
|
}
|
|
}
|
|
if analyzer == nil {
|
|
result, saveErr := s.upsertRuleResult(ctx, commentID, rule, "failed", "AI 分析不可用", nil, ptrTime(time.Now().UTC()))
|
|
if saveErr != nil {
|
|
return RuleResult{}, saveErr
|
|
}
|
|
return result, ErrUnavailable
|
|
}
|
|
theme, reason, err := analyzer.MatchTheme(ctx, work.Title, work.Body, rule.Topic)
|
|
if err != nil {
|
|
result, saveErr := s.upsertRuleResult(ctx, commentID, rule, "failed", err.Error(), nil, ptrTime(time.Now().UTC()))
|
|
if saveErr != nil {
|
|
return RuleResult{}, saveErr
|
|
}
|
|
return result, err
|
|
}
|
|
if !theme {
|
|
return s.upsertRuleResult(ctx, commentID, rule, "topic_no_match", strings.TrimSpace(reason), nil, ptrTime(time.Now().UTC()))
|
|
}
|
|
keywordsOK, matched := MatchKeywords(comment.Content, rule.IncludeKeywords, rule.ExcludeKeywords)
|
|
if !keywordsOK {
|
|
return s.upsertRuleResult(ctx, commentID, rule, "keyword_no_match", "关键词不满足", matched, ptrTime(time.Now().UTC()))
|
|
}
|
|
lead, reason, err := analyzer.MatchLead(ctx, work.Title+"\n"+work.Body, comment.Content, rule.AIRequirement)
|
|
if err != nil {
|
|
result, saveErr := s.upsertRuleResult(ctx, commentID, rule, "failed", err.Error(), matched, ptrTime(time.Now().UTC()))
|
|
if saveErr != nil {
|
|
return RuleResult{}, saveErr
|
|
}
|
|
return result, err
|
|
}
|
|
if !lead {
|
|
return s.upsertRuleResult(ctx, commentID, rule, "not_lead", strings.TrimSpace(reason), matched, ptrTime(time.Now().UTC()))
|
|
}
|
|
return s.upsertRuleResult(ctx, commentID, rule, "lead", strings.TrimSpace(reason), matched, ptrTime(time.Now().UTC()))
|
|
}
|
|
|
|
func scanRuleResult(scanner interface{ Scan(...any) error }) (RuleResult, error) {
|
|
var r RuleResult
|
|
var matched, snapshot []byte
|
|
var analysed sql.NullTime
|
|
if err := scanner.Scan(&r.CommentID, &r.RuleID, &r.Status, &r.Reason, &matched, &snapshot, &analysed); err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
var err error
|
|
r.MatchedKeywords, err = decodeStringList(matched)
|
|
if err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
if err := jsonUnmarshal(snapshot, &r.RuleSnapshot); err != nil {
|
|
return RuleResult{}, err
|
|
}
|
|
r.AnalysedAt = nullableTime(analysed)
|
|
return r, nil
|
|
}
|
|
|
|
func jsonUnmarshal(data []byte, value any) error {
|
|
if len(data) == 0 {
|
|
return nil
|
|
}
|
|
return json.Unmarshal(data, value)
|
|
}
|
|
|
|
func (s *Store) ListRuleResults(ctx context.Context, commentID, ruleID string) ([]RuleResult, error) {
|
|
query := `SELECT comment_id,rule_id,status,reason,matched_keywords,rule_snapshot,analysed_at FROM creator_comment_rule_result`
|
|
where := []string{}
|
|
args := []any{}
|
|
if commentID != "" {
|
|
args = append(args, commentID)
|
|
where = append(where, fmt.Sprintf("comment_id = $%d", len(args)))
|
|
}
|
|
if ruleID != "" {
|
|
args = append(args, ruleID)
|
|
where = append(where, fmt.Sprintf("rule_id = $%d", len(args)))
|
|
}
|
|
if len(where) > 0 {
|
|
query += ` WHERE ` + strings.Join(where, ` AND `)
|
|
}
|
|
query += ` ORDER BY analysed_at DESC NULLS LAST,comment_id,rule_id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]RuleResult, 0)
|
|
for rows.Next() {
|
|
item, err := scanRuleResult(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) ListLeads(ctx context.Context, platform string) ([]Lead, error) {
|
|
comments, err := s.ListComments(ctx, platform, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
results, err := s.ListRuleResults(ctx, "", "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
byComment := map[string][]RuleResult{}
|
|
for _, result := range results {
|
|
if result.Status == "lead" {
|
|
byComment[result.CommentID] = append(byComment[result.CommentID], result)
|
|
}
|
|
}
|
|
leads := make([]Lead, 0)
|
|
for _, comment := range comments {
|
|
matched := byComment[comment.ID]
|
|
if len(matched) == 0 {
|
|
continue
|
|
}
|
|
ids := make([]string, 0, len(matched))
|
|
for _, result := range matched {
|
|
ids = append(ids, result.RuleID)
|
|
}
|
|
leads = append(leads, Lead{Comment: comment, RuleIDs: ids, Results: matched})
|
|
}
|
|
return leads, nil
|
|
}
|