refactor: update duty user sync

* doris:support search sql with macro

* Update doris.go

---------

Co-authored-by: Yening Qin <710leo@gmail.com>

alert/dispatch/dispatch.go

@@ -595,6 +595,10 @@ func (e *Dispatch) handleSub(sub *models.AlertSubscribe, event models.AlertCurEv
 		return
 	}
 
+	if !sub.MatchCate(event.Cate) {
+		return
+	}
+
 	if !common.MatchTags(event.TagsMap, sub.ITags) {
 		return
 	}

alert/pipeline/processor/aisummary/ai_summary.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"text/template"
 	"time"
@@ -143,7 +144,11 @@ func (c *AISummaryConfig) generateAISummary(eventInfo string) (string, error) {
 
 	// 合并自定义参数
 	for k, v := range c.CustomParams {
-		reqParams[k] = v
+		converted, err := convertCustomParam(v)
+		if err != nil {
+			return "", fmt.Errorf("failed to convert custom param %s: %v", k, err)
+		}
+		reqParams[k] = converted
 	}
 
 	// 序列化请求体
@@ -196,3 +201,44 @@ func (c *AISummaryConfig) generateAISummary(eventInfo string) (string, error) {
 
 	return chatResp.Choices[0].Message.Content, nil
 }
+
+// convertCustomParam 将前端传入的参数转换为正确的类型
+func convertCustomParam(value interface{}) (interface{}, error) {
+	if value == nil {
+		return nil, nil
+	}
+
+	// 如果是字符串，尝试转换为其他类型
+	if str, ok := value.(string); ok {
+		// 尝试转换为数字
+		if f, err := strconv.ParseFloat(str, 64); err == nil {
+			// 检查是否为整数
+			if f == float64(int64(f)) {
+				return int64(f), nil
+			}
+			return f, nil
+		}
+
+		// 尝试转换为布尔值
+		if b, err := strconv.ParseBool(str); err == nil {
+			return b, nil
+		}
+
+		// 尝试解析为JSON数组
+		if strings.HasPrefix(strings.TrimSpace(str), "[") {
+			var arr []interface{}
+			if err := json.Unmarshal([]byte(str), &arr); err == nil {
+				return arr, nil
+			}
+		}
+
+		// 尝试解析为JSON对象
+		if strings.HasPrefix(strings.TrimSpace(str), "{") {
+			var obj map[string]interface{}
+			if err := json.Unmarshal([]byte(str), &obj); err == nil {
+				return obj, nil
+			}
+		}
+	}
+	return value, nil
+}

alert/pipeline/processor/aisummary/ai_summary_test.go

@@ -67,3 +67,73 @@ func TestAISummaryConfig_Process(t *testing.T) {
 	t.Logf("原始注释: %v", result.AnnotationsJSON["description"])
 	t.Logf("AI总结: %s", result.AnnotationsJSON["ai_summary"])
 }
+
+func TestConvertCustomParam(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    interface{}
+		expected interface{}
+		hasError bool
+	}{
+		{
+			name:     "nil value",
+			input:    nil,
+			expected: nil,
+			hasError: false,
+		},
+		{
+			name:     "string number to int64",
+			input:    "123",
+			expected: int64(123),
+			hasError: false,
+		},
+		{
+			name:     "string float to float64",
+			input:    "123.45",
+			expected: 123.45,
+			hasError: false,
+		},
+		{
+			name:     "string boolean to bool",
+			input:    "true",
+			expected: true,
+			hasError: false,
+		},
+		{
+			name:     "string false to bool",
+			input:    "false",
+			expected: false,
+			hasError: false,
+		},
+		{
+			name:     "JSON array string to slice",
+			input:    `["a", "b", "c"]`,
+			expected: []interface{}{"a", "b", "c"},
+			hasError: false,
+		},
+		{
+			name:     "JSON object string to map",
+			input:    `{"key": "value", "num": 123}`,
+			expected: map[string]interface{}{"key": "value", "num": float64(123)},
+			hasError: false,
+		},
+		{
+			name:     "plain string remains string",
+			input:    "hello world",
+			expected: "hello world",
+			hasError: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			converted, err := convertCustomParam(test.input)
+			if test.hasError {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			assert.Equal(t, test.expected, converted)
+		})
+	}
+}

