165 lines
8.1 KiB
Python
165 lines
8.1 KiB
Python
import json
|
|
from copy import deepcopy
|
|
from datetime import datetime, timedelta
|
|
from urllib.parse import urljoin
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from .crawler import BROWSER_HEADERS, CrawlError, ExtractedEvent, _extract_json_ld, _parse_date_range, _weekly_dates, validate_public_url
|
|
|
|
|
|
def parse_strategy(value) -> dict:
|
|
if isinstance(value, dict):
|
|
return deepcopy(value)
|
|
try:
|
|
return json.loads(value or "{}")
|
|
except (TypeError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def json_path(value, path: str, default=None):
|
|
if not path:
|
|
return value
|
|
current = value
|
|
for component in path.split("."):
|
|
if current is None:
|
|
return default
|
|
if isinstance(current, list):
|
|
try: current = current[int(component)]
|
|
except (ValueError, IndexError): return default
|
|
elif isinstance(current, dict):
|
|
current = current.get(component, default)
|
|
else:
|
|
return default
|
|
return current
|
|
|
|
|
|
def _value(node, selector: str, attribute: str = "") -> str:
|
|
if not selector:
|
|
return ""
|
|
target = node if selector == ":scope" else node.select_one(selector)
|
|
if not target:
|
|
return ""
|
|
if attribute:
|
|
return str(target.get(attribute, ""))
|
|
return target.get_text(" ", strip=True)
|
|
|
|
|
|
def _request(client, strategy: dict, page: int | None = None):
|
|
url = str(strategy.get("request_url") or strategy.get("url") or "")
|
|
validate_public_url(url)
|
|
method = strategy.get("method", "GET").upper()
|
|
params = deepcopy(strategy.get("params") or {})
|
|
body = deepcopy(strategy.get("body") or {})
|
|
page_parameter = strategy.get("page_parameter", "")
|
|
if page is not None and page_parameter:
|
|
(params if method == "GET" else body)[page_parameter] = str(page)
|
|
headers = {**BROWSER_HEADERS, **(strategy.get("headers") or {})}
|
|
response = client.request(method, url, params=params, data=body if method != "GET" else None, headers=headers)
|
|
response.raise_for_status()
|
|
if len(response.content) > int(strategy.get("max_bytes", 5_000_000)):
|
|
raise CrawlError("Strategy response exceeds configured size limit.")
|
|
return response
|
|
|
|
|
|
def _events_from_html(html: str, strategy: dict, source: dict) -> list[ExtractedEvent]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
now = datetime.now(ZoneInfo(source.get("timezone") or "UTC"))
|
|
if strategy.get("item_selector") == "json-ld":
|
|
return _extract_json_ld(soup, source["url"], source, now)
|
|
events = []
|
|
for item in soup.select(strategy.get("item_selector", "")):
|
|
title = _value(item, strategy.get("title_selector", ""), strategy.get("title_attribute", ""))
|
|
raw_date = _value(item, strategy.get("date_selector", ""), strategy.get("date_attribute", ""))
|
|
for old, new in (strategy.get("date_replacements") or {}).items(): raw_date = raw_date.replace(old, new)
|
|
raw_time = _value(item, strategy.get("time_selector", ""), strategy.get("time_attribute", ""))
|
|
if not title or not raw_date: continue
|
|
config = {**source, "date_format": strategy.get("date_format", "")}
|
|
if strategy.get("recurring_weekly"):
|
|
for start in _weekly_dates(raw_date, now):
|
|
events.append(ExtractedEvent(title=title, starts_at=start, description=_value(item, strategy.get("description_selector", ""), strategy.get("description_attribute", "")), url=urljoin(source["url"], _value(item, strategy.get("link_selector", ""), strategy.get("link_attribute", "href"))), address=_value(item, strategy.get("location_selector", ""), strategy.get("location_attribute", ""))))
|
|
continue
|
|
try: start, end = _parse_date_range(f"{raw_date} {raw_time}".strip(), config, now)
|
|
except (ValueError, OverflowError): continue
|
|
duration = strategy.get("duration_minutes")
|
|
if duration and not end: end = start + timedelta(minutes=int(duration))
|
|
link = _value(item, strategy.get("link_selector", ""), strategy.get("link_attribute", "href"))
|
|
description = _value(item, strategy.get("description_selector", ""), strategy.get("description_attribute", ""))
|
|
location = _value(item, strategy.get("location_selector", ""), strategy.get("location_attribute", ""))
|
|
events.append(ExtractedEvent(title=title, starts_at=start, ends_at=end, description=description, url=urljoin(source["url"], link), address=location))
|
|
return events
|
|
|
|
|
|
def _index_join(items: list, join_items: list, item_key: str, join_key: str) -> dict:
|
|
index = {}
|
|
for item in join_items:
|
|
keys = json_path(item, join_key, [])
|
|
if not isinstance(keys, list): keys = [keys]
|
|
for key in keys:
|
|
if key is not None: index[str(key)] = item
|
|
return index
|
|
|
|
|
|
def _events_from_json(payload, strategy: dict, source: dict, join_index: dict | None = None) -> list[ExtractedEvent]:
|
|
items = json_path(payload, strategy.get("items_path", ""), [])
|
|
if not isinstance(items, list): return []
|
|
now = datetime.now(ZoneInfo(source.get("timezone") or "UTC"))
|
|
events = []
|
|
for item in items:
|
|
if strategy.get("disabled_path") and json_path(item, strategy["disabled_path"]): continue
|
|
joined = join_index.get(str(json_path(item, strategy.get("join_item_key", "")))) if join_index else None
|
|
def field(name, default=""):
|
|
item_path = strategy.get(f"{name}_path", "")
|
|
join_path = strategy.get(f"join_{name}_path", "")
|
|
value = json_path(item, item_path, None) if item_path else None
|
|
if (value is None or value == "") and joined is not None and join_path:
|
|
value = json_path(joined, join_path, default)
|
|
return value if value is not None else default
|
|
title = str(field("title") or strategy.get("title_fallback", "Untitled event"))
|
|
date_value = str(field("date"))
|
|
if not date_value: continue
|
|
try: start, end = _parse_date_range(date_value, source, now)
|
|
except (ValueError, OverflowError): continue
|
|
end_value = field("end")
|
|
if end_value:
|
|
try: end, _ = _parse_date_range(str(end_value), source, now)
|
|
except (ValueError, OverflowError): pass
|
|
duration = field("duration") or strategy.get("duration_minutes")
|
|
if duration and not end: end = start + timedelta(minutes=int(duration))
|
|
events.append(ExtractedEvent(title=title, starts_at=start, ends_at=end, description=str(field("description")), url=urljoin(source["url"], str(field("link"))), address=str(field("location"))))
|
|
return events
|
|
|
|
|
|
def execute_strategy(source: dict, client) -> list[ExtractedEvent]:
|
|
strategy = parse_strategy(source.get("strategy_json"))
|
|
mode = strategy.get("mode", "html")
|
|
if mode == "html":
|
|
response = _request(client, strategy)
|
|
return _events_from_html(response.text, strategy, source)
|
|
if mode not in {"json", "html_fragment"}:
|
|
raise CrawlError(f"Unsupported extraction mode: {mode}")
|
|
join_index = None
|
|
join = strategy.get("join")
|
|
if join:
|
|
join_response = _request(client, join)
|
|
join_payload = join_response.json()
|
|
join_items = json_path(join_payload, join.get("items_path", ""), [])
|
|
join_index = _index_join([], join_items if isinstance(join_items, list) else [], strategy.get("join_item_key", ""), join.get("key_path", ""))
|
|
events = []
|
|
page = int(strategy.get("page_start", 1))
|
|
max_pages = min(int(strategy.get("max_pages", 1)), 20)
|
|
for _ in range(max_pages):
|
|
response = _request(client, strategy, page)
|
|
payload = response.json()
|
|
if mode == "html_fragment":
|
|
fragment = json_path(payload, strategy.get("html_path", "html"), "")
|
|
page_events = _events_from_html(str(fragment), strategy, source)
|
|
else:
|
|
page_events = _events_from_json(payload, strategy, source, join_index)
|
|
events.extend(page_events)
|
|
total_pages = json_path(payload, strategy.get("total_pages_path", ""), None)
|
|
if not strategy.get("page_parameter") or not page_events or (total_pages is not None and page >= int(total_pages)): break
|
|
page += 1
|
|
return events
|