Update web_crawler.py

Improve code efficiency, readability, and maintainability in web_crawler.py

README.md

@@ -190,33 +190,6 @@ result = crawler.run(
 print(result.extracted_content)
 ```
 
-### Extract Structured Data from Web Pages With Proxy and BaseUrl
-
-```python
-from crawl4ai import WebCrawler
-from crawl4ai.extraction_strategy import LLMExtractionStrategy
-
-def create_crawler():
-    crawler = WebCrawler(verbose=True, proxy="http://127.0.0.1:7890")
-    crawler.warmup()
-    return crawler
-
-crawler = create_crawler()
-
-crawler.warmup()
-
-result = crawler.run(
-    url="https://www.nbcnews.com/business",
-    extraction_strategy=LLMExtractionStrategy(
-        provider="openai/gpt-4o",
-        api_token="sk-",
-        base_url="https://api.openai.com/v1"
-    )
-)
-
-print(result.markdown)
-```
-
 ## Documentation 📚
 
 For detailed documentation, including installation instructions, advanced features, and API reference, visit our [Documentation Website](https://crawl4ai.com/mkdocs/).

crawl4ai/crawler_strategy.py

@@ -82,8 +82,6 @@ class LocalSeleniumCrawlerStrategy(CrawlerStrategy):
         print("[LOG] 🚀 Initializing LocalSeleniumCrawlerStrategy")
         self.options = Options()
         self.options.headless = True
-        if kwargs.get("proxy"):
-            self.options.add_argument("--proxy-server={}".format(kwargs.get("proxy")))
         if kwargs.get("user_agent"):
             self.options.add_argument("--user-agent=" + kwargs.get("user_agent"))
         else:
@@ -244,7 +242,6 @@ class LocalSeleniumCrawlerStrategy(CrawlerStrategy):
                 driver.quit()
             
             # Execute JS code if provided
-            self.js_code = kwargs.get("js_code", self.js_code)
             if self.js_code and type(self.js_code) == str:
                 self.driver.execute_script(self.js_code)
                 # Optionally, wait for some condition after executing the JS code

crawl4ai/extraction_strategy.py

@@ -79,7 +79,6 @@ class LLMExtractionStrategy(ExtractionStrategy):
         self.overlap_rate = kwargs.get("overlap_rate", OVERLAP_RATE)
         self.word_token_rate = kwargs.get("word_token_rate", WORD_TOKEN_RATE)
         self.apply_chunking = kwargs.get("apply_chunking", True)
-        self.base_url = kwargs.get("base_url", None)
         if not self.apply_chunking:
             self.chunk_token_threshold = 1e9
         
@@ -111,7 +110,7 @@ class LLMExtractionStrategy(ExtractionStrategy):
                 "{" + variable + "}", variable_values[variable]
             )
         
-        response = perform_completion_with_backoff(self.provider, prompt_with_variables, self.api_token, base_url=self.base_url) # , json_response=self.extract_type == "schema")
+        response = perform_completion_with_backoff(self.provider, prompt_with_variables, self.api_token) # , json_response=self.extract_type == "schema")
         try:
             blocks = extract_xml_data(["blocks"], response.choices[0].message.content)['blocks']
             blocks = json.loads(blocks)

crawl4ai/prompts.py

@@ -29,7 +29,7 @@ To generate the JSON objects:
 
 5. Make sure the generated JSON is complete and parsable, with no errors or omissions.
 
-6. Make sure to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
+6. Make sur to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
 
 Please provide your output within <blocks> tags, like this:
 
@@ -87,7 +87,7 @@ To generate the JSON objects:
 
 5. Make sure the generated JSON is complete and parsable, with no errors or omissions.
 
-6. Make sure to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
+6. Make sur to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
 
 7. Never alter the extracted content, just copy and paste it as it is.
 
@@ -142,7 +142,7 @@ To generate the JSON objects:
 
 5. Make sure the generated JSON is complete and parsable, with no errors or omissions.
 
-6. Make sure to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
+6. Make sur to escape any special characters in the HTML content, and also single or double quote to avoid JSON parsing issues.
 
 7. Never alter the extracted content, just copy and paste it as it is.
 
@@ -201,4 +201,4 @@ Avoid Common Mistakes:
 - Do not generate the Python coee show me how to do the task, this is your task to extract the information and return it in JSON format.
 
 Result
-Output the final list of JSON objects, wrapped in <blocks>...</blocks> XML tags. Make sure to close the tag properly."""
+Output the final list of JSON objects, wrapped in <blocks>...</blocks> XML tags. Make sure to close the tag properly."""

crawl4ai/utils.py

@@ -634,12 +634,7 @@ def get_content_of_website_optimized(url: str, html: str, word_count_threshold:
         return node
 
     body = flatten_nested_elements(body)
-    base64_pattern = re.compile(r'data:image/[^;]+;base64,([^"]+)')
-    for img in imgs:
-        src = img.get('src', '')
-        if base64_pattern.match(src):
-            # Replace base64 data with empty string
-            img['src'] = base64_pattern.sub('', src)
+
     cleaned_html = str(body).replace('\n\n', '\n').replace('  ', ' ')
     cleaned_html = sanitize_html(cleaned_html)
 
@@ -721,7 +716,7 @@ def extract_xml_data(tags, string):
     return data
     
 # Function to perform the completion with exponential backoff
-def perform_completion_with_backoff(provider, prompt_with_variables, api_token, json_response = False, base_url=None):
+def perform_completion_with_backoff(provider, prompt_with_variables, api_token, json_response = False):
     from litellm import completion 
     from litellm.exceptions import RateLimitError
     max_attempts = 3
@@ -740,7 +735,6 @@ def perform_completion_with_backoff(provider, prompt_with_variables, api_token,
                 ],
                 temperature=0.01,
                 api_key=api_token,
-                base_url=base_url,
                 **extra_args
             )
             return response  # Return the successful response
@@ -761,7 +755,7 @@ def perform_completion_with_backoff(provider, prompt_with_variables, api_token,
                     "content": ["Rate limit error. Please try again later."]
                 }]
     
-def extract_blocks(url, html, provider = DEFAULT_PROVIDER, api_token = None, base_url = None):
+def extract_blocks(url, html, provider = DEFAULT_PROVIDER, api_token = None):
     # 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
     
@@ -776,7 +770,7 @@ def extract_blocks(url, html, provider = DEFAULT_PROVIDER, api_token = None, bas
             "{" + variable + "}", variable_values[variable]
         )
         
-    response = perform_completion_with_backoff(provider, prompt_with_variables, api_token, base_url=base_url)
+    response = perform_completion_with_backoff(provider, prompt_with_variables, api_token)
         
     try:
         blocks = extract_xml_data(["blocks"], response.choices[0].message.content)['blocks']
@@ -870,17 +864,17 @@ def merge_chunks_based_on_token_threshold(chunks, token_threshold):
 
     return merged_sections
 
-def process_sections(url: str, sections: list, provider: str, api_token: str, base_url=None) -> list:
+def process_sections(url: str, sections: list, provider: str, api_token: str) -> list:
     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))
+            extracted_content.extend(extract_blocks(url, section, provider, api_token))
             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]
+            futures = [executor.submit(extract_blocks, url, section, provider, api_token) for section in sections]
             for future in as_completed(futures):
                 extracted_content.extend(future.result())
     

crawl4ai/web_crawler.py

@@ -16,41 +16,23 @@ warnings.filterwarnings("ignore", message='Field "model_name" has conflict with
 
 
 class WebCrawler:
-    def __init__(
-        self,
-        # db_path: str = None,
-        crawler_strategy: CrawlerStrategy = None,
-        always_by_pass_cache: bool = False,
-        verbose: bool = False,
-        proxy: str = None,
-    ):
-        # self.db_path = db_path
-        self.crawler_strategy = crawler_strategy or LocalSeleniumCrawlerStrategy(verbose=verbose, proxy=proxy)
+    def __init__(self, crawler_strategy: CrawlerStrategy = None, always_by_pass_cache: bool = False, verbose: bool = False):
+        self.crawler_strategy = crawler_strategy or LocalSeleniumCrawlerStrategy(verbose=verbose)
         self.always_by_pass_cache = always_by_pass_cache
-
-        # Create the .crawl4ai folder in the user's home directory if it doesn't exist
         self.crawl4ai_folder = os.path.join(Path.home(), ".crawl4ai")
         os.makedirs(self.crawl4ai_folder, exist_ok=True)
         os.makedirs(f"{self.crawl4ai_folder}/cache", exist_ok=True)
-
-        # If db_path is not provided, use the default path
-        # if not db_path:
-            # self.db_path = f"{self.crawl4ai_folder}/crawl4ai.db"
-        
-        # flush_db()
         init_db()
-        
         self.ready = False
         
     def warmup(self):
         print("[LOG] 🌤️  Warming up the WebCrawler")
-        result = self.run(
+        self.run(
             url='https://google.com/',
             word_count_threshold=5,
-            extraction_strategy= NoExtractionStrategy(),
+            extraction_strategy=NoExtractionStrategy(),
             bypass_cache=False,
-            verbose = False,
-            # warmup=True
+            verbose=False
         )
         self.ready = True
         print("[LOG] 🌞 WebCrawler is ready to crawl")
@@ -140,12 +122,8 @@ class WebCrawler:
                 if not isinstance(chunking_strategy, ChunkingStrategy):
                     raise ValueError("Unsupported chunking strategy")
                 
-                # if word_count_threshold < MIN_WORD_THRESHOLD:
-                #     word_count_threshold = MIN_WORD_THRESHOLD
-                    
                 word_count_threshold = max(word_count_threshold, 0)
 
-                # Check cache first
                 cached = None
                 screenshot_data = None
                 extracted_content = None
@@ -170,7 +148,7 @@ class WebCrawler:
                     html = sanitize_input_encode(self.crawler_strategy.crawl(url, **kwargs))
                     t2 = time.time()
                     if verbose:
-                        print(f"[LOG] 🚀 Crawling done for {url}, success: {bool(html)}, time taken: {t2 - t1} seconds")
+                        print(f"[LOG] 🚀 Crawling done for {url}, success: {bool(html)}, time taken: {t2 - t1:.2f} seconds")
                     if screenshot:
                         screenshot_data = self.crawler_strategy.take_screenshot()
 
@@ -201,13 +179,10 @@ class WebCrawler:
             t = time.time()
             # Extract content from HTML
             try:
-                # t1 = time.time()
-                # result = get_content_of_website(url, html, word_count_threshold, css_selector=css_selector, only_text=kwargs.get("only_text", False))
-                # print(f"[LOG] 🚀 Crawling done for {url}, success: True, time taken: {time.time() - t1} seconds")
                 t1 = time.time()
                 result = get_content_of_website_optimized(url, html, word_count_threshold, css_selector=css_selector, only_text=kwargs.get("only_text", False))
                 if verbose:
-                    print(f"[LOG] 🚀 Content extracted for {url}, success: True, time taken: {time.time() - t1} seconds")
+                    print(f"[LOG] 🚀 Content extracted for {url}, success: True, time taken: {time.time() - t1:.2f} seconds")
                 
                 if result is None:
                     raise ValueError(f"Failed to extract content from the website: {url}")
@@ -229,7 +204,7 @@ class WebCrawler:
                 extracted_content = json.dumps(extracted_content, indent=4, default=str)
 
                 if verbose:
-                    print(f"[LOG] 🚀 Extraction done for {url}, time taken: {time.time() - t} seconds.")
+                    print(f"[LOG] 🚀 Extraction done for {url}, time taken: {time.time() - t:.2f} seconds.")
                 
             screenshot = None if not screenshot else screenshot