How to Scrape Web Data for Your ML Project with BeautifulSoup and Python 3.13

mkdir ml-web-scraper
cd ml-web-scraper
python3.13 -m venv scraper-env
source scraper-env/bin/activate  # On Windows: scraper-env\Scripts\activate

pip install beautifulsoup4==4.12.3 requests==2.31.0 pandas==2.1.4 lxml==4.9.4 fake-useragent==1.4.0

import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import time
import pandas as pd

def create_session():
    """Create a session with realistic headers"""
    session = requests.Session()
    ua = UserAgent()
    session.headers.update({
        'User-Agent': ua.random,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
        'Accept-Encoding': 'gzip, deflate',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1'
    })
    return session

def scrape_quotes():
    """Scrape quotes from quotes.toscrape.com - perfect for testing"""
    session = create_session()
    base_url = "http://quotes.toscrape.com/page/{}"
    all_quotes = []
    
    for page in range(1, 6):  # First 5 pages
        print(f"Scraping page {page}...")
        
        response = session.get(base_url.format(page))
        soup = BeautifulSoup(response.content, 'lxml')
        
        quotes = soup.find_all('div', class_='quote')
        
        for quote in quotes:
            text = quote.find('span', class_='text').text.strip()
            author = quote.find('small', class_='author').text.strip()
            tags = [tag.text for tag in quote.find_all('a', class_='tag')]
            
            all_quotes.append({
                'text': text,
                'author': author,
                'tags': ', '.join(tags),
                'page': page
            })
        
        # Be respectful - wait between requests
        time.sleep(1)
    
    return all_quotes

if __name__ == "__main__":
    quotes = scrape_quotes()
    df = pd.DataFrame(quotes)
    df.to_csv('quotes_data.csv', index=False)
    print(f"Scraped {len(quotes)} quotes successfully!")
    print(df.head())

python basic_scraper.py

import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import time
import random
from urllib.parse import urljoin, urlparse
import csv

