initial commit

worker/worker.py

@@ -0,0 +1,83 @@
+import asyncio
+import os
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from urllib.parse import urljoin
+
+import httpx
+
+DATA_DIR = Path(os.getenv("DATA_DIR", "/data"))
+RAW_RADAR_DIR = DATA_DIR / "raw" / "radar"
+DWD_RADAR_BASE_URL = os.getenv("DWD_RADAR_BASE_URL", "https://opendata.dwd.de/weather/radar/composite").rstrip("/")
+RADAR_PRODUCT = os.getenv("RADAR_PRODUCT", "rv")
+RADAR_SYNC_LIMIT = int(os.getenv("RADAR_SYNC_LIMIT", "12"))
+RADAR_SYNC_INTERVAL_SECONDS = int(os.getenv("RADAR_SYNC_INTERVAL_SECONDS", "300"))
+RADAR_RETENTION_HOURS = int(os.getenv("RADAR_RETENTION_HOURS", "48"))
+RADAR_FILE_RE = re.compile(r"^[A-Za-z0-9_.-]+\.(?:tar|tar\.bz2|bz2|gz)$", re.IGNORECASE)
+
+
+def now_utc() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def log(message: str) -> None:
+    print(f"[{now_utc().isoformat()}] {message}", flush=True)
+
+
+def parse_apache_index(html: str) -> list[str]:
+    pattern = re.compile(r'href="(?P<href>[^"]+\.(?:tar|tar\.bz2|bz2|gz))"', re.IGNORECASE)
+    return [m.group("href") for m in pattern.finditer(html) if not m.group("href").startswith("../")]
+
+
+def cleanup() -> None:
+    cutoff = now_utc() - timedelta(hours=RADAR_RETENTION_HOURS)
+    if not RAW_RADAR_DIR.exists():
+        return
+    for path in RAW_RADAR_DIR.rglob("*"):
+        if not path.is_file() or not RADAR_FILE_RE.fullmatch(path.name):
+            continue
+        modified = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc)
+        if modified < cutoff:
+            path.unlink(missing_ok=True)
+            log(f"deleted old radar file {path.name}")
+
+
+async def sync_once() -> None:
+    base_url = f"{DWD_RADAR_BASE_URL}/{RADAR_PRODUCT}/"
+    target_dir = RAW_RADAR_DIR / "composite" / RADAR_PRODUCT
+    target_dir.mkdir(parents=True, exist_ok=True)
+
+    async with httpx.AsyncClient(timeout=90.0, follow_redirects=True) as client:
+        response = await client.get(base_url)
+        response.raise_for_status()
+        files = parse_apache_index(response.text)[-RADAR_SYNC_LIMIT:]
+        if not files:
+            log(f"no radar files found at {base_url}")
+            return
+        downloaded = 0
+        for filename in files:
+            target = target_dir / filename
+            if target.exists() and target.stat().st_size > 0:
+                continue
+            file_response = await client.get(urljoin(base_url, filename))
+            file_response.raise_for_status()
+            target.write_bytes(file_response.content)
+            downloaded += 1
+            log(f"downloaded {filename}")
+        cleanup()
+        log(f"sync complete: product={RADAR_PRODUCT}, checked={len(files)}, downloaded={downloaded}")
+
+
+async def main() -> None:
+    log("radar worker started")
+    while True:
+        try:
+            await sync_once()
+        except Exception as exc:
+            log(f"sync failed: {exc}")
+        await asyncio.sleep(RADAR_SYNC_INTERVAL_SECONDS)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())