242 lines
9.4 KiB
Python
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
|