mirror of
https://github.com/DullJZ/s3-balance.git
synced 2026-06-26 21:41:21 +08:00
Merge branch 'dev' into add-api
This commit is contained in:
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/DullJZ/s3-balance/internal/database"
|
||||
"github.com/DullJZ/s3-balance/internal/metrics"
|
||||
"github.com/DullJZ/s3-balance/internal/middleware"
|
||||
"github.com/DullJZ/s3-balance/internal/scheduler"
|
||||
"github.com/DullJZ/s3-balance/internal/storage"
|
||||
"github.com/DullJZ/s3-balance/pkg/presigner"
|
||||
"github.com/gorilla/mux"
|
||||
@@ -81,6 +82,11 @@ func main() {
|
||||
// 启动定期清理过期上传会话的任务
|
||||
startSessionCleaner(ctx, storageService)
|
||||
|
||||
// 启动月度统计归档任务(每小时检查一次)
|
||||
monthlyArchiver := scheduler.NewMonthlyArchiver(storageService, 1*time.Hour)
|
||||
monthlyArchiver.Start()
|
||||
defer monthlyArchiver.Stop()
|
||||
|
||||
// 创建S3兼容API处理器
|
||||
s3Handler := api.NewS3Handler(
|
||||
bucketManager,
|
||||
|
||||
@@ -141,11 +141,12 @@ s3api:
|
||||
virtual_host: false
|
||||
|
||||
# 工作模式:
|
||||
# false:预签名重定向模式,客户端直接与后端存储交互
|
||||
# true (默认):代理模式,数据通过S3 Balance服务器传输
|
||||
# false:预签名重定向模式,客户端下载直接重定向到与后端存储
|
||||
# true (默认):代理模式,数据通过S3 Balance服务器中转传输
|
||||
# 该选项仅适用于下载,上传操作始终为全代理模式
|
||||
proxy_mode: true
|
||||
|
||||
# 是否需要认证(开启后使用 Basic Auth,凭据来自 access_key/secret_key)
|
||||
|
||||
# 是否需要认证(使用配置的 access_key/secret_key)
|
||||
auth_required: true
|
||||
|
||||
# 用于签名验证的Host(可选)
|
||||
|
||||
2
go.mod
2
go.mod
@@ -3,7 +3,7 @@ module github.com/DullJZ/s3-balance
|
||||
go 1.24.5
|
||||
|
||||
require (
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251004111253-b3ec227d3796
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251103105435-c25eac6b580b
|
||||
github.com/aws/aws-sdk-go-v2 v1.39.2
|
||||
github.com/aws/aws-sdk-go-v2/config v1.31.1
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.18.5
|
||||
|
||||
4
go.sum
4
go.sum
@@ -1,7 +1,7 @@
|
||||
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
|
||||
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251004111253-b3ec227d3796 h1:0Lipgc3EHF2QOKpCziXApbVocdyzZ/3a52xluuWraXg=
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251004111253-b3ec227d3796/go.mod h1:OEx+/bRlDdI0oj/Bb1Plsq+1+qU1qal3/g9phixhU6Y=
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251103105435-c25eac6b580b h1:BHue7N77inSdaDUUZSO/gMmc3+4ZGdQA3ORdcLHnxtg=
|
||||
github.com/DullJZ/s3-validate v0.0.0-20251103105435-c25eac6b580b/go.mod h1:OEx+/bRlDdI0oj/Bb1Plsq+1+qU1qal3/g9phixhU6Y=
|
||||
github.com/aws/aws-sdk-go-v2 v1.39.2 h1:EJLg8IdbzgeD7xgvZ+I8M1e0fL0ptn/M47lianzth0I=
|
||||
github.com/aws/aws-sdk-go-v2 v1.39.2/go.mod h1:sDioUELIUO9Znk23YVmIk86/9DOpkbyyVb1i/gUNFXY=
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0 h1:6GMWV6CNpA/6fbFHnoAjrv4+LGfyTqZz2LtCHnspgDg=
|
||||
|
||||
@@ -76,7 +76,7 @@ func (h *S3Handler) RegisterS3Routes(router *mux.Router) {
|
||||
|
||||
// 带认证/虚拟主机的路由
|
||||
protected := router.NewRoute().PathPrefix("/{bucket}").Subrouter()
|
||||
protected.StrictSlash(true)
|
||||
// 注意:不使用 StrictSlash(true) 以避免 301 重定向,兼容WinSCP
|
||||
|
||||
// Bucket operations
|
||||
protected.HandleFunc("", h.handleBucketOperations).Methods("GET", "HEAD", "PUT", "DELETE")
|
||||
|
||||
193
internal/api/stats_handler.go
Normal file
193
internal/api/stats_handler.go
Normal file
@@ -0,0 +1,193 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/DullJZ/s3-balance/internal/storage"
|
||||
"github.com/gorilla/mux"
|
||||
)
|
||||
|
||||
// StatsHandler 统计数据处理器
|
||||
type StatsHandler struct {
|
||||
storage *storage.Service
|
||||
}
|
||||
|
||||
// NewStatsHandler 创建统计处理器
|
||||
func NewStatsHandler(storage *storage.Service) *StatsHandler {
|
||||
return &StatsHandler{
|
||||
storage: storage,
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterRoutes 注册统计API路由
|
||||
func (h *StatsHandler) RegisterRoutes(router *mux.Router) {
|
||||
router.HandleFunc("/api/stats/monthly", h.GetCurrentMonthStats).Methods("GET")
|
||||
router.HandleFunc("/api/stats/monthly/{year}/{month}", h.GetMonthlyStats).Methods("GET")
|
||||
router.HandleFunc("/api/stats/monthly/range", h.GetMonthlyStatsRange).Methods("GET")
|
||||
router.HandleFunc("/api/stats/bucket/{bucket}/history", h.GetBucketHistory).Methods("GET")
|
||||
}
|
||||
|
||||
// MonthlyStatsResponse 月度统计响应
|
||||
type MonthlyStatsResponse struct {
|
||||
Year int `json:"year"`
|
||||
Month int `json:"month"`
|
||||
Bucket string `json:"bucket"`
|
||||
Stats BucketOperationCounts `json:"stats"`
|
||||
}
|
||||
|
||||
// BucketOperationCounts 存储桶操作计数
|
||||
type BucketOperationCounts struct {
|
||||
OperationCountA int64 `json:"operation_count_a"`
|
||||
OperationCountB int64 `json:"operation_count_b"`
|
||||
Total int64 `json:"total"`
|
||||
}
|
||||
|
||||
// GetCurrentMonthStats 获取当前月份的统计
|
||||
func (h *StatsHandler) GetCurrentMonthStats(w http.ResponseWriter, r *http.Request) {
|
||||
stats, err := h.storage.GetCurrentMonthStats()
|
||||
if err != nil {
|
||||
log.Printf("Failed to get current month stats: %v", err)
|
||||
http.Error(w, "Failed to fetch statistics", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
response := h.formatMonthlyStats(stats)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}
|
||||
|
||||
// GetMonthlyStats 获取指定月份的统计
|
||||
func (h *StatsHandler) GetMonthlyStats(w http.ResponseWriter, r *http.Request) {
|
||||
vars := mux.Vars(r)
|
||||
year, err := strconv.Atoi(vars["year"])
|
||||
if err != nil {
|
||||
http.Error(w, "Invalid year", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
month, err := strconv.Atoi(vars["month"])
|
||||
if err != nil || month < 1 || month > 12 {
|
||||
http.Error(w, "Invalid month", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
stats, err := h.storage.GetMonthlyStats(year, month)
|
||||
if err != nil {
|
||||
log.Printf("Failed to get monthly stats: %v", err)
|
||||
http.Error(w, "Failed to fetch statistics", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
response := h.formatMonthlyStats(stats)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}
|
||||
|
||||
// GetMonthlyStatsRange 获取时间范围内的统计
|
||||
func (h *StatsHandler) GetMonthlyStatsRange(w http.ResponseWriter, r *http.Request) {
|
||||
query := r.URL.Query()
|
||||
|
||||
startYear, err := strconv.Atoi(query.Get("start_year"))
|
||||
if err != nil {
|
||||
http.Error(w, "Invalid start_year", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
startMonth, err := strconv.Atoi(query.Get("start_month"))
|
||||
if err != nil || startMonth < 1 || startMonth > 12 {
|
||||
http.Error(w, "Invalid start_month", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
endYear, err := strconv.Atoi(query.Get("end_year"))
|
||||
if err != nil {
|
||||
http.Error(w, "Invalid end_year", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
endMonth, err := strconv.Atoi(query.Get("end_month"))
|
||||
if err != nil || endMonth < 1 || endMonth > 12 {
|
||||
http.Error(w, "Invalid end_month", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
stats, err := h.storage.GetMonthlyStatsRange(startYear, startMonth, endYear, endMonth)
|
||||
if err != nil {
|
||||
log.Printf("Failed to get monthly stats range: %v", err)
|
||||
http.Error(w, "Failed to fetch statistics", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
response := h.formatMonthlyStats(stats)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}
|
||||
|
||||
// GetBucketHistory 获取指定存储桶的历史统计
|
||||
func (h *StatsHandler) GetBucketHistory(w http.ResponseWriter, r *http.Request) {
|
||||
vars := mux.Vars(r)
|
||||
bucket := vars["bucket"]
|
||||
|
||||
// 获取查询参数中的月份数,默认12个月
|
||||
months := 12
|
||||
if monthsStr := r.URL.Query().Get("months"); monthsStr != "" {
|
||||
if m, err := strconv.Atoi(monthsStr); err == nil && m > 0 {
|
||||
months = m
|
||||
}
|
||||
}
|
||||
|
||||
stats, err := h.storage.GetBucketMonthlyHistory(bucket, months)
|
||||
if err != nil {
|
||||
log.Printf("Failed to get bucket history: %v", err)
|
||||
http.Error(w, "Failed to fetch statistics", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
response := h.formatMonthlyStats(stats)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(response)
|
||||
}
|
||||
|
||||
// formatMonthlyStats 格式化月度统计数据
|
||||
func (h *StatsHandler) formatMonthlyStats(stats []storage.BucketMonthlyStats) []MonthlyStatsResponse {
|
||||
result := make([]MonthlyStatsResponse, 0, len(stats))
|
||||
|
||||
for _, stat := range stats {
|
||||
result = append(result, MonthlyStatsResponse{
|
||||
Year: stat.Year,
|
||||
Month: stat.Month,
|
||||
Bucket: stat.BucketName,
|
||||
Stats: BucketOperationCounts{
|
||||
OperationCountA: stat.OperationCountA,
|
||||
OperationCountB: stat.OperationCountB,
|
||||
Total: stat.OperationCountA + stat.OperationCountB,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// ArchiveCurrentMonth 手动触发归档当前月份(管理API)
|
||||
func (h *StatsHandler) ArchiveCurrentMonth(w http.ResponseWriter, r *http.Request) {
|
||||
now := time.Now()
|
||||
year, month := now.Year(), int(now.Month())
|
||||
|
||||
if err := h.storage.ArchiveMonthlyStats(year, month); err != nil {
|
||||
log.Printf("Failed to archive monthly stats: %v", err)
|
||||
http.Error(w, "Failed to archive statistics", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]string{
|
||||
"status": "success",
|
||||
"message": "Monthly statistics archived successfully",
|
||||
"year": strconv.Itoa(year),
|
||||
"month": strconv.Itoa(month),
|
||||
})
|
||||
}
|
||||
@@ -181,6 +181,7 @@ func AutoMigrate() error {
|
||||
models := []interface{}{
|
||||
&storage.Object{},
|
||||
&storage.BucketStats{},
|
||||
&storage.BucketMonthlyStats{},
|
||||
&storage.UploadSession{},
|
||||
&storage.AccessLog{},
|
||||
&storage.VirtualBucketMapping{},
|
||||
|
||||
96
internal/scheduler/monthly_archiver.go
Normal file
96
internal/scheduler/monthly_archiver.go
Normal file
@@ -0,0 +1,96 @@
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/DullJZ/s3-balance/internal/storage"
|
||||
)
|
||||
|
||||
// MonthlyArchiver 月度统计归档器
|
||||
type MonthlyArchiver struct {
|
||||
storage *storage.Service
|
||||
ticker *time.Ticker
|
||||
stopChan chan struct{}
|
||||
lastArchivedDate string // 格式: "2025-01" - 记录上次归档的月份
|
||||
}
|
||||
|
||||
// NewMonthlyArchiver 创建月度归档器
|
||||
func NewMonthlyArchiver(storage *storage.Service, checkInterval time.Duration) *MonthlyArchiver {
|
||||
return &MonthlyArchiver{
|
||||
storage: storage,
|
||||
ticker: time.NewTicker(checkInterval),
|
||||
stopChan: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Start 启动月度归档定期任务
|
||||
func (m *MonthlyArchiver) Start() {
|
||||
log.Println("Starting monthly statistics archiver...")
|
||||
|
||||
// 启动时立即归档上个月的数据(如果还没有归档)
|
||||
m.archiveLastMonth()
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-m.ticker.C:
|
||||
m.checkAndArchive()
|
||||
case <-m.stopChan:
|
||||
log.Println("Monthly statistics archiver stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop 停止归档任务
|
||||
func (m *MonthlyArchiver) Stop() {
|
||||
close(m.stopChan)
|
||||
m.ticker.Stop()
|
||||
}
|
||||
|
||||
// checkAndArchive 检查并归档统计数据
|
||||
func (m *MonthlyArchiver) checkAndArchive() {
|
||||
now := time.Now()
|
||||
lastMonth := now.AddDate(0, -1, 0)
|
||||
lastMonthKey := lastMonth.Format("2006-01")
|
||||
|
||||
// 如果是每月的第一天,且上个月还未归档,则归档上个月的数据
|
||||
if now.Day() == 1 && m.lastArchivedDate != lastMonthKey {
|
||||
m.archiveLastMonth()
|
||||
m.lastArchivedDate = lastMonthKey
|
||||
}
|
||||
|
||||
// 每天都归档当前月份(实时更新)
|
||||
m.archiveCurrentMonth()
|
||||
}
|
||||
|
||||
// archiveLastMonth 归档上个月的数据
|
||||
func (m *MonthlyArchiver) archiveLastMonth() {
|
||||
now := time.Now()
|
||||
lastMonth := now.AddDate(0, -1, 0)
|
||||
year, month := lastMonth.Year(), int(lastMonth.Month())
|
||||
|
||||
log.Printf("Archiving monthly stats for %d-%02d...", year, month)
|
||||
|
||||
if err := m.storage.ArchiveMonthlyStats(year, month); err != nil {
|
||||
log.Printf("Failed to archive monthly stats for %d-%02d: %v", year, month, err)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("Successfully archived monthly stats for %d-%02d", year, month)
|
||||
}
|
||||
|
||||
// archiveCurrentMonth 归档当前月份(实时更新)
|
||||
func (m *MonthlyArchiver) archiveCurrentMonth() {
|
||||
now := time.Now()
|
||||
year, month := now.Year(), int(now.Month())
|
||||
|
||||
if err := m.storage.ArchiveMonthlyStats(year, month); err != nil {
|
||||
log.Printf("Failed to update current month stats for %d-%02d: %v", year, month, err)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("Updated current month stats for %d-%02d", year, month)
|
||||
}
|
||||
@@ -45,6 +45,23 @@ func (BucketStats) TableName() string {
|
||||
return "bucket_stats"
|
||||
}
|
||||
|
||||
// BucketMonthlyStats 存储桶月度统计信息模型
|
||||
type BucketMonthlyStats struct {
|
||||
ID uint `gorm:"primaryKey" json:"id"`
|
||||
BucketName string `gorm:"uniqueIndex:idx_bucket_month;size:255;not null" json:"bucket_name"`
|
||||
Year int `gorm:"uniqueIndex:idx_bucket_month;not null" json:"year"`
|
||||
Month int `gorm:"uniqueIndex:idx_bucket_month;not null" json:"month"`
|
||||
OperationCountA int64 `gorm:"not null;default:0" json:"operation_count_a"`
|
||||
OperationCountB int64 `gorm:"not null;default:0" json:"operation_count_b"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// TableName 指定表名
|
||||
func (BucketMonthlyStats) TableName() string {
|
||||
return "bucket_monthly_stats"
|
||||
}
|
||||
|
||||
// VirtualBucketMapping 虚拟存储桶文件级映射模型
|
||||
type VirtualBucketMapping struct {
|
||||
ID uint `gorm:"primaryKey" json:"id"`
|
||||
|
||||
@@ -304,18 +304,29 @@ func (s *Service) IncrementBucketOperation(bucketName, category string) (int64,
|
||||
return 0, fmt.Errorf("unknown operation category: %s", category)
|
||||
}
|
||||
|
||||
if err := s.db.Model(&BucketStats{}).
|
||||
Where("bucket_name = ?", bucketName).
|
||||
UpdateColumn(field, gorm.Expr(field+" + ?", 1)).Error; err != nil {
|
||||
return 0, fmt.Errorf("failed to increment %s for bucket %s: %w", field, bucketName, err)
|
||||
}
|
||||
|
||||
// 使用事务确保原子性
|
||||
var count int64
|
||||
if err := s.db.Model(&BucketStats{}).
|
||||
Where("bucket_name = ?", bucketName).
|
||||
Select(field).
|
||||
Scan(&count).Error; err != nil {
|
||||
return 0, fmt.Errorf("failed to fetch updated %s for bucket %s: %w", field, bucketName, err)
|
||||
err := s.db.Transaction(func(tx *gorm.DB) error {
|
||||
// 原子递增
|
||||
if err := tx.Model(&BucketStats{}).
|
||||
Where("bucket_name = ?", bucketName).
|
||||
UpdateColumn(field, gorm.Expr(field+" + ?", 1)).Error; err != nil {
|
||||
return fmt.Errorf("failed to increment %s for bucket %s: %w", field, bucketName, err)
|
||||
}
|
||||
|
||||
// 在同一事务中读取最新值
|
||||
if err := tx.Model(&BucketStats{}).
|
||||
Where("bucket_name = ?", bucketName).
|
||||
Select(field).
|
||||
Scan(&count).Error; err != nil {
|
||||
return fmt.Errorf("failed to fetch updated %s for bucket %s: %w", field, bucketName, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return count, nil
|
||||
@@ -630,3 +641,175 @@ func (s *Service) DeleteVirtualBucketFileMapping(virtualBucketName, objectKey st
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ArchiveMonthlyStats 归档指定月份的统计数据(存储增量值,非累计值)
|
||||
// 如果该月份的记录已存在,则更新;否则创建新记录
|
||||
func (s *Service) ArchiveMonthlyStats(year, month int) error {
|
||||
// 获取当前所有bucket的累计统计
|
||||
var currentStats []BucketStats
|
||||
if err := s.db.Find(¤tStats).Error; err != nil {
|
||||
return fmt.Errorf("failed to fetch bucket stats: %w", err)
|
||||
}
|
||||
|
||||
// 获取上个月的累计值(从上月归档数据推算)
|
||||
lastYear, lastMonth := year, month-1
|
||||
if lastMonth == 0 {
|
||||
lastMonth = 12
|
||||
lastYear--
|
||||
}
|
||||
|
||||
// 查询上个月及之前的所有归档数据,用于推算上月末的累计值
|
||||
var lastMonthArchived []BucketMonthlyStats
|
||||
lastMonthMap := make(map[string]int64) // bucket_name -> last_month_cumulative_a
|
||||
lastMonthMapB := make(map[string]int64) // bucket_name -> last_month_cumulative_b
|
||||
|
||||
if err := s.db.Where("year < ? OR (year = ? AND month <= ?)", lastYear, lastYear, lastMonth).
|
||||
Order("year ASC, month ASC").
|
||||
Find(&lastMonthArchived).Error; err == nil {
|
||||
|
||||
// 累加历史增量得到上月末累计值
|
||||
cumulativeA := make(map[string]int64)
|
||||
cumulativeB := make(map[string]int64)
|
||||
|
||||
for _, archived := range lastMonthArchived {
|
||||
cumulativeA[archived.BucketName] += archived.OperationCountA
|
||||
cumulativeB[archived.BucketName] += archived.OperationCountB
|
||||
}
|
||||
|
||||
lastMonthMap = cumulativeA
|
||||
lastMonthMapB = cumulativeB
|
||||
}
|
||||
|
||||
// 对每个bucket,计算本月增量并存储
|
||||
for _, stat := range currentStats {
|
||||
lastCumulativeA := lastMonthMap[stat.BucketName]
|
||||
lastCumulativeB := lastMonthMapB[stat.BucketName]
|
||||
|
||||
incrementA := stat.OperationCountA - lastCumulativeA
|
||||
incrementB := stat.OperationCountB - lastCumulativeB
|
||||
|
||||
// 如果是首次运行(没有历史数据),incrementA/B 可能等于累计值
|
||||
// 这是预期行为:首月记录的就是从0到当前的增量
|
||||
|
||||
// 边界情况:如果计算出负值,说明数据不一致,设置为0
|
||||
if incrementA < 0 {
|
||||
incrementA = 0
|
||||
}
|
||||
if incrementB < 0 {
|
||||
incrementB = 0
|
||||
}
|
||||
|
||||
monthlyStats := BucketMonthlyStats{
|
||||
BucketName: stat.BucketName,
|
||||
Year: year,
|
||||
Month: month,
|
||||
OperationCountA: incrementA,
|
||||
OperationCountB: incrementB,
|
||||
}
|
||||
|
||||
// 使用 UPSERT 逻辑:如果存在则更新,否则创建
|
||||
if err := s.db.Where("bucket_name = ? AND year = ? AND month = ?",
|
||||
stat.BucketName, year, month).
|
||||
Assign(BucketMonthlyStats{
|
||||
OperationCountA: incrementA,
|
||||
OperationCountB: incrementB,
|
||||
}).
|
||||
FirstOrCreate(&monthlyStats).Error; err != nil {
|
||||
return fmt.Errorf("failed to archive monthly stats for bucket %s: %w", stat.BucketName, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetMonthlyStats 获取指定月份的统计数据
|
||||
func (s *Service) GetMonthlyStats(year, month int) ([]BucketMonthlyStats, error) {
|
||||
var stats []BucketMonthlyStats
|
||||
if err := s.db.Where("year = ? AND month = ?", year, month).
|
||||
Find(&stats).Error; err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch monthly stats: %w", err)
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// GetMonthlyStatsRange 获取指定时间范围的统计数据
|
||||
func (s *Service) GetMonthlyStatsRange(startYear, startMonth, endYear, endMonth int) ([]BucketMonthlyStats, error) {
|
||||
var stats []BucketMonthlyStats
|
||||
if err := s.db.Where("(year > ? OR (year = ? AND month >= ?)) AND (year < ? OR (year = ? AND month <= ?))",
|
||||
startYear, startYear, startMonth, endYear, endYear, endMonth).
|
||||
Order("year, month, bucket_name").
|
||||
Find(&stats).Error; err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch monthly stats range: %w", err)
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// GetCurrentMonthStats 获取当前月份的实时统计(从 bucket_stats 计算增量)
|
||||
func (s *Service) GetCurrentMonthStats() ([]BucketMonthlyStats, error) {
|
||||
now := time.Now()
|
||||
year, month := now.Year(), int(now.Month())
|
||||
|
||||
// 获取上个月末的累计值(通过累加所有历史增量)
|
||||
lastYear, lastMonth := year, month-1
|
||||
if lastMonth == 0 {
|
||||
lastMonth = 12
|
||||
lastYear--
|
||||
}
|
||||
|
||||
var historicalStats []BucketMonthlyStats
|
||||
lastMonthCumulativeA := make(map[string]int64)
|
||||
lastMonthCumulativeB := make(map[string]int64)
|
||||
|
||||
if err := s.db.Where("year < ? OR (year = ? AND month <= ?)", lastYear, lastYear, lastMonth).
|
||||
Find(&historicalStats).Error; err == nil {
|
||||
// 累加所有历史增量得到上月末累计值
|
||||
for _, stat := range historicalStats {
|
||||
lastMonthCumulativeA[stat.BucketName] += stat.OperationCountA
|
||||
lastMonthCumulativeB[stat.BucketName] += stat.OperationCountB
|
||||
}
|
||||
}
|
||||
|
||||
// 获取当前累计数据
|
||||
var currentStats []BucketStats
|
||||
if err := s.db.Find(¤tStats).Error; err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch current bucket stats: %w", err)
|
||||
}
|
||||
|
||||
// 计算当前月份的增量
|
||||
result := make([]BucketMonthlyStats, 0, len(currentStats))
|
||||
for _, current := range currentStats {
|
||||
incrementA := current.OperationCountA - lastMonthCumulativeA[current.BucketName]
|
||||
incrementB := current.OperationCountB - lastMonthCumulativeB[current.BucketName]
|
||||
|
||||
// 边界情况:如果计算出负值,说明数据不一致,设置为0
|
||||
if incrementA < 0 {
|
||||
incrementA = 0
|
||||
}
|
||||
if incrementB < 0 {
|
||||
incrementB = 0
|
||||
}
|
||||
|
||||
result = append(result, BucketMonthlyStats{
|
||||
BucketName: current.BucketName,
|
||||
Year: year,
|
||||
Month: month,
|
||||
OperationCountA: incrementA,
|
||||
OperationCountB: incrementB,
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// GetBucketMonthlyHistory 获取指定存储桶的月度历史统计
|
||||
func (s *Service) GetBucketMonthlyHistory(bucketName string, months int) ([]BucketMonthlyStats, error) {
|
||||
var stats []BucketMonthlyStats
|
||||
if err := s.db.Where("bucket_name = ?", bucketName).
|
||||
Order("year DESC, month DESC").
|
||||
Limit(months).
|
||||
Find(&stats).Error; err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch bucket monthly history: %w", err)
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user