red-future/data-engine
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补偿数据逻辑调整

Browse Source
This commit is contained in:
李明康
2026-04-09 09:48:22 +08:00
parent 5d72dd3aff
commit 556ec5075e
3 changed files with 99 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
model/dto/copydata/sync_task_log_dto.go
View File

@@ -48,6 +48,8 @@ type SyncTaskLogItem struct {
ErrorMessage string `json:"errorMessage,omitempty"`
ErrorCode string `json:"errorCode,omitempty"`
ResultSummary interface{} `json:"resultSummary,omitempty"`
PageInfo interface{} `json:"pageInfo,omitempty"`
RequestParams interface{} `json:"requestParams,omitempty"`
NextRetryTime interface{} `json:"nextRetryTime,omitempty"`
CompletedAt interface{} `json:"completedAt,omitempty"`
DurationMs int64 `json:"durationMs"`

82
scheduler/run_sync_task_log_task.go
View File

@@ -15,16 +15,19 @@ import (
"github.com/sirupsen/logrus"
)
// CompensationScheduler 补偿调度器，负责扫描和补偿失败的分页同步任务
type CompensationScheduler struct {
syncService *sync.SyncService
}
// NewCompensationScheduler 创建补偿调度器实例
func NewCompensationScheduler() *CompensationScheduler {
return &CompensationScheduler{
syncService: sync.NewSyncService(),
}
}
// RunCompensationOnce 执行一次补偿任务（用于手动触发或定时任务调用）
func (s *CompensationScheduler) RunCompensationOnce() {
ctx := gctx.New()
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"})
@@ -34,6 +37,7 @@ func (s *CompensationScheduler) RunCompensationOnce() {
logrus.Info("=== 补偿任务执行完毕 ===")
}
// processCompensation 处理补偿逻辑：扫描失败的分页任务并逐个补偿
func (s *CompensationScheduler) processCompensation(ctx context.Context) {
logrus.Info(">>> 开始扫描需要补偿的失败分页任务...")
@@ -99,6 +103,8 @@ func (s *CompensationScheduler) processCompensation(ctx context.Context) {
len(failedPageTasks), successCount, failCount)
}
// compensatePageTask 补偿单个分页任务：重新请求API并插入数据
// 返回 true 表示补偿成功，false 表示补偿失败
func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask *taskDto.SyncTaskLogItem) bool {
retryCount := pageTask.RetryCount + 1
@@ -107,7 +113,10 @@ func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask
Status: "retrying",
RetryCount: &retryCount,
}
dao.SyncTaskLog.Update(ctx, updateReq)
if err := dao.SyncTaskLog.Update(ctx, updateReq); err != nil {
logrus.Errorf("更新任务状态为 retrying 失败：%v", err)
return false
}
startTime := s.parseTime(pageTask.StartTime)
endTime := s.parseTime(pageTask.EndTime)
@@ -119,6 +128,8 @@ func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask
return false
}
pageSize := s.extractPageSize(pageTask)
req := &sync.AccountReportRequest{
AdvertiserID: pageTask.AdvertiserID,
StartTime: startTime.UnixMilli(),
@@ -128,7 +139,7 @@ func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask
QueryVersion: 1,
PageInfo: &sync.PageInfo{
CurrentPage: pageNumber,
PageSize: 100,
PageSize: pageSize,
},
}
@@ -136,7 +147,7 @@ func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask
parentTaskID := s.extractParentTaskID(pageTask.TaskID)
pageTaskID := fmt.Sprintf("%s_page_%d", parentTaskID, pageNumber)
result, err := s.syncService.SyncSinglePageWithTask(ctx, req, true, maxRetries, pageTaskID, pageNumber)
result, _, err := s.syncService.SyncSinglePageWithTask(ctx, req, true, maxRetries, pageTaskID, pageNumber)
if err != nil {
logrus.Errorf("补偿分页任务 %s 失败（第 %d 次）：%v", pageTask.TaskID, retryCount, err)
@@ -148,6 +159,7 @@ func (s *CompensationScheduler) compensatePageTask(ctx context.Context, pageTask
return true
}
// markPageTaskFailed 标记分页任务为失败状态，并设置下次重试时间（指数退避策略）
func (s *CompensationScheduler) markPageTaskFailed(ctx context.Context, taskID int64, retryCount int, errMsg, errCode string) {
backoffMinutes := s.calculateBackoff(retryCount)
nextRetry := time.Now().Add(time.Duration(backoffMinutes) * time.Minute)
@@ -162,6 +174,7 @@ func (s *CompensationScheduler) markPageTaskFailed(ctx context.Context, taskID i
dao.SyncTaskLog.Update(ctx, updateReq)
}
// checkAndUpdateParentTaskStatus 检查主任务的所有分页任务状态，如果全部成功则更新主任务状态
func (s *CompensationScheduler) checkAndUpdateParentTaskStatus(ctx context.Context, parentTaskID string) {
logrus.Infof(">>> 检查主任务 %s 的所有分页任务状态...", parentTaskID)
@@ -226,6 +239,8 @@ func (s *CompensationScheduler) checkAndUpdateParentTaskStatus(ctx context.Conte
}
}
// extractParentTaskID 从分页任务ID中提取主任务ID
// 例如：从 "12345_1234567890_account_page_2" 提取 "12345_1234567890_account"
func (s *CompensationScheduler) extractParentTaskID(taskID string) string {
if idx := strings.LastIndex(taskID, "_page_"); idx > 0 {
return taskID[:idx]
@@ -233,6 +248,8 @@ func (s *CompensationScheduler) extractParentTaskID(taskID string) string {
return ""
}
// extractPageNumber 从分页任务ID中提取页码
// 例如：从 "12345_1234567890_account_page_2" 提取 2
func (s *CompensationScheduler) extractPageNumber(taskID string) int {
if idx := strings.LastIndex(taskID, "_page_"); idx > 0 {
var pageNum int
@@ -242,6 +259,63 @@ func (s *CompensationScheduler) extractPageNumber(taskID string) int {
return 0
}
// extractPageSize 从任务日志的 PageInfo 或 RequestParams 字段中提取每页大小
// 优先级：PageInfo.page_size > RequestParams.page_size > 默认值10
func (s *CompensationScheduler) extractPageSize(pageTask *taskDto.SyncTaskLogItem) int {
// 在 extractPageSize 方法开头添加
logrus.Infof("DEBUG - PageInfo 类型: %T, 值: %+v", pageTask.PageInfo, pageTask.PageInfo)
logrus.Infof("DEBUG - RequestParams 类型: %T, 值: %+v", pageTask.RequestParams, pageTask.RequestParams)
// 优先从 PageInfo 中提取
if pageTask.PageInfo != nil {
switch v := pageTask.PageInfo.(type) {
case map[string]interface{}:
// 尝试 float64 类型（JSON 数字默认类型）
if pageSize, ok := v["page_size"].(float64); ok {
return int(pageSize)
}
// 尝试 string 类型
if pageSizeStr, ok := v["page_size"].(string); ok {
var pageSize int
fmt.Sscanf(pageSizeStr, "%d", &pageSize)
if pageSize > 0 {
return pageSize
}
}
case map[string]string:
if pageSizeStr, ok := v["page_size"]; ok {
var pageSize int
fmt.Sscanf(pageSizeStr, "%d", &pageSize)
if pageSize > 0 {
return pageSize
}
}
}
}
// 其次从 RequestParams 中提取
if pageTask.RequestParams != nil {
switch v := pageTask.RequestParams.(type) {
case map[string]interface{}:
if pageSize, ok := v["page_size"].(float64); ok {
return int(pageSize)
}
if pageSizeStr, ok := v["page_size"].(string); ok {
var pageSize int
fmt.Sscanf(pageSizeStr, "%d", &pageSize)
if pageSize > 0 {
return pageSize
}
}
}
}
// 默认值改为 10
return 10
}
// calculateBackoff 根据重试次数计算退避时间（分钟）
// 重试次数：1->5分钟, 2->15分钟, 3->30分钟, 4->60分钟, 5+->120分钟
func (s *CompensationScheduler) calculateBackoff(retryCount int) int {
backoffs := []int{5, 15, 30, 60, 120}
if retryCount <= len(backoffs) {
@@ -250,6 +324,7 @@ func (s *CompensationScheduler) calculateBackoff(retryCount int) int {
return backoffs[len(backoffs)-1]
}
// parseTime 解析时间字段，支持 time.Time 和字符串格式
func (s *CompensationScheduler) parseTime(t interface{}) time.Time {
switch v := t.(type) {
case time.Time:
@@ -262,6 +337,7 @@ func (s *CompensationScheduler) parseTime(t interface{}) time.Time {
return time.Now()
}
// sendAlert 发送告警通知（当前仅记录错误日志）
func (s *CompensationScheduler) sendAlert(task *taskDto.SyncTaskLogItem) {
logrus.Errorf("【告警】分页任务 %s 需要人工介入：广告主=%d, 错误=%s",
task.TaskID, task.AdvertiserID, task.ErrorMessage)

22
sync/sync_service.go
View File

@@ -59,8 +59,19 @@ func (s *SyncService) SyncAccountReport(ctx context.Context, req *AccountReportR
var responseData *AccountReportResponse
if useMock {
logrus.Info("使用 Mock 数据同步快手广告账户报表")
logrus.Infof("使用 Mock 数据同步快手广告账户报表 - page=%d, pageSize=%d",
req.PageInfo.CurrentPage, req.PageInfo.PageSize)
responseData = s.mockGen.GenerateAccountReportResponseWithPage(req.PageInfo.CurrentPage, req.PageInfo.PageSize)
logrus.Infof("✓ Mock 数据生成完成 - TotalCount=%d, Detail数组长度=%d",
responseData.Data.TotalCount, len(responseData.Data.Detail))
if len(responseData.Data.Detail) > 0 {
firstItem := responseData.Data.Detail[0]
lastItem := responseData.Data.Detail[len(responseData.Data.Detail)-1]
logrus.Infof(" 第一条: campaignId=%v, creativeId=%v", firstItem.CampaignId, firstItem.CreativeId)
logrus.Infof(" 最后一条: campaignId=%v, creativeId=%v", lastItem.CampaignId, lastItem.CreativeId)
}
} else {
logrus.Info("从真实 API 同步快手广告账户报表")
respBytes, err := s.httpClient.Post(ctx, "/rest/openapi/gw/esp/report/accountReport", req)
@@ -96,6 +107,9 @@ func (s *SyncService) SyncAccountReport(ctx context.Context, req *AccountReportR
if len(responseData.Data.Detail) > 0 {
detailItems := s.converter.ConvertToDetailItems(responseData.Data.Detail, "account_report", req.PageInfo.CurrentPage)
logrus.Infof("▶ 准备插入 %d 条明细数据（page=%d）", len(detailItems), req.PageInfo.CurrentPage)
detailResult, saveErr := s.saveDetailData(ctx, detailItems)
if saveErr != nil {
logrus.Errorf("保存明细数据失败：%v", saveErr)
@@ -105,7 +119,7 @@ func (s *SyncService) SyncAccountReport(ctx context.Context, req *AccountReportR
result.DetailCount = len(detailItems)
result.DetailSuccessCount = detailResult.SuccessCount
result.DetailFailCount = detailResult.FailCount
logrus.Infof("成功保存 %d 条明细数据（成功=%d, 失败=%d）", len(detailItems), detailResult.SuccessCount, detailResult.FailCount)
logrus.Infof("✓ 成功保存 %d 条明细数据（成功=%d, 失败=%d）", len(detailItems), detailResult.SuccessCount, detailResult.FailCount)
}
}
@@ -202,7 +216,7 @@ func (s *SyncService) SyncAccountReportWithPagination(ctx context.Context, req *
RetryCount: 0,
}
result, pageLogID, err := s.syncSinglePageWithTask(ctx, req, useMock, maxRetries, pageTaskID, currentPage)
result, pageLogID, err := s.SyncSinglePageWithTask(ctx, req, useMock, maxRetries, pageTaskID, currentPage)
pageDuration := time.Since(pageStartTime).Milliseconds()
pageResult.DurationMs = pageDuration
pageResult.PageTaskLogID = pageLogID
@@ -311,7 +325,7 @@ func (s *SyncService) SyncAccountReportWithPagination(ctx context.Context, req *
return aggregatedResult, aggregatedResult.Error
}
func (s *SyncService) syncSinglePageWithTask(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int, pageTaskID string, pageNumber int) (*SyncResult, int64, error) {
func (s *SyncService) SyncSinglePageWithTask(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int, pageTaskID string, pageNumber int) (*SyncResult, int64, error) {
pageStartTime := time.Now()
pageLogReq := &taskDto.CreateSyncTaskLogReq{