ayrisdev/crawl4ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(browser): reorganize browser strategies and improve Docker implementation

Browse Source 
Reorganize browser strategy code into separate modules for better maintainability and separation of concerns. Improve Docker implementation with:
- Add Alpine and Debian-based Dockerfiles for better container options
- Enhance Docker registry to share configuration with BuiltinBrowserStrategy
- Add CPU and memory limits to container configuration
- Improve error handling and logging
- Update documentation and examples

BREAKING CHANGE: DockerConfig, DockerRegistry, and DockerUtils have been moved to new locations and their APIs have been updated.
This commit is contained in:
UncleCode
2025-03-27 21:35:13 +08:00
parent 7f93e88379
commit c635f6b9a2
20 changed files with 2502 additions and 1786 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
crawl4ai/browser/__init__.py
View File

@@ -6,5 +6,17 @@ for browser creation and interaction.
from .manager import BrowserManager
from .profiles import BrowserProfileManager
from .models import DockerConfig
from .docker_registry import DockerRegistry
from .docker_utils import DockerUtils
from .strategies import (
BaseBrowserStrategy,
PlaywrightBrowserStrategy,
CDPBrowserStrategy,
BuiltinBrowserStrategy,
DockerBrowserStrategy
)
__all__ = ['BrowserManager', 'BrowserProfileManager']
__all__ = ['BrowserManager', 'BrowserProfileManager', 'DockerConfig', 'DockerRegistry', 'DockerUtils', 'BaseBrowserStrategy',
'PlaywrightBrowserStrategy', 'CDPBrowserStrategy', 'BuiltinBrowserStrategy',
'DockerBrowserStrategy']

34
crawl4ai/browser/docker/alpine/connect.Dockerfile Normal file
View File

