FranklinWH-Toolbox/service/task/queue/fwhQueue/fwhTaskCallFile.go

package fwhQueue

import (
	"context"
	"encoding/json"
	"fmt"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/jobs"
	"fwh-toolbox-server/repositories"
	"path"
	"strconv"
	"strings"
	"time"

	"github.com/hibiken/asynq"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

const maxCallRetry = 3

// handleCallFile 是文件下载子任务的 Asynq handler
// Asynq 本身会做重试（MaxRetry=3），所以 handler 内部不再做外层重试循环
// 返回 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 解析失败无法定位 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 stageErr *syncItemError
	switch info.FileType {
	case jobs.FileTypeEmsLog:
		stageErr = handleEmsLogDownload(ctx, mainTaskId, info)
	case jobs.FileTypeFaultRecord:
		stageErr = handleFaultRecordDownload(ctx, mainTaskId, info)
	default:
		stageErr = &syncItemError{stage: domain.StageBuild, reason: fmt.Sprintf("unknown file type: %s", info.FileType)}
	}

	if stageErr != nil {
		// 判断是否还有重试机会
		retryCount, _ := asynq.GetRetryCount(ctx)
		maxRetry, _ := asynq.GetMaxRetry(ctx)
		if retryCount >= maxRetry-1 {
			// 最后一次重试仍失败：记 failed item（带 stage），推进完成计数
			finishItem(mainObjId, payload.ItemKey, domain.ItemFailed, stageErr.stage, stageErr.reason)
			return nil // 不再重试
		}
		return stageErr // 触发 Asynq 重试
	}

	// 成功
	finishItem(mainObjId, payload.ItemKey, domain.ItemSuccess, "", "")
	return nil
}

// ==================== EmsLog 下载 ====================

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 &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, sErr := pollEmsLogResult(ctx, info.Sn, info.Date, info.FileSupType)
	if sErr != nil {
		return sErr
	}

	// 3. 下载并保存文件
	mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
	return downloadAndSaveEmsLog(ctx, mainObjId, info, recordInfo)
}

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)

	for {
		select {
		case <-ctx.Done():
			return nil, &syncItemError{stage: domain.StagePoll, reason: ctx.Err().Error()}
		case <-timeout:
			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)
				if err != nil {
					continue
				}
				if recordInfo.Status == -1 {
					return nil, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[EmsLog] 召唤失败(sn=%s)", sn)}
				}
				if recordInfo.Status == 1 {
					return recordInfo, nil
				}
				// status == 0: 还在处理中
				break
			}
		}
	}
}

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++ {
		dataBytes, err = repositories.QueryDownloadEmsLogFile(recordInfo.FileUrl)
		if err == nil {
			break
		}
		zap.L().Warn("[EmsLog] 文件下载重试", zap.String("sn", info.Sn), zap.Int("attempt", attempt))
	}
	if err != nil {
		return &syncItemError{stage: domain.StageDownload, reason: fmt.Sprintf("[EmsLog] 文件下载失败(sn=%s): %s", info.Sn, err.Error())}
	}

	// 提取文件名
	baseStr := path.Base(recordInfo.FileUrl)
	if baseStr == "." || baseStr == "/" {
		baseStr = ""
	}

	// 解析召唤时间
	loc, err := time.LoadLocation("America/Los_Angeles")
	if err != nil {
		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 &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 &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 &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[EmsLog] 路径写入失败(sn=%s): %s", info.Sn, err.Error())}
	}

	return nil
}

// ==================== FaultRecord 下载 ====================

func handleFaultRecordDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) *syncItemError {
	// 1. 解析参数
	timeRange := strings.Split(info.Date, ",")
	if len(timeRange) != 2 {
		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 &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 &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)
		}
		zap.L().Info("[FaultRecord] 无满足时间范围的录波数据，需要召唤",
			zap.String("sn", info.Sn),
			zap.String("startTime", timeRange[0]),
			zap.String("endTime", timeRange[1]),
		)
	}

	indexList, err := parseIndexList(info.FileSupType)
	if err != nil {
		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 &syncItemError{stage: domain.StageCall, reason: fmt.Sprintf("[FaultRecord] 召唤失败(sn=%s): %s", info.Sn, err.Error())}
	}

	// 4. 轮询等待召唤结果
	faultId, sErr := pollFaultRecordResult(ctx, info.DevType, info.Sn, timeRange[0], timeRange[1])
	if sErr != nil {
		return sErr
	}

	// 5. 下载所有子记录文件
	mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
	return downloadAndSaveFaultRecords(ctx, mainObjId, faultId)
}

