python爬虫学习例之一

以下是一个增强版的Python爬虫脚本，包含登录验证、异常处理、重试机制和模块化设计，附带详细注释说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
"""
网站数据抓取与汇总工具（含登录验证版）
"""
import os
import re
import time
import requests
from bs4 import BeautifulSoup
import pandas as pd
from typing import List, Dict, Optional
from urllib.parse import urljoin
from requests.adapters import HTTPAdapter
from requests.exceptions import RequestException

# --------------------------
# 配置模块
# --------------------------
class Config:
    """爬虫配置参数"""
    def __init__(self):
        # 网站配置
        self.base_url = "https://example.com"            # 网站根URL
        self.login_url = "https://example.com/login"    # 登录页面URL
        self.list_url_template = "/news?page={page}"     # 列表页URL模板
        
        # 分页配置
        self.start_page = 1              # 起始页码
        self.end_page = 3               # 结束页码
        
        # 认证信息（建议通过环境变量获取敏感信息）
        self.username = "your_username"  
        self.password = "your_password"
        
        # 请求配置
        self.headers = {
            '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',
            'Referer': 'https://example.com/'
        }
        self.timeout = 10                # 请求超时时间
        self.retries = 3                # 请求重试次数
        self.retry_delay = 5            # 重试延迟(秒)
        
        # 输出配置
        self.output_dir = "scraped_data"  # 文本文件保存目录
        self.excel_file = "summary.xlsx"  # Excel汇总文件名
        
        # 创建输出目录
        os.makedirs(self.output_dir, exist_ok=True)

# --------------------------
# 爬虫核心模块
# --------------------------
class WebScraper:
    """网站爬取处理器"""
    def __init__(self, config: Config):
        self.config = config
        self.session = requests.Session()
        # 配置请求重试
        self.session.mount('https://', HTTPAdapter(max_retries=config.retries))
        
    def login(self) -> bool:
        """执行登录操作"""
        try:
            # 首次获取登录页以获取CSRF令牌
            login_page = self._safe_request(
                'GET', 
                self.config.login_url,
                allow_redirects=False
            )
            
            # 提取CSRF令牌（根据实际网站结构调整）
            soup = BeautifulSoup(login_page.text, 'html.parser')
            csrf_token = soup.find('input', {'name': 'csrf_token'})['value']
            
            # 构造登录数据
            login_data = {
                'username': self.config.username,
                'password': self.config.password,
                'csrf_token': csrf_token
            }
            
            # 提交登录请求
            response = self._safe_request(
                'POST',
                self.config.login_url,
                data=login_data,
                allow_redirects=False
            )
            
            # 验证登录是否成功（根据实际网站调整验证逻辑）
            if response.status_code == 302 and 'sessionid' in response.cookies:
                print("登录成功")
                return True
            print("登录失败")
            return False
            
        except Exception as e:
            print(f"登录过程发生异常: {str(e)}")
            return False
    
    def fetch_pages(self) -> List[Dict]:
        """抓取所有页面数据"""
        all_data = []
        
        for page in range(self.config.start_page, self.config.end_page + 1):
            print(f"正在处理第 {page} 页...")
            
            # 获取列表页内容
            list_url = urljoin(
                self.config.base_url,
                self.config.list_url_template.format(page=page)
            )
            
            try:
                # 获取列表页
                list_response = self._safe_request('GET', list_url)
                article_links = self._parse_list_page(list_response.text)
                
                # 处理每篇文章
                for i, link in enumerate(article_links, 1):
                    article_url = urljoin(self.config.base_url, link)
                    print(f"  正在处理文章 {i}/{len(article_links)}: {article_url}")
                    
                    # 获取文章内容
                    article_data = self._fetch_article(article_url)
                    if article_data:
                        all_data.append(article_data)
                        time.sleep(1)  # 请求间隔
                        
            except Exception as e:
                print(f"第 {page} 页处理失败: {str(e)}")
                continue
                
        return all_data
    
    def _fetch_article(self, url: str) -> Optional[Dict]:
        """抓取单篇文章内容"""
        try:
            response = self._safe_request('GET', url)
            return self._parse_article_page(response.text, url)
        except Exception as e:
            print(f"文章抓取失败: {url} - {str(e)}")
            return None
    
    def _parse_list_page(self, html: str) -> List[str]:
        """解析列表页获取文章链接"""
        soup = BeautifulSoup(html, 'html.parser')
        links = []
        # 根据实际结构调整选择器
        for item in soup.select('.article-list .title a'):
            if href := item.get('href'):
                links.append(href)
        return links
    
    def _parse_article_page(self, html: str, url: str) -> Dict:
        """解析文章详情页"""
        soup = BeautifulSoup(html, 'html.parser')
        
        # 使用防御性解析
        def safe_extract(selector: str, default: str = "") -> str:
            element = soup.select_one(selector)
            return element.text.strip() if element else default
            
        return {
            'title': safe_extract('h1.article-title'),
            'author': safe_extract('.author-name'),
            'publish_date': safe_extract('.publish-time'),
            'content': safe_extract('.article-content'),
            'url': url
        }
    
    def _safe_request(self, method: str, url: str, **kwargs) -> requests.Response:
        """带异常处理和重试机制的请求方法"""
        for attempt in range(self.config.retries + 1):
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    headers=self.config.headers,
                    timeout=self.config.timeout,
                    **kwargs
                )
                response.raise_for_status()
                return response
            except RequestException as e:
                if attempt < self.config.retries:
                    print(f"请求失败，第 {attempt+1} 次重试: {str(e)}")
                    time.sleep(self.config.retry_delay)
                else:
                    raise