@@ -0,0 +1,34 @@
# ---------- Dockerfile ----------
FROM alpine:latest
# Combine everything in one RUN to keep layers minimal.
RUN apk update && apk upgrade && \
apk add --no-cache \
chromium \
nss \
freetype \
harfbuzz \
ca-certificates \
ttf-freefont \
socat \
curl && \
addgroup -S chromium && adduser -S chromium -G chromium && \
mkdir -p /data && chown chromium:chromium /data && \
rm -rf /var/cache/apk/*
# Copy start script, then chown/chmod in one step
COPY start.sh /home/chromium/start.sh
RUN chown chromium:chromium /home/chromium/start.sh && \
chmod +x /home/chromium/start.sh
USER chromium
WORKDIR /home/chromium
# Expose port used by socat (mapping 9222→9223 or whichever you prefer)
EXPOSE 9223
# Simple healthcheck: is the remote debug endpoint responding?
HEALTHCHECK --interval=30s --timeout=5s --retries=3 CMD curl -f http://localhost:9222/json/version || exit 1
CMD ["./start.sh"]

23
crawl4ai/browser/docker/alpine/launch.Dockerfile Normal file
View File

@@ -0,0 +1,23 @@
# ---------- Dockerfile (Idle Version) ----------
FROM alpine:latest
# Install only Chromium and its dependencies in a single layer
RUN apk update && apk upgrade && \
apk add --no-cache \
chromium \
nss \
freetype \
harfbuzz \
ca-certificates \
ttf-freefont && \
addgroup -S chromium && adduser -S chromium -G chromium && \
mkdir -p /data && chown chromium:chromium /data && \
rm -rf /var/cache/apk/*
# Switch to a non-root user for security
USER chromium
WORKDIR /home/chromium
# Idle: container does nothing except stay alive
CMD ["tail", "-f", "/dev/null"]

32
crawl4ai/browser/docker/connect.Dockerfile
View File

@@ -3,16 +3,14 @@ FROM ubuntu:22.04
# Install dependencies with comprehensive Chromium support
RUN apt-get update && apt-get install -y --no-install-recommends \
wget \
curl \
gnupg \
ca-certificates \
fonts-liberation \
# Sound support
# Core dependencies
libasound2 \
# Accessibility support
libatspi2.0-0 \
libatk1.0-0 \
libatk-bridge2.0-0 \
# Graphics and rendering
libdrm2 \
libgbm1 \
libgtk-3-0 \
@@ -21,16 +19,12 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
libxext6 \
libxfixes3 \
libxrandr2 \
# X11 and window system
libx11-6 \
libxcb1 \
libxkbcommon0 \
# Text and internationalization
libpango-1.0-0 \
libcairo2 \
# Printing support
libcups2 \
# System libraries
libdbus-1-3 \
libnss3 \
libnspr4 \
@@ -38,24 +32,24 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
# Utilities
xdg-utils \
socat \
# Process management
procps \
# Clean up
&& rm -rf /var/lib/apt/lists/*
# Install Chrome
RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - && \
echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list && \
apt-get update && \
apt-get install -y google-chrome-stable && \
rm -rf /var/lib/apt/lists/*
# Install Chromium with codecs
RUN apt-get update && \
apt-get install -y \
chromium-browser \
chromium-codecs-ffmpeg-extra \
&& rm -rf /var/lib/apt/lists/*
# Create data directory for user data
# Create Chrome alias for compatibility
RUN ln -s /usr/bin/chromium-browser /usr/bin/google-chrome
# Create data directory
RUN mkdir -p /data && chmod 777 /data
# Add a startup script
# Add startup script
COPY start.sh /start.sh
RUN chmod +x /start.sh
# Set entrypoint
ENTRYPOINT ["/start.sh"]

23
crawl4ai/browser/docker/debian/connect.Dockerfile Normal file
View File

@@ -0,0 +1,23 @@
# Use Debian 12 (Bookworm) slim for a small, stable base image
FROM debian:bookworm-slim
ENV DEBIAN_FRONTEND=noninteractive
# Install Chromium, socat, and basic fonts
RUN apt-get update && apt-get install -y --no-install-recommends \
chromium \
wget \
curl \
socat \
fonts-freefont-ttf \
fonts-noto-color-emoji && \
apt-get clean && rm -rf /var/lib/apt/lists/*
# Copy start.sh and make it executable
COPY start.sh /start.sh
RUN chmod +x /start.sh
# Expose socat port (use host mapping, e.g. -p 9225:9223)
EXPOSE 9223
ENTRYPOINT ["/start.sh"]

6
crawl4ai/browser/docker/launch.Dockerfile
View File

@@ -43,9 +43,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
# Clean up
&& rm -rf /var/lib/apt/lists/*
# Install Chrome
RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - && \
echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list && \
# Install Chrome (new method)
RUN curl -fsSL https://dl.google.com/linux/linux_signing_key.pub | gpg --dearmor -o /usr/share/keyrings/googlechrome-linux-keyring.gpg && \
echo "deb [arch=amd64 signed-by=/usr/share/keyrings/googlechrome-linux-keyring.gpg] http://dl.google.com/linux/chrome/deb/ stable main" | tee /etc/apt/sources.list.d/google-chrome.list && \
apt-get update && \
apt-get install -y google-chrome-stable && \
rm -rf /var/lib/apt/lists/*

133
crawl4ai/browser/docker_config.py
View File

@@ -1,133 +0,0 @@
"""Docker configuration module for Crawl4AI browser automation.
This module provides configuration classes for Docker-based browser automation,
allowing flexible configuration of Docker containers for browsing.
"""
from typing import Dict, List, Optional, Union
class DockerConfig:
"""Configuration for Docker-based browser automation.
This class contains Docker-specific settings to avoid cluttering BrowserConfig.
Attributes:
mode (str): Docker operation mode - "connect" or "launch".
- "connect": Uses a container with Chrome already running
- "launch": Dynamically configures and starts Chrome in container
image (str): Docker image to use. If None, defaults from DockerUtils are used.
registry_file (str): Path to container registry file for persistence.
persistent (bool): Keep container running after browser closes.
remove_on_exit (bool): Remove container on exit when not persistent.
network (str): Docker network to use.
volumes (List[str]): Volume mappings (e.g., ["host_path:container_path"]).
env_vars (Dict[str, str]): Environment variables to set in container.
extra_args (List[str]): Additional docker run arguments.
host_port (int): Host port to map to container's 9223 port.
user_data_dir (str): Path to user data directory on host.
container_user_data_dir (str): Path to user data directory in container.
"""
def __init__(
self,
mode: str = "connect", # "connect" or "launch"
image: Optional[str] = None, # Docker image to use
registry_file: Optional[str] = None, # Path to registry file
persistent: bool = False, # Keep container running after browser closes
remove_on_exit: bool = True, # Remove container on exit when not persistent
network: Optional[str] = None, # Docker network to use
volumes: List[str] = None, # Volume mappings
env_vars: Dict[str, str] = None, # Environment variables
extra_args: List[str] = None, # Additional docker run arguments
host_port: Optional[int] = None, # Host port to map to container's 9223
user_data_dir: Optional[str] = None, # Path to user data directory on host
container_user_data_dir: str = "/data", # Path to user data directory in container
):
"""Initialize Docker configuration.
Args:
mode: Docker operation mode ("connect" or "launch")
image: Docker image to use
registry_file: Path to container registry file
persistent: Whether to keep container running after browser closes
remove_on_exit: Whether to remove container on exit when not persistent
network: Docker network to use
volumes: Volume mappings as list of strings
env_vars: Environment variables as dictionary
extra_args: Additional docker run arguments
host_port: Host port to map to container's 9223
user_data_dir: Path to user data directory on host
container_user_data_dir: Path to user data directory in container
"""
self.mode = mode
self.image = image # If None, defaults will be used from DockerUtils
self.registry_file = registry_file
self.persistent = persistent
self.remove_on_exit = remove_on_exit
self.network = network
self.volumes = volumes or []
self.env_vars = env_vars or {}
self.extra_args = extra_args or []
self.host_port = host_port
self.user_data_dir = user_data_dir
self.container_user_data_dir = container_user_data_dir
def to_dict(self) -> Dict:
"""Convert this configuration to a dictionary.
Returns:
Dictionary representation of this configuration
"""
return {
"mode": self.mode,
"image": self.image,
"registry_file": self.registry_file,
"persistent": self.persistent,
"remove_on_exit": self.remove_on_exit,
"network": self.network,
"volumes": self.volumes,
"env_vars": self.env_vars,
"extra_args": self.extra_args,
"host_port": self.host_port,
"user_data_dir": self.user_data_dir,
"container_user_data_dir": self.container_user_data_dir
}
@staticmethod
def from_kwargs(kwargs: Dict) -> "DockerConfig":
"""Create a DockerConfig from a dictionary of keyword arguments.
Args:
kwargs: Dictionary of configuration options
Returns:
New DockerConfig instance
"""
return DockerConfig(
mode=kwargs.get("mode", "connect"),
image=kwargs.get("image"),
registry_file=kwargs.get("registry_file"),
persistent=kwargs.get("persistent", False),
remove_on_exit=kwargs.get("remove_on_exit", True),
network=kwargs.get("network"),
volumes=kwargs.get("volumes"),
env_vars=kwargs.get("env_vars"),
extra_args=kwargs.get("extra_args"),
host_port=kwargs.get("host_port"),
user_data_dir=kwargs.get("user_data_dir"),
container_user_data_dir=kwargs.get("container_user_data_dir", "/data")
)
def clone(self, **kwargs) -> "DockerConfig":
"""Create a copy of this configuration with updated values.
Args:
**kwargs: Key-value pairs of configuration options to update
Returns:
DockerConfig: A new instance with the specified updates
"""
config_dict = self.to_dict()
config_dict.update(kwargs)
return DockerConfig.from_kwargs(config_dict)

140
crawl4ai/browser/docker_registry.py
View File

@@ -31,9 +31,10 @@ class DockerRegistry:
Args:
registry_file: Path to the registry file. If None, uses default path.
"""
self.registry_file = registry_file or os.path.join(get_home_folder(), "docker_browser_registry.json")
self.containers = {}
self.port_map = {}
# Use the same file path as BuiltinBrowserStrategy by default
self.registry_file = registry_file or os.path.join(get_home_folder(), "builtin-browser", "browser_config.json")
self.containers = {} # Still maintain this for backward compatibility
self.port_map = {} # Will be populated from the shared file
self.last_port = 9222
self.load()
@@ -43,11 +44,35 @@ class DockerRegistry:
try:
with open(self.registry_file, 'r') as f:
registry_data = json.load(f)
self.containers = registry_data.get("containers", {})
self.port_map = registry_data.get("ports", {})
self.last_port = registry_data.get("last_port", 9222)
except Exception:
# Initialize port_map if not present
if "port_map" not in registry_data:
registry_data["port_map"] = {}
self.port_map = registry_data.get("port_map", {})
# Extract container information from port_map entries of type "docker"
self.containers = {}
for port_str, browser_info in self.port_map.items():
if browser_info.get("browser_type") == "docker" and "container_id" in browser_info:
container_id = browser_info["container_id"]
self.containers[container_id] = {
"host_port": int(port_str),
"config_hash": browser_info.get("config_hash", ""),
"created_at": browser_info.get("created_at", time.time())
}
# Get last port if available
if "last_port" in registry_data:
self.last_port = registry_data["last_port"]
else:
# Find highest port in port_map
ports = [int(p) for p in self.port_map.keys() if p.isdigit()]
self.last_port = max(ports + [9222])
except Exception as e:
# Reset to defaults on error
print(f"Error loading registry: {e}")
self.containers = {}
self.port_map = {}
self.last_port = 9222
@@ -59,28 +84,75 @@ class DockerRegistry:
def save(self):
"""Save container registry to file."""
# First load the current file to avoid overwriting other browser types
current_data = {"port_map": {}, "last_port": self.last_port}
if os.path.exists(self.registry_file):
try:
with open(self.registry_file, 'r') as f:
current_data = json.load(f)
except Exception:
pass
# Create a new port_map dictionary
updated_port_map = {}
# First, copy all non-docker entries from the existing port_map
for port_str, browser_info in current_data.get("port_map", {}).items():
if browser_info.get("browser_type") != "docker":
updated_port_map[port_str] = browser_info
# Then add all current docker container entries
for container_id, container_info in self.containers.items():
port_str = str(container_info["host_port"])
updated_port_map[port_str] = {
"browser_type": "docker",
"container_id": container_id,
"cdp_url": f"http://localhost:{port_str}",
"config_hash": container_info["config_hash"],
"created_at": container_info["created_at"]
}
# Replace the port_map with our updated version
current_data["port_map"] = updated_port_map
# Update last_port
current_data["last_port"] = self.last_port
# Ensure directory exists
os.makedirs(os.path.dirname(self.registry_file), exist_ok=True)
# Save the updated data
with open(self.registry_file, 'w') as f:
json.dump({
"containers": self.containers,
"ports": self.port_map,
"last_port": self.last_port
}, f, indent=2)
json.dump(current_data, f, indent=2)
def register_container(self, container_id: str, host_port: int, config_hash: str):
def register_container(self, container_id: str, host_port: int, config_hash: str, cdp_json_config: Optional[str] = None):
"""Register a container with its configuration hash and port mapping.
Args:
container_id: Docker container ID
host_port: Host port mapped to container
config_hash: Hash of configuration used to create container
cdp_json_config: CDP JSON configuration if available
"""
self.containers[container_id] = {
"host_port": host_port,
"config_hash": config_hash,
"created_at": time.time()
}
self.port_map[str(host_port)] = container_id
# Update port_map to maintain compatibility with BuiltinBrowserStrategy
port_str = str(host_port)
self.port_map[port_str] = {
"browser_type": "docker",
"container_id": container_id,
"cdp_url": f"http://localhost:{port_str}",
"config_hash": config_hash,
"created_at": time.time()
}
if cdp_json_config:
self.port_map[port_str]["cdp_json_config"] = cdp_json_config
self.save()
def unregister_container(self, container_id: str):
@@ -91,12 +163,18 @@ class DockerRegistry:
"""
if container_id in self.containers:
host_port = self.containers[container_id]["host_port"]
if str(host_port) in self.port_map:
del self.port_map[str(host_port)]
port_str = str(host_port)
# Remove from port_map
if port_str in self.port_map:
del self.port_map[port_str]
# Remove from containers
del self.containers[container_id]
self.save()
def find_container_by_config(self, config_hash: str, docker_utils) -> Optional[str]:
async def find_container_by_config(self, config_hash: str, docker_utils) -> Optional[str]:
"""Find a container that matches the given configuration hash.
Args:
@@ -106,9 +184,16 @@ class DockerRegistry:
Returns:
Container ID if found, None otherwise
"""
for container_id, data in self.containers.items():
if data["config_hash"] == config_hash and docker_utils.is_container_running(container_id):
return container_id
# Search through port_map for entries with matching config_hash
for port_str, browser_info in self.port_map.items():
if (browser_info.get("browser_type") == "docker" and
browser_info.get("config_hash") == config_hash and
"container_id" in browser_info):
container_id = browser_info["container_id"]
if await docker_utils.is_container_running(container_id):
return container_id
return None
def get_container_host_port(self, container_id: str) -> Optional[int]:
@@ -137,7 +222,7 @@ class DockerRegistry:
port = self.last_port + 1
# Check if port is in use (either in our registry or system-wide)
while port in self.port_map or docker_utils.is_port_in_use(port):
while str(port) in self.port_map or docker_utils.is_port_in_use(port):
port += 1
# Update last port
@@ -166,9 +251,14 @@ class DockerRegistry:
docker_utils: DockerUtils instance to check container status
"""
to_remove = []
for container_id in self.containers:
if not docker_utils.is_container_running(container_id):
to_remove.append(container_id)
# Find containers that are no longer running
for port_str, browser_info in self.port_map.items():
if browser_info.get("browser_type") == "docker" and "container_id" in browser_info:
container_id = browser_info["container_id"]
if not docker_utils.is_container_running(container_id):
to_remove.append(container_id)
# Remove stale containers
for container_id in to_remove:
self.unregister_container(container_id)

443
crawl4ai/browser/docker_utils.py
View File

@@ -8,13 +8,14 @@ import socket
import subprocess
from typing import Dict, List, Optional, Tuple, Union
class DockerUtils:
"""Utility class for Docker operations in browser automation.
This class provides methods for managing Docker images, containers,
and related operations needed for browser automation. It handles
image building, container lifecycle, port management, and registry operations.
Attributes:
DOCKER_FOLDER (str): Path to folder containing Docker files
DOCKER_CONNECT_FILE (str): Path to Dockerfile for connect mode
@@ -24,38 +25,38 @@ class DockerUtils:
DEFAULT_LAUNCH_IMAGE (str): Default image name for launch mode
logger: Optional logger instance
"""
# File paths for Docker resources
DOCKER_FOLDER = os.path.join(os.path.dirname(__file__), "docker")
DOCKER_CONNECT_FILE = os.path.join(DOCKER_FOLDER, "connect.Dockerfile")
DOCKER_LAUNCH_FILE = os.path.join(DOCKER_FOLDER, "launch.Dockerfile")
DOCKER_START_SCRIPT = os.path.join(DOCKER_FOLDER, "start.sh")
# Default image names
DEFAULT_CONNECT_IMAGE = "crawl4ai/browser-connect:latest"
DEFAULT_LAUNCH_IMAGE = "crawl4ai/browser-launch:latest"
def __init__(self, logger=None):
"""Initialize Docker utilities.
Args:
logger: Optional logger for recording operations
"""
self.logger = logger
# Image Management Methods
async def check_image_exists(self, image_name: str) -> bool:
"""Check if a Docker image exists.
Args:
image_name: Name of the Docker image to check
Returns:
bool: True if the image exists, False otherwise
"""
cmd = ["docker", "image", "inspect", image_name]
try:
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
@@ -64,18 +65,24 @@ class DockerUtils:
return process.returncode == 0
except Exception as e:
if self.logger:
self.logger.debug(f"Error checking if image exists: {str(e)}", tag="DOCKER")
self.logger.debug(
f"Error checking if image exists: {str(e)}", tag="DOCKER"
)
return False
async def build_docker_image(self, image_name: str, dockerfile_path: str,
files_to_copy: Dict[str, str] = None) -> bool:
async def build_docker_image(
self,
image_name: str,
dockerfile_path: str,
files_to_copy: Dict[str, str] = None,
) -> bool:
"""Build a Docker image from a Dockerfile.
Args:
image_name: Name to give the built image
dockerfile_path: Path to the Dockerfile
files_to_copy: Dict of {dest_name: source_path} for files to copy to build context
Returns:
bool: True if image was built successfully, False otherwise
"""
@@ -83,103 +90,119 @@ class DockerUtils:
with tempfile.TemporaryDirectory() as temp_dir:
# Copy the Dockerfile
shutil.copy(dockerfile_path, os.path.join(temp_dir, "Dockerfile"))
# Copy any additional files needed
if files_to_copy:
for dest_name, source_path in files_to_copy.items():
shutil.copy(source_path, os.path.join(temp_dir, dest_name))
# Build the image
cmd = [
"docker", "build",
"-t", image_name,
temp_dir
]
cmd = ["docker", "build", "-t", image_name, temp_dir]
if self.logger:
self.logger.debug(f"Building Docker image with command: {' '.join(cmd)}", tag="DOCKER")
self.logger.debug(
f"Building Docker image with command: {' '.join(cmd)}", tag="DOCKER"
)
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
if self.logger:
self.logger.error(
message="Failed to build Docker image: {error}",
tag="DOCKER",
params={"error": stderr.decode()}
params={"error": stderr.decode()},
)
return False
if self.logger:
self.logger.success(f"Successfully built Docker image: {image_name}", tag="DOCKER")
self.logger.success(
f"Successfully built Docker image: {image_name}", tag="DOCKER"
)
return True
async def ensure_docker_image_exists(self, image_name: str, mode: str = "connect") -> str:
async def ensure_docker_image_exists(
self, image_name: str, mode: str = "connect"
) -> str:
"""Ensure the required Docker image exists, creating it if necessary.
Args:
image_name: Name of the Docker image
mode: Either "connect" or "launch" to determine which image to build
Returns:
str: Name of the available Docker image
Raises:
Exception: If image doesn't exist and can't be built
"""
# If image name is not specified, use default based on mode
if not image_name:
image_name = self.DEFAULT_CONNECT_IMAGE if mode == "connect" else self.DEFAULT_LAUNCH_IMAGE
image_name = (
self.DEFAULT_CONNECT_IMAGE
if mode == "connect"
else self.DEFAULT_LAUNCH_IMAGE
)
# Check if the image already exists
if await self.check_image_exists(image_name):
if self.logger:
self.logger.debug(f"Docker image {image_name} already exists", tag="DOCKER")
self.logger.debug(
f"Docker image {image_name} already exists", tag="DOCKER"
)
return image_name
# If we're using a custom image that doesn't exist, warn and fail
if (image_name != self.DEFAULT_CONNECT_IMAGE and image_name != self.DEFAULT_LAUNCH_IMAGE):
if (
image_name != self.DEFAULT_CONNECT_IMAGE
and image_name != self.DEFAULT_LAUNCH_IMAGE
):
if self.logger:
self.logger.warning(
f"Custom Docker image {image_name} not found and cannot be automatically created",
tag="DOCKER"
tag="DOCKER",
)
raise Exception(f"Docker image {image_name} not found")
# Build the appropriate default image
if self.logger:
self.logger.info(f"Docker image {image_name} not found, creating it now...", tag="DOCKER")
self.logger.info(
f"Docker image {image_name} not found, creating it now...", tag="DOCKER"
)
if mode == "connect":
success = await self.build_docker_image(
image_name,
self.DOCKER_CONNECT_FILE,
{"start.sh": self.DOCKER_START_SCRIPT}
image_name,
self.DOCKER_CONNECT_FILE,
{"start.sh": self.DOCKER_START_SCRIPT},
)
else:
success = await self.build_docker_image(
image_name,
self.DOCKER_LAUNCH_FILE
)
success = await self.build_docker_image(image_name, self.DOCKER_LAUNCH_FILE)
if not success:
raise Exception(f"Failed to create Docker image {image_name}")
return image_name
# Container Management Methods
async def create_container(self, image_name: str, host_port: int,
container_name: Optional[str] = None,
volumes: List[str] = None,
network: Optional[str] = None,
env_vars: Dict[str, str] = None,
extra_args: List[str] = None) -> Optional[str]:
async def create_container(
self,
image_name: str,
host_port: int,
container_name: Optional[str] = None,
volumes: List[str] = None,
network: Optional[str] = None,
env_vars: Dict[str, str] = None,
cpu_limit: float = 1.0,
memory_limit: str = "1.5g",
extra_args: List[str] = None,
) -> Optional[str]:
"""Create a new Docker container.
Args:
image_name: Docker image to use
host_port: Port on host to map to container port 9223
@@ -187,111 +210,134 @@ class DockerUtils:
volumes: List of volume mappings (e.g., ["host_path:container_path"])
network: Optional Docker network to use
env_vars: Dictionary of environment variables
cpu_limit: CPU limit for the container
memory_limit: Memory limit for the container
extra_args: Additional docker run arguments
Returns:
str: Container ID if successful, None otherwise
"""
# Prepare container command
cmd = [
"docker", "run",
"docker",
"run",
"--detach",
]
# Add container name if specified
if container_name:
cmd.extend(["--name", container_name])
# Add port mapping
cmd.extend(["-p", f"{host_port}:9223"])
# Add volumes
if volumes:
for volume in volumes:
cmd.extend(["-v", volume])
# Add network if specified
if network:
cmd.extend(["--network", network])
# Add environment variables
if env_vars:
for key, value in env_vars.items():
cmd.extend(["-e", f"{key}={value}"])
# Add CPU and memory limits
if cpu_limit:
cmd.extend(["--cpus", str(cpu_limit)])
if memory_limit:
cmd.extend(["--memory", memory_limit])
cmd.extend(["--memory-swap", memory_limit])
if self.logger:
self.logger.debug(
f"Setting CPU limit: {cpu_limit}, Memory limit: {memory_limit}",
tag="DOCKER",
)
# Add extra args
if extra_args:
cmd.extend(extra_args)
# Add image
cmd.append(image_name)
if self.logger:
self.logger.debug(f"Creating Docker container with command: {' '.join(cmd)}", tag="DOCKER")
self.logger.debug(
f"Creating Docker container with command: {' '.join(cmd)}", tag="DOCKER"
)
# Run docker command
try:
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
if self.logger:
self.logger.error(
message="Failed to create Docker container: {error}",
tag="DOCKER",
params={"error": stderr.decode()}
params={"error": stderr.decode()},
)
return None
# Get container ID
container_id = stdout.decode().strip()
if self.logger:
self.logger.success(f"Created Docker container: {container_id[:12]}", tag="DOCKER")
self.logger.success(
f"Created Docker container: {container_id[:12]}", tag="DOCKER"
)
return container_id
except Exception as e:
if self.logger:
self.logger.error(
message="Error creating Docker container: {error}",
tag="DOCKER",
params={"error": str(e)}
params={"error": str(e)},
)
return None
async def is_container_running(self, container_id: str) -> bool:
"""Check if a container is running.
Args:
container_id: ID of the container to check
Returns:
bool: True if the container is running, False otherwise
"""
cmd = ["docker", "inspect", "--format", "{{.State.Running}}", container_id]
try:
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, _ = await process.communicate()
return process.returncode == 0 and stdout.decode().strip() == "true"
except Exception as e:
if self.logger:
self.logger.debug(f"Error checking if container is running: {str(e)}", tag="DOCKER")
self.logger.debug(
f"Error checking if container is running: {str(e)}", tag="DOCKER"
)
return False
async def wait_for_container_ready(self, container_id: str, timeout: int = 30) -> bool:
async def wait_for_container_ready(
self, container_id: str, timeout: int = 30
) -> bool:
"""Wait for the container to be in running state.
Args:
container_id: ID of the container to wait for
timeout: Maximum time to wait in seconds
Returns:
bool: True if container is ready, False if timeout occurred
"""
@@ -299,46 +345,51 @@ class DockerUtils:
if await self.is_container_running(container_id):
return True
await asyncio.sleep(1)
if self.logger:
self.logger.warning(f"Container {container_id[:12]} not ready after {timeout}s timeout", tag="DOCKER")
self.logger.warning(
f"Container {container_id[:12]} not ready after {timeout}s timeout",
tag="DOCKER",
)
return False
async def stop_container(self, container_id: str) -> bool:
"""Stop a Docker container.
Args:
container_id: ID of the container to stop
Returns:
bool: True if stopped successfully, False otherwise
"""
cmd = ["docker", "stop", container_id]
try:
process = await asyncio.create_subprocess_exec(*cmd)
await process.communicate()
if self.logger:
self.logger.debug(f"Stopped container: {container_id[:12]}", tag="DOCKER")
self.logger.debug(
f"Stopped container: {container_id[:12]}", tag="DOCKER"
)
return process.returncode == 0
except Exception as e:
if self.logger:
self.logger.warning(
message="Failed to stop container: {error}",
tag="DOCKER",
params={"error": str(e)}
params={"error": str(e)},
)
return False
async def remove_container(self, container_id: str, force: bool = True) -> bool:
"""Remove a Docker container.
Args:
container_id: ID of the container to remove
force: Whether to force removal
Returns:
bool: True if removed successfully, False otherwise
"""
@@ -346,35 +397,38 @@ class DockerUtils:
if force:
cmd.append("-f")
cmd.append(container_id)
try:
process = await asyncio.create_subprocess_exec(*cmd)
await process.communicate()
if self.logger:
self.logger.debug(f"Removed container: {container_id[:12]}", tag="DOCKER")
self.logger.debug(
f"Removed container: {container_id[:12]}", tag="DOCKER"
)
return process.returncode == 0
except Exception as e:
if self.logger:
self.logger.warning(
message="Failed to remove container: {error}",
tag="DOCKER",
params={"error": str(e)}
params={"error": str(e)},
)
return False
# Container Command Execution Methods
async def exec_in_container(self, container_id: str, command: List[str],
detach: bool = False) -> Tuple[int, str, str]:
async def exec_in_container(
self, container_id: str, command: List[str], detach: bool = False
) -> Tuple[int, str, str]:
"""Execute a command in a running container.
Args:
container_id: ID of the container
command: Command to execute as a list of strings
detach: Whether to run the command in detached mode
Returns:
Tuple of (return_code, stdout, stderr)
"""
@@ -383,181 +437,206 @@ class DockerUtils:
cmd.append("-d")
cmd.append(container_id)
cmd.extend(command)
try:
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
return process.returncode, stdout.decode(), stderr.decode()
except Exception as e:
if self.logger:
self.logger.error(
message="Error executing command in container: {error}",
tag="DOCKER",
params={"error": str(e)}
params={"error": str(e)},
)
return -1, "", str(e)
async def start_socat_in_container(self, container_id: str) -> bool:
"""Start socat in the container to map port 9222 to 9223.
Args:
container_id: ID of the container
Returns:
bool: True if socat started successfully, False otherwise
"""
# Command to run socat as a background process
cmd = ["socat", "TCP-LISTEN:9223,fork", "TCP:localhost:9222"]
returncode, _, stderr = await self.exec_in_container(container_id, cmd, detach=True)
returncode, _, stderr = await self.exec_in_container(
container_id, cmd, detach=True
)
if returncode != 0:
if self.logger:
self.logger.error(
message="Failed to start socat in container: {error}",
tag="DOCKER",
params={"error": stderr}
params={"error": stderr},
)
return False
if self.logger:
self.logger.debug(f"Started socat in container: {container_id[:12]}", tag="DOCKER")
self.logger.debug(
f"Started socat in container: {container_id[:12]}", tag="DOCKER"
)
# Wait a moment for socat to start
await asyncio.sleep(1)
return True
async def launch_chrome_in_container(self, container_id: str, browser_args: List[str]) -> bool:
async def launch_chrome_in_container(
self, container_id: str, browser_args: List[str]
) -> bool:
"""Launch Chrome inside the container with specified arguments.
Args:
container_id: ID of the container
browser_args: Chrome command line arguments
Returns:
bool: True if Chrome started successfully, False otherwise
"""
# Build Chrome command
chrome_cmd = ["google-chrome"]
chrome_cmd.extend(browser_args)
returncode, _, stderr = await self.exec_in_container(container_id, chrome_cmd, detach=True)
returncode, _, stderr = await self.exec_in_container(
container_id, chrome_cmd, detach=True
)
if returncode != 0:
if self.logger:
self.logger.error(
message="Failed to launch Chrome in container: {error}",
tag="DOCKER",
params={"error": stderr}
params={"error": stderr},
)
return False
if self.logger:
self.logger.debug(f"Launched Chrome in container: {container_id[:12]}", tag="DOCKER")
self.logger.debug(
f"Launched Chrome in container: {container_id[:12]}", tag="DOCKER"
)
return True
async def get_process_id_in_container(self, container_id: str, process_name: str) -> Optional[int]:
async def get_process_id_in_container(
self, container_id: str, process_name: str
) -> Optional[int]:
"""Get the process ID for a process in the container.
Args:
container_id: ID of the container
process_name: Name pattern to search for
Returns:
int: Process ID if found, None otherwise
"""
cmd = ["pgrep", "-f", process_name]
returncode, stdout, _ = await self.exec_in_container(container_id, cmd)
if returncode == 0 and stdout.strip():
pid = int(stdout.strip().split("\n")[0])
return pid
return None
async def stop_process_in_container(self, container_id: str, pid: int) -> bool:
"""Stop a process in the container by PID.
Args:
container_id: ID of the container
pid: Process ID to stop
Returns:
bool: True if process was stopped, False otherwise
"""
cmd = ["kill", "-TERM", str(pid)]
returncode, _, stderr = await self.exec_in_container(container_id, cmd)
if returncode != 0:
if self.logger:
self.logger.warning(
message="Failed to stop process in container: {error}",
tag="DOCKER",
params={"error": stderr}
params={"error": stderr},
)
return False
if self.logger:
self.logger.debug(f"Stopped process {pid} in container: {container_id[:12]}", tag="DOCKER")
self.logger.debug(
f"Stopped process {pid} in container: {container_id[:12]}", tag="DOCKER"
)
return True
# Network and Port Methods
async def wait_for_cdp_ready(self, host_port: int, timeout: int = 30) -> bool:
async def wait_for_cdp_ready(self, host_port: int, timeout: int = 10) -> dict:
"""Wait for the CDP endpoint to be ready.
Args:
host_port: Port to check for CDP endpoint
timeout: Maximum time to wait in seconds
Returns:
bool: True if CDP endpoint is ready, False if timeout occurred
dict: CDP JSON config if ready, None if timeout occurred
"""
import aiohttp
url = f"http://localhost:{host_port}/json/version"
for _ in range(timeout):
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=1) as response:
if response.status == 200:
if self.logger:
self.logger.debug(f"CDP endpoint ready on port {host_port}", tag="DOCKER")
return True
self.logger.debug(
f"CDP endpoint ready on port {host_port}",
tag="DOCKER",
)
cdp_json_config = await response.json()
if self.logger:
self.logger.debug(
f"CDP JSON config: {cdp_json_config}", tag="DOCKER"
)
return cdp_json_config
except Exception:
pass
await asyncio.sleep(1)
if self.logger:
self.logger.warning(f"CDP endpoint not ready on port {host_port} after {timeout}s timeout", tag="DOCKER")
return False
self.logger.warning(
f"CDP endpoint not ready on port {host_port} after {timeout}s timeout",
tag="DOCKER",
)
return None
def is_port_in_use(self, port: int) -> bool:
"""Check if a port is already in use on the host.
Args:
port: Port number to check
Returns:
bool: True if port is in use, False otherwise
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex(('localhost', port)) == 0
return s.connect_ex(("localhost", port)) == 0
def get_next_available_port(self, start_port: int = 9223) -> int:
"""Get the next available port starting from a given port.
Args:
start_port: Port number to start checking from
Returns:
int: First available port number
"""
@@ -565,18 +644,18 @@ class DockerUtils:
while self.is_port_in_use(port):
port += 1
return port
# Configuration Hash Methods
def generate_config_hash(self, config_dict: Dict) -> str:
"""Generate a hash of the configuration for container matching.
Args:
config_dict: Dictionary of configuration parameters
Returns:
str: Hash string uniquely identifying this configuration
"""
# Convert to canonical JSON string and hash
config_json = json.dumps(config_dict, sort_keys=True)
return hashlib.sha256(config_json.encode()).hexdigest()
return hashlib.sha256(config_json.encode()).hexdigest()

9
crawl4ai/browser/manager.py
View File

@@ -18,15 +18,10 @@ from .strategies import (
BaseBrowserStrategy,
PlaywrightBrowserStrategy,
CDPBrowserStrategy,
BuiltinBrowserStrategy
BuiltinBrowserStrategy,
DockerBrowserStrategy
)
# Import DockerBrowserStrategy if available
try:
from .docker_strategy import DockerBrowserStrategy
except ImportError:
DockerBrowserStrategy = None
class BrowserManager:
"""Main interface for browser management in Crawl4AI.

143
crawl4ai/browser/models.py
View File

@@ -0,0 +1,143 @@
"""Docker configuration module for Crawl4AI browser automation.
This module provides configuration classes for Docker-based browser automation,
allowing flexible configuration of Docker containers for browsing.
"""
from typing import Dict, List, Optional
class DockerConfig:
"""Configuration for Docker-based browser automation.
This class contains Docker-specific settings to avoid cluttering BrowserConfig.
Attributes:
mode (str): Docker operation mode - "connect" or "launch".
- "connect": Uses a container with Chrome already running
- "launch": Dynamically configures and starts Chrome in container
image (str): Docker image to use. If None, defaults from DockerUtils are used.
registry_file (str): Path to container registry file for persistence.
persistent (bool): Keep container running after browser closes.
remove_on_exit (bool): Remove container on exit when not persistent.
network (str): Docker network to use.
volumes (List[str]): Volume mappings (e.g., ["host_path:container_path"]).
env_vars (Dict[str, str]): Environment variables to set in container.
extra_args (List[str]): Additional docker run arguments.
host_port (int): Host port to map to container's 9223 port.
user_data_dir (str): Path to user data directory on host.
container_user_data_dir (str): Path to user data directory in container.
"""
def __init__(
self,
mode: str = "connect", # "connect" or "launch"
image: Optional[str] = None, # Docker image to use
registry_file: Optional[str] = None, # Path to registry file
persistent: bool = False, # Keep container running after browser closes
remove_on_exit: bool = True, # Remove container on exit when not persistent
network: Optional[str] = None, # Docker network to use
volumes: List[str] = None, # Volume mappings
cpu_limit: float = 1.0, # CPU limit for the container
memory_limit: str = "1.5g", # Memory limit for the container
env_vars: Dict[str, str] = None, # Environment variables
host_port: Optional[int] = None, # Host port to map to container's 9223
user_data_dir: Optional[str] = None, # Path to user data directory on host
container_user_data_dir: str = "/data", # Path to user data directory in container
extra_args: List[str] = None, # Additional docker run arguments
):
"""Initialize Docker configuration.
Args:
mode: Docker operation mode ("connect" or "launch")
image: Docker image to use
registry_file: Path to container registry file
persistent: Whether to keep container running after browser closes
remove_on_exit: Whether to remove container on exit when not persistent
network: Docker network to use
volumes: Volume mappings as list of strings
cpu_limit: CPU limit for the container
memory_limit: Memory limit for the container
env_vars: Environment variables as dictionary
extra_args: Additional docker run arguments
host_port: Host port to map to container's 9223
user_data_dir: Path to user data directory on host
container_user_data_dir: Path to user data directory in container
"""
self.mode = mode
self.image = image # If None, defaults will be used from DockerUtils
self.registry_file = registry_file
self.persistent = persistent
self.remove_on_exit = remove_on_exit
self.network = network
self.volumes = volumes or []
self.cpu_limit = cpu_limit
self.memory_limit = memory_limit
self.env_vars = env_vars or {}
self.extra_args = extra_args or []
self.host_port = host_port
self.user_data_dir = user_data_dir
self.container_user_data_dir = container_user_data_dir
def to_dict(self) -> Dict:
"""Convert this configuration to a dictionary.
Returns:
Dictionary representation of this configuration
"""
return {
"mode": self.mode,
"image": self.image,
"registry_file": self.registry_file,
"persistent": self.persistent,
"remove_on_exit": self.remove_on_exit,
"network": self.network,
"volumes": self.volumes,
"cpu_limit": self.cpu_limit,
"memory_limit": self.memory_limit,
"env_vars": self.env_vars,
"extra_args": self.extra_args,
"host_port": self.host_port,
"user_data_dir": self.user_data_dir,
"container_user_data_dir": self.container_user_data_dir
}
@staticmethod
def from_kwargs(kwargs: Dict) -> "DockerConfig":
"""Create a DockerConfig from a dictionary of keyword arguments.
Args:
kwargs: Dictionary of configuration options
Returns:
New DockerConfig instance
"""
return DockerConfig(
mode=kwargs.get("mode", "connect"),
image=kwargs.get("image"),
registry_file=kwargs.get("registry_file"),
persistent=kwargs.get("persistent", False),
remove_on_exit=kwargs.get("remove_on_exit", True),
network=kwargs.get("network"),
volumes=kwargs.get("volumes"),
cpu_limit=kwargs.get("cpu_limit", 1.0),
memory_limit=kwargs.get("memory_limit", "1.5g"),
env_vars=kwargs.get("env_vars"),
extra_args=kwargs.get("extra_args"),
host_port=kwargs.get("host_port"),
user_data_dir=kwargs.get("user_data_dir"),
container_user_data_dir=kwargs.get("container_user_data_dir", "/data")
)
def clone(self, **kwargs) -> "DockerConfig":
"""Create a copy of this configuration with updated values.
Args:
**kwargs: Key-value pairs of configuration options to update
Returns:
DockerConfig: A new instance with the specified updates
"""
config_dict = self.to_dict()
config_dict.update(kwargs)
return DockerConfig.from_kwargs(config_dict)

1256
crawl4ai/browser/strategies.py
View File

File diff suppressed because it is too large Load Diff

13
crawl4ai/browser/strategies/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
from .base import BaseBrowserStrategy
from .cdp import CDPBrowserStrategy
from .docker_strategy import DockerBrowserStrategy
from .playwright import PlaywrightBrowserStrategy
from .builtin import BuiltinBrowserStrategy
__all__ = [
"BrowserStrategy",
"CDPBrowserStrategy",
"DockerBrowserStrategy",
"PlaywrightBrowserStrategy",
"BuiltinBrowserStrategy",
]

270
crawl4ai/browser/strategies/base.py Normal file
View File

@@ -0,0 +1,270 @@
"""Browser strategies module for Crawl4AI.
This module implements the browser strategy pattern for different
browser implementations, including Playwright, CDP, and builtin browsers.
"""
from abc import ABC, abstractmethod
import asyncio
import json
import hashlib
from typing import Optional, Tuple, List
from playwright.async_api import BrowserContext, Page
from ...async_logger import AsyncLogger
from ...async_configs import BrowserConfig, CrawlerRunConfig
from ...config import DOWNLOAD_PAGE_TIMEOUT
from ...js_snippet import load_js_script
class BaseBrowserStrategy(ABC):
"""Base class for all browser strategies.
This abstract class defines the interface that all browser strategies
must implement. It handles common functionality like context caching.
"""
def __init__(self, config: BrowserConfig, logger: Optional[AsyncLogger] = None):
"""Initialize the strategy with configuration and logger.
Args:
config: Browser configuration
logger: Logger for recording events and errors
"""
self.config = config
self.logger = logger
self.browser = None
self.default_context = None
self.contexts_by_config = {}
self._contexts_lock = asyncio.Lock()
self.playwright = None
@abstractmethod
async def start(self):
"""Start the browser.
Returns:
self: For method chaining
"""
pass
@abstractmethod
async def get_page(self, crawlerRunConfig: CrawlerRunConfig) -> Tuple[Page, BrowserContext]:
"""Get a page with specified configuration.
Args:
crawlerRunConfig: Crawler run configuration
Returns:
Tuple of (Page, BrowserContext)
"""
pass
async def get_pages(self, crawlerRunConfig: CrawlerRunConfig, count: int = 1) -> List[Tuple[Page, BrowserContext]]:
"""Get multiple pages with the same configuration.
Args:
crawlerRunConfig: Configuration for the pages
count: Number of pages to create
Returns:
List of (Page, Context) tuples
"""
pages = []
for _ in range(count):
page, context = await self.get_page(crawlerRunConfig)
pages.append((page, context))
return pages
@abstractmethod
async def close(self):
"""Close the browser and clean up resources."""
pass
def _make_config_signature(self, crawlerRunConfig: CrawlerRunConfig) -> str:
"""Create a signature hash from configuration for context caching.
Args:
crawlerRunConfig: Crawler run configuration
Returns:
str: Unique hash for this configuration
"""
config_dict = crawlerRunConfig.__dict__.copy()
# Exclude items that do not affect browser-level setup
ephemeral_keys = [
"session_id",
"js_code",
"scraping_strategy",
"extraction_strategy",
"chunking_strategy",
"cache_mode",
"content_filter",
"semaphore_count",
"url"
]
for key in ephemeral_keys:
if key in config_dict:
del config_dict[key]
# Convert to canonical JSON string
signature_json = json.dumps(config_dict, sort_keys=True, default=str)
# Hash the JSON so we get a compact, unique string
signature_hash = hashlib.sha256(signature_json.encode("utf-8")).hexdigest()
return signature_hash
async def create_browser_context(self, crawlerRunConfig: Optional[CrawlerRunConfig] = None) -> BrowserContext:
"""Creates and returns a new browser context with configured settings.
Args:
crawlerRunConfig: Configuration object for the crawler run
Returns:
BrowserContext: Browser context object with the specified configurations
"""
if not self.browser:
raise ValueError("Browser must be initialized before creating context")
# Base settings
user_agent = self.config.headers.get("User-Agent", self.config.user_agent)
viewport_settings = {
"width": self.config.viewport_width,
"height": self.config.viewport_height,
}
proxy_settings = {"server": self.config.proxy} if self.config.proxy else None
# Define blocked extensions for resource optimization
blocked_extensions = [
# Images
"jpg", "jpeg", "png", "gif", "webp", "svg", "ico", "bmp", "tiff", "psd",
# Fonts
"woff", "woff2", "ttf", "otf", "eot",
# Media
"mp4", "webm", "ogg", "avi", "mov", "wmv", "flv", "m4v", "mp3", "wav", "aac",
"m4a", "opus", "flac",
# Documents
"pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx",
# Archives
"zip", "rar", "7z", "tar", "gz",
# Scripts and data
"xml", "swf", "wasm",
]
# Common context settings
context_settings = {
"user_agent": user_agent,
"viewport": viewport_settings,
"proxy": proxy_settings,
"accept_downloads": self.config.accept_downloads,
"ignore_https_errors": self.config.ignore_https_errors,
"device_scale_factor": 1.0,
"java_script_enabled": self.config.java_script_enabled,
}
# Apply text mode settings if enabled
if self.config.text_mode:
text_mode_settings = {
"has_touch": False,
"is_mobile": False,
# Disable javascript in text mode
"java_script_enabled": False
}
# Update context settings with text mode settings
context_settings.update(text_mode_settings)
if self.logger:
self.logger.debug("Text mode enabled for browser context", tag="BROWSER")
# Handle storage state properly - this is key for persistence
if self.config.storage_state:
context_settings["storage_state"] = self.config.storage_state
if self.logger:
if isinstance(self.config.storage_state, str):
self.logger.debug(f"Using storage state from file: {self.config.storage_state}", tag="BROWSER")
else:
self.logger.debug("Using storage state from config object", tag="BROWSER")
# If user_data_dir is specified, browser persistence should be automatic
if self.config.user_data_dir and self.logger:
self.logger.debug(f"Using user data directory: {self.config.user_data_dir}", tag="BROWSER")
# Apply crawler-specific configurations if provided
if crawlerRunConfig:
# Check if there is value for crawlerRunConfig.proxy_config set add that to context
if crawlerRunConfig.proxy_config:
proxy_settings = {
"server": crawlerRunConfig.proxy_config.server,
}
if crawlerRunConfig.proxy_config.username:
proxy_settings.update({
"username": crawlerRunConfig.proxy_config.username,
"password": crawlerRunConfig.proxy_config.password,
})
context_settings["proxy"] = proxy_settings
# Create and return the context
try:
# Create the context with appropriate settings
context = await self.browser.new_context(**context_settings)
# Apply text mode resource blocking if enabled
if self.config.text_mode:
# Create and apply route patterns for each extension
for ext in blocked_extensions:
await context.route(f"**/*.{ext}", lambda route: route.abort())
return context
except Exception as e:
if self.logger:
self.logger.error(f"Error creating browser context: {str(e)}", tag="BROWSER")
# Fallback to basic context creation if the advanced settings fail
return await self.browser.new_context()
async def setup_context(self, context: BrowserContext, crawlerRunConfig: Optional[CrawlerRunConfig] = None):
"""Set up a browser context with the configured options.
Args:
context: The browser context to set up
crawlerRunConfig: Configuration object containing all browser settings
"""
if self.config.headers:
await context.set_extra_http_headers(self.config.headers)
if self.config.cookies:
await context.add_cookies(self.config.cookies)
if self.config.accept_downloads:
context.set_default_timeout(DOWNLOAD_PAGE_TIMEOUT)
context.set_default_navigation_timeout(DOWNLOAD_PAGE_TIMEOUT)
if self.config.downloads_path:
context._impl_obj._options["accept_downloads"] = True
context._impl_obj._options["downloads_path"] = self.config.downloads_path
# Handle user agent and browser hints
if self.config.user_agent:
combined_headers = {
"User-Agent": self.config.user_agent,
"sec-ch-ua": self.config.browser_hint,
}
combined_headers.update(self.config.headers)
await context.set_extra_http_headers(combined_headers)
# Add default cookie
await context.add_cookies(
[
{
"name": "cookiesEnabled",
"value": "true",
"url": crawlerRunConfig and crawlerRunConfig.url or "https://crawl4ai.com/",
}
]
)
# Handle navigator overrides
if crawlerRunConfig:
if (
crawlerRunConfig.override_navigator
or crawlerRunConfig.simulate_user
or crawlerRunConfig.magic
):
await context.add_init_script(load_js_script("navigator_overrider"))

