ecki
859 lines
31 KiB
Python
859 lines
31 KiB
Python
import json
|
|
import re
|
|
from datetime import datetime
|
|
from difflib import SequenceMatcher
|
|
from html import unescape
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
from app.models import RegionScope, WatchItem, WatchSource, WatchType
|
|
from app.providers.utils import normalize_search_text
|
|
|
|
|
|
MONTH_ALIASES = {
|
|
"jan": 1,
|
|
"januar": 1,
|
|
"feb": 2,
|
|
"februar": 2,
|
|
"maer": 3,
|
|
"maerz": 3,
|
|
"mar": 3,
|
|
"maerz": 3,
|
|
"apr": 4,
|
|
"april": 4,
|
|
"mai": 5,
|
|
"jun": 6,
|
|
"juni": 6,
|
|
"jul": 7,
|
|
"juli": 7,
|
|
"aug": 8,
|
|
"august": 8,
|
|
"sep": 9,
|
|
"sept": 9,
|
|
"september": 9,
|
|
"okt": 10,
|
|
"oktober": 10,
|
|
"nov": 11,
|
|
"november": 11,
|
|
"dez": 12,
|
|
"dezember": 12,
|
|
}
|
|
|
|
FOLLOW_LINK_KEYWORDS = (
|
|
"event",
|
|
"gig",
|
|
"konzert",
|
|
"live",
|
|
"rausgegangen",
|
|
"show",
|
|
"termin",
|
|
"ticket",
|
|
"tour",
|
|
)
|
|
|
|
MAX_FOLLOWED_LINKS = 6
|
|
|
|
MOJIBAKE_MARKERS = ("Â", "Ã", "â", *[chr(code) for code in range(0x80, 0xA0)])
|
|
|
|
|
|
def clean_display_text(value: str | None) -> str:
|
|
if not value:
|
|
return ""
|
|
|
|
cleaned = " ".join(str(value).split())
|
|
if not any(marker in cleaned for marker in MOJIBAKE_MARKERS):
|
|
return cleaned
|
|
|
|
try:
|
|
repaired = cleaned.encode("latin1").decode("utf-8")
|
|
except UnicodeError:
|
|
return cleaned
|
|
return " ".join(repaired.split())
|
|
|
|
|
|
class SourceScanner:
|
|
headers = {
|
|
"User-Agent": "eventlens/0.1 (+https://local)",
|
|
"Accept": "text/html,application/xhtml+xml,application/json",
|
|
"Accept-Language": "de-DE,de;q=0.9,en;q=0.7",
|
|
}
|
|
|
|
def __init__(self):
|
|
self.last_message = ""
|
|
|
|
def scan(self, watch_item: WatchItem, source: WatchSource) -> list[dict]:
|
|
self.last_message = ""
|
|
response = requests.get(
|
|
source.url,
|
|
headers=self.headers,
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
content_type = response.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
results = self._scan_json(watch_item, source, response.json())
|
|
if not self.last_message:
|
|
self.last_message = self._build_scan_message(results, "JSON-Daten ausgewertet.")
|
|
return results
|
|
|
|
results = self._scan_html(watch_item, source, response.text, response.url)
|
|
if not self.last_message:
|
|
self.last_message = self._build_scan_message(results, "HTML-Seite ausgewertet.")
|
|
return results
|
|
|
|
def _scan_json(self, watch_item: WatchItem, source: WatchSource, payload) -> list[dict]:
|
|
events = self._extract_jsonld_events(payload)
|
|
return self._events_from_jsonld(watch_item, source, events)
|
|
|
|
def _scan_html(
|
|
self,
|
|
watch_item: WatchItem,
|
|
source: WatchSource,
|
|
html: str,
|
|
base_url: str | None = None,
|
|
follow_links: bool = True,
|
|
) -> list[dict]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
source_url = base_url or source.url
|
|
jsonld_events = []
|
|
|
|
for script in soup.find_all("script", type="application/ld+json"):
|
|
raw_payload = script.string or script.get_text()
|
|
if not raw_payload:
|
|
continue
|
|
try:
|
|
payload = json.loads(unescape(raw_payload))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
jsonld_events.extend(self._extract_jsonld_events(payload))
|
|
|
|
jsonld_results = self._events_from_jsonld(watch_item, source, jsonld_events)
|
|
jsonld_message = self.last_message
|
|
html_results = self._events_from_html_text(watch_item, source, soup, source_url)
|
|
html_message = self.last_message
|
|
combined_results = self._dedupe_results([*jsonld_results, *html_results])
|
|
if combined_results:
|
|
if jsonld_results and html_results:
|
|
self.last_message = (
|
|
f"{len(combined_results)} passende Events aus strukturierten Daten "
|
|
"und Terminzeilen gefunden."
|
|
)
|
|
elif jsonld_results:
|
|
self.last_message = jsonld_message
|
|
else:
|
|
self.last_message = html_message
|
|
return combined_results
|
|
|
|
if follow_links:
|
|
linked_results = self._events_from_linked_pages(watch_item, source, soup, source_url)
|
|
if linked_results:
|
|
return linked_results
|
|
|
|
return []
|
|
|
|
def _dedupe_results(self, results: list[dict]) -> list[dict]:
|
|
deduped: list[dict] = []
|
|
seen_keys: set[tuple[str, str, str, str]] = set()
|
|
|
|
for result in results:
|
|
event_date = result.get("event_date")
|
|
if isinstance(event_date, datetime):
|
|
date_key = event_date.date().isoformat()
|
|
else:
|
|
date_key = str(event_date or "")
|
|
|
|
key = (
|
|
date_key,
|
|
normalize_search_text(result.get("matched_term") or result.get("title") or ""),
|
|
normalize_search_text(result.get("venue_name") or ""),
|
|
normalize_search_text(result.get("city") or ""),
|
|
)
|
|
if key in seen_keys:
|
|
continue
|
|
seen_keys.add(key)
|
|
deduped.append(result)
|
|
|
|
return deduped
|
|
|
|
def _extract_jsonld_events(self, payload) -> list[dict]:
|
|
events: list[dict] = []
|
|
if isinstance(payload, list):
|
|
for item in payload:
|
|
events.extend(self._extract_jsonld_events(item))
|
|
return events
|
|
|
|
if not isinstance(payload, dict):
|
|
return events
|
|
|
|
graph = payload.get("@graph")
|
|
if isinstance(graph, list):
|
|
for item in graph:
|
|
events.extend(self._extract_jsonld_events(item))
|
|
|
|
item_type = payload.get("@type")
|
|
if isinstance(item_type, list):
|
|
is_event = "Event" in item_type
|
|
else:
|
|
is_event = item_type == "Event"
|
|
if is_event:
|
|
events.append(payload)
|
|
|
|
return events
|
|
|
|
def _events_from_jsonld(
|
|
self,
|
|
watch_item: WatchItem,
|
|
source: WatchSource,
|
|
events: list[dict],
|
|
) -> list[dict]:
|
|
results: list[dict] = []
|
|
normalized_term = normalize_search_text(watch_item.name)
|
|
matching_name_count = 0
|
|
outside_region_count = 0
|
|
past_count = 0
|
|
|
|
for event in events:
|
|
title = clean_display_text(event.get("name"))
|
|
performers = self._extract_performer_names(event)
|
|
haystack = normalize_search_text(" ".join([title] + performers))
|
|
if not self._term_matches_normalized(normalized_term, haystack):
|
|
continue
|
|
matching_name_count += 1
|
|
|
|
location = event.get("location") or {}
|
|
address = location.get("address") or {}
|
|
city = clean_display_text(address.get("addressLocality") or location.get("name"))
|
|
if watch_item.region_scope == RegionScope.hamburg and normalize_search_text(city) != "hamburg":
|
|
outside_region_count += 1
|
|
continue
|
|
|
|
event_date = self._parse_datetime(event.get("startDate"))
|
|
if event_date and event_date.date() < datetime.utcnow().date():
|
|
past_count += 1
|
|
continue
|
|
ticket_url = event.get("url") or source.url
|
|
|
|
results.append(
|
|
{
|
|
"external_id": str(event.get("@id") or ticket_url or f"{source.id}:{title}"),
|
|
"title": title or watch_item.name,
|
|
"matched_term": watch_item.name,
|
|
"venue_name": clean_display_text(location.get("name") or source.label),
|
|
"city": city,
|
|
"country_code": "DE",
|
|
"event_date": event_date,
|
|
"ticket_url": ticket_url,
|
|
"image_url": self._extract_image(event),
|
|
"raw_payload": event,
|
|
}
|
|
)
|
|
|
|
if results:
|
|
self.last_message = f"{len(results)} passende Events in strukturierten Daten gefunden."
|
|
elif outside_region_count:
|
|
self.last_message = (
|
|
f"Strukturierte Daten gefunden, aber kein Termin fuer '{watch_item.name}' "
|
|
"in Hamburg."
|
|
)
|
|
elif past_count:
|
|
self.last_message = (
|
|
f"Strukturierte Daten gefunden, aber nur vergangene Termine fuer '{watch_item.name}'."
|
|
)
|
|
elif matching_name_count:
|
|
self.last_message = (
|
|
f"Strukturierte Daten gefunden, aber keine passenden Termine fuer '{watch_item.name}'."
|
|
)
|
|
|
|
return results
|
|
|
|
def _events_from_html_text(
|
|
self,
|
|
watch_item: WatchItem,
|
|
source: WatchSource,
|
|
soup: BeautifulSoup,
|
|
base_url: str | None = None,
|
|
) -> list[dict]:
|
|
text = soup.get_text(" ", strip=True)
|
|
normalized_text = normalize_search_text(text)
|
|
normalized_term = normalize_search_text(watch_item.name)
|
|
if not self._term_matches_normalized(normalized_term, normalized_text):
|
|
self.last_message = (
|
|
f"Seite erreichbar, aber der Name '{watch_item.name}' wurde nicht gefunden."
|
|
)
|
|
return []
|
|
|
|
results: list[dict] = []
|
|
seen_keys: set[str] = set()
|
|
line_results = self._events_from_text_lines(watch_item, source, soup, base_url)
|
|
if line_results:
|
|
self.last_message = f"{len(line_results)} passende Events aus Terminzeilen gefunden."
|
|
return line_results
|
|
|
|
contexts_with_date = 0
|
|
past_contexts = 0
|
|
outside_region_contexts = 0
|
|
for context in self._find_matching_contexts(soup, watch_item):
|
|
context_text = clean_display_text(context.get_text(" ", strip=True))
|
|
event_date = self._find_nearest_date(context_text, watch_item.name)
|
|
if event_date is None:
|
|
continue
|
|
contexts_with_date += 1
|
|
if event_date.date() < datetime.utcnow().date():
|
|
past_contexts += 1
|
|
continue
|
|
if (
|
|
watch_item.region_scope == RegionScope.hamburg
|
|
and "hamburg" not in normalize_search_text(context_text)
|
|
):
|
|
outside_region_contexts += 1
|
|
continue
|
|
|
|
title = self._find_title(context, watch_item.name)
|
|
context_url = base_url or source.url
|
|
link = self._find_nearest_link(context, watch_item.name, context_url) or context_url
|
|
key = f"{source.id}:{normalize_search_text(title)}:{event_date.date().isoformat()}"
|
|
if key in seen_keys:
|
|
continue
|
|
seen_keys.add(key)
|
|
|
|
results.append(
|
|
{
|
|
"external_id": key,
|
|
"title": title,
|
|
"matched_term": watch_item.name,
|
|
"venue_name": clean_display_text(self._find_venue(context_text, source.label)),
|
|
"city": "Hamburg" if watch_item.region_scope == RegionScope.hamburg else None,
|
|
"country_code": "DE",
|
|
"event_date": event_date,
|
|
"ticket_url": link,
|
|
"image_url": None,
|
|
"raw_payload": {
|
|
"source_url": context_url,
|
|
"parser": "html_text",
|
|
"context": clean_display_text(context_text[:1000]),
|
|
},
|
|
}
|
|
)
|
|
|
|
if results:
|
|
self.last_message = f"{len(results)} passende Events gefunden."
|
|
elif outside_region_contexts:
|
|
self.last_message = (
|
|
f"Seite erreichbar, Termine fuer '{watch_item.name}' gefunden, "
|
|
"aber keiner in Hamburg."
|
|
)
|
|
elif past_contexts:
|
|
self.last_message = (
|
|
f"Seite erreichbar, aber nur vergangene Termine fuer '{watch_item.name}' gefunden."
|
|
)
|
|
elif contexts_with_date == 0:
|
|
self.last_message = (
|
|
f"Seite erreichbar, Name '{watch_item.name}' gefunden, "
|
|
"aber keine auswertbaren Termine."
|
|
)
|
|
|
|
return results
|
|
|
|
def _events_from_text_lines(
|
|
self,
|
|
watch_item: WatchItem,
|
|
source: WatchSource,
|
|
soup: BeautifulSoup,
|
|
base_url: str | None,
|
|
) -> list[dict]:
|
|
results: list[dict] = []
|
|
seen_keys: set[str] = set()
|
|
title = self._find_title(soup, watch_item.name)
|
|
context_url = base_url or source.url
|
|
|
|
for line in self._extract_text_lines(soup):
|
|
if (
|
|
watch_item.watch_type == WatchType.event
|
|
and not self._term_matches_normalized(
|
|
normalize_search_text(watch_item.name),
|
|
normalize_search_text(line),
|
|
)
|
|
):
|
|
continue
|
|
event_date = self._find_nearest_date(line, watch_item.name)
|
|
if event_date is None:
|
|
continue
|
|
if event_date.date() < datetime.utcnow().date():
|
|
continue
|
|
if (
|
|
watch_item.region_scope == RegionScope.hamburg
|
|
and "hamburg" not in normalize_search_text(line)
|
|
):
|
|
continue
|
|
|
|
city = clean_display_text(self._find_city_in_line(line, watch_item.region_scope))
|
|
venue_name = clean_display_text(self._find_venue_from_line(line, source.label))
|
|
key = f"{source.id}:{normalize_search_text(line)}:{event_date.date().isoformat()}"
|
|
if key in seen_keys:
|
|
continue
|
|
seen_keys.add(key)
|
|
|
|
results.append(
|
|
{
|
|
"external_id": key,
|
|
"title": title,
|
|
"matched_term": watch_item.name,
|
|
"venue_name": venue_name,
|
|
"city": city,
|
|
"country_code": "DE",
|
|
"event_date": event_date,
|
|
"ticket_url": self._find_link_for_line(soup, line, context_url) or context_url,
|
|
"image_url": None,
|
|
"raw_payload": {
|
|
"source_url": context_url,
|
|
"parser": "html_line",
|
|
"context": clean_display_text(line[:1000]),
|
|
},
|
|
}
|
|
)
|
|
|
|
return results
|
|
|
|
def _extract_text_lines(self, soup: BeautifulSoup) -> list[str]:
|
|
lines = []
|
|
seen_lines: set[str] = set()
|
|
for line in soup.get_text("\n", strip=True).splitlines():
|
|
cleaned = clean_display_text(line)
|
|
if len(cleaned) < 8:
|
|
continue
|
|
if not self._find_nearest_date(cleaned, ""):
|
|
continue
|
|
if cleaned in seen_lines:
|
|
continue
|
|
seen_lines.add(cleaned)
|
|
lines.append(cleaned)
|
|
|
|
for node in soup.find_all(
|
|
["article", "li", "p", "tr", "section", "div", "span", "h1", "h2", "h3", "h4"]
|
|
):
|
|
cleaned = clean_display_text(node.get_text(" ", strip=True))
|
|
if len(cleaned) < 8 or len(cleaned) > 500:
|
|
continue
|
|
if not self._find_nearest_date(cleaned, ""):
|
|
continue
|
|
if cleaned in seen_lines:
|
|
continue
|
|
seen_lines.add(cleaned)
|
|
lines.append(cleaned)
|
|
return lines
|
|
|
|
def _find_city_in_line(self, line: str, region_scope: RegionScope) -> str | None:
|
|
if region_scope == RegionScope.hamburg and "hamburg" in normalize_search_text(line):
|
|
return "Hamburg"
|
|
|
|
match = re.search(
|
|
r"\b\d{1,2}\.\d{1,2}\.(?:\d{2,4}\.?)?\s+([^/]+)",
|
|
line,
|
|
re.IGNORECASE,
|
|
)
|
|
if match:
|
|
return clean_display_text(match.group(1).strip(" /-"))
|
|
return None
|
|
|
|
def _find_venue_from_line(self, line: str, default: str) -> str:
|
|
if "/" not in line:
|
|
return clean_display_text(default)
|
|
venue = clean_display_text(line.split("/", 1)[1].strip())
|
|
return venue or clean_display_text(default)
|
|
|
|
def _find_link_for_line(self, soup: BeautifulSoup, line: str, base_url: str) -> str | None:
|
|
normalized_line = normalize_search_text(line)
|
|
venue = normalize_search_text(self._find_venue_from_line(line, ""))
|
|
|
|
candidates = []
|
|
for link in soup.find_all("a", href=True):
|
|
link_text = link.get_text(" ", strip=True)
|
|
normalized_link_text = normalize_search_text(link_text)
|
|
if not normalized_link_text:
|
|
continue
|
|
score = 0
|
|
if normalized_link_text and normalized_link_text in normalized_line:
|
|
score += 2
|
|
if venue and normalized_link_text in venue:
|
|
score += 2
|
|
if score:
|
|
candidates.append((score, len(normalized_link_text), link["href"]))
|
|
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(key=lambda item: (item[0], item[1]), reverse=True)
|
|
return urljoin(base_url, candidates[0][2])
|
|
|
|
def _events_from_linked_pages(
|
|
self,
|
|
watch_item: WatchItem,
|
|
source: WatchSource,
|
|
soup: BeautifulSoup,
|
|
base_url: str,
|
|
) -> list[dict]:
|
|
results: list[dict] = []
|
|
seen_event_keys: set[str] = set()
|
|
checked_links = 0
|
|
|
|
for link_url in self._candidate_follow_links(soup, base_url):
|
|
try:
|
|
response = requests.get(link_url, headers=self.headers, timeout=30)
|
|
response.raise_for_status()
|
|
except requests.RequestException:
|
|
continue
|
|
|
|
checked_links += 1
|
|
content_type = response.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
linked_results = self._scan_json(watch_item, source, response.json())
|
|
elif "text/html" in content_type or "application/xhtml+xml" in content_type or not content_type:
|
|
linked_results = self._scan_html(
|
|
watch_item,
|
|
source,
|
|
response.text,
|
|
response.url,
|
|
follow_links=False,
|
|
)
|
|
else:
|
|
continue
|
|
|
|
for event in linked_results:
|
|
key = event["external_id"]
|
|
if key in seen_event_keys:
|
|
continue
|
|
seen_event_keys.add(key)
|
|
results.append(event)
|
|
|
|
if results:
|
|
self.last_message = f"{len(results)} passende Events auf verlinkten Seiten gefunden."
|
|
elif checked_links:
|
|
self.last_message = (
|
|
f"Seite erreichbar, {checked_links} relevante Links geprueft, "
|
|
"aber keine passenden Events gefunden."
|
|
)
|
|
|
|
return results
|
|
|
|
def _build_scan_message(self, results: list[dict], fallback: str) -> str:
|
|
if results:
|
|
return f"{len(results)} passende Events gefunden."
|
|
return f"{fallback} Keine passenden Events gefunden."
|
|
|
|
def _candidate_follow_links(self, soup: BeautifulSoup, base_url: str) -> list[str]:
|
|
base_host = urlparse(base_url).netloc.lower()
|
|
scored_links: list[tuple[int, str]] = []
|
|
seen_urls: set[str] = set()
|
|
|
|
for link in soup.find_all("a", href=True):
|
|
href = link["href"].strip()
|
|
if not href or href.startswith(("#", "mailto:", "tel:", "javascript:")):
|
|
continue
|
|
|
|
link_url = urljoin(base_url, href)
|
|
if link_url in seen_urls or link_url.rstrip("/") == base_url.rstrip("/"):
|
|
continue
|
|
seen_urls.add(link_url)
|
|
|
|
parsed = urlparse(link_url)
|
|
if parsed.scheme not in {"http", "https"}:
|
|
continue
|
|
|
|
link_text = link.get_text(" ", strip=True)
|
|
haystack = normalize_search_text(" ".join([link_text, href, parsed.netloc, parsed.path]))
|
|
keyword_hits = sum(1 for keyword in FOLLOW_LINK_KEYWORDS if keyword in haystack)
|
|
if keyword_hits == 0:
|
|
continue
|
|
|
|
same_host_bonus = 1 if parsed.netloc.lower() == base_host else 0
|
|
scored_links.append((keyword_hits + same_host_bonus, link_url))
|
|
|
|
scored_links.sort(key=lambda item: item[0], reverse=True)
|
|
return [link_url for _, link_url in scored_links[:MAX_FOLLOWED_LINKS]]
|
|
|
|
def _extract_performer_names(self, event: dict) -> list[str]:
|
|
performer = event.get("performer") or event.get("performers")
|
|
if isinstance(performer, dict):
|
|
return [clean_display_text(performer.get("name"))]
|
|
if isinstance(performer, list):
|
|
return [clean_display_text(item.get("name")) for item in performer if isinstance(item, dict)]
|
|
return []
|
|
|
|
def _extract_image(self, event: dict) -> str | None:
|
|
image = event.get("image")
|
|
if isinstance(image, str):
|
|
return image
|
|
if isinstance(image, list):
|
|
for item in image:
|
|
if isinstance(item, str):
|
|
return item
|
|
return None
|
|
|
|
def _parse_datetime(self, value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
|
|
except ValueError:
|
|
pass
|
|
for fmt in ("%d.%m.%Y", "%Y-%m-%d"):
|
|
try:
|
|
return datetime.strptime(value[:10], fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
def _find_nearest_date(self, text: str, term: str) -> datetime | None:
|
|
normalized_term = normalize_search_text(term)
|
|
normalized_text = normalize_search_text(text)
|
|
term_index = normalized_text.find(normalized_term)
|
|
search_area = text
|
|
if term_index >= 0:
|
|
start = max(0, term_index - 300)
|
|
end = min(len(text), term_index + 500)
|
|
search_area = text[start:end]
|
|
|
|
explicit_candidates: list[datetime] = []
|
|
inferred_candidates: list[datetime] = []
|
|
|
|
for pattern in (
|
|
r"\b(\d{1,2}\.\d{1,2}\.\d{4})(?=\D|$)",
|
|
r"\b(\d{1,2}\.\d{1,2}\.\d{2}\.?)(?=\D|$)",
|
|
):
|
|
for match in re.finditer(pattern, search_area):
|
|
parsed = self._parse_german_date(match.group(1))
|
|
if parsed:
|
|
explicit_candidates.append(parsed)
|
|
|
|
for match in re.finditer(r"\b(\d{1,2}\.\d{1,2}\.)(?!\d)", search_area):
|
|
parsed = self._parse_german_date(match.group(1))
|
|
if parsed:
|
|
inferred_candidates.append(parsed)
|
|
|
|
month_name_pattern = (
|
|
r"jan(?:uar)?|feb(?:ruar)?|m(?:ae|ä)r(?:z)?|apr(?:il)?|mai|jun(?:i)?|"
|
|
r"jul(?:i)?|aug(?:ust)?|sep(?:t|tember)?|okt(?:ober)?|nov(?:ember)?|dez(?:ember)?"
|
|
)
|
|
for match in re.finditer(
|
|
rf"\b(\d{{1,2}})\.?\s+({month_name_pattern})\.?\s*(\d{{4}})?\b",
|
|
search_area,
|
|
re.IGNORECASE,
|
|
):
|
|
parsed = self._parse_named_month_date(match.group(1), match.group(2), match.group(3))
|
|
if parsed:
|
|
if match.group(3):
|
|
explicit_candidates.append(parsed)
|
|
else:
|
|
inferred_candidates.append(parsed)
|
|
for match in re.finditer(
|
|
rf"\b({month_name_pattern})\.?\s+(\d{{1,2}})\.?\s*(\d{{4}})?\b",
|
|
search_area,
|
|
re.IGNORECASE,
|
|
):
|
|
parsed = self._parse_named_month_date(match.group(2), match.group(1), match.group(3))
|
|
if parsed:
|
|
if match.group(3):
|
|
explicit_candidates.append(parsed)
|
|
else:
|
|
inferred_candidates.append(parsed)
|
|
|
|
if explicit_candidates:
|
|
future_explicit_candidates = [
|
|
candidate
|
|
for candidate in explicit_candidates
|
|
if candidate.date() >= datetime.utcnow().date()
|
|
]
|
|
if future_explicit_candidates:
|
|
return sorted(future_explicit_candidates)[0]
|
|
return sorted(explicit_candidates)[0]
|
|
|
|
future_candidates = [
|
|
candidate for candidate in inferred_candidates if candidate.date() >= datetime.utcnow().date()
|
|
]
|
|
if future_candidates:
|
|
return sorted(future_candidates)[0]
|
|
return sorted(inferred_candidates)[0] if inferred_candidates else None
|
|
|
|
def _parse_german_date(self, value: str) -> datetime | None:
|
|
cleaned = value.strip()
|
|
cleaned = cleaned.rstrip(".") if re.fullmatch(r"\d{1,2}\.\d{1,2}\.\d{2}\.", cleaned) else cleaned
|
|
current_year = datetime.utcnow().year
|
|
candidates = [cleaned]
|
|
if re.fullmatch(r"\d{1,2}\.\d{1,2}\.", cleaned):
|
|
candidates.append(f"{cleaned}{current_year}")
|
|
candidates.append(f"{cleaned}{current_year + 1}")
|
|
elif re.fullmatch(r"\d{1,2}\.\d{1,2}\.\d{2}", cleaned):
|
|
day, month, year = cleaned.split(".")
|
|
candidates.append(f"{day}.{month}.20{year}")
|
|
|
|
for candidate in candidates:
|
|
try:
|
|
parsed = datetime.strptime(candidate, "%d.%m.%Y")
|
|
parsed = parsed.replace(hour=12)
|
|
if parsed.date() < datetime.utcnow().date() and candidate != cleaned:
|
|
continue
|
|
return parsed
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
def _parse_named_month_date(
|
|
self,
|
|
day_value: str,
|
|
month_value: str,
|
|
year_value: str | None,
|
|
) -> datetime | None:
|
|
month = MONTH_ALIASES.get(normalize_search_text(month_value).rstrip("."))
|
|
if month is None:
|
|
return None
|
|
|
|
day = int(day_value)
|
|
current_year = datetime.utcnow().year
|
|
years = [int(year_value)] if year_value else [current_year, current_year + 1]
|
|
for year in years:
|
|
try:
|
|
parsed = datetime(year, month, day)
|
|
except ValueError:
|
|
continue
|
|
if year_value or parsed.date() >= datetime.utcnow().date():
|
|
return parsed
|
|
return None
|
|
|
|
def _find_matching_contexts(self, soup: BeautifulSoup, watch_item: WatchItem) -> list:
|
|
normalized_term = normalize_search_text(watch_item.name)
|
|
selectors = [
|
|
"li.card",
|
|
".tourplan .row",
|
|
"[class*=event]",
|
|
"[class*=termin]",
|
|
"article",
|
|
"a[href]",
|
|
"tr",
|
|
"li",
|
|
".row",
|
|
]
|
|
candidates = []
|
|
seen_nodes = set()
|
|
|
|
for selector in selectors:
|
|
for node in soup.select(selector):
|
|
if id(node) in seen_nodes:
|
|
continue
|
|
seen_nodes.add(id(node))
|
|
text = node.get_text(" ", strip=True)
|
|
if not self._term_matches_normalized(normalized_term, normalize_search_text(text)):
|
|
continue
|
|
if len(text) > 3500:
|
|
continue
|
|
if self._find_nearest_date(text, watch_item.name):
|
|
candidates.append(node)
|
|
|
|
if candidates:
|
|
return candidates
|
|
|
|
fallback = self._find_best_context(soup, watch_item.name)
|
|
return [fallback] if fallback is not None else []
|
|
|
|
def _find_venue(self, text: str, default: str) -> str:
|
|
lines = [line.strip() for line in re.split(r"\s{2,}|\n|\r", text) if line.strip()]
|
|
for line in lines:
|
|
normalized = normalize_search_text(line)
|
|
if "hamburg" in normalized and len(line) <= 120:
|
|
return clean_display_text(line)
|
|
return clean_display_text(default)
|
|
|
|
def _find_best_context(self, soup: BeautifulSoup, term: str):
|
|
normalized_term = normalize_search_text(term)
|
|
candidates = []
|
|
for node in soup.find_all(string=True):
|
|
if self._term_matches_normalized(normalized_term, normalize_search_text(str(node))):
|
|
parent = node.parent
|
|
if parent is None:
|
|
continue
|
|
best_parent = self._climb_to_context_with_date(parent, term)
|
|
text = best_parent.get_text(" ", strip=True)
|
|
candidates.append(
|
|
(
|
|
0 if self._find_nearest_date(text, term) else 1,
|
|
len(text),
|
|
best_parent,
|
|
)
|
|
)
|
|
|
|
if not candidates:
|
|
return None
|
|
|
|
candidates.sort(key=lambda item: (item[0], item[1]))
|
|
return candidates[0][2]
|
|
|
|
def _climb_to_context_with_date(self, node, term: str):
|
|
current = node
|
|
best = node
|
|
for _ in range(6):
|
|
if current is None:
|
|
break
|
|
context_text = current.get_text(" ", strip=True)
|
|
if self._find_nearest_date(context_text, term):
|
|
return current
|
|
best = current
|
|
current = current.parent
|
|
return best
|
|
|
|
def _find_title(self, soup: BeautifulSoup, term: str) -> str:
|
|
if soup is None:
|
|
return term
|
|
normalized_term = normalize_search_text(term)
|
|
for heading in soup.find_all(["h1", "h2", "h3", "h4", "strong", "b", "a"]):
|
|
title = clean_display_text(heading.get_text(" ", strip=True))
|
|
if self._term_matches_normalized(normalized_term, normalize_search_text(title)):
|
|
return title
|
|
|
|
text = clean_display_text(soup.get_text(" ", strip=True))
|
|
dated_match = re.search(
|
|
r"(.{0,40}\d{1,2}\.\d{1,2}\.(?:\d{2,4})?.{0,100}"
|
|
+ re.escape(term)
|
|
+ r".{0,100})",
|
|
text,
|
|
re.IGNORECASE,
|
|
)
|
|
if dated_match:
|
|
return clean_display_text(dated_match.group(1))
|
|
|
|
match = re.search(r"(.{0,80}" + re.escape(term) + r".{0,80})", text, re.IGNORECASE)
|
|
if match:
|
|
return clean_display_text(match.group(1))
|
|
return clean_display_text(term)
|
|
|
|
def _find_nearest_link(self, soup: BeautifulSoup, term: str, base_url: str) -> str | None:
|
|
normalized_term = normalize_search_text(term)
|
|
for link in soup.find_all("a", href=True):
|
|
if self._term_matches_normalized(
|
|
normalized_term,
|
|
normalize_search_text(link.get_text(" ", strip=True)),
|
|
):
|
|
return urljoin(base_url, link["href"])
|
|
return None
|
|
|
|
def _term_matches_normalized(self, normalized_term: str, normalized_text: str) -> bool:
|
|
if not normalized_term or not normalized_text:
|
|
return False
|
|
if normalized_term in normalized_text:
|
|
return True
|
|
if len(normalized_term) < 8:
|
|
return False
|
|
|
|
term_tokens = normalized_term.split()
|
|
text_tokens = normalized_text.split()
|
|
if not term_tokens or len(text_tokens) < len(term_tokens):
|
|
return False
|
|
|
|
window_size = len(term_tokens)
|
|
for index in range(len(text_tokens) - window_size + 1):
|
|
candidate = " ".join(text_tokens[index : index + window_size])
|
|
if SequenceMatcher(None, normalized_term, candidate).ratio() >= 0.9:
|
|
return True
|
|
|
|
return False
|