347 lines
12 KiB
Go
347 lines
12 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"cmautobuy/admin/model"
|
|
"cmautobuy/admin/repository"
|
|
)
|
|
|
|
const MaxAIMatchBatchItems = 100
|
|
|
|
type AIMatchBatchView struct {
|
|
BatchID string `json:"batch_id"`
|
|
Status string `json:"status"`
|
|
StatusText string `json:"status_text"`
|
|
ProviderName string `json:"provider_name"`
|
|
Model string `json:"model"`
|
|
ErrorMessage string `json:"error_message"`
|
|
Total int `json:"total"`
|
|
Processed int `json:"processed"`
|
|
Success int `json:"success"`
|
|
Reused int `json:"reused"`
|
|
Manual int `json:"manual"`
|
|
Failed int `json:"failed"`
|
|
CreatedAt string `json:"created_at"`
|
|
StartedAt string `json:"started_at"`
|
|
FinishedAt string `json:"finished_at"`
|
|
Running bool `json:"running"`
|
|
Items []AIMatchBatchItemView `json:"items"`
|
|
}
|
|
|
|
type AIMatchBatchItemView struct {
|
|
SybID string `json:"syb_id"`
|
|
Status string `json:"status"`
|
|
StatusText string `json:"status_text"`
|
|
Outcome string `json:"outcome"`
|
|
Message string `json:"message"`
|
|
SourceText string `json:"source_text"`
|
|
OptionKey string `json:"option_key"`
|
|
ConfidenceText string `json:"confidence_text"`
|
|
FinishedAt string `json:"finished_at"`
|
|
}
|
|
|
|
func CreateAIMatchBatch(db *sql.DB, actor *model.User, sybIDs []string, snapshot AIMatchSnapshot, now time.Time) (*model.AIMatchBatch, error) {
|
|
if actor == nil || !actor.IsActive() {
|
|
return nil, ErrUnauthenticated
|
|
}
|
|
ids := make([]string, 0, len(sybIDs))
|
|
seen := map[string]bool{}
|
|
for _, raw := range sybIDs {
|
|
id := strings.TrimSpace(raw)
|
|
if id == "" || seen[id] {
|
|
continue
|
|
}
|
|
if utf8.RuneCountInString(id) > 191 {
|
|
return nil, invalidFieldInput("ids", "货运单明细编号过长")
|
|
}
|
|
seen[id] = true
|
|
ids = append(ids, id)
|
|
}
|
|
if len(ids) == 0 {
|
|
return nil, invalidFieldInput("ids", "请先勾选要匹配的顺运宝商品")
|
|
}
|
|
if len(ids) > MaxAIMatchBatchItems {
|
|
return nil, invalidFieldInput("ids", "一次最多匹配 %d 条商品", MaxAIMatchBatchItems)
|
|
}
|
|
if snapshot.Provider.ProviderID == "" || snapshot.Client == nil || snapshot.Secret == "" {
|
|
return nil, fmt.Errorf("AI 服务商运行快照不可用")
|
|
}
|
|
batchID, err := randomID("AIM-", 16)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
createdAt := now.UTC().Format(model.TimeLayout)
|
|
batch := &model.AIMatchBatch{BatchID: batchID, Status: "queued", CreatedByUserID: actor.UserID,
|
|
TotalCount: len(ids), ProviderID: snapshot.Provider.ProviderID, ProviderName: snapshot.Provider.Name,
|
|
ProviderBaseURL: snapshot.Provider.BaseURL, Model: snapshot.Provider.Model,
|
|
TimeoutSeconds: snapshot.Provider.TimeoutSeconds, MaxConcurrency: snapshot.Provider.MaxConcurrency,
|
|
ConfidenceThresholdBPS: snapshot.Provider.ConfidenceThresholdBPS,
|
|
ConfigFingerprint: snapshot.ConfigFingerprint, RulesVersion: SpecMatchRulesVersion,
|
|
PromptVersion: AISpecMatchPromptVersion, CreatedAt: createdAt}
|
|
tx, err := db.Begin()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
leaders := map[string]string{}
|
|
items := make([]model.AIMatchBatchItem, 0, len(ids))
|
|
for position, sybID := range ids {
|
|
orderContext, err := repository.GetSybOrderContext(tx, sybID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if orderContext == nil {
|
|
return nil, invalidFieldInput("ids", "顺运宝明细 %s 已不存在，请刷新页面后重试", sybID)
|
|
}
|
|
identity := aiSpecMatchIdentity(*orderContext)
|
|
leader := leaders[identity]
|
|
if leader == "" {
|
|
leader, leaders[identity] = sybID, sybID
|
|
}
|
|
itemID, err := randomID("AII-", 16)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, model.AIMatchBatchItem{ItemID: itemID, BatchID: batchID, SybID: sybID,
|
|
IdentityHash: identity, LeaderSybID: leader, ContextVersion: mappingContextVersion(*orderContext),
|
|
Position: position + 1, Status: "queued"})
|
|
}
|
|
if err := repository.InsertAIMatchBatch(tx, *batch, items); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
return batch, nil
|
|
}
|
|
|
|
func aiSpecMatchIdentity(c repository.SybOrderContext) string {
|
|
payload, _ := json.Marshal([]string{c.Order.ShopeeGoodsID, c.Order.SpecKey, c.Order.ProductSpec,
|
|
c.PddGoodsID, c.PddUpdatedAt, c.PddSkusJSON, SpecMatchRulesVersion})
|
|
return fmt.Sprintf("%x", sha256.Sum256(payload))
|
|
}
|
|
|
|
func RunAIMatchBatch(ctx context.Context, db *sql.DB, actor model.User, batchID string, snapshot AIMatchSnapshot) error {
|
|
if err := repository.MarkAIMatchBatchRunning(db, batchID, model.NowISO()); err != nil {
|
|
return err
|
|
}
|
|
leaders, err := repository.ListAIMatchLeaderItems(db, batchID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
concurrency := snapshot.Provider.MaxConcurrency
|
|
if concurrency < 1 {
|
|
concurrency = 1
|
|
}
|
|
if concurrency > 16 {
|
|
concurrency = 16
|
|
}
|
|
jobs := make(chan model.AIMatchBatchItem)
|
|
var wg sync.WaitGroup
|
|
var firstErr error
|
|
var errMu sync.Mutex
|
|
for i := 0; i < concurrency; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for leader := range jobs {
|
|
if err := runAIMatchIdentity(ctx, db, actor, batchID, leader, snapshot); err != nil {
|
|
errMu.Lock()
|
|
if firstErr == nil {
|
|
firstErr = err
|
|
}
|
|
errMu.Unlock()
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
for _, leader := range leaders {
|
|
jobs <- leader
|
|
}
|
|
close(jobs)
|
|
wg.Wait()
|
|
if firstErr != nil {
|
|
return firstErr
|
|
}
|
|
counts, err := repository.CountAIMatchBatchItems(db, batchID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
batch, err := repository.GetAIMatchBatch(db, batchID)
|
|
if err != nil || batch == nil {
|
|
return fmt.Errorf("完成 AI 匹配批次前无法读取批次: %w", err)
|
|
}
|
|
status := "succeeded"
|
|
if counts.Failed == batch.TotalCount {
|
|
status = "failed"
|
|
} else if counts.Manual > 0 || counts.Failed > 0 || counts.Processed < batch.TotalCount {
|
|
status = "partial"
|
|
}
|
|
return repository.FinishAIMatchBatch(db, batchID, status, "", model.NowISO(), counts)
|
|
}
|
|
|
|
func runAIMatchIdentity(ctx context.Context, db *sql.DB, actor model.User, batchID string, leader model.AIMatchBatchItem, snapshot AIMatchSnapshot) error {
|
|
startedAt := model.NowISO()
|
|
if err := repository.MarkAIMatchItemRunning(db, leader.ItemID, startedAt); err != nil {
|
|
return err
|
|
}
|
|
result, matchErr := MatchSybSpecWithAI(ctx, db, &actor, snapshot, leader.SybID, leader.ContextVersion)
|
|
leaderResult := batchItemFromMatch(leader, result, matchErr)
|
|
leaderResult.StartedAt, leaderResult.FinishedAt = startedAt, model.NowISO()
|
|
if err := repository.FinishAIMatchItem(db, leaderResult); err != nil {
|
|
return err
|
|
}
|
|
items, err := repository.ListAIMatchIdentityItems(db, batchID, leader.IdentityHash)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, item := range items {
|
|
if item.ItemID == leader.ItemID || item.Status != "queued" {
|
|
continue
|
|
}
|
|
follower := reuseAIMatchLeaderResult(db, item, leaderResult)
|
|
follower.FinishedAt = model.NowISO()
|
|
if err := repository.FinishAIMatchItem(db, follower); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
counts, err := repository.CountAIMatchBatchItems(db, batchID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return repository.UpdateAIMatchBatchCounts(db, batchID, counts)
|
|
}
|
|
|
|
func reuseAIMatchLeaderResult(db *sql.DB, item, leader model.AIMatchBatchItem) model.AIMatchBatchItem {
|
|
current, err := repository.GetSybOrderContext(db, item.SybID)
|
|
if err != nil || current == nil || aiSpecMatchIdentity(*current) != item.IdentityHash {
|
|
item.Status, item.Outcome, item.Message = "stale", "stale", "同规格处理期间数据已变化，请刷新后重试"
|
|
return item
|
|
}
|
|
item.Status, item.Outcome = leader.Status, leader.Outcome
|
|
item.Message = "同一商品规格复用批次结果：" + leader.Message
|
|
item.MappingSource, item.OptionKey = leader.MappingSource, leader.OptionKey
|
|
item.ConfidenceBPS, item.ConfidenceSet = leader.ConfidenceBPS, leader.ConfidenceSet
|
|
if leader.Status == "succeeded" || leader.Status == "reused" {
|
|
if !mappingIsValid(*current) {
|
|
item.Status, item.Outcome, item.Message = "stale", "stale", "批次结果保存后当前映射已失效，请重试"
|
|
item.MappingSource, item.OptionKey, item.ConfidenceSet = "", "", false
|
|
return item
|
|
}
|
|
item.Status, item.Outcome = "reused", "reused"
|
|
item.MappingSource, item.OptionKey = current.MappingSource, current.MappingOptionKey
|
|
if item.MappingSource == "" {
|
|
item.MappingSource = "manual"
|
|
}
|
|
item.ConfidenceBPS, item.ConfidenceSet = current.MappingConfidenceBPS, current.MappingConfidenceSet
|
|
}
|
|
return item
|
|
}
|
|
|
|
func batchItemFromMatch(item model.AIMatchBatchItem, result AISpecMatchResult, err error) model.AIMatchBatchItem {
|
|
item.Outcome, item.Message = result.Outcome, truncateRunes(result.Message, 500)
|
|
item.MappingSource, item.OptionKey = result.Source, result.OptionKey
|
|
item.ConfidenceBPS, item.ConfidenceSet = result.ConfidenceBPS, result.ConfidenceSet
|
|
item.ModelCalled = result.ModelCalled
|
|
if err != nil {
|
|
item.Status, item.Outcome, item.Message = "failed", "failed", truncateRunes(err.Error(), 500)
|
|
return item
|
|
}
|
|
switch result.Outcome {
|
|
case "ai_saved", "rule_saved":
|
|
item.Status = "succeeded"
|
|
case "reused", "manual_exists":
|
|
item.Status = "reused"
|
|
case "stale":
|
|
item.Status = "stale"
|
|
case "failed":
|
|
item.Status = "failed"
|
|
default:
|
|
item.Status = "manual"
|
|
}
|
|
return item
|
|
}
|
|
|
|
func InterruptRunningAIMatchBatches(db *sql.DB, now time.Time) (int, error) {
|
|
ids, err := repository.ListUnfinishedAIMatchBatchIDs(db)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
for _, id := range ids {
|
|
if err := repository.InterruptAIMatchBatch(db, id, now.UTC().Format(model.TimeLayout)); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
return len(ids), nil
|
|
}
|
|
|
|
func MarkAIMatchBatchInterrupted(db *sql.DB, batchID string) error {
|
|
return repository.InterruptAIMatchBatch(db, batchID, model.NowISO())
|
|
}
|
|
|
|
func GetAIMatchBatchView(db *sql.DB, actor *model.User, batchID string) (*AIMatchBatchView, error) {
|
|
if actor == nil || !actor.IsActive() {
|
|
return nil, ErrUnauthenticated
|
|
}
|
|
batch, err := repository.GetAIMatchBatch(db, strings.TrimSpace(batchID))
|
|
if err != nil || batch == nil {
|
|
return nil, err
|
|
}
|
|
if !actor.IsAdmin() && batch.CreatedByUserID != actor.UserID {
|
|
return nil, ErrForbidden
|
|
}
|
|
items, err := repository.ListAIMatchBatchItems(db, batch.BatchID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
view := &AIMatchBatchView{BatchID: batch.BatchID, Status: batch.Status,
|
|
StatusText: aiMatchBatchStatusText(batch.Status), ProviderName: batch.ProviderName, Model: batch.Model,
|
|
ErrorMessage: batch.ErrorMessage, Total: batch.TotalCount, Processed: batch.ProcessedCount,
|
|
Success: batch.SuccessCount, Reused: batch.ReusedCount, Manual: batch.ManualCount,
|
|
Failed: batch.FailedCount, CreatedAt: batch.CreatedAt, StartedAt: batch.StartedAt,
|
|
FinishedAt: batch.FinishedAt, Running: batch.Status == "queued" || batch.Status == "running",
|
|
Items: make([]AIMatchBatchItemView, 0, len(items))}
|
|
for _, item := range items {
|
|
confidence := ""
|
|
if item.ConfidenceSet {
|
|
confidence = fmt.Sprintf("%.2f%%", float64(item.ConfidenceBPS)/100)
|
|
}
|
|
view.Items = append(view.Items, AIMatchBatchItemView{SybID: item.SybID, Status: item.Status,
|
|
StatusText: aiMatchItemStatusText(item.Status), Outcome: item.Outcome,
|
|
Message: item.Message, SourceText: mappingSourceText(item.MappingSource), OptionKey: item.OptionKey,
|
|
ConfidenceText: confidence, FinishedAt: item.FinishedAt})
|
|
}
|
|
return view, nil
|
|
}
|
|
|
|
func LatestAIMatchBatchID(db *sql.DB, actor *model.User) (string, error) {
|
|
if actor == nil || !actor.IsActive() {
|
|
return "", ErrUnauthenticated
|
|
}
|
|
return repository.GetLatestAIMatchBatchID(db, actor.UserID, actor.IsAdmin())
|
|
}
|
|
|
|
func aiMatchBatchStatusText(status string) string {
|
|
return map[string]string{"queued": "等待中", "running": "匹配中", "partial": "部分完成",
|
|
"succeeded": "已完成", "failed": "失败", "interrupted": "已中断"}[status]
|
|
}
|
|
|
|
func aiMatchItemStatusText(status string) string {
|
|
return map[string]string{"queued": "等待中", "running": "匹配中", "succeeded": "已保存",
|
|
"reused": "已复用", "manual": "需人工", "failed": "失败", "stale": "数据已变化",
|
|
"interrupted": "已中断"}[status]
|
|
}
|
|
|
|
func mappingSourceText(source string) string {
|
|
return map[string]string{"manual": "人工匹配", "rule": "规则匹配", "ai": "AI匹配"}[source]
|
|
}
|