fix:重构BatchDownload的错误手机机制，避免过程报错被忽略

service/domain/batchDownloadDomain.go

@@ -1,6 +1,9 @@
 package domain
 
 import (
+	"fmt"
+	"time"
+
 	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
@@ -12,6 +15,9 @@ type BatchDownloadInfo struct {
 	Options     BatchDownloadOptions `bson:"options" json:"options"`
 	Progress    Progress             `bson:"progress" json:"progress"`
 	FailReasons []string             `bson:"failReasons" json:"failReasons"`
+	Items       []BatchDownloadItem  `bson:"items" json:"items"`
+	TotalCount  int                  `bson:"totalCount" json:"totalCount"`
+	DoneCount   int                  `bson:"doneCount" json:"doneCount"`
 	Owner       *[]string            `bson:"owner,omitempty" json:"owner,omitempty"`
 }
 
@@ -43,6 +49,9 @@ type BatchDownloadConfig struct {
 	Progress    Progress                     `bson:"progress" json:"progress"`
 	Data        map[string]BatchDownloadData `bson:"data" json:"data"` // map-key为SN
 	FailReasons []string                     `bson:"failReasons" json:"failReasons"`
+	Items       []BatchDownloadItem          `bson:"items" json:"items"`
+	TotalCount  int                          `bson:"totalCount" json:"totalCount"`
+	DoneCount   int                          `bson:"doneCount" json:"doneCount"`
 	Owner       *[]string                    `bson:"owner,omitempty" json:"owner,omitempty"`
 }
 
@@ -121,3 +130,44 @@ type BatchDownloadAHubUpdateLogInfo struct {
 	Result      string                                  `json:"result"`  // 升级结果
 	DetailList  []BatchDownloadUpdateLogAHubVersionInfo `json:"detailList"`
 }
+
+// ==================== BatchDownload 结构化结果单元 ====================
+
+// ItemStatus 单个下载工作单元的状态
+type ItemStatus = string
+
+const (
+	ItemPending ItemStatus = "pending" // 待处理
+	ItemSuccess ItemStatus = "success" // 成功
+	ItemFailed  ItemStatus = "failed"  // 失败
+	ItemSkipped ItemStatus = "skipped" // 无满足条件的数据，非错误
+)
+
+// item 失败发生的阶段
+const (
+	StageBuild    = "build"    // 构建期（查可召唤列表/站点信息、入队等）
+	StageQuery    = "query"    // 查询数据
+	StageCall     = "call"     // 召唤
+	StagePoll     = "poll"     // 轮询召唤结果
+	StageDownload = "download" // 下载文件
+	StageSave     = "save"     // 保存/写库
+)
+
+// BatchDownloadItem 是批量下载中的一个最小工作单元的结构化结果
+type BatchDownloadItem struct {
+	Key       string     `bson:"key" json:"key"`         // 唯一键 fileType|sn|devType|date|subType
+	Sn        string     `bson:"sn" json:"sn"`           // 归属 SN
+	DevType   int        `bson:"devType" json:"devType"` // -1 表示不涉及设备类型
+	FileType  string     `bson:"fileType" json:"fileType"`
+	SubType   string     `bson:"subType" json:"subType"` // logType / indexList / ""
+	Date      string     `bson:"date" json:"date"`
+	Status    ItemStatus `bson:"status" json:"status"`
+	Stage     string     `bson:"stage" json:"stage"`   // 失败发生的阶段
+	Reason    string     `bson:"reason" json:"reason"` // 失败补充文本
+	UpdatedAt time.Time  `bson:"updatedAt" json:"updatedAt"`
+}
+
+// BuildItemKey 生成 item 唯一键。主任务构建与子任务上报必须使用同一套规则。
+func BuildItemKey(fileType, sn string, devType int, date, subType string) string {
+	return fmt.Sprintf("%s|%s|%d|%s|%s", fileType, sn, devType, date, subType)
+}

service/repositories/api/fwh_pro/fwh_faultRecord.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 )
@@ -49,21 +50,39 @@ func queryFaultRecordCallableList(devType int, sn string) (*domain.FwhResponse[[
 	params.Add("callType", strconv.Itoa(devType))
 	params.Add("lang", "ZH_CN")
 
-	response, err := FwhHttpGetUnmarshal[[]domain.FaultRecordCallableInfo](getFullEnergyUrl("/hes-gateway/manage/getWaveRecordingCatalog"), params)
-	// 此处无数据也会失败，暂时定义失败为无数据，不报错
-	if err != nil {
-		zap.L().Error("[queryFaultRecordCallableList] Response failed",
+	// 请求失败重试：仅对请求错误（HTTP 层失败）重试 3 次、间隔 5s。
+	// 注意：单纯"无数据"由后端以 200 + code(400,故障目录为空) 返回（err==nil），不会进入重试分支，
+	// 由导出层 QueryFaultRecordCallableList 识别并返回空列表。
+	const maxRetry = 3
+	const retryInterval = 5 * time.Second
+
+	var response *domain.FwhResponse[[]domain.FaultRecordCallableInfo]
+	var err error
+	for attempt := 0; attempt < maxRetry; attempt++ {
+		response, err = FwhHttpGetUnmarshal[[]domain.FaultRecordCallableInfo](getFullEnergyUrl("/hes-gateway/manage/getWaveRecordingCatalog"), params)
+		if err == nil {
+			return response, nil
+		}
+		zap.L().Warn("[queryFaultRecordCallableList] 请求失败，准备重试",
 			zap.Int("devType", devType),
 			zap.String("sn", sn),
+			zap.Int("attempt", attempt+1),
+			zap.Int("maxRetry", maxRetry),
 			zap.String("err", err.Error()),
 		)
-		// 创造假的空数据
-		return &domain.FwhResponse[[]domain.FaultRecordCallableInfo]{
-			Code:   200,
-			Result: &[]domain.FaultRecordCallableInfo{},
-		}, nil
+		if attempt < maxRetry-1 {
+			time.Sleep(retryInterval)
+		}
 	}
-	return response, nil
+
+	// 重试仍失败：返回真实错误（不再吞错误伪造空数据），由上层记录到任务错误信息
+	zap.L().Error("[queryFaultRecordCallableList] 请求失败，已达最大重试次数",
+		zap.Int("devType", devType),
+		zap.String("sn", sn),
+		zap.Int("maxRetry", maxRetry),
+		zap.String("err", err.Error()),
+	)
+	return nil, fmt.Errorf("[queryFaultRecordCallableList] 请求失败(devType=%d, sn=%s, 已重试%d次): %w", devType, sn, maxRetry, err)
 }
 
 // 查询设备可召唤数据列表 - 仅自动录播

service/repositories/api/fwh_test/fwh_faultRecord.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 )
@@ -49,21 +50,39 @@ func queryFaultRecordCallableList(devType int, sn string) (*domain.FwhResponse[[
 	params.Add("callType", strconv.Itoa(devType))
 	params.Add("lang", "ZH_CN")
 
-	response, err := FwhHttpGetUnmarshal[[]domain.FaultRecordCallableInfo](getFullEnergyUrl("/hes-gateway/manage/getWaveRecordingCatalog"), params)
-	// 此处无数据也会失败，暂时定义失败为无数据，不报错
-	if err != nil {
-		zap.L().Error("[queryFaultRecordCallableList] Response failed",
+	// 请求失败重试：仅对请求错误（HTTP 层失败）重试 3 次、间隔 5s。
+	// 注意：单纯"无数据"由后端以 200 + code(400,故障目录为空) 返回（err==nil），不会进入重试分支，
+	// 由导出层 QueryFaultRecordCallableList 识别并返回空列表。
+	const maxRetry = 3
+	const retryInterval = 5 * time.Second
+
+	var response *domain.FwhResponse[[]domain.FaultRecordCallableInfo]
+	var err error
+	for attempt := 0; attempt < maxRetry; attempt++ {
+		response, err = FwhHttpGetUnmarshal[[]domain.FaultRecordCallableInfo](getFullEnergyUrl("/hes-gateway/manage/getWaveRecordingCatalog"), params)
+		if err == nil {
+			return response, nil
+		}
+		zap.L().Warn("[queryFaultRecordCallableList] 请求失败，准备重试",
 			zap.Int("devType", devType),
 			zap.String("sn", sn),
+			zap.Int("attempt", attempt+1),
+			zap.Int("maxRetry", maxRetry),
 			zap.String("err", err.Error()),
 		)
-		// 创造假的空数据
-		return &domain.FwhResponse[[]domain.FaultRecordCallableInfo]{
-			Code:   200,
-			Result: &[]domain.FaultRecordCallableInfo{},
-		}, nil
+		if attempt < maxRetry-1 {
+			time.Sleep(retryInterval)
+		}
 	}
-	return response, nil
+
+	// 重试仍失败：返回真实错误（不再吞错误伪造空数据），由上层记录到任务错误信息
+	zap.L().Error("[queryFaultRecordCallableList] 请求失败，已达最大重试次数",
+		zap.Int("devType", devType),
+		zap.String("sn", sn),
+		zap.Int("maxRetry", maxRetry),
+		zap.String("err", err.Error()),
+	)
+	return nil, fmt.Errorf("[queryFaultRecordCallableList] 请求失败(devType=%d, sn=%s, 已重试%d次): %w", devType, sn, maxRetry, err)
 }
 
 // 查询设备可召唤数据列表 - 仅自动录播

service/repositories/mongo/batchDownloadRepo.go

@@ -9,6 +9,7 @@ import (
 
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
@@ -41,6 +42,9 @@ func GetBatchDownloadManageList(page, size uint, sn, progress *string) (*dto.Res
 			"options":     1,
 			"progress":    1,
 			"failReasons": 1,
+			"items":       1,
+			"totalCount":  1,
+			"doneCount":   1,
 		}).
 		SetSkip(int64((page - 1) * size)).
 		SetLimit(int64(size)).
@@ -461,3 +465,97 @@ func InsertBatchDownloadPathSettings(id primitive.ObjectID, sn string, path stri
 
 	return nil
 }
+
+// ==================== 结构化 item 结果（B2 完成判定） ====================
+
+// InitBatchDownloadItems 初始化 items 列表，并设置 totalCount / doneCount。
+// doneCount 初始化为已经处于终态的 item 数量（构建期就 failed/skipped 的），
+// 使得后续所有 item 转终态后 doneCount 恰好等于 totalCount。
+func InitBatchDownloadItems(id primitive.ObjectID, items []domain.BatchDownloadItem) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	collection := database.GetCollection(CollectionName_BatchDownloadConfig)
+
+	doneInit := 0
+	now := time.Now()
+	for i := range items {
+		if items[i].UpdatedAt.IsZero() {
+			items[i].UpdatedAt = now
+		}
+		if items[i].Status != domain.ItemPending {
+			doneInit++
+		}
+	}
+
+	filter := bson.M{"_id": id}
+	update := bson.M{
+		"$set": bson.M{
+			"items":      items,
+			"totalCount": len(items),
+			"doneCount":  doneInit,
+			"progress":   domain.ProgressRunning,
+		},
+	}
+	opts := options.Update().SetUpsert(true)
+
+	_, err := collection.UpdateOne(ctx, filter, update, opts)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// FinishBatchDownloadItem 原子地把某个 pending item 置为终态并递增 doneCount。
+// 幂等：filter 要求该 item 当前为 pending，重复调用（Asynq 重试/重复回调）不会重复计数。
+// 返回 (doneCount, totalCount, error)；若该 item 已非 pending（幂等跳过），返回 (-1, -1, nil)。
+func FinishBatchDownloadItem(id primitive.ObjectID, key string, status domain.ItemStatus, stage, reason string) (int, int, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	collection := database.GetCollection(CollectionName_BatchDownloadConfig)
+
+	filter := bson.M{
+		"_id": id,
+		"items": bson.M{
+			"$elemMatch": bson.M{"key": key, "status": domain.ItemPending},
+		},
+	}
+	update := bson.M{
+		"$set": bson.M{
+			"items.$.status":    status,
+			"items.$.stage":     stage,
+			"items.$.reason":    reason,
+			"items.$.updatedAt": time.Now(),
+		},
+		"$inc": bson.M{"doneCount": 1},
+	}
+	opts := options.FindOneAndUpdate().SetReturnDocument(options.After)
+
+	var doc domain.BatchDownloadConfig
+	err := collection.FindOneAndUpdate(ctx, filter, update, opts).Decode(&doc)
+	if err != nil {
+		if err == mongo.ErrNoDocuments {
+			// item 已非 pending（重复回调/重试），幂等跳过
+			return -1, -1, nil
+		}
+		return 0, 0, err
+	}
+	return doc.DoneCount, doc.TotalCount, nil
+}
+
+// GetBatchDownloadItems 读取某任务的全部 item 结果，用于聚合派生终态。
+func GetBatchDownloadItems(id primitive.ObjectID) ([]domain.BatchDownloadItem, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	collection := database.GetCollection(CollectionName_BatchDownloadConfig)
+
+	var cfg domain.BatchDownloadConfig
+	err := collection.FindOne(ctx, bson.M{"_id": id},
+		options.FindOne().SetProjection(bson.M{"items": 1})).Decode(&cfg)
+	if err != nil {
+		return nil, err
+	}
+	return cfg.Items, nil
+}

service/repositories/repositories.go

@@ -378,6 +378,9 @@ var InsertBatchDownloadConfig = mongo.InsertBatchDownloadConfig
 var UpdateBatchDownloadDescription = mongo.UpdateBatchDownloadDescription
 var UpdateBatchDownloadOwner = mongo.UpdateBatchDownloadOwner
 var UpdateBatchDownloadResult = mongo.UpdateBatchDownloadResult
+var InitBatchDownloadItems = mongo.InitBatchDownloadItems
+var FinishBatchDownloadItem = mongo.FinishBatchDownloadItem
+var GetBatchDownloadItems = mongo.GetBatchDownloadItems
 var DeleteBatchDownloadConfig = mongo.DeleteBatchDownloadConfig
 var InsertBatchDownloadData = mongo.InsertBatchDownloadData
 var GetBatchDownloadData = mongo.GetBatchDownloadData

service/task/queue/fwhQueue/fwhTaskBatchDownload.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"fwh-toolbox-server/cache"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/dto"
 	"fwh-toolbox-server/jobs"
@@ -15,6 +14,7 @@ import (
 	"time"
 
 	"github.com/hibiken/asynq"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.uber.org/zap"
 )
 
@@ -33,115 +33,141 @@ func handleBatchDownload(ctx context.Context, task *asynq.Task) error {
 		return err
 	}
 
-	// 执行主逻辑
-	reasons, err := executeBatchDownload(ctx, payload)
-
-	// 记录结果
-	if err != nil {
-		zap.L().Error("[BatchDownload] 任务执行失败", zap.String("id", mainTaskId), zap.String("err", err.Error()))
+	// 执行主逻辑。
+	// 注意：任务终态（complete/failed）由 item 聚合（maybeAggregate）异步派生写入，
+	// 这里只处理"任务根本无法开始"的致命错误 → ProgressError。
+	if err := executeBatchDownload(ctx, payload); err != nil {
+		zap.L().Error("[BatchDownload] 任务执行失败(致命)", zap.String("id", mainTaskId), zap.String("err", err.Error()))
 		repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressError, []string{err.Error()})
 		return err
 	}
 
-	if len(reasons) == 0 {
-		zap.L().Info("[BatchDownload] 任务执行成功", zap.String("id", mainTaskId))
-		repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressComplete, []string{})
-	} else {
-		zap.L().Warn("[BatchDownload] 任务部分失败", zap.String("id", mainTaskId), zap.Strings("reasons", reasons))
-		repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressFailed, reasons)
-	}
-
 	return nil // 返回 nil 让 Asynq 标记任务为 completed，不触发重试
 }
 
-// executeBatchDownload 执行批量下载逻辑
-func executeBatchDownload(ctx context.Context, payload domain.TaskBatchDownload) ([]string, error) {
+// executeBatchDownload 执行批量下载逻辑。
+// 返回 error 仅表示"任务根本无法开始"的致命错误（→ ProgressError）；
+// 进入执行阶段后的所有成败都以 item 形式记录，终态由聚合派生。
+func executeBatchDownload(ctx context.Context, payload domain.TaskBatchDownload) error {
 	mainTaskId := payload.Id.Hex()
 
-	// 解析时间范围
+	// 解析时间范围（致命错误：早退）
 	startTime, err := time.Parse(time.DateOnly, payload.TimePeriod[0])
 	if err != nil {
-		return nil, fmt.Errorf("startTime 格式错误: %w", err)
+		return fmt.Errorf("startTime 格式错误: %w", err)
 	}
 	endTime, err := time.Parse(time.DateOnly, payload.TimePeriod[1])
 	if err != nil {
-		return nil, fmt.Errorf("endTime 格式错误: %w", err)
+		return fmt.Errorf("endTime 格式错误: %w", err)
 	}
 	endTime = endTime.Add(23*time.Hour + 59*time.Minute + 59*time.Second)
 	if startTime.After(endTime) {
-		return nil, errors.New("startTime > endTime")
+		return errors.New("startTime > endTime")
 	}
 
 	// 按天拆分时间段
 	periods := splitTimePeriods(startTime, endTime)
 
-	// ==================== 阶段1：创建文件下载子任务 ====================
-	callFileTasks := buildCallFileTasks(payload, periods, startTime, endTime)
-	totalSubTasks := len(callFileTasks)
-
-	if totalSubTasks > 0 {
-		// 初始化 Redis 计数器（待完成数）
-		counterKey := callFileCounterPrefix + mainTaskId
-		failuresKey := callFileFailuresPrefix + mainTaskId
-
-		cache.Set(ctx, counterKey, totalSubTasks, 2*time.Hour)
-		cache.Delete(ctx, failuresKey)
-
-		// 入队所有子任务
-		for i, cfTask := range callFileTasks {
-			taskPayload, _ := json.Marshal(cfTask)
-			subTaskID := fmt.Sprintf("callfile-%s-%d", mainTaskId, i)
-			t := asynq.NewTask(TaskTypeCallFile, taskPayload,
-				asynq.MaxRetry(3),
-				asynq.Timeout(35*time.Minute),
-				asynq.Queue(QueueDownload),
-				asynq.TaskID(subTaskID),
-			)
-			if _, err := TaskClient.Enqueue(t); err != nil {
-				zap.L().Error("[BatchDownload] 子任务入队失败",
-					zap.String("id", mainTaskId),
-					zap.Int("index", i),
-					zap.String("err", err.Error()),
-				)
-			}
-		}
+	// ==================== 阶段0：规划所有 item ====================
+	// planItems 枚举所有同步 + 异步工作单元为 pending；构建期查询失败/无数据直接写 failed/skipped。
+	items, asyncTasks := planItems(payload, periods, startTime, endTime)
+
+	if err := repositories.InitBatchDownloadItems(payload.Id, items); err != nil {
+		return fmt.Errorf("初始化 items 失败: %w", err)
 	}
 
-	// ==================== 阶段2：同步处理其他数据任务 ====================
-	if err := executeSyncTasks(ctx, payload, periods); err != nil {
-		return nil, err
+	// 边界：若所有 item 在规划期就已是终态（无异步子任务、无同步任务），主动触发一次聚合。
+	doneInit := 0
+	for _, it := range items {
+		if it.Status != domain.ItemPending {
+			doneInit++
+		}
+	}
+	if doneInit == len(items) {
+		maybeAggregate(payload.Id, doneInit, len(items))
+		return nil
 	}
 
-	// ==================== 阶段3：等待子任务完成 ====================
-	if totalSubTasks > 0 {
-		failReasons, err := waitForCallFileTasks(ctx, mainTaskId)
-		if err != nil {
-			return nil, err
+	// ==================== 阶段1：入队异步文件下载子任务 ====================
+	for i, cfTask := range asyncTasks {
+		taskPayload, _ := json.Marshal(cfTask)
+		subTaskID := fmt.Sprintf("callfile-%s-%d", mainTaskId, i)
+		t := asynq.NewTask(TaskTypeCallFile, taskPayload,
+			asynq.MaxRetry(3),
+			asynq.Timeout(35*time.Minute),
+			asynq.Queue(QueueDownload),
+			asynq.TaskID(subTaskID),
+		)
+		if _, err := TaskClient.Enqueue(t); err != nil {
+			// 入队失败：直接把该 item 记为 failed(build)，doneCount 照常推进，避免永久 pending 挂起。
+			zap.L().Error("[BatchDownload] 子任务入队失败",
+				zap.String("id", mainTaskId),
+				zap.Int("index", i),
+				zap.String("key", cfTask.ItemKey),
+				zap.String("err", err.Error()),
+			)
+			finishItem(payload.Id, cfTask.ItemKey, domain.ItemFailed, domain.StageBuild, "子任务入队失败: "+err.Error())
 		}
-		return failReasons, nil
 	}
 
-	return nil, nil
+	// ==================== 阶段2：同步处理其他数据任务 ====================
+	// 同步任务在主任务内联执行，失败记 item failed 而非中止；每个 item 终态都会驱动 maybeAggregate。
+	executeSyncTasks(ctx, payload, periods)
+
+	return nil
 }
 
 // callFilePayload 是传递给子任务的 payload 结构
 type callFilePayload struct {
 	MainTaskId string                           `json:"mainTaskId"`
+	ItemKey    string                           `json:"itemKey"` // 对应 BatchDownloadItem.Key，子任务据此上报结果
 	TaskInfo   domain.BatchDownloadFileTaskInfo `json:"taskInfo"`
 }
 
-// buildCallFileTasks 构建所有需要异步下载的文件任务列表
-func buildCallFileTasks(payload domain.TaskBatchDownload, periods []timePeriod, startTime, endTime time.Time) []callFilePayload {
+// 同步任务的 item fileType 标识（与 executeSyncTasks 内的 key 生成保持一致）
+const (
+	fileTypePowerFlow     = "PowerFlow"
+	fileTypeMqttLog       = "MqttLog"
+	fileTypeEventAlarm    = "EventAlarm"
+	fileTypeAlarmSnapshot = "AlarmSnapshot"
+	fileTypeUpdateLog     = "UpdateLog"
+)
+
+// planItems 枚举本次批量下载的所有工作单元：
+//   - 返回全部 items（pending / 构建期 failed / skipped）
+//   - 返回需要异步入队的文件下载子任务列表（EmsLog / FaultRecord，payload 带 ItemKey）
+//
+// 同步类任务（PowerFlow/MqttLog/EventAlarm/AlarmSnapshot/UpdateLog）只产出 pending item，
+// 由 executeSyncTasks 内联执行并上报。
+func planItems(payload domain.TaskBatchDownload, periods []timePeriod, startTime, endTime time.Time) ([]domain.BatchDownloadItem, []callFilePayload) {
 	mainTaskId := payload.Id.Hex()
-	var tasks []callFilePayload
+	var items []domain.BatchDownloadItem
+	var asyncTasks []callFilePayload
+
+	newItem := func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem {
+		return domain.BatchDownloadItem{
+			Key:      domain.BuildItemKey(fileType, sn, devType, date, subType),
+			Sn:       sn,
+			DevType:  devType,
+			FileType: fileType,
+			SubType:  subType,
+			Date:     date,
+			Status:   status,
+			Stage:    stage,
+			Reason:   reason,
+		}
+	}
 
 	for _, sn := range payload.Sn {
-		// EmsLog 任务
+		// ---------- EmsLog（异步） ----------
 		for _, period := range periods {
 			dateStr := period.Start.Format(time.DateOnly)
 			for _, logType := range payload.Options.EmsLogs {
-				tasks = append(tasks, callFilePayload{
+				key := domain.BuildItemKey(jobs.FileTypeEmsLog, sn, -1, dateStr, logType)
+				items = append(items, newItem(jobs.FileTypeEmsLog, sn, -1, dateStr, logType, domain.ItemPending, "", ""))
+				asyncTasks = append(asyncTasks, callFilePayload{
 					MainTaskId: mainTaskId,
+					ItemKey:    key,
 					TaskInfo: domain.BatchDownloadFileTaskInfo{
 						DevType:     -1,
 						Sn:          sn,
@@ -153,25 +179,61 @@ func buildCallFileTasks(payload domain.TaskBatchDownload, periods []timePeriod,
 			}
 		}
 
-		// FaultRecord 任务
+		// ---------- FaultRecord（异步） ----------
 		if payload.Options.FaultRecord {
-			frTasks := buildFaultRecordTasks(mainTaskId, sn, startTime, endTime)
-			tasks = append(tasks, frTasks...)
+			frItems, frTasks := planFaultRecordItems(mainTaskId, sn, startTime, endTime, newItem)
+			items = append(items, frItems...)
+			asyncTasks = append(asyncTasks, frTasks...)
+		}
+
+		// ---------- 同步类 ----------
+		if payload.Options.PowerFlow {
+			for _, period := range periods {
+				dateStr := period.Start.Format(time.DateOnly)
+				items = append(items, newItem(fileTypePowerFlow, sn, -1, dateStr, "", domain.ItemPending, "", ""))
+			}
+		}
+		for _, logType := range payload.Options.MqttLogs {
+			for _, period := range periods {
+				dateStr := period.Start.Format(time.DateOnly)
+				items = append(items, newItem(fileTypeMqttLog, sn, -1, dateStr, logType, domain.ItemPending, "", ""))
+			}
+		}
+		if payload.Options.EventAlarm {
+			for _, period := range periods {
+				dateStr := period.Start.Format(time.DateOnly)
+				items = append(items, newItem(fileTypeEventAlarm, sn, -1, dateStr, "", domain.ItemPending, "", ""))
+			}
+		}
+		if payload.Options.AlarmSnapshot {
+			for _, period := range periods {
+				dateStr := period.Start.Format(time.DateOnly)
+				items = append(items, newItem(fileTypeAlarmSnapshot, sn, -1, dateStr, "", domain.ItemPending, "", ""))
+			}
+		}
+		if payload.Options.UpdateLog {
+			items = append(items, newItem(fileTypeUpdateLog, sn, -1, "", "", domain.ItemPending, "", ""))
 		}
 	}
 
-	return tasks
+	return items, asyncTasks
 }
 
-// buildFaultRecordTasks 为单个 SN 构建故障录波下载任务
-func buildFaultRecordTasks(mainTaskId string, sn string, startTime, endTime time.Time) []callFilePayload {
+// planFaultRecordItems 为单个 gateway SN 规划故障录波 item（aGate + 其下 aPower）。
+// 构建期查询失败 → failed(build)；无满足时间范围的数据 → skipped；有数据 → pending + 异步 payload。
+func planFaultRecordItems(mainTaskId string, sn string, startTime, endTime time.Time,
+	newItem func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem,
+) ([]domain.BatchDownloadItem, []callFilePayload) {
+	var items []domain.BatchDownloadItem
 	var tasks []callFilePayload
 
-	// 查询 aGate 可召唤列表
-	aGateTasks := buildFaultRecordTasksForDev(mainTaskId, 1, sn, startTime, endTime)
-	tasks = append(tasks, aGateTasks...)
+	// aGate 可召唤列表
+	frItems, frTasks := planFaultRecordItemsForDev(mainTaskId, 1, sn, startTime, endTime, newItem)
+	items = append(items, frItems...)
+	tasks = append(tasks, frTasks...)
 
-	// 查询 aPower 子设备
+	// aPower 子设备
+	dateRange := startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime)
 	siteBaseInfoList, err := repositories.QuerySiteBaseInfoList(dto.SiteBaseInfoPara{
 		SiteList: []string{sn},
 		PageSize: 10,
@@ -179,33 +241,39 @@ func buildFaultRecordTasks(mainTaskId string, sn string, startTime, endTime time
 	})
 	if err != nil {
 		zap.L().Error("[BatchDownload] 查询站点基本信息失败", zap.String("sn", sn), zap.String("err", err.Error()))
-		return tasks
+		// 站点信息查询失败 → 无法枚举 aPower，记一条 failed(build) 归属该 SN
+		items = append(items, newItem(jobs.FileTypeFaultRecord, sn, 0, dateRange, "", domain.ItemFailed, domain.StageBuild, "查询站点基本信息失败: "+err.Error()))
+		return items, tasks
 	}
 
-	if siteBaseInfoList != nil && len(*siteBaseInfoList.Result) > 0 {
+	if siteBaseInfoList != nil && siteBaseInfoList.Result != nil && len(*siteBaseInfoList.Result) > 0 {
 		aPowerSnList := strings.Split((*siteBaseInfoList.Result)[0].FhpSn, ",")
 		for _, aPowerSn := range aPowerSnList {
 			aPowerSn = strings.TrimSpace(aPowerSn)
 			if aPowerSn == "" {
 				continue
 			}
-			aPowerTasks := buildFaultRecordTasksForDev(mainTaskId, 0, aPowerSn, startTime, endTime)
-			tasks = append(tasks, aPowerTasks...)
+			apItems, apTasks := planFaultRecordItemsForDev(mainTaskId, 0, aPowerSn, startTime, endTime, newItem)
+			items = append(items, apItems...)
+			tasks = append(tasks, apTasks...)
 		}
 	}
 
-	return tasks
+	return items, tasks
 }
 
-// buildFaultRecordTasksForDev 为指定设备构建故障录波下载任务
-func buildFaultRecordTasksForDev(mainTaskId string, devType int, sn string, startTime, endTime time.Time) []callFilePayload {
+// planFaultRecordItemsForDev 为指定设备规划故障录波 item。
+func planFaultRecordItemsForDev(mainTaskId string, devType int, sn string, startTime, endTime time.Time,
+	newItem func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem,
+) ([]domain.BatchDownloadItem, []callFilePayload) {
+	dateRange := startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime)
+
 	callInfoList, err := repositories.QueryFaultRecordCallableList(devType, sn)
 	if err != nil {
-		zap.L().Error("[BatchDownload] 查询故障录波召唤选项失败", zap.String("sn", sn), zap.String("err", err.Error()))
-		return nil
-	}
-	if len(callInfoList) == 0 {
-		return nil
+		zap.L().Error("[BatchDownload] 查询故障录波召唤选项失败", zap.Int("devType", devType), zap.String("sn", sn), zap.String("err", err.Error()))
+		// 构建期查询失败 → failed(build)，错误得以上报
+		item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, "", domain.ItemFailed, domain.StageBuild, "查询故障录波召唤选项失败: "+err.Error())
+		return []domain.BatchDownloadItem{item}, nil
 	}
 
 	// 筛选日期范围内的记录
@@ -226,41 +294,42 @@ func buildFaultRecordTasksForDev(mainTaskId string, devType int, sn string, star
 		}
 	}
 
-	// 没有符合日期范围的记录，不创建任务
+	// 无符合日期范围的记录 → skipped（非错误）
 	if indexList == "" {
-		return nil
+		item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, "", domain.ItemSkipped, "", "无满足时间范围的录波数据")
+		return []domain.BatchDownloadItem{item}, nil
 	}
 
-	return []callFilePayload{{
+	// 有数据 → pending + 异步 payload
+	key := domain.BuildItemKey(jobs.FileTypeFaultRecord, sn, devType, dateRange, indexList)
+	item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, indexList, domain.ItemPending, "", "")
+	task := callFilePayload{
 		MainTaskId: mainTaskId,
+		ItemKey:    key,
 		TaskInfo: domain.BatchDownloadFileTaskInfo{
 			DevType:     devType,
 			Sn:          sn,
-			Date:        startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime),
+			Date:        dateRange,
 			FileType:    jobs.FileTypeFaultRecord,
 			FileSupType: indexList,
 		},
-	}}
+	}
+	return []domain.BatchDownloadItem{item}, []callFilePayload{task}
 }
 
-// executeSyncTasks 同步执行非文件下载类任务（PowerFlow、MqttLogs、EventAlarm 等）
-func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, periods []timePeriod) error {
+// executeSyncTasks 同步执行非文件下载类任务（PowerFlow、MqttLogs、EventAlarm 等）。
+// 每个工作单元成功/失败都通过 finishItem 上报，失败不中止整个任务。
+func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, periods []timePeriod) {
 	for _, sn := range payload.Sn {
 		// PowerFlow
 		if payload.Options.PowerFlow {
 			for _, period := range periods {
 				dateStr := period.Start.Format(time.DateOnly)
-				res, err := repositories.QuerySitePowerFlow(sn, period.Start)
-				if err != nil {
-					zap.L().Error("[BatchDownload] 查询功率流数据失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("err", err.Error()))
-					return err
-				}
-				objName, err := jobs.SaveBatchDownloadPowerFlow(sn, dateStr, *res.Result)
-				if err != nil {
-					return err
-				}
-				if err := repositories.InsertBatchDownloadPathPowerFlow(payload.Id, sn, dateStr, objName); err != nil {
-					return err
+				key := domain.BuildItemKey(fileTypePowerFlow, sn, -1, dateStr, "")
+				if err := syncPowerFlow(payload.Id, sn, dateStr, period); err != nil {
+					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
+				} else {
+					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
 				}
 			}
 		}
@@ -268,17 +337,12 @@ func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, per
 		// MqttLogs
 		for _, logType := range payload.Options.MqttLogs {
 			for _, period := range periods {
-				logList, err := repositories.QueryMqttLogListOnTimeRange(sn, domain.MqttCmdType(logType), period.Start, period.End)
-				if err != nil {
-					return err
-				}
 				dateStr := period.Start.Format(time.DateOnly)
-				objName, err := jobs.SaveBatchDownloadMqttLog(sn, logType, dateStr, logList)
-				if err != nil {
-					return err
-				}
-				if err := repositories.InsertBatchDownloadPathMqttLogs(payload.Id, sn, logType, dateStr, objName); err != nil {
-					return err
+				key := domain.BuildItemKey(fileTypeMqttLog, sn, -1, dateStr, logType)
+				if err := syncMqttLog(payload.Id, sn, logType, dateStr, period); err != nil {
+					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
+				} else {
+					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
 				}
 			}
 		}
@@ -286,17 +350,12 @@ func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, per
 		// EventAlarm
 		if payload.Options.EventAlarm {
 			for _, period := range periods {
-				eventAlarmList, err := repositories.QueryEventOrAlarmListOnTimeRange(sn, period.Start, period.End, repositories.EventAndAlarm)
-				if err != nil {
-					return err
-				}
 				dateStr := period.Start.Format(time.DateOnly)
-				objName, err := jobs.SaveBatchDownloadEventAlarm(sn, dateStr, eventAlarmList)
-				if err != nil {
-					return err
-				}
-				if err := repositories.InsertBatchDownloadPathEventAlarm(payload.Id, sn, dateStr, objName); err != nil {
-					return err
+				key := domain.BuildItemKey(fileTypeEventAlarm, sn, -1, dateStr, "")
+				if err := syncEventAlarm(payload.Id, sn, dateStr, period); err != nil {
+					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
+				} else {
+					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
 				}
 			}
 		}
@@ -305,74 +364,101 @@ func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, per
 		if payload.Options.AlarmSnapshot {
 			for _, period := range periods {
 				dateStr := period.Start.Format(time.DateOnly)
-				alarmSnapshotList, err := jobs.GetAlarmSnapshotData(ctx, sn, dateStr)
-				if err != nil {
-					return err
-				}
-				objName, err := jobs.SaveBatchDownloadAlarmSnapshot(sn, dateStr, alarmSnapshotList)
-				if err != nil {
-					return err
-				}
-				if err := repositories.InsertBatchDownloadPathAlarmSnapshot(payload.Id, sn, dateStr, objName); err != nil {
-					return err
+				key := domain.BuildItemKey(fileTypeAlarmSnapshot, sn, -1, dateStr, "")
+				if err := syncAlarmSnapshot(ctx, payload.Id, sn, dateStr); err != nil {
+					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
+				} else {
+					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
 				}
 			}
 		}
 
 		// UpdateLog
 		if payload.Options.UpdateLog {
+			key := domain.BuildItemKey(fileTypeUpdateLog, sn, -1, "", "")
 			if err := executeSyncUpdateLog(sn, payload); err != nil {
-				return err
+				finishItem(payload.Id, key, domain.ItemFailed, domain.StageQuery, err.Error())
+			} else {
+				finishItem(payload.Id, key, domain.ItemSuccess, "", "")
 			}
 		}
 	}
-	return nil
 }
 
-// waitForCallFileTasks 等待所有子任务完成，收集失败原因
-func waitForCallFileTasks(ctx context.Context, mainTaskId string) ([]string, error) {
-	counterKey := callFileCounterPrefix + mainTaskId
-	failuresKey := callFileFailuresPrefix + mainTaskId
-
-	ticker := time.NewTicker(30 * time.Second)
-	defer ticker.Stop()
-
-	timeout := time.After(90 * time.Minute) // 90 分钟总超时
-
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		case <-timeout:
-			return nil, fmt.Errorf("等待子任务完成超时（90分钟）")
-		case <-ticker.C:
-			val, err := cache.Get(ctx, counterKey)
-			if err != nil {
-				// key 不存在或读取失败
-				if err.Error() == "redis: nil" {
-					// key 不存在，所有子任务已完成
-					val = "0"
-				} else {
-					zap.L().Error("[BatchDownload] 读取计数器失败", zap.String("key", counterKey), zap.String("err", err.Error()))
-					continue
-				}
-			}
+// syncItemError 携带失败阶段与原因，便于 finishItem 记录 stage
+type syncItemError struct {
+	stage  string
+	reason string
+}
 
-			remaining, _ := strconv.Atoi(val)
-			if remaining <= 0 {
-				// 收集失败原因
-				failures, err := cache.LRange(ctx, failuresKey, 0, -1)
-				if err != nil {
-					return nil, fmt.Errorf("读取失败原因列表失败: %w", err)
-				}
+func (e *syncItemError) Error() string { return e.reason }
 
-				// 清理
-				cache.Delete(ctx, counterKey, failuresKey)
+func newSyncErr(stage string, err error) *syncItemError {
+	return &syncItemError{stage: stage, reason: err.Error()}
+}
 
-				return failures, nil
-			}
-		}
+// syncPowerFlow 处理单个 PowerFlow 工作单元
+func syncPowerFlow(id primitive.ObjectID, sn, dateStr string, period timePeriod) *syncItemError {
+	res, err := repositories.QuerySitePowerFlow(sn, period.Start)
+	if err != nil {
+		return newSyncErr(domain.StageQuery, err)
 	}
+	objName, err := jobs.SaveBatchDownloadPowerFlow(sn, dateStr, *res.Result)
+	if err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	if err := repositories.InsertBatchDownloadPathPowerFlow(id, sn, dateStr, objName); err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	return nil
+}
+
+// syncMqttLog 处理单个 MqttLog 工作单元
+func syncMqttLog(id primitive.ObjectID, sn, logType, dateStr string, period timePeriod) *syncItemError {
+	logList, err := repositories.QueryMqttLogListOnTimeRange(sn, domain.MqttCmdType(logType), period.Start, period.End)
+	if err != nil {
+		return newSyncErr(domain.StageQuery, err)
+	}
+	objName, err := jobs.SaveBatchDownloadMqttLog(sn, logType, dateStr, logList)
+	if err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	if err := repositories.InsertBatchDownloadPathMqttLogs(id, sn, logType, dateStr, objName); err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	return nil
+}
+
+// syncEventAlarm 处理单个 EventAlarm 工作单元
+func syncEventAlarm(id primitive.ObjectID, sn, dateStr string, period timePeriod) *syncItemError {
+	eventAlarmList, err := repositories.QueryEventOrAlarmListOnTimeRange(sn, period.Start, period.End, repositories.EventAndAlarm)
+	if err != nil {
+		return newSyncErr(domain.StageQuery, err)
+	}
+	objName, err := jobs.SaveBatchDownloadEventAlarm(sn, dateStr, eventAlarmList)
+	if err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	if err := repositories.InsertBatchDownloadPathEventAlarm(id, sn, dateStr, objName); err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	return nil
+}
+
+// syncAlarmSnapshot 处理单个 AlarmSnapshot 工作单元
+func syncAlarmSnapshot(ctx context.Context, id primitive.ObjectID, sn, dateStr string) *syncItemError {
+	alarmSnapshotList, err := jobs.GetAlarmSnapshotData(ctx, sn, dateStr)
+	if err != nil {
+		return newSyncErr(domain.StageQuery, err)
+	}
+	objName, err := jobs.SaveBatchDownloadAlarmSnapshot(sn, dateStr, alarmSnapshotList)
+	if err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	if err := repositories.InsertBatchDownloadPathAlarmSnapshot(id, sn, dateStr, objName); err != nil {
+		return newSyncErr(domain.StageSave, err)
+	}
+	return nil
 }
 
 // ==================== 辅助函数 ====================
@@ -549,3 +635,64 @@ func executeSyncUpdateLog(sn string, payload domain.TaskBatchDownload) error {
 	}
 	return repositories.InsertBatchDownloadPathUpdateLog(payload.Id, sn, objName)
 }
+
+// ==================== 聚合与完成判定（B2 + 策略C） ====================
+
+// deriveProgressAndSummary 从 items 聚合派生任务终态与摘要 failReasons（策略C：摘要，明细在 items）
+func deriveProgressAndSummary(items []domain.BatchDownloadItem) (domain.Progress, []string) {
+	failed := 0
+	var samples []string
+	for _, it := range items {
+		if it.Status == domain.ItemFailed {
+			failed++
+			if len(samples) < 5 { // 摘要仅保留前若干条样本
+				samples = append(samples, fmt.Sprintf("[%s] sn=%s date=%s stage=%s: %s",
+					it.FileType, it.Sn, it.Date, it.Stage, it.Reason))
+			}
+		}
+	}
+
+	if failed == 0 {
+		return domain.ProgressComplete, []string{}
+	}
+
+	summary := append([]string{fmt.Sprintf("共 %d 项失败（详见明细 items）", failed)}, samples...)
+	return domain.ProgressFailed, summary
+}
+
+// maybeAggregate 在 done==total（本任务全部 item 已终态）时聚合派生终态并写库。
+// 由 FinishBatchDownloadItem 的原子性保证只有一个调用会命中 done==total，聚合仅触发一次。
+// done/total 为 -1 表示幂等跳过（重复回调），不处理。
+func maybeAggregate(id primitive.ObjectID, done, total int) {
+	if done < 0 || total < 0 {
+		return
+	}
+	if done != total {
+		return
+	}
+
+	items, err := repositories.GetBatchDownloadItems(id)
+	if err != nil {
+		zap.L().Error("[BatchDownload] 聚合读取 items 失败", zap.String("id", id.Hex()), zap.String("err", err.Error()))
+		return
+	}
+
+	progress, summary := deriveProgressAndSummary(items)
+	if err := repositories.UpdateBatchDownloadResult(id, progress, summary); err != nil {
+		zap.L().Error("[BatchDownload] 聚合写入结果失败", zap.String("id", id.Hex()), zap.String("err", err.Error()))
+		return
+	}
+	zap.L().Info("[BatchDownload] 任务聚合完成", zap.String("id", id.Hex()), zap.String("progress", string(progress)))
+}
+
+// finishItem 是对 repositories.FinishBatchDownloadItem + maybeAggregate 的封装，
+// 主任务同步任务与子任务共用，保证每个 item 转终态后驱动一次完成检查。
+func finishItem(id primitive.ObjectID, key string, status domain.ItemStatus, stage, reason string) {
+	done, total, err := repositories.FinishBatchDownloadItem(id, key, status, stage, reason)
+	if err != nil {
+		zap.L().Error("[BatchDownload] 更新 item 结果失败",
+			zap.String("id", id.Hex()), zap.String("key", key), zap.String("err", err.Error()))
+		return
+	}
+	maybeAggregate(id, done, total)
+}

service/task/queue/fwhQueue/fwhTaskCallFile.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"fwh-toolbox-server/cache"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/jobs"
 	"fwh-toolbox-server/repositories"
@@ -22,78 +21,58 @@ const maxCallRetry = 3
 
 // handleCallFile 是文件下载子任务的 Asynq handler
 // Asynq 本身会做重试（MaxRetry=3），所以 handler 内部不再做外层重试循环
-// 返回 error 会触发 Asynq 重试，返回 nil 表示成功
+// 返回 error 会触发 Asynq 重试，返回 nil 表示成功（结果已通过 finishItem 上报）
 func handleCallFile(ctx context.Context, task *asynq.Task) error {
 	var payload callFilePayload
 	if err := json.Unmarshal(task.Payload(), &payload); err != nil {
-		// payload 解析失败不重试
-		reportCallFileResult(ctx, payload.MainTaskId, fmt.Sprintf("unmarshal callFile payload: %s", err.Error()))
+		// payload 解析失败无法定位 item，只能记日志，不重试
+		zap.L().Error("[CallFile] unmarshal payload 失败", zap.String("err", err.Error()))
 		return nil
 	}
 
 	info := payload.TaskInfo
 	mainTaskId := payload.MainTaskId
+	mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
 
-	var err error
+	var stageErr *syncItemError
 	switch info.FileType {
 	case jobs.FileTypeEmsLog:
-		err = handleEmsLogDownload(ctx, mainTaskId, info)
+		stageErr = handleEmsLogDownload(ctx, mainTaskId, info)
 	case jobs.FileTypeFaultRecord:
-		err = handleFaultRecordDownload(ctx, mainTaskId, info)
+		stageErr = handleFaultRecordDownload(ctx, mainTaskId, info)
 	default:
-		err = fmt.Errorf("unknown file type: %s", info.FileType)
+		stageErr = &syncItemError{stage: domain.StageBuild, reason: fmt.Sprintf("unknown file type: %s", info.FileType)}
 	}
 
-	if err != nil {
+	if stageErr != nil {
 		// 判断是否还有重试机会
 		retryCount, _ := asynq.GetRetryCount(ctx)
 		maxRetry, _ := asynq.GetMaxRetry(ctx)
 		if retryCount >= maxRetry-1 {
-			// 最后一次重试仍失败，记录失败原因并完成计数
-			reportCallFileResult(ctx, mainTaskId, err.Error())
-			return nil // 返回 nil 不再重试
+			// 最后一次重试仍失败：记 failed item（带 stage），推进完成计数
+			finishItem(mainObjId, payload.ItemKey, domain.ItemFailed, stageErr.stage, stageErr.reason)
+			return nil // 不再重试
 		}
-		return err // 返回 error 触发 Asynq 重试
+		return stageErr // 触发 Asynq 重试
 	}
 
-	// 成功：递减计数器
-	reportCallFileResult(ctx, mainTaskId, "")
+	// 成功
+	finishItem(mainObjId, payload.ItemKey, domain.ItemSuccess, "", "")
 	return nil
 }
 
-// reportCallFileResult 报告子任务完成结果
-// reason 为空表示成功，非空表示失败
-func reportCallFileResult(ctx context.Context, mainTaskId string, reason string) {
-	if mainTaskId == "" {
-		return
-	}
-
-	counterKey := callFileCounterPrefix + mainTaskId
-	failuresKey := callFileFailuresPrefix + mainTaskId
-
-	// 递减计数器
-	cache.Decr(ctx, counterKey)
-
-	// 如果失败，记录失败原因
-	if reason != "" {
-		cache.LPush(ctx, failuresKey, reason)
-		// 设置 failures key 过期时间
-		cache.Expire(ctx, failuresKey, 2*time.Hour)
-	}
-}
-
 // ==================== EmsLog 下载 ====================
 
-func handleEmsLogDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) error {
+func handleEmsLogDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) *syncItemError {
 	// 1. 召唤 EmsLog
 	if err := callEmsLog(info.Sn, info.Date, info.FileSupType); err != nil {
-		return fmt.Errorf("[EmsLog] 召唤失败(sn=%s, date=%s, type=%s): %w", info.Sn, info.Date, info.FileSupType, err)
+		return &syncItemError{stage: domain.StageCall, reason: fmt.Sprintf("[EmsLog] 召唤失败(sn=%s, date=%s, type=%s): %s", info.Sn, info.Date, info.FileSupType, err.Error())}
 	}
 
 	// 2. 轮询等待召唤结果
-	recordInfo, err := pollEmsLogResult(ctx, info.Sn, info.Date, info.FileSupType)
-	if err != nil {
-		return err
+	recordInfo, sErr := pollEmsLogResult(ctx, info.Sn, info.Date, info.FileSupType)
+	if sErr != nil {
+		return sErr
 	}
 
 	// 3. 下载并保存文件
@@ -101,7 +80,7 @@ func handleEmsLogDownload(ctx context.Context, mainTaskId string, info domain.Ba
 	return downloadAndSaveEmsLog(ctx, mainObjId, info, recordInfo)
 }
 
-func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domain.EmsLogRecordInfo, error) {
+func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domain.EmsLogRecordInfo, *syncItemError) {
 	ticker := time.NewTicker(time.Minute)
 	defer ticker.Stop()
 	timeout := time.After(30 * time.Minute)
@@ -109,9 +88,9 @@ func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domai
 	for {
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return nil, &syncItemError{stage: domain.StagePoll, reason: ctx.Err().Error()}
 		case <-timeout:
-			return nil, fmt.Errorf("[EmsLog] 轮询超时(sn=%s, date=%s)", sn, date)
+			return nil, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[EmsLog] 轮询超时(sn=%s, date=%s)", sn, date)}
 		case <-ticker.C:
 			for attempt := 0; attempt < maxCallRetry; attempt++ {
 				recordInfo, err := getEmsLogUrl(sn, date, fileSupType)
@@ -119,7 +98,7 @@ func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domai
 					continue
 				}
 				if recordInfo.Status == -1 {
-					return nil, fmt.Errorf("[EmsLog] 召唤失败(sn=%s)", sn)
+					return nil, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[EmsLog] 召唤失败(sn=%s)", sn)}
 				}
 				if recordInfo.Status == 1 {
 					return recordInfo, nil
@@ -131,7 +110,7 @@ func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domai
 	}
 }
 
-func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, info domain.BatchDownloadFileTaskInfo, recordInfo *domain.EmsLogRecordInfo) error {
+func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, info domain.BatchDownloadFileTaskInfo, recordInfo *domain.EmsLogRecordInfo) *syncItemError {
 	var dataBytes []byte
 	var err error
 	for attempt := 0; attempt < maxCallRetry; attempt++ {
@@ -142,7 +121,7 @@ func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, i
 		zap.L().Warn("[EmsLog] 文件下载重试", zap.String("sn", info.Sn), zap.Int("attempt", attempt))
 	}
 	if err != nil {
-		return fmt.Errorf("[EmsLog] 文件下载失败(sn=%s): %w", info.Sn, err)
+		return &syncItemError{stage: domain.StageDownload, reason: fmt.Sprintf("[EmsLog] 文件下载失败(sn=%s): %s", info.Sn, err.Error())}
 	}
 
 	// 提取文件名
@@ -154,22 +133,22 @@ func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, i
 	// 解析召唤时间
 	loc, err := time.LoadLocation("America/Los_Angeles")
 	if err != nil {
-		return fmt.Errorf("[EmsLog] 时区加载失败: %w", err)
+		return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[EmsLog] 时区加载失败: %s", err.Error())}
 	}
 	callTime, err := time.ParseInLocation(time.DateTime, recordInfo.CallTime, loc)
 	if err != nil {
-		return fmt.Errorf("[EmsLog] CallTime解析失败: %w", err)
+		return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[EmsLog] CallTime解析失败: %s", err.Error())}
 	}
 
 	// 保存到 MinIO
 	objName, err := jobs.SaveBatchDownloadEmsLog(info.Sn, dataBytes, baseStr, info.FileSupType, info.Date, callTime)
 	if err != nil {
-		return fmt.Errorf("[EmsLog] 文件保存失败(sn=%s): %w", info.Sn, err)
+		return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[EmsLog] 文件保存失败(sn=%s): %s", info.Sn, err.Error())}
 	}
 
 	// 写入路径记录
 	if err := repositories.InsertBatchDownloadPathEmsLogs(mainTaskId, info.Sn, info.Date, info.FileSupType, objName); err != nil {
-		return fmt.Errorf("[EmsLog] 路径写入失败(sn=%s): %w", info.Sn, err)
+		return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[EmsLog] 路径写入失败(sn=%s): %s", info.Sn, err.Error())}
 	}
 
 	return nil
@@ -177,28 +156,27 @@ func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, i
 
 // ==================== FaultRecord 下载 ====================
 
-func handleFaultRecordDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) error {
+func handleFaultRecordDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) *syncItemError {
 	// 1. 解析参数
 	timeRange := strings.Split(info.Date, ",")
 	if len(timeRange) != 2 {
-		return fmt.Errorf("[FaultRecord] timeRange 格式错误: %s", info.Date)
+		return &syncItemError{stage: domain.StageBuild, reason: fmt.Sprintf("[FaultRecord] timeRange 格式错误: %s", info.Date)}
 	}
 
 	// 2. 查询故障录波列表，判断是否存在满足时间范围的数据
 	existingRecords, err := repositories.QueryFaultRecordList(info.Sn, timeRange[0], timeRange[1])
 	if err != nil {
 		zap.L().Error("[FaultRecord] 查询已有录波列表失败", zap.String("sn", info.Sn), zap.String("err", err.Error()))
-		return err
+		return &syncItemError{stage: domain.StageQuery, reason: fmt.Sprintf("[FaultRecord] 查询已有录波列表失败(sn=%s): %s", info.Sn, err.Error())}
 	} else {
 		// 判断列表中是否存在 earliestFaultTime 在时间范围内的记录
 		faultID, err := checkFaultRecordInTimeRange(existingRecords, info.DevType, info.Sn, timeRange[0], timeRange[1])
 		if err != nil {
 			zap.L().Error("[FaultRecord] 判断时间范围失败", zap.String("sn", info.Sn), zap.String("err", err.Error()))
-			return err
+			return &syncItemError{stage: domain.StageQuery, reason: fmt.Sprintf("[FaultRecord] 判断时间范围失败(sn=%s): %s", info.Sn, err.Error())}
 		}
 		if faultID >= 0 {
-			// 故障录波文件存在，直接下载
-			// 下载所有子记录文件
+			// 故障录波文件存在，直接下载所有子记录文件
 			mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
 			return downloadAndSaveFaultRecords(ctx, mainObjId, faultID)
 		}
@@ -211,18 +189,18 @@ func handleFaultRecordDownload(ctx context.Context, mainTaskId string, info doma
 
 	indexList, err := parseIndexList(info.FileSupType)
 	if err != nil {
-		return fmt.Errorf("[FaultRecord] indexList 解析错误: %w", err)
+		return &syncItemError{stage: domain.StageBuild, reason: fmt.Sprintf("[FaultRecord] indexList 解析错误: %s", err.Error())}
 	}
 
 	// 3. 召唤故障录波
 	if err := callFaultRecord(info.DevType, info.Sn, timeRange[0], timeRange[1], indexList); err != nil {
-		return fmt.Errorf("[FaultRecord] 召唤失败(sn=%s): %w", info.Sn, err)
+		return &syncItemError{stage: domain.StageCall, reason: fmt.Sprintf("[FaultRecord] 召唤失败(sn=%s): %s", info.Sn, err.Error())}
 	}
 
 	// 4. 轮询等待召唤结果
-	faultId, err := pollFaultRecordResult(ctx, info.DevType, info.Sn, timeRange[0], timeRange[1])
-	if err != nil {
-		return err
+	faultId, sErr := pollFaultRecordResult(ctx, info.DevType, info.Sn, timeRange[0], timeRange[1])
+	if sErr != nil {
+		return sErr
 	}
 
 	// 5. 下载所有子记录文件
@@ -292,7 +270,7 @@ func checkFaultRecordInTimeRange(records []domain.FaultRecordInfo, devType int,
 	return -1, nil
 }
 
-func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endTime string) (int64, error) {
+func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endTime string) (int64, *syncItemError) {
 	ticker := time.NewTicker(time.Minute)
 	defer ticker.Stop()
 	timeout := time.After(30 * time.Minute)
@@ -300,9 +278,9 @@ func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endT
 	for {
 		select {
 		case <-ctx.Done():
-			return 0, ctx.Err()
+			return 0, &syncItemError{stage: domain.StagePoll, reason: ctx.Err().Error()}
 		case <-timeout:
-			return 0, fmt.Errorf("[FaultRecord] 轮询超时(sn=%s)", sn)
+			return 0, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[FaultRecord] 轮询超时(sn=%s)", sn)}
 		case <-ticker.C:
 			for attempt := 0; attempt < maxCallRetry; attempt++ {
 				status, faultId, reason, err := getFaultRecordUrl(devType, sn, startTime, endTime)
@@ -310,7 +288,7 @@ func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endT
 					continue
 				}
 				if status == -1 {
-					return 0, fmt.Errorf("[FaultRecord] 召唤失败(sn=%s): %s", sn, reason)
+					return 0, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[FaultRecord] 召唤失败(sn=%s): %s", sn, reason)}
 				}
 				if status == 2 {
 					return faultId, nil
@@ -322,10 +300,10 @@ func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endT
 	}
 }
 
-func downloadAndSaveFaultRecords(ctx context.Context, mainTaskId primitive.ObjectID, faultId int64) error {
+func downloadAndSaveFaultRecords(ctx context.Context, mainTaskId primitive.ObjectID, faultId int64) *syncItemError {
 	subRecordList, err := repositories.QueryFaultRecordSubList(faultId)
 	if err != nil {
-		return fmt.Errorf("[FaultRecord] 子记录查询失败: %w", err)
+		return &syncItemError{stage: domain.StageDownload, reason: fmt.Sprintf("[FaultRecord] 子记录查询失败(faultId=%d): %s", faultId, err.Error())}
 	}
 
 	for _, subRecord := range subRecordList {
@@ -343,18 +321,18 @@ func downloadAndSaveFaultRecords(ctx context.Context, mainTaskId primitive.Objec
 			)
 		}
 		if downloadErr != nil {
-			return fmt.Errorf("[FaultRecord] 文件下载失败(sn=%s, id=%d): %w", subRecord.GatewayId, subRecord.Id, downloadErr)
+			return &syncItemError{stage: domain.StageDownload, reason: fmt.Sprintf("[FaultRecord] 文件下载失败(sn=%s, id=%d): %s", subRecord.GatewayId, subRecord.Id, downloadErr.Error())}
 		}
 
 		// 保存到 MinIO
 		objName, err := jobs.SaveBatchDownloadFaultRecord(subRecord.GatewayId, dataBytes, subRecord.FileName+".xls")
 		if err != nil {
-			return fmt.Errorf("[FaultRecord] 文件保存失败(sn=%s): %w", subRecord.GatewayId, err)
+			return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[FaultRecord] 文件保存失败(sn=%s): %s", subRecord.GatewayId, err.Error())}
 		}
 
 		// 写入路径记录
 		if err := repositories.InsertBatchDownloadPathFaultRecord(mainTaskId, subRecord.GatewayId, subRecord.FileName, objName); err != nil {
-			return fmt.Errorf("[FaultRecord] 路径写入失败(sn=%s): %w", subRecord.GatewayId, err)
+			return &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[FaultRecord] 路径写入失败(sn=%s): %s", subRecord.GatewayId, err.Error())}
 		}
 	}
 

service/task/queue/fwhQueue/fwhTaskQueue.go

@@ -27,14 +27,6 @@ const (
 	QueueDownload = "fwh:download" // 文件下载队列（可并发）
 )
 
-// Redis key 前缀
-const (
-	// 子任务计数器 key: fwh:callfile:counter:{mainTaskId}
-	callFileCounterPrefix = "fwh:callfile:counter:"
-	// 子任务失败原因 key: fwh:callfile:failures:{mainTaskId}
-	callFileFailuresPrefix = "fwh:callfile:failures:"
-)
-
 // 模块级变量
 var (
 	TaskClient *asynq.Client

src/service/batchDownload/interface.ts

@@ -26,6 +26,26 @@ export interface BatchDownloadOptions {
 
 export interface BatchDownloadItemPage extends ResPage<BatchDownloadInfo[]> {}
 
+// 单个下载工作单元的状态
+export type ItemStatus = 'pending' | 'success' | 'failed' | 'skipped'
+
+// 失败发生的阶段
+export type ItemStage = '' | 'build' | 'query' | 'call' | 'poll' | 'download' | 'save'
+
+// 结构化的下载工作单元结果（策略C：items 为权威明细，failReasons 为摘要）
+export interface BatchDownloadItem {
+  key: string // 唯一键 fileType|sn|devType|date|subType
+  sn: string
+  devType: number // -1 表示不涉及设备类型
+  fileType: string
+  subType: string
+  date: string
+  status: ItemStatus
+  stage: ItemStage
+  reason: string
+  updatedAt: string
+}
+
 export interface BatchDownloadInfo {
   id: string
   sn: string[]
@@ -33,7 +53,10 @@ export interface BatchDownloadInfo {
   timePeriod: [string, string]
   options: BatchDownloadOptions
   progress: string
-  failReasons: string[]
+  failReasons: string[] // 策略C：摘要文本，明细见 items
+  items?: BatchDownloadItem[] // 结构化明细
+  totalCount?: number // 工作单元总数
+  doneCount?: number // 已完成（终态）单元数
   owner?: string // 未设置则只有管理员能查看
 }
 

src/views/batch-download/index.vue

@@ -62,7 +62,7 @@
 
 <script setup lang="tsx">
 import { useRouter } from 'vue-router'
-import { NButton, NDataTable, NForm, NNumberAnimation, NInput, NEllipsis, NDrawer, NDrawerContent } from 'naive-ui'
+import { NButton, NDataTable, NForm, NNumberAnimation, NInput, NEllipsis, NTooltip, NDrawer, NDrawerContent } from 'naive-ui'
 import type { DataTableColumns, PaginationProps, DropdownProps, EllipsisProps } from 'naive-ui'
 import { useDataTable } from '@/composable/comp/useDataTable'
 import { useResettableReactive } from '@/composable/useResettable'
@@ -240,7 +240,17 @@ function createColumns(): DataTableColumns<BatchDownloadInfo> {
     {
       title: '进度',
       key: 'progress',
-      width: 100,
+      width: 120,
+      render(rowData) {
+        const total = rowData.totalCount ?? 0
+        const done = rowData.doneCount ?? 0
+        return h('div', { style: { whiteSpace: 'pre-line' } }, [
+          h('div', rowData.progress),
+          total > 0
+            ? h('div', { style: { fontSize: '12px', color: '#888' } }, `${done}/${total}`)
+            : null,
+        ])
+      },
     },
     {
       title: '所有者',
@@ -254,22 +264,55 @@ function createColumns(): DataTableColumns<BatchDownloadInfo> {
       title: '任务结果',
       key: 'failReasons',
       render(rowData, rowIndex) {
-        return h(NEllipsis, {
-          lineClamp: 3,
-          tooltip: {
-            placement: "left-start",   // ⬅️ Tooltip 显示在右侧
-            style: {
-              maxHeight: "200px",       // ⬅️ 多行时滚动
-              overflowY: "auto",
-              whiteSpace: "pre-line",   // ⬅️ \n 正常换行
-              maxWidth: "480px"
-            }
+        // 策略C：主文本用 failReasons 摘要；hover 展示 items 中失败项的结构化明细
+        const failedItems = (rowData.items ?? []).filter((it) => it.status === 'failed')
+
+        // 主文本：优先摘要，其次根据 items 状态给出简述
+        let summaryText = rowData.failReasons?.join('\n') ?? ''
+        if (!summaryText) {
+          if (failedItems.length > 0) {
+            summaryText = `共 ${failedItems.length} 项失败`
+          } else if (rowData.progress === 'complete') {
+            summaryText = '全部成功'
           }
-        } as EllipsisProps,
+        }
+
+        const summaryNode = h(
+          NEllipsis,
+          { lineClamp: 3 } as EllipsisProps,
+          { default: () => h('span', { style: { whiteSpace: 'pre-line' } }, summaryText) },
+        )
+
+        // 无失败明细：仅展示摘要
+        if (failedItems.length === 0) {
+          return summaryNode
+        }
+
+        // 有失败明细：用 Tooltip 展示结构化明细（每项 类型/SN/设备/日期/阶段/原因）
+        const detailLines = failedItems
+          .map(
+            (it) =>
+              `[${it.fileType}] sn=${it.sn}${it.devType >= 0 ? ` devType=${it.devType}` : ''}` +
+              `${it.date ? ` date=${it.date}` : ''}${it.stage ? ` stage=${it.stage}` : ''}\n  ${it.reason}`,
+          )
+          .join('\n')
+
+        return h(
+          NTooltip,
           {
-            default: () =>
-              h("span", { style: { whiteSpace: "pre-line" } }, rowData.failReasons?.join("\n"))
-          })
+            placement: 'left-start',
+            style: {
+              maxHeight: '300px',
+              overflowY: 'auto',
+              whiteSpace: 'pre-line',
+              maxWidth: '520px',
+            },
+          },
+          {
+            trigger: () => summaryNode,
+            default: () => h('span', { style: { whiteSpace: 'pre-line' } }, detailLines),
+          },
+        )
       },
     },
     {