A Coding Implementation of Crawl4AI for Web Crawling, Markdown Generation, JavaScript Execution, and LLM-Based Structured Extraction

import subprocess
import sys


print(" Installing system dependencies...")
subprocess.run(['apt-get', 'update', '-qq'], capture_output=True)
subprocess.run(['apt-get', 'install', '-y', '-qq',
               'libnss3', 'libnspr4', 'libatk1.0-0', 'libatk-bridge2.0-0',
               'libcups2', 'libdrm2', 'libxkbcommon0', 'libxcomposite1',
               'libxdamage1', 'libxfixes3', 'libxrandr2', 'libgbm1',
               'libasound2', 'libpango-1.0-0', 'libcairo2'], capture_output=True)
print(" System dependencies installed!")


print("n Installing Python packages...")
subprocess.run([sys.executable, '-m', 'pip', 'install', '-U', 'crawl4ai', 'nest_asyncio', 'pydantic', '-q'])
print(" Python packages installed!")


print("n Installing Playwright browsers (this may take a minute)...")
subprocess.run([sys.executable, '-m', 'playwright', 'install', 'chromium'], capture_output=True)
subprocess.run([sys.executable, '-m', 'playwright', 'install-deps', 'chromium'], capture_output=True)
print(" Playwright browsers installed!")


import nest_asyncio
nest_asyncio.apply()


import asyncio
import json
from typing import List, Optional
from pydantic import BaseModel, Field


print("n" + "="*60)
print(" INSTALLATION COMPLETE! Ready to crawl!")
print("="*60)


print("n" + "="*60)
print(" PART 2: BASIC CRAWLING")
print("="*60)


from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode


async def basic_crawl():
   """The simplest possible crawl - fetch a webpage and get markdown."""
   print("n Running basic crawl on example.com...")
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(url="https://example.com")
      
       print(f"n Crawl successful: {result.success}")
       print(f" Title: {result.metadata.get('title', 'N/A')}")
       print(f" Markdown length: {len(result.markdown.raw_markdown)} characters")
       print(f"n--- First 500 chars of markdown ---")
       print(result.markdown.raw_markdown[:500])
      
   return result


result = asyncio.run(basic_crawl())


print("n" + "="*60)
print(" PART 3: CONFIGURED CRAWLING")
print("="*60)


async def configured_crawl():
   """Crawling with custom browser and crawler configurations."""
   print("n Running configured crawl with custom settings...")
  
   browser_config = BrowserConfig(
       headless=True,
       verbose=True,
       viewport_width=1920,
       viewport_height=1080,
       user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
   )
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       word_count_threshold=10,
       page_timeout=30000,
       wait_until="networkidle",
       verbose=True
   )
  
   async with AsyncWebCrawler(config=browser_config) as crawler:
       result = await crawler.arun(
           url="https://httpbin.org/html",
           config=run_config
       )
      
       print(f"n Success: {result.success}")
       print(f" Status code: {result.status_code}")
       print(f"n--- Content Preview ---")
       print(result.markdown.raw_markdown[:400])
      
   return result


result = asyncio.run(configured_crawl())


print("n" + "="*60)
print(" PART 4: MARKDOWN GENERATION")
print("="*60)


from crawl4ai.content_filter_strategy import PruningContentFilter, BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator


async def markdown_generation_demo():
   """Demonstrates raw vs fit markdown with content filtering."""
   print("n Demonstrating markdown generation strategies...")
  
   browser_config = BrowserConfig(headless=True, verbose=False)
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       markdown_generator=DefaultMarkdownGenerator(
           content_filter=PruningContentFilter(
               threshold=0.4,
               threshold_type="fixed",
               min_word_threshold=20
           )
       )
   )
  
   async with AsyncWebCrawler(config=browser_config) as crawler:
       result = await crawler.arun(
           url="https://en.wikipedia.org/wiki/Web_scraping",
           config=run_config
       )
      
       raw_len = len(result.markdown.raw_markdown)
       fit_len = len(result.markdown.fit_markdown) if result.markdown.fit_markdown else 0
      
       print(f"n Markdown Comparison:")
       print(f"   Raw Markdown:  {raw_len:,} characters")
       print(f"   Fit Markdown:  {fit_len:,} characters")
       print(f"   Reduction:     {((raw_len - fit_len) / raw_len * 100):.1f}%")
      
       print(f"n--- Fit Markdown Preview (first 600 chars) ---")
       print(result.markdown.fit_markdown[:600] if result.markdown.fit_markdown else "N/A")
      
   return result


