feat(storage): 添加阿里云OSS存储支持

- 在StorageConfig中新增AliOSS字段以配置阿里云OSS参数
- 新增AliOSSConfig结构体定义阿里云OSS相关配置项
- 在fs包中实现AliOSSAdapter适配器用于操作阿里云OSS
- 实现GetFileList方法从阿里云OSS获取并缓存文件列表
- 添加定时清理过期缓存的功能
- 更新adapter.go根据存储类型选择对应的适配器实例

fs/ali_adapter.go

@@ -0,0 +1,203 @@
+package fs
+
+import (
+	"ZhenTuLocalPassiveAdapter/config"
+	"ZhenTuLocalPassiveAdapter/dto"
+	"ZhenTuLocalPassiveAdapter/logger"
+	"ZhenTuLocalPassiveAdapter/util"
+	"context"
+	"fmt"
+	"path"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/aliyun/aliyun-oss-go-sdk/oss"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.uber.org/zap"
+)
+
+var aliOssCache sync.Map
+
+type AliOSSAdapter struct {
+	StorageConfig config.StorageConfig
+	ossClient     *oss.Client
+}
+
+func (a *AliOSSAdapter) getClient() (*oss.Client, error) {
+	if a.ossClient == nil {
+		client, err := oss.New(
+			a.StorageConfig.AliOSS.Endpoint,
+			a.StorageConfig.AliOSS.AccessKeyId,
+			a.StorageConfig.AliOSS.AccessKeySecret,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("创建阿里云OSS客户端失败: %w", err)
+		}
+		a.ossClient = client
+	}
+	return a.ossClient, nil
+}
+
+func (a *AliOSSAdapter) GetFileList(ctx context.Context, dirPath string, relDt time.Time) ([]dto.File, error) {
+	_, span := tracer.Start(ctx, "GetFileList_alioss")
+	defer span.End()
+
+	span.SetAttributes(attribute.String("path", dirPath))
+	span.SetAttributes(attribute.String("relativeDate", relDt.Format("2006-01-02")))
+	if a.StorageConfig.AliOSS.Bucket == "" {
+		span.SetAttributes(attribute.String("error", "未配置阿里云OSS存储桶"))
+		span.SetStatus(codes.Error, "未配置阿里云OSS存储桶")
+		return nil, fmt.Errorf("未配置阿里云OSS存储桶")
+	}
+
+	cacheKey := fmt.Sprintf("%s_%s", dirPath, relDt.Format("2006-01-02"))
+	if cachedInterface, ok := aliOssCache.Load(cacheKey); ok {
+		cachedItem := cachedInterface.(cacheItem)
+		logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now())))
+		if time.Now().Before(cachedItem.expires) {
+			logger.Debug("获取已缓存列表", zap.String("cacheKey", cacheKey))
+			span.SetAttributes(attribute.Bool("cache.hit", true))
+			return cachedItem.data, nil
+		}
+	}
+
+	mutexKey := fmt.Sprintf("lock_%s", cacheKey)
+	mutex, _ := aliOssCache.LoadOrStore(mutexKey, &sync.Mutex{})
+	lock := mutex.(*sync.Mutex)
+	defer func() {
+		// 解锁后删除锁(避免内存泄漏)
+		aliOssCache.Delete(mutexKey)
+		lock.Unlock()
+	}()
+	lock.Lock()
+
+	if cachedInterface, ok := aliOssCache.Load(cacheKey); ok {
+		cachedItem := cachedInterface.(cacheItem)
+		logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now())))
+		if time.Now().Before(cachedItem.expires) {
+			logger.Debug("过锁后获取已缓存列表", zap.String("cacheKey", cacheKey))
+			span.SetAttributes(attribute.Bool("aliOssCache.hit", true))
+			return cachedItem.data, nil
+		}
+	}
+
+	client, err := a.getClient()
+	if err != nil {
+		span.SetAttributes(attribute.String("error", err.Error()))
+		span.SetStatus(codes.Error, "创建阿里云OSS客户端失败")
+		return nil, err
+	}
+
+	bucket, err := client.Bucket(a.StorageConfig.AliOSS.Bucket)
+	if err != nil {
+		span.SetAttributes(attribute.String("error", err.Error()))
+		span.SetStatus(codes.Error, "获取存储桶失败")
+		return nil, fmt.Errorf("获取存储桶失败: %w", err)
+	}
+
+	var fileList []dto.File
+	prefix := path.Join(a.StorageConfig.AliOSS.Prefix, dirPath)
+	marker := ""
+
+	for {
+		lsRes, err := bucket.ListObjects(
+			oss.Prefix(prefix),
+			oss.Marker(marker),
+			oss.MaxKeys(1000),
+		)
+		if err != nil {
+			span.SetAttributes(attribute.String("error", err.Error()))
+			span.SetStatus(codes.Error, "文件列表读取失败")
+			return nil, fmt.Errorf("文件列表读取失败: %w", err)
+		}
+
+		for _, object := range lsRes.Objects {
+			key := object.Key
+			if !util.IsVideoFile(path.Base(key)) {
+				continue
+			}
+			startTime, stopTime, err := util.ParseStartStopTime(path.Base(key), relDt)
+			if err != nil {
+				continue
+			}
+			if stopTime.IsZero() {
+				stopTime = startTime
+			}
+			if startTime.Equal(stopTime) {
+				stopTime = stopTime.Add(time.Second * time.Duration(config.Config.Record.Duration))
+			}
+
+			// 生成预签名URL(有效期10分钟)
+			signedURL, err := bucket.SignURL(key, oss.HTTPGet, 600)
+			if err != nil {
+				span.SetAttributes(attribute.String("error", err.Error()))
+				span.SetStatus(codes.Error, "生成预签名URL失败")
+				logger.Error("生成预签名URL失败", zap.Error(err))
+				continue
+			}
+
+			fileList = append(fileList, dto.File{
+				BasePath:  a.StorageConfig.AliOSS.Bucket,
+				Name:      path.Base(key),
+				Path:      path.Dir(key),
+				Url:       signedURL,
+				StartTime: startTime,
+				EndTime:   stopTime,
+			})
+		}
+
+		if !lsRes.IsTruncated {
+			break
+		}
+		marker = lsRes.NextMarker
+	}
+
+	span.SetAttributes(attribute.Int("file.count", len(fileList)))
+	sort.Slice(fileList, func(i, j int) bool {
+		return fileList[i].StartTime.Before(fileList[j].StartTime)
+	})
+	span.SetStatus(codes.Ok, "文件读取成功")
+
+	cacheItem := cacheItem{
+		data:    fileList,
+		expires: time.Now().Add(30 * time.Second),
+	}
+	aliOssCache.Store(cacheKey, cacheItem)
+	logger.Debug("缓存文件列表", zap.String("cacheKey", cacheKey))
+
+	return fileList, nil
+}
+
+// 添加定时清理缓存的初始化函数
+func init() {
+	go func() {
+		ticker := time.NewTicker(1 * time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				cleanupAliOssCache()
+			}
+		}
+	}()
+}
+
+// 添加缓存清理函数
+func cleanupAliOssCache() {
+	var keysToDelete []interface{}
+	aliOssCache.Range(func(key, value interface{}) bool {
+		item, ok := value.(cacheItem)
+		if !ok {
+			return true
+		}
+		if time.Now().After(item.expires) {
+			keysToDelete = append(keysToDelete, key)
+		}
+		return true
+	})
+	for _, key := range keysToDelete {
+		aliOssCache.Delete(key)
+	}
+}