diff --git a/src/crawlee/_utils/forms.py b/src/crawlee/_utils/forms.py new file mode 100644 index 0000000000..3a0d569df0 --- /dev/null +++ b/src/crawlee/_utils/forms.py @@ -0,0 +1,451 @@ +from __future__ import annotations + +import codecs +import re +from contextlib import suppress +from dataclasses import dataclass +from typing import TYPE_CHECKING, NamedTuple, TypedDict +from urllib.parse import urlencode + +from lxml.html import HTMLParser, document_fromstring +from yarl import URL + +from crawlee._request import Request +from crawlee._types import HttpHeaders +from crawlee._utils.crypto import compute_short_hash +from crawlee._utils.urls import convert_to_absolute_url + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping, Sequence + + from lxml.html import HtmlElement + from typing_extensions import NotRequired, Unpack + + from crawlee._types import EnqueueStrategy, HttpMethod, JsonSerializable + +_CHARSET_PATTERN = re.compile(r'charset\s*=\s*["\']?([^"\'\s;]+)', re.IGNORECASE) + +# Legacy charsets browsers replace with a superset, per the WHATWG Encoding Standard. +_ENCODING_SUPERSETS = { + 'ascii': 'cp1252', + 'iso8859-1': 'cp1252', + 'iso8859-9': 'cp1254', + 'iso8859-11': 'cp874', + 'tis-620': 'cp874', + 'gb2312': 'gbk', + 'shift_jis': 'cp932', + 'euc_kr': 'cp949', +} + +_FIELD_TAGS = ('input', 'button', 'select', 'textarea') +_BUTTON_INPUT_TYPES = ('submit', 'image', 'reset', 'button') + +# Browsers percent-encode these characters in multipart field names. +_MULTIPART_NAME_ESCAPES = str.maketrans({'"': '%22', '\r': '%0D', '\n': '%0A'}) + + +class FormRequestOptions(TypedDict): + """Options for the `Request` created from a form, other than the URL, method and payload taken from the form.""" + + label: NotRequired[str | None] + session_id: NotRequired[str | None] + unique_key: NotRequired[str | None] + keep_url_fragment: NotRequired[bool] + use_extended_unique_key: NotRequired[bool] + always_enqueue: NotRequired[bool] + user_data: NotRequired[Mapping[str, JsonSerializable]] + no_retry: NotRequired[bool] + enqueue_strategy: NotRequired[EnqueueStrategy] + max_retries: NotRequired[int | None] + + +def forms_to_requests( + forms: Iterable[HtmlElement], + page_url: str, + content_type: str | None, + *, + form_data: Mapping[str, str | Sequence[str] | None] | None = None, + click_data: Mapping[str, str] | None = None, + dont_click: bool = False, + headers: HttpHeaders | dict[str, str] | None = None, + **kwargs: Unpack[FormRequestOptions], +) -> list[Request]: + """Create a `Request` submitting each form the way a browser does. + + Forms with no submit button matching `click_data`, dialog forms and forms not submitting over HTTP(S) are skipped. + + Args: + forms: The form elements, each within the lxml tree of the whole page. + page_url: The URL of the page. + content_type: The `Content-Type` header of the page response, used to find the page encoding. + form_data: Field values overriding those in the form. A `None` value drops the field. + click_data: Attributes identifying the submit button to click. Defaults to the first one. + dont_click: Submit the form without clicking any button. + headers: The HTTP headers of the request. The `Content-Type` of the form is added to them. + **kwargs: Additional options passed to `Request.from_url`. + """ + forms = list(forms) + if not forms: + return [] + + page = _analyze_page(forms[0].getroottree().getroot(), page_url, content_type) + + requests = [] + for form in forms: + request = _form_to_request( + form, + page, + form_data=form_data, + click_data=click_data, + dont_click=dont_click, + headers=headers, + **kwargs, + ) + if request is not None: + requests.append(request) + return requests + + +def parse_html(body: bytes, encoding: str | None) -> HtmlElement: + """Parse a page with lxml, decoding it with the given encoding, or the one lxml detects if it's `None`.""" + try: + parser = HTMLParser(encoding=encoding) + except LookupError: + # A codec Python knows but libxml2 doesn't, so let lxml detect the encoding itself. + parser = HTMLParser() + return document_fromstring(body, parser=parser) + + +def response_charset(content_type: str | None) -> str | None: + """Get the charset from a `Content-Type` header, if it names a known one.""" + charset = _find_charset(content_type) + if charset is None: + return None + with suppress(LookupError): + return codecs.lookup(charset).name + return None + + +class _Field(NamedTuple): + """A single entry the form submits.""" + + name: str + value: str + is_file: bool = False + + +@dataclass +class _Page: + """The parts of a page shared by all its forms, computed in a single pass over the document.""" + + url: str + base_url: str + charsets: list[str] + """The page encoding candidates, from the `Content-Type` header and then the `` tags.""" + elements_by_form: dict[HtmlElement, list[HtmlElement]] + """The fields and buttons belonging to each form, in document order.""" + + +def _analyze_page(root: HtmlElement, page_url: str, content_type: str | None) -> _Page: + """Resolve the base URL, encoding candidates and form owners of all fields on the page.""" + try: + base_url = convert_to_absolute_url(page_url, root.xpath('string(//base[@href][1]/@href)')) + except ValueError: + base_url = page_url + + # The first element with a given ID wins, as in `getElementById`. + elements_by_id: dict[str, HtmlElement] = {} + for element in root.xpath('//*[@id]'): + elements_by_id.setdefault(element.get('id'), element) + + elements_by_form: dict[HtmlElement, list[HtmlElement]] = {} + for element in root.iter(*_FIELD_TAGS): + owner = _form_owner(element, elements_by_id) + if owner is not None: + elements_by_form.setdefault(owner, []).append(element) + + return _Page( + url=page_url, + base_url=base_url, + charsets=_page_charsets(root, content_type), + elements_by_form=elements_by_form, + ) + + +def _page_charsets(root: HtmlElement, content_type: str | None) -> list[str]: + """Collect the charsets the page declares, in the order browsers trust them.""" + charsets: list[str] = [] + + header_charset = _find_charset(content_type) + if header_charset is not None: + charsets.append(header_charset) + + charsets.extend(root.xpath('//meta[@charset]/@charset')) + + http_equiv_contents = root.xpath( + '//meta[translate(@http-equiv, "CONTENT-TYP", "content-typ")="content-type"]/@content' + ) + for content in http_equiv_contents: + meta_charset = _find_charset(content) + if meta_charset is not None: + charsets.append(meta_charset) + + return charsets + + +def _find_charset(value: str | None) -> str | None: + """Find the `charset=` parameter in a `Content-Type` value.""" + if not value: + return None + match = _CHARSET_PATTERN.search(value) + return match.group(1) if match else None + + +def _form_to_request( + form: HtmlElement, + page: _Page, + *, + form_data: Mapping[str, str | Sequence[str] | None] | None, + click_data: Mapping[str, str] | None, + dont_click: bool, + headers: HttpHeaders | dict[str, str] | None, + **kwargs: Unpack[FormRequestOptions], +) -> Request | None: + """Create a `Request` submitting a single form, or `None` if a browser wouldn't send one.""" + elements = page.elements_by_form.get(form, []) + + if dont_click: + button = None + else: + button = _find_clickable(elements, click_data) + if click_data and button is None: + return None + + method = (_submission_attribute(form, button, 'method') or 'get').upper() + enctype = _submission_attribute(form, button, 'enctype').lower() + action = _submission_attribute(form, button, 'action').strip() + + # A dialog form only closes its `` on the client. + if method == 'DIALOG': + return None + + url = _resolve_action(page, action) + if url is None: + return None + + fields = _collect_fields(elements, button, form_data) + encoding = _form_encoding(form, page.charsets) + request_headers = HttpHeaders(headers or {}) + + if method != 'POST': + get_url = _url_with_query(url, fields, encoding) + return Request.from_url(get_url, method='GET', headers=request_headers, **kwargs) + + payload, content_type = _encode_body(fields, enctype, encoding) + request_method: HttpMethod = 'POST' + kwargs.setdefault('use_extended_unique_key', True) + return Request.from_url( + url, + method=request_method, + headers=HttpHeaders({'Content-Type': content_type}) | request_headers, + payload=payload, + **kwargs, + ) + + +def _submission_attribute(form: HtmlElement, button: HtmlElement | None, name: str) -> str: + """Get a form attribute like `action`, which the clicked button can override with its `form*` counterpart.""" + if button is not None: + button_value = button.get(f'form{name}') + if button_value: + return button_value + return form.get(name) or '' + + +def _resolve_action(page: _Page, action: str) -> str | None: + """Resolve the form action to an absolute URL, or `None` if it can't be submitted over HTTP(S).""" + if not action: + return page.url + + try: + url = convert_to_absolute_url(page.base_url, action) + except ValueError: + return None + + if URL(url).scheme not in ('http', 'https'): + return None + return url + + +def _url_with_query(url: str, fields: list[_Field], encoding: str) -> str: + """Replace the query of the URL with the fields, as a GET form does.""" + query = urlencode([(field.name, field.value) for field in fields], encoding=encoding, errors='xmlcharrefreplace') + parsed = URL(url) + # Build the URL from raw parts, as `with_query` would encode the query again as UTF-8. + return str( + URL.build( + scheme=parsed.scheme, + authority=parsed.raw_authority, + path=parsed.raw_path, + query_string=query, + fragment=parsed.raw_fragment, + encoded=True, + ) + ) + + +def _encode_body(fields: list[_Field], enctype: str, encoding: str) -> tuple[bytes, str]: + """Encode the fields as a POST body, returning it with its `Content-Type` header value.""" + if enctype == 'multipart/form-data': + return _encode_multipart(fields, encoding) + + if enctype == 'text/plain': + text = ''.join(f'{field.name}={field.value}\r\n' for field in fields) + return text.encode(encoding, 'xmlcharrefreplace'), 'text/plain' + + pairs = [(field.name, field.value) for field in fields] + body = urlencode(pairs, encoding=encoding, errors='xmlcharrefreplace').encode() + return body, 'application/x-www-form-urlencoded' + + +def _form_owner(element: HtmlElement, elements_by_id: Mapping[str, HtmlElement]) -> HtmlElement | None: + """Get the form an element belongs to, honouring its `form` attribute.""" + form_id = element.get('form') + if form_id is None: + return next(element.iterancestors('form'), None) + + owner = elements_by_id.get(form_id) + if owner is None or owner.tag != 'form': + return None + return owner + + +def _find_clickable(elements: list[HtmlElement], click_data: Mapping[str, str] | None) -> HtmlElement | None: + """Find the submit button to click, matching all attributes in `click_data`.""" + for element in elements: + if not _is_submit_button(element) or _is_disabled(element): + continue + if not click_data: + return element + if all(element.get(key) == value for key, value in click_data.items()): + return element + return None + + +def _is_submit_button(element: HtmlElement) -> bool: + """Check whether the element submits the form when clicked.""" + button_type = element.get('type', '').lower() + if element.tag == 'button': + return button_type in ('', 'submit') + if element.tag == 'input': + return button_type in ('submit', 'image') + return False + + +def _is_disabled(element: HtmlElement) -> bool: + """Check whether the element, or a `
` containing it, is disabled.""" + if 'disabled' in element.attrib: + return True + return any('disabled' in fieldset.attrib for fieldset in element.iterancestors('fieldset')) + + +def _collect_fields( + elements: list[HtmlElement], + button: HtmlElement | None, + form_data: Mapping[str, str | Sequence[str] | None] | None, +) -> list[_Field]: + """Collect the entries the form submits: its fields, the clicked button and the `form_data` overrides.""" + fields: list[_Field] = [] + for element in elements: + if element.get('name') and not _is_disabled(element): + fields.extend(_element_fields(element)) + + if button is not None: + fields.extend(_button_fields(button)) + + if form_data: + fields = _apply_form_data(fields, form_data) + + return fields + + +def _element_fields(element: HtmlElement) -> list[_Field]: + """Get the entries a single enabled, named field submits.""" + name = element.get('name') + + if element.tag == 'select': + values = element.value if element.multiple else [element.value] + return [_Field(name, value) for value in values if value is not None] + + if element.tag == 'textarea': + # Browsers drop the newline right after ` + +
+ + + + + + + + """ + + request = await _single(form_requests, html) + + assert _form_values(request) == [ + ('text', 't'), + ('checked', 'c1'), + ('no-value', 'on'), + ('radio', 'r2'), + ('first', 'o1'), + ('multi', 'm1'), + ('multi', 'm3'), + ('area', 'long text'), + ('file', ''), + ('empty', ''), + ('hidden', ''), + ] + + +async def test_select_options(form_requests: FormRequests) -> None: + """A select submits its last selected option, or the first enabled one when none is selected.""" + html = """ +
+ + + +
+ """ + + request = await _single(form_requests, html) + + assert _form_values(request) == [('last', 'b'), ('first-enabled', 'on')] + + +async def test_fields_linked_by_form_attribute(form_requests: FormRequests) -> None: + """Fields are assigned to forms by their `form` attribute.""" + html = """ + + + + +
+ """ + + search, login = await form_requests(html) + + assert _form_values(search) == [('q', 'x'), ('lang', 'en'), ('go', '1')] + assert _form_values(login) == [('other', 'o')] + + +async def test_disabled_fieldset_around_form(form_requests: FormRequests) -> None: + """A disabled `
` around the form disables its fields.""" + html = '
' + + request = await _single(form_requests, html) + + assert _form_values(request) == [] + + +async def test_dialog_form_is_skipped(form_requests: FormRequests) -> None: + """Dialog forms send no request.""" + html = '
' + + assert [urlsplit(request.url).path for request in await form_requests(html)] == ['/ok'] + + +@pytest.mark.parametrize( + ('html', 'content_type', 'encoding'), + [ + pytest.param( + '
', + None, + 'utf-8', + id='accept-charset', + ), + pytest.param( + '
', + 'text/html; charset=windows-1251', + 'windows-1251', + id='header', + ), + pytest.param( + '
', + None, + 'windows-1251', + id='meta-charset', + ), + pytest.param( + '' + '
', + None, + 'windows-1251', + id='meta-http-equiv', + ), + ], +) +@pytest.mark.parametrize('method', [pytest.param('get', id='get'), pytest.param('post', id='post')]) +async def test_form_encoding( + form_requests: FormRequests, html: str, content_type: str | None, encoding: str, method: str +) -> None: + """Values are encoded in the form or page charset, with unsupported characters as character references.""" + html = html.format(value='привіт ✓').replace(' None: + """Values from `form_data` replace, drop and add fields.""" + html = '
' + + request = await _single(form_requests, html, form_data={'replace': 'new', 'drop': None, 'tags': ['a', 'b']}) + + assert _form_values(request) == [('keep', 'k'), ('replace', 'new'), ('tags', 'a'), ('tags', 'b')] + + +async def test_first_submit_button_is_clicked(form_requests: FormRequests) -> None: + """The first enabled submit button is included by default.""" + html = """ +
+ + + +
+ + +
+ """ + + request = await _single(form_requests, html) + + assert _form_values(request) == [('q', 'x'), ('go', 'first')] + + +async def test_click_data_selects_button(form_requests: FormRequests) -> None: + """`click_data` picks the button to click and its form overrides apply.""" + html = """ +
+ + + +
+ """ + + request = await _single(form_requests, html, click_data={'value': 'delete'}) + + assert request.method == 'POST' + assert request.url == 'https://example.com/delete' + assert _form_values(request) == [('q', 'x'), ('action', 'delete')] + + +async def test_dont_click(form_requests: FormRequests) -> None: + """`dont_click` leaves out every button.""" + html = '
' + + request = await _single(form_requests, html, dont_click=True) + + assert _form_values(request) == [('q', 'x')] + + +async def test_image_button_sends_coordinates(form_requests: FormRequests) -> None: + """A clicked image button sends its click coordinates.""" + html = '
' + + request = await _single(form_requests, html) + + assert _form_values(request) == [('map.x', '0'), ('map.y', '0')] + + +async def test_click_data_skips_unmatched_forms(form_requests: FormRequests) -> None: + """Forms with no button matching `click_data` are skipped.""" + html = '
' + + assert [urlsplit(request.url).path for request in await form_requests(html, click_data={'name': 'go'})] == ['/save'] + assert await form_requests(html, click_data={'name': 'missing'}) == [] + + +async def test_non_http_action_is_skipped(form_requests: FormRequests) -> None: + """Forms whose action isn't a valid HTTP(S) URL are skipped.""" + html = ( + '
' + '
' + ) + + assert [urlsplit(request.url).path for request in await form_requests(html)] == ['/ok'] + + +async def test_multipart_form(form_requests: FormRequests) -> None: + """A multipart form is encoded with a deterministic boundary and empty file parts.""" + html = ( + '
' + '' + '
' + ) + + request = await _single(form_requests, html) + again = await _single(form_requests, html) + + content_type = request.headers['content-type'] + assert content_type.startswith('multipart/form-data; boundary=') + boundary = content_type.split('boundary=')[1] + assert ( + request.payload + == ( + f'--{boundary}\r\n' + 'Content-Disposition: form-data; name="title"\r\n\r\nhi\r\n' + f'--{boundary}\r\n' + 'Content-Disposition: form-data; name="doc"; filename=""\r\n' + 'Content-Type: application/octet-stream\r\n\r\n\r\n' + f'--{boundary}--\r\n' + ).encode() + ) + assert request.unique_key == again.unique_key + + +async def test_text_plain_form(form_requests: FormRequests) -> None: + """A `text/plain` form sends one field per line.""" + html = '
' + + request = await _single(form_requests, html) + + assert request.headers['content-type'] == 'text/plain' + assert request.payload == b'a=1\r\nb=2\r\n' + + +@pytest.mark.parametrize( + ('html', 'expected_url'), + [ + pytest.param('
', 'https://example.com/page/index.html', id='no-action'), + pytest.param('
', 'https://example.com/up', id='relative'), + pytest.param( + '
', + 'https://other.com/dir/go', + id='base-href', + ), + pytest.param( + '
', + 'https://example.com/page/index.html', + id='base-href-no-action', + ), + pytest.param( + '
', + 'https://example.com/page/go', + id='invalid-base-href', + ), + ], +) +async def test_action_resolution(form_requests: FormRequests, html: str, expected_url: str) -> None: + """The action is resolved against ``, and a missing action submits to the page URL.""" + request = await _single(form_requests, html) + + assert request.url == expected_url + + +async def test_action_resolves_against_loaded_url(form_requests: FormRequests) -> None: + """The action is resolved against the URL the page was loaded from after redirects.""" + page_request = Request.from_url(_PAGE_URL, loaded_url='https://example.com/moved/index.html') + + request = await _single(form_requests, '
', page_request=page_request) + + assert request.url == 'https://example.com/moved/go' + + +async def test_textarea_leading_newline(form_requests: FormRequests) -> None: + """The newline right after `') + + assert _form_values(request) == [('a', '\nhello')] + + +async def test_selector_filters_forms(form_requests: FormRequests) -> None: + """All forms are used by default and `selector` narrows them.""" + html = '
' + + assert [urlsplit(request.url).path for request in await form_requests(html)] == ['/a', '/c'] + assert [urlsplit(request.url).path for request in await form_requests(html, selector='#c, #b')] == ['/c'] + + +async def test_request_options_are_passed(form_requests: FormRequests) -> None: + """Headers and request options are applied to the request.""" + html = '
' + + request = await _single(form_requests, html, headers={'X-Custom': '1'}, label='detail') + + assert request.headers['x-custom'] == '1' + assert request.headers['content-type'] == 'application/x-www-form-urlencoded' + assert request.label == 'detail' + + +async def test_xml_declaration(form_requests: FormRequests) -> None: + """Pages starting with an XML declaration are supported.""" + html = '
' + + assert [urlsplit(request.url).path for request in await form_requests(html)] == ['/ok'] + + +@pytest.mark.parametrize( + ('charset', 'value', 'expected_query'), + [ + pytest.param('us-ascii', 'é€', 'q=%E9%80', id='ascii'), + pytest.param('iso-8859-1', 'é€', 'q=%E9%80', id='latin1'), + pytest.param('shift_jis', '①', 'q=%87%40', id='shift-jis'), + ], +) +async def test_legacy_charset_uses_superset( + form_requests: FormRequests, charset: str, value: str, expected_query: str +) -> None: + """Legacy charsets are submitted in the superset browsers use instead.""" + html = f'
' + + request = await _single(form_requests, html, content_type=f'text/html; charset={charset}') + + assert request.url.split('?')[1] == expected_query + + +@pytest.mark.parametrize( + 'charset', + [ + pytest.param('base64', id='binary-codec'), + pytest.param('undefined', id='undefined'), + pytest.param('idna', id='idna'), + ], +) +async def test_non_text_charset_is_ignored(form_requests: FormRequests, charset: str) -> None: + """Charsets naming codecs that can't encode form data fall back to UTF-8.""" + html = f'
' + + request = await _single(form_requests, html) + + assert request.url == 'https://example.com/page/index.html?q=%C3%A9'