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