| |
| from __future__ import annotations |
|
|
| import os |
| import sys |
| from datetime import datetime, timezone |
|
|
| import requests |
|
|
| OPENF1_BASE_URL = os.getenv("OPENF1_BASE_URL", "https://api.openf1.org/v1") |
| INGEST_URL = os.getenv("INGEST_URL") |
|
|
|
|
| def is_race_day(date_iso: str) -> bool: |
| params = {"date_start>=": date_iso, "date_end<=": date_iso} |
| response = requests.get(f"{OPENF1_BASE_URL}/sessions", params=params, timeout=30) |
| response.raise_for_status() |
| sessions = response.json() |
| return any( |
| (session.get("session_type") == "Race" or session.get("session_name") == "Race") |
| for session in sessions |
| ) |
|
|
|
|
| def has_practice_sessions(date_iso: str) -> bool: |
| params = {"date_start>=": date_iso, "date_end<=": date_iso} |
| response = requests.get(f"{OPENF1_BASE_URL}/sessions", params=params, timeout=30) |
| response.raise_for_status() |
| sessions = response.json() |
| return any( |
| "practice" in (session.get("session_type") or session.get("session_name") or "").lower() |
| for session in sessions |
| ) |
|
|
|
|
| def trigger_ingest(date_iso: str) -> None: |
| if not INGEST_URL: |
| raise RuntimeError("INGEST_URL is not set") |
| response = requests.post(INGEST_URL, params={"date": date_iso}, timeout=60) |
| response.raise_for_status() |
| print("Ingest triggered:", response.json()) |
|
|
|
|
| def main() -> int: |
| date_iso = os.getenv("RACE_DATE") |
| if not date_iso: |
| date_iso = datetime.now(timezone.utc).date().isoformat() |
|
|
| if has_practice_sessions(date_iso): |
| print(f"Practice sessions detected for {date_iso}. Triggering practice ingest.") |
| trigger_ingest(date_iso) |
| return 0 |
|
|
| if is_race_day(date_iso): |
| print(f"Race day detected for {date_iso}. Triggering ingest.") |
| trigger_ingest(date_iso) |
| else: |
| print(f"No race session found for {date_iso}. Skipping.") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