result = asyncio.run(markdown_generation_demo())

print("n" + "="*60)
print(" PART 5: BM25 QUERY-BASED FILTERING")
print("="*60)


async def bm25_filtering_demo():
   """Using BM25 algorithm to extract content relevant to a specific query."""
   print("n Extracting content relevant to a specific query...")
  
   query = "legal aspects privacy data protection"
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       markdown_generator=DefaultMarkdownGenerator(
           content_filter=BM25ContentFilter(
               user_query=query,
               bm25_threshold=1.2
           )
       )
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://en.wikipedia.org/wiki/Web_scraping",
           config=run_config
       )
      
       print(f"n Query: '{query}'")
       print(f" Fit markdown length: {len(result.markdown.fit_markdown or '')} chars")
       print(f"n--- Query-Relevant Content Preview ---")
       print(result.markdown.fit_markdown[:800] if result.markdown.fit_markdown else "No relevant content found")
      
   return result


result = asyncio.run(bm25_filtering_demo())


print("n" + "="*60)
print(" PART 6: CSS-BASED EXTRACTION (No LLM)")
print("="*60)


from crawl4ai import JsonCssExtractionStrategy


async def css_extraction_demo():
   """Extract structured data using CSS selectors - fast and reliable."""
   print("n Extracting data using CSS selectors...")
  
   schema = {
       "name": "Wikipedia Headings",
       "baseSelector": "div.mw-parser-output h2",
       "fields": [
           {
               "name": "heading_text",
               "selector": "span.mw-headline",
               "type": "text"
           },
           {
               "name": "heading_id",
               "selector": "span.mw-headline",
               "type": "attribute",
               "attribute": "id"
           }
       ]
   }
  
   extraction_strategy = JsonCssExtractionStrategy(schema, verbose=False)
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       extraction_strategy=extraction_strategy
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://en.wikipedia.org/wiki/Python_(programming_language)",
           config=run_config
       )
      
       if result.extracted_content:
           data = json.loads(result.extracted_content)
           print(f"n Extracted {len(data)} section headings")
           print(f"n--- Extracted Headings ---")
           for item in data[:10]:
               heading = item.get('heading_text', 'N/A')
               heading_id = item.get('heading_id', 'N/A')
               if heading:
                   print(f"  • {heading} (#{heading_id})")
       else:
           print(" No data extracted")
          
   return result


result = asyncio.run(css_extraction_demo())


print("n" + "="*60)
print(" PART 7: ADVANCED CSS EXTRACTION - Hacker News")
print("="*60)


async def advanced_css_extraction():
   """Extract stories from Hacker News with nested selectors."""
   print("n Extracting stories from Hacker News...")
  
   schema = {
       "name": "Hacker News Stories",
       "baseSelector": "tr.athing",
       "fields": [
           {
               "name": "rank",
               "selector": "span.rank",
               "type": "text"
           },
           {
               "name": "title",
               "selector": "span.titleline > a",
               "type": "text"
           },
           {
               "name": "url",
               "selector": "span.titleline > a",
               "type": "attribute",
               "attribute": "href"
           },
           {
               "name": "site",
               "selector": "span.sitestr",
               "type": "text"
           }
       ]
   }
  
   extraction_strategy = JsonCssExtractionStrategy(schema)
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       extraction_strategy=extraction_strategy
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://news.ycombinator.com",
           config=run_config
       )
      
       if result.extracted_content:
           stories = json.loads(result.extracted_content)
           print(f"n Extracted {len(stories)} stories from Hacker News")
           print(f"n--- Top 10 Stories ---")
           for story in stories[:10]:
               rank = story.get('rank', '?').strip('.') if story.get('rank') else '?'
               title = story.get('title', 'N/A')[:55]
               site = story.get('site', 'N/A')
               print(f"  #{rank:<3} {title:<55} ({site})")
              
   return result


result = asyncio.run(advanced_css_extraction())

print("n" + "="*60)
print(" PART 8: JAVASCRIPT EXECUTION")
print("="*60)


