diff --git a/scrapling/engines/camo.py b/scrapling/engines/camo.py index 0062ffc..0807a97 100644 --- a/scrapling/engines/camo.py +++ b/scrapling/engines/camo.py @@ -106,14 +106,8 @@ class CamoufoxEngine: waiter = page.locator(self.wait_selector) waiter.first.wait_for(state=self.wait_selector_state) - content_type = res.headers.get('content-type', '') - # Parse charset from content-type - encoding = 'utf-8' # default encoding - content_type_lower = content_type.lower() - if 'charset=' in content_type_lower: - encoding = content_type_lower.split('charset=')[-1].split(';')[0].strip().strip('"').strip("'") - if 'utf-8' in encoding: - encoding = 'utf-8' + # This will be parsed inside `Response` + encoding = res.headers.get('content-type', '') or 'utf-8' # default encoding status_text = res.status_text # PlayWright API sometimes give empty status text for some reason! diff --git a/scrapling/engines/pw.py b/scrapling/engines/pw.py index 6788078..78c1d39 100644 --- a/scrapling/engines/pw.py +++ b/scrapling/engines/pw.py @@ -216,11 +216,8 @@ class PlaywrightEngine: waiter = page.locator(self.wait_selector) waiter.first.wait_for(state=self.wait_selector_state) - content_type = res.headers.get('content-type', '') - # Parse charset from content-type - encoding = 'utf-8' # default encoding - if 'charset=' in content_type.lower(): - encoding = content_type.lower().split('charset=')[-1].split(';')[0].strip() + # This will be parsed inside `Response` + encoding = res.headers.get('content-type', '') or 'utf-8' # default encoding status_text = res.status_text # PlayWright API sometimes give empty status text for some reason! diff --git a/scrapling/engines/toolbelt/custom.py b/scrapling/engines/toolbelt/custom.py index 8957d1a..d056620 100644 --- a/scrapling/engines/toolbelt/custom.py +++ b/scrapling/engines/toolbelt/custom.py @@ -3,11 +3,78 @@ Functions related to custom types or type checking """ import inspect import logging +from email.message import Message from scrapling.core.custom_types import MappingProxyType from scrapling.parser import Adaptor, SQLiteStorageSystem from scrapling.core.utils import setup_basic_logging, cache -from scrapling.core._types import Any, List, Type, Union, Optional, Dict, Callable +from scrapling.core._types import Any, List, Type, Union, Optional, Dict, Callable, Tuple + + +class ResponseEncoding: + DEFAULT_ENCODING = "utf-8" + ISO_8859_1_CONTENT_TYPES = {"text/plain", "text/html", "text/css", "text/javascript"} + + @classmethod + @cache(maxsize=None) + def __parse_content_type(cls, header_value: str) -> Tuple[str, Dict[str, str]]: + """Parse content type and parameters from a content-type header value. + + Uses `email.message.Message` for robust header parsing according to RFC 2045. + + :param header_value: Raw content-type header string + :return: Tuple of (content_type, parameters_dict) + """ + # Create a Message object and set the Content-Type header then get the content type and parameters + msg = Message() + msg['content-type'] = header_value + + content_type = msg.get_content_type() + params = dict(msg.get_params(failobj=[])) + + # Remove the content-type from params if present somehow + params.pop('content-type', None) + + return content_type, params + + @classmethod + @cache(maxsize=None) + def get_value(cls, content_type: Optional[str]) -> str: + """Determine the appropriate character encoding from a content-type header. + + The encoding is determined by these rules in order: + 1. If no content-type is provided, use UTF-8 + 2. If charset parameter is present, use that encoding + 3. If content-type is text/*, use ISO-8859-1 per HTTP/1.1 spec + 4. If content-type is application/json, use UTF-8 per RFC 4627 + 5. Default to UTF-8 if nothing else matches + + :param content_type: Content-Type header value or None + :return: String naming the character encoding + """ + if not content_type: + return cls.DEFAULT_ENCODING + + try: + content_type, params = cls.__parse_content_type(content_type) + + # First check for explicit charset parameter + if "charset" in params: + encoding = params["charset"].strip("'\"") + "test".encode(encoding) # Validate encoding + return encoding + + # Apply content-type specific rules + if content_type in cls.ISO_8859_1_CONTENT_TYPES: + return "ISO-8859-1" + + if content_type == "application/json": + return cls.DEFAULT_ENCODING + + return cls.DEFAULT_ENCODING + + except (ValueError, LookupError, UnicodeEncodeError): + return cls.DEFAULT_ENCODING class Response(Adaptor): @@ -20,6 +87,7 @@ class Response(Adaptor): self.cookies = cookies self.headers = headers self.request_headers = request_headers + encoding = ResponseEncoding.get_value(encoding) super().__init__(text=text, body=body, url=automatch_domain or url, encoding=encoding, **adaptor_arguments) # For back-ward compatibility self.adaptor = self