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-08 09:03:20 +08:00
parent 14db0dd2fe
commit 5f2c9c3855
19 changed files with 1058 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

328
sync/sync_service.go
View File

@@ -1,14 +1,15 @@
package sync
import (
dao "cid/dao/copydata"
dto "cid/model/dto/copydata"
taskDto "cid/model/dto/copydata"
"cid/service/copydata"
"context"
"encoding/json"
"fmt"
"time"
"gitea.com/red-future/common/beans"
"github.com/sirupsen/logrus"
)
@@ -27,32 +28,44 @@ func NewSyncService() *SyncService {
}
type SyncResult struct {
SumSuccess bool `json:"sum_success"`
SumID int64 `json:"sum_id"`
DetailSuccess bool `json:"detail_success"`
DetailCount int `json:"detail_count"`
DetailSuccessCount int64 `json:"detail_success_count"`
DetailFailCount int64 `json:"detail_fail_count"`
Error error `json:"error"`
SumSuccess bool `json:"sum_success"`
SumID int64 `json:"sum_id"`
DetailSuccess bool `json:"detail_success"`
DetailCount int `json:"detail_count"`
DetailSuccessCount int64 `json:"detail_success_count"`
DetailFailCount int64 `json:"detail_fail_count"`
Error error `json:"error"`
TaskLogID int64 `json:"task_log_id"`
PageResults []*PageSyncResult `json:"page_results,omitempty"`
}
func (s *SyncService) SyncCampaignReport(ctx context.Context, req *CampaignReportRequest, useMock bool) (*SyncResult, error) {
type PageSyncResult struct {
PageNumber int `json:"page_number"`
PageTaskLogID int64 `json:"page_task_log_id"`
Success bool `json:"success"`
RecordCount int `json:"record_count"`
DurationMs int64 `json:"duration_ms"`
ErrorMessage string `json:"error_message,omitempty"`
RetryCount int `json:"retry_count"`
}
func (s *SyncService) SyncAccountReport(ctx context.Context, req *AccountReportRequest, useMock bool) (*SyncResult, error) {
result := &SyncResult{}
var responseData *CampaignReportResponse
var responseData *AccountReportResponse
if useMock {
logrus.Info("使用 Mock 数据同步快手广告计划报表")
responseData = s.mockGen.GenerateCampaignReportResponse()
logrus.Info("使用 Mock 数据同步快手广告账户报表")
responseData = s.mockGen.GenerateAccountReportResponse()
} else {
logrus.Info("从真实 API 同步快手广告计划报表")
respBytes, err := s.httpClient.Post(ctx, "/rest/openapi/gw/esp/report/campaignReport", req)
logrus.Info("从真实 API 同步快手广告账户报表")
respBytes, err := s.httpClient.Post(ctx, "/rest/openapi/gw/esp/report/accountReport", req)
if err != nil {
result.Error = fmt.Errorf("调用 API 失败：%w", err)
return result, result.Error
}
responseData = &CampaignReportResponse{}
responseData = &AccountReportResponse{}
if err := json.Unmarshal(respBytes, responseData); err != nil {
result.Error = fmt.Errorf("解析响应失败：%w", err)
return result, result.Error
@@ -65,9 +78,7 @@ func (s *SyncService) SyncCampaignReport(ctx context.Context, req *CampaignRepor
}
if responseData.Data.Sum != nil {
sumItem := s.converter.ConvertToSumItem(responseData.Data.Sum, "campaign_report")
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"})
sumItem := s.converter.ConvertToSumItem(responseData.Data.Sum, "account_report", req.PageInfo.CurrentPage)
sumResult, saveErr := s.saveSumData(ctx, sumItem)
if saveErr != nil {
logrus.Errorf("保存汇总数据失败：%v", saveErr)
@@ -80,7 +91,7 @@ func (s *SyncService) SyncCampaignReport(ctx context.Context, req *CampaignRepor
}
if len(responseData.Data.Detail) > 0 {
detailItems := s.converter.ConvertToDetailItems(responseData.Data.Detail, "campaign_report")
detailItems := s.converter.ConvertToDetailItems(responseData.Data.Detail, "account_report", req.PageInfo.CurrentPage)
detailResult, saveErr := s.saveDetailData(ctx, detailItems)
if saveErr != nil {
logrus.Errorf("保存明细数据失败：%v", saveErr)
@@ -90,39 +101,117 @@ func (s *SyncService) SyncCampaignReport(ctx context.Context, req *CampaignRepor
result.DetailCount = len(detailItems)
result.DetailSuccessCount = detailResult.SuccessCount
result.DetailFailCount = detailResult.FailCount
logrus.Infof("成功保存明细数据，成功=%d, 失败=%d", detailResult.SuccessCount, detailResult.FailCount)
logrus.Infof("成功保存 %d 条明细数据（成功=%d, 失败=%d）", len(detailItems), detailResult.SuccessCount, detailResult.FailCount)
}
}
return result, result.Error
}
// SyncCampaignReportWithPagination 带分页处理的同步方法（支持全量数据抽取）
func (s *SyncService) SyncCampaignReportWithPagination(ctx context.Context, req *CampaignReportRequest, useMock bool, maxRetries int) (*SyncResult, error) {
aggregatedResult := &SyncResult{
SumSuccess: false,
SumID: 0,
// SyncAccountReportWithPagination 带分页处理的同步方法（支持全量数据抽取）
func (s *SyncService) SyncAccountReportWithPagination(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int) (*SyncResult, error) {
startTime := time.Now()
parentTaskID := fmt.Sprintf("%d_%d_account", req.AdvertiserID, req.StartTime)
logReq := &taskDto.CreateSyncTaskLogReq{
TaskID: parentTaskID,
TaskType: "account_report",
AdvertiserID: req.AdvertiserID,
StartTime: time.UnixMilli(req.StartTime),
EndTime: time.UnixMilli(req.EndTime),
Status: "pending",
MaxRetry: maxRetries,
RequestParams: req,
}
parentLogID, err := dao.SyncTaskLog.Create(ctx, logReq)
if err != nil {
logrus.Errorf("创建主任务日志失败：%v", err)
}
updateParentLog := func(status, errMsg, errorCode string, summary interface{}) {
if parentLogID == 0 {
return
}
duration := time.Since(startTime).Milliseconds()
updateReq := &taskDto.UpdateSyncTaskLogReq{
ID: parentLogID,
Status: status,
ErrorMessage: errMsg,
ErrorCode: errorCode,
DurationMs: &duration,
}
if status == "success" || status == "manual_review" {
completedAt := time.Now()
updateReq.CompletedAt = completedAt
}
if summary != nil {
updateReq.ResultSummary = summary
}
if err := dao.SyncTaskLog.Update(ctx, updateReq); err != nil {
logrus.Errorf("更新主任务日志失败：%v", err)
}
}
updateParentLog("running", "", "", nil)
aggregatedResult := &SyncResult{
SumSuccess: false,
SumID: 0,
TaskLogID: parentLogID,
PageResults: make([]*PageSyncResult, 0),
}
allDetailItems := make([]*dto.CidAccountReportDetailItem, 0)
totalCount := 0
currentPage := 1
pageSize := 100
successPages := 0
failedPages := 0
if req.PageInfo == nil {
req.PageInfo = &PageInfo{}
}
var totalPages int
for {
logrus.Infof(">>> 正在同步第 %d 页数据...", currentPage)
req.PageInfo.CurrentPage = currentPage
req.PageInfo.PageSize = pageSize
result, err := s.SyncWithRetry(ctx, req, useMock, maxRetries)
pageTaskID := fmt.Sprintf("%s_page_%d", parentTaskID, currentPage)
pageStartTime := time.Now()
pageResult := &PageSyncResult{
PageNumber: currentPage,
Success: false,
RecordCount: 0,
RetryCount: 0,
}
result, err := s.syncSinglePageWithTask(ctx, req, useMock, maxRetries, pageTaskID, currentPage)
pageDuration := time.Since(pageStartTime).Milliseconds()
pageResult.DurationMs = pageDuration
if err != nil {
logrus.Errorf("第 %d 页同步失败：%v", currentPage, err)
return aggregatedResult, err
pageResult.ErrorMessage = err.Error()
failedPages++
aggregatedResult.PageResults = append(aggregatedResult.PageResults, pageResult)
if failedPages > maxRetries {
logrus.Warnf("失败页数超过阈值 %d，终止同步", maxRetries)
break
}
currentPage++
time.Sleep(500 * time.Millisecond)
continue
}
if result.SumSuccess && aggregatedResult.SumID == 0 {
@@ -132,17 +221,18 @@ func (s *SyncService) SyncCampaignReportWithPagination(ctx context.Context, req
}
if result.DetailSuccess && result.DetailCount > 0 {
detailItems := s.extractDetailItems(req, useMock)
if len(detailItems) > 0 {
allDetailItems = append(allDetailItems, detailItems...)
totalCount += len(detailItems)
logrus.Infof("✓ 第 %d 页获取到 %d 条明细数据，累计 %d 条", currentPage, len(detailItems), totalCount)
}
totalCount += result.DetailCount
pageResult.Success = true
pageResult.RecordCount = result.DetailCount
successPages++
logrus.Infof("✓ 第 %d 页获取到 %d 条明细数据，累计 %d 条", currentPage, result.DetailCount, totalCount)
}
aggregatedResult.PageResults = append(aggregatedResult.PageResults, pageResult)
currentData := s.fetchCurrentData(req, useMock)
if currentData != nil && currentData.TotalCount > 0 {
totalPages := (currentData.TotalCount + pageSize - 1) / pageSize
totalPages = (currentData.TotalCount + pageSize - 1) / pageSize
logrus.Infof("总记录数：%d, 总页数：%d, 当前页：%d/%d",
currentData.TotalCount, totalPages, currentPage, totalPages)
@@ -161,70 +251,148 @@ func (s *SyncService) SyncCampaignReportWithPagination(ctx context.Context, req
time.Sleep(300 * time.Millisecond)
}
if len(allDetailItems) > 0 {
logrus.Infof("开始批量保存 %d 条明细数据...", len(allDetailItems))
detailResult, saveErr := s.saveDetailData(ctx, allDetailItems)
if saveErr != nil {
logrus.Errorf("批量保存明细数据失败：%v", saveErr)
aggregatedResult.Error = fmt.Errorf("批量保存明细数据失败：%w", saveErr)
logrus.Infof("分页同步完成 - 成功:%d页, 失败:%d页, 总明细:%d条",
successPages, failedPages, totalCount)
// 统计所有子任务的结果
totalDetailCount := 0
var totalSuccessCount int64
var totalFailCount int64
for _, pageResult := range aggregatedResult.PageResults {
if pageResult.Success {
totalDetailCount += pageResult.RecordCount
totalSuccessCount++
} else {
aggregatedResult.DetailSuccess = true
aggregatedResult.DetailCount = len(allDetailItems)
aggregatedResult.DetailSuccessCount = detailResult.SuccessCount
aggregatedResult.DetailFailCount = detailResult.FailCount
logrus.Infof("✓ 批量保存明细数据完成，成功=%d, 失败=%d",
detailResult.SuccessCount, detailResult.FailCount)
totalFailCount++
}
}
aggregatedResult.DetailCount = totalDetailCount
aggregatedResult.DetailSuccessCount = totalSuccessCount
aggregatedResult.DetailFailCount = totalFailCount
if failedPages > 0 {
logrus.Warnf("存在 %d 个失败的页面，主任务标记为部分失败", failedPages)
summary := map[string]interface{}{
"sum_id": aggregatedResult.SumID,
"detail_count": totalDetailCount,
"total_pages": totalPages,
"success_pages": successPages,
"failed_pages": failedPages,
"page_results": aggregatedResult.PageResults,
}
updateParentLog("partial_failed", fmt.Sprintf("%d 个页面同步失败", failedPages), "PAGE_SYNC_FAILED", summary)
} else {
logrus.Info("没有明细数据需要保存")
logrus.Info("✓ 所有页面同步成功")
summary := map[string]interface{}{
"sum_id": aggregatedResult.SumID,
"detail_count": totalDetailCount,
"total_pages": totalPages,
"success_pages": successPages,
"failed_pages": 0,
"page_results": aggregatedResult.PageResults,
}
updateParentLog("success", "", "", summary)
}
return aggregatedResult, aggregatedResult.Error
}
func (s *SyncService) extractDetailItems(req *CampaignReportRequest, useMock bool) []*dto.CidAccountReportDetailItem {
if useMock {
responseData := s.mockGen.GenerateCampaignReportResponse()
if responseData == nil || responseData.Data == nil || len(responseData.Data.Detail) == 0 {
return nil
}
return s.converter.ConvertToDetailItems(responseData.Data.Detail, "campaign_report")
}
respBytes, err := s.httpClient.Post(context.Background(), "/rest/openapi/gw/esp/report/campaignReport", req)
if err != nil {
logrus.Errorf("重新获取数据失败：%v", err)
return nil
}
responseData := &CampaignReportResponse{}
if err := json.Unmarshal(respBytes, responseData); err != nil {
logrus.Errorf("解析响应失败：%v", err)
return nil
}
if responseData.Code != 0 || responseData.Data == nil || len(responseData.Data.Detail) == 0 {
return nil
}
return s.converter.ConvertToDetailItems(responseData.Data.Detail, "campaign_report")
func (s *SyncService) SyncSinglePageWithTask(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int, pageTaskID string, pageNumber int) (*SyncResult, error) {
return s.syncSinglePageWithTask(ctx, req, useMock, maxRetries, pageTaskID, pageNumber)
}
func (s *SyncService) fetchCurrentData(req *CampaignReportRequest, useMock bool) *CampaignReportData {
func (s *SyncService) syncSinglePageWithTask(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int, pageTaskID string, pageNumber int) (*SyncResult, error) {
pageStartTime := time.Now()
pageLogReq := &taskDto.CreateSyncTaskLogReq{
TaskID: pageTaskID,
TaskType: "account_report_page",
AdvertiserID: req.AdvertiserID,
StartTime: time.UnixMilli(req.StartTime),
EndTime: time.UnixMilli(req.EndTime),
Status: "pending",
MaxRetry: maxRetries,
PageInfo: req.PageInfo,
RequestParams: map[string]interface{}{
"page_number": pageNumber,
"page_size": req.PageInfo.PageSize,
},
}
pageLogID, err := dao.SyncTaskLog.Create(ctx, pageLogReq)
if err != nil {
logrus.Errorf("创建分页任务日志失败：%v", err)
}
updatePageLog := func(status, errMsg, errorCode string, retryCount int) {
if pageLogID == 0 {
return
}
duration := time.Since(pageStartTime).Milliseconds()
updateReq := &taskDto.UpdateSyncTaskLogReq{
ID: pageLogID,
Status: status,
ErrorMessage: errMsg,
ErrorCode: errorCode,
DurationMs: &duration,
}
if retryCount > 0 {
updateReq.RetryCount = &retryCount
}
if status == "success" || status == "failed" {
completedAt := time.Now()
updateReq.CompletedAt = completedAt
}
if err := dao.SyncTaskLog.Update(ctx, updateReq); err != nil {
logrus.Errorf("更新分页任务日志失败：%v", err)
}
}
updatePageLog("running", "", "", 0)
logrus.Infof(">>> 开始同步第 %d 页数据...", pageNumber)
result, err := s.SyncWithRetry(ctx, req, useMock, maxRetries)
if err != nil {
updatePageLog("failed", err.Error(), "PAGE_SYNC_FAILED", 0)
return result, err
}
summary := map[string]interface{}{
"page_number": pageNumber,
"detail_count": result.DetailCount,
"sum_saved": result.SumSuccess,
}
updatePageLog("success", "", "", 0)
logrus.Debugf("分页任务 %s 完成: %v", pageTaskID, summary)
return result, nil
}
func (s *SyncService) fetchCurrentData(req *AccountReportRequest, useMock bool) *AccountReportData {
if useMock {
responseData := s.mockGen.GenerateCampaignReportResponse()
responseData := s.mockGen.GenerateAccountReportResponse()
if responseData != nil && responseData.Data != nil {
return responseData.Data
}
return nil
}
respBytes, err := s.httpClient.Post(context.Background(), "/rest/openapi/gw/esp/report/campaignReport", req)
respBytes, err := s.httpClient.Post(context.Background(), "/rest/openapi/gw/esp/report/accountReport", req)
if err != nil {
return nil
}
responseData := &CampaignReportResponse{}
responseData := &AccountReportResponse{}
if err := json.Unmarshal(respBytes, responseData); err != nil {
return nil
}
@@ -247,12 +415,12 @@ func (s *SyncService) saveDetailData(ctx context.Context, items []*dto.CidAccoun
return copydata.CidAccountReportDetail.BatchCreate(ctx, req)
}
func (s *SyncService) SyncWithRetry(ctx context.Context, req *CampaignReportRequest, useMock bool, maxRetries int) (*SyncResult, error) {
func (s *SyncService) SyncWithRetry(ctx context.Context, req *AccountReportRequest, useMock bool, maxRetries int) (*SyncResult, error) {
var lastResult *SyncResult
var lastErr error
for attempt := 0; attempt <= maxRetries; attempt++ {
result, err := s.SyncCampaignReport(ctx, req, useMock)
result, err := s.SyncAccountReport(ctx, req, useMock)
lastResult = result
lastErr = err