async def javascript_execution_demo():
   """Execute JavaScript on pages before extraction."""
   print("n Executing JavaScript before crawling...")
  
   js_code = """
   // Scroll down to trigger lazy loading
   window.scrollTo(0, document.body.scrollHeight);
  
   // Wait for content to load
   await new Promise(r => setTimeout(r, 1000));
  
   // Scroll back up
   window.scrollTo(0, 0);
  
   // Add a marker to verify JS ran
   document.body.setAttribute('data-crawl4ai', 'executed');
   """
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       js_code=[js_code],
       wait_for="css:body",
       delay_before_return_html=1.0
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://httpbin.org/html",
           config=run_config
       )
      
       print(f"n Page crawled with JS execution")
       print(f" Status: {result.status_code}")
       print(f" Content length: {len(result.markdown.raw_markdown)} chars")
      
   return result


result = asyncio.run(javascript_execution_demo())


print("n" + "="*60)
print(" PART 9: LLM-BASED EXTRACTION")
print("="*60)


from crawl4ai import LLMExtractionStrategy, LLMConfig


class Article(BaseModel):
   title: str = Field(description="The article title")
   summary: str = Field(description="A brief summary")
   topics: List[str] = Field(description="Main topics covered")


async def llm_extraction_demo():
   """Use LLM to intelligently extract and structure data."""
   print("n LLM-based extraction setup...")
  
   import os
   api_key = os.getenv('OPENAI_API_KEY')
  
   if not api_key:
       print("n No OPENAI_API_KEY found. Showing setup code only.")
       print("nTo enable LLM extraction, run:")
       print("   import os")
       print("   os.environ['OPENAI_API_KEY'] = 'sk-your-key-here'")
       print("n--- Example Code ---")
       example_code = '''
from crawl4ai import LLMExtractionStrategy, LLMConfig
from pydantic import BaseModel, Field


class Product(BaseModel):
   name: str = Field(description="Product name")
   price: str = Field(description="Product price")


llm_strategy = LLMExtractionStrategy(
   llm_config=LLMConfig(
       provider="openai/gpt-4o-mini",  # or "ollama/llama3"
       api_token=os.getenv('OPENAI_API_KEY')
   ),
   schema=Product.model_json_schema(),
   extraction_type="schema",
   instruction="Extract all products with prices."
)


run_config = CrawlerRunConfig(
   extraction_strategy=llm_strategy,
   cache_mode=CacheMode.BYPASS
)


async with AsyncWebCrawler() as crawler:
   result = await crawler.arun(url="https://example.com", config=run_config)
   products = json.loads(result.extracted_content)
'''
       print(example_code)
       return None
  
   llm_strategy = LLMExtractionStrategy(
       llm_config=LLMConfig(
           provider="openai/gpt-4o-mini",
           api_token=api_key
       ),
       schema=Article.model_json_schema(),
       extraction_type="schema",
       instruction="Extract article titles and summaries."
   )
  
   run_config = CrawlerRunConfig(
       extraction_strategy=llm_strategy,
       cache_mode=CacheMode.BYPASS
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://news.ycombinator.com",
           config=run_config
       )
      
       if result.extracted_content:
           data = json.loads(result.extracted_content)
           print(f"n LLM extracted:")
           print(json.dumps(data, indent=2)[:1000])
          
   return result


result = asyncio.run(llm_extraction_demo())

print("n" + "="*60)
print(" PART 10: DEEP CRAWLING")
print("="*60)


from crawl4ai.deep_crawling import BFSDeepCrawlStrategy
from crawl4ai.deep_crawling.filters import FilterChain, URLPatternFilter, DomainFilter


async def deep_crawl_demo():
   """Crawl multiple pages starting from a seed URL using BFS."""
   print("n Starting deep crawl with BFS strategy...")
  
   filter_chain = FilterChain([
       DomainFilter(
           allowed_domains=["docs.crawl4ai.com"],
           blocked_domains=[]
       ),
       URLPatternFilter(
           patterns=["*quickstart*", "*installation*", "*examples*"]
       )
   ])
  
   deep_crawl_strategy = BFSDeepCrawlStrategy(
       max_depth=2,
       max_pages=5,
       filter_chain=filter_chain,
       include_external=False
   )
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       deep_crawl_strategy=deep_crawl_strategy
   )
  
   pages_crawled = []
  
   async with AsyncWebCrawler() as crawler:
       results = await crawler.arun(
           url="https://docs.crawl4ai.com/",
           config=run_config
       )
      
       if isinstance(results, list):
           for result in results:
               pages_crawled.append(result.url)
               print(f"   Crawled: {result.url}")
               print(f"      Content: {len(result.markdown.raw_markdown)} chars")
       else:
           pages_crawled.append(results.url)
           print(f"   Crawled: {results.url}")
           print(f"      Content: {len(results.markdown.raw_markdown)} chars")
  
   print(f"n Total pages crawled: {len(pages_crawled)}")
   return pages_crawled