394
crawl4ai/browser/strategies/builtin.py Normal file
View File

@@ -0,0 +1,394 @@
import asyncio
import os
import time
import json
import subprocess
import shutil
import signal
from typing import Optional, Dict, Any
from ...async_logger import AsyncLogger
from ...async_configs import BrowserConfig
from ...utils import get_home_folder
from ..utils import get_browser_executable, is_windows, is_browser_running
from .cdp import CDPBrowserStrategy
class BuiltinBrowserStrategy(CDPBrowserStrategy):
"""Built-in browser strategy.
This strategy extends the CDP strategy to use the built-in browser.
"""
def __init__(self, config: BrowserConfig, logger: Optional[AsyncLogger] = None):
"""Initialize the built-in browser strategy.
Args:
config: Browser configuration
logger: Logger for recording events and errors
"""
super().__init__(config, logger)
self.builtin_browser_dir = os.path.join(get_home_folder(), "builtin-browser") if not self.config.user_data_dir else self.config.user_data_dir
self.builtin_config_file = os.path.join(self.builtin_browser_dir, "browser_config.json")
# Raise error if user data dir is already engaged
if self._check_user_dir_is_engaged(self.builtin_browser_dir):
raise Exception(f"User data directory {self.builtin_browser_dir} is already engaged by another browser instance.")
os.makedirs(self.builtin_browser_dir, exist_ok=True)
def _check_user_dir_is_engaged(self, user_data_dir: str) -> bool:
"""Check if the user data directory is already in use.
Returns:
bool: True if the directory is engaged, False otherwise
"""
# Load browser config file, then iterate in port_map values, check "user_data_dir" key if it matches
# the current user data directory
if os.path.exists(self.builtin_config_file):
try:
with open(self.builtin_config_file, 'r') as f:
browser_info_dict = json.load(f)
# Check if user data dir is already engaged
for port_str, browser_info in browser_info_dict.get("port_map", {}).items():
if browser_info.get("user_data_dir") == user_data_dir:
return True
except Exception as e:
if self.logger:
self.logger.error(f"Error reading built-in browser config: {str(e)}", tag="BUILTIN")
return False
async def start(self):
"""Start or connect to the built-in browser.
Returns:
self: For method chaining
"""
# Check for existing built-in browser (get_browser_info already checks if running)
browser_info = self.get_browser_info()
if browser_info:
if self.logger:
self.logger.info(f"Using existing built-in browser at {browser_info.get('cdp_url')}", tag="BROWSER")
self.config.cdp_url = browser_info.get('cdp_url')
else:
if self.logger:
self.logger.info("Built-in browser not found, launching new instance...", tag="BROWSER")
cdp_url = await self.launch_builtin_browser(
browser_type=self.config.browser_type,
debugging_port=self.config.debugging_port,
headless=self.config.headless,
)
if not cdp_url:
if self.logger:
self.logger.warning("Failed to launch built-in browser, falling back to regular CDP strategy", tag="BROWSER")
return await super().start()
self.config.cdp_url = cdp_url
# Call parent class implementation with updated CDP URL
return await super().start()
@classmethod
def get_builtin_browser_info(cls, debugging_port: int, config_file: str, logger: Optional[AsyncLogger] = None) -> Optional[Dict[str, Any]]:
"""Get information about the built-in browser for a specific debugging port.
Args:
debugging_port: The debugging port to look for
config_file: Path to the config file
logger: Optional logger for recording events
Returns:
dict: Browser information or None if no running browser is configured for this port
"""
if not os.path.exists(config_file):
return None
try:
with open(config_file, 'r') as f:
browser_info_dict = json.load(f)
# Get browser info from port map
if isinstance(browser_info_dict, dict) and "port_map" in browser_info_dict:
port_str = str(debugging_port)
if port_str in browser_info_dict["port_map"]:
browser_info = browser_info_dict["port_map"][port_str]
# Check if the browser is still running
if not is_browser_running(browser_info.get('pid')):
if logger:
logger.warning(f"Built-in browser on port {debugging_port} is not running", tag="BUILTIN")
# Remove this port from the dictionary
del browser_info_dict["port_map"][port_str]
with open(config_file, 'w') as f:
json.dump(browser_info_dict, f, indent=2)
return None
return browser_info
return None
except Exception as e:
if logger:
logger.error(f"Error reading built-in browser config: {str(e)}", tag="BUILTIN")
return None
def get_browser_info(self) -> Optional[Dict[str, Any]]:
"""Get information about the current built-in browser instance.
Returns:
dict: Browser information or None if no running browser is configured
"""
return self.get_builtin_browser_info(
debugging_port=self.config.debugging_port,
config_file=self.builtin_config_file,
logger=self.logger
)
async def launch_builtin_browser(self,
browser_type: str = "chromium",
debugging_port: int = 9222,
headless: bool = True) -> Optional[str]:
"""Launch a browser in the background for use as the built-in browser.
Args:
browser_type: Type of browser to launch ('chromium' or 'firefox')
debugging_port: Port to use for CDP debugging
headless: Whether to run in headless mode
Returns:
str: CDP URL for the browser, or None if launch failed
"""
# Check if there's an existing browser still running
browser_info = self.get_builtin_browser_info(
debugging_port=debugging_port,
config_file=self.builtin_config_file,
logger=self.logger
)
if browser_info:
if self.logger:
self.logger.info(f"Built-in browser is already running on port {debugging_port}", tag="BUILTIN")
return browser_info.get('cdp_url')
# Create a user data directory for the built-in browser
user_data_dir = os.path.join(self.builtin_browser_dir, "user_data")
# Raise error if user data dir is already engaged
if self._check_user_dir_is_engaged(user_data_dir):
raise Exception(f"User data directory {user_data_dir} is already engaged by another browser instance.")
# Create the user data directory if it doesn't exist
os.makedirs(user_data_dir, exist_ok=True)
# Prepare browser launch arguments
browser_path = await get_browser_executable(browser_type)
if browser_type == "chromium":
args = [
browser_path,
f"--remote-debugging-port={debugging_port}",
f"--user-data-dir={user_data_dir}",
]
if headless:
args.append("--headless=new")
elif browser_type == "firefox":
args = [
browser_path,
"--remote-debugging-port",
str(debugging_port),
"--profile",
user_data_dir,
]
if headless:
args.append("--headless")
else:
if self.logger:
self.logger.error(f"Browser type {browser_type} not supported for built-in browser", tag="BUILTIN")
return None
try:
# Start the browser process detached
if is_windows():
process = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
creationflags=subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP
)
else:
process = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
preexec_fn=os.setpgrp # Start in a new process group
)
# Wait briefly to ensure the process starts successfully
await asyncio.sleep(2.0)
# Check if the process is still running
if process.poll() is not None:
if self.logger:
self.logger.error(f"Browser process exited immediately with code {process.returncode}", tag="BUILTIN")
return None
# Construct CDP URL
cdp_url = f"http://localhost:{debugging_port}"
# Try to verify browser is responsive by fetching version info
import aiohttp
json_url = f"{cdp_url}/json/version"
config_json = None
try:
async with aiohttp.ClientSession() as session:
for _ in range(10): # Try multiple times
try:
async with session.get(json_url) as response:
if response.status == 200:
config_json = await response.json()
break
except Exception:
pass
await asyncio.sleep(0.5)
except Exception as e:
if self.logger:
self.logger.warning(f"Could not verify browser: {str(e)}", tag="BUILTIN")
# Create browser info
browser_info = {
'pid': process.pid,
'cdp_url': cdp_url,
'user_data_dir': user_data_dir,
'browser_type': browser_type,
'debugging_port': debugging_port,
'start_time': time.time(),
'config': config_json
}
# Read existing config file if it exists
port_map = {}
if os.path.exists(self.builtin_config_file):
try:
with open(self.builtin_config_file, 'r') as f:
existing_data = json.load(f)
# Check if it already uses port mapping
if isinstance(existing_data, dict) and "port_map" in existing_data:
port_map = existing_data["port_map"]
# Convert legacy format to port mapping
elif isinstance(existing_data, dict) and "debugging_port" in existing_data:
old_port = str(existing_data.get("debugging_port"))
if self._is_browser_running(existing_data.get("pid")):
port_map[old_port] = existing_data
except Exception as e:
if self.logger:
self.logger.warning(f"Could not read existing config: {str(e)}", tag="BUILTIN")
# Add/update this browser in the port map
port_map[str(debugging_port)] = browser_info
# Write updated config
with open(self.builtin_config_file, 'w') as f:
json.dump({"port_map": port_map}, f, indent=2)
# Detach from the browser process - don't keep any references
# This is important to allow the Python script to exit while the browser continues running
process = None
if self.logger:
self.logger.success(f"Built-in browser launched at CDP URL: {cdp_url}", tag="BUILTIN")
return cdp_url
except Exception as e:
if self.logger:
self.logger.error(f"Error launching built-in browser: {str(e)}", tag="BUILTIN")
return None
async def kill_builtin_browser(self) -> bool:
"""Kill the built-in browser if it's running.
Returns:
bool: True if the browser was killed, False otherwise
"""
browser_info = self.get_browser_info()
if not browser_info:
if self.logger:
self.logger.warning(f"No built-in browser found on port {self.config.debugging_port}", tag="BUILTIN")
return False
pid = browser_info.get('pid')
if not pid:
return False
try:
if is_windows():
subprocess.run(["taskkill", "/F", "/PID", str(pid)], check=True)
else:
os.kill(pid, signal.SIGTERM)
# Wait for termination
for _ in range(5):
if not is_browser_running(pid):
break
await asyncio.sleep(0.5)
else:
# Force kill if still running
os.kill(pid, signal.SIGKILL)
# Update config file to remove this browser
with open(self.builtin_config_file, 'r') as f:
browser_info_dict = json.load(f)
# Remove this port from the dictionary
port_str = str(self.config.debugging_port)
if port_str in browser_info_dict.get("port_map", {}):
del browser_info_dict["port_map"][port_str]
with open(self.builtin_config_file, 'w') as f:
json.dump(browser_info_dict, f, indent=2)
# Remove user data directory if it exists
if os.path.exists(self.builtin_browser_dir):
shutil.rmtree(self.builtin_browser_dir)
# Clear the browser info cache
self.browser = None
self.temp_dir = None
self.shutting_down = True
if self.logger:
self.logger.success("Built-in browser terminated", tag="BUILTIN")
return True
except Exception as e:
if self.logger:
self.logger.error(f"Error killing built-in browser: {str(e)}", tag="BUILTIN")
return False
async def get_builtin_browser_status(self) -> Dict[str, Any]:
"""Get status information about the built-in browser.
Returns:
dict: Status information with running, cdp_url, and info fields
"""
browser_info = self.get_browser_info()
if not browser_info:
return {
'running': False,
'cdp_url': None,
'info': None,
'port': self.config.debugging_port
}
return {
'running': True,
'cdp_url': browser_info.get('cdp_url'),
'info': browser_info,
'port': self.config.debugging_port
}
# Override the close method to handle built-in browser cleanup
async def close(self):
"""Close the built-in browser and clean up resources."""
# Call parent class close method
await super().close()
# Clean up built-in browser if we created it
if self.shutting_down:
await self.kill_builtin_browser()

