262 lines
10 KiB
Go
262 lines
10 KiB
Go
package multitable
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strconv"
|
|
"time"
|
|
|
|
"git.iwork-ai.com/xdc/iqudo-top1/internal/config"
|
|
"git.iwork-ai.com/xdc/iqudo-top1/internal/model"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type Projector struct {
|
|
db *gorm.DB
|
|
client *Client
|
|
tables config.MultiTableTables
|
|
}
|
|
|
|
func NewProjector(db *gorm.DB, client *Client, tables config.MultiTableTables) *Projector {
|
|
return &Projector{db: db, client: client, tables: tables}
|
|
}
|
|
|
|
func (p *Projector) Project(ctx context.Context, event model.MultiTableOutbox) error {
|
|
switch event.Resource {
|
|
case "scenario":
|
|
return p.scenario(ctx, event)
|
|
case "scenario_field":
|
|
return p.scenarioField(ctx, event)
|
|
case "scenario_rule":
|
|
return p.scenarioRule(ctx, event)
|
|
case "sop":
|
|
return p.sop(ctx, event)
|
|
case "sop_run":
|
|
if err := p.run(ctx, event); err != nil {
|
|
return err
|
|
}
|
|
if event.Action == "feedback" {
|
|
return p.feedback(ctx, event)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *Projector) scenario(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var item model.Scenario
|
|
if err := p.db.Where("id = ? AND tenant_id = ?", event.ResourceID, event.TenantID).First(&item).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := p.client.Upsert(ctx, p.tables.Scenarios, id(item.ID), common(item.ID, item.TenantID, statusFor(item.Status), item.UpdatedAt, map[string]interface{}{
|
|
"名称": item.Name, "行业": item.Industry, "适用角色": item.RoleName, "目标": clip(item.Goal), "触发条件": clip(item.TriggerText), "可见范围": item.Visibility, "业务状态": item.Status, "创建人ID": id(item.CreatedBy),
|
|
})); err != nil {
|
|
return err
|
|
}
|
|
if item.Status != "archived" {
|
|
return nil
|
|
}
|
|
fields := make([]model.ScenarioField, 0)
|
|
if err := p.db.Where("scenario_id = ? AND tenant_id = ?", item.ID, item.TenantID).Find(&fields).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, field := range fields {
|
|
if err := p.projectScenarioField(ctx, field, "已归档"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
sops := make([]model.SOP, 0)
|
|
if err := p.db.Where("scenario_id = ? AND tenant_id = ?", item.ID, item.TenantID).Find(&sops).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, sop := range sops {
|
|
if err := p.sop(ctx, model.MultiTableOutbox{TenantID: item.TenantID, ResourceID: sop.ID}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *Projector) scenarioField(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var item model.ScenarioField
|
|
if err := p.db.Where("id = ? AND tenant_id = ?", event.ResourceID, event.TenantID).First(&item).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) && event.Action == "delete" {
|
|
return p.deletedScenarioField(ctx, event)
|
|
}
|
|
return err
|
|
}
|
|
return p.projectScenarioField(ctx, item, "正常")
|
|
}
|
|
|
|
func (p *Projector) projectScenarioField(ctx context.Context, item model.ScenarioField, syncStatus string) error {
|
|
return p.client.Upsert(ctx, p.tables.ScenarioFields, id(item.ID), common(item.ID, item.TenantID, syncStatus, item.UpdatedAt, map[string]interface{}{
|
|
"场景来源ID": id(item.ScenarioID), "字段标识": item.FieldKey, "字段名称": item.FieldName, "外部数据路径": item.SourcePath, "字段类型": item.FieldType, "是否必填": yesNo(item.Required), "选项": jsonText(item.Options), "校验规则": jsonText(item.Validation), "排序": item.SortOrder,
|
|
}))
|
|
}
|
|
|
|
func (p *Projector) scenarioRule(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var item model.ScenarioRule
|
|
err := p.db.Where("id = ? AND tenant_id = ?", event.ResourceID, event.TenantID).First(&item).Error
|
|
if err == nil {
|
|
return p.client.Upsert(ctx, p.tables.ScenarioRules, id(item.ID), scenarioRuleProjection(item, "正常"))
|
|
}
|
|
if !errors.Is(err, gorm.ErrRecordNotFound) || event.Action != "archive" {
|
|
return err
|
|
}
|
|
var payload struct {
|
|
ScenarioID uint64 `json:"scenario_id"`
|
|
RuleKey string `json:"rule_key"`
|
|
Name string `json:"name"`
|
|
Priority int `json:"priority"`
|
|
Status string `json:"status"`
|
|
}
|
|
if err := json.Unmarshal(event.Payload, &payload); err != nil {
|
|
return err
|
|
}
|
|
return p.client.Upsert(ctx, p.tables.ScenarioRules, id(event.ResourceID), common(event.ResourceID, event.TenantID, "已归档", event.UpdatedAt, map[string]interface{}{
|
|
"场景来源ID": id(payload.ScenarioID), "规则标识": payload.RuleKey, "名称": payload.Name, "优先级": payload.Priority, "规则状态": payload.Status,
|
|
}))
|
|
}
|
|
|
|
func (p *Projector) deletedScenarioField(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var payload struct {
|
|
ScenarioID uint64 `json:"scenario_id"`
|
|
FieldKey string `json:"field_key"`
|
|
FieldName string `json:"field_name"`
|
|
SourcePath string `json:"source_path"`
|
|
FieldType string `json:"field_type"`
|
|
Required bool `json:"required"`
|
|
Options json.RawMessage `json:"options"`
|
|
Validation json.RawMessage `json:"validation"`
|
|
SortOrder int `json:"sort_order"`
|
|
}
|
|
if err := json.Unmarshal(event.Payload, &payload); err != nil {
|
|
return err
|
|
}
|
|
if payload.ScenarioID == 0 || payload.FieldKey == "" || payload.FieldName == "" || payload.FieldType == "" {
|
|
return errors.New("deleted scenario field audit payload is incomplete")
|
|
}
|
|
item := model.ScenarioField{Base: model.Base{ID: event.ResourceID, UpdatedAt: event.UpdatedAt}, TenantID: event.TenantID, ScenarioID: payload.ScenarioID, FieldKey: payload.FieldKey, FieldName: payload.FieldName, SourcePath: payload.SourcePath, FieldType: payload.FieldType, Required: payload.Required, Options: datatypes.JSON(payload.Options), Validation: datatypes.JSON(payload.Validation), SortOrder: payload.SortOrder}
|
|
return p.projectScenarioField(ctx, item, "已归档")
|
|
}
|
|
|
|
func (p *Projector) sop(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var sop model.SOP
|
|
if err := p.db.Where("id = ? AND tenant_id = ?", event.ResourceID, event.TenantID).First(&sop).Error; err != nil {
|
|
return err
|
|
}
|
|
nodes := make([]model.SOPNode, 0)
|
|
edges := make([]model.SOPEdge, 0)
|
|
if err := p.db.Where("sop_id = ? AND tenant_id = ?", sop.ID, sop.TenantID).Order("id").Find(&nodes).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := p.db.Where("sop_id = ? AND tenant_id = ?", sop.ID, sop.TenantID).Order("priority, id").Find(&edges).Error; err != nil {
|
|
return err
|
|
}
|
|
state := statusFor(sop.Status)
|
|
if sop.Status == "archived" {
|
|
state = "已归档"
|
|
}
|
|
if err := p.client.Upsert(ctx, p.tables.SOPs, id(sop.ID), common(sop.ID, sop.TenantID, state, sop.UpdatedAt, map[string]interface{}{
|
|
"场景来源ID": id(sop.ScenarioID), "SOP名称": sop.Name, "SOP状态": sop.Status, "开始节点": sop.StartNodeKey, "节点数": len(nodes), "路径数": len(edges),
|
|
})); err != nil {
|
|
return err
|
|
}
|
|
for _, node := range nodes {
|
|
if err := p.client.Upsert(ctx, p.tables.SOPNodes, id(node.ID), common(node.ID, node.TenantID, state, node.UpdatedAt, map[string]interface{}{
|
|
"SOP来源ID": id(sop.ID), "节点标识": node.NodeKey, "节点类型": node.Type, "标题": node.Title, "标准话术或操作提示": clip(node.Content), "配置": jsonText(node.Config), "排序": node.PositionY,
|
|
})); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, edge := range edges {
|
|
if err := p.client.Upsert(ctx, p.tables.SOPEdges, id(edge.ID), common(edge.ID, edge.TenantID, state, edge.UpdatedAt, map[string]interface{}{
|
|
"SOP来源ID": id(sop.ID), "起点节点标识": edge.SourceNodeKey, "终点节点标识": edge.TargetNodeKey, "条件": jsonText(edge.Condition), "优先级": edge.Priority,
|
|
})); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *Projector) run(ctx context.Context, event model.MultiTableOutbox) error {
|
|
var row struct {
|
|
model.SOPRun
|
|
SOPName string
|
|
ScenarioName string
|
|
}
|
|
err := p.db.Table("sop_runs r").Select("r.*, s.name AS sop_name, sc.name AS scenario_name").Joins("JOIN sops s ON s.id = r.sop_id").Joins("JOIN scenarios sc ON sc.id = s.scenario_id").Where("r.id = ? AND r.tenant_id = ?", event.ResourceID, event.TenantID).Scan(&row).Error
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if row.ID == 0 {
|
|
return gorm.ErrRecordNotFound
|
|
}
|
|
var answerCount int
|
|
var answers map[string]interface{}
|
|
_ = json.Unmarshal(row.Answers, &answers)
|
|
answerCount = len(answers)
|
|
return p.client.Upsert(ctx, p.tables.Runs, id(row.ID), common(row.ID, row.TenantID, "正常", row.UpdatedAt, map[string]interface{}{
|
|
"SOP来源ID": id(row.SOPID), "场景名称": row.ScenarioName, "执行人ID": id(row.OperatorID), "执行状态": row.Status, "执行结果": row.Result, "最终结果": jsonText(row.FinalResult), "已采集字段数": answerCount, "开始时间": formatTime(&row.StartedAt), "完成时间": formatTime(row.CompletedAt),
|
|
}))
|
|
}
|
|
|
|
func (p *Projector) feedback(ctx context.Context, event model.MultiTableOutbox) error {
|
|
items := make([]model.SOPFeedback, 0)
|
|
if err := p.db.Where("run_id = ? AND tenant_id = ?", event.ResourceID, event.TenantID).Find(&items).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, item := range items {
|
|
if err := p.client.Upsert(ctx, p.tables.Feedback, id(item.ID), common(item.ID, item.TenantID, "正常", item.UpdatedAt, map[string]interface{}{
|
|
"执行来源ID": id(item.RunID), "提交人ID": id(item.UserID), "评分": item.Score, "反馈内容": clip(item.Comment), "提交时间": item.CreatedAt.Format(time.RFC3339),
|
|
})); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func common(sourceID, tenantID uint64, status string, updated time.Time, extra map[string]interface{}) map[string]interface{} {
|
|
data := map[string]interface{}{"来源ID": id(sourceID), "业务租户ID": tenantID, "同步状态": status, "来源更新时间": updated.Format(time.RFC3339)}
|
|
for key, value := range extra {
|
|
data[key] = value
|
|
}
|
|
return data
|
|
}
|
|
|
|
func scenarioRuleProjection(item model.ScenarioRule, status string) map[string]interface{} {
|
|
return common(item.ID, item.TenantID, status, item.UpdatedAt, map[string]interface{}{"场景来源ID": id(item.ScenarioID), "规则标识": item.RuleKey, "名称": item.Name, "优先级": item.Priority, "规则状态": item.Status})
|
|
}
|
|
|
|
func id(value uint64) string { return strconv.FormatUint(value, 10) }
|
|
func formatTime(value *time.Time) string {
|
|
if value == nil {
|
|
return ""
|
|
}
|
|
return value.Format(time.RFC3339)
|
|
}
|
|
func statusFor(value string) string {
|
|
if value == "archived" {
|
|
return "已归档"
|
|
}
|
|
if value == "offline" {
|
|
return "已下线"
|
|
}
|
|
return "正常"
|
|
}
|
|
func jsonText(value []byte) string { return clip(string(value)) }
|
|
func stringValue(value interface{}) string { result, _ := value.(string); return result }
|
|
func clip(value string) string {
|
|
if len(value) > 4000 {
|
|
return value[:4000]
|
|
}
|
|
return value
|
|
}
|
|
func yesNo(value bool) string {
|
|
if value {
|
|
return "是"
|
|
}
|
|
return "否"
|
|
}
|