pages = asyncio.run(deep_crawl_demo())


print("n" + "="*60)
print(" PART 11: MULTI-URL CONCURRENT CRAWLING")
print("="*60)


async def multi_url_crawl():
   """Crawl multiple URLs concurrently for maximum efficiency."""
   print("n Crawling multiple URLs concurrently...")
  
   urls = [
       "https://httpbin.org/html",
       "https://httpbin.org/robots.txt",
       "https://httpbin.org/json",
       "https://example.com",
       "https://httpbin.org/headers"
   ]
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       verbose=False
   )
  
   async with AsyncWebCrawler() as crawler:
       results = await crawler.arun_many(
           urls=urls,
           config=run_config
       )
      
       print(f"n Results Summary:")
       print(f"{'URL':<40} {'Status':<10} {'Content':<15}")
       print("-" * 65)
      
       for result in results:
           url_short = result.url[:38] + ".." if len(result.url) > 40 else result.url
           status = "" if result.success else ""
           content_len = f"{len(result.markdown.raw_markdown):,} chars" if result.success else "N/A"
           print(f"{url_short:<40} {status:<10} {content_len:<15}")
          
   return results


results = asyncio.run(multi_url_crawl())


print("n" + "="*60)
print(" PART 12: SCREENSHOTS & MEDIA")
print("="*60)


async def screenshot_demo():
   """Capture screenshots and extract media from pages."""
   print("n Capturing screenshot and extracting media...")
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       screenshot=True,
       pdf=False,
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://en.wikipedia.org/wiki/Web_scraping",
           config=run_config
       )
      
       print(f"n Crawl complete!")
       print(f" Screenshot captured: {result.screenshot is not None}")
      
       if result.screenshot:
           print(f"   Screenshot size: {len(result.screenshot)} bytes (base64)")
          
       if result.media and 'images' in result.media:
           images = result.media['images']
           print(f"n Found {len(images)} images:")
           for img in images[:5]:
               print(f"   • {img.get('src', 'N/A')[:60]}...")
              
   return result


result = asyncio.run(screenshot_demo())

print("n" + "="*60)
print(" PART 13: LINK EXTRACTION")
print("="*60)


async def link_extraction_demo():
   """Extract and analyze all links from a page."""
   print("n Extracting and analyzing links...")
  
   run_config = CrawlerRunConfig(cache_mode=CacheMode.BYPASS)
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://docs.crawl4ai.com/",
           config=run_config
       )
      
       internal_links = result.links.get('internal', [])
       external_links = result.links.get('external', [])
      
       print(f"n Link Analysis:")
       print(f"   Internal links: {len(internal_links)}")
       print(f"   External links: {len(external_links)}")
      
       print(f"n--- Sample Internal Links (first 5) ---")
       for link in internal_links[:5]:
           print(f"   • {link.get('href', 'N/A')[:60]}")
          
       print(f"n--- Sample External Links (first 5) ---")
       for link in external_links[:5]:
           print(f"   • {link.get('href', 'N/A')[:60]}")
          
   return result


result = asyncio.run(link_extraction_demo())


print("n" + "="*60)
print(" PART 14: CONTENT SELECTION")
print("="*60)


async def content_selection_demo():
   """Target specific content using CSS selectors."""
   print("n Targeting specific content with CSS selectors...")
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       css_selector="article, main, .content, #content, #mw-content-text",
       excluded_tags=["nav", "footer", "header", "aside"],
       remove_overlay_elements=True
   )
  
   async with AsyncWebCrawler() as crawler:
       result = await crawler.arun(
           url="https://en.wikipedia.org/wiki/Web_scraping",
           config=run_config
       )
      
       print(f"n Content extracted with targeting")
       print(f" Markdown length: {len(result.markdown.raw_markdown):,} chars")
       print(f"n--- Preview (first 500 chars) ---")
       print(result.markdown.raw_markdown[:500])
      
   return result


result = asyncio.run(content_selection_demo())


