290 lines
16 KiB
Python
290 lines
16 KiB
Python
import hashlib
|
||
import ipaddress
|
||
import json
|
||
import re
|
||
import socket
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta
|
||
from urllib.parse import urljoin, urlparse
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import httpx
|
||
from bs4 import BeautifulSoup
|
||
from dateutil import parser as date_parser
|
||
|
||
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"
|
||
BROWSER_HEADERS = {
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
||
"Accept-Language": "en-AU,en;q=0.9",
|
||
"Cache-Control": "no-cache",
|
||
"DNT": "1",
|
||
"From": "what2do local event index",
|
||
}
|
||
SOCIAL_WIDGET_MARKERS = ("instagram", "sbi_", "facebook", "social-feed", "tiktok")
|
||
|
||
@dataclass
|
||
class ExtractedEvent:
|
||
title: str
|
||
starts_at: datetime
|
||
ends_at: datetime | None = None
|
||
description: str = ""
|
||
url: str = ""
|
||
address: str = ""
|
||
|
||
class CrawlError(Exception):
|
||
pass
|
||
|
||
def validate_public_url(url: str) -> None:
|
||
parsed = urlparse(url)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
||
raise CrawlError("Only absolute http:// and https:// URLs are allowed.")
|
||
try:
|
||
infos = socket.getaddrinfo(parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80))
|
||
except socket.gaierror as exc:
|
||
raise CrawlError(f"Host cannot be resolved: {exc}") from exc
|
||
for info in infos:
|
||
ip = ipaddress.ip_address(info[4][0])
|
||
if not ip.is_global:
|
||
raise CrawlError("URLs resolving to private, loopback, link-local, or reserved addresses are blocked.")
|
||
|
||
def _http_client(timeout: int = 20) -> httpx.Client:
|
||
return httpx.Client(follow_redirects=True, timeout=timeout, headers=BROWSER_HEADERS)
|
||
|
||
def fetch_html(url: str) -> str:
|
||
validate_public_url(url)
|
||
with _http_client() as client:
|
||
response = client.get(url)
|
||
response.raise_for_status()
|
||
validate_public_url(str(response.url))
|
||
content_type = response.headers.get("content-type", "")
|
||
if "html" not in content_type.lower():
|
||
raise CrawlError(f"Expected HTML but received {content_type or 'an unknown content type'}.")
|
||
if len(response.content) > 5_000_000:
|
||
raise CrawlError("Page exceeds the 5 MB crawl limit.")
|
||
return response.text
|
||
|
||
def _text(node, selector: str) -> str:
|
||
if not selector:
|
||
return ""
|
||
found = node.select_one(selector)
|
||
return found.get_text(" ", strip=True) if found else ""
|
||
|
||
def _link(node, selector: str, base_url: str) -> str:
|
||
found = node.select_one(selector) if selector else None
|
||
target = found or node
|
||
for attribute in ("href", "data-permalink", "data-url", "data-href"):
|
||
value = target.get(attribute)
|
||
if value:
|
||
return urljoin(base_url, str(value))
|
||
return base_url
|
||
|
||
def _parse_date_range(value: str, source: dict, now: datetime) -> tuple[datetime, datetime | None]:
|
||
value = " ".join(value.replace("–", "-").replace("—", "-").split())
|
||
same_month = re.fullmatch(r"(\d{1,2})\s*-\s*(\d{1,2})\s+([A-Za-z]+)\s+(\d{4})(?:\s+(.+))?", value)
|
||
cross_month = re.fullmatch(r"(\d{1,2})\s+([A-Za-z]+)\s*-\s*(\d{1,2})\s+([A-Za-z]+)\s+(\d{4})(?:\s+(.+))?", value)
|
||
if same_month:
|
||
start_day, end_day, month, year, time_text = same_month.groups()
|
||
suffix = f" {time_text}" if time_text else ""
|
||
start = _parse_date(f"{start_day} {month} {year}{suffix}", source, now)
|
||
end = _parse_date(f"{end_day} {month} {year}{suffix}", source, now)
|
||
return start, end
|
||
if cross_month:
|
||
start_day, start_month, end_day, end_month, year, time_text = cross_month.groups()
|
||
suffix = f" {time_text}" if time_text else ""
|
||
start = _parse_date(f"{start_day} {start_month} {year}{suffix}", source, now)
|
||
end = _parse_date(f"{end_day} {end_month} {year}{suffix}", source, now)
|
||
return start, end
|
||
return _parse_date(value, source, now), None
|
||
|
||
def _parse_date(value: str, source: dict, now: datetime) -> datetime:
|
||
value = " ".join(value.split())
|
||
tz = ZoneInfo(source.get("timezone") or "UTC")
|
||
fmt = source.get("date_format", "").strip()
|
||
if fmt:
|
||
candidates = [fmt]
|
||
# Event lists commonly omit a year. Resolve to the next occurrence.
|
||
for candidate in candidates:
|
||
has_year = "%Y" in candidate or "%y" in candidate
|
||
parse_value = value if has_year else f"{value} {now.year}"
|
||
parse_format = candidate if has_year else f"{candidate} %Y"
|
||
parsed = datetime.strptime(parse_value, parse_format).replace(tzinfo=tz)
|
||
if not has_year:
|
||
if parsed < now - timedelta(days=2):
|
||
parsed = parsed.replace(year=now.year + 1)
|
||
return parsed
|
||
if not re.search(r"\b\d{1,4}\b", value):
|
||
raise ValueError("Date text must contain a numeric day or year")
|
||
parsed = date_parser.parse(value, fuzzy=True, default=now.replace(tzinfo=None, second=0, microsecond=0), tzinfos={"INFO": None, "IPAC": None, "IMAX": None})
|
||
if parsed.tzinfo is None:
|
||
parsed = parsed.replace(tzinfo=tz)
|
||
if not now.year - 1 <= parsed.year <= now.year + 10:
|
||
raise ValueError("Parsed year is outside the plausible event range")
|
||
return parsed.replace(microsecond=0)
|
||
|
||
def _extract_json_ld(soup: BeautifulSoup, base_url: str, source: dict, now: datetime) -> list[ExtractedEvent]:
|
||
result = []
|
||
def walk(value):
|
||
if isinstance(value, list):
|
||
for item in value: yield from walk(item)
|
||
elif isinstance(value, dict):
|
||
if value.get("@type") == "Event" or "Event" in (value.get("@type") or []):
|
||
yield value
|
||
for key in ("@graph", "itemListElement"):
|
||
if key in value: yield from walk(value[key])
|
||
for script in soup.select('script[type="application/ld+json"]'):
|
||
try:
|
||
data = json.loads(script.string or script.get_text())
|
||
except (json.JSONDecodeError, TypeError):
|
||
continue
|
||
for event in walk(data):
|
||
location = event.get("location") or {}
|
||
address = location.get("address", "") if isinstance(location, dict) else ""
|
||
if isinstance(address, dict):
|
||
address = ", ".join(str(address.get(k, "")) for k in ("streetAddress", "addressLocality", "addressRegion", "postalCode") if address.get(k))
|
||
try:
|
||
start = _parse_date(str(event["startDate"]), source, now)
|
||
end = _parse_date(str(event["endDate"]), source, now) if event.get("endDate") else None
|
||
except (KeyError, ValueError, OverflowError):
|
||
continue
|
||
result.append(ExtractedEvent(title=str(event.get("name", "Untitled event")), starts_at=start, ends_at=end, description=str(event.get("description", "")), url=urljoin(base_url, str(event.get("url", base_url))), address=str(address)))
|
||
return result
|
||
|
||
|
||
def _weekly_dates(day_text: str, now: datetime, weeks: int = 12) -> list[datetime]:
|
||
names = {name.lower(): i for i, name in enumerate(("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"))}
|
||
match = next(((name, idx) for name, idx in names.items() if name in day_text.lower()), None)
|
||
if not match:
|
||
return []
|
||
delta = (match[1] - now.weekday()) % 7
|
||
first = (now + timedelta(days=delta)).replace(hour=0, minute=0, second=0, microsecond=0)
|
||
return [first + timedelta(weeks=i) for i in range(weeks)]
|
||
|
||
def _selector_for(node) -> str:
|
||
if node.get("id"):
|
||
return f"#{node['id']}"
|
||
classes = [value for value in node.get("class", []) if not value.startswith(("elementor-element-", "post-"))]
|
||
return node.name + "".join(f".{value}" for value in classes[:2])
|
||
|
||
def detect_rules(html: str, url: str, timezone: str = "Australia/Sydney") -> tuple[dict, list[ExtractedEvent], str]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
now = datetime.now(ZoneInfo(timezone))
|
||
json_source = {"url": url, "timezone": timezone, "item_selector": "json-ld", "title_selector": "", "date_selector": "", "time_selector": "", "link_selector": "", "description_selector": "", "location_selector": "", "date_format": "", "recurring_weekly": 0}
|
||
json_events = extract_events(html, json_source)
|
||
if json_events:
|
||
return json_source, json_events, "Found schema.org JSON-LD Event data."
|
||
|
||
date_pattern = re.compile(r"(?:\b(?:mon|tue|wed|thu|fri|sat|sun)(?:day)?\b|\b\d{1,2}[\s/.-](?:\d{1,2}|jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)|\b(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{1,2})", re.I)
|
||
candidates = []
|
||
container_selectors = ("article", "li", ".event", ".events", "[class*='event']", "[class*='Event']", ".event-list__item", ".item", "[class*='item']", ".wpupg-item", ".wpupg-item.wpupg-type-event", "[data-permalink]", ".wp-block-post", ".elementor-loop-item", ".tribe-events-calendar-list__event-row", ".em-event", ".mec-event-article")
|
||
for selector in container_selectors:
|
||
try: nodes = soup.select(selector)
|
||
except Exception: continue
|
||
if not 2 <= len(nodes) <= 300:
|
||
continue
|
||
nodes = [node for node in nodes if not any(marker in " ".join(str(value) for value in (node.get("class") or [])).casefold() for marker in SOCIAL_WIDGET_MARKERS)]
|
||
if len(nodes) < 2: continue
|
||
dated = [node for node in nodes if date_pattern.search(node.get_text(" ", strip=True))]
|
||
titled = [node for node in dated if node.find(["h1", "h2", "h3", "h4", "a"]) or node.select_one("[class*='title'], [itemprop='name']")]
|
||
score = len(titled) * 3 + len(dated) - max(0, len(nodes) - len(dated))
|
||
if len(titled) >= 2:
|
||
candidates.append((score, selector, titled))
|
||
if not candidates:
|
||
raise CrawlError("No repeated event-like HTML containers were detected. Try JSON-LD or configure CSS selectors manually.")
|
||
trials = []
|
||
for _, item_selector, nodes in sorted(candidates, reverse=True)[:8]:
|
||
title_selectors = []
|
||
date_selectors = []
|
||
link_selectors = []
|
||
for sample in nodes[:8]:
|
||
for node in sample.select("h1, h2, h3, h4, a, [class*='title'], [itemprop='name']"):
|
||
selector = _selector_for(node)
|
||
if selector not in title_selectors: title_selectors.append(selector)
|
||
for node in sample.find_all(["time", "span", "div", "li", "p", "ul"]):
|
||
if date_pattern.search(node.get_text(" ", strip=True)):
|
||
selector = _selector_for(node)
|
||
if selector not in date_selectors: date_selectors.append(selector)
|
||
for node in sample.find_all("a"):
|
||
selector = _selector_for(node)
|
||
if selector not in link_selectors: link_selectors.append(selector)
|
||
if any(sample.get(attribute) for attribute in ("data-permalink", "data-url", "data-href")) and ":scope" not in link_selectors:
|
||
link_selectors.insert(0, ":scope")
|
||
time_selectors = []
|
||
for sample in nodes[:8]:
|
||
for node in sample.find_all(["time", "span", "div", "p"]):
|
||
if re.search(r"\b\d{1,2}(?::\d{2})?\s*(?:am|pm)\b", node.get_text(" ", strip=True), re.IGNORECASE):
|
||
selector = _selector_for(node)
|
||
if selector not in time_selectors: time_selectors.append(selector)
|
||
for title_selector in title_selectors[:6]:
|
||
for date_selector in date_selectors[:10]:
|
||
for time_selector in [""] + time_selectors[:4]:
|
||
for recurring in (0, 1):
|
||
rules = {"url": url, "timezone": timezone, "item_selector": item_selector, "title_selector": title_selector, "date_selector": date_selector, "time_selector": time_selector, "link_selector": link_selectors[0] if link_selectors else "", "description_selector": "", "location_selector": "", "date_format": "", "recurring_weekly": recurring}
|
||
events = extract_events(html, rules)
|
||
unique_titles = len({event.title for event in events})
|
||
explicit_dates = sum(1 for event in events if event.starts_at.second == 0 and event.starts_at.microsecond == 0 and now.year - 1 <= event.starts_at.year <= now.year + 10)
|
||
if events and unique_titles >= 2 and explicit_dates == len(events):
|
||
occurrence_penalty = 12 if recurring else 1
|
||
trials.append((unique_titles * 10 + len(events) / occurrence_penalty, rules, events, len(nodes)))
|
||
if not trials:
|
||
raise CrawlError("Event-like containers were found, but no selector combination produced parseable events. Adjust selectors manually.")
|
||
_, rules, events, container_count = max(trials, key=lambda value: value[0])
|
||
winning_nodes = soup.select(rules["item_selector"])
|
||
sample = winning_nodes[0] if winning_nodes else None
|
||
if sample:
|
||
description_node = sample.find(["p", "summary"])
|
||
location_node = next((node for node in sample.find_all(["span", "div", "p"]) if re.search(r"\b(?:venue|location|address|theatre|hall|hotel|club)\b", node.get_text(" ", strip=True), re.I)), None)
|
||
rules["description_selector"] = _selector_for(description_node) if description_node else ""
|
||
rules["location_selector"] = _selector_for(location_node) if location_node else ""
|
||
return rules, events, f"Guessed CSS selectors from {container_count} repeated event-like containers."
|
||
|
||
def extract_events(html: str, source: dict) -> list[ExtractedEvent]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
tz = ZoneInfo(source.get("timezone") or "UTC")
|
||
now = datetime.now(tz)
|
||
item_selector = source.get("item_selector", "").strip()
|
||
if not item_selector or item_selector == "json-ld":
|
||
json_events = _extract_json_ld(soup, source["url"], source, now)
|
||
if json_events or item_selector == "json-ld":
|
||
return json_events
|
||
events = []
|
||
for item in soup.select(item_selector):
|
||
title = _text(item, source.get("title_selector", ""))
|
||
raw_date = _text(item, source.get("date_selector", ""))
|
||
raw_time = _text(item, source.get("time_selector", ""))
|
||
if not title or not raw_date:
|
||
continue
|
||
description = _text(item, source.get("description_selector", ""))
|
||
address = _text(item, source.get("location_selector", ""))
|
||
link = _link(item, source.get("link_selector", ""), source["url"])
|
||
if source.get("recurring_weekly"):
|
||
for start in _weekly_dates(raw_date, now):
|
||
events.append(ExtractedEvent(title=title, starts_at=start, description=description, url=link, address=address))
|
||
continue
|
||
try:
|
||
start, end = _parse_date_range(f"{raw_date} {raw_time}".strip(), source, now)
|
||
except (ValueError, OverflowError):
|
||
continue
|
||
events.append(ExtractedEvent(title=title, starts_at=start, ends_at=end, description=description, url=link, address=address))
|
||
return events
|
||
|
||
def infer_page_fields(html: str, url: str) -> dict:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
title = ""
|
||
og_site = soup.select_one('meta[property="og:site_name"]')
|
||
if og_site:
|
||
content = og_site.get("content")
|
||
title = str(content).strip() if content else ""
|
||
if not title and soup.title:
|
||
title = re.split(r"[|–—-]", soup.title.get_text(" ", strip=True))[0].strip()
|
||
heading = soup.find("h1")
|
||
source_name = heading.get_text(" ", strip=True) if heading else title or urlparse(url).hostname or "Event source"
|
||
text = soup.get_text(" ", strip=True)
|
||
address_match = re.search(r"\b\d{1,5}[A-Za-z/-]*\s+[A-Za-z][A-Za-z .'-]+(?:Street|St|Road|Rd|Lane|Ln|Avenue|Ave|Drive|Dr|Way|Place|Pl),?\s+[A-Za-z ]+\s+(?:NSW|VIC|QLD|SA|WA|TAS|NT|ACT)\s+\d{4}\b", text, re.I)
|
||
address = address_match.group(0).strip() if address_match else ""
|
||
return {"name": source_name, "venue_name": title or source_name, "venue_address": address}
|
||
|
||
def external_key(event: ExtractedEvent) -> str:
|
||
value = f"{event.title.casefold()}|{event.starts_at.isoformat()}|{event.address.casefold()}"
|
||
return hashlib.sha256(value.encode()).hexdigest()
|