nathan/vctp2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve aggregations
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Nathan Coad
2026-01-15 09:57:05 +11:00
parent 457d9395f0
commit 50e9921955
8 changed files with 261 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
internal/report/snapshots.go
View File

@@ -8,6 +8,7 @@ import (
"log/slog"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
@@ -341,6 +342,7 @@ func ListSnapshotsByRange(ctx context.Context, database db.Database, snapshotTyp
startUnix := start.Unix()
endUnix := end.Unix()
loc := start.Location()
var rows *sqlx.Rows
var err error
@@ -386,7 +388,7 @@ ORDER BY snapshot_time ASC, table_name ASC
}
records = append(records, SnapshotRecord{
TableName: tableName,
SnapshotTime: time.Unix(snapshotTime, 0),
SnapshotTime: time.Unix(snapshotTime, 0).In(loc),
SnapshotType: recordType,
SnapshotCount: snapshotCnt,
})
@@ -394,6 +396,65 @@ ORDER BY snapshot_time ASC, table_name ASC
return records, rows.Err()
}
func SnapshotRecordsWithFallback(ctx context.Context, database db.Database, snapshotType, prefix, layout string, start, end time.Time) ([]SnapshotRecord, error) {
records, err := ListSnapshotsByRange(ctx, database, snapshotType, start, end)
if err == nil && len(records) > 0 {
return records, nil
}
fallback, err2 := recordsFromTableNames(ctx, database, snapshotType, prefix, layout, start, end)
if err2 != nil {
if err != nil {
return nil, err
}
return nil, err2
}
if len(fallback) > 0 {
return fallback, nil
}
return records, err
}
func recordsFromTableNames(ctx context.Context, database db.Database, snapshotType, prefix, layout string, start, end time.Time) ([]SnapshotRecord, error) {
tables, err := ListTablesByPrefix(ctx, database, prefix)
if err != nil {
return nil, err
}
records := make([]SnapshotRecord, 0, len(tables))
for _, table := range tables {
if !strings.HasPrefix(table, prefix) {
continue
}
suffix := strings.TrimPrefix(table, prefix)
var ts time.Time
switch layout {
case "epoch":
val, err := strconv.ParseInt(suffix, 10, 64)
if err != nil {
continue
}
ts = time.Unix(val, 0)
default:
parsed, err := time.ParseInLocation(layout, suffix, time.Local)
if err != nil {
continue
}
ts = parsed
}
if !ts.Before(start) && ts.Before(end) {
records = append(records, SnapshotRecord{
TableName: table,
SnapshotTime: ts,
SnapshotType: snapshotType,
})
}
}
sort.Slice(records, func(i, j int) bool {
return records[i].SnapshotTime.Before(records[j].SnapshotTime)
})
return records, nil
}
func LatestSnapshotTime(ctx context.Context, database db.Database, snapshotType string) (time.Time, error) {
dbConn := database.DB()
driver := strings.ToLower(dbConn.DriverName())
@@ -665,7 +726,7 @@ func SaveTableReport(logger *slog.Logger, Database db.Database, ctx context.Cont
func addTotalsChartSheet(logger *slog.Logger, database db.Database, ctx context.Context, xlsx *excelize.File, tableName string) {
if strings.HasPrefix(tableName, "inventory_daily_summary_") {
suffix := strings.TrimPrefix(tableName, "inventory_daily_summary_")
dayStart, err := time.Parse("20060102", suffix)
dayStart, err := time.ParseInLocation("20060102", suffix, time.Local)
if err != nil {
return
}
@@ -673,7 +734,7 @@ func addTotalsChartSheet(logger *slog.Logger, database db.Database, ctx context.
if err := EnsureSnapshotRegistry(ctx, database); err != nil {
return
}
records, err := ListSnapshotsByRange(ctx, database, "hourly", dayStart, dayEnd)
records, err := SnapshotRecordsWithFallback(ctx, database, "hourly", "inventory_hourly_", "epoch", dayStart, dayEnd)
if err != nil || len(records) == 0 {
return
}
@@ -687,7 +748,7 @@ func addTotalsChartSheet(logger *slog.Logger, database db.Database, ctx context.
if strings.HasPrefix(tableName, "inventory_monthly_summary_") {
suffix := strings.TrimPrefix(tableName, "inventory_monthly_summary_")
monthStart, err := time.Parse("200601", suffix)
monthStart, err := time.ParseInLocation("200601", suffix, time.Local)
if err != nil {
return
}
@@ -695,7 +756,7 @@ func addTotalsChartSheet(logger *slog.Logger, database db.Database, ctx context.
if err := EnsureSnapshotRegistry(ctx, database); err != nil {
return
}
records, err := ListSnapshotsByRange(ctx, database, "daily", monthStart, monthEnd)
records, err := SnapshotRecordsWithFallback(ctx, database, "daily", "inventory_daily_summary_", "20060102", monthStart, monthEnd)
if err != nil || len(records) == 0 {
return
}

6
internal/tasks/dailyAggregate.go
View File

@@ -52,10 +52,11 @@ func (c *CronTask) aggregateDailySummary(ctx context.Context, targetTime time.Ti
}
}
hourlySnapshots, err := report.ListSnapshotsByRange(ctx, c.Database, "hourly", dayStart, dayEnd)
hourlySnapshots, err := report.SnapshotRecordsWithFallback(ctx, c.Database, "hourly", "inventory_hourly_", "epoch", dayStart, dayEnd)
if err != nil {
return err
}
hourlySnapshots = filterRecordsInRange(hourlySnapshots, dayStart, dayEnd)
hourlySnapshots = filterSnapshotsWithRows(ctx, dbConn, hourlySnapshots)
if len(hourlySnapshots) == 0 {
return fmt.Errorf("no hourly snapshot tables found for %s", dayStart.Format("2006-01-02"))
@@ -85,8 +86,9 @@ func (c *CronTask) aggregateDailySummary(ctx context.Context, targetTime time.Ti
prevStart := dayStart.AddDate(0, 0, -1)
prevEnd := dayStart
prevSnapshots, err := report.ListSnapshotsByRange(ctx, c.Database, "hourly", prevStart, prevEnd)
prevSnapshots, err := report.SnapshotRecordsWithFallback(ctx, c.Database, "hourly", "inventory_hourly_", "epoch", prevStart, prevEnd)
if err == nil && len(prevSnapshots) > 0 {
prevSnapshots = filterRecordsInRange(prevSnapshots, prevStart, prevEnd)
prevSnapshots = filterSnapshotsWithRows(ctx, dbConn, prevSnapshots)
prevTables := make([]string, 0, len(prevSnapshots))
for _, snapshot := range prevSnapshots {

31
internal/tasks/inventorySnapshots.go
View File

@@ -370,6 +370,16 @@ func filterSnapshotsWithRows(ctx context.Context, dbConn *sqlx.DB, snapshots []r
return filtered
}
func filterRecordsInRange(records []report.SnapshotRecord, start, end time.Time) []report.SnapshotRecord {
filtered := records[:0]
for _, r := range records {
if !r.SnapshotTime.Before(start) && r.SnapshotTime.Before(end) {
filtered = append(filtered, r)
}
}
return filtered
}
type columnDef struct {
Name string
Type string
@@ -437,18 +447,17 @@ func normalizeResourcePool(value string) string {
if trimmed == "" {
return ""
}
switch {
case strings.EqualFold(trimmed, "tin"):
return "Tin"
case strings.EqualFold(trimmed, "bronze"):
return "Bronze"
case strings.EqualFold(trimmed, "silver"):
return "Silver"
case strings.EqualFold(trimmed, "gold"):
return "Gold"
default:
return trimmed
lower := strings.ToLower(trimmed)
canonical := map[string]string{
"tin": "Tin",
"bronze": "Bronze",
"silver": "Silver",
"gold": "Gold",
}
if val, ok := canonical[lower]; ok {
return val
}
return trimmed
}
func (c *CronTask) reportsDir() string {

3
internal/tasks/monthlyAggregate.go
View File

@@ -35,10 +35,11 @@ func (c *CronTask) aggregateMonthlySummary(ctx context.Context, targetMonth time
monthStart := time.Date(targetMonth.Year(), targetMonth.Month(), 1, 0, 0, 0, 0, targetMonth.Location())
monthEnd := monthStart.AddDate(0, 1, 0)
dailySnapshots, err := report.ListSnapshotsByRange(ctx, c.Database, "hourly", monthStart, monthEnd)
dailySnapshots, err := report.SnapshotRecordsWithFallback(ctx, c.Database, "daily", "inventory_daily_summary_", "20060102", monthStart, monthEnd)
if err != nil {
return err
}
dailySnapshots = filterRecordsInRange(dailySnapshots, monthStart, monthEnd)
dbConn := c.Database.DB()
dailySnapshots = filterSnapshotsWithRows(ctx, dbConn, dailySnapshots)