import time from urllib.parse import urlparse from concurrent.futures import ThreadPoolExecutor, as_completed from bs4 import BeautifulSoup, Comment, element, Tag, NavigableString import json import html import re import os import platform from .prompts import PROMPT_EXTRACT_BLOCKS from .config import * from pathlib import Path from typing import Dict, Any from urllib.parse import urljoin import requests from requests.exceptions import InvalidSchema from typing import Optional, Tuple, Dict, Any import xxhash from colorama import Fore, Style, init import textwrap import cProfile import pstats from functools import wraps import asyncio class InvalidCSSSelectorError(Exception): pass def create_box_message(message: str, type: str = "info", width: int = 120, add_newlines: bool = True, double_line: bool = False) -> str: """ Create a styled message box with colored borders and formatted text. How it works: 1. Determines box style and colors based on the message type (e.g., info, warning). 2. Wraps text to fit within the specified width. 3. Constructs a box using characters (single or double lines) with appropriate formatting. 4. Adds optional newlines before and after the box. Args: message (str): The message to display inside the box. type (str): Type of the message (e.g., "info", "warning", "error", "success"). Defaults to "info". width (int): Width of the box. Defaults to 120. add_newlines (bool): Whether to add newlines before and after the box. Defaults to True. double_line (bool): Whether to use double lines for the box border. Defaults to False. Returns: str: A formatted string containing the styled message box. """ init() # Define border and text colors for different types styles = { "warning": (Fore.YELLOW, Fore.LIGHTYELLOW_EX, "⚠"), "info": (Fore.BLUE, Fore.LIGHTBLUE_EX, "ℹ"), "success": (Fore.GREEN, Fore.LIGHTGREEN_EX, "✓"), "error": (Fore.RED, Fore.LIGHTRED_EX, "×"), } border_color, text_color, prefix = styles.get(type.lower(), styles["info"]) # Define box characters based on line style box_chars = { "single": ("─", "│", "┌", "┐", "└", "┘"), "double": ("═", "║", "╔", "╗", "╚", "╝") } line_style = "double" if double_line else "single" h_line, v_line, tl, tr, bl, br = box_chars[line_style] # Process lines with lighter text color formatted_lines = [] raw_lines = message.split('\n') if raw_lines: first_line = f"{prefix} {raw_lines[0].strip()}" wrapped_first = textwrap.fill(first_line, width=width-4) formatted_lines.extend(wrapped_first.split('\n')) for line in raw_lines[1:]: if line.strip(): wrapped = textwrap.fill(f" {line.strip()}", width=width-4) formatted_lines.extend(wrapped.split('\n')) else: formatted_lines.append("") # Create the box with colored borders and lighter text horizontal_line = h_line * (width - 1) box = [ f"{border_color}{tl}{horizontal_line}{tr}", *[f"{border_color}{v_line}{text_color} {line:<{width-2}}{border_color}{v_line}" for line in formatted_lines], f"{border_color}{bl}{horizontal_line}{br}{Style.RESET_ALL}" ] result = "\n".join(box) if add_newlines: result = f"\n{result}\n" return result def calculate_semaphore_count(): """ Calculate the optimal semaphore count based on system resources. How it works: 1. Determines the number of CPU cores and total system memory. 2. Sets a base count as half of the available CPU cores. 3. Limits the count based on memory, assuming 2GB per semaphore instance. 4. Returns the minimum value between CPU and memory-based limits. Returns: int: The calculated semaphore count. """ cpu_count = os.cpu_count() memory_gb = get_system_memory() / (1024 ** 3) # Convert to GB base_count = max(1, cpu_count // 2) memory_based_cap = int(memory_gb / 2) # Assume 2GB per instance return min(base_count, memory_based_cap) def get_system_memory(): """ Get the total system memory in bytes. How it works: 1. Detects the operating system. 2. Reads memory information from system-specific commands or files. 3. Converts the memory to bytes for uniformity. Returns: int: The total system memory in bytes. Raises: OSError: If the operating system is unsupported. """ system = platform.system() if system == "Linux": with open('/proc/meminfo', 'r') as mem: for line in mem: if line.startswith('MemTotal:'): return int(line.split()[1]) * 1024 # Convert KB to bytes elif system == "Darwin": # macOS import subprocess output = subprocess.check_output(['sysctl', '-n', 'hw.memsize']).decode('utf-8') return int(output.strip()) elif system == "Windows": import ctypes kernel32 = ctypes.windll.kernel32 c_ulonglong = ctypes.c_ulonglong class MEMORYSTATUSEX(ctypes.Structure): _fields_ = [ ('dwLength', ctypes.c_ulong), ('dwMemoryLoad', ctypes.c_ulong), ('ullTotalPhys', c_ulonglong), ('ullAvailPhys', c_ulonglong), ('ullTotalPageFile', c_ulonglong), ('ullAvailPageFile', c_ulonglong), ('ullTotalVirtual', c_ulonglong), ('ullAvailVirtual', c_ulonglong), ('ullAvailExtendedVirtual', c_ulonglong), ] memoryStatus = MEMORYSTATUSEX() memoryStatus.dwLength = ctypes.sizeof(MEMORYSTATUSEX) kernel32.GlobalMemoryStatusEx(ctypes.byref(memoryStatus)) return memoryStatus.ullTotalPhys else: raise OSError("Unsupported operating system") def get_home_folder(): """ Get or create the home folder for Crawl4AI configuration and cache. How it works: 1. Uses environment variables or defaults to the user's home directory. 2. Creates `.crawl4ai` and its subdirectories (`cache`, `models`) if they don't exist. 3. Returns the path to the home folder. Returns: str: The path to the Crawl4AI home folder. """ home_folder = os.path.join(os.getenv("CRAWL4_AI_BASE_DIRECTORY", os.getenv("CRAWL4_AI_BASE_DIRECTORY", Path.home())), ".crawl4ai") os.makedirs(home_folder, exist_ok=True) os.makedirs(f"{home_folder}/cache", exist_ok=True) os.makedirs(f"{home_folder}/models", exist_ok=True) return home_folder def beautify_html(escaped_html): """ Beautifies an escaped HTML string. Parameters: escaped_html (str): A string containing escaped HTML. Returns: str: A beautifully formatted HTML string. """ # Unescape the HTML string unescaped_html = html.unescape(escaped_html) # Use BeautifulSoup to parse and prettify the HTML soup = BeautifulSoup(unescaped_html, 'html.parser') pretty_html = soup.prettify() return pretty_html def split_and_parse_json_objects(json_string): """ Splits a JSON string which is a list of objects and tries to parse each object. Parameters: json_string (str): A string representation of a list of JSON objects, e.g., '[{...}, {...}, ...]'. Returns: tuple: A tuple containing two lists: - First list contains all successfully parsed JSON objects. - Second list contains the string representations of all segments that couldn't be parsed. """ # Trim the leading '[' and trailing ']' if json_string.startswith('[') and json_string.endswith(']'): json_string = json_string[1:-1].strip() # Split the string into segments that look like individual JSON objects segments = [] depth = 0 start_index = 0 for i, char in enumerate(json_string): if char == '{': if depth == 0: start_index = i depth += 1 elif char == '}': depth -= 1 if depth == 0: segments.append(json_string[start_index:i+1]) # Try parsing each segment parsed_objects = [] unparsed_segments = [] for segment in segments: try: obj = json.loads(segment) parsed_objects.append(obj) except json.JSONDecodeError: unparsed_segments.append(segment) return parsed_objects, unparsed_segments def sanitize_html(html): """ Sanitize an HTML string by escaping quotes. How it works: 1. Replaces all unwanted and special characters with an empty string. 2. Escapes double and single quotes for safe usage. Args: html (str): The HTML string to sanitize. Returns: str: The sanitized HTML string. """ # Replace all unwanted and special characters with an empty string sanitized_html = html # sanitized_html = re.sub(r'[^\w\s.,;:!?=\[\]{}()<>\/\\\-"]', '', html) # Escape all double and single quotes sanitized_html = sanitized_html.replace('"', '\\"').replace("'", "\\'") return sanitized_html def sanitize_input_encode(text: str) -> str: """Sanitize input to handle potential encoding issues.""" try: try: if not text: return '' # Attempt to encode and decode as UTF-8 to handle potential encoding issues return text.encode('utf-8', errors='ignore').decode('utf-8') except UnicodeEncodeError as e: print(f"Warning: Encoding issue detected. Some characters may be lost. Error: {e}") # Fall back to ASCII if UTF-8 fails return text.encode('ascii', errors='ignore').decode('ascii') except Exception as e: raise ValueError(f"Error sanitizing input: {str(e)}") from e def escape_json_string(s): """ Escapes characters in a string to be JSON safe. Parameters: s (str): The input string to be escaped. Returns: str: The escaped string, safe for JSON encoding. """ # Replace problematic backslash first s = s.replace('\\', '\\\\') # Replace the double quote s = s.replace('"', '\\"') # Escape control characters s = s.replace('\b', '\\b') s = s.replace('\f', '\\f') s = s.replace('\n', '\\n') s = s.replace('\r', '\\r') s = s.replace('\t', '\\t') # Additional problematic characters # Unicode control characters s = re.sub(r'[\x00-\x1f\x7f-\x9f]', lambda x: '\\u{:04x}'.format(ord(x.group())), s) return s def replace_inline_tags(soup, tags, only_text=False): """ Replace inline HTML tags with Markdown-style equivalents. How it works: 1. Maps specific tags (e.g., , ) to Markdown syntax. 2. Finds and replaces all occurrences of these tags in the provided BeautifulSoup object. 3. Optionally replaces tags with their text content only. Args: soup (BeautifulSoup): Parsed HTML content. tags (List[str]): List of tags to replace. only_text (bool): Whether to replace tags with plain text. Defaults to False. Returns: BeautifulSoup: Updated BeautifulSoup object with replaced tags. """ tag_replacements = { 'b': lambda tag: f"**{tag.text}**", 'i': lambda tag: f"*{tag.text}*", 'u': lambda tag: f"__{tag.text}__", 'span': lambda tag: f"{tag.text}", 'del': lambda tag: f"~~{tag.text}~~", 'ins': lambda tag: f"++{tag.text}++", 'sub': lambda tag: f"~{tag.text}~", 'sup': lambda tag: f"^^{tag.text}^^", 'strong': lambda tag: f"**{tag.text}**", 'em': lambda tag: f"*{tag.text}*", 'code': lambda tag: f"`{tag.text}`", 'kbd': lambda tag: f"`{tag.text}`", 'var': lambda tag: f"_{tag.text}_", 's': lambda tag: f"~~{tag.text}~~", 'q': lambda tag: f'"{tag.text}"', 'abbr': lambda tag: f"{tag.text} ({tag.get('title', '')})", 'cite': lambda tag: f"_{tag.text}_", 'dfn': lambda tag: f"_{tag.text}_", 'time': lambda tag: f"{tag.text}", 'small': lambda tag: f"{tag.text}", 'mark': lambda tag: f"=={tag.text}==" } replacement_data = [(tag, tag_replacements.get(tag, lambda t: t.text)) for tag in tags] for tag_name, replacement_func in replacement_data: for tag in soup.find_all(tag_name): replacement_text = tag.text if only_text else replacement_func(tag) tag.replace_with(replacement_text) return soup # for tag_name in tags: # for tag in soup.find_all(tag_name): # if not only_text: # replacement_text = tag_replacements.get(tag_name, lambda t: t.text)(tag) # tag.replace_with(replacement_text) # else: # tag.replace_with(tag.text) # return soup def get_content_of_website(url, html, word_count_threshold = MIN_WORD_THRESHOLD, css_selector = None, **kwargs): """ Extract structured content, media, and links from website HTML. How it works: 1. Parses the HTML content using BeautifulSoup. 2. Extracts internal/external links and media (images, videos, audios). 3. Cleans the content by removing unwanted tags and attributes. 4. Converts cleaned HTML to Markdown. 5. Collects metadata and returns the extracted information. Args: url (str): The website URL. html (str): The HTML content of the website. word_count_threshold (int): Minimum word count for content inclusion. Defaults to MIN_WORD_THRESHOLD. css_selector (Optional[str]): CSS selector to extract specific content. Defaults to None. Returns: Dict[str, Any]: Extracted content including Markdown, cleaned HTML, media, links, and metadata. """ try: if not html: return None # Parse HTML content with BeautifulSoup soup = BeautifulSoup(html, 'html.parser') # Get the content within the tag body = soup.body # If css_selector is provided, extract content based on the selector if css_selector: selected_elements = body.select(css_selector) if not selected_elements: raise InvalidCSSSelectorError(f"Invalid CSS selector , No elements found for CSS selector: {css_selector}") div_tag = soup.new_tag('div') for el in selected_elements: div_tag.append(el) body = div_tag links = { 'internal': [], 'external': [] } # Extract all internal and external links for a in body.find_all('a', href=True): href = a['href'] url_base = url.split('/')[2] if href.startswith('http') and url_base not in href: links['external'].append({ 'href': href, 'text': a.get_text() }) else: links['internal'].append( { 'href': href, 'text': a.get_text() } ) # Remove script, style, and other tags that don't carry useful content from body for tag in body.find_all(['script', 'style', 'link', 'meta', 'noscript']): tag.decompose() # Remove all attributes from remaining tags in body, except for img tags for tag in body.find_all(): if tag.name != 'img': tag.attrs = {} # Extract all img tgas int0 [{src: '', alt: ''}] media = { 'images': [], 'videos': [], 'audios': [] } for img in body.find_all('img'): media['images'].append({ 'src': img.get('src'), 'alt': img.get('alt'), "type": "image" }) # Extract all video tags into [{src: '', alt: ''}] for video in body.find_all('video'): media['videos'].append({ 'src': video.get('src'), 'alt': video.get('alt'), "type": "video" }) # Extract all audio tags into [{src: '', alt: ''}] for audio in body.find_all('audio'): media['audios'].append({ 'src': audio.get('src'), 'alt': audio.get('alt'), "type": "audio" }) # Replace images with their alt text or remove them if no alt text is available for img in body.find_all('img'): alt_text = img.get('alt') if alt_text: img.replace_with(soup.new_string(alt_text)) else: img.decompose() # Create a function that replace content of all"pre" tag with its inner text def replace_pre_tags_with_text(node): for child in node.find_all('pre'): # set child inner html to its text child.string = child.get_text() return node # Replace all "pre" tags with their inner text body = replace_pre_tags_with_text(body) # Replace inline tags with their text content body = replace_inline_tags( body, ['b', 'i', 'u', 'span', 'del', 'ins', 'sub', 'sup', 'strong', 'em', 'code', 'kbd', 'var', 's', 'q', 'abbr', 'cite', 'dfn', 'time', 'small', 'mark'], only_text=kwargs.get('only_text', False) ) # Recursively remove empty elements, their parent elements, and elements with word count below threshold def remove_empty_and_low_word_count_elements(node, word_count_threshold): for child in node.contents: if isinstance(child, element.Tag): remove_empty_and_low_word_count_elements(child, word_count_threshold) word_count = len(child.get_text(strip=True).split()) if (len(child.contents) == 0 and not child.get_text(strip=True)) or word_count < word_count_threshold: child.decompose() return node body = remove_empty_and_low_word_count_elements(body, word_count_threshold) def remove_small_text_tags(body: Tag, word_count_threshold: int = MIN_WORD_THRESHOLD): # We'll use a list to collect all tags that don't meet the word count requirement tags_to_remove = [] # Traverse all tags in the body for tag in body.find_all(True): # True here means all tags # Check if the tag contains text and if it's not just whitespace if tag.string and tag.string.strip(): # Split the text by spaces and count the words word_count = len(tag.string.strip().split()) # If the word count is less than the threshold, mark the tag for removal if word_count < word_count_threshold: tags_to_remove.append(tag) # Remove all marked tags from the tree for tag in tags_to_remove: tag.decompose() # or tag.extract() to remove and get the element return body # Remove small text tags body = remove_small_text_tags(body, word_count_threshold) def is_empty_or_whitespace(tag: Tag): if isinstance(tag, NavigableString): return not tag.strip() # Check if the tag itself is empty or all its children are empty/whitespace if not tag.contents: return True return all(is_empty_or_whitespace(child) for child in tag.contents) def remove_empty_tags(body: Tag): # Continue processing until no more changes are made changes = True while changes: changes = False # Collect all tags that are empty or contain only whitespace empty_tags = [tag for tag in body.find_all(True) if is_empty_or_whitespace(tag)] for tag in empty_tags: # If a tag is empty, decompose it tag.decompose() changes = True # Mark that a change was made return body # Remove empty tags body = remove_empty_tags(body) # Flatten nested elements with only one child of the same type def flatten_nested_elements(node): for child in node.contents: if isinstance(child, element.Tag): flatten_nested_elements(child) if len(child.contents) == 1 and child.contents[0].name == child.name: # print('Flattening:', child.name) child_content = child.contents[0] child.replace_with(child_content) return node body = flatten_nested_elements(body) # Remove comments for comment in soup.find_all(string=lambda text: isinstance(text, Comment)): comment.extract() # Remove consecutive empty newlines and replace multiple spaces with a single space cleaned_html = str(body).replace('\n\n', '\n').replace(' ', ' ') # Sanitize the cleaned HTML content cleaned_html = sanitize_html(cleaned_html) # sanitized_html = escape_json_string(cleaned_html) # Convert cleaned HTML to Markdown h = html2text.HTML2Text() h = CustomHTML2Text() h.ignore_links = True markdown = h.handle(cleaned_html) markdown = markdown.replace(' ```', '```') try: meta = extract_metadata(html, soup) except Exception as e: print('Error extracting metadata:', str(e)) meta = {} # Return the Markdown content return{ 'markdown': markdown, 'cleaned_html': cleaned_html, 'success': True, 'media': media, 'links': links, 'metadata': meta } except Exception as e: print('Error processing HTML content:', str(e)) raise InvalidCSSSelectorError(f"Invalid CSS selector: {css_selector}") from e def get_content_of_website_optimized(url: str, html: str, word_count_threshold: int = MIN_WORD_THRESHOLD, css_selector: str = None, **kwargs) -> Dict[str, Any]: if not html: return None soup = BeautifulSoup(html, 'html.parser') body = soup.body image_description_min_word_threshold = kwargs.get('image_description_min_word_threshold', IMAGE_DESCRIPTION_MIN_WORD_THRESHOLD) for tag in kwargs.get('excluded_tags', []) or []: for el in body.select(tag): el.decompose() if css_selector: selected_elements = body.select(css_selector) if not selected_elements: raise InvalidCSSSelectorError(f"Invalid CSS selector, No elements found for CSS selector: {css_selector}") body = soup.new_tag('div') for el in selected_elements: body.append(el) links = {'internal': [], 'external': []} media = {'images': [], 'videos': [], 'audios': []} # Extract meaningful text for media files from closest parent def find_closest_parent_with_useful_text(tag): current_tag = tag while current_tag: current_tag = current_tag.parent # Get the text content from the parent tag if current_tag: text_content = current_tag.get_text(separator=' ',strip=True) # Check if the text content has at least word_count_threshold if len(text_content.split()) >= image_description_min_word_threshold: return text_content return None def process_image(img, url, index, total_images): #Check if an image has valid display and inside undesired html elements def is_valid_image(img, parent, parent_classes): style = img.get('style', '') src = img.get('src', '') classes_to_check = ['button', 'icon', 'logo'] tags_to_check = ['button', 'input'] return all([ 'display:none' not in style, src, not any(s in var for var in [src, img.get('alt', ''), *parent_classes] for s in classes_to_check), parent.name not in tags_to_check ]) #Score an image for it's usefulness def score_image_for_usefulness(img, base_url, index, images_count): # Function to parse image height/width value and units def parse_dimension(dimension): if dimension: match = re.match(r"(\d+)(\D*)", dimension) if match: number = int(match.group(1)) unit = match.group(2) or 'px' # Default unit is 'px' if not specified return number, unit return None, None # Fetch image file metadata to extract size and extension def fetch_image_file_size(img, base_url): #If src is relative path construct full URL, if not it may be CDN URL img_url = urljoin(base_url,img.get('src')) try: response = requests.head(img_url) if response.status_code == 200: return response.headers.get('Content-Length',None) else: print(f"Failed to retrieve file size for {img_url}") return None except InvalidSchema as e: return None finally: return image_height = img.get('height') height_value, height_unit = parse_dimension(image_height) image_width = img.get('width') width_value, width_unit = parse_dimension(image_width) image_size = 0 #int(fetch_image_file_size(img,base_url) or 0) image_format = os.path.splitext(img.get('src',''))[1].lower() # Remove . from format image_format = image_format.strip('.') score = 0 if height_value: if height_unit == 'px' and height_value > 150: score += 1 if height_unit in ['%','vh','vmin','vmax'] and height_value >30: score += 1 if width_value: if width_unit == 'px' and width_value > 150: score += 1 if width_unit in ['%','vh','vmin','vmax'] and width_value >30: score += 1 if image_size > 10000: score += 1 if img.get('alt') != '': score+=1 if any(image_format==format for format in ['jpg','png','webp']): score+=1 if index/images_count<0.5: score+=1 return score if not is_valid_image(img, img.parent, img.parent.get('class', [])): return None score = score_image_for_usefulness(img, url, index, total_images) if score <= IMAGE_SCORE_THRESHOLD: return None return { 'src': img.get('src', '').replace('\\"', '"').strip(), 'alt': img.get('alt', ''), 'desc': find_closest_parent_with_useful_text(img), 'score': score, 'type': 'image' } def process_element(element: element.PageElement) -> bool: try: if isinstance(element, NavigableString): if isinstance(element, Comment): element.extract() return False if element.name in ['script', 'style', 'link', 'meta', 'noscript']: element.decompose() return False keep_element = False if element.name == 'a' and element.get('href'): href = element['href'] url_base = url.split('/')[2] link_data = {'href': href, 'text': element.get_text()} if href.startswith('http') and url_base not in href: links['external'].append(link_data) else: links['internal'].append(link_data) keep_element = True elif element.name == 'img': return True # Always keep image elements elif element.name in ['video', 'audio']: media[f"{element.name}s"].append({ 'src': element.get('src'), 'alt': element.get('alt'), 'type': element.name, 'description': find_closest_parent_with_useful_text(element) }) source_tags = element.find_all('source') for source_tag in source_tags: media[f"{element.name}s"].append({ 'src': source_tag.get('src'), 'alt': element.get('alt'), 'type': element.name, 'description': find_closest_parent_with_useful_text(element) }) return True # Always keep video and audio elements if element.name != 'pre': if element.name in ['b', 'i', 'u', 'span', 'del', 'ins', 'sub', 'sup', 'strong', 'em', 'code', 'kbd', 'var', 's', 'q', 'abbr', 'cite', 'dfn', 'time', 'small', 'mark']: if kwargs.get('only_text', False): element.replace_with(element.get_text()) else: element.unwrap() elif element.name != 'img': element.attrs = {} # Process children for child in list(element.children): if isinstance(child, NavigableString) and not isinstance(child, Comment): if len(child.strip()) > 0: keep_element = True else: if process_element(child): keep_element = True # Check word count if not keep_element: word_count = len(element.get_text(strip=True).split()) keep_element = word_count >= word_count_threshold if not keep_element: element.decompose() return keep_element except Exception as e: print('Error processing element:', str(e)) return False #process images by filtering and extracting contextual text from the page imgs = body.find_all('img') media['images'] = [ result for result in (process_image(img, url, i, len(imgs)) for i, img in enumerate(imgs)) if result is not None ] process_element(body) def flatten_nested_elements(node): if isinstance(node, NavigableString): return node if len(node.contents) == 1 and isinstance(node.contents[0], element.Tag) and node.contents[0].name == node.name: return flatten_nested_elements(node.contents[0]) node.contents = [flatten_nested_elements(child) for child in node.contents] return node body = flatten_nested_elements(body) base64_pattern = re.compile(r'data:image/[^;]+;base64,([^"]+)') for img in imgs: try: src = img.get('src', '') if base64_pattern.match(src): img['src'] = base64_pattern.sub('', src) except: pass cleaned_html = str(body).replace('\n\n', '\n').replace(' ', ' ') cleaned_html = sanitize_html(cleaned_html) h = CustomHTML2Text() h.ignore_links = True markdown = h.handle(cleaned_html) markdown = markdown.replace(' ```', '```') try: meta = extract_metadata(html, soup) except Exception as e: print('Error extracting metadata:', str(e)) meta = {} return { 'markdown': markdown, 'cleaned_html': cleaned_html, 'success': True, 'media': media, 'links': links, 'metadata': meta } def extract_metadata(html, soup=None): """ Extract optimized content, media, and links from website HTML. How it works: 1. Similar to `get_content_of_website`, but optimized for performance. 2. Filters and scores images for usefulness. 3. Extracts contextual descriptions for media files. 4. Handles excluded tags and CSS selectors. 5. Cleans HTML and converts it to Markdown. Args: url (str): The website URL. html (str): The HTML content of the website. word_count_threshold (int): Minimum word count for content inclusion. Defaults to MIN_WORD_THRESHOLD. css_selector (Optional[str]): CSS selector to extract specific content. Defaults to None. **kwargs: Additional options for customization. Returns: Dict[str, Any]: Extracted content including Markdown, cleaned HTML, media, links, and metadata. """ metadata = {} if not html and not soup: return {} if not soup: soup = BeautifulSoup(html, 'lxml') head = soup.head if not head: return metadata # Title title_tag = head.find('title') metadata['title'] = title_tag.string.strip() if title_tag and title_tag.string else None # Meta description description_tag = head.find('meta', attrs={'name': 'description'}) metadata['description'] = description_tag.get('content', '').strip() if description_tag else None # Meta keywords keywords_tag = head.find('meta', attrs={'name': 'keywords'}) metadata['keywords'] = keywords_tag.get('content', '').strip() if keywords_tag else None # Meta author author_tag = head.find('meta', attrs={'name': 'author'}) metadata['author'] = author_tag.get('content', '').strip() if author_tag else None # Open Graph metadata og_tags = head.find_all('meta', attrs={'property': re.compile(r'^og:')}) for tag in og_tags: property_name = tag.get('property', '').strip() content = tag.get('content', '').strip() if property_name and content: metadata[property_name] = content # Twitter Card metadata twitter_tags = head.find_all('meta', attrs={'name': re.compile(r'^twitter:')}) for tag in twitter_tags: property_name = tag.get('name', '').strip() content = tag.get('content', '').strip() if property_name and content: metadata[property_name] = content return metadata def extract_xml_tags(string): """ Extracts XML tags from a string. Args: string (str): The input string containing XML tags. Returns: List[str]: A list of XML tags extracted from the input string. """ tags = re.findall(r'<(\w+)>', string) return list(set(tags)) def extract_xml_data(tags, string): """ Extract data for specified XML tags from a string. How it works: 1. Searches the string for each tag using regex. 2. Extracts the content within the tags. 3. Returns a dictionary of tag-content pairs. Args: tags (List[str]): The list of XML tags to extract. string (str): The input string containing XML data. Returns: Dict[str, str]: A dictionary with tag names as keys and extracted content as values. """ data = {} for tag in tags: pattern = f"<{tag}>(.*?)" match = re.search(pattern, string, re.DOTALL) if match: data[tag] = match.group(1).strip() else: data[tag] = "" return data def perform_completion_with_backoff( provider, prompt_with_variables, api_token, json_response = False, base_url=None, **kwargs ): """ Perform an API completion request with exponential backoff. How it works: 1. Sends a completion request to the API. 2. Retries on rate-limit errors with exponential delays. 3. Returns the API response or an error after all retries. Args: provider (str): The name of the API provider. prompt_with_variables (str): The input prompt for the completion request. api_token (str): The API token for authentication. json_response (bool): Whether to request a JSON response. Defaults to False. base_url (Optional[str]): The base URL for the API. Defaults to None. **kwargs: Additional arguments for the API request. Returns: dict: The API response or an error message after all retries. """ from litellm import completion from litellm.exceptions import RateLimitError max_attempts = 3 base_delay = 2 # Base delay in seconds, you can adjust this based on your needs extra_args = { "temperature": 0.01, 'api_key': api_token, 'base_url': base_url } if json_response: extra_args["response_format"] = { "type": "json_object" } if kwargs.get("extra_args"): extra_args.update(kwargs["extra_args"]) for attempt in range(max_attempts): try: response =completion( model=provider, messages=[ {"role": "user", "content": prompt_with_variables} ], **extra_args ) return response # Return the successful response except RateLimitError as e: print("Rate limit error:", str(e)) # Check if we have exhausted our max attempts if attempt < max_attempts - 1: # Calculate the delay and wait delay = base_delay * (2 ** attempt) # Exponential backoff formula print(f"Waiting for {delay} seconds before retrying...") time.sleep(delay) else: # Return an error response after exhausting all retries return [{ "index": 0, "tags": ["error"], "content": ["Rate limit error. Please try again later."] }] def extract_blocks(url, html, provider = DEFAULT_PROVIDER, api_token = None, base_url = None): """ Extract content blocks from website HTML using an AI provider. How it works: 1. Prepares a prompt by sanitizing and escaping HTML. 2. Sends the prompt to an AI provider with optional retries. 3. Parses the response to extract structured blocks or errors. Args: url (str): The website URL. html (str): The HTML content of the website. provider (str): The AI provider for content extraction. Defaults to DEFAULT_PROVIDER. api_token (Optional[str]): The API token for authentication. Defaults to None. base_url (Optional[str]): The base URL for the API. Defaults to None. Returns: List[dict]: A list of extracted content blocks. """ # api_token = os.getenv('GROQ_API_KEY', None) if not api_token else api_token api_token = PROVIDER_MODELS.get(provider, None) if not api_token else api_token variable_values = { "URL": url, "HTML": escape_json_string(sanitize_html(html)), } prompt_with_variables = PROMPT_EXTRACT_BLOCKS for variable in variable_values: prompt_with_variables = prompt_with_variables.replace( "{" + variable + "}", variable_values[variable] ) response = perform_completion_with_backoff(provider, prompt_with_variables, api_token, base_url=base_url) try: blocks = extract_xml_data(["blocks"], response.choices[0].message.content)['blocks'] blocks = json.loads(blocks) ## Add error: False to the blocks for block in blocks: block['error'] = False except Exception as e: parsed, unparsed = split_and_parse_json_objects(response.choices[0].message.content) blocks = parsed # Append all unparsed segments as onr error block and content is list of unparsed segments if unparsed: blocks.append({ "index": 0, "error": True, "tags": ["error"], "content": unparsed }) return blocks def extract_blocks_batch(batch_data, provider = "groq/llama3-70b-8192", api_token = None): """ Extract content blocks from a batch of website HTMLs. How it works: 1. Prepares prompts for each URL and HTML pair. 2. Sends the prompts to the AI provider in a batch request. 3. Parses the responses to extract structured blocks or errors. Args: batch_data (List[Tuple[str, str]]): A list of (URL, HTML) pairs. provider (str): The AI provider for content extraction. Defaults to "groq/llama3-70b-8192". api_token (Optional[str]): The API token for authentication. Defaults to None. Returns: List[dict]: A list of extracted content blocks from all batch items. """ api_token = os.getenv('GROQ_API_KEY', None) if not api_token else api_token from litellm import batch_completion messages = [] for url, html in batch_data: variable_values = { "URL": url, "HTML": html, } prompt_with_variables = PROMPT_EXTRACT_BLOCKS for variable in variable_values: prompt_with_variables = prompt_with_variables.replace( "{" + variable + "}", variable_values[variable] ) messages.append([{"role": "user", "content": prompt_with_variables}]) responses = batch_completion( model = provider, messages = messages, temperature = 0.01 ) all_blocks = [] for response in responses: try: blocks = extract_xml_data(["blocks"], response.choices[0].message.content)['blocks'] blocks = json.loads(blocks) except Exception as e: blocks = [{ "index": 0, "tags": ["error"], "content": ["Error extracting blocks from the HTML content. Choose another provider/model or try again."], "questions": ["What went wrong during the block extraction process?"] }] all_blocks.append(blocks) return sum(all_blocks, []) def merge_chunks_based_on_token_threshold(chunks, token_threshold): """ Merges small chunks into larger ones based on the total token threshold. :param chunks: List of text chunks to be merged based on token count. :param token_threshold: Max number of tokens for each merged chunk. :return: List of merged text chunks. """ merged_sections = [] current_chunk = [] total_token_so_far = 0 for chunk in chunks: chunk_token_count = len(chunk.split()) * 1.3 # Estimate token count with a factor if total_token_so_far + chunk_token_count < token_threshold: current_chunk.append(chunk) total_token_so_far += chunk_token_count else: if current_chunk: merged_sections.append('\n\n'.join(current_chunk)) current_chunk = [chunk] total_token_so_far = chunk_token_count # Add the last chunk if it exists if current_chunk: merged_sections.append('\n\n'.join(current_chunk)) return merged_sections def process_sections(url: str, sections: list, provider: str, api_token: str, base_url=None) -> list: """ Process sections of HTML content sequentially or in parallel. How it works: 1. Sequentially processes sections with delays for "groq/" providers. 2. Uses ThreadPoolExecutor for parallel processing with other providers. 3. Extracts content blocks for each section. Args: url (str): The website URL. sections (List[str]): The list of HTML sections to process. provider (str): The AI provider for content extraction. api_token (str): The API token for authentication. base_url (Optional[str]): The base URL for the API. Defaults to None. Returns: List[dict]: The list of extracted content blocks from all sections. """ extracted_content = [] if provider.startswith("groq/"): # Sequential processing with a delay for section in sections: extracted_content.extend(extract_blocks(url, section, provider, api_token, base_url=base_url)) time.sleep(0.5) # 500 ms delay between each processing else: # Parallel processing using ThreadPoolExecutor with ThreadPoolExecutor() as executor: futures = [executor.submit(extract_blocks, url, section, provider, api_token, base_url=base_url) for section in sections] for future in as_completed(futures): extracted_content.extend(future.result()) return extracted_content def wrap_text(draw, text, font, max_width): """ Wrap text to fit within a specified width for rendering. How it works: 1. Splits the text into words. 2. Constructs lines that fit within the maximum width using the provided font. 3. Returns the wrapped text as a single string. Args: draw (ImageDraw.Draw): The drawing context for measuring text size. text (str): The text to wrap. font (ImageFont.FreeTypeFont): The font to use for measuring text size. max_width (int): The maximum width for each line. Returns: str: The wrapped text. """ # Wrap the text to fit within the specified width lines = [] words = text.split() while words: line = '' while words and draw.textbbox((0, 0), line + words[0], font=font)[2] <= max_width: line += (words.pop(0) + ' ') lines.append(line) return '\n'.join(lines) def format_html(html_string): """ Prettify an HTML string using BeautifulSoup. How it works: 1. Parses the HTML string with BeautifulSoup. 2. Formats the HTML with proper indentation. 3. Returns the prettified HTML string. Args: html_string (str): The HTML string to format. Returns: str: The prettified HTML string. """ soup = BeautifulSoup(html_string, 'lxml.parser') return soup.prettify() def fast_format_html(html_string): """ A fast HTML formatter that uses string operations instead of parsing. Args: html_string (str): The HTML string to format Returns: str: The formatted HTML string """ # Initialize variables indent = 0 indent_str = " " # Two spaces for indentation formatted = [] in_content = False # Split by < and > to separate tags and content parts = html_string.replace('>', '>\n').replace('<', '\n<').split('\n') for part in parts: if not part.strip(): continue # Handle closing tags if part.startswith(''): formatted.append(indent_str * indent + part) # Handle opening tags elif part.startswith('<'): formatted.append(indent_str * indent + part) indent += 1 # Handle content between tags else: content = part.strip() if content: formatted.append(indent_str * indent + content) return '\n'.join(formatted) def normalize_url(href, base_url): """Normalize URLs to ensure consistent format""" from urllib.parse import urljoin, urlparse # Parse base URL to get components parsed_base = urlparse(base_url) if not parsed_base.scheme or not parsed_base.netloc: raise ValueError(f"Invalid base URL format: {base_url}") # Use urljoin to handle all cases normalized = urljoin(base_url, href.strip()) return normalized def normalize_url_tmp(href, base_url): """Normalize URLs to ensure consistent format""" # Extract protocol and domain from base URL try: base_parts = base_url.split('/') protocol = base_parts[0] domain = base_parts[2] except IndexError: raise ValueError(f"Invalid base URL format: {base_url}") # Handle special protocols special_protocols = {'mailto:', 'tel:', 'ftp:', 'file:', 'data:', 'javascript:'} if any(href.lower().startswith(proto) for proto in special_protocols): return href.strip() # Handle anchor links if href.startswith('#'): return f"{base_url}{href}" # Handle protocol-relative URLs if href.startswith('//'): return f"{protocol}{href}" # Handle root-relative URLs if href.startswith('/'): return f"{protocol}//{domain}{href}" # Handle relative URLs if not href.startswith(('http://', 'https://')): # Remove leading './' if present href = href.lstrip('./') return f"{protocol}//{domain}/{href}" return href.strip() def get_base_domain(url: str) -> str: """ Extract the base domain from a given URL, handling common edge cases. How it works: 1. Parses the URL to extract the domain. 2. Removes the port number and 'www' prefix. 3. Handles special domains (e.g., 'co.uk') to extract the correct base. Args: url (str): The URL to extract the base domain from. Returns: str: The extracted base domain or an empty string if parsing fails. """ try: # Get domain from URL domain = urlparse(url).netloc.lower() if not domain: return "" # Remove port if present domain = domain.split(':')[0] # Remove www domain = re.sub(r'^www\.', '', domain) # Extract last two parts of domain (handles co.uk etc) parts = domain.split('.') if len(parts) > 2 and parts[-2] in { 'co', 'com', 'org', 'gov', 'edu', 'net', 'mil', 'int', 'ac', 'ad', 'ae', 'af', 'ag' }: return '.'.join(parts[-3:]) return '.'.join(parts[-2:]) except Exception: return "" def is_external_url(url: str, base_domain: str) -> bool: """ Extract the base domain from a given URL, handling common edge cases. How it works: 1. Parses the URL to extract the domain. 2. Removes the port number and 'www' prefix. 3. Handles special domains (e.g., 'co.uk') to extract the correct base. Args: url (str): The URL to extract the base domain from. Returns: str: The extracted base domain or an empty string if parsing fails. """ special = {'mailto:', 'tel:', 'ftp:', 'file:', 'data:', 'javascript:'} if any(url.lower().startswith(p) for p in special): return True try: parsed = urlparse(url) if not parsed.netloc: # Relative URL return False # Strip 'www.' from both domains for comparison url_domain = parsed.netloc.lower().replace('www.', '') base = base_domain.lower().replace('www.', '') # Check if URL domain ends with base domain return not url_domain.endswith(base) except Exception: return False def clean_tokens(tokens: list[str]) -> list[str]: """ Clean a list of tokens by removing noise, stop words, and short tokens. How it works: 1. Defines a set of noise words and stop words. 2. Filters tokens based on length and exclusion criteria. 3. Excludes tokens starting with certain symbols (e.g., "↑", "▲"). Args: tokens (list[str]): The list of tokens to clean. Returns: list[str]: The cleaned list of tokens. """ # Set of tokens to remove noise = {'ccp', 'up', '↑', '▲', '⬆️', 'a', 'an', 'at', 'by', 'in', 'of', 'on', 'to', 'the'} STOP_WORDS = { 'a', 'an', 'and', 'are', 'as', 'at', 'be', 'by', 'for', 'from', 'has', 'he', 'in', 'is', 'it', 'its', 'of', 'on', 'that', 'the', 'to', 'was', 'were', 'will', 'with', # Pronouns 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'me', 'him', 'her', 'us', 'them', 'my', 'your', 'his', 'her', 'its', 'our', 'their', 'mine', 'yours', 'hers', 'ours', 'theirs', 'myself', 'yourself', 'himself', 'herself', 'itself', 'ourselves', 'themselves', # Common verbs 'am', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'having', 'do', 'does', 'did', 'doing', # Prepositions 'about', 'above', 'across', 'after', 'against', 'along', 'among', 'around', 'at', 'before', 'behind', 'below', 'beneath', 'beside', 'between', 'beyond', 'by', 'down', 'during', 'except', 'for', 'from', 'in', 'inside', 'into', 'near', 'of', 'off', 'on', 'out', 'outside', 'over', 'past', 'through', 'to', 'toward', 'under', 'underneath', 'until', 'up', 'upon', 'with', 'within', # Conjunctions 'and', 'but', 'or', 'nor', 'for', 'yet', 'so', 'although', 'because', 'since', 'unless', # Articles 'a', 'an', 'the', # Other common words 'this', 'that', 'these', 'those', 'what', 'which', 'who', 'whom', 'whose', 'when', 'where', 'why', 'how', 'all', 'any', 'both', 'each', 'few', 'more', 'most', 'other', 'some', 'such', 'can', 'cannot', "can't", 'could', "couldn't", 'may', 'might', 'must', "mustn't", 'shall', 'should', "shouldn't", 'will', "won't", 'would', "wouldn't", 'not', "n't", 'no', 'nor', 'none' } # Single comprehension, more efficient than multiple passes return [token for token in tokens if len(token) > 2 and token not in noise and token not in STOP_WORDS and not token.startswith('↑') and not token.startswith('▲') and not token.startswith('⬆')] def profile_and_time(func): """ Decorator to profile a function's execution time and performance. How it works: 1. Records the start time before executing the function. 2. Profiles the function's execution using `cProfile`. 3. Prints the elapsed time and profiling statistics. Args: func (Callable): The function to decorate. Returns: Callable: The decorated function with profiling and timing enabled. """ @wraps(func) def wrapper(self, *args, **kwargs): # Start timer start_time = time.perf_counter() # Setup profiler profiler = cProfile.Profile() profiler.enable() # Run function result = func(self, *args, **kwargs) # Stop profiler profiler.disable() # Calculate elapsed time elapsed_time = time.perf_counter() - start_time # Print timing print(f"[PROFILER] Scraping completed in {elapsed_time:.2f} seconds") # Print profiling stats stats = pstats.Stats(profiler) stats.sort_stats('cumulative') # Sort by cumulative time stats.print_stats(20) # Print top 20 time-consuming functions return result return wrapper def generate_content_hash(content: str) -> str: """Generate a unique hash for content""" return xxhash.xxh64(content.encode()).hexdigest() # return hashlib.sha256(content.encode()).hexdigest() def ensure_content_dirs(base_path: str) -> Dict[str, str]: """Create content directories if they don't exist""" dirs = { 'html': 'html_content', 'cleaned': 'cleaned_html', 'markdown': 'markdown_content', 'extracted': 'extracted_content', 'screenshots': 'screenshots', 'screenshot': 'screenshots' } content_paths = {} for key, dirname in dirs.items(): path = os.path.join(base_path, dirname) os.makedirs(path, exist_ok=True) content_paths[key] = path return content_paths def configure_windows_event_loop(): """ Configure the Windows event loop to use ProactorEventLoop. This resolves the NotImplementedError that occurs on Windows when using asyncio subprocesses. This function should only be called on Windows systems and before any async operations. On non-Windows systems, this function does nothing. Example: ```python from crawl4ai.async_configs import configure_windows_event_loop # Call this before any async operations if you're on Windows configure_windows_event_loop() ``` """ if platform.system() == 'Windows': asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) def get_error_context(exc_info, context_lines: int = 5): """ Extract error context with more reliable line number tracking. Args: exc_info: The exception info from sys.exc_info() context_lines: Number of lines to show before and after the error Returns: dict: Error context information """ import traceback import linecache import os # Get the full traceback tb = traceback.extract_tb(exc_info[2]) # Get the last frame (where the error occurred) last_frame = tb[-1] filename = last_frame.filename line_no = last_frame.lineno func_name = last_frame.name # Get the source code context using linecache # This is more reliable than inspect.getsourcelines context_start = max(1, line_no - context_lines) context_end = line_no + context_lines + 1 # Build the context lines with line numbers context_lines = [] for i in range(context_start, context_end): line = linecache.getline(filename, i) if line: # Remove any trailing whitespace/newlines and add the pointer for error line line = line.rstrip() pointer = '→' if i == line_no else ' ' context_lines.append(f"{i:4d} {pointer} {line}") # Join the lines with newlines code_context = '\n'.join(context_lines) # Get relative path for cleaner output try: rel_path = os.path.relpath(filename) except ValueError: # Fallback if relpath fails (can happen on Windows with different drives) rel_path = filename return { "filename": rel_path, "line_no": line_no, "function": func_name, "code_context": code_context }