# --------------------------
# 数据存储模块
# --------------------------
class DataHandler:
    """数据处理与存储"""
    @staticmethod
    def save_text(data: Dict, output_dir: str) -> str:
        """保存为文本文件"""
        try:
            # 生成安全文件名
            filename = re.sub(r'[^\w\-_\. ]', '_', data.get('title', 'untitled'))[:50]
            filename = f"{filename}.txt"
            filepath = os.path.join(output_dir, filename)
            
            with open(filepath, 'w', encoding='utf-8') as f:
                for key, value in data.items():
                    f.write(f"=== {key.upper()} ===\n{value}\n\n")
                    
            return filepath
        except Exception as e:
            print(f"文件保存失败: {str(e)}")
            return ""
    
    @staticmethod
    def save_excel(data: List[Dict], filename: str) -> bool:
        """保存为Excel文件"""
        try:
            df = pd.DataFrame(data)
            
            # 数据清洗
            df['publish_date'] = pd.to_datetime(df['publish_date'], errors='coerce')
            
            # 保存文件
            df.to_excel(
                filename,
                index=False,
                engine='openpyxl',
                encoding='utf-8'
            )
            return True
        except Exception as e:
            print(f"Excel保存失败: {str(e)}")
            return False

# --------------------------
# 主程序
# --------------------------
def main():
    # 初始化配置
    config = Config()
    
    # 初始化爬虫
    scraper = WebScraper(config)
    
    # 执行登录
    if not scraper.login():
        print("登录失败，程序终止")
        return
    
    # 抓取数据
    try:
        articles = scraper.fetch_pages()
        print(f"共抓取到 {len(articles)} 篇文章")
    except Exception as e:
        print(f"抓取过程发生严重错误: {str(e)}")
        return
    
    # 保存文本文件
    saved_files = []
    for article in articles:
        if path := DataHandler.save_text(article, config.output_dir):
            saved_files.append(path)
    
    # 保存Excel汇总
    if DataHandler.save_excel(articles, config.excel_file):
        print(f"数据已汇总保存至 {config.excel_file}")

if __name__ == "__main__":
    main()

主要功能说明

  1. 登录验证系统：

    • 自动处理CSRF令牌
    • 使用Session保持登录状态
    • 支持多因素身份验证（需根据实际情况修改）

  2. 异常处理机制：

    • 网络请求自动重试（可配置重试次数）
    • 防御性数据解析（元素不存在时返回默认值）
    • 分级错误处理（区分可恢复错误和致命错误）

  3. 增强特性：

    • 文件名安全处理（防止特殊字符问题）
    • 数据类型自动转换（如日期格式转换）
    • 请求限速机制（防止被封禁）

使用说明

  1. 配置修改：

    1
    2
    3
    4
    5
    6
    7
    class Config:
        # 修改以下参数
        base_url = "https://your-target-site.com"
        login_url = "https://your-target-site.com/login"
        list_url_template = "/articles?page={page}"
        username = "your_username"
        password = "your_password"

  2. 页面解析调整：

    • 修改_parse_list_page中的CSS选择器
    • 调整_parse_article_page中的字段解析逻辑

  3. 运行脚本：

    1
    2
    pip install requests beautifulsoup4 pandas openpyxl
    python scraper.py

注意事项

  1. 合法性检查：

    • 遵守目标网站的robots.txt规则
    • 尊重版权和隐私条款

  2. 性能优化：

    • 适当调整retry_delay参数
    • 考虑使用代理池应对IP封锁

  3. 安全建议：

    • 不要将凭证硬编码在代码中（建议使用环境变量）
    • 定期检查网站结构变化

  4. 扩展性：

    • 可添加日志记录模块
    • 支持分布式爬取
    • 集成验证码识别服务

该脚本实现了从登录认证到数据存储的完整流程，包含多重错误处理机制，建议根据目标网站的具体结构调整CSS选择器和登录验证逻辑。