class RobustScraper:
    def __init__(self, base_delay=1, max_delay=5, max_retries=3):
        self.session = requests.Session()
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.max_retries = max_retries
        self.setup_session()
    
    def setup_session(self):
        """Configure session with rotating user agents"""
        ua = UserAgent()
        self.session.headers.update({
            'User-Agent': ua.random,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        })
    
    def smart_delay(self):
        """Random delay to avoid detection"""
        delay = random.uniform(self.base_delay, self.max_delay)
        time.sleep(delay)
    
    def get_with_retries(self, url):
        """Get URL with exponential backoff retry logic"""
        for attempt in range(self.max_retries):
            try:
                response = self.session.get(url, timeout=10)
                
                if response.status_code == 200:
                    return response
                elif response.status_code == 429:  # Too many requests
                    wait_time = (2 ** attempt) * 60  # Exponential backoff
                    print(f"Rate limited. Waiting {wait_time} seconds...")
                    time.sleep(wait_time)
                else:
                    print(f"HTTP {response.status_code} for {url}")
                    
            except requests.exceptions.RequestException as e:
                print(f"Attempt {attempt + 1} failed: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(2 ** attempt)
        
        return None
    
    def scrape_hacker_news_titles(self, pages=3):
        """Scrape Hacker News titles - good for text classification"""
        base_url = "https://news.ycombinator.com/"
        all_stories = []
        
        for page in range(pages):
            if page == 0:
                url = base_url
            else:
                url = f"{base_url}?p={page + 1}"
            
            print(f"Scraping page {page + 1}...")
            
            response = self.get_with_retries(url)
            if not response:
                continue
                
            soup = BeautifulSoup(response.content, 'lxml')
            story_links = soup.find_all('span', class_='titleline')
            
            for story in story_links:
                title_element = story.find('a')
                if title_element:
                    title = title_element.text.strip()
                    link = title_element.get('href', '')
                    
                    # Make absolute URLs
                    if link.startswith('item?'):
                        link = urljoin(base_url, link)
                    
                    all_stories.append({
                        'title': title,
                        'url': link,
                        'source': 'hackernews',
                        'page': page + 1
                    })
            
            self.smart_delay()
        
        return all_stories
    
    def save_to_csv(self, data, filename):
        """Save data to CSV with UTF-8 encoding"""
        if not data:
            print("No data to save")
            return
            
        with open(filename, 'w', newline='', encoding='utf-8') as file:
            writer = csv.DictWriter(file, fieldnames=data[0].keys())
            writer.writeheader()
            writer.writerows(data)
        
        print(f"Saved {len(data)} records to {filename}")

# Usage example
if __name__ == "__main__":
    scraper = RobustScraper()
    stories = scraper.scrape_hacker_news_titles(pages=5)
    scraper.save_to_csv(stories, 'hackernews_titles.csv')

python robust_scraper.py

import requests
from bs4 import BeautifulSoup
import os
from urllib.parse import urljoin, urlparse
from PIL import Image
import hashlib

class ImageScraper:
    def __init__(self, download_dir="images"):
        self.download_dir = download_dir
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
        })
        
        # Create download directory
        os.makedirs(download_dir, exist_ok=True)
    
    def download_image(self, img_url, filename=None):
        """Download single image with validation"""
        try:
            response = self.session.get(img_url, stream=True, timeout=10)
            response.raise_for_status()
            
            # Generate filename if not provided
            if not filename:
                parsed_url = urlparse(img_url)
                filename = os.path.basename(parsed_url.path)
                if not filename or '.' not in filename:
                    filename = f"image_{hashlib.md5(img_url.encode()).hexdigest()[:8]}.jpg"
            
            filepath = os.path.join(self.download_dir, filename)
            
            # Download and validate
            with open(filepath, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
            
            # Verify it's a valid image
            try:
                with Image.open(filepath) as img:
                    width, height = img.size
                    if width < 100 or height < 100:  # Skip tiny images
                        os.remove(filepath)
                        return None
                    return {'filepath': filepath, 'width': width, 'height': height}
            except Exception:
                os.remove(filepath)
                return None
                
        except Exception as e:
            print(f"Failed to download {img_url}: {e}")
            return None
    
    def scrape_unsplash_photos(self, query="technology", count=20):
        """Scrape photos from Unsplash - great for ML training"""
        base_url = f"https://unsplash.com/s/photos/{query}"
        downloaded = []
        
        response = self.session.get(base_url)
        soup = BeautifulSoup(response.content, 'lxml')
        
        # Find image containers
        img_elements = soup.find_all('img', {'data-test': 'photo-grid-single-photo-img'})
        
        print(f"Found {len(img_elements)} images for '{query}'")
        
        for i, img in enumerate(img_elements[:count]):
            if i >= count:
                break
                
            img_url = img.get('src')
            if img_url and 'images.unsplash.com' in img_url:
                # Get higher resolution version
                img_url = img_url.replace('w=400', 'w=800').replace('h=400', 'h=800')
                
                filename = f"{query}_{i+1:03d}.jpg"
                result = self.download_image(img_url, filename)
                
                if result:
                    result['query'] = query
                    result['source_url'] = img_url
                    downloaded.append(result)
                    print(f"Downloaded {len(downloaded)}/{count}: {filename}")
                
                # Be respectful
                import time
                time.sleep(0.5)
        
        return downloaded

# Usage example
if __name__ == "__main__":
    scraper = ImageScraper()
    
    # Download tech-related images
    tech_images = scraper.scrape_unsplash_photos("artificial intelligence", 15)
    
    print(f"\nDownloaded {len(tech_images)} images")
    for img in tech_images[:3]:
        print(f"- {img['filepath']}: {img['width']}x{img['height']}")

import pandas as pd
import re
from datetime import datetime
import html

class DataCleaner:
    def __init__(self):
        self.cleaning_stats = {}
    
    def clean_text(self, text):
        """Clean scraped text for ML training"""
        if pd.isna(text) or text == '':
            return ''
        
        # Convert to string
        text = str(text)
        
        # Decode HTML entities
        text = html.unescape(text)
        
        # Remove extra whitespace and newlines
        text = re.sub(r'\s+', ' ', text).strip()
        
        # Remove common scraping artifacts
        text = re.sub(r'\[.*?\]', '', text)  # Remove [brackets]
        text = re.sub(r'Advertisement', '', text, flags=re.IGNORECASE)
        text = re.sub(r'Read more.*', '', text, flags=re.IGNORECASE)
        
        return text
    
    def remove_duplicates(self, df, text_column):
        """Remove duplicate content - common in web scraping"""
        initial_count = len(df)
        
        # Remove exact duplicates
        df = df.drop_duplicates(subset=[text_column])
        
        # Remove near-duplicates (same first 50 characters)
        if len(df) > 0:
            df['_temp_start'] = df[text_column].str[:50]
            df = df.drop_duplicates(subset=['_temp_start'])
            df = df.drop('_temp_start', axis=1)
        
        removed = initial_count - len(df)
        self.cleaning_stats['duplicates_removed'] = removed
        
        return df
    
    def filter_quality(self, df, text_column, min_length=10, max_length=10000):
        """Filter out low-quality content"""
        initial_count = len(df)
        
        # Remove empty or too short/long content
        df = df[df[text_column].str.len().between(min_length, max_length)]
        
        # Remove content that's mostly numbers or symbols
        df = df[~df[text_column].str.match(r'^[\d\s\W]+$')]
        
        removed = initial_count - len(df)
        self.cleaning_stats['quality_filtered'] = removed
        
        return df
    
    def clean_dataset(self, input_file, output_file, text_column='text'):
        """Complete cleaning pipeline"""
        print(f"Cleaning {input_file}...")
        
        # Load data
        df = pd.read_csv(input_file)
        initial_count = len(df)
        print(f"Initial records: {initial_count}")
        
        # Clean text content
        df[text_column] = df[text_column].apply(self.clean_text)
        
        # Remove duplicates
        df = self.remove_duplicates(df, text_column)
        print(f"After deduplication: {len(df)}")
        
        # Filter quality
        df = self.filter_quality(df, text_column)
        print(f"After quality filter: {len(df)}")
        
        # Add metadata
        df['cleaned_at'] = datetime.now().isoformat()
        df['character_count'] = df[text_column].str.len()
        df['word_count'] = df[text_column].str.split().str.len()
        
        # Save cleaned data
        df.to_csv(output_file, index=False)
        
        # Report results
        final_count = len(df)
        retention_rate = (final_count / initial_count) * 100
        
        print(f"\nCleaning complete!")
        print(f"Retained: {final_count}/{initial_count} ({retention_rate:.1f}%)")
        print(f"Average length: {df['character_count'].mean():.0f} characters")
        print(f"Saved to: {output_file}")
        
        return df

# Usage example
if __name__ == "__main__":
    cleaner = DataCleaner()
    
    # Clean the quotes data we scraped earlier
    if os.path.exists('quotes_data.csv'):
        cleaned_df = cleaner.clean_dataset('quotes_data.csv', 'quotes_clean.csv', 'text')
        
        # Show sample of cleaned data
        print("\nSample cleaned data:")
        print(cleaned_df[['text', 'author', 'word_count']].head())

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import time
import csv
from fake_useragent import UserAgent

class ConcurrentScraper:
    def __init__(self, max_concurrent=5, delay_between_batches=2):
        self.max_concurrent = max_concurrent
        self.delay_between_batches = delay_between_batches
        self.ua = UserAgent()
    
    async def fetch_url(self, session, url, semaphore):
        """Fetch single URL with semaphore limiting"""
        async with semaphore:
            try:
                headers = {'User-Agent': self.ua.random}
                async with session.get(url, headers=headers, timeout=10) as response:
                    if response.status == 200:
                        content = await response.text()
                        return {'url': url, 'content': content, 'status': 'success'}
                    else:
                        return {'url': url, 'content': None, 'status': f'error_{response.status}'}
            except Exception as e:
                return {'url': url, 'content': None, 'status': f'error_{str(e)}'}
    
    async def scrape_multiple_urls(self, urls):
        """Scrape multiple URLs concurrently"""
        semaphore = asyncio.Semaphore(self.max_concurrent)
        results = []
        
        # Process URLs in batches to avoid overwhelming servers
        batch_size = self.max_concurrent * 3
        
        async with aiohttp.ClientSession() as session:
            for i in range(0, len(urls), batch_size):
                batch = urls[i:i + batch_size]
                print(f"Processing batch {i//batch_size + 1}: {len(batch)} URLs")
                
                tasks = [self.fetch_url(session, url, semaphore) for url in batch]
                batch_results = await asyncio.gather(*tasks)
                results.extend(batch_results)
                
                # Respectful delay between batches
                if i + batch_size < len(urls):
                    await asyncio.sleep(self.delay_between_batches)
        
        return results
    
    def extract_reddit_posts(self, html_content, subreddit):
        """Extract post data from Reddit HTML"""
        soup = BeautifulSoup(html_content, 'lxml')
        posts = []
        
        # Reddit post containers
        post_elements = soup.find_all('div', {'data-testid': 'post-container'})
        
        for post in post_elements:
            try:
                title_elem = post.find('h3')
                title = title_elem.text.strip() if title_elem else ''
                
                # Skip if no title
                if not title:
                    continue
                
                # Extract post text if available
                text_elem = post.find('div', {'data-testid': 'post-content'})
                text = text_elem.text.strip() if text_elem else ''
                
                posts.append({
                    'subreddit': subreddit,
                    'title': title,
                    'text': text,
                    'scraped_at': time.time()
                })
                
            except Exception as e:
                continue  # Skip problematic posts
        
        return posts

async def main():
    """Example: Scrape multiple subreddit pages concurrently"""
    scraper = ConcurrentScraper(max_concurrent=3)
    
    # Generate URLs for different subreddits
    subreddits = ['MachineLearning', 'Python', 'artificial', 'datascience']
    urls = []
    
    for subreddit in subreddits:
        for page in range(1, 4):  # First 3 pages of each
            url = f"https://old.reddit.com/r/{subreddit}/?page={page}"
            urls.append(url)
    
    print(f"Starting concurrent scrape of {len(urls)} URLs...")
    start_time = time.time()
    
    # Scrape all URLs concurrently
    results = await scraper.scrape_multiple_urls(urls)
    
    # Extract post data from successful responses
    all_posts = []
    successful_scrapes = 0
    
    for result in results:
        if result['status'] == 'success' and result['content']:
            subreddit = result['url'].split('/r/')[1].split('/')[0]
            posts = scraper.extract_reddit_posts(result['content'], subreddit)
            all_posts.extend(posts)
            successful_scrapes += 1
    
    # Save results
    if all_posts:
        with open('reddit_posts.csv', 'w', newline='', encoding='utf-8') as file:
            writer = csv.DictWriter(file, fieldnames=['subreddit', 'title', 'text', 'scraped_at'])
            writer.writeheader()
            writer.writerows(all_posts)
    
    # Report results
    elapsed = time.time() - start_time
    print(f"\nScraping complete in {elapsed:.1f} seconds")
    print(f"Successful requests: {successful_scrapes}/{len(urls)}")
    print(f"Posts extracted: {len(all_posts)}")
    print(f"Average: {len(all_posts)/elapsed:.1f} posts/second")

# Run the concurrent scraper
if __name__ == "__main__":
    asyncio.run(main())

python concurrent_scraper.py

# Bad - crashes when structure changes
title = soup.find('h1', class_='title').text

# Good - handles structure changes gracefully  
title_elem = soup.find('h1', class_='title')
title = title_elem.text.strip() if title_elem else 'No title found'

# After scraping, always check
if len(scraped_data) == 0:
    print("WARNING: No data scraped! Check your selectors.")
    
# Check data quality
sample_item = scraped_data[0] if scraped_data else {}
print(f"Sample data: {sample_item}")

# Check robots.txt first
import urllib.robotparser

def can_scrape(url):
    rp = urllib.robotparser.RobotFileParser()
    rp.set_url(f"{url}/robots.txt")
    rp.read()
    return rp.can_fetch('*', url)