359
crawl4ai/browser/strategies/cdp.py Normal file
View File

@@ -0,0 +1,359 @@
"""Browser strategies module for Crawl4AI.
This module implements the browser strategy pattern for different
browser implementations, including Playwright, CDP, and builtin browsers.
"""
import asyncio
import os
import time
import json
import subprocess
import shutil
from typing import Optional, Tuple, List
from playwright.async_api import BrowserContext, Page
from ...async_logger import AsyncLogger
from ...async_configs import BrowserConfig, CrawlerRunConfig
from ..utils import get_playwright, get_browser_executable, create_temp_directory, is_windows
from .base import BaseBrowserStrategy
class CDPBrowserStrategy(BaseBrowserStrategy):
"""CDP-based browser strategy.
This strategy connects to an existing browser using CDP protocol or
launches and connects to a browser using CDP.
"""
def __init__(self, config: BrowserConfig, logger: Optional[AsyncLogger] = None):
"""Initialize the CDP browser strategy.
Args:
config: Browser configuration
logger: Logger for recording events and errors
"""
super().__init__(config, logger)
self.sessions = {}
self.session_ttl = 1800 # 30 minutes
self.browser_process = None
self.temp_dir = None
self.shutting_down = False
async def start(self):
"""Start or connect to the browser using CDP.
Returns:
self: For method chaining
"""
self.playwright = await get_playwright()
# Get or create CDP URL
cdp_url = await self._get_or_create_cdp_url()
# Connect to the browser using CDP
self.browser = await self.playwright.chromium.connect_over_cdp(cdp_url)
# Get or create default context
contexts = self.browser.contexts
if contexts:
self.default_context = contexts[0]
else:
self.default_context = await self.create_browser_context()
await self.setup_context(self.default_context)
return self
async def _get_or_create_cdp_url(self) -> str:
"""Get existing CDP URL or launch a browser and return its CDP URL.
Returns:
str: CDP URL for connecting to the browser
"""
# If CDP URL is provided, just return it
if self.config.cdp_url:
return self.config.cdp_url
# Create temp dir if needed
if not self.config.user_data_dir:
self.temp_dir = create_temp_directory()
user_data_dir = self.temp_dir
else:
user_data_dir = self.config.user_data_dir
# Get browser args based on OS and browser type
args = await self._get_browser_args(user_data_dir)
# Start browser process
try:
# Use DETACHED_PROCESS flag on Windows to fully detach the process
# On Unix, we'll use preexec_fn=os.setpgrp to start the process in a new process group
if is_windows():
self.browser_process = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
creationflags=subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP
)
else:
self.browser_process = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
preexec_fn=os.setpgrp # Start in a new process group
)
# Monitor for a short time to make sure it starts properly
await asyncio.sleep(0.5) # Give browser time to start
await self._initial_startup_check()
await asyncio.sleep(2) # Give browser more time to start
return f"http://localhost:{self.config.debugging_port}"
except Exception as e:
await self._cleanup_process()
raise Exception(f"Failed to start browser: {e}")
async def _initial_startup_check(self):
"""Perform a quick check to make sure the browser started successfully."""
if not self.browser_process:
return
# Check that process started without immediate termination
await asyncio.sleep(0.5)
if self.browser_process.poll() is not None:
# Process already terminated
stdout, stderr = b"", b""
try:
stdout, stderr = self.browser_process.communicate(timeout=0.5)
except subprocess.TimeoutExpired:
pass
if self.logger:
self.logger.error(
message="Browser process terminated during startup | Code: {code} | STDOUT: {stdout} | STDERR: {stderr}",
tag="ERROR",
params={
"code": self.browser_process.returncode,
"stdout": stdout.decode() if stdout else "",
"stderr": stderr.decode() if stderr else "",
},
)
async def _get_browser_args(self, user_data_dir: str) -> List[str]:
"""Returns browser-specific command line arguments.
Args:
user_data_dir: Path to user data directory
Returns:
List of command-line arguments for the browser
"""
browser_path = await get_browser_executable(self.config.browser_type)
base_args = [browser_path]
if self.config.browser_type == "chromium":
args = [
f"--remote-debugging-port={self.config.debugging_port}",
f"--user-data-dir={user_data_dir}",
]
if self.config.headless:
args.append("--headless=new")
elif self.config.browser_type == "firefox":
args = [
"--remote-debugging-port",
str(self.config.debugging_port),
"--profile",
user_data_dir,
]
if self.config.headless:
args.append("--headless")
else:
raise NotImplementedError(f"Browser type {self.config.browser_type} not supported")
return base_args + args
async def _cleanup_process(self):
"""Cleanup browser process and temporary directory."""
# Set shutting_down flag BEFORE any termination actions
self.shutting_down = True
if self.browser_process:
try:
# Only terminate if we have proper control over the process
if not self.browser_process.poll():
# Process is still running
self.browser_process.terminate()
# Wait for process to end gracefully
for _ in range(10): # 10 attempts, 100ms each
if self.browser_process.poll() is not None:
break
await asyncio.sleep(0.1)
# Force kill if still running
if self.browser_process.poll() is None:
if is_windows():
# On Windows we might need taskkill for detached processes
try:
subprocess.run(["taskkill", "/F", "/PID", str(self.browser_process.pid)])
except Exception:
self.browser_process.kill()
else:
self.browser_process.kill()
await asyncio.sleep(0.1) # Brief wait for kill to take effect
except Exception as e:
if self.logger:
self.logger.error(
message="Error terminating browser: {error}",
tag="ERROR",
params={"error": str(e)},
)
if self.temp_dir and os.path.exists(self.temp_dir):
try:
shutil.rmtree(self.temp_dir)
except Exception as e:
if self.logger:
self.logger.error(
message="Error removing temporary directory: {error}",
tag="ERROR",
params={"error": str(e)},
)
async def create_browser_context(self, crawlerRunConfig: Optional[CrawlerRunConfig] = None) -> BrowserContext:
"""Create a new browser context.
Uses the base class implementation which handles all configurations.
Args:
crawlerRunConfig: Configuration object for the crawler run
Returns:
BrowserContext: Browser context object
"""
# Handle user_data_dir for CDP browsers
if self.config.user_data_dir:
# For CDP-based browsers, storage persistence is typically handled by the user_data_dir
# at the browser level, but we'll create a storage_state location for Playwright as well
storage_path = os.path.join(self.config.user_data_dir, "storage_state.json")
if not os.path.exists(storage_path):
# Create parent directory if it doesn't exist
os.makedirs(os.path.dirname(storage_path), exist_ok=True)
with open(storage_path, "w") as f:
json.dump({}, f)
self.config.storage_state = storage_path
# Use the base class implementation
return await super().create_browser_context(crawlerRunConfig)
def _cleanup_expired_sessions(self):
"""Clean up expired sessions based on TTL."""
current_time = time.time()
expired_sessions = [
sid
for sid, (_, _, last_used) in self.sessions.items()
if current_time - last_used > self.session_ttl
]
for sid in expired_sessions:
asyncio.create_task(self._kill_session(sid))
async def _kill_session(self, session_id: str):
"""Kill a browser session and clean up resources.
Args:
session_id: The session ID to kill
"""
if session_id in self.sessions:
context, page, _ = self.sessions[session_id]
await page.close()
del self.sessions[session_id]
async def get_page(self, crawlerRunConfig: CrawlerRunConfig) -> Tuple[Page, BrowserContext]:
"""Get a page for the given configuration.
Args:
crawlerRunConfig: Configuration object for the crawler run
Returns:
Tuple of (Page, BrowserContext)
"""
self._cleanup_expired_sessions()
# If a session_id is provided and we already have it, reuse that page + context
if crawlerRunConfig.session_id and crawlerRunConfig.session_id in self.sessions:
context, page, _ = self.sessions[crawlerRunConfig.session_id]
# Update last-used timestamp
self.sessions[crawlerRunConfig.session_id] = (context, page, time.time())
return page, context
# For CDP, we typically use the shared default_context
context = self.default_context
pages = context.pages
page = next((p for p in pages if p.url == crawlerRunConfig.url), None)
if not page:
page = await context.new_page()
# If a session_id is specified, store this session so we can reuse later
if crawlerRunConfig.session_id:
self.sessions[crawlerRunConfig.session_id] = (context, page, time.time())
return page, context
async def close(self):
"""Close the browser and clean up resources."""
# Skip cleanup if using external CDP URL and not launched by us
if self.config.cdp_url and not self.browser_process:
return
if self.config.sleep_on_close:
await asyncio.sleep(0.5)
# If we have a user_data_dir configured, ensure persistence of storage state
if self.config.user_data_dir and self.browser and self.default_context:
for context in self.browser.contexts:
try:
await context.storage_state(path=os.path.join(self.config.user_data_dir, "Default", "storage_state.json"))
if self.logger:
self.logger.debug("Ensuring storage state is persisted before closing browser", tag="BROWSER")
except Exception as e:
if self.logger:
self.logger.warning(
message="Failed to ensure storage persistence: {error}",
tag="BROWSER",
params={"error": str(e)}
)
# Close all sessions
session_ids = list(self.sessions.keys())
for session_id in session_ids:
await self._kill_session(session_id)
# Close browser
if self.browser:
await self.browser.close()
self.browser = None
# Clean up managed browser if we created it
if self.browser_process:
await asyncio.sleep(0.5)
await self._cleanup_process()
self.browser_process = None
# Close temporary directory
if self.temp_dir and os.path.exists(self.temp_dir):
try:
shutil.rmtree(self.temp_dir)
self.temp_dir = None
except Exception as e:
if self.logger:
self.logger.error(
message="Error removing temporary directory: {error}",
tag="ERROR",
params={"error": str(e)},
)
# Stop playwright
if self.playwright:
await self.playwright.stop()
self.playwright = None