center/integration/init.go

@@ -187,7 +187,7 @@ func Init(ctx *ctx.Context, builtinIntegrationsDir string) {
 						CreatedBy:   SYSTEM,
 						UpdatedBy:   SYSTEM,
 					}
-					BuiltinPayloadInFile.addBuiltinPayload(&builtinAlert)
+					BuiltinPayloadInFile.AddBuiltinPayload(&builtinAlert)
 
 				}
 			}
@@ -245,7 +245,7 @@ func Init(ctx *ctx.Context, builtinIntegrationsDir string) {
 					CreatedBy:   SYSTEM,
 					UpdatedBy:   SYSTEM,
 				}
-				BuiltinPayloadInFile.addBuiltinPayload(&builtinDashboard)
+				BuiltinPayloadInFile.AddBuiltinPayload(&builtinDashboard)
 			}
 		} else if err != nil {
 			logger.Warningf("read builtin component dash dir fail %s %v", component.Ident, err)
@@ -314,7 +314,7 @@ func NewBuiltinPayloadInFileType() *BuiltinPayloadInFileType {
 	}
 }
 
-func (b *BuiltinPayloadInFileType) addBuiltinPayload(bp *models.BuiltinPayload) {
+func (b *BuiltinPayloadInFileType) AddBuiltinPayload(bp *models.BuiltinPayload) {
 	if _, exists := b.Data[bp.ComponentID]; !exists {
 		b.Data[bp.ComponentID] = make(map[string]map[string][]*models.BuiltinPayload)
 	}

center/router/router_alert_rule.go

@@ -543,6 +543,20 @@ func (rt *Router) alertRulePutFields(c *gin.Context) {
 			}
 		}
 
+		// 检测是否是批量更新通知规则的字段，如果是清理掉旧版本的配置
+		for k := range f.Fields {
+			if k == "notify_rule_ids" {
+				f.Fields["notify_version"] = 1
+				f.Fields["notify_channels"] = ""
+				f.Fields["notify_groups"] = ""
+				f.Fields["callbacks"] = ""
+			}
+
+			if k == "notify_channels" {
+				f.Fields["notify_version"] = 0
+			}
+		}
+
 		for k, v := range f.Fields {
 			// 检查 v 是否为各种切片类型
 			switch v.(type) {

center/router/router_alert_subscribe.go

@@ -288,6 +288,7 @@ func (rt *Router) alertSubscribePut(c *gin.Context) {
 			"busi_groups",
 			"note",
 			"notify_rule_ids",
+			"notify_version",
 		))
 	}
 

center/router/router_recording_rule.go

@@ -149,6 +149,12 @@ func (rt *Router) recordingRulePutFields(c *gin.Context) {
 		f.Fields["datasource_queries"] = string(bytes)
 	}
 
+	if datasourceIds, ok := f.Fields["datasource_ids"]; ok {
+		bytes, err := json.Marshal(datasourceIds)
+		ginx.Dangerous(err)
+		f.Fields["datasource_ids"] = string(bytes)
+	}
+
 	for i := 0; i < len(f.Ids); i++ {
 		ar, err := models.RecordingRuleGetById(rt.Ctx, f.Ids[i])
 		ginx.Dangerous(err)

datasource/doris/doris.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ccfos/nightingale/v6/datasource"
 	"github.com/ccfos/nightingale/v6/dskit/doris"
 	"github.com/ccfos/nightingale/v6/dskit/types"
+	"github.com/ccfos/nightingale/v6/pkg/macros"
 	"github.com/ccfos/nightingale/v6/models"
 
 	"github.com/mitchellh/mapstructure"
@@ -27,11 +28,16 @@ type Doris struct {
 }
 
 type QueryParam struct {
-	Ref      string          `json:"ref" mapstructure:"ref"`
-	Database string          `json:"database" mapstructure:"database"`
-	Table    string          `json:"table" mapstructure:"table"`
-	SQL      string          `json:"sql" mapstructure:"sql"`
-	Keys     datasource.Keys `json:"keys" mapstructure:"keys"`
+	Ref        string          `json:"ref" mapstructure:"ref"`
+	Database   string          `json:"database" mapstructure:"database"`
+	Table      string          `json:"table" mapstructure:"table"`
+	SQL        string          `json:"sql" mapstructure:"sql"`
+	Keys       datasource.Keys `json:"keys" mapstructure:"keys"`
+	Limit      int             `json:"limit" mapstructure:"limit"`
+	From       int64           `json:"from" mapstructure:"from"`
+	To         int64           `json:"to" mapstructure:"to"`
+	TimeField  string          `json:"time_field" mapstructure:"time_field"`
+	TimeFormat string          `json:"time_format" mapstructure:"time_format"`
 }
 
 func (d *Doris) InitClient() error {
@@ -66,7 +72,7 @@ func (d *Doris) Validate(ctx context.Context) error {
 func (d *Doris) Equal(p datasource.Datasource) bool {
 	newest, ok := p.(*Doris)
 	if !ok {
-		logger.Errorf("unexpected plugin type, expected is ck")
+		logger.Errorf("unexpected plugin type, expected is doris")
 		return false
 	}
 
@@ -174,6 +180,14 @@ func (d *Doris) QueryLog(ctx context.Context, query interface{}) ([]interface{},
 		return nil, 0, err
 	}
 
+	if strings.Contains(dorisQueryParam.SQL, "$__") {
+		var err error
+		dorisQueryParam.SQL, err = macros.Macro(dorisQueryParam.SQL, dorisQueryParam.From, dorisQueryParam.To)
+		if err != nil {
+			return nil, 0, err
+		}
+	}
+
 	items, err := d.QueryLogs(ctx, &doris.QueryParam{
 		Database: dorisQueryParam.Database,
 		Sql:      dorisQueryParam.SQL,
@@ -187,7 +201,7 @@ func (d *Doris) QueryLog(ctx context.Context, query interface{}) ([]interface{},
 		logs = append(logs, items[i])
 	}
 
-	return logs, 0, nil
+	return logs, int64(len(logs)), nil
 }
 
 func (d *Doris) DescribeTable(ctx context.Context, query interface{}) ([]*types.ColumnProperty, error) {

docker/compose-postgres/initsql_for_postgres/a-n9e-for-Postgres.sql

@@ -956,7 +956,7 @@ CREATE TABLE notify_rule (
     id bigserial PRIMARY KEY,
     name varchar(255) NOT NULL,
     description text,
-    enable smallint NOT NULL DEFAULT 0,
+    enable boolean DEFAULT false,
     user_group_ids varchar(255) NOT NULL DEFAULT '',
     notify_configs text,
     pipeline_configs text,
@@ -971,7 +971,7 @@ CREATE TABLE notify_channel (
     name varchar(255) NOT NULL,
     ident varchar(255) NOT NULL,
     description text,
-    enable smallint NOT NULL DEFAULT 0,
+    enable boolean DEFAULT false,
     param_config text,
     request_type varchar(50) NOT NULL,
     request_config text,

dscache/sync.go

@@ -22,6 +22,8 @@ import (
 
 var FromAPIHook func()
 
+var DatasourceProcessHook func(items []datasource.DatasourceInfo) []datasource.DatasourceInfo
+
 func Init(ctx *ctx.Context, fromAPI bool) {
 	go getDatasourcesFromDBLoop(ctx, fromAPI)
 }
@@ -100,6 +102,10 @@ func getDatasourcesFromDBLoop(ctx *ctx.Context, fromAPI bool) {
 				atomic.StoreInt64(&PromDefaultDatasourceId, 0)
 			}
 
+			if DatasourceProcessHook != nil {
+				dss = DatasourceProcessHook(dss)
+			}
+
 			PutDatasources(dss)
 		} else {
 			FromAPIHook()

dskit/doris/doris.go

@@ -20,8 +20,8 @@ import (
 
 // Doris struct to hold connection details and the connection object
 type Doris struct {
-	Addr            string `json:"doris.addr" mapstructure:"doris.addr"`         // be node
-	FeAddr          string `json:"doris.fe_addr" mapstructure:"doris.fe_addr"`   // fe node
+	Addr            string `json:"doris.addr" mapstructure:"doris.addr"`         // fe mysql endpoint
+	FeAddr          string `json:"doris.fe_addr" mapstructure:"doris.fe_addr"`   // fe http endpoint
 	User            string `json:"doris.user" mapstructure:"doris.user"`         //
 	Password        string `json:"doris.password" mapstructure:"doris.password"` //
 	Timeout         int    `json:"doris.timeout" mapstructure:"doris.timeout"`

memsto/notify_channel_cache.go

@@ -240,17 +240,17 @@ func (ncc *NotifyChannelCacheType) startHttpChannel(chID int64, channel *models.
 		go ncc.startNotifyConsumer(chID, queue, quitCh)
 	}
 
-	logger.Infof("started %d notify consumers for channel %d", concurrency, chID)
+	logger.Debugf("started %d notify consumers for channel %d", concurrency, chID)
 }
 
 // 启动通知消费者协程
 func (ncc *NotifyChannelCacheType) startNotifyConsumer(channelID int64, queue *list.SafeListLimited, quitCh chan struct{}) {
-	logger.Infof("starting notify consumer for channel %d", channelID)
+	logger.Debugf("starting notify consumer for channel %d", channelID)
 
 	for {
 		select {
 		case <-quitCh:
-			logger.Infof("notify consumer for channel %d stopped", channelID)
+			logger.Debugf("notify consumer for channel %d stopped", channelID)
 			return
 		default:
 			// 从队列中取出任务
@@ -502,7 +502,11 @@ func (ncc *NotifyChannelCacheType) startEmailSender(chID int64, smtp *models.SMT
 					m.Mail.GetHeader("Subject"), m.Mail.GetHeader("To"))
 			}
 
-			// sender.NotifyRecord(ncc.ctx, m.Events, m.NotifyRuleId, models.Email, strings.Join(m.Mail.GetHeader("To"), ","), "", err)
+			// 记录通知详情
+			if ncc.notifyRecordFunc != nil {
+				target := strings.Join(m.Mail.GetHeader("To"), ",")
+				ncc.notifyRecordFunc(ncc.ctx, m.Events, m.NotifyRuleId, "Email", target, "success", err)
+			}
 			size++
 
 			if size >= conf.Batch {

models/alert_subscribe.go

@@ -116,7 +116,18 @@ func (s *AlertSubscribe) Verify() error {
 		return errors.New("severities is required")
 	}
 
-	if len(s.NotifyRuleIds) > 0 {
+	if s.NotifyVersion == 1 {
+		if len(s.NotifyRuleIds) == 0 {
+			return errors.New("no notify rules selected")
+		}
+
+		s.UserGroupIds = ""
+		s.RedefineChannels = 0
+		s.NewChannels = ""
+		s.RedefineWebhooks = 0
+		s.Webhooks = ""
+		s.RedefineSeverity = 0
+		s.NewSeverity = 0
 		return nil
 	}
 
@@ -132,8 +143,8 @@ func (s *AlertSubscribe) Verify() error {
 		}
 	}
 
-	if s.NotifyVersion == 1 && len(s.NotifyRuleIds) == 0 {
-		return errors.New("no notify rules selected")
+	if s.NotifyVersion == 0 {
+		s.NotifyRuleIds = []int64{}
 	}
 
 	return nil
@@ -381,6 +392,17 @@ func (s *AlertSubscribe) MatchProd(prod string) bool {
 	return s.Prod == prod
 }
 
+func (s *AlertSubscribe) MatchCate(cate string) bool {
+	if s.Cate == "" {
+		return true
+	}
+
+	if s.Cate == "host" {
+		return cate == "host"
+	}
+	return true
+}
+
 func (s *AlertSubscribe) MatchCluster(dsId int64) bool {
 	// 没有配置数据源, 或者事件不需要关联数据源
 	// do not match any datasource or event not related to datasource

models/migrate/migrate.go

@@ -72,8 +72,10 @@ func MigrateTables(db *gorm.DB) error {
 
 	if isPostgres(db) {
 		dts = append(dts, &models.PostgresBuiltinComponent{})
+		DropUniqueFiledLimit(db, &models.PostgresBuiltinComponent{}, "idx_ident", "idx_ident")
 	} else {
 		dts = append(dts, &models.BuiltinComponent{})
+		DropUniqueFiledLimit(db, &models.BuiltinComponent{}, "idx_ident", "idx_ident")
 	}
 
 	if !db.Migrator().HasColumn(&imodels.TaskSchedulerHealth{}, "scheduler") {
@@ -124,11 +126,17 @@ func MigrateTables(db *gorm.DB) error {
 	DropUniqueFiledLimit(db, &Configs{}, "ckey", "configs_ckey_key")
 	// 删除 builtin_metrics 表的 idx_collector_typ_name 唯一索引
 	DropUniqueFiledLimit(db, &models.BuiltinMetric{}, "idx_collector_typ_name", "idx_collector_typ_name")
+
 	InsertPermPoints(db)
 	return nil
 }
 
 func DropUniqueFiledLimit(db *gorm.DB, dst interface{}, uniqueFiled string, pgUniqueFiled string) { // UNIQUE KEY (`ckey`)
+	// 先检查表是否存在，如果不存在则直接返回
+	if !db.Migrator().HasTable(dst) {
+		return
+	}
+
 	if db.Migrator().HasIndex(dst, uniqueFiled) {
 		err := db.Migrator().DropIndex(dst, uniqueFiled) //mysql  DROP INDEX
 		if err != nil {

models/notify_channel.go

@@ -21,10 +21,10 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"syscall"
 	"time"
 	"unicode/utf8"
 
+	"github.com/ccfos/nightingale/v6/pkg/cmdx"
 	"github.com/ccfos/nightingale/v6/pkg/ctx"
 	"github.com/ccfos/nightingale/v6/pkg/poster"
 	"github.com/ccfos/nightingale/v6/pkg/tplx"
@@ -33,7 +33,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/toolkits/pkg/file"
 	"github.com/toolkits/pkg/logger"
-	"github.com/toolkits/pkg/sys"
 	"gopkg.in/gomail.v2"
 )
 
@@ -196,10 +195,8 @@ func (ncc *NotifyChannelConfig) SendScript(events []*AlertCurEvent, tpl map[stri
 	cmd.Stdout = &buf
 	cmd.Stderr = &buf
 
-	err := startCmd(cmd)
-	if err != nil {
-		return "", "", fmt.Errorf("failed to start script: %v", err)
-	}
+	err, isTimeout := cmdx.RunTimeout(cmd, time.Duration(config.Timeout)*time.Millisecond)
+	logger.Infof("event_script_notify_result: exec %s output: %s isTimeout: %v err: %v", fpath, buf.String(), isTimeout, err)
 
 	res := buf.String()
 
@@ -218,8 +215,6 @@ func (ncc *NotifyChannelConfig) SendScript(events []*AlertCurEvent, tpl map[stri
 		res = res[:validLen] + "..."
 	}
 
-	err, isTimeout := sys.WrapTimeout(cmd, time.Duration(config.Timeout)*time.Second)
-	logger.Infof("event_script_notify_result: exec %s output: %s isTimeout: %v err: %v", fpath, buf.String(), isTimeout, err)
 	if isTimeout {
 		if err == nil {
 			return cmd.String(), res, errors.New("timeout and killed process")
@@ -257,11 +252,6 @@ func getStdinBytes(events []*AlertCurEvent, tpl map[string]interface{}, params m
 	return jsonBytes
 }
 
-func startCmd(c *exec.Cmd) error {
-	c.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
-	return c.Start()
-}
-
 func NotifyChannelStatistics(ctx *ctx.Context) (*Statistics, error) {
 	if !ctx.IsCenter {
 		s, err := poster.GetByUrls[*Statistics](ctx, "/v1/n9e/statistic?name=notify_channel")
@@ -932,11 +922,6 @@ func (ncc *NotifyChannelConfig) ValidateFlashDutyRequestConfig() error {
 }
 
 func (ncc *NotifyChannelConfig) Update(ctx *ctx.Context, ref NotifyChannelConfig) error {
-	// ref.FE2DB()
-	if ncc.Ident != ref.Ident {
-		return errors.New("cannot update ident")
-	}
-
 	ref.ID = ncc.ID
 	ref.CreateAt = ncc.CreateAt
 	ref.CreateBy = ncc.CreateBy

models/user_token.go

@@ -8,11 +8,11 @@ import (
 
 type UserToken struct {
 	Id        int64  `json:"id" gorm:"primaryKey"`
-	Username  string `json:"username" gorm:"type:varchar(255) not null default ''"`
-	TokenName string `json:"token_name" gorm:"type:varchar(255) not null default ''"`
-	Token     string `json:"token" gorm:"type:varchar(255) not null default ''"`
-	CreateAt  int64  `json:"create_at" gorm:"type:bigint not null default 0"`
-	LastUsed  int64  `json:"last_used" gorm:"type:bigint not null default 0"`
+	Username  string `json:"username" gorm:"type:varchar(255); not null; default ''"`
+	TokenName string `json:"token_name" gorm:"type:varchar(255); not null; default ''"`
+	Token     string `json:"token" gorm:"type:varchar(255); not null; default ''"`
+	CreateAt  int64  `json:"create_at" gorm:"type:bigint; not null; default 0"`
+	LastUsed  int64  `json:"last_used" gorm:"type:bigint; not null; default 0"`
 }
 
 func (UserToken) TableName() string {

pkg/cmdx/cmd_notwindows.go

@@ -0,0 +1,37 @@
+//go:build !windows
+// +build !windows
+
+package cmdx
+
+import (
+	"os/exec"
+	"syscall"
+	"time"
+)
+
+func CmdWait(cmd *exec.Cmd, timeout time.Duration) (error, bool) {
+	var err error
+
+	done := make(chan error)
+	go func() {
+		done <- cmd.Wait()
+	}()
+
+	select {
+	case <-time.After(timeout):
+		go func() {
+			<-done // allow goroutine to exit
+		}()
+
+		// IMPORTANT: cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} is necessary before cmd.Start()
+		err = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+		return err, true
+	case err = <-done:
+		return err, false
+	}
+}
+
+func CmdStart(cmd *exec.Cmd) error {
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	return cmd.Start()
+}

pkg/cmdx/cmd_windows.go

@@ -0,0 +1,35 @@
+//go:build windows
+// +build windows
+
+package cmdx
+
+import (
+	"os/exec"
+	"syscall"
+	"time"
+)
+
+func CmdWait(cmd *exec.Cmd, timeout time.Duration) (error, bool) {
+	var err error
+
+	done := make(chan error)
+	go func() {
+		done <- cmd.Wait()
+	}()
+
+	select {
+	case <-time.After(timeout):
+		go func() {
+			<-done // allow goroutine to exit
+		}()
+
+		err = cmd.Process.Signal(syscall.SIGKILL)
+		return err, true
+	case err = <-done:
+		return err, false
+	}
+}
+
+func CmdStart(cmd *exec.Cmd) error {
+	return cmd.Start()
+}

pkg/cmdx/cmdx.go

@@ -0,0 +1,15 @@
+package cmdx
+
+import (
+	"os/exec"
+	"time"
+)
+
+func RunTimeout(cmd *exec.Cmd, timeout time.Duration) (error, bool) {
+	err := CmdStart(cmd)
+	if err != nil {
+		return err, false
+	}
+
+	return CmdWait(cmd, timeout)
+}

pkg/flashduty/sync_user.go

@@ -89,7 +89,7 @@ func diffMap(m1, m2 map[int64]*models.User) []models.User {
 func updateUser(appKey string, m1, m2 map[int64]*models.User) {
 	for i := range m1 {
 		if _, ok := m2[i]; ok {
-			if m1[i].Email != m2[i].Email || m1[i].Phone != m2[i].Phone || m1[i].Username != m2[i].Username {
+			if m1[i].Email != m2[i].Email || !PhoneIsSame(m1[i].Phone, m2[i].Phone) || m1[i].Username != m2[i].Username {
 				var flashdutyUser User
 
 				flashdutyUser = User{
@@ -110,6 +110,30 @@ func updateUser(appKey string, m1, m2 map[int64]*models.User) {
 	}
 }
 
+func PhoneIsSame(phone1, phone2 string) bool {
+	// 兼容不同国家/地区前缀，例如 +86、+1、+44 等，以及包含空格或短横线的格式
+	normalize := func(p string) string {
+		p = strings.TrimSpace(p)
+		p = strings.ReplaceAll(p, " ", "")
+		p = strings.ReplaceAll(p, "-", "")
+		p = strings.TrimPrefix(p, "+")
+		return p
+	}
+
+	p1 := normalize(phone1)
+	p2 := normalize(phone2)
+
+	if p1 == p2 {
+		return true
+	}
+
+	// 如果长度相差不超过 3 且较长的以较短的结尾，则认为是相同号码（忽略最多 3 位国家区号差异）
+	if len(p1) > len(p2) {
+		return len(p1)-len(p2) <= 3 && strings.HasSuffix(p1, p2)
+	}
+	return len(p2)-len(p1) <= 3 && strings.HasSuffix(p2, p1)
+}
+
 type User struct {
 	Email      string  `json:"email,omitempty"`
 	Phone      string  `json:"phone,omitempty"`

pkg/flashduty/sync_user_test.go

@@ -0,0 +1,67 @@
+package flashduty
+
+import "testing"
+
+func TestPhoneIsSame(t *testing.T) {
+	tests := []struct {
+		name   string
+		phone1 string
+		phone2 string
+		same   bool
+	}{
+		{
+			name:   "blank",
+			phone1: "",
+			phone2: "",
+			same:   true,
+		},
+		{
+			name:   "China +86 prefix",
+			phone1: "+8613812345678",
+			phone2: "13812345678",
+			same:   true,
+		},
+		{
+			name:   "China +86 with spaces and hyphens",
+			phone1: "+86 138-1234-5678",
+			phone2: "13812345678",
+			same:   true,
+		},
+		{
+			name:   "USA +1 prefix",
+			phone1: "+1 234-567-8900",
+			phone2: "2345678900",
+			same:   true,
+		},
+		{
+			name:   "UK +44 prefix",
+			phone1: "+442078765432",
+			phone2: "2078765432",
+			same:   true,
+		},
+		{
+			name:   "India +91 prefix",
+			phone1: "+919876543210",
+			phone2: "9876543210",
+			same:   true,
+		},
+		{
+			name:   "Germany +49 prefix",
+			phone1: "+4915123456789",
+			phone2: "15123456789",
+			same:   true,
+		},
+		{
+			name:   "Different numbers",
+			phone1: "+8613812345678",
+			phone2: "13812345679",
+			same:   false,
+		},
+	}
+
+	for _, tt := range tests {
+		if got := PhoneIsSame(tt.phone1, tt.phone2); got != tt.same {
+			t.Errorf("%s: expected %v, got %v", tt.name, tt.same, got)
+		}
+	}
+}