feat：新增数据导出器
This commit is contained in:
parent
6fb4655a15
commit
24dca2f489
@ -38,8 +38,11 @@ func main() {
|
||||
server.SetupLogger()
|
||||
|
||||
// 初始化数据库连接
|
||||
_ = database.GetDB() // 确保数据库连接已初始化
|
||||
_ = database.GetDB() // 确保PostgreSQL连接已初始化
|
||||
defer database.Close()
|
||||
// 初始化MySQL连接（如果配置存在）
|
||||
_ = database.GetMySQL()
|
||||
defer database.CloseMySQL()
|
||||
|
||||
// 启动前自检
|
||||
if !*noSelftest {
|
||||
@ -115,21 +118,42 @@ func main() {
|
||||
}
|
||||
|
||||
// 根据命令行参数启动服务
|
||||
startExporterBackground := func(wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
wg.Add(1)
|
||||
}
|
||||
go func() {
|
||||
defer func() {
|
||||
if wg != nil {
|
||||
wg.Done()
|
||||
}
|
||||
}()
|
||||
log.Println("启动数据导出器（10分钟）...")
|
||||
ctx := context.Background()
|
||||
exporter := tools.NewExporter()
|
||||
if err := exporter.Start(ctx); err != nil {
|
||||
log.Printf("导出器退出: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
if *webOnly {
|
||||
// 只启动Web服务器
|
||||
// 只启动Web服务器 + 导出器
|
||||
startExporterBackground(nil)
|
||||
log.Println("启动Web服务器模式...")
|
||||
if err := server.StartGinServer(); err != nil {
|
||||
log.Fatalf("启动Web服务器失败: %v", err)
|
||||
}
|
||||
} else if *udpOnly {
|
||||
// 只启动UDP服务器
|
||||
// 只启动UDP服务器 + 导出器
|
||||
startExporterBackground(nil)
|
||||
log.Println("启动UDP服务器模式...")
|
||||
if err := server.StartUDPServer(); err != nil {
|
||||
log.Fatalf("启动UDP服务器失败: %v", err)
|
||||
}
|
||||
} else {
|
||||
// 同时启动UDP和Web服务器
|
||||
log.Println("启动完整模式：UDP + Web服务器...")
|
||||
// 同时启动UDP和Web服务器 + 导出器
|
||||
log.Println("启动完整模式：UDP + Web服务器 + 导出器...")
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
@ -152,6 +176,7 @@ func main() {
|
||||
}
|
||||
}()
|
||||
|
||||
startExporterBackground(&wg)
|
||||
wg.Wait()
|
||||
}
|
||||
}
|
||||
|
||||
@ -11,3 +11,11 @@ database:
|
||||
|
||||
forecast:
|
||||
caiyun_token: "ZAcZq49qzibr10F0"
|
||||
|
||||
mysql:
|
||||
host: "127.0.0.1"
|
||||
port: 3306
|
||||
user: "remote"
|
||||
password: "your_password"
|
||||
dbname: "rtk_data"
|
||||
params: "parseTime=true&loc=Asia%2FShanghai"
|
||||
|
||||
2
go.mod
2
go.mod
@ -6,11 +6,13 @@ toolchain go1.24.5
|
||||
|
||||
require (
|
||||
github.com/gin-gonic/gin v1.10.1
|
||||
github.com/go-sql-driver/mysql v1.8.1
|
||||
github.com/lib/pq v1.10.9
|
||||
gopkg.in/yaml.v3 v3.0.1
|
||||
)
|
||||
|
||||
require (
|
||||
filippo.io/edwards25519 v1.1.0 // indirect
|
||||
github.com/bytedance/sonic v1.11.6 // indirect
|
||||
github.com/bytedance/sonic/loader v0.1.1 // indirect
|
||||
github.com/cloudwego/base64x v0.1.4 // indirect
|
||||
|
||||
13
go.sum
13
go.sum
@ -1,3 +1,5 @@
|
||||
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
|
||||
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
|
||||
github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc0=
|
||||
github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4=
|
||||
github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM=
|
||||
@ -7,6 +9,7 @@ github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJ
|
||||
github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
|
||||
github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/gabriel-vasile/mimetype v1.4.3 h1:in2uUcidCuFcDKtdcBxlR0rJ1+fsokWf+uqxgUFjbI0=
|
||||
github.com/gabriel-vasile/mimetype v1.4.3/go.mod h1:d8uq/6HKRL6CGdk+aubisF/M5GcPfT7nKyLpA0lbSSk=
|
||||
@ -14,14 +17,20 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE
|
||||
github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
|
||||
github.com/gin-gonic/gin v1.10.1 h1:T0ujvqyCSqRopADpgPgiTT63DUQVSfojyME59Ei63pQ=
|
||||
github.com/gin-gonic/gin v1.10.1/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y=
|
||||
github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
|
||||
github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
|
||||
github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
|
||||
github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
|
||||
github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
|
||||
github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
|
||||
github.com/go-playground/validator/v10 v10.20.0 h1:K9ISHbSaI0lyB2eWMPJo+kOS/FBExVwjEviJTixqxL8=
|
||||
github.com/go-playground/validator/v10 v10.20.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM=
|
||||
github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
|
||||
github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
|
||||
github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
|
||||
github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
|
||||
github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
|
||||
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
|
||||
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
|
||||
@ -42,6 +51,7 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
|
||||
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
@ -53,6 +63,7 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
|
||||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
|
||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
||||
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
|
||||
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
|
||||
@ -71,6 +82,8 @@ golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
|
||||
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
|
||||
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/protobuf v1.34.1 h1:9ddQBjfCyZPOHPUiPxpYESBLc+T8P3E+Vo4IbKZgFWg=
|
||||
google.golang.org/protobuf v1.34.1/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
|
||||
|
||||
@ -28,10 +28,21 @@ type ForecastConfig struct {
|
||||
CaiyunToken string `yaml:"caiyun_token"`
|
||||
}
|
||||
|
||||
// MySQLConfig MySQL 连接配置（用于 rtk_data）
|
||||
type MySQLConfig struct {
|
||||
Host string `yaml:"host"`
|
||||
Port int `yaml:"port"`
|
||||
User string `yaml:"user"`
|
||||
Password string `yaml:"password"`
|
||||
DBName string `yaml:"dbname"`
|
||||
Params string `yaml:"params"` // 例如: parseTime=true&loc=Asia%2FShanghai
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
Server ServerConfig `yaml:"server"`
|
||||
Database DatabaseConfig `yaml:"database"`
|
||||
Forecast ForecastConfig `yaml:"forecast"`
|
||||
MySQL MySQLConfig `yaml:"mysql"`
|
||||
}
|
||||
|
||||
var (
|
||||
@ -89,6 +100,9 @@ func (c *Config) validate() error {
|
||||
if c.Database.SSLMode == "" {
|
||||
c.Database.SSLMode = "disable" // 默认禁用SSL
|
||||
}
|
||||
if c.MySQL.Port <= 0 {
|
||||
c.MySQL.Port = 3306
|
||||
}
|
||||
// CaiyunToken 允许为空：表示不启用彩云定时任务
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -6,6 +6,7 @@ import (
|
||||
"sync"
|
||||
"weatherstation/internal/config"
|
||||
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
_ "github.com/lib/pq"
|
||||
)
|
||||
|
||||
@ -48,3 +49,40 @@ func Close() error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// -------------------- MySQL 连接（rtk_data） --------------------
|
||||
|
||||
var (
|
||||
mysqlInstance *sql.DB
|
||||
mysqlOnce sync.Once
|
||||
)
|
||||
|
||||
// GetMySQL 返回 MySQL 连接单例（rtk_data）
|
||||
func GetMySQL() *sql.DB {
|
||||
mysqlOnce.Do(func() {
|
||||
cfg := config.GetConfig().MySQL
|
||||
var dsn string
|
||||
if cfg.Params != "" {
|
||||
dsn = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.DBName, cfg.Params)
|
||||
} else {
|
||||
dsn = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.DBName)
|
||||
}
|
||||
var err error
|
||||
mysqlInstance, err = sql.Open("mysql", dsn)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("无法连接到 MySQL: %v", err))
|
||||
}
|
||||
if err = mysqlInstance.Ping(); err != nil {
|
||||
panic(fmt.Sprintf("MySQL 连接测试失败: %v", err))
|
||||
}
|
||||
})
|
||||
return mysqlInstance
|
||||
}
|
||||
|
||||
// CloseMySQL 关闭 MySQL 连接
|
||||
func CloseMySQL() error {
|
||||
if mysqlInstance != nil {
|
||||
return mysqlInstance.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
338
internal/tools/exporter.go
Normal file
338
internal/tools/exporter.go
Normal file
@ -0,0 +1,338 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"math"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
"weatherstation/internal/database"
|
||||
)
|
||||
|
||||
// Exporter 负责每10分钟导出 CSV（含ZTD融合）
|
||||
type Exporter struct {
|
||||
pg *sql.DB
|
||||
my *sql.DB
|
||||
loc *time.Location // Asia/Shanghai
|
||||
}
|
||||
|
||||
func NewExporter() *Exporter {
|
||||
loc, _ := time.LoadLocation("Asia/Shanghai")
|
||||
if loc == nil {
|
||||
loc = time.FixedZone("CST", 8*3600)
|
||||
}
|
||||
return &Exporter{
|
||||
pg: database.GetDB(),
|
||||
my: database.GetMySQL(),
|
||||
loc: loc,
|
||||
}
|
||||
}
|
||||
|
||||
// Start 启动调度循环（阻塞）
|
||||
func (e *Exporter) Start(ctx context.Context) error {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
now := time.Now().In(e.loc)
|
||||
next := alignToNextBucketEnd(now, 10).Add(30 * time.Second)
|
||||
log.Printf("exporter: now=%s next_run=%s", now.Format("2006-01-02 15:04:05"), next.Format("2006-01-02 15:04:05"))
|
||||
delay := time.Until(next)
|
||||
if delay > 0 {
|
||||
timer := time.NewTimer(delay)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
return ctx.Err()
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
|
||||
end := time.Now().In(e.loc).Truncate(time.Minute)
|
||||
end = alignToPrevBucketEnd(end, 10)
|
||||
start := end.Add(-10 * time.Minute)
|
||||
|
||||
if err := e.exportBucket(ctx, start, end); err != nil {
|
||||
log.Printf("export bucket %s-%s failed: %v", start.Format("2006-01-02 15:04:05"), end.Format("2006-01-02 15:04:05"), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// exportBucket 导出一个10分钟桶（CST）
|
||||
func (e *Exporter) exportBucket(ctx context.Context, bucketStart, bucketEnd time.Time) error {
|
||||
utcDay := bucketEnd.UTC().Format("2006-01-02")
|
||||
outDir := filepath.Join("export_data")
|
||||
histDir := filepath.Join("export_data", "history")
|
||||
if err := os.MkdirAll(histDir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
activePath := filepath.Join(outDir, fmt.Sprintf("weather_data_%s.csv", utcDay))
|
||||
log.Printf("exporter: begin bucket start=%s end=%s -> %s", bucketStart.Format("2006-01-02 15:04:05"), bucketEnd.Format("2006-01-02 15:04:05"), activePath)
|
||||
|
||||
// 轮转上一 UTC 日的文件（若存在且未压缩）
|
||||
if err := rotatePreviousUTC(outDir, histDir, utcDay); err != nil {
|
||||
log.Printf("rotate previous day failed: %v", err)
|
||||
}
|
||||
|
||||
needHeader := ensureFileWithHeader(activePath)
|
||||
|
||||
rows, err := e.pg.QueryContext(ctx, `
|
||||
SELECT
|
||||
s.latitude,
|
||||
s.longitude,
|
||||
s.device_id,
|
||||
s.altitude,
|
||||
r.pressure_hpa_x100,
|
||||
r.temp_c_x100,
|
||||
r.wind_speed_ms_x1000,
|
||||
r.wind_dir_deg,
|
||||
r.humidity_pct,
|
||||
r.bucket_start
|
||||
FROM stations s
|
||||
JOIN rs485_weather_10min r ON r.station_id = s.station_id AND r.bucket_start = $1
|
||||
WHERE s.device_type = 'WH65LP'
|
||||
AND s.latitude IS NOT NULL AND s.longitude IS NOT NULL
|
||||
AND s.latitude <> 0 AND s.longitude <> 0
|
||||
ORDER BY s.device_id
|
||||
`, bucketStart)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
f, err := os.OpenFile(activePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
if needHeader {
|
||||
if _, err := f.WriteString(headerLine() + "\n"); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
var total, ztdHit, ztdMiss int
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
lat, lon, elev sql.NullFloat64
|
||||
deviceID string
|
||||
pX100, tX100 sql.NullInt64
|
||||
wsX1000 sql.NullInt64
|
||||
wdDeg sql.NullInt64
|
||||
rh sql.NullInt64
|
||||
bucketStartTS time.Time
|
||||
)
|
||||
if err := rows.Scan(&lat, &lon, &deviceID, &elev, &pX100, &tX100, &wsX1000, &wdDeg, &rh, &bucketStartTS); err != nil {
|
||||
log.Printf("scan row failed: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// date_time 使用 Go 端按 CST 格式化
|
||||
dateTimeStr := bucketStartTS.In(e.loc).Format("2006-01-02 15:04:05")
|
||||
|
||||
var pressureStr, tempStr, wsStr, wdStr, rhStr string
|
||||
if pX100.Valid {
|
||||
pressureStr = fmtFloat(float64(pX100.Int64)/100.0, 2)
|
||||
}
|
||||
if tX100.Valid {
|
||||
tempStr = fmtFloat(float64(tX100.Int64)/100.0, 2)
|
||||
}
|
||||
if wsX1000.Valid {
|
||||
wsStr = fmtFloat(float64(wsX1000.Int64)/1000.0, 3)
|
||||
}
|
||||
if wdDeg.Valid {
|
||||
wdStr = fmtFloat(float64(wdDeg.Int64), 0)
|
||||
}
|
||||
if rh.Valid {
|
||||
rhStr = fmtFloat(float64(rh.Int64), 0)
|
||||
}
|
||||
|
||||
ztdStr := e.lookupZTD(ctx, deviceID, bucketEnd)
|
||||
if ztdStr != "" {
|
||||
ztdHit++
|
||||
} else {
|
||||
ztdMiss++
|
||||
}
|
||||
|
||||
var b strings.Builder
|
||||
b.WriteString(fmtNullFloat(lat))
|
||||
b.WriteByte(',')
|
||||
b.WriteString(fmtNullFloat(lon))
|
||||
b.WriteByte(',')
|
||||
b.WriteString(deviceID)
|
||||
b.WriteByte(',')
|
||||
b.WriteByte(',') // station_name 留空
|
||||
b.WriteString(dateTimeStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteString(fmtNullFloat(elev))
|
||||
b.WriteByte(',')
|
||||
b.WriteString(pressureStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteString(tempStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteByte(',') // dewpoint 留空
|
||||
b.WriteString(wsStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteString(wdStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteString(rhStr)
|
||||
b.WriteByte(',')
|
||||
b.WriteString(ztdStr)
|
||||
b.WriteByte(',') // pwv 留空
|
||||
|
||||
if _, err := f.WriteString(b.String() + "\n"); err != nil {
|
||||
log.Printf("write csv failed: %v", err)
|
||||
}
|
||||
total++
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("exporter: done bucket start=%s end=%s, rows=%d, ztd_hit=%d, ztd_miss=%d", bucketStart.Format("2006-01-02 15:04:05"), bucketEnd.Format("2006-01-02 15:04:05"), total, ztdHit, ztdMiss)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Exporter) lookupZTD(ctx context.Context, deviceID string, bucketEnd time.Time) string {
|
||||
if e.my == nil {
|
||||
return ""
|
||||
}
|
||||
var ztd sql.NullFloat64
|
||||
var ts time.Time
|
||||
err := e.my.QueryRowContext(ctx, `
|
||||
SELECT ztd, timestamp FROM rtk_data
|
||||
WHERE station_id = ?
|
||||
AND ABS(TIMESTAMPDIFF(MINUTE, timestamp, ?)) <= 5
|
||||
LIMIT 1
|
||||
`, deviceID, bucketEnd).Scan(&ztd, &ts)
|
||||
if err != nil {
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
log.Printf("mysql query ztd failed: %v", err)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
if !ztd.Valid {
|
||||
return ""
|
||||
}
|
||||
return fmtFloat(ztd.Float64*1000.0, -1)
|
||||
}
|
||||
|
||||
func ensureFileWithHeader(path string) bool {
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return false
|
||||
}
|
||||
dir := filepath.Dir(path)
|
||||
_ = os.MkdirAll(dir, 0o755)
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
log.Printf("create csv failed: %v", err)
|
||||
return false
|
||||
}
|
||||
_ = f.Close()
|
||||
return true
|
||||
}
|
||||
|
||||
func headerLine() string {
|
||||
return "latitude,longitude,station_id,station_name,date_time,elevation,pressure,temperature,dewpoint,wind_speed,wind_direction,relative_humidity,ztd,pwv"
|
||||
}
|
||||
|
||||
func alignToNextBucketEnd(t time.Time, minutes int) time.Time {
|
||||
m := t.Minute()
|
||||
next := (m/minutes + 1) * minutes
|
||||
dt := time.Duration(next-m) * time.Minute
|
||||
return t.Truncate(time.Minute).Add(dt).Truncate(time.Minute)
|
||||
}
|
||||
|
||||
func alignToPrevBucketEnd(t time.Time, minutes int) time.Time {
|
||||
m := t.Minute()
|
||||
prev := (m / minutes) * minutes
|
||||
return t.Truncate(time.Minute).Add(time.Duration(prev-m) * time.Minute).Add(time.Duration(minutes) * time.Minute)
|
||||
}
|
||||
|
||||
func fmtNullFloat(v sql.NullFloat64) string {
|
||||
if v.Valid {
|
||||
return fmtFloat(v.Float64, -1)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// fmtFloat: prec < 0 表示用不固定小数（去除无意义零），否则保留指定小数位
|
||||
func fmtFloat(fv float64, prec int) string {
|
||||
if prec >= 0 {
|
||||
return fmt.Sprintf("%.*f", prec, fv)
|
||||
}
|
||||
s := fmt.Sprintf("%.10f", fv)
|
||||
s = strings.TrimRight(s, "0")
|
||||
s = strings.TrimRight(s, ".")
|
||||
if s == "-0" {
|
||||
s = "0"
|
||||
}
|
||||
if s == "" || s == "-" || s == "+" || s == "." {
|
||||
return "0"
|
||||
}
|
||||
if math.Abs(fv) < 1e-9 {
|
||||
return "0"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// rotatePreviousUTC 将上一UTC日的活跃CSV压缩到history目录
|
||||
func rotatePreviousUTC(outDir, histDir, currentUTC string) error {
|
||||
// 计算昨日 UTC 日期
|
||||
curDay, err := time.Parse("2006-01-02", currentUTC)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
yesterday := curDay.Add(-24 * time.Hour).Format("2006-01-02")
|
||||
prevPath := filepath.Join(outDir, fmt.Sprintf("weather_data_%s.csv", yesterday))
|
||||
gzPath := filepath.Join(histDir, fmt.Sprintf("weather_data_%s.csv.gz", yesterday))
|
||||
|
||||
if _, err := os.Stat(prevPath); err != nil {
|
||||
return nil
|
||||
}
|
||||
if _, err := os.Stat(gzPath); err == nil {
|
||||
// 已压缩
|
||||
return nil
|
||||
}
|
||||
return gzipFile(prevPath, gzPath)
|
||||
}
|
||||
|
||||
func gzipFile(src, dst string) error {
|
||||
srcF, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer srcF.Close()
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
dstF, err := os.Create(dst)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_ = dstF.Close()
|
||||
}()
|
||||
|
||||
gw := gzip.NewWriter(dstF)
|
||||
gw.Name = filepath.Base(src)
|
||||
gw.ModTime = time.Now()
|
||||
defer func() { _ = gw.Close() }()
|
||||
|
||||
if _, err := io.Copy(gw, srcF); err != nil {
|
||||
return err
|
||||
}
|
||||
// 压缩成功后删除原文件
|
||||
return os.Remove(src)
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user