data-engine/dao/tencent/account_relation_dao.go

package tencent

import (
	"context"
	consts "dataengine/consts/public"
	entity "dataengine/model/entity/tencent"
	"time"

	"gitea.com/red-future/common/db/gfdb"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/sirupsen/logrus"
)

type accountRelationDao struct{}

var AccountRelation = new(accountRelationDao)

// Upsert 插入或更新账户关系（根据account_id判断）
func (d *accountRelationDao) Upsert(ctx context.Context, item *entity.AccountRelation) error {
	now := time.Now()

	// 检查是否已存在
	var existing entity.AccountRelation
	err := gfdb.DB(ctx).Model(ctx, consts.TencentAccountRelationTable).
		Where("tenant_id", item.TenantId).
		Where(entity.AccountRelationCols.AccountID, item.AccountID).
		WhereNull(entity.AccountRelationCols.DeletedAt).
		Scan(&existing)

	// Scan找不到记录时err不为nil，但这是正常情况，需要继续执行插入
	if err != nil && existing.Id == 0 {
		// 记录不存在，执行插入
		item.CreatedAt = &now
		item.UpdatedAt = &now

		_, err = gfdb.DB(ctx).Model(ctx, consts.TencentAccountRelationTable).
			Data(item).
			Insert()

		return err
	}

	// 记录存在，执行更新
	item.UpdatedAt = &now
	_, err = gfdb.DB(ctx).Model(ctx, consts.TencentAccountRelationTable).
		Where("id", existing.Id).
		Data(g.Map{
			entity.AccountRelationCols.CorporationName: item.CorporationName,
			entity.AccountRelationCols.CommentDataList: item.CommentDataList,
			entity.AccountRelationCols.IsAdx:           item.IsAdx,
			entity.AccountRelationCols.IsBid:           item.IsBid,
			entity.AccountRelationCols.IsMp:            item.IsMp,
			entity.AccountRelationCols.UpdatedAt:       now,
		}).
		Update()

	return err
}

// BatchUpsert 批量插入或更新（使用 OnConflict 实现 Upsert）
func (d *accountRelationDao) BatchUpsert(ctx context.Context, items []*entity.AccountRelation) (successCount int, err error) {
	if len(items) == 0 {
		return 0, nil
	}

	logrus.Infof("开始批量Upsert: %d 条记录", len(items))

	// 分批处理，每批100条
	batchSize := 100
	successCount = 0

	for i := 0; i < len(items); i += batchSize {
		end := i + batchSize
		if end > len(items) {
			end = len(items)
		}

		batch := items[i:end]

		logrus.Infof("处理第 %d-%d 条记录", i+1, end)

		// 执行批量插入，使用 OnConflict 实现 Upsert
		result, err := gfdb.DB(ctx).Model(ctx, consts.TencentAccountRelationTable).
			Data(batch).
			OnConflict(entity.AccountRelationCols.AccountID).
			Save()

		if err != nil {
			logrus.Errorf("批量Upsert失败: %v，尝试逐条处理", err)
			// 批量失败，逐条处理
			for _, item := range batch {
				if upsertErr := d.Upsert(ctx, item); upsertErr != nil {
					logrus.Errorf("逐条Upsert失败: account_id=%d, err=%v", item.AccountID, upsertErr)
					continue
				}
				successCount++
			}
		} else {
			affected, _ := result.RowsAffected()
			successCount += int(affected)
			logrus.Infof("批量Upsert成功: 影响 %d 条记录", affected)
		}
	}

	logrus.Infof("批量Upsert完成: 成功 %d 条", successCount)
	return successCount, nil
}

// ListAll 获取所有账户关系
func (d *accountRelationDao) ListAll(ctx context.Context) ([]entity.AccountRelation, error) {
	var list []entity.AccountRelation
	err := gfdb.DB(ctx).Model(ctx, consts.TencentAccountRelationTable).
		WhereNull(entity.AccountRelationCols.DeletedAt).
		OrderAsc(entity.AccountRelationCols.AccountID).
		Scan(&list)

	return list, err
}