feat(browsers): Add option to block requests to specific domains
This commit is contained in:
@@ -11,7 +11,7 @@ from msgspec import Struct, structs, convert, ValidationError
|
||||
from playwright.sync_api import Route
|
||||
|
||||
from scrapling.core.utils import log
|
||||
from scrapling.core._types import Dict, Tuple
|
||||
from scrapling.core._types import Dict, Set, Tuple, Optional, Callable
|
||||
from scrapling.engines.constants import DEFAULT_DISABLED_RESOURCES
|
||||
|
||||
__BYPASSES_DIR__ = Path(__file__).parent / "bypasses"
|
||||
@@ -23,30 +23,58 @@ class ProxyDict(Struct):
|
||||
password: str = ""
|
||||
|
||||
|
||||
def intercept_route(route: Route):
|
||||
"""This is just a route handler, but it drops requests that its type falls in `DEFAULT_DISABLED_RESOURCES`
|
||||
def create_intercept_handler(disable_resources: bool, blocked_domains: Optional[Set[str]] = None) -> Callable:
|
||||
"""Create a route handler that blocks both resource types and specific domains.
|
||||
|
||||
:param route: PlayWright `Route` object of the current page
|
||||
:return: PlayWright `Route` object
|
||||
:param disable_resources: Whether to block default resource types.
|
||||
:param blocked_domains: Set of domain names to block requests to.
|
||||
:return: A sync route handler function.
|
||||
"""
|
||||
if route.request.resource_type in DEFAULT_DISABLED_RESOURCES:
|
||||
log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"')
|
||||
route.abort()
|
||||
else:
|
||||
route.continue_()
|
||||
disabled_resources = DEFAULT_DISABLED_RESOURCES if disable_resources else set()
|
||||
domains = blocked_domains or set()
|
||||
|
||||
def handler(route: Route):
|
||||
if route.request.resource_type in disabled_resources:
|
||||
log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"')
|
||||
route.abort()
|
||||
elif domains:
|
||||
hostname = urlparse(route.request.url).hostname or ""
|
||||
if any(hostname == d or hostname.endswith("." + d) for d in domains):
|
||||
log.debug(f'Blocking request to blocked domain "{hostname}" ({route.request.url})')
|
||||
route.abort()
|
||||
else:
|
||||
route.continue_()
|
||||
else:
|
||||
route.continue_()
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
async def async_intercept_route(route: async_Route):
|
||||
"""This is just a route handler, but it drops requests that its type falls in `DEFAULT_DISABLED_RESOURCES`
|
||||
def create_async_intercept_handler(disable_resources: bool, blocked_domains: Optional[Set[str]] = None) -> Callable:
|
||||
"""Create an async route handler that blocks both resource types and specific domains.
|
||||
|
||||
:param route: PlayWright `Route` object of the current page
|
||||
:return: PlayWright `Route` object
|
||||
:param disable_resources: Whether to block default resource types.
|
||||
:param blocked_domains: Set of domain names to block requests to.
|
||||
:return: An async route handler function.
|
||||
"""
|
||||
if route.request.resource_type in DEFAULT_DISABLED_RESOURCES:
|
||||
log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"')
|
||||
await route.abort()
|
||||
else:
|
||||
await route.continue_()
|
||||
disabled_resources = DEFAULT_DISABLED_RESOURCES if disable_resources else set()
|
||||
domains = blocked_domains or set()
|
||||
|
||||
async def handler(route: async_Route):
|
||||
if route.request.resource_type in disabled_resources:
|
||||
log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"')
|
||||
await route.abort()
|
||||
elif domains:
|
||||
hostname = urlparse(route.request.url).hostname or ""
|
||||
if any(hostname == d or hostname.endswith("." + d) for d in domains):
|
||||
log.debug(f'Blocking request to blocked domain "{hostname}" ({route.request.url})')
|
||||
await route.abort()
|
||||
else:
|
||||
await route.continue_()
|
||||
else:
|
||||
await route.continue_()
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
def construct_proxy_dict(proxy_string: str | Dict[str, str] | Tuple) -> Dict:
|
||||
|
||||
Reference in New Issue
Block a user