АДРЕСНЫЙ РЕЖИМ - ллм декомпоз
This commit is contained in:
@@ -0,0 +1,13 @@
|
||||
FROM python:3.11-slim
|
||||
|
||||
ENV PYTHONDONTWRITEBYTECODE=1
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
COPY requirements.txt ./
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY worker.py ./
|
||||
|
||||
CMD ["python", "worker.py"]
|
||||
@@ -0,0 +1,2 @@
|
||||
psycopg2-binary==2.9.9
|
||||
requests==2.32.3
|
||||
@@ -0,0 +1,446 @@
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import psycopg2
|
||||
from psycopg2.extras import Json
|
||||
import requests
|
||||
|
||||
|
||||
DATABASE_URL = os.environ.get("DATABASE_URL")
|
||||
WAQI_TOKEN = os.environ.get("WAQI_TOKEN")
|
||||
WAQI_BASE_URL = os.environ.get("WAQI_BASE_URL", "https://api.waqi.info")
|
||||
|
||||
|
||||
def get_int_env(name, default):
|
||||
value = os.environ.get(name)
|
||||
if value is None or value == "":
|
||||
return default
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError as exc:
|
||||
raise RuntimeError(f"Invalid int for {name}: {value}") from exc
|
||||
|
||||
|
||||
DELAY_MINUTES = get_int_env("DETECT_DELAY_MINUTES", 60)
|
||||
STUCK_MINUTES = get_int_env("DETECT_STUCK_MINUTES", 120)
|
||||
SPIKE_DELTA = get_int_env("DETECT_SPIKE_DELTA", 20)
|
||||
SPIKE_WINDOW_MINUTES = get_int_env("DETECT_SPIKE_WINDOW_MINUTES", 60)
|
||||
DEDUP_MINUTES = get_int_env("DETECT_DEDUP_MINUTES", 60)
|
||||
|
||||
|
||||
def run_query():
|
||||
if not DATABASE_URL:
|
||||
raise RuntimeError("DATABASE_URL is not set")
|
||||
with psycopg2.connect(DATABASE_URL) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT 1;")
|
||||
cur.fetchone()
|
||||
|
||||
|
||||
def wait_for_db():
|
||||
while True:
|
||||
try:
|
||||
run_query()
|
||||
print("db ready", flush=True)
|
||||
return
|
||||
except Exception as exc:
|
||||
print(f"db not ready: {exc}", flush=True)
|
||||
time.sleep(5)
|
||||
|
||||
|
||||
def get_env_float(name):
|
||||
value = os.environ.get(name)
|
||||
if value is None or value == "":
|
||||
raise RuntimeError(f"{name} is not set")
|
||||
return float(value)
|
||||
|
||||
|
||||
def parse_tz_offset(tz_str):
|
||||
if not tz_str:
|
||||
return timezone.utc
|
||||
if tz_str in ("UTC", "GMT", "Z"):
|
||||
return timezone.utc
|
||||
if tz_str.startswith(("+", "-")):
|
||||
sign = 1 if tz_str[0] == "+" else -1
|
||||
parts = tz_str[1:].split(":")
|
||||
try:
|
||||
hours = int(parts[0])
|
||||
minutes = int(parts[1]) if len(parts) > 1 else 0
|
||||
except ValueError:
|
||||
return timezone.utc
|
||||
return timezone(sign * timedelta(hours=hours, minutes=minutes))
|
||||
return timezone.utc
|
||||
|
||||
|
||||
def parse_time_string(stime, tzinfo):
|
||||
if not stime:
|
||||
return None
|
||||
try:
|
||||
if "T" not in stime and " " in stime:
|
||||
stime = stime.replace(" ", "T")
|
||||
dt = datetime.fromisoformat(stime)
|
||||
except ValueError:
|
||||
try:
|
||||
dt = datetime.strptime(stime, "%Y-%m-%d %H:%M:%S")
|
||||
except ValueError:
|
||||
return None
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=tzinfo)
|
||||
return dt.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def dt_iso(dt):
|
||||
if dt is None:
|
||||
return None
|
||||
return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def parse_observed_ts(station_time):
|
||||
tzinfo = timezone.utc
|
||||
stime = None
|
||||
if isinstance(station_time, dict):
|
||||
tzinfo = parse_tz_offset(station_time.get("tz"))
|
||||
stime = (
|
||||
station_time.get("stime")
|
||||
or station_time.get("s")
|
||||
or station_time.get("vtime")
|
||||
)
|
||||
elif isinstance(station_time, str):
|
||||
stime = station_time
|
||||
parsed = parse_time_string(stime, tzinfo)
|
||||
return parsed or datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def parse_aqi(value):
|
||||
if value is None or value == "-":
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def build_waqi_url():
|
||||
if not WAQI_TOKEN:
|
||||
raise RuntimeError("WAQI_TOKEN is not set")
|
||||
lat1 = get_env_float("BBOX_LAT1")
|
||||
lon1 = get_env_float("BBOX_LON1")
|
||||
lat2 = get_env_float("BBOX_LAT2")
|
||||
lon2 = get_env_float("BBOX_LON2")
|
||||
return (
|
||||
f"{WAQI_BASE_URL}/map/bounds/?latlng="
|
||||
f"{lat1},{lon1},{lat2},{lon2}&token={WAQI_TOKEN}"
|
||||
)
|
||||
|
||||
|
||||
def fetch_waqi():
|
||||
url = build_waqi_url()
|
||||
last_error = None
|
||||
for attempt in range(3):
|
||||
try:
|
||||
response = requests.get(url, timeout=20)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
if payload.get("status") != "ok":
|
||||
raise RuntimeError(
|
||||
f"waqi status {payload.get('status')}: {payload.get('data')}"
|
||||
)
|
||||
return payload.get("data", [])
|
||||
except Exception as exc:
|
||||
last_error = exc
|
||||
time.sleep(2 * (attempt + 1))
|
||||
raise RuntimeError(f"WAQI fetch failed: {last_error}")
|
||||
|
||||
|
||||
def upsert_station(cur, provider_uid, name, lat, lon, source_meta):
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO stations (provider_uid, name, lat, lon, source_meta)
|
||||
VALUES (%s, %s, %s, %s, %s)
|
||||
ON CONFLICT (provider_uid) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
lat = EXCLUDED.lat,
|
||||
lon = EXCLUDED.lon,
|
||||
source_meta = EXCLUDED.source_meta
|
||||
RETURNING id;
|
||||
""",
|
||||
(provider_uid, name, lat, lon, Json(source_meta)),
|
||||
)
|
||||
return cur.fetchone()[0]
|
||||
|
||||
|
||||
def insert_measurement(cur, station_id, observed_ts, ingested_ts, aqi, raw):
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO measurements (station_id, observed_ts, ingested_ts, aqi, raw)
|
||||
VALUES (%s, %s, %s, %s, %s)
|
||||
ON CONFLICT (station_id, observed_ts) DO NOTHING;
|
||||
""",
|
||||
(station_id, observed_ts, ingested_ts, aqi, Json(raw)),
|
||||
)
|
||||
return cur.rowcount
|
||||
|
||||
|
||||
def anomaly_exists(cur, station_id, anomaly_type, metric, since_ts):
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT 1
|
||||
FROM anomalies
|
||||
WHERE station_id = %s AND type = %s AND metric = %s AND created_at >= %s
|
||||
LIMIT 1;
|
||||
""",
|
||||
(station_id, anomaly_type, metric, since_ts),
|
||||
)
|
||||
return cur.fetchone() is not None
|
||||
|
||||
|
||||
def insert_anomaly(
|
||||
cur,
|
||||
station_id,
|
||||
anomaly_type,
|
||||
severity,
|
||||
confidence,
|
||||
ts_start,
|
||||
ts_end,
|
||||
metric,
|
||||
evidence,
|
||||
):
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO anomalies (
|
||||
station_id, type, severity, confidence, ts_start, ts_end, metric, evidence
|
||||
)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s);
|
||||
""",
|
||||
(
|
||||
station_id,
|
||||
anomaly_type,
|
||||
severity,
|
||||
confidence,
|
||||
ts_start,
|
||||
ts_end,
|
||||
metric,
|
||||
Json(evidence),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def detect_delay(cur, now):
|
||||
delay_threshold = timedelta(minutes=DELAY_MINUTES)
|
||||
dedup_since = now - timedelta(minutes=DEDUP_MINUTES)
|
||||
inserted = 0
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT s.id, MAX(m.observed_ts) AS last_ts
|
||||
FROM stations s
|
||||
LEFT JOIN measurements m ON m.station_id = s.id
|
||||
GROUP BY s.id;
|
||||
"""
|
||||
)
|
||||
for station_id, last_ts in cur.fetchall():
|
||||
if last_ts is None:
|
||||
continue
|
||||
age = now - last_ts
|
||||
if age <= delay_threshold:
|
||||
continue
|
||||
if anomaly_exists(cur, station_id, "delay", "delay", dedup_since):
|
||||
continue
|
||||
age_minutes = int(age.total_seconds() / 60)
|
||||
severity = min(100, int((age_minutes / DELAY_MINUTES) * 50) + 50)
|
||||
evidence = {
|
||||
"last_observed_ts": dt_iso(last_ts),
|
||||
"age_minutes": age_minutes,
|
||||
"threshold_minutes": DELAY_MINUTES,
|
||||
}
|
||||
insert_anomaly(
|
||||
cur,
|
||||
station_id,
|
||||
"delay",
|
||||
severity,
|
||||
0.8,
|
||||
last_ts,
|
||||
now,
|
||||
"delay",
|
||||
evidence,
|
||||
)
|
||||
inserted += 1
|
||||
return inserted
|
||||
|
||||
|
||||
def detect_stuck(cur, now):
|
||||
window = timedelta(minutes=STUCK_MINUTES)
|
||||
window_start = now - window
|
||||
dedup_since = now - timedelta(minutes=DEDUP_MINUTES)
|
||||
inserted = 0
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT station_id,
|
||||
MIN(aqi) AS min_aqi,
|
||||
MAX(aqi) AS max_aqi,
|
||||
COUNT(*) AS count_aqi,
|
||||
MIN(observed_ts) AS min_ts,
|
||||
MAX(observed_ts) AS max_ts
|
||||
FROM measurements
|
||||
WHERE observed_ts >= %s AND observed_ts <= %s AND aqi IS NOT NULL
|
||||
GROUP BY station_id;
|
||||
""",
|
||||
(window_start, now),
|
||||
)
|
||||
min_span_seconds = max(0, (STUCK_MINUTES - 5) * 60)
|
||||
for station_id, min_aqi, max_aqi, count_aqi, min_ts, max_ts in cur.fetchall():
|
||||
if count_aqi < 3:
|
||||
continue
|
||||
if min_aqi != max_aqi:
|
||||
continue
|
||||
if max_ts is None or min_ts is None:
|
||||
continue
|
||||
span_seconds = (max_ts - min_ts).total_seconds()
|
||||
if span_seconds < min_span_seconds:
|
||||
continue
|
||||
if anomaly_exists(cur, station_id, "stuck", "aqi", dedup_since):
|
||||
continue
|
||||
evidence = {
|
||||
"value": int(min_aqi),
|
||||
"window_minutes": STUCK_MINUTES,
|
||||
"first_ts": dt_iso(min_ts),
|
||||
"last_ts": dt_iso(max_ts),
|
||||
"count": int(count_aqi),
|
||||
}
|
||||
insert_anomaly(
|
||||
cur,
|
||||
station_id,
|
||||
"stuck",
|
||||
60,
|
||||
0.7,
|
||||
min_ts,
|
||||
max_ts,
|
||||
"aqi",
|
||||
evidence,
|
||||
)
|
||||
inserted += 1
|
||||
return inserted
|
||||
|
||||
|
||||
def detect_spike(cur, now):
|
||||
window = timedelta(minutes=SPIKE_WINDOW_MINUTES)
|
||||
dedup_since = now - timedelta(minutes=DEDUP_MINUTES)
|
||||
inserted = 0
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT DISTINCT ON (station_id) station_id, observed_ts, aqi
|
||||
FROM measurements
|
||||
WHERE aqi IS NOT NULL
|
||||
ORDER BY station_id, observed_ts DESC;
|
||||
"""
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
for station_id, latest_ts, latest_aqi in rows:
|
||||
target_ts = latest_ts - window
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT observed_ts, aqi
|
||||
FROM measurements
|
||||
WHERE station_id = %s AND aqi IS NOT NULL AND observed_ts <= %s
|
||||
ORDER BY observed_ts DESC
|
||||
LIMIT 1;
|
||||
""",
|
||||
(station_id, target_ts),
|
||||
)
|
||||
prev = cur.fetchone()
|
||||
if not prev:
|
||||
continue
|
||||
prev_ts, prev_aqi = prev
|
||||
delta = latest_aqi - prev_aqi
|
||||
if delta < SPIKE_DELTA:
|
||||
continue
|
||||
if anomaly_exists(cur, station_id, "spike", "aqi", dedup_since):
|
||||
continue
|
||||
severity = min(100, int((delta / SPIKE_DELTA) * 50) + 50)
|
||||
evidence = {
|
||||
"delta": int(delta),
|
||||
"prev_aqi": int(prev_aqi),
|
||||
"prev_ts": dt_iso(prev_ts),
|
||||
"curr_aqi": int(latest_aqi),
|
||||
"curr_ts": dt_iso(latest_ts),
|
||||
"window_minutes": SPIKE_WINDOW_MINUTES,
|
||||
}
|
||||
insert_anomaly(
|
||||
cur,
|
||||
station_id,
|
||||
"spike",
|
||||
severity,
|
||||
0.9,
|
||||
prev_ts,
|
||||
latest_ts,
|
||||
"aqi",
|
||||
evidence,
|
||||
)
|
||||
inserted += 1
|
||||
return inserted
|
||||
|
||||
|
||||
def detect_anomalies():
|
||||
now = datetime.now(timezone.utc)
|
||||
inserted = 0
|
||||
with psycopg2.connect(DATABASE_URL) as conn:
|
||||
with conn.cursor() as cur:
|
||||
# delay anomalies disabled; "Устаревшие" handled in Overview
|
||||
inserted += detect_stuck(cur, now)
|
||||
inserted += detect_spike(cur, now)
|
||||
print(f"anomalies inserted={inserted}", flush=True)
|
||||
|
||||
|
||||
def ingest_once():
|
||||
data = fetch_waqi()
|
||||
if not data:
|
||||
print("waqi empty", flush=True)
|
||||
return
|
||||
ingested_ts = datetime.now(timezone.utc)
|
||||
inserted_measurements = 0
|
||||
seen_stations = {}
|
||||
with psycopg2.connect(DATABASE_URL) as conn:
|
||||
with conn.cursor() as cur:
|
||||
for item in data:
|
||||
provider_uid = item.get("uid")
|
||||
lat = item.get("lat")
|
||||
lon = item.get("lon")
|
||||
if provider_uid is None or lat is None or lon is None:
|
||||
continue
|
||||
station = item.get("station") or {}
|
||||
name = station.get("name")
|
||||
source_meta = station
|
||||
if provider_uid in seen_stations:
|
||||
station_id = seen_stations[provider_uid]
|
||||
else:
|
||||
station_id = upsert_station(
|
||||
cur, provider_uid, name, lat, lon, source_meta
|
||||
)
|
||||
seen_stations[provider_uid] = station_id
|
||||
station_time = station.get("time") or item.get("time")
|
||||
observed_ts = parse_observed_ts(station_time)
|
||||
aqi = parse_aqi(item.get("aqi"))
|
||||
inserted_measurements += insert_measurement(
|
||||
cur, station_id, observed_ts, ingested_ts, aqi, item
|
||||
)
|
||||
print(
|
||||
f"waqi ingest: stations={len(seen_stations)} "
|
||||
f"measurements_inserted={inserted_measurements}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
wait_for_db()
|
||||
poll_seconds = int(os.environ.get("POLL_SECONDS", "300"))
|
||||
while True:
|
||||
try:
|
||||
ingest_once()
|
||||
detect_anomalies()
|
||||
except Exception as exc:
|
||||
print(f"ingest failed: {exc}", flush=True)
|
||||
time.sleep(poll_seconds)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user