`;
}).join("");
}
@@ -196,6 +226,21 @@ function routeStr(f) {
return `${o} → ${d}`;
}
+function categoryBadge(category) {
+ if (!category) return "";
+ const cls = {
+ "Passenger": "cat-pax",
+ "Cargo": "cat-cargo",
+ "Military": "cat-mil",
+ }[category] || "cat-other";
+ const labels = {
+ "Passenger": "P",
+ "Cargo": "C",
+ "Military": "M",
+ };
+ return `
${labels[category] || category}`;
+}
+
function statusBadge(status) {
const map = {
scheduled: "badge-scheduled",
@@ -217,9 +262,10 @@ function statusBadge(status) {
function delayCell(min) {
if (min == null) return "—";
+ if (min > 30) return `
+${min}`;
if (min > 0) return `
+${min}`;
if (min < 0) return `
${min}`;
- return "0";
+ return `
0`;
}
function fmtDate(d) {
@@ -249,7 +295,7 @@ function esc(s) {
function setLoading(on) {
if (on) {
document.getElementById("table-body").innerHTML =
- `
| Загрузка… |
`;
+ `
| Загрузка… |
`;
document.getElementById("cards-container").innerHTML =
`
Загрузка…
`;
}
@@ -257,7 +303,7 @@ function setLoading(on) {
function showError(msg) {
document.getElementById("table-body").innerHTML =
- `
| Ошибка: ${esc(msg)} |
`;
+ `
| Ошибка: ${esc(msg)} |
`;
document.getElementById("cards-container").innerHTML =
`
Ошибка: ${esc(msg)}
`;
}
diff --git a/tasks/flightradar24/ingest/tracks_fr24/config.py b/tasks/flightradar24/ingest/tracks_fr24/config.py
index 7867679..300c614 100644
--- a/tasks/flightradar24/ingest/tracks_fr24/config.py
+++ b/tasks/flightradar24/ingest/tracks_fr24/config.py
@@ -15,14 +15,20 @@ class Config:
FR24_API_KEY: str = os.getenv("FR24_API_KEY", "")
FR24_API_BASE: str = "https://fr24api.flightradar24.com"
- # Airports to track
- AIRPORTS: str = "SVO,DME,VKO,ZIA"
+ # Airports to track (comma-separated IATA codes)
+ AIRPORTS: str = os.getenv("FR24_AIRPORTS", "SVO,DME,VKO,ZIA")
+
+ # Airport direction prefix: "both:" means inbound+outbound
+ AIRPORT_DIRECTION_PREFIX: str = os.getenv("FR24_AIRPORT_DIR_PREFIX", "both:")
# Rate limit: 10 req/min for Explorer tier → 6s between requests
RATE_LIMIT_SEC: float = float(os.getenv("FR24_RATE_LIMIT_SEC", "6.0"))
- # Pagination page size
- PAGE_SIZE: int = 100
+ # Pagination page size (for /full endpoint max 20000)
+ PAGE_SIZE: int = int(os.getenv("FR24_PAGE_SIZE", "20000"))
+
+ # Whether to fetch tracks after summaries (costs extra credits)
+ FETCH_TRACKS: bool = os.getenv("FR24_FETCH_TRACKS", "false").lower() == "true"
@property
def DB_DSN(self) -> str:
diff --git a/tasks/flightradar24/ingest/tracks_fr24/fr24_worker.py b/tasks/flightradar24/ingest/tracks_fr24/fr24_worker.py
index 175264a..a7ca09d 100644
--- a/tasks/flightradar24/ingest/tracks_fr24/fr24_worker.py
+++ b/tasks/flightradar24/ingest/tracks_fr24/fr24_worker.py
@@ -1,16 +1,19 @@
"""
FR24 tracks worker.
-1. GET /api/flight-summary/light for each day → list of fr24_ids
-2. GET /api/flight-tracks?flight_id={fr24_id} → track points
-3. Upsert into fr24_ext.flight_tracks_fr24 + fr24_ext.track_points_fr24
+1. GET /api/flight-summary/full for each day → actual flight data (up to 20000)
+2. Optionally: GET /api/flight-tracks?flight_id={fr24_id} → track points
+3. Upsert into fr24_ext.flight_actual (actual data)
+4. Upsert into fr24_ext.flight_tracks_fr24 + fr24_ext.track_points_fr24 (tracks)
+5. Enrich fr24_ext.schedule with actual times + delays
"""
import logging
+import re
import time
from datetime import date, datetime, timezone
-from typing import Iterator, List, Dict, Optional
+from typing import Dict, Iterator, List, Optional
-import requests
import psycopg2
+import requests
from config import config
@@ -37,7 +40,7 @@ def _throttle():
def _get(path: str, params: dict = None) -> dict:
_throttle()
url = f"{config.FR24_API_BASE}{path}"
- resp = requests.get(url, headers=HEADERS, params=params, timeout=30)
+ resp = requests.get(url, headers=HEADERS, params=params, timeout=60)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 60))
log.warning("Rate limited, sleeping %ds", retry_after)
@@ -47,39 +50,36 @@ def _get(path: str, params: dict = None) -> dict:
return resp.json()
-def iter_flight_summaries(target_date: date) -> Iterator[Dict]:
- """Paginate through flight-summary/light for all 4 airports."""
+def _build_airports_param() -> str:
+ """Build airports param: both:SVO,both:DME,both:VKO,both:ZIA"""
+ prefix = config.AIRPORT_DIRECTION_PREFIX
+ codes = [a.strip() for a in config.AIRPORTS.split(",") if a.strip()]
+ return ",".join(f"{prefix}{code}" for code in codes)
+
+
+def fetch_flight_summaries(target_date: date) -> List[Dict]:
+ """Fetch all flights from flight-summary/full for a single day."""
dt_from = f"{target_date}T00:00:00"
- dt_to = f"{target_date}T23:59:59"
- offset = 0
+ dt_to = f"{target_date}T23:59:59"
+ airports_param = _build_airports_param()
- while True:
- data = _get("/api/flight-summary/light", params={
- "flight_datetime_from": dt_from,
- "flight_datetime_to": dt_to,
- "airports": config.AIRPORTS,
- "limit": config.PAGE_SIZE,
- "offset": offset,
- })
+ data = _get("/api/flight-summary/full", params={
+ "flight_datetime_from": dt_from,
+ "flight_datetime_to": dt_to,
+ "airports": airports_param,
+ "limit": config.PAGE_SIZE,
+ })
- items = data.get("data", data) if isinstance(data, dict) else data
- if not items:
- break
-
- for item in items:
- yield item
-
- # pagination: if fewer items than page size, we're done
- if len(items) < config.PAGE_SIZE:
- break
- offset += len(items)
+ items = data.get("data", data) if isinstance(data, dict) else data
+ if not items:
+ return []
+ return items if isinstance(items, list) else []
def fetch_track(fr24_id: str) -> Optional[List[Dict]]:
"""Fetch track points for a single flight."""
try:
data = _get("/api/flight-tracks", params={"flight_id": fr24_id})
- # response is a list of {fr24_id, tracks: [...]}
if isinstance(data, list) and data:
return data[0].get("tracks", [])
if isinstance(data, dict):
@@ -90,8 +90,70 @@ def fetch_track(fr24_id: str) -> Optional[List[Dict]]:
return None
+# ── DB upsert: flight_actual ─────────────────────────────────────────────────
+
+def upsert_flight_actual(conn, item: Dict, target_date: date) -> Optional[int]:
+ """Insert/update actual flight data. Returns DB id."""
+ fr24_id = item.get("fr24_id")
+ if not fr24_id:
+ return None
+
+ with conn.cursor() as cur:
+ cur.execute(
+ """
+ INSERT INTO fr24_ext.flight_actual
+ (fr24_id, flight, callsign, operated_as, origin_icao, dest_icao,
+ datetime_takeoff, datetime_landed, flight_time,
+ runway_takeoff, runway_landed, actual_distance, category,
+ flight_ended, first_seen, last_seen, flight_date)
+ VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+ ON CONFLICT (fr24_id) DO UPDATE SET
+ flight = EXCLUDED.flight,
+ callsign = EXCLUDED.callsign,
+ operated_as = EXCLUDED.operated_as,
+ origin_icao = EXCLUDED.origin_icao,
+ dest_icao = EXCLUDED.dest_icao,
+ datetime_takeoff = EXCLUDED.datetime_takeoff,
+ datetime_landed = EXCLUDED.datetime_landed,
+ flight_time = EXCLUDED.flight_time,
+ runway_takeoff = EXCLUDED.runway_takeoff,
+ runway_landed = EXCLUDED.runway_landed,
+ actual_distance = EXCLUDED.actual_distance,
+ category = EXCLUDED.category,
+ flight_ended = EXCLUDED.flight_ended,
+ first_seen = EXCLUDED.first_seen,
+ last_seen = EXCLUDED.last_seen,
+ fetched_at = now()
+ RETURNING id
+ """,
+ (
+ fr24_id,
+ item.get("flight"),
+ item.get("callsign"),
+ item.get("operated_as"),
+ item.get("origin_icao"),
+ item.get("destination_icao"),
+ item.get("datetime_takeoff"),
+ item.get("datetime_landed"),
+ item.get("flight_time"),
+ item.get("runway_takeoff"),
+ item.get("runway_landed"),
+ item.get("distance"),
+ item.get("category"),
+ item.get("flight_ended", False),
+ item.get("first_seen"),
+ item.get("last_seen"),
+ target_date,
+ ),
+ )
+ row = cur.fetchone()
+ return row[0] if row else None
+
+
+# ── DB upsert: flight_tracks_fr24 (existing, kept for tracks) ────────────────
+
def upsert_flight(conn, summary: Dict, target_date: date) -> Optional[int]:
- """Insert/update flight header, return DB id."""
+ """Insert/update flight header for tracks. Return DB id."""
with conn.cursor() as cur:
cur.execute(
"""
@@ -117,8 +179,8 @@ def upsert_flight(conn, summary: Dict, target_date: date) -> Optional[int]:
summary.get("callsign"),
summary.get("type"),
summary.get("reg"),
- summary.get("orig_icao"),
- summary.get("dest_icao"),
+ summary.get("origin_icao"),
+ summary.get("destination_icao"),
summary.get("datetime_takeoff"),
summary.get("datetime_landed"),
target_date,
@@ -162,37 +224,136 @@ def upsert_track_points(conn, track_id: int, points: List[Dict]):
)
+# ── Enrich schedule with actual times ────────────────────────────────────────
+
+def _normalize_flight_number(fn: str) -> str:
+ """
+ Normalize flight number for matching.
+ 'SU 1234' → 'SU1234', 'SU1234' → 'SU1234'
+ """
+ if not fn:
+ return ""
+ return re.sub(r"\s+", "", fn.strip().upper())
+
+
+def enrich_schedule(conn, target_date: date) -> int:
+ """
+ Match flight_actual rows to schedule rows by flight number + date.
+ Update schedule with actual times, delays, fr24_id, and category.
+ Returns number of schedule rows updated.
+ """
+ with conn.cursor() as cur:
+ # Match by normalized flight number + flight_date
+ # For departures: match on origin_icao → airport is origin
+ # For arrivals: match on dest_icao → airport is destination
+ cur.execute(
+ """
+ WITH matches AS (
+ SELECT
+ s.schedule_id,
+ fa.fr24_id,
+ fa.datetime_takeoff AS actual_takeoff,
+ fa.datetime_landed AS actual_landed,
+ fa.category AS flight_category,
+ CASE
+ WHEN fa.datetime_takeoff IS NOT NULL AND s.scheduled_at IS NOT NULL
+ THEN EXTRACT(EPOCH FROM (fa.datetime_takeoff - s.scheduled_at))::int / 60
+ END AS delay_takeoff_min,
+ CASE
+ WHEN fa.datetime_landed IS NOT NULL AND s.scheduled_at IS NOT NULL
+ THEN EXTRACT(EPOCH FROM (fa.datetime_landed - s.scheduled_at))::int / 60
+ END AS delay_landed_min
+ FROM fr24_ext.schedule s
+ JOIN fr24_ext.flight_actual fa
+ ON UPPER(REPLACE(fa.flight, ' ', '')) = UPPER(REPLACE(s.flight_number, ' ', ''))
+ AND fa.flight_date = s.flight_date
+ WHERE s.flight_date = %s
+ AND (
+ (s.direction = 'departure' AND fa.origin_icao IN ('UUEE','UUDD','UUWW','UUBW'))
+ OR
+ (s.direction = 'arrival' AND fa.dest_icao IN ('UUEE','UUDD','UUWW','UUBW'))
+ )
+ )
+ UPDATE fr24_ext.schedule s
+ SET
+ actual_takeoff = m.actual_takeoff,
+ actual_landed = m.actual_landed,
+ fr24_id = m.fr24_id,
+ flight_category = m.flight_category,
+ delay_takeoff_min = m.delay_takeoff_min,
+ delay_landed_min = m.delay_landed_min,
+ fetched_at = now()
+ FROM matches m
+ WHERE s.schedule_id = m.schedule_id
+ """,
+ (target_date,),
+ )
+ updated = cur.rowcount
+ return updated
+
+
+# ── Main entry ───────────────────────────────────────────────────────────────
+
def run(target_date: date, conn) -> Dict:
- """Main entry: load all tracks for target_date. Returns stats dict."""
- log.info("FR24 tracks: starting for %s", target_date)
- stats = {"date": str(target_date), "flights_found": 0, "tracks_loaded": 0, "errors": 0}
+ """Main entry: load flight summaries + optionally tracks. Returns stats dict."""
+ log.info("FR24 worker: starting for %s", target_date)
+ stats = {
+ "date": str(target_date),
+ "flights_found": 0,
+ "flights_upserted": 0,
+ "tracks_loaded": 0,
+ "schedule_enriched": 0,
+ "errors": 0,
+ }
- summaries = list(iter_flight_summaries(target_date))
- stats["flights_found"] = len(summaries)
- log.info("FR24 tracks: found %d flights in summary", len(summaries))
+ # 1. Fetch flight summaries from /full endpoint
+ try:
+ summaries = fetch_flight_summaries(target_date)
+ stats["flights_found"] = len(summaries)
+ log.info("FR24 worker: found %d flights", len(summaries))
+ except Exception as e:
+ log.error("FR24 worker: failed to fetch summaries: %s", e)
+ stats["errors"] += 1
+ return stats
- for summary in summaries:
- fr24_id = summary.get("fr24_id")
+ # 2. Upsert into flight_actual + optionally fetch tracks
+ for item in summaries:
+ fr24_id = item.get("fr24_id")
if not fr24_id:
continue
try:
- track_id = upsert_flight(conn, summary, target_date)
- if track_id is None:
- continue
+ actual_id = upsert_flight_actual(conn, item, target_date)
+ if actual_id:
+ stats["flights_upserted"] += 1
- points = fetch_track(fr24_id)
- if points is None:
- stats["errors"] += 1
- continue
+ # Optionally fetch tracks (costs extra credits)
+ if config.FETCH_TRACKS:
+ track_id = upsert_flight(conn, item, target_date)
+ if track_id:
+ points = fetch_track(fr24_id)
+ if points is not None:
+ upsert_track_points(conn, track_id, points)
+ stats["tracks_loaded"] += 1
+ else:
+ stats["errors"] += 1
- upsert_track_points(conn, track_id, points)
conn.commit()
- stats["tracks_loaded"] += 1
- log.debug("FR24: %s → %d points", fr24_id, len(points))
+ log.debug("FR24: %s upserted", fr24_id)
except Exception as e:
conn.rollback()
stats["errors"] += 1
log.error("FR24: error processing %s: %s", fr24_id, e)
- log.info("FR24 tracks done: %s", stats)
+ # 3. Enrich schedule with actual times
+ try:
+ enriched = enrich_schedule(conn, target_date)
+ conn.commit()
+ stats["schedule_enriched"] = enriched
+ log.info("FR24 worker: enriched %d schedule rows", enriched)
+ except Exception as e:
+ conn.rollback()
+ log.error("FR24 worker: schedule enrichment failed: %s", e)
+ stats["errors"] += 1
+
+ log.info("FR24 worker done: %s", stats)
return stats