print("n" + "="*60)
print(" PART 15: SESSION MANAGEMENT")
print("="*60)


async def session_management_demo():
   """Maintain browser sessions across multiple requests."""
   print("n Demonstrating session management...")
  
   browser_config = BrowserConfig(headless=True)
  
   async with AsyncWebCrawler(config=browser_config) as crawler:
       session_id = "my_session"
      
       result1 = await crawler.arun(
           url="https://httpbin.org/cookies/set?session=demo123",
           config=CrawlerRunConfig(
               cache_mode=CacheMode.BYPASS,
               session_id=session_id
           )
       )
       print(f"  Step 1: Set cookies - Success: {result1.success}")
      
       result2 = await crawler.arun(
           url="https://httpbin.org/cookies",
           config=CrawlerRunConfig(
               cache_mode=CacheMode.BYPASS,
               session_id=session_id
           )
       )
       print(f"  Step 2: Read cookies - Success: {result2.success}")
       print(f"n Cookie Response:")
       print(result2.markdown.raw_markdown[:300])
      
   return result2


result = asyncio.run(session_management_demo())

print("n" + "="*60)
print(" PART 16: COMPLETE REAL-WORLD EXAMPLE")
print("="*60)


async def complete_example():
   """Complete example combining CSS extraction with content filtering."""
   print("n Running complete example: Hacker News scraper with filtering")
  
   schema = {
       "name": "HN Stories",
       "baseSelector": "tr.athing",
       "fields": [
           {"name": "rank", "selector": "span.rank", "type": "text"},
           {"name": "title", "selector": "span.titleline > a", "type": "text"},
           {"name": "url", "selector": "span.titleline > a", "type": "attribute", "attribute": "href"},
           {"name": "site", "selector": "span.sitestr", "type": "text"}
       ]
   }
  
   browser_config = BrowserConfig(
       headless=True,
       viewport_width=1920,
       viewport_height=1080
   )
  
   run_config = CrawlerRunConfig(
       cache_mode=CacheMode.BYPASS,
       extraction_strategy=JsonCssExtractionStrategy(schema),
       markdown_generator=DefaultMarkdownGenerator(
           content_filter=PruningContentFilter(threshold=0.4)
       )
   )
  
   async with AsyncWebCrawler(config=browser_config) as crawler:
       result = await crawler.arun(
           url="https://news.ycombinator.com",
           config=run_config
       )
      
       if result.extracted_content:
           stories = json.loads(result.extracted_content)
          
           print(f"n Successfully extracted {len(stories)} stories!")
           print(f"n{'='*70}")
           print(" TOP HACKER NEWS STORIES")
           print("="*70)
          
           for story in stories[:15]:
               rank = story.get('rank', '?').strip('.') if story.get('rank') else '?'
               title = story.get('title', 'No title')[:50]
               site = story.get('site', 'N/A')
               url = story.get('url', '')[:30]
               print(f"  #{rank:<3} {title:<50} ({site})")
              
           print("="*70)
          
           return stories
  
   return []


stories = asyncio.run(complete_example())


print("n" + "="*60)
print(" BONUS: SAVING RESULTS")
print("="*60)


if stories:
   with open('hacker_news_stories.json', 'w') as f:
       json.dump(stories, f, indent=2)
   print(f" Saved {len(stories)} stories to 'hacker_news_stories.json'")
   print("nTo download in Colab:")
   print("   from google.colab import files")
   print("   files.download('hacker_news_stories.json')")


print("n" + "="*60)
print(" TUTORIAL COMPLETE!")
print("="*60)


print("""
 What you learned:


1. Basic crawling with AsyncWebCrawler
2. Browser & crawler configuration
3. Markdown generation (raw vs fit)
4. BM25 query-based content filtering
5. CSS-based structured data extraction
6. Advanced CSS extraction (Hacker News)
7. JavaScript execution for dynamic content
8. LLM-based extraction setup
9. Deep crawling with BFS strategy
10. Multi-URL concurrent crawling
11. Screenshots & media extraction
12. Link extraction & analysis
13. Content targeting with CSS selectors
14. Session management
15. Complete real-world scraping example


 RESOURCES:
 • Docs: https://docs.crawl4ai.com/
 • GitHub: https://github.com/unclecode/crawl4ai
 • Discord: https://discord.gg/jP8KfhDhyN


 Happy Crawling with Crawl4AI!
""")