arun_many with proxy gives weird cleaned_html and markdown values (example a1b76c2fce11d755)

[ThePianistop66]

I'm trying to scrape a website. Yesterday it was working fine without a proxy. When I added the proxy, now it returns weird strings like this instead of the actual data: a1b76c2fce11d755

Python 3.11.11 with Crawl4AI 0.4.248
Windows 11

If someone could point me in the right direction, I would really appreciate it

Here is the code I have written:

import aiofiles
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CrawlerMonitor, DisplayMode, MemoryAdaptiveDispatcher, RateLimiter, CrawlResult
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator
import asyncio, os, re, logging, requests
from bs4 import BeautifulSoup
from typing import Union, List, Dict

ENV = 'TEST'

class CustomFormatter(logging.Formatter):
    def format(self, record):
        if record.levelno == logging.INFO:
            return f"\n{'=' * 40} {record.getMessage()} {'=' * 40}\n"
        return super().format(record)
    
# Set up custom logging format
handler = logging.StreamHandler()
handler.setFormatter(CustomFormatter())
logging.root.addHandler(handler)
logging.root.setLevel(logging.INFO)

# Set up logging format
logging.basicConfig(
    format='%(asctime)s %(levelname)s %(message)s',
    level=logging.INFO
)
    
md_generator = DefaultMarkdownGenerator(
        options={
            "ignore_links": True,
            "escape_html": False,
            "body_width": 80
        }
    )

crawler_config = CrawlerRunConfig(
    stream=False,
    markdown_generator= md_generator
)

# Proxy configuration
username = "USERNAME HERE"
password = "PASSWORD HERE"
proxy_host = "PROXY SCRAPE HERE"
proxy_port = "PORT HERE"

# Configure browser with proper Playwright proxy format
browser_config = BrowserConfig(
    proxy_config={
        "server": f"http://{proxy_host}:{proxy_port}",
        "username": username,
        "password": password
    }
)

monitor = CrawlerMonitor(
    max_visible_rows=15,
    display_mode = DisplayMode.AGGREGATED
)

rate_limiter = RateLimiter(
        base_delay = (1.0,3.0),
        max_delay = 30.0
    )

dispatcher = MemoryAdaptiveDispatcher(
    rate_limiter = rate_limiter,
    monitor = monitor
)

def get_ubik_urls(version: str) -> list[dict['title': str, 'href':str]]:
    """
    Retrieves a list of URLs and their titles from a Ubik support page for a specific version,
    excluding certain predefined pages.
    
    This function scrapes the Ubik support page to collect hyperlink URLs and their titles
    from the French QC content page, excluding pages with titles like 'Contenu', 'Index',
    and 'Recherche'. It constructs the base URL using the provided version number and returns
    a list of dictionaries containing the title and href for each valid link.
    
    Args:
        version (str): The Ubik version number to fetch URLs for
                      (e.g., "2.0", "3.1", etc.)
    
    Returns:
        list[dict]: A list of dictionaries, each containing:
                   - 'title': The link text (str)
                   - 'href': The URL of the link (str)
    
    Raises:
        requests.RequestException: If there's an error fetching the webpage
        AttributeError: If the page structure doesn't match expected HTML format
    """
    
    # Construct the base URL and content list URL using the version
    BASE_URL = f'https://support.assystrx.com/ubik/{version}/FR/QC/'
    LIST_OF_URLS = f'{BASE_URL}masterf1_ubik_fr_content.htm'
    
    # Pages to skip
    TO_SKIP = ['Contenu', 'Index', 'Recherche']
    # Initialize empty list to store found URLs
    urls = []
    
    # Log the start of URL fetching process
    logging.info(f"Started fetching URLs for Ubik version {version}")
    
    # Fetch the webpage content
    response = requests.get(LIST_OF_URLS)
    
    # Parse the HTML content using BeautifulSoup
    soup = BeautifulSoup(response.content, 'html.parser')
    
    # Find all anchor tags (links) in the HTML
    links = soup.find_all('a')
    # Extract the href attribute from each link and add to urls list
    
    for link in links:
        title = link.text
        href = link.get('href')
        if title not in TO_SKIP:
            urls.append({
                'title': title,
                'href': f'{BASE_URL}{href}'
                })
    
    # Log the completion with number of URLs found
    logging.info(f'Completed: Found {len(urls)} URLs')
    if ENV == 'TEST':
        return urls[10:15]
    else:
        return urls

async def crawl_ubik(urls: List[Dict[str, str]]) -> List[Dict[str, Union[str, object]]]:
    """
    Asynchronously crawl multiple URLs and collect results.

    Args:
        urls (List[Dict[str, str]]): List of dictionaries containing 'title' and 'href' keys.

    Returns:
        List[Dict[str, Union[str, object]]]: List of dictionaries with 'title' and 'result' keys.
    """
    # Log the start of crawling with number of URLs
    logging.info(f"Starting Ubik crawl for {len(urls)} URLs")

    async with AsyncWebCrawler(config = browser_config) as crawler:
        # Extract URLs from input and run async crawling
        logging.info("Extracting URLs and running async crawling")
        results: list[CrawlResult] = await crawler.arun_many(
            urls = [url['href'] for url in urls],
            config = crawler_config,
            dispatcher = dispatcher,
        )
        # Log successful crawl completion
        logging.info(f"Successfully crawled {len(results)} URLs")
        
        # Combine original titles with crawling results
        processed_results = [{'title': url['title'], 'result': result} for url, result in zip(urls, results)]
        
        return processed_results

def sanitize_filename(filename: str) -> str:
   """
   Sanitize filename by removing or replacing incompatible characters.

   Args:
       filename (str): Original filename.

   Returns:
       str: Sanitized filename safe for Windows file systems.
   """
   # Remove or replace characters not allowed in Windows filenames
   sanitized = re.sub(r'[<>:"/\\|?*]', '_', filename)
   
   # Trim leading/trailing spaces and periods
   sanitized = sanitized.strip(' .')
   
   # Limit filename length (Windows max path is 255 characters)
   max_length = 255
   sanitized = sanitized[:max_length]
   
   return sanitized

async def write_ubik_md_files_async(results: List[Dict[str, Union[str, object]]]) -> None:
    """
    Asynchronously write markdown files for Ubik results using aiofiles.
    
    Args:
        results (List[Dict[str, Union[str, object]]]): A list of dictionaries containing 
            result information with 'title' and 'result' keys.
    """
    folder_path = 'ubik'
    logging.info(f"Starting to write {len(results)} Markdown files asynchronously")
    
    async def write_single_file(result: Dict[str, Union[str, object]]) -> None:
        file_name = sanitize_filename(result['title']) + '.md'
        full_path = os.path.join('pharmacie', folder_path, file_name)
        
        try:
            async with aiofiles.open(full_path, 'w', encoding='UTF-8') as f:
                await f.write(result['result'].markdown)
            logging.info(f"Successfully wrote {file_name}")
        except IOError as e:
            logging.error(f"Error writing file {file_name}: {e}")
    
    # Create tasks for all files
    tasks = [write_single_file(result) for result in results]
    
    # Wait for all tasks to complete
    await asyncio.gather(*tasks)
    
    logging.info(f"Finished writing {len(results)} Markdown files")

async def main():
    urls = get_ubik_urls('6.0')
    results = await crawl_ubik(urls)
    await write_ubik_md_files_async(results)

if __name__ == '__main__':
    asyncio.run(main())