Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1bcbbf6f44 | ||
|
|
8f973853bd | ||
|
|
6632c38c0c | ||
|
|
8c0bd734c6 | ||
|
|
087b0ec637 | ||
|
|
e90e35cfbc |
@@ -8,11 +8,13 @@
|
|||||||
"sslmode": "disable"
|
"sslmode": "disable"
|
||||||
},
|
},
|
||||||
"valkey": {
|
"valkey": {
|
||||||
"host": "127.0.0.1",
|
"host": "127.0.0.1",
|
||||||
"port": "6379",
|
"port": "6379",
|
||||||
"max_conns": 50,
|
"max_conns": 50,
|
||||||
"timeout_ms": 5000,
|
"timeout_ms": 5000,
|
||||||
"password": "the_valkey_password"
|
"password": "the_valkey_password"
|
||||||
},
|
},
|
||||||
"temp": "value"
|
"temp": "value",
|
||||||
|
"dataset_id": "",
|
||||||
|
"excluded_dataset_ids": ""
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const configFilePath = "config/conf.json"
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
Database struct {
|
Database struct {
|
||||||
Host string `json:"host"`
|
Host string `json:"host"`
|
||||||
@@ -21,14 +23,16 @@ type Config struct {
|
|||||||
Port string `json:"port"`
|
Port string `json:"port"`
|
||||||
MaxConns int `json:"max_conns"`
|
MaxConns int `json:"max_conns"`
|
||||||
TimeoutMs int `json:"timeout_ms"`
|
TimeoutMs int `json:"timeout_ms"`
|
||||||
Password string `json:"password"` // Add this line
|
Password string `json:"password"`
|
||||||
} `json:"valkey"`
|
} `json:"valkey"`
|
||||||
Temp string `json:"temp"`
|
Temp string `json:"temp"`
|
||||||
|
DatasetId string `json:"dataset_id"`
|
||||||
|
ExcludedDatasetIds string `json:"excluded_dataset_ids"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func LoadConfig(file string) (Config, error) {
|
func LoadConfig() (Config, error) {
|
||||||
var config Config
|
var config Config
|
||||||
configFile, err := os.Open(file)
|
configFile, err := os.Open(configFilePath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return config, fmt.Errorf("failed to open config file: %w", err)
|
return config, fmt.Errorf("failed to open config file: %w", err)
|
||||||
}
|
}
|
||||||
@@ -80,5 +84,13 @@ func LoadConfig(file string) (Config, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Override datasetId and excludedDatasetIds with environment variables
|
||||||
|
if datasetId := os.Getenv("DATASET_ID"); datasetId != "" {
|
||||||
|
config.DatasetId = datasetId
|
||||||
|
}
|
||||||
|
if excludedDatasetIds := os.Getenv("EXCLUDED_DATASET_IDS"); excludedDatasetIds != "" {
|
||||||
|
config.ExcludedDatasetIds = excludedDatasetIds
|
||||||
|
}
|
||||||
|
|
||||||
return config, nil
|
return config, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import (
|
|||||||
|
|
||||||
func ConnectToPostgreSQL() (*sql.DB, error) {
|
func ConnectToPostgreSQL() (*sql.DB, error) {
|
||||||
fmt.Println("Connecting to PostgreSQL...")
|
fmt.Println("Connecting to PostgreSQL...")
|
||||||
config, err := LoadConfig("config/conf.json")
|
config, err := LoadConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -57,7 +57,7 @@ func DisconnectFromPostgreSQL(db *sql.DB) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func PrintDBConfig() {
|
func PrintDBConfig() {
|
||||||
config, err := LoadConfig("config/conf.json")
|
config, err := LoadConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println("Error loading config:", err)
|
fmt.Println("Error loading config:", err)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ type ValkeyConfig struct {
|
|||||||
Port string `json:"port"`
|
Port string `json:"port"`
|
||||||
MaxConns int `json:"max_conns"`
|
MaxConns int `json:"max_conns"`
|
||||||
TimeoutMs int `json:"timeout_ms"`
|
TimeoutMs int `json:"timeout_ms"`
|
||||||
Password string `json:"password"` // Add this line
|
Password string `json:"password"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
||||||
@@ -56,9 +56,9 @@ func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
|||||||
return config, nil
|
return config, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func ConnectToValkey(configPath string) (valkey.Client, error) {
|
func ConnectToValkey() (valkey.Client, error) {
|
||||||
fmt.Println("Loading configuration...")
|
fmt.Println("Loading configuration...")
|
||||||
config, err := LoadConfig(configPath)
|
config, err := LoadConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to load config: %v", err)
|
return nil, fmt.Errorf("failed to load config: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
18
data/data.go
18
data/data.go
@@ -1,9 +1,10 @@
|
|||||||
package data
|
package data
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"log"
|
|
||||||
"encoding/xml"
|
"encoding/xml"
|
||||||
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"strings"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Data struct {
|
type Data struct {
|
||||||
@@ -126,13 +127,20 @@ type Data struct {
|
|||||||
} `xml:"ServiceDelivery"`
|
} `xml:"ServiceDelivery"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func FetchData(timestamp string) (*Data, error) {
|
func FetchData(timestamp, datasetId, excludedDatasetIds string) (*Data, error) {
|
||||||
client := &http.Client{}
|
client := &http.Client{}
|
||||||
requestorId := "ti1-" + timestamp
|
requestorId := "ti1-" + timestamp
|
||||||
|
|
||||||
url := "https://api.entur.io/realtime/v1/rest/et?useOriginalId=true&maxSize=100000&requestorId=" + requestorId
|
baseURL := "https://api.entur.io/realtime/v1/rest/et?useOriginalId=true&maxSize=100000&requestorId=" + requestorId
|
||||||
log.Println("Fetching data from URL:", url)
|
|
||||||
resp, err := client.Get(url)
|
if datasetId != "" {
|
||||||
|
baseURL += "&datasetId=" + datasetId
|
||||||
|
} else if excludedDatasetIds != "" {
|
||||||
|
baseURL += "&excludedDatasetIds=" + strings.ReplaceAll(excludedDatasetIds, ",", "&excludedDatasetIds=")
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Println("Fetching data from URL:", baseURL)
|
||||||
|
resp, err := client.Get(baseURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,13 +3,37 @@ package database
|
|||||||
import (
|
import (
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"ti1/config"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func GetDatasetVariable(config config.Config) string {
|
||||||
|
if config.DatasetId != "" {
|
||||||
|
fmt.Println(config.DatasetId)
|
||||||
|
return config.DatasetId
|
||||||
|
} else if config.ExcludedDatasetIds != "" {
|
||||||
|
result := "EX." + config.ExcludedDatasetIds
|
||||||
|
fmt.Println(result)
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
fmt.Println("")
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
func InsertServiceDelivery(db *sql.DB, responseTimestamp string, recordedAtTime string) (int, error) {
|
func InsertServiceDelivery(db *sql.DB, responseTimestamp string, recordedAtTime string) (int, error) {
|
||||||
fmt.Println("Inserting ServiceDelivery...")
|
fmt.Println("Inserting ServiceDelivery...")
|
||||||
var id int
|
var id int
|
||||||
|
|
||||||
err := db.QueryRow("INSERT INTO public.ServiceDelivery (ResponseTimestamp, RecordedAtTime) VALUES ($1, $2) RETURNING ID", responseTimestamp, recordedAtTime).Scan(&id)
|
// Load configuration
|
||||||
|
config, err := config.LoadConfig()
|
||||||
|
if err != nil {
|
||||||
|
fmt.Println("Error loading config:", err)
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get dataset variable
|
||||||
|
datasetVariable := GetDatasetVariable(config)
|
||||||
|
|
||||||
|
err = db.QueryRow("INSERT INTO public.ServiceDelivery (ResponseTimestamp, RecordedAtTime, Source) VALUES ($1, $2, $3) RETURNING ID", responseTimestamp, recordedAtTime, datasetVariable).Scan(&id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println(err)
|
fmt.Println(err)
|
||||||
return 0, err
|
return 0, err
|
||||||
|
|||||||
@@ -68,6 +68,7 @@ func SetupDB() error {
|
|||||||
id INTEGER PRIMARY KEY DEFAULT nextval('public.servicedelivery_id_seq'),
|
id INTEGER PRIMARY KEY DEFAULT nextval('public.servicedelivery_id_seq'),
|
||||||
responsetimestamp TIMESTAMPTZ,
|
responsetimestamp TIMESTAMPTZ,
|
||||||
recordedattime TIMESTAMPTZ,
|
recordedattime TIMESTAMPTZ,
|
||||||
|
source VARCHAR,
|
||||||
data JSON
|
data JSON
|
||||||
);`,
|
);`,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -22,7 +22,7 @@ func DBData(data *data.Data) {
|
|||||||
defer db.Close()
|
defer db.Close()
|
||||||
|
|
||||||
// Connect to Valkey
|
// Connect to Valkey
|
||||||
valkeyClient, err := config.ConnectToValkey("config/conf.json")
|
valkeyClient, err := config.ConnectToValkey()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Failed to connect to Valkey: %v", err)
|
log.Fatalf("Failed to connect to Valkey: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
11
main.go
11
main.go
@@ -2,6 +2,7 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"log"
|
"log"
|
||||||
|
"ti1/config"
|
||||||
"ti1/data"
|
"ti1/data"
|
||||||
"ti1/database"
|
"ti1/database"
|
||||||
"ti1/export"
|
"ti1/export"
|
||||||
@@ -12,8 +13,14 @@ func main() {
|
|||||||
log.Println("ti1 v0.2.1")
|
log.Println("ti1 v0.2.1")
|
||||||
log.Println("Starting...")
|
log.Println("Starting...")
|
||||||
|
|
||||||
|
// Load configuration
|
||||||
|
cfg, err := config.LoadConfig()
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("Failed to load config: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
// Setup the database
|
// Setup the database
|
||||||
err := database.SetupDB()
|
err = database.SetupDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Database setup failed: %v", err)
|
log.Fatalf("Database setup failed: %v", err)
|
||||||
}
|
}
|
||||||
@@ -25,7 +32,7 @@ func main() {
|
|||||||
for {
|
for {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
|
||||||
data, err := data.FetchData(starttimestamp)
|
data, err := data.FetchData(starttimestamp, cfg.DatasetId, cfg.ExcludedDatasetIds)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal(err)
|
log.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user