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: parsed = datetime.strptime(value, candidate).replace(tzinfo=tz) if "%Y" not in candidate and "%y" not in candidate: parsed = parsed.replace(year=now.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), 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()