Learn the fundamentals of web scraping using Python with requests and BeautifulSoup to extract data from websites programmatically.
Web scraping is the process of automatically extracting data from websites. This lesson covers the fundamentals of web scraping using Python's requests library for HTTP operations and BeautifulSoup for HTML parsing, including ethical considerations and best practices.
Web scraping involves programmatically accessing websites and extracting data from their HTML content. Ethical considerations are paramount when building scrapers.
Ethical Scraping Principles:
1import requests
2import time
3from urllib.robotparser import RobotFileParser
4
5def check_robots_txt(url):
6 """Check if scraping is allowed by robots.txt"""
7 rp = RobotFileParser()
8 rp.set_url(f"{url}/robots.txt")
9 rp.read()
10 return rp.can_fetch("*", url)
11
12class EthicalScraper:
13 def __init__(self, delay=1.0):
14 self.delay = delay
15 self.session = requests.Session()
16
17 # Set respectful headers
18 self.session.headers.update({
19 'User-Agent': 'Educational-Scraper/1.0',
20 'Accept': 'text/html,application/xhtml+xml',
21 'Accept-Language': 'en-US,en;q=0.5',
22 })
23
24 def scrape_page(self, url):
25 # Check robots.txt first
26 if not check_robots_txt(url):
27 print(f"Scraping not allowed by robots.txt for {url}")
28 return None
29
30 # Implement delay
31 time.sleep(self.delay)
32
33 try:
34 response = self.session.get(url, timeout=10)
35 response.raise_for_status()
36 return response
37 except requests.RequestException as e:
38 print(f"Error scraping {url}: {e}")
39 return NoneBest Practices:
Remember: just because you can scrape doesn't mean you should. Always consider the website's perspective and server load.
BeautifulSoup provides a powerful interface for parsing HTML and XML documents, making it easy to extract data from web pages.
BeautifulSoup Fundamentals:
1from bs4 import BeautifulSoup
2import requests
3
4# Parse HTML content
5html_content = """
6<html>
7<head><title>Sample Page</title></head>
8<body>
9 <div class="container">
10 <h1 id="main-title">News Articles</h1>
11 <article class="news-item">
12 <h2>Article Title 1</h2>
13 <p class="author">By John Doe</p>
14 <p class="content">Article content here...</p>
15 <div class="tags">
16 <span>Technology</span>
17 <span>AI</span>
18 </div>
19 </article>
20 <article class="news-item">
21 <h2>Article Title 2</h2>
22 <p class="author">By Jane Smith</p>
23 <p class="content">More article content...</p>
24 </article>
25 </div>
26</body>
27</html>
28"""
29
30soup = BeautifulSoup(html_content, 'html.parser')
31
32# Find elements by tag
33title = soup.find('title').text
34print(f"Page title: {title}")
35
36# Find by class
37articles = soup.find_all('article', class_='news-item')
38for article in articles:
39 title = article.find('h2').text
40 author = article.find('p', class_='author').text
41 content = article.find('p', class_='content').text
42 print(f"Title: {title}")
43 print(f"Author: {author}")
44 print(f"Content: {content[:50]}...")
45
46# CSS Selectors
47titles = soup.select('article h2') # All h2 inside article tags
48authors = soup.select('.author') # All elements with class 'author'
49tags = soup.select('.tags span') # All span tags inside tags div
50
51# Extract attributes
52for link in soup.find_all('a'):
53 href = link.get('href')
54 text = link.text
55 print(f"Link: {text} -> {href}")
56
57# Navigate the tree
58first_article = soup.find('article')
59next_sibling = first_article.find_next_sibling('article')
60parent = first_article.parentAdvanced Parsing Techniques:
1# Handle missing elements safely
2def safe_extract(element, selector, default="Not found"):
3 found = element.select_one(selector)
4 return found.text.strip() if found else default
5
6# Extract structured data
7def extract_article_data(article_element):
8 return {
9 'title': safe_extract(article_element, 'h2'),
10 'author': safe_extract(article_element, '.author'),
11 'content': safe_extract(article_element, '.content'),
12 'tags': [tag.text for tag in article_element.select('.tags span')],
13 'url': article_element.find('a', href=True)['href'] if article_element.find('a', href=True) else None
14 }
15
16# Process all articles
17articles_data = []
18for article in soup.find_all('article', class_='news-item'):
19 data = extract_article_data(article)
20 articles_data.append(data)BeautifulSoup handles malformed HTML gracefully and provides intuitive methods for navigating and extracting data.
Effective web scraping requires proper HTTP request management and robust response handling to deal with various scenarios.
Advanced Request Management:
1import requests
2from requests.adapters import HTTPAdapter
3from urllib3.util.retry import Retry
4import time
5
6class RobustScraper:
7 def __init__(self):
8 self.session = requests.Session()
9
10 # Configure retry strategy
11 retry_strategy = Retry(
12 total=3,
13 backoff_factor=1,
14 status_forcelist=[429, 500, 502, 503, 504],
15 method_whitelist=["HEAD", "GET", "OPTIONS"]
16 )
17
18 adapter = HTTPAdapter(max_retries=retry_strategy)
19 self.session.mount("http://", adapter)
20 self.session.mount("https://", adapter)
21
22 # Set proper headers
23 self.session.headers.update({
24 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
25 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
26 'Accept-Language': 'en-US,en;q=0.5',
27 'Accept-Encoding': 'gzip, deflate',
28 'Connection': 'keep-alive',
29 })
30
31 def get_page(self, url, **kwargs):
32 """Get page with comprehensive error handling"""
33 try:
34 response = self.session.get(url, timeout=10, **kwargs)
35
36 # Check status code
37 if response.status_code == 200:
38 return response
39 elif response.status_code == 404:
40 print(f"Page not found: {url}")
41 elif response.status_code == 403:
42 print(f"Access forbidden: {url}")
43 elif response.status_code == 429:
44 print(f"Rate limited: {url}")
45 # Implement exponential backoff
46 time.sleep(60) # Wait 1 minute
47 else:
48 print(f"HTTP {response.status_code}: {url}")
49
50 return None
51
52 except requests.exceptions.ConnectionError:
53 print(f"Connection error: {url}")
54 except requests.exceptions.Timeout:
55 print(f"Timeout: {url}")
56 except requests.exceptions.RequestException as e:
57 print(f"Request error: {e}")
58
59 return None
60
61 def handle_different_content_types(self, response):
62 """Handle various content types"""
63 content_type = response.headers.get('content-type', '').lower()
64
65 if 'application/json' in content_type:
66 return response.json()
67 elif 'text/html' in content_type:
68 return BeautifulSoup(response.content, 'html.parser')
69 elif 'text/xml' in content_type or 'application/xml' in content_type:
70 return BeautifulSoup(response.content, 'xml')
71 else:
72 return response.text
73
74 def scrape_with_cookies(self, url):
75 """Handle cookies and sessions"""
76 # First request to get cookies
77 response = self.session.get(url)
78
79 # Cookies are automatically stored in session
80 # Subsequent requests will include cookies
81 return self.session.get(f"{url}/protected-page")Response Processing:
1def process_response(response):
2 """Comprehensive response processing"""
3 if not response:
4 return None
5
6 # Check encoding
7 if response.encoding != response.apparent_encoding:
8 response.encoding = response.apparent_encoding
9
10 # Handle different content types
11 content_type = response.headers.get('content-type', '')
12
13 if 'charset' not in content_type and 'text' in content_type:
14 # Detect encoding if not specified
15 import chardet
16 detected = chardet.detect(response.content)
17 response.encoding = detected['encoding']
18
19 return response.textProper request management ensures your scraper works reliably across different websites and handles edge cases gracefully.
Extracted web data often requires cleaning, validation, and transformation before storage. Implementing robust data processing pipelines is crucial for reliable scrapers.
Data Cleaning and Validation:
1import re
2import html
3from datetime import datetime
4from typing import Optional, List, Dict, Any
5
6class DataProcessor:
7 """Clean and process scraped data"""
8
9 @staticmethod
10 def clean_text(text: str) -> str:
11 """Clean and normalize text content"""
12 if not text:
13 return ""
14
15 # Decode HTML entities
16 text = html.unescape(text)
17
18 # Remove extra whitespace
19 text = re.sub(r's+', ' ', text.strip())
20
21 # Remove common unwanted characters
22 text = text.replace(' ', ' ') # Non-breaking space
23 text = text.replace('', '') # Zero-width space
24 text = text.replace('
', ' ') # Line separator
25 text = text.replace('
', ' ') # Paragraph separator
26
27 return text
28
29 @staticmethod
30 def clean_url(url: str, base_url: str = "") -> Optional[str]:
31 """Clean and validate URLs"""
32 if not url:
33 return None
34
35 # Remove whitespace
36 url = url.strip()
37
38 # Handle relative URLs
39 if url.startswith('//'):
40 url = 'https:' + url
41 elif url.startswith('/'):
42 url = base_url.rstrip('/') + url
43 elif not url.startswith(('http://', 'https://')):
44 url = base_url.rstrip('/') + '/' + url
45
46 # Validate URL format
47 if not re.match(r'https?://.+', url):
48 return None
49
50 return url
51
52 @staticmethod
53 def parse_date(date_str: str) -> Optional[str]:
54 """Parse various date formats"""
55 if not date_str:
56 return None
57
58 # Common date patterns
59 patterns = [
60 '%Y-%m-%d',
61 '%Y-%m-%d %H:%M:%S',
62 '%B %d, %Y',
63 '%b %d, %Y',
64 '%d %B %Y',
65 '%d %b %Y',
66 '%m/%d/%Y',
67 '%d/%m/%Y'
68 ]
69
70 # Clean date string
71 date_str = re.sub(r'[^ws:/-]', '', date_str).strip()
72
73 for pattern in patterns:
74 try:
75 parsed_date = datetime.strptime(date_str, pattern)
76 return parsed_date.strftime('%Y-%m-%d')
77 except ValueError:
78 continue
79
80 return None
81
82 @staticmethod
83 def validate_email(email: str) -> bool:
84 """Validate email format"""
85 pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
86 return bool(re.match(pattern, email))
87
88 @staticmethod
89 def extract_numbers(text: str) -> List[float]:
90 """Extract numeric values from text"""
91 pattern = r'[-+]?d*.?d+'
92 matches = re.findall(pattern, text)
93 return [float(match) for match in matches]
94
95class DataStorage:
96 """Handle data storage in various formats"""
97
98 @staticmethod
99 def save_to_json(data: List[Dict], filename: str, indent: int = 2):
100 """Save data to JSON with proper encoding"""
101 import json
102
103 with open(filename, 'w', encoding='utf-8') as f:
104 json.dump(data, f, indent=indent, ensure_ascii=False, default=str)
105
106 @staticmethod
107 def save_to_csv(data: List[Dict], filename: str):
108 """Save data to CSV with proper handling"""
109 import csv
110
111 if not data:
112 return
113
114 fieldnames = data[0].keys()
115
116 with open(filename, 'w', newline='', encoding='utf-8') as f:
117 writer = csv.DictWriter(f, fieldnames=fieldnames)
118 writer.writeheader()
119
120 for row in data:
121 # Clean data for CSV
122 cleaned_row = {}
123 for key, value in row.items():
124 if isinstance(value, list):
125 cleaned_row[key] = ', '.join(map(str, value))
126 elif isinstance(value, str):
127 cleaned_row[key] = value.replace('
128', ' ').replace('
', '')
129 else:
130 cleaned_row[key] = str(value) if value is not None else ''
131
132 writer.writerow(cleaned_row)
133
134 @staticmethod
135 def save_to_database(data: List[Dict], table_name: str, db_path: str = 'scraped_data.db'):
136 """Save data to SQLite database"""
137 import sqlite3
138
139 if not data:
140 return
141
142 conn = sqlite3.connect(db_path)
143
144 # Create table dynamically based on data structure
145 columns = list(data[0].keys())
146 create_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join([f'{col} TEXT' for col in columns])})"
147
148 conn.execute(create_sql)
149
150 # Insert data
151 placeholders = ', '.join(['?' for _ in columns])
152 insert_sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"
153
154 for row in data:
155 values = [str(row.get(col, '')) for col in columns]
156 conn.execute(insert_sql, values)
157
158 conn.commit()
159 conn.close()
160
161# Usage example
162def process_scraped_articles(raw_articles):
163 """Process and store scraped articles"""
164 processor = DataProcessor()
165 cleaned_articles = []
166
167 for article in raw_articles:
168 cleaned = {
169 'title': processor.clean_text(article.get('title', '')),
170 'author': processor.clean_text(article.get('author', '')),
171 'content': processor.clean_text(article.get('content', '')),
172 'url': processor.clean_url(article.get('url', '')),
173 'publish_date': processor.parse_date(article.get('date', '')),
174 'tags': [processor.clean_text(tag) for tag in article.get('tags', [])],
175 'scraped_at': datetime.now().isoformat()
176 }
177
178 # Validate required fields
179 if cleaned['title'] and cleaned['content']:
180 cleaned_articles.append(cleaned)
181
182 # Store in multiple formats
183 storage = DataStorage()
184 storage.save_to_json(cleaned_articles, 'articles.json')
185 storage.save_to_csv(cleaned_articles, 'articles.csv')
186 storage.save_to_database(cleaned_articles, 'articles')
187
188 return cleaned_articlesProper data processing ensures your scraped data is clean, consistent, and ready for analysis or further use.
Build a comprehensive news scraper that extracts articles, analyzes content, and stores structured data
1import requests
2from bs4 import BeautifulSoup
3import time
4import csv
5import json
6import re
7from datetime import datetime, timedelta
8from urllib.parse import urljoin, urlparse
9from typing import Dict, List, Optional, Any
10import logging
11
12# Configure logging
13logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
14logger = logging.getLogger(__name__)
15
16class RateLimiter:
17 """Rate limiter to be respectful to websites"""
18
19 def __init__(self, min_delay: float = 1.0):
20 self.min_delay = min_delay
21 self.last_request_time = 0
22
23 def wait(self):
24 """Wait appropriate time before next request"""
25 current_time = time.time()
26 time_since_last = current_time - self.last_request_time
27
28 if time_since_last < self.min_delay:
29 sleep_time = self.min_delay - time_since_last
30 logger.info(f"Rate limiting: sleeping for {sleep_time:.2f} seconds")
31 time.sleep(sleep_time)
32
33 self.last_request_time = time.time()
34
35class WebScraper:
36 """Base web scraping class with common functionality"""
37
38 def __init__(self, delay: float = 1.0):
39 self.session = requests.Session()
40 self.rate_limiter = RateLimiter(delay)
41
42 # Set user agent to identify as a legitimate browser
43 self.session.headers.update({
44 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
45 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
46 'Accept-Language': 'en-US,en;q=0.5',
47 'Accept-Encoding': 'gzip, deflate',
48 'Connection': 'keep-alive',
49 })
50
51 def get_page(self, url: str, timeout: int = 10) -> Optional[BeautifulSoup]:
52 """Fetch and parse a web page with error handling"""
53 try:
54 self.rate_limiter.wait() # Respect rate limiting
55
56 logger.info(f"Fetching: {url}")
57 response = self.session.get(url, timeout=timeout)
58 response.raise_for_status() # Raise exception for bad status codes
59
60 # Check if content is HTML
61 content_type = response.headers.get('content-type', '').lower()
62 if 'text/html' not in content_type:
63 logger.warning(f"Non-HTML content type: {content_type}")
64 return None
65
66 # Parse HTML
67 soup = BeautifulSoup(response.content, 'html.parser')
68 return soup
69
70 except requests.exceptions.RequestException as e:
71 logger.error(f"Request failed for {url}: {e}")
72 return None
73 except Exception as e:
74 logger.error(f"Unexpected error parsing {url}: {e}")
75 return None
76
77 def clean_text(self, text: str) -> str:
78 """Clean and normalize extracted text"""
79 if not text:
80 return ""
81
82 # Remove extra whitespace and normalize
83 text = re.sub(r'\s+', ' ', text.strip())
84
85 # Remove common unwanted characters
86 text = text.replace('\xa0', ' ') # Non-breaking space
87 text = text.replace('\u200b', '') # Zero-width space
88
89 return text
90
91class NewsArticle:
92 """Data class for news articles"""
93
94 def __init__(self):
95 self.title: str = ""
96 self.url: str = ""
97 self.author: str = ""
98 self.publish_date: str = ""
99 self.content: str = ""
100 self.summary: str = ""
101 self.tags: List[str] = []
102 self.word_count: int = 0
103 self.scraped_at: str = datetime.now().isoformat()
104
105 def to_dict(self) -> Dict[str, Any]:
106 """Convert article to dictionary"""
107 return {
108 'title': self.title,
109 'url': self.url,
110 'author': self.author,
111 'publish_date': self.publish_date,
112 'content': self.content,
113 'summary': self.summary,
114 'tags': self.tags,
115 'word_count': self.word_count,
116 'scraped_at': self.scraped_at
117 }
118
119class NewsScraper(WebScraper):
120 """News article scraper with multiple site support"""
121
122 def __init__(self, delay: float = 2.0):
123 super().__init__(delay)
124 self.scraped_articles: List[NewsArticle] = []
125
126 def scrape_generic_article(self, url: str) -> Optional[NewsArticle]:
127 """Scrape an article using generic selectors"""
128 soup = self.get_page(url)
129 if not soup:
130 return None
131
132 article = NewsArticle()
133 article.url = url
134
135 try:
136 # Extract title using various common selectors
137 title_selectors = [
138 'h1.entry-title', 'h1.article-title', 'h1.post-title',
139 'h1[class*="title"]', 'h1', '.headline h1', 'title'
140 ]
141
142 for selector in title_selectors:
143 title_element = soup.select_one(selector)
144 if title_element:
145 article.title = self.clean_text(title_element.get_text())
146 break
147
148 # Extract author
149 author_selectors = [
150 '.author', '.by-author', '[rel="author"]',
151 '.article-author', '.post-author', '[class*="author"]'
152 ]
153
154 for selector in author_selectors:
155 author_element = soup.select_one(selector)
156 if author_element:
157 article.author = self.clean_text(author_element.get_text())
158 break
159
160 # Extract publish date
161 date_selectors = [
162 'time[datetime]', '.publish-date', '.article-date',
163 '.post-date', '[class*="date"]', 'time'
164 ]
165
166 for selector in date_selectors:
167 date_element = soup.select_one(selector)
168 if date_element:
169 # Try to get datetime attribute first
170 date_str = date_element.get('datetime', date_element.get_text())
171 article.publish_date = self.clean_text(date_str)
172 break
173
174 # Extract main content
175 content_selectors = [
176 '.entry-content', '.article-content', '.post-content',
177 '.content', 'article', '.story-body', '[class*="content"]'
178 ]
179
180 content_text = ""
181 for selector in content_selectors:
182 content_element = soup.select_one(selector)
183 if content_element:
184 # Remove unwanted elements
185 for unwanted in content_element.find_all(['script', 'style', 'nav', 'aside', 'footer']):
186 unwanted.decompose()
187
188 paragraphs = content_element.find_all(['p', 'div'], string=True)
189 content_text = '\n'.join([self.clean_text(p.get_text()) for p in paragraphs if p.get_text().strip()])
190 break
191
192 article.content = content_text
193 article.word_count = len(content_text.split())
194
195 # Extract tags/categories
196 tag_selectors = [
197 '.tags a', '.categories a', '.article-tags a',
198 '[class*="tag"] a', '.meta-tags a'
199 ]
200
201 tags = []
202 for selector in tag_selectors:
203 tag_elements = soup.select(selector)
204 for tag_elem in tag_elements:
205 tag_text = self.clean_text(tag_elem.get_text())
206 if tag_text and tag_text not in tags:
207 tags.append(tag_text)
208
209 article.tags = tags[:10] # Limit to 10 tags
210
211 # Generate summary (first 200 characters of content)
212 if article.content:
213 article.summary = article.content[:200] + "..." if len(article.content) > 200 else article.content
214
215 return article
216
217 except Exception as e:
218 logger.error(f"Error extracting article data from {url}: {e}")
219 return None
220
221 def scrape_news_site_listing(self, listing_url: str, max_articles: int = 10) -> List[str]:
222 """Scrape article URLs from a news site listing page"""
223 soup = self.get_page(listing_url)
224 if not soup:
225 return []
226
227 article_links = []
228
229 # Common selectors for article links
230 link_selectors = [
231 'article h2 a', 'article h3 a', 'article .headline a',
232 '.entry-title a', '.post-title a', '.article-title a',
233 'h2 a[href*="/article"]', 'h3 a[href*="/news"]',
234 '.story-headline a', '.headline a'
235 ]
236
237 base_url = f"{urlparse(listing_url).scheme}://{urlparse(listing_url).netloc}"
238
239 for selector in link_selectors:
240 links = soup.select(selector)
241 for link in links:
242 href = link.get('href')
243 if href:
244 # Convert relative URLs to absolute
245 full_url = urljoin(base_url, href)
246 if full_url not in article_links and len(article_links) < max_articles:
247 article_links.append(full_url)
248
249 if len(article_links) >= max_articles:
250 break
251
252 logger.info(f"Found {len(article_links)} article links from {listing_url}")
253 return article_links
254
255 def scrape_articles_from_urls(self, urls: List[str]) -> List[NewsArticle]:
256 """Scrape multiple articles from a list of URLs"""
257 articles = []
258
259 for i, url in enumerate(urls, 1):
260 logger.info(f"Scraping article {i}/{len(urls)}: {url}")
261
262 article = self.scrape_generic_article(url)
263 if article and article.title and article.content:
264 articles.append(article)
265 logger.info(f"Successfully scraped: {article.title[:50]}...")
266 else:
267 logger.warning(f"Failed to scrape article from: {url}")
268
269 return articles
270
271 def analyze_articles(self, articles: List[NewsArticle]) -> Dict[str, Any]:
272 """Analyze scraped articles and generate statistics"""
273 if not articles:
274 return {}
275
276 analysis = {
277 'total_articles': len(articles),
278 'average_word_count': sum(a.word_count for a in articles) / len(articles),
279 'total_words': sum(a.word_count for a in articles),
280 'date_range': {},
281 'top_authors': {},
282 'common_tags': {},
283 'word_count_distribution': {
284 'short': 0, # < 300 words
285 'medium': 0, # 300-800 words
286 'long': 0 # > 800 words
287 }
288 }
289
290 # Analyze authors
291 for article in articles:
292 if article.author:
293 analysis['top_authors'][article.author] = analysis['top_authors'].get(article.author, 0) + 1
294
295 # Analyze tags
296 for article in articles:
297 for tag in article.tags:
298 analysis['common_tags'][tag] = analysis['common_tags'].get(tag, 0) + 1
299
300 # Analyze word count distribution
301 for article in articles:
302 if article.word_count < 300:
303 analysis['word_count_distribution']['short'] += 1
304 elif article.word_count < 800:
305 analysis['word_count_distribution']['medium'] += 1
306 else:
307 analysis['word_count_distribution']['long'] += 1
308
309 # Sort authors and tags by frequency
310 analysis['top_authors'] = dict(sorted(analysis['top_authors'].items(), key=lambda x: x[1], reverse=True)[:10])
311 analysis['common_tags'] = dict(sorted(analysis['common_tags'].items(), key=lambda x: x[1], reverse=True)[:15])
312
313 return analysis
314
315class DataStorage:
316 """Handle data storage in various formats"""
317
318 @staticmethod
319 def save_to_json(articles: List[NewsArticle], filename: str):
320 """Save articles to JSON file"""
321 data = [article.to_dict() for article in articles]
322
323 with open(filename, 'w', encoding='utf-8') as f:
324 json.dump(data, f, indent=2, ensure_ascii=False)
325
326 logger.info(f"Saved {len(articles)} articles to {filename}")
327
328 @staticmethod
329 def save_to_csv(articles: List[NewsArticle], filename: str):
330 """Save articles to CSV file"""
331 if not articles:
332 return
333
334 fieldnames = ['title', 'url', 'author', 'publish_date', 'summary', 'word_count', 'tags', 'scraped_at']
335
336 with open(filename, 'w', newline='', encoding='utf-8') as f:
337 writer = csv.DictWriter(f, fieldnames=fieldnames)
338 writer.writeheader()
339
340 for article in articles:
341 row = article.to_dict()
342 row['tags'] = ', '.join(row['tags']) # Convert list to string for CSV
343 row['summary'] = row['summary'].replace('\n', ' ') # Clean newlines
344 writer.writerow({k: row[k] for k in fieldnames})
345
346 logger.info(f"Saved {len(articles)} articles to {filename}")
347
348 @staticmethod
349 def print_analysis(analysis: Dict[str, Any]):
350 """Print formatted analysis results"""
351 print("\n" + "="*50)
352 print(" ARTICLE ANALYSIS REPORT")
353 print("="*50)
354
355 print(f"\nTotal Articles Scraped: {analysis['total_articles']}")
356 print(f"Average Word Count: {analysis['average_word_count']:.1f}")
357 print(f"Total Words: {analysis['total_words']:,}")
358
359 print("\nWord Count Distribution:")
360 dist = analysis['word_count_distribution']
361 print(f" Short articles (<300 words): {dist['short']}")
362 print(f" Medium articles (300-800 words): {dist['medium']}")
363 print(f" Long articles (>800 words): {dist['long']}")
364
365 if analysis['top_authors']:
366 print("\nTop Authors:")
367 for author, count in list(analysis['top_authors'].items())[:5]:
368 print(f" {author}: {count} articles")
369
370 if analysis['common_tags']:
371 print("\nMost Common Tags:")
372 for tag, count in list(analysis['common_tags'].items())[:10]:
373 print(f" {tag}: {count} occurrences")
374
375# Example usage demonstrating comprehensive web scraping
376def main():
377 """Main function demonstrating the news scraping system"""
378 print("=== News Scraper Demo ===")
379
380 try:
381 # Initialize scraper with 2-second delay between requests
382 scraper = NewsScraper(delay=2.0)
383
384 # Example URLs (replace with actual news sites)
385 # Note: Always check robots.txt and terms of service before scraping
386 example_urls = [
387 "https://example-news-site.com/technology/ai-breakthrough-2023",
388 "https://example-news-site.com/science/climate-research-update",
389 "https://example-news-site.com/tech/programming-trends-2023"
390 ]
391
392 print(f"\nScraping {len(example_urls)} example articles...")
393
394 # In a real implementation, you would:
395 # 1. First scrape a listing page to get article URLs
396 # 2. Then scrape individual articles
397
398 # For demonstration, we'll show the structure:
399 print("\n=== Scraping Process ===")
400 print("1. Fetching article listing page...")
401 print("2. Extracting article URLs...")
402 print("3. Scraping individual articles with rate limiting...")
403 print("4. Processing and analyzing content...")
404 print("5. Storing results in multiple formats...")
405
406 # Simulate scraped articles for demonstration
407 demo_articles = []
408 for i, url in enumerate(example_urls, 1):
409 article = NewsArticle()
410 article.url = url
411 article.title = f"Example Article {i}: Technology Trends"
412 article.author = f"Reporter {i}"
413 article.publish_date = datetime.now().strftime('%Y-%m-%d')
414 article.content = f"This is example content for article {i}. " * 50 # Simulate content
415 article.word_count = len(article.content.split())
416 article.summary = article.content[:200] + "..."
417 article.tags = [f"tag{i}", "technology", "news"]
418 demo_articles.append(article)
419
420 print(f"\nSuccessfully processed {len(demo_articles)} articles")
421
422 # Analyze articles
423 print("\n=== Analyzing Articles ===")
424 analysis = scraper.analyze_articles(demo_articles)
425 DataStorage.print_analysis(analysis)
426
427 # Save data
428 print("\n=== Saving Data ===")
429 DataStorage.save_to_json(demo_articles, "scraped_articles.json")
430 DataStorage.save_to_csv(demo_articles, "scraped_articles.csv")
431
432 print("\n=== Web Scraping Features Demonstrated ===")
433 print("- Ethical scraping with rate limiting and proper headers")
434 print("- Robust error handling for network and parsing issues")
435 print("- Generic content extraction from various website structures")
436 print("- Data cleaning and normalization techniques")
437 print("- Content analysis and statistical reporting")
438 print("- Multiple data storage formats (JSON, CSV)")
439 print("- Structured data models for scraped content")
440 print("- Comprehensive logging and monitoring")
441
442 print("\n=== Important Reminders ===")
443 print("- Always check robots.txt before scraping")
444 print("- Respect website terms of service")
445 print("- Implement appropriate delays between requests")
446 print("- Handle errors gracefully and log activities")
447 print("- Consider using APIs when available instead of scraping")
448
449 except Exception as e:
450 logger.error(f"Demo error: {e}")
451
452if __name__ == "__main__":
453 main()Your progress is saved in this browser, on this device only.