auto-sync: 2026-04-21 18:50:01
This commit is contained in:
45
tasks/flightradar24/db/migrations/007_flight_actual.sql
Normal file
45
tasks/flightradar24/db/migrations/007_flight_actual.sql
Normal file
@@ -0,0 +1,45 @@
|
||||
-- Migration 007: flight_actual table + schedule enrichment columns
|
||||
|
||||
-- ── New table: actual flight data from FR24 flight-summary/full ────────────
|
||||
|
||||
CREATE TABLE IF NOT EXISTS fr24_ext.flight_actual (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
fr24_id VARCHAR(30) NOT NULL UNIQUE,
|
||||
flight VARCHAR(20),
|
||||
callsign VARCHAR(20),
|
||||
operated_as VARCHAR(5), -- ICAO airline code
|
||||
origin_icao VARCHAR(5),
|
||||
dest_icao VARCHAR(5),
|
||||
datetime_takeoff TIMESTAMPTZ,
|
||||
datetime_landed TIMESTAMPTZ,
|
||||
flight_time INTEGER, -- seconds
|
||||
runway_takeoff VARCHAR(10),
|
||||
runway_landed VARCHAR(10),
|
||||
actual_distance FLOAT, -- km
|
||||
category VARCHAR(20), -- Passenger/Cargo/Military
|
||||
flight_ended BOOLEAN DEFAULT false,
|
||||
first_seen TIMESTAMPTZ,
|
||||
last_seen TIMESTAMPTZ,
|
||||
flight_date DATE NOT NULL,
|
||||
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_flight_actual_date ON fr24_ext.flight_actual (flight_date);
|
||||
CREATE INDEX IF NOT EXISTS idx_flight_actual_flight ON fr24_ext.flight_actual (flight, flight_date);
|
||||
CREATE INDEX IF NOT EXISTS idx_flight_actual_orig ON fr24_ext.flight_actual (origin_icao, flight_date);
|
||||
CREATE INDEX IF NOT EXISTS idx_flight_actual_dest ON fr24_ext.flight_actual (dest_icao, flight_date);
|
||||
CREATE INDEX IF NOT EXISTS idx_flight_actual_category ON fr24_ext.flight_actual (category);
|
||||
|
||||
-- ── Enrich schedule table with actual times + delays ────────────────────────
|
||||
|
||||
ALTER TABLE fr24_ext.schedule
|
||||
ADD COLUMN IF NOT EXISTS actual_takeoff TIMESTAMPTZ,
|
||||
ADD COLUMN IF NOT EXISTS actual_landed TIMESTAMPTZ,
|
||||
ADD COLUMN IF NOT EXISTS delay_takeoff_min INTEGER,
|
||||
ADD COLUMN IF NOT EXISTS delay_landed_min INTEGER,
|
||||
ADD COLUMN IF NOT EXISTS fr24_id VARCHAR(30),
|
||||
ADD COLUMN IF NOT EXISTS flight_category VARCHAR(20);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_schedule_fr24_id ON fr24_ext.schedule (fr24_id);
|
||||
|
||||
COMMENT ON TABLE fr24_ext.flight_actual IS 'Actual flight data from FR24 flight-summary/full endpoint';
|
||||
@@ -468,7 +468,10 @@ def schedule_data():
|
||||
flight_number, airline_name, airport_iata, direction,
|
||||
origin_iata, destination_iata,
|
||||
scheduled_at, actual_at, status, icao24,
|
||||
flight_date, duration_min, thread_title
|
||||
flight_date, duration_min, thread_title,
|
||||
actual_takeoff, actual_landed,
|
||||
delay_takeoff_min, delay_landed_min,
|
||||
fr24_id, flight_category
|
||||
FROM fr24_ext.schedule
|
||||
WHERE {where}
|
||||
ORDER BY scheduled_at DESC
|
||||
@@ -485,6 +488,14 @@ def schedule_data():
|
||||
int((actual - sched).total_seconds() / 60)
|
||||
if actual and sched else None
|
||||
)
|
||||
# Actual times from FR24 flight-summary/full enrichment
|
||||
actual_takeoff = r.get("actual_takeoff")
|
||||
actual_landed = r.get("actual_landed")
|
||||
delay_takeoff = r.get("delay_takeoff_min")
|
||||
delay_landed = r.get("delay_landed_min")
|
||||
fr24_id = r.get("fr24_id")
|
||||
flight_cat = r.get("flight_category")
|
||||
|
||||
flights.append({
|
||||
"flight_number": r["flight_number"],
|
||||
"airline": r["airline_name"],
|
||||
@@ -501,6 +512,13 @@ def schedule_data():
|
||||
"duration_min": r["duration_min"],
|
||||
"status": r["status"],
|
||||
"icao24": r["icao24"],
|
||||
# New fields from FR24 enrichment
|
||||
"actual_takeoff": actual_takeoff.isoformat() if hasattr(actual_takeoff, 'isoformat') else None,
|
||||
"actual_landed": actual_landed.isoformat() if hasattr(actual_landed, 'isoformat') else None,
|
||||
"delay_takeoff_min": delay_takeoff,
|
||||
"delay_landed_min": delay_landed,
|
||||
"fr24_id": fr24_id,
|
||||
"flight_category": flight_cat,
|
||||
})
|
||||
|
||||
return ok({"total": total, "flights": flights})
|
||||
@@ -519,7 +537,9 @@ def schedule_export():
|
||||
SELECT
|
||||
flight_date, flight_number, airline_name, airport_iata, direction,
|
||||
origin_iata, destination_iata,
|
||||
scheduled_at, actual_at, status, icao24, duration_min
|
||||
scheduled_at, actual_at, status, icao24, duration_min,
|
||||
actual_takeoff, actual_landed, delay_takeoff_min, delay_landed_min,
|
||||
fr24_id, flight_category
|
||||
FROM fr24_ext.schedule
|
||||
WHERE {where}
|
||||
ORDER BY scheduled_at DESC
|
||||
@@ -534,6 +554,8 @@ def schedule_export():
|
||||
"Date", "Flight", "Airline", "Airport", "Direction",
|
||||
"Origin", "Destination", "Scheduled", "Actual",
|
||||
"Delay (min)", "Duration (min)", "Status", "ICAO24",
|
||||
"Actual Takeoff", "Actual Landed", "Delay Takeoff (min)", "Delay Landed (min)",
|
||||
"FR24 ID", "Category",
|
||||
])
|
||||
for r in rows:
|
||||
sched = r["scheduled_at"]
|
||||
@@ -556,6 +578,12 @@ def schedule_export():
|
||||
r["duration_min"] or "",
|
||||
r["status"] or "",
|
||||
r["icao24"] or "",
|
||||
r["actual_takeoff"].isoformat() if hasattr(r.get("actual_takeoff"), 'isoformat') else "",
|
||||
r["actual_landed"].isoformat() if hasattr(r.get("actual_landed"), 'isoformat') else "",
|
||||
r["delay_takeoff_min"] if r.get("delay_takeoff_min") is not None else "",
|
||||
r["delay_landed_min"] if r.get("delay_landed_min") is not None else "",
|
||||
r["fr24_id"] or "",
|
||||
r["flight_category"] or "",
|
||||
])
|
||||
|
||||
buf.seek(0)
|
||||
|
||||
@@ -161,6 +161,34 @@
|
||||
|
||||
.delay-pos { color: #d29922; }
|
||||
.delay-neg { color: #3fb950; }
|
||||
.delay-ok { color: #3fb950; }
|
||||
.delay-critical { color: #f85149; font-weight: 700; }
|
||||
|
||||
/* category badges */
|
||||
.cat-badge {
|
||||
border-radius: 3px;
|
||||
display: inline-block;
|
||||
font-size: 10px;
|
||||
font-weight: 700;
|
||||
padding: 1px 5px;
|
||||
margin-left: 4px;
|
||||
}
|
||||
.cat-pax { background: #0d4429; color: #3fb950; }
|
||||
.cat-cargo { background: #3d2b00; color: #d29922; }
|
||||
.cat-mil { background: #3d0c0c; color: #f85149; }
|
||||
.cat-other { background: #21262d; color: #8b949e; }
|
||||
|
||||
/* track link */
|
||||
.track-link {
|
||||
color: #58a6ff;
|
||||
text-decoration: none;
|
||||
font-size: 13px;
|
||||
}
|
||||
.track-link:hover { text-decoration: underline; }
|
||||
|
||||
/* actual time display */
|
||||
.act-time { color: #c9d1d9; }
|
||||
.act-landed { color: #8b949e; font-size: 11px; }
|
||||
|
||||
/* ── pagination ── */
|
||||
.pagination {
|
||||
@@ -300,11 +328,13 @@
|
||||
<th>Запланировано</th>
|
||||
<th>Фактически</th>
|
||||
<th>Задержка</th>
|
||||
<th>Тип</th>
|
||||
<th>Статус</th>
|
||||
<th>Трек</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody id="table-body">
|
||||
<tr><td colspan="10" class="state-msg">Загрузка…</td></tr>
|
||||
<tr><td colspan="12" class="state-msg">Загрузка…</td></tr>
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
@@ -105,7 +105,7 @@ async function loadData() {
|
||||
function renderTable(flights) {
|
||||
const tbody = document.getElementById("table-body");
|
||||
if (!flights.length) {
|
||||
tbody.innerHTML = `<tr><td colspan="10" class="state-msg">Нет данных по выбранным фильтрам</td></tr>`;
|
||||
tbody.innerHTML = `<tr><td colspan="12" class="state-msg">Нет данных по выбранным фильтрам</td></tr>`;
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -121,17 +121,39 @@ function renderTable(flights) {
|
||||
const badge = statusBadge(f.status);
|
||||
const dateStr = fmtDateShort(f.scheduled_at);
|
||||
|
||||
// Actual times from FR24 enrichment
|
||||
const actTakeoff = f.actual_takeoff ? fmtTime(f.actual_takeoff) : "";
|
||||
const actLanded = f.actual_landed ? fmtTime(f.actual_landed) : "";
|
||||
const delayTakeoff = delayCell(f.delay_takeoff_min);
|
||||
const delayLanded = delayCell(f.delay_landed_min);
|
||||
|
||||
// Category badge
|
||||
const catBadge = categoryBadge(f.flight_category);
|
||||
|
||||
// FR24 track link
|
||||
const trackLink = f.fr24_id
|
||||
? `<a href="https://www.flightradar24.com/data/flights/${f.fr24_id}" target="_blank" class="track-link" title="Трек FR24">✈</a>`
|
||||
: "";
|
||||
|
||||
// Enriched actual time display: show actual takeoff/landed if available
|
||||
const actualDisplay = actTakeoff || actLanded
|
||||
? `<span class="act-time">${actTakeoff || "—"}</span>` +
|
||||
(f.direction === "arrival" && actLanded ? ` <span class="act-landed">⇣${actLanded}</span>` : "")
|
||||
: actual;
|
||||
|
||||
return `<tr>
|
||||
<td>${dateStr}</td>
|
||||
<td><strong>${esc(f.flight_number)}</strong></td>
|
||||
<td><strong>${esc(f.flight_number)}</strong> ${trackLink}</td>
|
||||
<td>${esc(f.airline || "—")}</td>
|
||||
<td>${esc(f.airport)}</td>
|
||||
<td>${dirIcon}</td>
|
||||
<td>${esc(route)}</td>
|
||||
<td>${sched}</td>
|
||||
<td>${actual}</td>
|
||||
<td>${delay}</td>
|
||||
<td>${actualDisplay}</td>
|
||||
<td>${delayTakeoff}</td>
|
||||
<td>${catBadge}</td>
|
||||
<td>${badge}</td>
|
||||
<td>${trackLink}</td>
|
||||
</tr>`;
|
||||
}).join("");
|
||||
}
|
||||
@@ -153,11 +175,17 @@ function renderCards(flights) {
|
||||
const actual = f.actual_at ? fmtTime(f.actual_at) : "—";
|
||||
const delay = f.delay_min != null ? `${f.delay_min > 0 ? "+" : ""}${f.delay_min} мин` : "—";
|
||||
const badge = statusBadge(f.status);
|
||||
const catBadge = categoryBadge(f.flight_category);
|
||||
const trackLink = f.fr24_id
|
||||
? `<a href="https://www.flightradar24.com/data/flights/${f.fr24_id}" target="_blank" style="color:#58a6ff">Трек</a>`
|
||||
: "";
|
||||
const actTakeoff = f.actual_takeoff ? fmtTime(f.actual_takeoff) : "";
|
||||
const actLanded = f.actual_landed ? fmtTime(f.actual_landed) : "";
|
||||
|
||||
return `<div class="card">
|
||||
<div class="card-header">
|
||||
<div>
|
||||
<div class="card-flight">${esc(f.flight_number)}</div>
|
||||
<div class="card-flight">${esc(f.flight_number)} ${catBadge} ${trackLink}</div>
|
||||
<div class="card-airline">${esc(f.airline || "—")}</div>
|
||||
</div>
|
||||
${badge}
|
||||
@@ -168,6 +196,8 @@ function renderCards(flights) {
|
||||
<div class="card-row"><span>Запланировано</span><span>${sched}</span></div>
|
||||
<div class="card-row"><span>Фактически</span><span>${actual}</span></div>
|
||||
<div class="card-row"><span>Задержка</span><span>${delay}</span></div>
|
||||
${actTakeoff ? `<div class="card-row"><span>Взлёт факт</span><span>${actTakeoff} ${delayCell(f.delay_takeoff_min)}</span></div>` : ""}
|
||||
${actLanded ? `<div class="card-row"><span>Посадка факт</span><span>${actLanded} ${delayCell(f.delay_landed_min)}</span></div>` : ""}
|
||||
</div>`;
|
||||
}).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 `<span class="cat-badge ${cls}" title="${category}">${labels[category] || category}</span>`;
|
||||
}
|
||||
|
||||
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 `<span class="delay-critical">+${min}</span>`;
|
||||
if (min > 0) return `<span class="delay-pos">+${min}</span>`;
|
||||
if (min < 0) return `<span class="delay-neg">${min}</span>`;
|
||||
return "0";
|
||||
return `<span class="delay-ok">0</span>`;
|
||||
}
|
||||
|
||||
function fmtDate(d) {
|
||||
@@ -249,7 +295,7 @@ function esc(s) {
|
||||
function setLoading(on) {
|
||||
if (on) {
|
||||
document.getElementById("table-body").innerHTML =
|
||||
`<tr><td colspan="10" class="state-msg">Загрузка…</td></tr>`;
|
||||
`<tr><td colspan="12" class="state-msg">Загрузка…</td></tr>`;
|
||||
document.getElementById("cards-container").innerHTML =
|
||||
`<div class="state-msg">Загрузка…</div>`;
|
||||
}
|
||||
@@ -257,7 +303,7 @@ function setLoading(on) {
|
||||
|
||||
function showError(msg) {
|
||||
document.getElementById("table-body").innerHTML =
|
||||
`<tr><td colspan="10" class="state-msg" style="color:#f85149">Ошибка: ${esc(msg)}</td></tr>`;
|
||||
`<tr><td colspan="12" class="state-msg" style="color:#f85149">Ошибка: ${esc(msg)}</td></tr>`;
|
||||
document.getElementById("cards-container").innerHTML =
|
||||
`<div class="state-msg" style="color:#f85149">Ошибка: ${esc(msg)}</div>`;
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user