205
crawl4ai/browser/docker_strategy.py → crawl4ai/browser/strategies/docker_strategy.py
View File

@@ -6,18 +6,15 @@ which offers better isolation, consistency across platforms, and easy scaling.
import os
import uuid
import asyncio
from typing import Dict, List, Optional, Tuple, Union
from pathlib import Path
from typing import List, Optional
from playwright.async_api import Page, BrowserContext
from ..async_logger import AsyncLogger
from ..async_configs import BrowserConfig, CrawlerRunConfig
from .docker_config import DockerConfig
from .docker_registry import DockerRegistry
from .docker_utils import DockerUtils
from .strategies import BuiltinBrowserStrategy
from ...async_logger import AsyncLogger
from ...async_configs import BrowserConfig
from ..models import DockerConfig
from ..docker_registry import DockerRegistry
from ..docker_utils import DockerUtils
from .builtin import BuiltinBrowserStrategy
class DockerBrowserStrategy(BuiltinBrowserStrategy):
@@ -53,6 +50,16 @@ class DockerBrowserStrategy(BuiltinBrowserStrategy):
self.docker_config = self.config.docker_config or DockerConfig()
self.container_id = None
self.container_name = f"crawl4ai-browser-{uuid.uuid4().hex[:8]}"
# Use the shared registry file path for consistency with BuiltinBrowserStrategy
registry_file = self.docker_config.registry_file
if registry_file is None and self.config.user_data_dir:
# Use the same registry file as BuiltinBrowserStrategy if possible
registry_file = os.path.join(
os.path.dirname(self.config.user_data_dir),
"browser_config.json"
)
self.registry = DockerRegistry(self.docker_config.registry_file)
self.docker_utils = DockerUtils(logger)
self.chrome_process_id = None
@@ -60,7 +67,77 @@ class DockerBrowserStrategy(BuiltinBrowserStrategy):
self.internal_cdp_port = 9222 # Chrome's internal CDP port
self.internal_mapped_port = 9223 # Port that socat maps to internally
self.shutting_down = False
async def start(self):
"""Start or connect to a browser running in a Docker container.
This method initializes Playwright and establishes a connection to
a browser running in a Docker container. Depending on the configured mode:
- "connect": Connects to a container with Chrome already running
- "launch": Creates a container and launches Chrome within it
Returns:
self: For method chaining
"""
# Initialize Playwright
from ..utils import get_playwright
self.playwright = await get_playwright()
if self.logger:
self.logger.info(
f"Starting Docker browser strategy in {self.docker_config.mode} mode",
tag="DOCKER"
)
try:
# Get CDP URL by creating or reusing a Docker container
# This handles the container management and browser startup
cdp_url = await self._get_or_create_cdp_url()
if not cdp_url:
raise Exception("Failed to establish CDP connection to Docker container")
if self.logger:
self.logger.info(f"Connecting to browser in Docker via CDP: {cdp_url}", tag="DOCKER")
# Connect to the browser using CDP
self.browser = await self.playwright.chromium.connect_over_cdp(cdp_url)
# Get existing context or create default context
contexts = self.browser.contexts
if contexts:
self.default_context = contexts[0]
if self.logger:
self.logger.debug("Using existing browser context", tag="DOCKER")
else:
if self.logger:
self.logger.debug("Creating new browser context", tag="DOCKER")
self.default_context = await self.create_browser_context()
await self.setup_context(self.default_context)
return self
except Exception as e:
# Clean up resources if startup fails
if self.container_id and not self.docker_config.persistent:
if self.logger:
self.logger.warning(
f"Cleaning up container after failed start: {self.container_id[:12]}",
tag="DOCKER"
)
await self.docker_utils.remove_container(self.container_id)
self.registry.unregister_container(self.container_id)
self.container_id = None
if self.playwright:
await self.playwright.stop()
self.playwright = None
# Re-raise the exception
if self.logger:
self.logger.error(f"Failed to start Docker browser: {str(e)}", tag="DOCKER")
raise
async def _generate_config_hash(self) -> str:
"""Generate a hash of the configuration for container matching.
@@ -87,7 +164,7 @@ class DockerBrowserStrategy(BuiltinBrowserStrategy):
# Use the utility method to generate the hash
return self.docker_utils.generate_config_hash(config_dict)
async def _get_or_create_cdp_url(self) -> str:
async def _get_or_create_cdp_url1(self) -> str:
"""Get CDP URL by either creating a new container or using an existing one.
Returns:
@@ -183,7 +260,109 @@ class DockerBrowserStrategy(BuiltinBrowserStrategy):
# Return CDP URL
return f"http://localhost:{host_port}"
async def _get_or_create_cdp_url(self) -> str:
"""Get CDP URL by either creating a new container or using an existing one.
Returns:
CDP URL for connecting to the browser
Raises:
Exception: If container creation or browser launch fails
"""
# If CDP URL is explicitly provided, use it
if self.config.cdp_url:
return self.config.cdp_url
# Ensure Docker image exists (will build if needed)
image_name = await self.docker_utils.ensure_docker_image_exists(
self.docker_config.image,
self.docker_config.mode
)
# Generate config hash for container matching
config_hash = await self._generate_config_hash()
# Look for existing container with matching config
container_id = await self.registry.find_container_by_config(config_hash, self.docker_utils)
if container_id:
# Use existing container
self.container_id = container_id
host_port = self.registry.get_container_host_port(container_id)
if self.logger:
self.logger.info(f"Using existing Docker container: {container_id[:12]}", tag="DOCKER")
else:
# Get a port for the new container
host_port = self.docker_config.host_port or self.registry.get_next_available_port(self.docker_utils)
# Prepare volumes list
volumes = list(self.docker_config.volumes)
# Add user data directory if specified
if self.docker_config.user_data_dir:
# Ensure user data directory exists
os.makedirs(self.docker_config.user_data_dir, exist_ok=True)
volumes.append(f"{self.docker_config.user_data_dir}:{self.docker_config.container_user_data_dir}")
# Update config user_data_dir to point to container path
self.config.user_data_dir = self.docker_config.container_user_data_dir
# Create a new container
container_id = await self.docker_utils.create_container(
image_name=image_name,
host_port=host_port,
container_name=self.container_name,
volumes=volumes,
network=self.docker_config.network,
env_vars=self.docker_config.env_vars,
cpu_limit=self.docker_config.cpu_limit,
memory_limit=self.docker_config.memory_limit,
extra_args=self.docker_config.extra_args
)
if not container_id:
raise Exception("Failed to create Docker container")
self.container_id = container_id
# Wait for container to be ready
await self.docker_utils.wait_for_container_ready(container_id)
# Handle specific setup based on mode
if self.docker_config.mode == "launch":
# In launch mode, we need to start socat and Chrome
await self.docker_utils.start_socat_in_container(container_id)
# Build browser arguments
browser_args = self._build_browser_args()
# Launch Chrome
await self.docker_utils.launch_chrome_in_container(container_id, browser_args)
# Get PIDs for later cleanup
self.chrome_process_id = await self.docker_utils.get_process_id_in_container(
container_id, "chrome"
)
self.socat_process_id = await self.docker_utils.get_process_id_in_container(
container_id, "socat"
)
# Wait for CDP to be ready
cdp_json_config = await self.docker_utils.wait_for_cdp_ready(host_port)
if cdp_json_config:
# Register the container in the shared registry
self.registry.register_container(container_id, host_port, config_hash, cdp_json_config)
else:
raise Exception("Failed to get CDP JSON config from Docker container")
if self.logger:
self.logger.success(f"Docker container ready: {container_id[:12]} on port {host_port}", tag="DOCKER")
# Return CDP URL
return f"http://localhost:{host_port}"
def _build_browser_args(self) -> List[str]:
"""Build Chrome command line arguments based on BrowserConfig.

284
crawl4ai/browser/strategies/playwright.py Normal file
View File

@@ -0,0 +1,284 @@
"""Browser strategies module for Crawl4AI.
This module implements the browser strategy pattern for different
browser implementations, including Playwright, CDP, and builtin browsers.
"""
import asyncio
import os
import time
import json
from typing import Optional, Tuple
from playwright.async_api import BrowserContext, Page, ProxySettings
from ...async_logger import AsyncLogger
from ...async_configs import BrowserConfig, CrawlerRunConfig
from ..utils import get_playwright, get_browser_disable_options
from playwright_stealth import StealthConfig
from .base import BaseBrowserStrategy
stealth_config = StealthConfig(
webdriver=True,
chrome_app=True,
chrome_csi=True,
chrome_load_times=True,
chrome_runtime=True,
navigator_languages=True,
navigator_plugins=True,
navigator_permissions=True,
webgl_vendor=True,
outerdimensions=True,
navigator_hardware_concurrency=True,
media_codecs=True,
)
class PlaywrightBrowserStrategy(BaseBrowserStrategy):
"""Standard Playwright browser strategy.
This strategy launches a new browser instance using Playwright
and manages browser contexts.
"""
def __init__(self, config: BrowserConfig, logger: Optional[AsyncLogger] = None):
"""Initialize the Playwright browser strategy.
Args:
config: Browser configuration
logger: Logger for recording events and errors
"""
super().__init__(config, logger)
# Add session management
self.sessions = {}
self.session_ttl = 1800 # 30 minutes
async def start(self):
"""Start the browser instance.
Returns:
self: For method chaining
"""
self.playwright = await get_playwright()
browser_args = self._build_browser_args()
# Launch appropriate browser type
if self.config.browser_type == "firefox":
self.browser = await self.playwright.firefox.launch(**browser_args)
elif self.config.browser_type == "webkit":
self.browser = await self.playwright.webkit.launch(**browser_args)
else:
self.browser = await self.playwright.chromium.launch(**browser_args)
self.default_context = self.browser
return self
def _build_browser_args(self) -> dict:
"""Build browser launch arguments from config.
Returns:
dict: Browser launch arguments
"""
args = [
"--disable-gpu",
"--disable-gpu-compositing",
"--disable-software-rasterizer",
"--no-sandbox",
"--disable-dev-shm-usage",
"--no-first-run",
"--no-default-browser-check",
"--disable-infobars",
"--window-position=0,0",
"--ignore-certificate-errors",
"--ignore-certificate-errors-spki-list",
"--disable-blink-features=AutomationControlled",
"--window-position=400,0",
"--disable-renderer-backgrounding",
"--disable-ipc-flooding-protection",
"--force-color-profile=srgb",
"--mute-audio",
"--disable-background-timer-throttling",
f"--window-size={self.config.viewport_width},{self.config.viewport_height}",
]
if self.config.light_mode:
args.extend(get_browser_disable_options())
if self.config.text_mode:
args.extend(
[
"--blink-settings=imagesEnabled=false",
"--disable-remote-fonts",
"--disable-images",
"--disable-javascript",
"--disable-software-rasterizer",
"--disable-dev-shm-usage",
]
)
if self.config.extra_args:
args.extend(self.config.extra_args)
browser_args = {"headless": self.config.headless, "args": args}
if self.config.chrome_channel:
browser_args["channel"] = self.config.chrome_channel
if self.config.accept_downloads:
browser_args["downloads_path"] = self.config.downloads_path or os.path.join(
os.getcwd(), "downloads"
)
os.makedirs(browser_args["downloads_path"], exist_ok=True)
if self.config.proxy or self.config.proxy_config:
proxy_settings = (
ProxySettings(server=self.config.proxy)
if self.config.proxy
else ProxySettings(
server=self.config.proxy_config.server,
username=self.config.proxy_config.username,
password=self.config.proxy_config.password,
)
)
browser_args["proxy"] = proxy_settings
return browser_args
async def create_browser_context(self, crawlerRunConfig: Optional[CrawlerRunConfig] = None) -> BrowserContext:
"""Creates and returns a new browser context with configured settings.
This implementation extends the base class version to handle user_data_dir specifically.
Args:
crawlerRunConfig: Configuration object for the crawler run
Returns:
BrowserContext: Browser context object with the specified configurations
"""
# Handle user_data_dir explicitly to ensure storage persistence
if self.config.user_data_dir:
# Create a storage state file path if none exists
storage_path = os.path.join(self.config.user_data_dir, "Default", "storage_state.json")
# Create the file if it doesn't exist
if not os.path.exists(storage_path):
os.makedirs(os.path.dirname(storage_path), exist_ok=True)
with open(storage_path, "w") as f:
json.dump({}, f)
# Override storage_state with our specific path
self.config.storage_state = storage_path
if self.logger:
self.logger.debug(f"Using persistent storage state at: {storage_path}", tag="BROWSER")
# Now call the base class implementation which handles everything else
return await super().create_browser_context(crawlerRunConfig)
def _cleanup_expired_sessions(self):
"""Clean up expired sessions based on TTL."""
current_time = time.time()
expired_sessions = [
sid
for sid, (_, _, last_used) in self.sessions.items()
if current_time - last_used > self.session_ttl
]
for sid in expired_sessions:
asyncio.create_task(self._kill_session(sid))
async def _kill_session(self, session_id: str):
"""Kill a browser session and clean up resources.
Args:
session_id: The session ID to kill
"""
if session_id in self.sessions:
context, page, _ = self.sessions[session_id]
await page.close()
del self.sessions[session_id]
async def get_page(self, crawlerRunConfig: CrawlerRunConfig) -> Tuple[Page, BrowserContext]:
"""Get a page for the given configuration.
Args:
crawlerRunConfig: Configuration object for the crawler run
Returns:
Tuple of (Page, BrowserContext)
"""
# Clean up expired sessions first
self._cleanup_expired_sessions()
# If a session_id is provided and we already have it, reuse that page + context
if crawlerRunConfig.session_id and crawlerRunConfig.session_id in self.sessions:
context, page, _ = self.sessions[crawlerRunConfig.session_id]
# Update last-used timestamp
self.sessions[crawlerRunConfig.session_id] = (context, page, time.time())
return page, context
# Otherwise, check if we have an existing context for this config
config_signature = self._make_config_signature(crawlerRunConfig)
async with self._contexts_lock:
if config_signature in self.contexts_by_config:
context = self.contexts_by_config[config_signature]
else:
# Create and setup a new context
context = await self.create_browser_context(crawlerRunConfig)
await self.setup_context(context, crawlerRunConfig)
self.contexts_by_config[config_signature] = context
# Create a new page from the chosen context
page = await context.new_page()
# If a session_id is specified, store this session so we can reuse later
if crawlerRunConfig.session_id:
self.sessions[crawlerRunConfig.session_id] = (context, page, time.time())
return page, context
async def close(self):
"""Close the browser and clean up resources."""
if self.config.sleep_on_close:
await asyncio.sleep(0.5)
# If we have a user_data_dir configured, ensure persistence of storage state
if self.config.user_data_dir and self.browser and self.default_context:
for context in self.browser.contexts:
try:
await context.storage_state(path=os.path.join(self.config.user_data_dir, "Default", "storage_state.json"))
if self.logger:
self.logger.debug("Ensuring storage state is persisted before closing browser", tag="BROWSER")
except Exception as e:
if self.logger:
self.logger.warning(
message="Failed to ensure storage persistence: {error}",
tag="BROWSER",
params={"error": str(e)}
)
# Close all sessions
session_ids = list(self.sessions.keys())
for session_id in session_ids:
await self._kill_session(session_id)
# Close all contexts we created
for ctx in self.contexts_by_config.values():
try:
await ctx.close()
except Exception as e:
if self.logger:
self.logger.error(
message="Error closing context: {error}",
tag="ERROR",
params={"error": str(e)}
)
self.contexts_by_config.clear()
if self.browser:
await self.browser.close()
self.browser = None
if self.playwright:
await self.playwright.stop()
self.playwright = None

487
docs/examples/crypto_analysis_example.py
View File

@@ -18,11 +18,20 @@ Key Features:
import asyncio
import pandas as pd
import numpy as np
import re
import plotly.express as px
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode, LXMLWebScrapingStrategy
from crawl4ai import (
AsyncWebCrawler,
BrowserConfig,
CrawlerRunConfig,
CacheMode,
LXMLWebScrapingStrategy,
)
from crawl4ai import CrawlResult
from typing import List
from IPython.display import HTML
__current_dir__ = __file__.rsplit("/", 1)[0]
class CryptoAlphaGenerator:
"""
@@ -31,134 +40,319 @@ class CryptoAlphaGenerator:
- Liquidity scores
- Momentum-risk ratios
- Machine learning-inspired trading signals
Methods:
analyze_tables(): Process raw tables into trading insights
create_visuals(): Generate institutional-grade visualizations
generate_insights(): Create plain English trading recommendations
"""
def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Convert crypto market data to machine-readable format
Handles currency symbols, units (B=Billions), and percentage values
Convert crypto market data to machine-readable format.
Handles currency symbols, units (B=Billions), and percentage values.
"""
# Clean numeric columns
df['Price'] = df['Price'].str.replace('[^\d.]', '', regex=True).astype(float)
df['Market Cap'] = df['Market Cap'].str.extract(r'\$([\d.]+)B')[0].astype(float) * 1e9
df['Volume(24h)'] = df['Volume(24h)'].str.extract(r'\$([\d.]+)B')[0].astype(float) * 1e9
# Make a copy to avoid SettingWithCopyWarning
df = df.copy()
# Clean Price column (handle currency symbols)
df["Price"] = df["Price"].astype(str).str.replace("[^\d.]", "", regex=True).astype(float)
# Handle Market Cap and Volume, considering both Billions and Trillions
def convert_large_numbers(value):
if pd.isna(value):
return float('nan')
value = str(value)
multiplier = 1
if 'B' in value:
multiplier = 1e9
elif 'T' in value:
multiplier = 1e12
# Handle cases where the value might already be numeric
cleaned_value = re.sub(r"[^\d.]", "", value)
return float(cleaned_value) * multiplier if cleaned_value else float('nan')
df["Market Cap"] = df["Market Cap"].apply(convert_large_numbers)
df["Volume(24h)"] = df["Volume(24h)"].apply(convert_large_numbers)
# Convert percentages to decimal values
for col in ['1h %', '24h %', '7d %']:
df[col] = df[col].str.replace('%', '').astype(float) / 100
for col in ["1h %", "24h %", "7d %"]:
if col in df.columns:
# First ensure it's string, then clean
df[col] = (
df[col].astype(str)
.str.replace("%", "")
.str.replace(",", ".")
.replace("nan", np.nan)
)
df[col] = pd.to_numeric(df[col], errors='coerce') / 100
return df
def calculate_metrics(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Compute advanced trading metrics used by quantitative funds:
1. Volume/Market Cap Ratio - Measures liquidity efficiency
(High ratio = Underestimated attention)
2. Volatility Score - Risk-adjusted momentum potential
(High ratio = Underestimated attention, and small-cap = higher growth potential)
2. Volatility Score - Risk-adjusted momentum potential - Shows how stable is the trend
(STD of 1h/24h/7d returns)
3. Momentum Score - Weighted average of returns
3. Momentum Score - Weighted average of returns - Shows how strong is the trend
(1h:30% + 24h:50% + 7d:20%)
4. Volume Anomaly - 3σ deviation detection
(Flags potential insider activity)
(Flags potential insider activity) - Unusual trading activity – Flags coins with volume spikes (potential insider buying or news).
"""
# Liquidity Metrics
df['Volume/Market Cap Ratio'] = df['Volume(24h)'] / df['Market Cap']
df["Volume/Market Cap Ratio"] = df["Volume(24h)"] / df["Market Cap"]
# Risk Metrics
df['Volatility Score'] = df[['1h %','24h %','7d %']].std(axis=1)
df["Volatility Score"] = df[["1h %", "24h %", "7d %"]].std(axis=1)
# Momentum Metrics
df['Momentum Score'] = (df['1h %']*0.3 + df['24h %']*0.5 + df['7d %']*0.2)
df["Momentum Score"] = df["1h %"] * 0.3 + df["24h %"] * 0.5 + df["7d %"] * 0.2
# Anomaly Detection
median_vol = df['Volume(24h)'].median()
df['Volume Anomaly'] = df['Volume(24h)'] > 3 * median_vol
median_vol = df["Volume(24h)"].median()
df["Volume Anomaly"] = df["Volume(24h)"] > 3 * median_vol
# Value Flags
df['Undervalued Flag'] = (df['Market Cap'] < 1e9) & (df['Momentum Score'] > 0.05)
df['Liquid Giant'] = (df['Volume/Market Cap Ratio'] > 0.15) & (df['Market Cap'] > 1e9)
# Undervalued Flag - Low market cap and high momentum
# (High growth potential and low attention)
df["Undervalued Flag"] = (df["Market Cap"] < 1e9) & (
df["Momentum Score"] > 0.05
)
# Liquid Giant Flag - High volume/market cap ratio and large market cap
# (High liquidity and large market cap = institutional interest)
df["Liquid Giant"] = (df["Volume/Market Cap Ratio"] > 0.15) & (
df["Market Cap"] > 1e9
)
return df
def create_visuals(self, df: pd.DataFrame) -> dict:
def generate_insights_simple(self, df: pd.DataFrame) -> str:
"""
Generate three institutional-grade visualizations:
1. 3D Market Map - X:Size, Y:Liquidity, Z:Momentum
2. Liquidity Tree - Color:Volume Efficiency
3. Momentum Leaderboard - Top sustainable movers
Generates an ultra-actionable crypto trading report with:
- Risk-tiered opportunities (High/Medium/Low)
- Concrete examples for each trade type
- Entry/exit strategies spelled out
- Visual cues for quick scanning
"""
# 3D Market Overview
fig1 = px.scatter_3d(
df,
x='Market Cap',
y='Volume/Market Cap Ratio',
z='Momentum Score',
size='Volatility Score',
color='Volume Anomaly',
hover_name='Name',
title='Smart Money Market Map: Spot Overlooked Opportunities',
labels={'Market Cap': 'Size (Log $)', 'Volume/Market Cap Ratio': 'Liquidity Power'},
log_x=True,
template='plotly_dark'
)
# Liquidity Efficiency Tree
fig2 = px.treemap(
df,
path=['Name'],
values='Market Cap',
color='Volume/Market Cap Ratio',
hover_data=['Momentum Score'],
title='Liquidity Forest: Green = High Trading Efficiency',
color_continuous_scale='RdYlGn'
)
# Momentum Leaders
fig3 = px.bar(
df.sort_values('Momentum Score', ascending=False).head(10),
x='Name',
y='Momentum Score',
color='Volatility Score',
title='Sustainable Momentum Leaders (Low Volatility + High Growth)',
text='7d %',
template='plotly_dark'
)
return {'market_map': fig1, 'liquidity_tree': fig2, 'momentum_leaders': fig3}
report = [
"🚀 **CRYPTO TRADING CHEAT SHEET** 🚀",
"*Based on quantitative signals + hedge fund tactics*",
"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
]
# 1. HIGH-RISK: Undervalued Small-Caps (Momentum Plays)
high_risk = df[df["Undervalued Flag"]].sort_values("Momentum Score", ascending=False)
if not high_risk.empty:
example_coin = high_risk.iloc[0]
report.extend([
"\n🔥 **HIGH-RISK: Rocket Fuel Small-Caps**",
f"*Example Trade:* {example_coin['Name']} (Price: ${example_coin['Price']:.6f})",
"📊 *Why?* Tiny market cap (<$1B) but STRONG momentum (+{:.0f}% last week)".format(example_coin['7d %']*100),
"🎯 *Strategy:*",
"1. Wait for 5-10% dip from recent high (${:.6f} → Buy under ${:.6f})".format(
example_coin['Price'] / (1 - example_coin['24h %']), # Approx recent high
example_coin['Price'] * 0.95
),
"2. Set stop-loss at -10% (${:.6f})".format(example_coin['Price'] * 0.90),
"3. Take profit at +20% (${:.6f})".format(example_coin['Price'] * 1.20),
"⚠️ *Risk Warning:* These can drop 30% fast! Never bet more than 5% of your portfolio.",
"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
])
# 2. MEDIUM-RISK: Liquid Giants (Swing Trades)
medium_risk = df[df["Liquid Giant"]].sort_values("Volume/Market Cap Ratio", ascending=False)
if not medium_risk.empty:
example_coin = medium_risk.iloc[0]
report.extend([
"\n💎 **MEDIUM-RISK: Liquid Giants (Safe Swing Trades)**",
f"*Example Trade:* {example_coin['Name']} (Market Cap: ${example_coin['Market Cap']/1e9:.1f}B)",
"📊 *Why?* Huge volume (${:.1f}M/day) makes it easy to enter/exit".format(example_coin['Volume(24h)']/1e6),
"🎯 *Strategy:*",
"1. Buy when 24h volume > 15% of market cap (Current: {:.0f}%)".format(example_coin['Volume/Market Cap Ratio']*100),
"2. Hold 1-4 weeks (Big coins trend longer)",
"3. Exit when momentum drops below 5% (Current: {:.0f}%)".format(example_coin['Momentum Score']*100),
"📉 *Pro Tip:* Watch Bitcoin's trend - if BTC drops 5%, these usually follow.",
"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
])
# 3. LOW-RISK: Stable Momentum (DCA Targets)
low_risk = df[
(df["Momentum Score"] > 0.05) &
(df["Volatility Score"] < 0.03)
].sort_values("Market Cap", ascending=False)
if not low_risk.empty:
example_coin = low_risk.iloc[0]
report.extend([
"\n🛡️ **LOW-RISK: Steady Climbers (DCA & Forget)**",
f"*Example Trade:* {example_coin['Name']} (Volatility: {example_coin['Volatility Score']:.2f}/5)",
"📊 *Why?* Rises steadily (+{:.0f}%/week) with LOW drama".format(example_coin['7d %']*100),
"🎯 *Strategy:*",
"1. Buy small amounts every Tuesday/Friday (DCA)",
"2. Hold for 3+ months (Compound gains work best here)",
"3. Sell 10% at every +25% milestone",
"💰 *Best For:* Long-term investors who hate sleepless nights",
"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
])
# Volume Spike Alerts
anomalies = df[df["Volume Anomaly"]].sort_values("Volume(24h)", ascending=False)
if not anomalies.empty:
example_coin = anomalies.iloc[0]
report.extend([
"\n🚨 **Volume Spike Alert (Possible News/Whale Action)**",
f"*Coin:* {example_coin['Name']} (Volume: ${example_coin['Volume(24h)']/1e6:.1f}M, usual: ${example_coin['Volume(24h)']/3/1e6:.1f}M)",
"🔍 *Check:* Twitter/CoinGecko for news before trading",
"⚡ *If no news:* Could be insider buying - watch price action:",
"- Break above today's high → Buy with tight stop-loss",
"- Fade back down → Avoid (may be a fakeout)"
])
# Pro Tip Footer
report.append("\n✨ *Pro Tip:* Bookmark this report & check back in 24h to see if signals held up.")
return "\n".join(report)
def generate_insights(self, df: pd.DataFrame) -> str:
"""
Create plain English trading insights explaining:
- Volume spikes and their implications
- Risk-reward ratios of top movers
- Liquidity warnings for large positions
Generates a tactical trading report with:
- Top 3 trades per risk level (High/Medium/Low)
- Auto-calculated entry/exit prices
- BTC chart toggle tip
"""
top_coin = df.sort_values('Momentum Score', ascending=False).iloc[0]
anomaly_coins = df[df['Volume Anomaly']].sort_values('Volume(24h)', ascending=False)
# Filter top candidates for each risk level
high_risk = (
df[df["Undervalued Flag"]]
.sort_values("Momentum Score", ascending=False)
.head(3)
)
medium_risk = (
df[df["Liquid Giant"]]
.sort_values("Volume/Market Cap Ratio", ascending=False)
.head(3)
)
low_risk = (
df[(df["Momentum Score"] > 0.05) & (df["Volatility Score"] < 0.03)]
.sort_values("Momentum Score", ascending=False)
.head(3)
)
report = ["# 🎯 Crypto Trading Tactical Report (Top 3 Per Risk Tier)"]
report = f"""
🚀 Top Alpha Opportunity: {top_coin['Name']}
- Momentum Score: {top_coin['Momentum Score']:.2%} (Top 1%)
- Risk-Reward Ratio: {top_coin['Momentum Score']/top_coin['Volatility Score']:.1f}
- Liquidity Warning: {'✅ Safe' if top_coin['Liquid Giant'] else '⚠️ Thin Markets'}
# 1. High-Risk Trades (Small-Cap Momentum)
if not high_risk.empty:
report.append("\n## 🔥 HIGH RISK: Small-Cap Rockets (5-50% Potential)")
for i, coin in high_risk.iterrows():
current_price = coin["Price"]
entry = current_price * 0.95 # -5% dip
stop_loss = current_price * 0.90 # -10%
take_profit = current_price * 1.20 # +20%
report.append(
f"\n### {coin['Name']} (Momentum: {coin['Momentum Score']:.1%})"
f"\n- **Current Price:** ${current_price:.4f}"
f"\n- **Entry:** < ${entry:.4f} (Wait for pullback)"
f"\n- **Stop-Loss:** ${stop_loss:.4f} (-10%)"
f"\n- **Target:** ${take_profit:.4f} (+20%)"
f"\n- **Risk/Reward:** 1:2"
f"\n- **Watch:** Volume spikes above {coin['Volume(24h)']/1e6:.1f}M"
)
# 2. Medium-Risk Trades (Liquid Giants)
if not medium_risk.empty:
report.append("\n## 💎 MEDIUM RISK: Liquid Swing Trades (10-30% Potential)")
for i, coin in medium_risk.iterrows():
current_price = coin["Price"]
entry = current_price * 0.98 # -2% dip
stop_loss = current_price * 0.94 # -6%
take_profit = current_price * 1.15 # +15%
report.append(
f"\n### {coin['Name']} (Liquidity Score: {coin['Volume/Market Cap Ratio']:.1%})"
f"\n- **Current Price:** ${current_price:.2f}"
f"\n- **Entry:** < ${entry:.2f} (Buy slight dips)"
f"\n- **Stop-Loss:** ${stop_loss:.2f} (-6%)"
f"\n- **Target:** ${take_profit:.2f} (+15%)"
f"\n- **Hold Time:** 1-3 weeks"
f"\n- **Key Metric:** Volume/Cap > 15%"
)
# 3. Low-Risk Trades (Stable Momentum)
if not low_risk.empty:
report.append("\n## 🛡️ LOW RISK: Steady Gainers (5-15% Potential)")
for i, coin in low_risk.iterrows():
current_price = coin["Price"]
entry = current_price * 0.99 # -1% dip
stop_loss = current_price * 0.97 # -3%
take_profit = current_price * 1.10 # +10%
report.append(
f"\n### {coin['Name']} (Stability Score: {1/coin['Volatility Score']:.1f}x)"
f"\n- **Current Price:** ${current_price:.2f}"
f"\n- **Entry:** < ${entry:.2f} (Safe zone)"
f"\n- **Stop-Loss:** ${stop_loss:.2f} (-3%)"
f"\n- **Target:** ${take_profit:.2f} (+10%)"
f"\n- **DCA Suggestion:** 3 buys over 72 hours"
)
# Volume Anomaly Alert
anomalies = df[df["Volume Anomaly"]].sort_values("Volume(24h)", ascending=False).head(2)
if not anomalies.empty:
report.append("\n⚠️ **Volume Spike Alerts**")
for i, coin in anomalies.iterrows():
report.append(
f"- {coin['Name']}: Volume {coin['Volume(24h)']/1e6:.1f}M "
f"(3x normal) | Price moved: {coin['24h %']:.1%}"
)
# Pro Tip
report.append(
"\n📊 **Chart Hack:** Hide BTC in visuals:\n"
"```python\n"
"# For 3D Map:\n"
"fig.update_traces(visible=False, selector={'name':'Bitcoin'})\n"
"# For Treemap:\n"
"df = df[df['Name'] != 'Bitcoin']\n"
"```"
)
return "\n".join(report)
def create_visuals(self, df: pd.DataFrame) -> dict:
"""Enhanced visuals with BTC toggle support"""
# 3D Market Map (with BTC toggle hint)
fig1 = px.scatter_3d(
df,
x="Market Cap",
y="Volume/Market Cap Ratio",
z="Momentum Score",
color="Name", # Color by name to allow toggling
hover_name="Name",
title="Market Map (Toggle BTC in legend to focus on alts)",
log_x=True
)
fig1.update_traces(
marker=dict(size=df["Volatility Score"]*100 + 5) # Dynamic sizing
)
🔥 Volume Spikes Detected ({len(anomaly_coins)} coins):
{anomaly_coins[['Name', 'Volume(24h)']].head(3).to_markdown(index=False)}
# Liquidity Tree (exclude BTC if too dominant)
if df[df["Name"] == "BitcoinBTC"]["Market Cap"].values[0] > df["Market Cap"].median() * 10:
df = df[df["Name"] != "BitcoinBTC"]
💡 Smart Money Tip: Coins with Volume/Cap > 15% and Momentum > 5%
historically outperform by 22% weekly returns.
"""
return report
fig2 = px.treemap(
df,
path=["Name"],
values="Market Cap",
color="Volume/Market Cap Ratio",
title="Liquidity Tree (BTC auto-removed if dominant)"
)
return {"market_map": fig1, "liquidity_tree": fig2}
async def main():
"""
@@ -171,60 +365,79 @@ async def main():
"""
# Configure browser with anti-detection features
browser_config = BrowserConfig(
headless=True,
stealth=True,
block_resources=["image", "media"]
headless=False,
)
# Initialize crawler with smart table detection
crawler = AsyncWebCrawler(config=browser_config)
await crawler.start()
try:
# Set up scraping parameters
crawl_config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS,
scraping_strategy=LXMLWebScrapingStrategy(
table_score_threshold=8, # Strict table detection
keep_data_attributes=True
)
table_score_threshold=8, # Strict table detection
keep_data_attributes=True,
scraping_strategy=LXMLWebScrapingStrategy(),
scan_full_page=True,
scroll_delay=0.2,
)
# Execute market data extraction
results: List[CrawlResult] = await crawler.arun(
url='https://coinmarketcap.com/?page=1',
config=crawl_config
)
# Process results
for result in results:
if result.success and result.media['tables']:
# Extract primary market table
raw_df = pd.DataFrame(
result.media['tables'][0]['rows'],
columns=result.media['tables'][0]['headers']
)
# Initialize analysis engine
analyzer = CryptoAlphaGenerator()
clean_df = analyzer.clean_data(raw_df)
analyzed_df = analyzer.calculate_metrics(clean_df)
# Generate outputs
visuals = analyzer.create_visuals(analyzed_df)
insights = analyzer.generate_insights(analyzed_df)
# Save visualizations
visuals['market_map'].write_html("market_map.html")
visuals['liquidity_tree'].write_html("liquidity_tree.html")
# Display results
print("🔑 Key Trading Insights:")
print(insights)
print("\n📊 Open 'market_map.html' for interactive analysis")
# # Execute market data extraction
# results: List[CrawlResult] = await crawler.arun(
# url="https://coinmarketcap.com/?page=1", config=crawl_config
# )
# # Process results
# raw_df = pd.DataFrame()
# for result in results:
# if result.success and result.media["tables"]:
# # Extract primary market table
# # DataFrame
# raw_df = pd.DataFrame(
# result.media["tables"][0]["rows"],
# columns=result.media["tables"][0]["headers"],
# )
# break
# This is for debugging only
# ////// Remove this in production from here..
# Save raw data for debugging
# raw_df.to_csv(f"{__current_dir__}/tmp/raw_crypto_data.csv", index=False)
# print("🔍 Raw data saved to 'raw_crypto_data.csv'")
# Read from file for debugging
raw_df = pd.read_csv(f"{__current_dir__}/tmp/raw_crypto_data.csv")
# ////// ..to here
# Select top 20
raw_df = raw_df.head(50)
# Remove "Buy" from name
raw_df["Name"] = raw_df["Name"].str.replace("Buy", "")
# Initialize analysis engine
analyzer = CryptoAlphaGenerator()
clean_df = analyzer.clean_data(raw_df)
analyzed_df = analyzer.calculate_metrics(clean_df)
# Generate outputs
visuals = analyzer.create_visuals(analyzed_df)
insights = analyzer.generate_insights(analyzed_df)
# Save visualizations
visuals["market_map"].write_html(f"{__current_dir__}/tmp/market_map.html")
visuals["liquidity_tree"].write_html(f"{__current_dir__}/tmp/liquidity_tree.html")
# Display results
print("🔑 Key Trading Insights:")
print(insights)
print("\n📊 Open 'market_map.html' for interactive analysis")
print("\n📊 Open 'liquidity_tree.html' for interactive analysis")
finally:
await crawler.close()
if __name__ == "__main__":
asyncio.run(main())
asyncio.run(main())

20
tests/browser/docker/test_docker_browser.py
View File

@@ -17,9 +17,9 @@ if __name__ == "__main__":
from crawl4ai.browser import BrowserManager
from crawl4ai.async_configs import BrowserConfig, CrawlerRunConfig
from crawl4ai.async_logger import AsyncLogger
from crawl4ai.browser.docker_config import DockerConfig
from crawl4ai.browser.docker_registry import DockerRegistry
from crawl4ai.browser.docker_utils import DockerUtils
from crawl4ai.browser import DockerConfig
from crawl4ai.browser import DockerRegistry
from crawl4ai.browser import DockerUtils
# Create a logger for clear terminal output
logger = AsyncLogger(verbose=True, log_file=None)
@@ -136,7 +136,7 @@ async def test_docker_components():
# Verify Chrome is installed in the container
returncode, stdout, stderr = await docker_utils.exec_in_container(
container_id, ["which", "google-chrome"]
container_id, ["which", "chromium"]
)
if returncode != 0:
@@ -149,7 +149,7 @@ async def test_docker_components():
# Test Chrome version
returncode, stdout, stderr = await docker_utils.exec_in_container(
container_id, ["google-chrome", "--version"]
container_id, ["chromium", "--version"]
)
if returncode != 0:
@@ -608,13 +608,13 @@ async def run_tests():
return
# First test Docker components
setup_result = await test_docker_components()
if not setup_result:
logger.error("Docker component tests failed - skipping browser tests", tag="TEST")
return
# setup_result = await test_docker_components()
# if not setup_result:
# logger.error("Docker component tests failed - skipping browser tests", tag="TEST")
# return
# Run browser tests
results.append(await test_docker_connect_mode())
# results.append(await test_docker_connect_mode())
results.append(await test_docker_launch_mode())
results.append(await test_docker_persistent_storage())
results.append(await test_docker_parallel_pages())