fix:修正查询SQL
This commit is contained in:
parent
6936734f7e
commit
e4b1c19064
@ -7,6 +7,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
"weatherstation/internal/database"
|
||||
"weatherstation/internal/selftest"
|
||||
"weatherstation/internal/server"
|
||||
"weatherstation/internal/tools"
|
||||
)
|
||||
@ -21,6 +22,9 @@ func main() {
|
||||
var bfFrom = flag.String("from", "", "回填起始时间,格式:YYYY-MM-DD HH:MM:SS")
|
||||
var bfTo = flag.String("to", "", "回填结束时间,格式:YYYY-MM-DD HH:MM:SS")
|
||||
var bfWrap = flag.Float64("wrap", 0, "回绕一圈对应毫米值(mm),<=0 则降级为仅计当前值")
|
||||
// 自检控制
|
||||
var noSelftest = flag.Bool("no-selftest", false, "跳过启动自检")
|
||||
var selftestOnly = flag.Bool("selftest_only", false, "仅执行自检后退出")
|
||||
flag.Parse()
|
||||
|
||||
// 设置日志
|
||||
@ -30,6 +34,17 @@ func main() {
|
||||
_ = database.GetDB() // 确保数据库连接已初始化
|
||||
defer database.Close()
|
||||
|
||||
// 启动前自检
|
||||
if !*noSelftest {
|
||||
if err := selftest.Run(context.Background()); err != nil {
|
||||
log.Fatalf("启动自检失败: %v", err)
|
||||
}
|
||||
if *selftestOnly {
|
||||
log.Println("自检完成,按 --selftest_only 要求退出")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Backfill 调试路径
|
||||
if *doBackfill {
|
||||
if *bfFrom == "" || *bfTo == "" {
|
||||
|
||||
@ -2,6 +2,7 @@ package database
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"log"
|
||||
"time"
|
||||
"weatherstation/pkg/types"
|
||||
)
|
||||
@ -113,6 +114,9 @@ func GetWeatherData(db *sql.DB, stationID string, startTime, endTime time.Time,
|
||||
|
||||
// GetSeriesFrom10Min 基于10分钟聚合表返回 10m/30m/1h 数据(风向向量平均、降雨求和、加权平均)
|
||||
func GetSeriesFrom10Min(db *sql.DB, stationID string, startTime, endTime time.Time, interval string) ([]types.WeatherPoint, error) {
|
||||
log.Printf("查询数据: stationID=%s, start=%v, end=%v, interval=%s",
|
||||
stationID, startTime.Format("2006-01-02 15:04:05"), endTime.Format("2006-01-02 15:04:05"), interval)
|
||||
|
||||
var query string
|
||||
switch interval {
|
||||
case "10min":
|
||||
@ -136,8 +140,14 @@ func GetSeriesFrom10Min(db *sql.DB, stationID string, startTime, endTime time.Ti
|
||||
query = buildAggFrom10MinQuery("1 hour")
|
||||
}
|
||||
|
||||
// // 调试输出完整SQL
|
||||
// debugSQL := fmt.Sprintf("-- SQL for %s\n%s\n-- Params: stationID=%s, start=%v, end=%v",
|
||||
// interval, query, stationID, startTime, endTime)
|
||||
// log.Println(debugSQL)
|
||||
|
||||
rows, err := db.Query(query, stationID, startTime, endTime)
|
||||
if err != nil {
|
||||
log.Printf("查询失败: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
@ -158,10 +168,18 @@ func buildAggFrom10MinQuery(interval string) string {
|
||||
return `
|
||||
WITH base AS (
|
||||
SELECT * FROM rs485_weather_10min
|
||||
WHERE station_id = $1 AND bucket_start >= $2 AND bucket_start < $3 + '` + interval + `'::interval
|
||||
WHERE station_id = $1 AND bucket_start >= $2 AND bucket_start <= $3
|
||||
), g AS (
|
||||
SELECT
|
||||
date_trunc('` + interval + `', bucket_start) AS grp,
|
||||
CASE '` + interval + `'
|
||||
WHEN '1 hour' THEN date_trunc('hour', bucket_start)
|
||||
WHEN '30 minutes' THEN
|
||||
date_trunc('hour', bucket_start) +
|
||||
CASE WHEN date_part('minute', bucket_start) >= 30
|
||||
THEN '30 minutes'::interval
|
||||
ELSE '0 minutes'::interval
|
||||
END
|
||||
END AS grp,
|
||||
SUM(temp_c_x100 * sample_count)::bigint AS w_temp,
|
||||
SUM(humidity_pct * sample_count)::bigint AS w_hum,
|
||||
SUM(pressure_hpa_x100 * sample_count)::bigint AS w_p,
|
||||
|
||||
80
internal/selftest/selftest.go
Normal file
80
internal/selftest/selftest.go
Normal file
@ -0,0 +1,80 @@
|
||||
package selftest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"weatherstation/internal/database"
|
||||
)
|
||||
|
||||
// Run 执行启动前自检
|
||||
// 1) 数据库可用 2) 关键表存在 3) 基础查询可执行
|
||||
func Run(ctx context.Context) error {
|
||||
db := database.GetDB()
|
||||
if err := pingDB(ctx, db); err != nil {
|
||||
return fmt.Errorf("数据库连通性失败: %w", err)
|
||||
}
|
||||
if err := ensureTables(ctx, db); err != nil {
|
||||
return fmt.Errorf("关键表缺失: %w", err)
|
||||
}
|
||||
if err := basicQuery(ctx, db); err != nil {
|
||||
return fmt.Errorf("基础查询失败: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func pingDB(ctx context.Context, db *sql.DB) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
defer cancel()
|
||||
return db.PingContext(ctx)
|
||||
}
|
||||
|
||||
func ensureTables(ctx context.Context, db *sql.DB) error {
|
||||
required := []string{
|
||||
"public.stations",
|
||||
"public.rs485_weather_data",
|
||||
"public.rs485_weather_10min",
|
||||
"public.forecast_hourly",
|
||||
}
|
||||
for _, t := range required {
|
||||
var exists bool
|
||||
if err := db.QueryRowContext(ctx, "SELECT to_regclass($1) IS NOT NULL", t).Scan(&exists); err != nil {
|
||||
return err
|
||||
}
|
||||
if !exists {
|
||||
return fmt.Errorf("缺少表: %s", t)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func basicQuery(ctx context.Context, db *sql.DB) error {
|
||||
// 若无站点则跳过后续测试
|
||||
var stationID string
|
||||
_ = db.QueryRowContext(ctx, "SELECT station_id FROM stations LIMIT 1").Scan(&stationID)
|
||||
|
||||
// 10分钟表可被查询(允许为空)
|
||||
var cnt int
|
||||
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM rs485_weather_10min").Scan(&cnt); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 若有站点,做一次安全时间窗聚合验证(不要求有数据,只要语句可执行)
|
||||
if stationID != "" {
|
||||
from := time.Now().Add(-2 * time.Hour).Truncate(time.Hour)
|
||||
to := time.Now().Truncate(time.Hour)
|
||||
_, err := db.QueryContext(ctx,
|
||||
`WITH base AS (
|
||||
SELECT * FROM rs485_weather_10min WHERE station_id=$1 AND bucket_start >= $2 AND bucket_start <= $3
|
||||
), g AS (
|
||||
SELECT date_trunc('hour', bucket_start) AS grp, SUM(sample_count) AS n_sum FROM base GROUP BY 1
|
||||
) SELECT COUNT(*) FROM g`, stationID, from, to,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -2,6 +2,7 @@ package server
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
@ -126,7 +127,10 @@ func getDataHandler(c *gin.Context) {
|
||||
// 获取数据(改为基于10分钟聚合表的再聚合)
|
||||
points, err := database.GetSeriesFrom10Min(database.GetDB(), stationID, start, end, interval)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "查询数据失败"})
|
||||
log.Printf("查询数据失败: %v", err) // 记录具体错误到服务端日志
|
||||
c.JSON(http.StatusInternalServerError, gin.H{
|
||||
"error": fmt.Sprintf("查询数据失败: %v", err),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user