|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from typing import TYPE_CHECKING, NamedTuple, TypedDict |
| 4 | +from urllib.parse import urlencode, urlsplit |
| 5 | + |
| 6 | +from pydantic import ValidationError |
| 7 | + |
| 8 | +from crawlee._request import Request |
| 9 | +from crawlee._types import HttpHeaders |
| 10 | +from crawlee._utils.crypto import compute_short_hash |
| 11 | +from crawlee._utils.html import resolve_encoding |
| 12 | +from crawlee._utils.urls import convert_to_absolute_url, validate_http_url |
| 13 | + |
| 14 | +if TYPE_CHECKING: |
| 15 | + from collections.abc import Mapping, Sequence |
| 16 | + |
| 17 | + from lxml.html import HtmlElement |
| 18 | + from typing_extensions import NotRequired, Unpack |
| 19 | + |
| 20 | + from crawlee._types import EnqueueStrategy, JsonSerializable |
| 21 | + |
| 22 | +_FIELD_TAGS = ('input', 'button', 'select', 'textarea') |
| 23 | +_BUTTON_INPUT_TYPES = ('submit', 'image', 'reset', 'button') |
| 24 | + |
| 25 | + |
| 26 | +class FormRequestOptions(TypedDict): |
| 27 | + """Options for the `Request` created from a form. |
| 28 | +
|
| 29 | + Mirrors `RequestOptions` without the URL, method and payload, which come from the form, and without `id` and |
| 30 | + `unique_key`, which can't be shared by several forms. |
| 31 | + """ |
| 32 | + |
| 33 | + label: NotRequired[str | None] |
| 34 | + """A label routing the request to a specific handler.""" |
| 35 | + |
| 36 | + headers: NotRequired[HttpHeaders | dict[str, str] | None] |
| 37 | + """HTTP headers of the request. A `Content-Type` set here replaces the one of a POST form.""" |
| 38 | + |
| 39 | + session_id: NotRequired[str | None] |
| 40 | + """ID of the `Session` the request is bound to.""" |
| 41 | + |
| 42 | + keep_url_fragment: NotRequired[bool] |
| 43 | + """Whether the URL fragment counts towards the unique key of the request.""" |
| 44 | + |
| 45 | + use_extended_unique_key: NotRequired[bool] |
| 46 | + """Whether the method and payload count towards the unique key. Defaults to `True` for POST forms.""" |
| 47 | + |
| 48 | + always_enqueue: NotRequired[bool] |
| 49 | + """Whether to enqueue the request even if it's already in the queue.""" |
| 50 | + |
| 51 | + user_data: NotRequired[Mapping[str, JsonSerializable]] |
| 52 | + """Custom data stored with the request.""" |
| 53 | + |
| 54 | + no_retry: NotRequired[bool] |
| 55 | + """Whether to skip retrying the request if it fails.""" |
| 56 | + |
| 57 | + enqueue_strategy: NotRequired[EnqueueStrategy] |
| 58 | + """The strategy used when enqueuing the request.""" |
| 59 | + |
| 60 | + max_retries: NotRequired[int | None] |
| 61 | + """The maximum number of retries of the request.""" |
| 62 | + |
| 63 | + |
| 64 | +def forms_to_requests( |
| 65 | + forms: Sequence[HtmlElement], |
| 66 | + page_url: str, |
| 67 | + page_encoding: str, |
| 68 | + *, |
| 69 | + fields: Mapping[str, str | Sequence[str] | None] | None = None, |
| 70 | + click: bool | Mapping[str, str] = True, |
| 71 | + **kwargs: Unpack[FormRequestOptions], |
| 72 | +) -> list[Request]: |
| 73 | + """Create a `Request` submitting each form the way a browser does. |
| 74 | +
|
| 75 | + Dialog forms and forms not submitting over HTTP(S) are skipped. POST forms send their `Content-Type` and use |
| 76 | + an extended unique key by default. |
| 77 | +
|
| 78 | + Args: |
| 79 | + forms: The form elements, each within the lxml tree of the whole page. |
| 80 | + page_url: The URL of the page. |
| 81 | + page_encoding: The Python codec the page was decoded with, which forms submit in by default. |
| 82 | + fields: Field values replacing those in the form or adding new fields. A `None` value drops the field and |
| 83 | + a sequence submits the field once per value. |
| 84 | + click: The submit button to click: `True` for the first one, `False` for none, or a mapping of attributes |
| 85 | + the button must have, in which case forms without such a button are skipped. |
| 86 | + **kwargs: Additional options passed to `Request.from_url`. |
| 87 | + """ |
| 88 | + if not forms: |
| 89 | + return [] |
| 90 | + |
| 91 | + root = forms[0].getroottree().getroot() |
| 92 | + base = root.find('.//base[@href]') |
| 93 | + try: |
| 94 | + base_url = convert_to_absolute_url(page_url, '' if base is None else base.get('href')) |
| 95 | + except ValueError: |
| 96 | + base_url = page_url |
| 97 | + |
| 98 | + elements_by_form = _elements_by_form(root) |
| 99 | + |
| 100 | + requests = [] |
| 101 | + for form in forms: |
| 102 | + request = _form_to_request( |
| 103 | + form, |
| 104 | + elements_by_form.get(form, []), |
| 105 | + page_url=page_url, |
| 106 | + base_url=base_url, |
| 107 | + page_encoding=page_encoding, |
| 108 | + fields=fields, |
| 109 | + click=click, |
| 110 | + **kwargs, |
| 111 | + ) |
| 112 | + if request is not None: |
| 113 | + requests.append(request) |
| 114 | + return requests |
| 115 | + |
| 116 | + |
| 117 | +class _Field(NamedTuple): |
| 118 | + """A single entry the form submits.""" |
| 119 | + |
| 120 | + name: str |
| 121 | + value: str |
| 122 | + is_file: bool = False |
| 123 | + |
| 124 | + |
| 125 | +def _elements_by_form(root: HtmlElement) -> dict[HtmlElement, list[HtmlElement]]: |
| 126 | + """Group the fields and buttons of the page by the form they belong to, in document order.""" |
| 127 | + elements = list(root.iter(*_FIELD_TAGS)) |
| 128 | + |
| 129 | + # The first element with a given ID wins, as in `getElementById`. Only the `form` attribute needs them. |
| 130 | + elements_by_id: dict[str, HtmlElement] = {} |
| 131 | + if any(element.get('form') is not None for element in elements): |
| 132 | + for element in root.xpath('//*[@id]'): |
| 133 | + elements_by_id.setdefault(element.get('id'), element) |
| 134 | + |
| 135 | + elements_by_form: dict[HtmlElement, list[HtmlElement]] = {} |
| 136 | + for element in elements: |
| 137 | + owner = _form_owner(element, elements_by_id) |
| 138 | + if owner is not None: |
| 139 | + elements_by_form.setdefault(owner, []).append(element) |
| 140 | + return elements_by_form |
| 141 | + |
| 142 | + |
| 143 | +def _form_owner(element: HtmlElement, elements_by_id: Mapping[str, HtmlElement]) -> HtmlElement | None: |
| 144 | + """Get the form an element belongs to, honouring its `form` attribute.""" |
| 145 | + form_id = element.get('form') |
| 146 | + if form_id is None: |
| 147 | + return next(element.iterancestors('form'), None) |
| 148 | + |
| 149 | + owner = elements_by_id.get(form_id) |
| 150 | + if owner is None or owner.tag != 'form': |
| 151 | + return None |
| 152 | + return owner |
| 153 | + |
| 154 | + |
| 155 | +def _form_to_request( |
| 156 | + form: HtmlElement, |
| 157 | + elements: list[HtmlElement], |
| 158 | + *, |
| 159 | + page_url: str, |
| 160 | + base_url: str, |
| 161 | + page_encoding: str, |
| 162 | + fields: Mapping[str, str | Sequence[str] | None] | None, |
| 163 | + click: bool | Mapping[str, str], |
| 164 | + **kwargs: Unpack[FormRequestOptions], |
| 165 | +) -> Request | None: |
| 166 | + """Create a `Request` submitting a single form, or `None` if a browser wouldn't send one.""" |
| 167 | + button = None |
| 168 | + if click is not False: |
| 169 | + button = _find_clickable(elements, {} if click is True else click) |
| 170 | + if click is not True and button is None: |
| 171 | + return None |
| 172 | + |
| 173 | + method = (_submission_attribute(form, button, 'method') or 'get').upper() |
| 174 | + enctype = _submission_attribute(form, button, 'enctype').lower() |
| 175 | + action = _submission_attribute(form, button, 'action').strip() |
| 176 | + |
| 177 | + # A dialog form only closes its `<dialog>` on the client. |
| 178 | + if method == 'DIALOG': |
| 179 | + return None |
| 180 | + |
| 181 | + url = _resolve_action(page_url, base_url, action) |
| 182 | + if url is None: |
| 183 | + return None |
| 184 | + |
| 185 | + entries = _collect_fields(elements, button, fields) |
| 186 | + encoding = _form_encoding(form, page_encoding) |
| 187 | + |
| 188 | + if method != 'POST': |
| 189 | + query = urlencode( |
| 190 | + [(entry.name, entry.value) for entry in entries], encoding=encoding, errors='xmlcharrefreplace' |
| 191 | + ) |
| 192 | + return Request.from_url(urlsplit(url)._replace(query=query).geturl(), method='GET', **kwargs) |
| 193 | + |
| 194 | + payload, content_type = _encode_body(entries, enctype, encoding) |
| 195 | + # The `Content-Type` passed in `headers` wins over the form's. |
| 196 | + kwargs['headers'] = HttpHeaders({'Content-Type': content_type}) | HttpHeaders(kwargs.get('headers') or {}) |
| 197 | + kwargs.setdefault('use_extended_unique_key', True) |
| 198 | + return Request.from_url(url, method='POST', payload=payload, **kwargs) |
| 199 | + |
| 200 | + |
| 201 | +def _find_clickable(elements: list[HtmlElement], attributes: Mapping[str, str]) -> HtmlElement | None: |
| 202 | + """Find the first enabled submit button having all the given attributes.""" |
| 203 | + for element in elements: |
| 204 | + if not _is_submit_button(element) or _is_disabled(element): |
| 205 | + continue |
| 206 | + if all(element.get(key) == value for key, value in attributes.items()): |
| 207 | + return element |
| 208 | + return None |
| 209 | + |
| 210 | + |
| 211 | +def _is_submit_button(element: HtmlElement) -> bool: |
| 212 | + """Check whether the element submits the form when clicked.""" |
| 213 | + button_type = element.get('type', '').lower() |
| 214 | + if element.tag == 'button': |
| 215 | + return button_type in ('', 'submit') |
| 216 | + if element.tag == 'input': |
| 217 | + return button_type in ('submit', 'image') |
| 218 | + return False |
| 219 | + |
| 220 | + |
| 221 | +def _is_disabled(element: HtmlElement) -> bool: |
| 222 | + """Check whether the element, or a `<fieldset>` containing it, is disabled.""" |
| 223 | + if 'disabled' in element.attrib: |
| 224 | + return True |
| 225 | + return any('disabled' in fieldset.attrib for fieldset in element.iterancestors('fieldset')) |
| 226 | + |
| 227 | + |
| 228 | +def _submission_attribute(form: HtmlElement, button: HtmlElement | None, name: str) -> str: |
| 229 | + """Get a form attribute like `action`, which the clicked button can override with its `form*` counterpart.""" |
| 230 | + if button is not None: |
| 231 | + button_value = button.get(f'form{name}') |
| 232 | + if button_value: |
| 233 | + return button_value |
| 234 | + return form.get(name) or '' |
| 235 | + |
| 236 | + |
| 237 | +def _resolve_action(page_url: str, base_url: str, action: str) -> str | None: |
| 238 | + """Resolve the form action to an absolute URL, or `None` if it isn't a valid HTTP(S) URL.""" |
| 239 | + if not action: |
| 240 | + return page_url |
| 241 | + |
| 242 | + try: |
| 243 | + return validate_http_url(convert_to_absolute_url(base_url, action)) |
| 244 | + except (ValueError, ValidationError): |
| 245 | + return None |
| 246 | + |
| 247 | + |
| 248 | +def _collect_fields( |
| 249 | + elements: list[HtmlElement], |
| 250 | + button: HtmlElement | None, |
| 251 | + fields: Mapping[str, str | Sequence[str] | None] | None, |
| 252 | +) -> list[_Field]: |
| 253 | + """Collect the entries the form submits: its fields, the clicked button and the `fields` overrides.""" |
| 254 | + # Only one radio button of a group is checked, the last one in the markup. |
| 255 | + checked_radios = {element.get('name'): element for element in elements if _is_radio(element) and element.checked} |
| 256 | + |
| 257 | + entries: list[_Field] = [] |
| 258 | + for element in elements: |
| 259 | + if element is button: |
| 260 | + entries.extend(_button_fields(button)) |
| 261 | + continue |
| 262 | + if not element.get('name') or _is_disabled(element): |
| 263 | + continue |
| 264 | + if _is_radio(element) and checked_radios.get(element.get('name')) is not element: |
| 265 | + continue |
| 266 | + entries.extend(_element_fields(element)) |
| 267 | + |
| 268 | + return _apply_fields(entries, fields or {}) |
| 269 | + |
| 270 | + |
| 271 | +def _is_radio(element: HtmlElement) -> bool: |
| 272 | + """Check whether the element is a radio button.""" |
| 273 | + return element.tag == 'input' and element.type == 'radio' |
| 274 | + |
| 275 | + |
| 276 | +def _button_fields(button: HtmlElement) -> list[_Field]: |
| 277 | + """Get the entries the clicked button adds.""" |
| 278 | + name = button.get('name') |
| 279 | + |
| 280 | + # An image button sends the click coordinates instead of its value. |
| 281 | + if button.get('type', '').lower() == 'image': |
| 282 | + prefix = f'{name}.' if name else '' |
| 283 | + return [_Field(f'{prefix}x', '0'), _Field(f'{prefix}y', '0')] |
| 284 | + |
| 285 | + if name: |
| 286 | + return [_Field(name, button.get('value', ''))] |
| 287 | + return [] |
| 288 | + |
| 289 | + |
| 290 | +def _element_fields(element: HtmlElement) -> list[_Field]: |
| 291 | + """Get the entries a single enabled, named field submits.""" |
| 292 | + name = element.get('name') |
| 293 | + |
| 294 | + if element.tag == 'select': |
| 295 | + values = element.value if element.multiple else [element.value] |
| 296 | + return [_Field(name, value) for value in values if value is not None] |
| 297 | + |
| 298 | + if element.tag == 'textarea': |
| 299 | + # Browsers drop the newline right after `<textarea>`, lxml keeps it. |
| 300 | + return [_Field(name, element.value.removeprefix('\n'))] |
| 301 | + |
| 302 | + # Buttons submit only when clicked, see `_button_fields`. |
| 303 | + if element.tag != 'input' or element.type in _BUTTON_INPUT_TYPES: |
| 304 | + return [] |
| 305 | + |
| 306 | + if element.type == 'file': |
| 307 | + return [_Field(name, '', is_file=True)] |
| 308 | + |
| 309 | + if element.checkable and not element.checked: |
| 310 | + return [] |
| 311 | + |
| 312 | + return [_Field(name, element.value or '')] |
| 313 | + |
| 314 | + |
| 315 | +def _apply_fields(entries: list[_Field], fields: Mapping[str, str | Sequence[str] | None]) -> list[_Field]: |
| 316 | + """Replace the entries named in `fields` with its values, dropping those set to `None`.""" |
| 317 | + result = [entry for entry in entries if entry.name not in fields] |
| 318 | + for name, value in fields.items(): |
| 319 | + if value is None: |
| 320 | + continue |
| 321 | + values = [value] if isinstance(value, str) else value |
| 322 | + result.extend(_Field(name, item) for item in values) |
| 323 | + return result |
| 324 | + |
| 325 | + |
| 326 | +def _form_encoding(form: HtmlElement, page_encoding: str) -> str: |
| 327 | + """Pick the encoding a browser submits the form in: the first known `accept-charset` label, else the page one.""" |
| 328 | + labels = (form.get('accept-charset') or '').replace(',', ' ').split() |
| 329 | + encoding = next(filter(None, map(resolve_encoding, labels)), page_encoding) |
| 330 | + # Browsers never submit in UTF-16, as it isn't ASCII-compatible. |
| 331 | + return 'utf-8' if encoding.startswith('utf-16') else encoding |
| 332 | + |
| 333 | + |
| 334 | +def _encode_body(entries: list[_Field], enctype: str, encoding: str) -> tuple[bytes, str]: |
| 335 | + """Encode the fields as a POST body, returning it with its `Content-Type` header value.""" |
| 336 | + if enctype == 'multipart/form-data': |
| 337 | + return _encode_multipart(entries, encoding) |
| 338 | + |
| 339 | + if enctype == 'text/plain': |
| 340 | + text = ''.join(f'{entry.name}={entry.value}\r\n' for entry in entries) |
| 341 | + return text.encode(encoding, 'xmlcharrefreplace'), 'text/plain' |
| 342 | + |
| 343 | + pairs = [(entry.name, entry.value) for entry in entries] |
| 344 | + body = urlencode(pairs, encoding=encoding, errors='xmlcharrefreplace').encode() |
| 345 | + return body, 'application/x-www-form-urlencoded' |
| 346 | + |
| 347 | + |
| 348 | +def _encode_multipart(entries: list[_Field], encoding: str) -> tuple[bytes, str]: |
| 349 | + """Encode fields as `multipart/form-data`, returning the body and the `Content-Type` header value.""" |
| 350 | + parts: list[bytes] = [] |
| 351 | + for entry in entries: |
| 352 | + disposition = f'form-data; name="{entry.name}"' |
| 353 | + if entry.is_file: |
| 354 | + head = f'Content-Disposition: {disposition}; filename=""\r\nContent-Type: application/octet-stream\r\n\r\n' |
| 355 | + else: |
| 356 | + head = f'Content-Disposition: {disposition}\r\n\r\n' |
| 357 | + parts.append((head + entry.value).encode(encoding, 'xmlcharrefreplace')) |
| 358 | + |
| 359 | + # A random boundary would change the payload on every call and break deduplication. |
| 360 | + boundary = f'----CrawleeFormBoundary{compute_short_hash(b"".join(parts), length=16)}' |
| 361 | + delimiter = f'--{boundary}\r\n'.encode() |
| 362 | + body = b''.join(delimiter + part + b'\r\n' for part in parts) + f'--{boundary}--\r\n'.encode() |
| 363 | + return body, f'multipart/form-data; boundary={boundary}' |
0 commit comments