Universally handling different response encoding and charsets for all responses
Thanks for @andyfcx
This commit is contained in:
@@ -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!
|
||||
|
||||
@@ -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!
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user