diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..056048c --- /dev/null +++ b/.env.example @@ -0,0 +1,19 @@ +# ============================================================================== +# NBA_Go Environment Configuration +# Target: Hetzner Coolify Server (178.105.149.129) +# ============================================================================== + +# Database Credentials +DB_USER=postgres +DB_PASSWORD=postgrespassword +DB_NAME=appdb +DB_SSLMODE=disable + +# HAProxy Ingress Endpoints (PostgreSQL 17 Replicated Cluster) +DB_WRITE_HOST=178.105.149.129 +DB_WRITE_PORT=5437 +DB_READ_HOST=178.105.149.129 +DB_READ_PORT=5438 + +# Administrative Auth Token +ADMIN_SECRET=nba_super_secret_admin_token_2025 diff --git a/.gitignore b/.gitignore index 3bf06a5..ed0b27f 100644 --- a/.gitignore +++ b/.gitignore @@ -47,4 +47,6 @@ docs/digital_ocean/digitaal ocean llms.txt .docs -NBA_Go \ No newline at end of file +NBA_Go + +.env.old \ No newline at end of file diff --git a/README.md b/README.md index e61649a..da475fd 100644 --- a/README.md +++ b/README.md @@ -160,17 +160,38 @@ Services will be available at: ### Importing Data -The application has a dual-mode entry point. To run the initial data import (migrations + scraping): +The application has a dual-mode entry point. To run data imports (migrations + scraping): +#### 1. Local CLI Execution (Recommended for targeted imports) +```bash +# Export environment variables from .env +export $(grep -v '^#' .env | xargs) + +# Run the import pipeline +go run . import-data +``` + +#### 2. Local Docker Stack ```bash docker-compose -f docker-compose.local.yml run --rm db-init ``` -This runs `main.go` with the `import-data` argument, which: +#### 3. Production / Coolify Deployment +```bash +docker compose --profile init run --rm db-init +``` -1. Runs all GORM AutoMigrate operations -2. Scrapes Basketball Reference for player advanced stats, totals, game schedules, and box scores -3. Upserts all data into PostgreSQL +--- + +### Ingestion Pipeline & Scraping Architecture + +The data import engine (`import.go` & `services/`) features a robust, resilient ingestion workflow designed to safely handle thousands of games: + +- **20-Day Temporal Session Chunks**: Large date ranges (such as full seasons) are automatically partitioned into 20-day sliding windows (~80ā€“120 games per chunk). +- **Immediate Incremental Persistence**: Scraped data (`line_scores`, `player_game_basic_stats`, `player_game_adv_stats`, `team_game_basic_stats`, `team_game_adv_stats`) is immediately committed and upserted into PostgreSQL at the end of each chunk rather than held in memory until the end of the run. +- **Inter-Chunk Cool-Off Period**: Enforces a 20-second pause ($\pm 25\%$ jitter) between chunks to avoid rate limiting and IP blocks from upstream sources. +- **Smart Skip & Idempotent Resumption**: Automatically checks `WHERE game_id NOT IN (SELECT DISTINCT game_id FROM line_scores WHERE deleted_at IS NULL)` so completed games/chunks are instantly skipped. +- **Graceful Interrupt Handling (`Ctrl+C`)**: Captures `SIGINT`/`SIGTERM` via `context.Context`. If interrupted, in-flight scraped games in the active chunk are flushed to the database before cleanly exiting without data loss. ### Stopping diff --git a/docker-compose.yml b/docker-compose.yml index dc14cca..ae8e484 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,8 +1,10 @@ -# "import-data" disabled services: db-init: build: . command: ["/nba_go", "import-data"] + profiles: + - init + - import networks: - coolify-shared # - api_network diff --git a/import.go b/import.go index 1f4e5e8..eba4043 100644 --- a/import.go +++ b/import.go @@ -1,9 +1,13 @@ package main import ( + "context" + "errors" "log" + "os" + "os/signal" + "syscall" "time" - "fmt" "github.com/nprasad2077/NBA_Go/services" "github.com/nprasad2077/NBA_Go/utils" @@ -12,7 +16,7 @@ import ( // importPlayerAdvanced fetches and stores advanced stats for seasons func importPlayerAdvanced(db *gorm.DB) { - for season := 2001; season <= 2002; season++ { + for season := 2001; season <= 2001; season++ { if err := services.FetchAndStorePlayerAdvancedScrapedStats(db, season, false); err != nil { log.Printf("advanced import failed for %d: %v", season, err) } @@ -24,7 +28,7 @@ func importPlayerAdvanced(db *gorm.DB) { // importPlayerAdvancedPlayoffs fetches and stores advanced stats for playoffs seasons func importPlayerAdvancedPlayoffs(db *gorm.DB) { - for season := 2001; season <= 2002; season++ { + for season := 2001; season <= 2001; season++ { if err := services.FetchAndStorePlayerAdvancedScrapedStats(db, season, true); err != nil { log.Printf("advanced import failed for %d: %v", season, err) } @@ -36,7 +40,7 @@ func importPlayerAdvancedPlayoffs(db *gorm.DB) { // importPlayerTotalsScrape fetches & stores scraped regular-season total stats func importPlayerTotalsScrape(db *gorm.DB) { - for season := 2001; season <= 2002; season++ { + for season := 2001; season <= 2001; season++ { if err := services.FetchAndStorePlayerTotalScrapedStats(db, season, false); err != nil { log.Printf("scraped totals import failed for %d: %v", season, err) } @@ -48,7 +52,7 @@ func importPlayerTotalsScrape(db *gorm.DB) { // importPlayerPlayoffsScrape fetches & stores scraped playoff total stats func importPlayerTotalsPlayoffsScrape(db *gorm.DB) { - for season := 2001; season <= 2002; season++ { + for season := 2001; season <= 2001; season++ { if err := services.FetchAndStorePlayerTotalScrapedStats(db, season, true); err != nil { log.Printf("scraped playoffs import failed for %d: %v", season, err) } @@ -62,7 +66,7 @@ func importPlayerTotalsPlayoffsScrape(db *gorm.DB) { func importGameSchedules(db *gorm.DB) { months := []string{"october", "november", "december", "january", "february", "march", "april", "may", "june"} - for season := 2001; season <= 2002; season++ { + for season := 2001; season <= 2001; season++ { log.Printf("--- Starting Game Schedule Import for Season: %d ---", season) for _, month := range months { if err := services.FetchAndStoreGameSchedule(db, season, month); err != nil { @@ -77,18 +81,23 @@ func importGameSchedules(db *gorm.DB) { } // importBoxScores fetches and stores all box score data (line scores, player/team stats) -// for games within a recent date range. +// in safe 20-day session chunks with cool-off pauses and immediate database upserts. func importBoxScores(db *gorm.DB) { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + from := time.Date(2000, time.October, 31, 0, 0, 0, 0, time.UTC) to := time.Date(2001, time.June, 30, 23, 59, 59, 0, time.UTC) - dateRangeComment := fmt.Sprintf("--- Starting Box Score Data Import for games between %s and %s ---", - from.Format("January 2, 2006"), - to.Format("January 2, 2006")) + chunkDays := 20 // 20-day session chunks + coolOff := 20 * time.Second // 20s cool-off with jitter between chunks + skipExisting := true // Automatically skip games that already have line scores - log.Println(dateRangeComment) - - if err := services.FetchAndStoreBoxScoreDataForDateRange(db, from, to); err != nil { + if err := services.FetchAndStoreBoxScoreDataChunked(ctx, db, from, to, chunkDays, coolOff, skipExisting); err != nil { + if errors.Is(err, context.Canceled) { + log.Println("šŸ‘‹ Box score import interrupted by user. Safe to resume anytime!") + return + } log.Fatalf("Box score import failed: %v", err) } @@ -175,8 +184,8 @@ func importPlayerShotCharts(db *gorm.DB) { // importMarkPlayoffGames marks games as playoff using the dedicated Basketball Reference playoff schedule. func importMarkPlayoffGames(db *gorm.DB) { - for season := 2001; season <= 2002; season++ { - if err := services.FetchAndMarkPlayoffGames(db, season+1); err != nil { + for season := 2001; season <= 2001; season++ { + if err := services.FetchAndMarkPlayoffGames(db, season); err != nil { log.Printf("playoff marking failed for %d: %v", season, err) } log.Printf("Playoff games marked for season: %d", season) diff --git a/services/box_score_scrape_service.go b/services/box_score_scrape_service.go index 080b4d2..6620269 100644 --- a/services/box_score_scrape_service.go +++ b/services/box_score_scrape_service.go @@ -1,6 +1,7 @@ package services import ( + "context" "fmt" "log" "net/http" @@ -43,53 +44,154 @@ func uncommentDoc(doc *goquery.Document) *goquery.Document { return doc } -// FetchAndStoreBoxScoreDataForDateRange fetches games and batch processes their box scores concurrently. +// FetchAndStoreBoxScoreDataForDateRange provides backward-compatibility by delegating +// to the chunked scraper with default 20-day chunks and 20s cool-off. func FetchAndStoreBoxScoreDataForDateRange(db *gorm.DB, from, to time.Time) error { - var games []models.Game - if err := db.Where("date >= ? AND date < ?", from, to.Add(24*time.Hour)).Find(&games).Error; err != nil { - return fmt.Errorf("failed to query games from DB: %w", err) + return FetchAndStoreBoxScoreDataChunked(context.Background(), db, from, to, 20, 20*time.Second, true) +} + +// FetchAndStoreBoxScoreDataChunked processes box scores in temporal chunks (e.g. 20 days) +// with immediate per-chunk database upserts, cool-off periods, and graceful interrupt handling. +func FetchAndStoreBoxScoreDataChunked( + ctx context.Context, + db *gorm.DB, + from, to time.Time, + chunkDays int, + coolOffBase time.Duration, + skipExisting bool, +) error { + if chunkDays <= 0 { + chunkDays = 20 } - if len(games) == 0 { - log.Println("No games found to process in the specified date range.") - return nil - } - log.Printf("Found %d games to process. Initializing concurrent scraping...", len(games)) + chunkDuration := time.Duration(chunkDays) * 24 * time.Hour + totalDays := int(to.Sub(from).Hours()/24) + 1 - // --- Concurrency Setup --- + log.Printf("šŸš€ Starting Chunked Box Score Import: %s to %s (%d total days, %d days/chunk)", + from.Format("2006-01-02"), to.Format("2006-01-02"), totalDays, chunkDays) + + currentStart := from + chunkIndex := 1 + totalGamesScraped := 0 + + for currentStart.Before(to) { + select { + case <-ctx.Done(): + log.Println("šŸ›‘ Interrupt received. Halting further chunk processing.") + return ctx.Err() + default: + } + + currentEnd := currentStart.Add(chunkDuration) + if currentEnd.After(to) { + currentEnd = to + } + + log.Printf("\nšŸ“¦ ========================================================") + log.Printf("šŸ“¦ [Chunk %d] Window: %s to %s", chunkIndex, currentStart.Format("2006-01-02"), currentEnd.Format("2006-01-02")) + log.Printf("šŸ“¦ ========================================================") + + // 1. Query games for this chunk + var games []models.Game + query := db.Where("date >= ? AND date <= ?", currentStart, currentEnd.Add(24*time.Hour)) + if skipExisting { + query = query.Where("game_id NOT IN (SELECT DISTINCT game_id FROM line_scores WHERE deleted_at IS NULL)") + } + + if err := query.Order("date ASC").Find(&games).Error; err != nil { + return fmt.Errorf("failed to query games for chunk %d: %w", chunkIndex, err) + } + + if len(games) == 0 { + log.Printf("ā© [Chunk %d] All games in this window are already scraped or none found. Skipping.", chunkIndex) + currentStart = currentEnd.Add(24 * time.Hour) + chunkIndex++ + continue + } + + log.Printf("Found %d pending games to scrape in Chunk %d. Starting worker pool...", len(games), chunkIndex) + + // 2. Concurrently scrape the chunk's games + results := processGamesWithWorkers(ctx, games) + + // 3. IMMEDIATELY UPSERT chunk results to DB + if len(results) > 0 { + log.Printf("šŸ’¾ Saving and upserting data for %d games from Chunk %d into PostgreSQL...", len(results), chunkIndex) + if err := persistScrapedResults(db, results); err != nil { + log.Printf("āŒ Failed to upsert results for chunk %d: %v", chunkIndex, err) + return err + } + totalGamesScraped += len(results) + log.Printf("āœ… [Chunk %d] Successfully saved %d games to database. (Total so far: %d)", + chunkIndex, len(results), totalGamesScraped) + } + + // Check if interrupted during chunk processing + if ctx.Err() != nil { + log.Printf("šŸ›‘ Process interrupted! All data scraped up to Chunk %d was safely committed to DB.", chunkIndex) + return ctx.Err() + } + + // 4. Cool-off pause between chunks + if currentEnd.Before(to) { + log.Printf("šŸ˜“ Cool-off period: Pausing before Chunk %d...", chunkIndex+1) + utils.SleepWithJitter(coolOffBase) + } + + currentStart = currentEnd.Add(24 * time.Hour) + chunkIndex++ + } + + log.Printf("\nšŸŽ‰ All Box Score Chunks Finished! Successfully processed %d total games.", totalGamesScraped) + return nil +} + +// processGamesWithWorkers runs the worker pool for a slice of games. +func processGamesWithWorkers(ctx context.Context, games []models.Game) []ScrapedResult { jobs := make(chan models.Game, len(games)) - results := make(chan ScrapedResult, len(games)) + resultsChan := make(chan ScrapedResult, len(games)) var wg sync.WaitGroup // Start worker goroutines for w := 1; w <= numWorkers; w++ { wg.Add(1) - go scrapeAndParseWorker(w, jobs, results, &wg) + go scrapeAndParseWorker(ctx, w, jobs, resultsChan, &wg) } // Send jobs to the workers for _, game := range games { - jobs <- game + select { + case <-ctx.Done(): + break + case jobs <- game: + } } close(jobs) // Wait for all workers to finish wg.Wait() - close(results) + close(resultsChan) - // --- Aggregation & Final Upsert --- - log.Println("All scraping complete. Aggregating results for final batch upsert...") + var validResults []ScrapedResult + for res := range resultsChan { + if res.Err != nil { + log.Printf("āš ļø Worker failed on game %s: %v", res.GameID, res.Err) + continue + } + validResults = append(validResults, res) + } + return validResults +} + +// persistScrapedResults extracts and batch-upserts line scores, player stats, and team stats. +func persistScrapedResults(db *gorm.DB, results []ScrapedResult) error { var allPlayerBasicStats []models.PlayerGameBasicStat var allPlayerAdvStats []models.PlayerGameAdvStat var allTeamBasicStats []models.TeamGameBasicStat var allTeamAdvStats []models.TeamGameAdvStat var allLineScores []models.LineScore - for res := range results { - if res.Err != nil { - log.Printf("A worker failed on game %s: %v", res.GameID, res.Err) - continue - } + for _, res := range results { allPlayerBasicStats = append(allPlayerBasicStats, res.PlayerBasicStats...) allPlayerAdvStats = append(allPlayerAdvStats, res.PlayerAdvStats...) allTeamBasicStats = append(allTeamBasicStats, res.TeamBasicStats...) @@ -97,7 +199,7 @@ func FetchAndStoreBoxScoreDataForDateRange(db *gorm.DB, from, to time.Time) erro allLineScores = append(allLineScores, res.LineScores...) } - // Upsert Line Scores first + // Upsert Line Scores if len(allLineScores) > 0 { if err := db.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "game_id"}, {Name: "team"}}, @@ -105,77 +207,93 @@ func FetchAndStoreBoxScoreDataForDateRange(db *gorm.DB, from, to time.Time) erro }).Create(&allLineScores).Error; err != nil { return fmt.Errorf("failed to upsert line scores: %w", err) } - log.Printf("Successfully upserted %d line scores.", len(allLineScores)) } - // Call the batch upsert function with the fully aggregated data + // Batch upsert Player and Team Stats if err := batchUpsertAll(db, allPlayerBasicStats, allPlayerAdvStats, allTeamBasicStats, allTeamAdvStats); err != nil { - return fmt.Errorf("final batch upsert failed: %w", err) + return fmt.Errorf("batch upsert failed: %w", err) } - log.Printf("Successfully upserted all box score data for %d games.", len(games)) return nil } // scrapeAndParseWorker is a worker goroutine that receives games, scrapes them, and sends back the result. -func scrapeAndParseWorker(id int, jobs <-chan models.Game, results chan<- ScrapedResult, wg *sync.WaitGroup) { +func scrapeAndParseWorker(ctx context.Context, id int, jobs <-chan models.Game, results chan<- ScrapedResult, wg *sync.WaitGroup) { defer wg.Done() - // --- Stagger the start of each worker --- - // Calculate an offset based on the worker's ID to spread out the initial requests. - // We divide the base delay by the number of workers to get an even interval. + // Stagger worker start if numWorkers > 1 { staggerAmount := time.Duration(int64(baseDelay) / int64(numWorkers)) initialDelay := time.Duration(id-1) * staggerAmount log.Printf("Worker %d: Staggering start with an initial delay of %v", id, initialDelay) - time.Sleep(initialDelay) + select { + case <-ctx.Done(): + return + case <-time.After(initialDelay): + } } - for game := range jobs { - log.Printf("šŸ Worker %d: Processing game %s", id, game.GameID) - fullURL := boxScoreURLBase + game.BoxScoreURL + for { + select { + case <-ctx.Done(): + log.Printf("Worker %d: Interrupted, finishing up...", id) + return + case game, ok := <-jobs: + if !ok { + return + } - utils.SleepWithJitter(baseDelay) - time.Sleep(2500 * time.Millisecond) + log.Printf("šŸ Worker %d: Processing game %s", id, game.GameID) + fullURL := boxScoreURLBase + game.BoxScoreURL - req, err := http.NewRequest("GET", fullURL, nil) - if err != nil { - results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("failed to create request: %w", err)} - continue - } - req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36") - resp, err := http.DefaultClient.Do(req) - if err != nil { - results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("request failed: %w", err)} - continue - } + utils.SleepWithJitter(baseDelay) + time.Sleep(2500 * time.Millisecond) - if resp.StatusCode != http.StatusOK { + select { + case <-ctx.Done(): + return + default: + } + + req, err := http.NewRequestWithContext(ctx, "GET", fullURL, nil) + if err != nil { + results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("failed to create request: %w", err)} + continue + } + req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36") + resp, err := http.DefaultClient.Do(req) + if err != nil { + results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("request failed: %w", err)} + continue + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("received non-200 status code: %s", resp.Status)} + continue + } + + doc, err := goquery.NewDocumentFromReader(resp.Body) resp.Body.Close() - results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("received non-200 status code: %s", resp.Status)} - continue - } + if err != nil { + results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("failed to parse document: %w", err)} + continue + } - doc, err := goquery.NewDocumentFromReader(resp.Body) - resp.Body.Close() - if err != nil { - results <- ScrapedResult{GameID: game.GameID, Err: fmt.Errorf("failed to parse document: %w", err)} - continue - } + doc = uncommentDoc(doc) - doc = uncommentDoc(doc) + lineScores := parseLineScore(doc, game.GameID) + pbs, pas, tbs, tas := parseBoxScores(doc, game.GameID) - lineScores := parseLineScore(doc, game.GameID) - pbs, pas, tbs, tas := parseBoxScores(doc, game.GameID) - - results <- ScrapedResult{ - PlayerBasicStats: pbs, - PlayerAdvStats: pas, - TeamBasicStats: tbs, - TeamAdvStats: tas, - LineScores: lineScores, - GameID: game.GameID, - Err: nil, + results <- ScrapedResult{ + PlayerBasicStats: pbs, + PlayerAdvStats: pas, + TeamBasicStats: tbs, + TeamAdvStats: tas, + LineScores: lineScores, + GameID: game.GameID, + Err: nil, + } } } }