// checkFaultRecordInTimeRange 检查故障录波列表中是否存在 earliestFaultTime 满足时间范围的记录，返回FaultID和错误信息，不在范围内FaultID = -1
func checkFaultRecordInTimeRange(records []domain.FaultRecordInfo, devType int, sn string, startTime string, endTime string) (int64, error) {
	start, err := time.Parse(time.DateTime, startTime)
	if err != nil {
		zap.L().Error("[checkFaultRecordInTimeRange] 解析startTime失败",
			zap.String("startTime", startTime),
			zap.String("err", err.Error()),
		)
		return -1, err // 解析失败，保守处理：继续召唤
	}
	end, err := time.Parse(time.DateTime, endTime)
	if err != nil {
		zap.L().Error("[checkFaultRecordInTimeRange] 解析endTime失败",
			zap.String("endTime", endTime),
			zap.String("err", err.Error()),
		)
		return -1, err
	}

	for _, record := range records {
		// 匹配设备类型和 SN
		if record.DeviceType != devType {
			continue
		}
		var recordSn string
		switch devType {
		case 0: // aPower
			if record.ApowerId != nil {
				recordSn = *record.ApowerId
			}
		case 1: // aGate
			recordSn = record.GatewayId
		case 2: // MPPT
			if record.ApowerId != nil {
				recordSn = *record.ApowerId
			}
		case 3: // aHub
			if record.PowerlinkId != nil {
				recordSn = *record.PowerlinkId
			}
		}
		if recordSn != sn {
			continue
		}

		// 解析 earliestFaultTime 并判断是否在时间范围内
		faultTime, err := time.Parse(time.DateTime, record.EarliestFaultTime)
		if err != nil {
			// 尝试时间戳格式
			continue
		}

		inRange := (faultTime.Equal(start) || faultTime.After(start)) &&
			(faultTime.Equal(end) || faultTime.Before(end))
		if inRange {
			return record.Id, nil
		}
	}

	return -1, nil
}

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)

	for {
		select {
		case <-ctx.Done():
			return 0, &syncItemError{stage: domain.StagePoll, reason: ctx.Err().Error()}
		case <-timeout:
			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)
				if err != nil {
					continue
				}
				if status == -1 {
					return 0, &syncItemError{stage: domain.StagePoll, reason: fmt.Sprintf("[FaultRecord] 召唤失败(sn=%s): %s", sn, reason)}
				}
				if status == 2 {
					return faultId, nil
				}
				// status == 0: 还在处理中
				break
			}
		}
	}
}

func downloadAndSaveFaultRecords(ctx context.Context, mainTaskId primitive.ObjectID, faultId int64) *syncItemError {
	subRecordList, err := repositories.QueryFaultRecordSubList(faultId)
	if err != nil {
		return &syncItemError{stage: domain.StageDownload, reason: fmt.Sprintf("[FaultRecord] 子记录查询失败(faultId=%d): %s", faultId, err.Error())}
	}

	for _, subRecord := range subRecordList {
		var dataBytes []byte
		var downloadErr error
		for attempt := 0; attempt < maxCallRetry; attempt++ {
			_, dataBytes, downloadErr = repositories.QueryDownloadFaultRecordData(faultId, subRecord.Id)
			if downloadErr == nil {
				break
			}
			zap.L().Warn("[FaultRecord] 文件下载重试",
				zap.String("sn", subRecord.GatewayId),
				zap.Int64("subId", subRecord.Id),
				zap.Int("attempt", attempt),
			)
		}
		if downloadErr != nil {
			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 &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 &syncItemError{stage: domain.StageSave, reason: fmt.Sprintf("[FaultRecord] 路径写入失败(sn=%s): %s", subRecord.GatewayId, err.Error())}
		}
	}

	return nil
}

// ==================== 辅助函数（复用原有逻辑） ====================

// EMS日志类型map
var mapEmsLogType = map[string]int{
	"EmsRunLog":     4,
	"EmsBattLog":    7,
	"EmsMonitorLog": 9,
	"Ems4gLog":      10,
	"EmsAwsLog":     11,
	"EmsMainLog":    12,
}

func callEmsLog(sn string, date string, logType string) error {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		return fmt.Errorf("logType不存在: %s", logType)
	}
	return repositories.QueryCallEmsLog(sn, date, logTypeNum)
}

func getEmsLogUrl(sn string, date string, logType string) (*domain.EmsLogRecordInfo, error) {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		return nil, fmt.Errorf("logType不存在: %s", logType)
	}
	return repositories.QueryEmsLogCallRecord(sn, date, logTypeNum)
}

func callFaultRecord(devType int, sn string, startTime string, endTime string, indexList []int) error {
	return repositories.QueryCellFaultRecord(devType, sn, indexList)
}

func getFaultRecordUrl(devType int, sn string, startTime string, endTime string) (int, int64, string, error) {
	recordList, err := repositories.QueryFaultRecordCallRecordList()
	if err != nil {
		return 0, 0, "", err
	}

	for _, record := range recordList {
		if record.DeviceType == devType {
			var devSn string
			switch devType {
			case 0:
				devSn = *record.ApowerId
			case 1:
				devSn = record.GatewayId
			case 2:
				devSn = *record.ApowerId
			case 3:
				devSn = *record.PowerlinkId
			default:
				continue
			}

			if devSn == sn {
				if *record.Status == -1 {
					return *record.Status, record.Id, *record.FaultReason, nil
				}
				return *record.Status, record.Id, "", nil
			}
		}
	}

	return 0, 0, "", fmt.Errorf("未找到召唤匹配记录(devType=%d, sn=%s)", devType, sn)
}

func parseIndexList(fileSupType string) ([]int, error) {
	indexStrList := strings.Split(fileSupType, ",")
	indexList := make([]int, 0, len(indexStrList))
	for _, s := range indexStrList {
		idx, err := strconv.Atoi(s)
		if err != nil {
			return nil, fmt.Errorf("无法解析 index '%s': %w", s, err)
		}
		indexList = append(indexList, idx)
	}
	return indexList, nil
}