banyaro/backend/scraper/events_vdh.py

242 lines
9.4 KiB
Python

"""
BAN YARO — VDH Veranstaltungs-Scraper
Scrapt Hundeveranstaltungen von vdh.de.
Bei Fehler oder 0 Ergebnissen: Fallback auf hartcodierte Events.
"""
import logging
import re
from datetime import datetime
import httpx
logger = logging.getLogger(__name__)
FALLBACK_EVENTS = [
{"titel": "VDH-Europasiegershow 2026", "datum": "2026-06-14", "ort_name": "Dortmund", "typ": "ausstellung", "link": "https://www.vdh.de/ausstellungen/", "external_id": "vdh-fallback-europasieger-2026"},
{"titel": "Internationale Hundeausstellung Hannover", "datum": "2026-06-27", "ort_name": "Hannover", "typ": "ausstellung", "link": "https://www.vdh.de/ausstellungen/", "external_id": "vdh-fallback-hannover-2026"},
{"titel": "VDH-Bundessiegerprüfung Agility", "datum": "2026-07-19", "ort_name": "Leipzig", "typ": "wettkampf", "link": "https://www.vdh.de/hundesport/termine/", "external_id": "vdh-fallback-agility-2026"},
{"titel": "Internationale Hundeausstellung München", "datum": "2026-08-01", "ort_name": "München", "typ": "ausstellung", "link": "https://www.vdh.de/ausstellungen/", "external_id": "vdh-fallback-muenchen-2026"},
{"titel": "Deutsche Meisterschaft Agility", "datum": "2026-10-24", "ort_name": "Hemsbach", "typ": "wettkampf", "link": "https://www.vdh.de/hundesport/termine/", "external_id": "vdh-fallback-dm-agility-2026"},
{"titel": "Deutsche Meisterschaft Para Agility", "datum": "2026-11-07", "ort_name": "Hückelhoven", "typ": "wettkampf", "link": "https://www.vdh.de/hundesport/termine/", "external_id": "vdh-fallback-para-agility-2026"},
{"titel": "Internationale Hundeausstellung Frankfurt", "datum": "2026-11-21", "ort_name": "Frankfurt am Main", "typ": "ausstellung", "link": "https://www.vdh.de/ausstellungen/", "external_id": "vdh-fallback-frankfurt-nov-2026"},
]
_TYP_MAP = {
"ausstellung": "ausstellung", "show": "ausstellung", "siegershow": "ausstellung",
"agility": "wettkampf", "wettkampf": "wettkampf", "turnier": "wettkampf",
"prüfung": "wettkampf", "meisterschaft": "wettkampf",
"training": "training", "treffen": "treffen", "markt": "markt",
}
def _guess_typ(text: str) -> str:
t = text.lower()
for keyword, typ in _TYP_MAP.items():
if keyword in t:
return typ
return "ausstellung"
def _parse_date(raw: str) -> str | None:
"""DD.MM.YYYY oder DD.MM.YYYY - DD.MM.YYYY → YYYY-MM-DD (erstes Datum)."""
raw = raw.strip().split(" - ")[0].strip()
m = re.match(r'^(\d{1,2})\.(\d{1,2})\.(\d{4})$', raw)
if m:
d, mo, y = m.groups()
return f"{y}-{int(mo):02d}-{int(d):02d}"
m = re.match(r'^(\d{4})-(\d{2})-(\d{2})$', raw)
if m:
return raw
return None
def _strip_tags(html: str) -> str:
"""Entfernt HTML-Tags."""
return re.sub(r'<[^>]+>', '', html).strip()
def _build_external_id(ev: dict) -> str:
raw = f"vdh-{ev['datum']}-{ev['titel']}"
key = re.sub(r'[^a-z0-9]+', '-', raw.lower()).strip('-')
return key[:120]
# ── PARSER 1: /ausstellungen/liste/typ/spezial/ ──────────────────────────────
# Struktur innerhalb div.ausstellung_liste:
# <div class="row">
# <div class="span6">
# <b>Rassen</b><br>DD.MM.YYYY<br> Verein<br> Straße<br> PLZ Ort<br>
# </div>
# <div class="span6">…Kontakt…</div>
# </div>
def _parse_spezial(html: str) -> list[dict]:
events = []
# Ausstellung_liste-Block extrahieren
m = re.search(r'<div class="ausstellung_liste">(.*?)(?=<div class="row">\s*<div class="span12">)',
html, re.DOTALL)
block = m.group(1) if m else html
# Jede linke span6 (erstes span6 pro row) extrahieren
# Pattern: <div class="row"> ... <div class="span6">INHALT</div>
row_pattern = re.compile(
r'<div class="row">\s*<div class="span6">(.*?)</div>',
re.DOTALL
)
for row_m in row_pattern.finditer(block):
cell = row_m.group(1)
# Titel aus <b>...</b>
title_m = re.search(r'<b>(.*?)</b>', cell, re.DOTALL)
if not title_m:
continue
title = _strip_tags(title_m.group(1)).strip()
if not title:
continue
# Datum: erste DD.MM.YYYY nach dem <b>-Block
after_b = cell[title_m.end():]
date_m = re.search(r'(\d{1,2}\.\d{1,2}\.\d{4})', after_b)
if not date_m:
continue
date_str = _parse_date(date_m.group(1))
if not date_str:
continue
# PLZ + Ort: "12345 Stadtname"
ort = ""
ort_m = re.search(r'(\d{5})\s+([^<\n\r]+)', after_b)
if ort_m:
ort = ort_m.group(2).strip()
events.append({
"titel": title,
"datum": date_str,
"ort_name": ort,
"link": "https://www.vdh.de/ausstellungen/",
})
return events
# ── PARSER 2: /hundesport/termine/ ───────────────────────────────────────────
# Struktur: <h2 class="h2ash1">Kategorie</h2> dann <ul><li>
# <li>DD.MM.YYYY - DD.MM.YYYY<br>Titel<br><b>Ort:</b> Stadt<br></li>
def _parse_sport(html: str) -> list[dict]:
events = []
# <li>-Blöcke extrahieren
li_pattern = re.compile(r'<li>(.*?)</li>', re.DOTALL)
for li_m in li_pattern.finditer(html):
cell = li_m.group(1)
# Datum: erstes DD.MM.YYYY oder DD.MM.YYYY - DD.MM.YYYY
date_m = re.search(r'(\d{1,2}\.\d{1,2}\.\d{4}(?:\s*-\s*\d{1,2}\.\d{1,2}\.\d{4})?)', cell)
if not date_m:
continue
date_str = _parse_date(date_m.group(1))
if not date_str:
continue
# Text nach dem Datum ohne Tags
after_date = cell[date_m.end():]
# "Ort:" aus <b>Ort:</b> Stadt entfernen wir für den Titel
parts = [p.strip() for p in re.split(r'<br\s*/?>|<b>[^<]*</b>', after_date) if p.strip()]
parts = [_strip_tags(p) for p in parts if _strip_tags(p)]
title = ""
ort = ""
for part in parts:
if re.match(r'^Ort:\s*', part, re.IGNORECASE):
ort = re.sub(r'^Ort:\s*', '', part, flags=re.IGNORECASE).strip()
elif not title and not re.match(r'^\d', part):
title = part
if not title:
continue
events.append({
"titel": title,
"datum": date_str,
"ort_name": ort,
"link": "https://www.vdh.de/hundesport/termine/",
})
return events
async def fetch_vdh_events() -> list[dict]:
"""
Scrapt VDH-Veranstaltungen von ausstellungen/liste und hundesport/termine.
Gibt eine Liste von Dicts zurück: {titel, datum, ort_name, typ, link, external_id}
"""
sources = [
("https://www.vdh.de/ausstellungen/liste/typ/spezial/", _parse_spezial),
("https://www.vdh.de/hundesport/termine/", _parse_sport),
]
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "de-DE,de;q=0.9,en;q=0.5",
}
raw_events: list[dict] = []
timeout = httpx.Timeout(connect=15.0, read=60.0, write=10.0, pool=10.0)
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
for url, parse_fn in sources:
try:
resp = await client.get(url, headers=headers)
resp.raise_for_status()
found = parse_fn(resp.text)
if found:
logger.info(f"VDH-Scraper: {len(found)} Events von {url}")
raw_events.extend(found)
else:
logger.info(f"VDH-Scraper: Keine Events auf {url}")
except httpx.HTTPStatusError as e:
logger.warning(f"VDH-Scraper HTTP-Fehler {e.response.status_code} für {url}: {e}")
except httpx.TimeoutException as e:
logger.warning(f"VDH-Scraper Timeout für {url}: {type(e).__name__}")
except Exception as e:
logger.warning(f"VDH-Scraper Fehler für {url}: {type(e).__name__}: {e}")
if not raw_events:
logger.warning("VDH-Scraper: Keine Daten — verwende Fallback-Events.")
return list(FALLBACK_EVENTS)
today = datetime.today().strftime("%Y-%m-%d")
result = []
seen_ids: set[str] = set()
for ev in raw_events:
datum = ev.get("datum", "")
if datum < today:
continue
titel = ev.get("titel", "").strip()
if not titel or len(titel) < 3:
continue
entry = {
"titel": titel,
"datum": datum,
"ort_name": ev.get("ort_name") or None,
"typ": _guess_typ(titel),
"link": ev.get("link", "https://www.vdh.de"),
"external_id": _build_external_id(ev),
}
if entry["external_id"] not in seen_ids:
seen_ids.add(entry["external_id"])
result.append(entry)
if not result:
logger.warning("VDH-Scraper: Nach Filterung 0 Events — verwende Fallback.")
return list(FALLBACK_EVENTS)
logger.info(f"VDH-Scraper: {len(result)} zukünftige Events nach Normalisierung.")
return result