44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
from app.config import DWD_RADAR_BASE_URL, RADAR_PRODUCT
|
|
from app.http_client import fetch_text
|
|
|
|
RADAR_FILE_RE = re.compile(r"^[A-Za-z0-9_.-]+\.(?:tar|tar\.bz2|bz2|gz)$", re.IGNORECASE)
|
|
|
|
|
|
def parse_apache_index(html: str, suffixes: tuple[str, ...] | None = None) -> list[dict[str, Any]]:
|
|
files: list[dict[str, Any]] = []
|
|
pattern = re.compile(
|
|
r'href="(?P<href>[^"]+)"[^>]*>[^<]+</a>\s*'
|
|
r'(?P<date>\d{2}-[A-Za-z]{3}-\d{4})\s+(?P<time>\d{2}:\d{2}(?::\d{2})?)\s+(?P<size>[0-9\-]+)?',
|
|
re.IGNORECASE,
|
|
)
|
|
for match in pattern.finditer(html):
|
|
href = match.group("href")
|
|
if href.startswith("../") or href.endswith("/"):
|
|
continue
|
|
if suffixes and not href.lower().endswith(tuple(s.lower() for s in suffixes)):
|
|
continue
|
|
files.append(
|
|
{
|
|
"name": href,
|
|
"modified": f"{match.group('date')} {match.group('time')}",
|
|
"size_bytes": None if match.group("size") in (None, "-") else int(match.group("size")),
|
|
}
|
|
)
|
|
return files
|
|
|
|
|
|
async def fetch_radar_index(product: str = RADAR_PRODUCT, limit: int = 12) -> dict[str, Any]:
|
|
base_url = f"{DWD_RADAR_BASE_URL}/{product}/"
|
|
html = await fetch_text(base_url, timeout=12.0)
|
|
files = parse_apache_index(html, suffixes=(".tar", ".tar.bz2", ".bz2", ".gz"))
|
|
files = files[-limit:]
|
|
for item in files:
|
|
item["url"] = urljoin(base_url, item["name"])
|
|
return {"source": base_url, "product": product, "count": len(files), "files": files}
|