一文詳解Python HTML/XML解析庫BeautifulSoup4

更新時間：2026年06月02日 09:26:38 作者：河阿里

這段文章詳細(xì)介紹了BeautifulSoup4的使用方法、解析器選擇與性能對比,并提供了實戰(zhàn)案例,文章強調(diào)在生產(chǎn)環(huán)境中推薦使用lxml解析器,同時提供了數(shù)據(jù)清洗、文檔樹修改及優(yōu)化技巧等內(nèi)容,需要的朋友可以參考下

第一章：BeautifulSoup4 是什么

1.1 簡單介紹BeautifulSoup4

BeautifulSoup4（簡稱 BS4）是 Python 中最強大的 HTML 和 XML 文檔解析庫之一。

自動編碼處理：自動將輸入文檔轉(zhuǎn)換為 Unicode 編碼，輸出為 UTF-8，無需手動處理編碼問題
容錯性強：能夠優(yōu)雅處理格式錯誤或結(jié)構(gòu)混亂的 HTML 文檔
直觀的 API：提供類似 jQuery 的選擇器語法，學(xué)習(xí)成本低
多種解析器支持：支持 html.parser、lxml、html5lib 等多種解析器

1.2 BeautifulSoup4 的應(yīng)用場景

競品數(shù)據(jù)監(jiān)控：實時抓取競爭對手產(chǎn)品價格、庫存信息
市場情報收集：從新聞網(wǎng)站、行業(yè)報告中提取市場趨勢數(shù)據(jù)
社交媒體分析：抓取微博、知乎等平臺的用戶評論和互動數(shù)據(jù)
金融數(shù)據(jù)采集：獲取股票行情、財經(jīng)新聞等實時數(shù)據(jù)
電商數(shù)據(jù)分析：爬取商品信息、用戶評價進(jìn)行市場分析

第二章：環(huán)境配置與安裝

2.1 基礎(chǔ)安裝

# 安裝 BeautifulSoup4 核心庫
pip install beautifulsoup4
# 安裝推薦的解析器（lxml 性能最佳）
pip install lxml
# 安裝 html5lib（容錯性最強，選裝）
pip install html5lib
# 安裝網(wǎng)絡(luò)請求庫（配合使用）
pip install requests

2.2 驗證安裝

import bs4
import lxml
import html5lib
import requests

print(f"BeautifulSoup4 版本: {bs4.__version__}")
print(f"lxml 版本: {lxml.__version__}")
print(f"html5lib 版本: {html5lib.__version__}")
print(f"requests 版本: {requests.__version__}")

第三章：解析器選擇與性能對比

3.1 三種主流解析器詳解

解析器	安裝命令	優(yōu)點	缺點	適用場景
html.parser	無需安裝	內(nèi)置標(biāo)準(zhǔn)庫，無需額外依賴	速度一般，容錯性中等	簡單任務(wù)，快速原型開發(fā)
lxml	`pip install lxml`	速度最快，功能強大	需要 C 語言依賴	生產(chǎn)環(huán)境推薦，大規(guī)模數(shù)據(jù)處理
html5lib	`pip install html5lib`	容錯性最強，最接近瀏覽器解析	速度最慢	處理復(fù)雜、不規(guī)范的 HTML

3.2 性能測試對比

import time
from bs4 import BeautifulSoup
import requests

# 測試網(wǎng)頁
url = "https://baidu.com"
html_content = requests.get(url).text

# 性能測試函數(shù)
def test_parser(parser_name, iterations=100):
    start_time = time.time()
    for _ in range(iterations):
        soup = BeautifulSoup(html_content, parser_name)
    end_time = time.time()
    return (end_time - start_time) / iterations * 1000  # 毫秒

print("解析器性能對比（平均耗時/次）:")
print(f"html.parser: {test_parser('html.parser'):.2f}ms")
print(f"lxml: {test_parser('lxml'):.2f}ms")
print(f"html5lib: {test_parser('html5lib'):.2f}ms")

性能結(jié)論：

lxml 比 html.parser 快 1.5-3 倍
html5lib 比 lxml 慢 2-10 倍，但容錯性最佳
推薦：生產(chǎn)環(huán)境使用 lxml，開發(fā)調(diào)試使用 html5lib

第四章：基礎(chǔ)用法

4.1 創(chuàng)建 BeautifulSoup 對象

from bs4 import BeautifulSoup

# 方式1：從字符串創(chuàng)建
html_doc = """
<html>
    <head><title>測試頁面</title></head>
    <body>
        <h1>歡迎使用 BeautifulSoup4</h1>
        <p class="content">這是第一個段落</p>
        <p class="content">這是第二個段落</p>
    </body>
</html>
"""

# 使用 lxml 解析器（推薦）
soup = BeautifulSoup(html_doc, 'lxml')

# 方式2：從文件創(chuàng)建
with open('example.html', 'r', encoding='utf-8') as file:
    soup = BeautifulSoup(file, 'lxml')

# 方式3：從 URL 創(chuàng)建（配合 requests）
import requests
response = requests.get('https://example.com')
soup = BeautifulSoup(response.text, 'lxml')

4.2 四種核心對象類型

BeautifulSoup 將文檔轉(zhuǎn)換為樹形結(jié)構(gòu)，包含四種 Python 對象：

from bs4 import BeautifulSoup, Comment

html = """
<html>
    <head><title>測試</title></head>
    <body>
        <!-- 這是一個注釋 -->
        <p class="Class">Hello World</p>
    </body>
</html>
"""

soup = BeautifulSoup(html, 'lxml')

# 1. Tag - 標(biāo)簽對象
tag = soup.p
print(type(tag))  # <class 'bs4.element.Tag'>
print(tag.name)   # 'p'
print(tag['class'])  # 訪問指定屬性
print(tag.attrs) # 訪問所有屬性

# 2. NavigableString - 標(biāo)簽內(nèi)的文本
text = tag.string
print(type(text))  # <class 'bs4.element.NavigableString'>
print(text)        # 'Hello World'

# 3. BeautifulSoup - 整個文檔對象
print(type(soup))  # <class 'bs4.BeautifulSoup'>

# 4. Comment - 特殊的 NavigableString
comment = soup.find(string=lambda text: isinstance(text, Comment))
print(type(comment))  # <class 'bs4.element.Comment'>
print(comment)        # ' 這是一個注釋 '

4.3 遍歷文檔樹

# 獲取子節(jié)點
print(soup.body.contents)  # 返回列表
print(list(soup.body.children))  # 返回迭代器

# 獲取后代節(jié)點
for descendant in soup.body.descendants:
    print(descendant)

# 獲取父節(jié)點
print(soup.p.parent)  # body 標(biāo)簽
print(soup.p.parents)  # 所有祖先節(jié)點

# 獲取兄弟節(jié)點
print(soup.p.next_sibling)  # 下一個兄弟
print(soup.p.previous_sibling)  # 上一個兄弟

第五章：搜索文檔樹（核心）

5.1 find() 與 find_all() 方法

5.1.1 基礎(chǔ)用法

from bs4 import BeautifulSoup

html = """
<div class="container">
    <h1>標(biāo)題</h1>
    <p class="text" id="p1">段落1</p>
    <p class="text" id="p2">段落2</p>
    <a  rel="external nofollow"  rel="external nofollow" >鏈接</a>
    <span>其他內(nèi)容</span>
</div>
"""

soup = BeautifulSoup(html, 'lxml')

# find() - 返回第一個匹配的元素
first_p = soup.find('p')
print(first_p)  # <p class="text" id="p1">段落1</p>

# find_all() - 返回所有匹配的元素列表
all_p = soup.find_all('p')
print(all_p)  # [<p class="text" id="p1">段落1</p>, <p class="text" id="p2">段落2</p>]
print(len(all_p))  # 2

# 使用 limit 參數(shù)限制返回數(shù)量
limited_p = soup.find_all('p', limit=1)
print(len(limited_p))  # 1

5.1.2 按屬性搜索

# 按 class 搜索（注意：class 是 Python 關(guān)鍵字，需用 class_）
by_class = soup.find_all('p', class_='text')
print(len(by_class))  # 2

# 按 id 搜索
by_id = soup.find(id='p1')
print(by_id.text)  # '段落1'

# 按多個屬性組合搜索
by_attrs = soup.find_all('p', attrs={'class': 'text', 'id': 'p2'})
print(by_attrs[0].text)  # '段落2'

# 按屬性值模式搜索（正則表達(dá)式）
import re
by_pattern = soup.find_all(id=re.compile('^p'))
print(len(by_pattern))  # 2

5.1.3 高級搜索技巧

# 按文本內(nèi)容搜索
by_text = soup.find_all(string='段落1')
print(by_text)  # ['段落1']

# 按文本模式搜索
by_text_pattern = soup.find_all(string=re.compile('段落'))
print(len(by_text_pattern))  # 2

# 使用函數(shù)作為過濾器
def has_class_but_no_id(tag):
    return tag.has_attr('class') and not tag.has_attr('id')

result = soup.find_all(has_class_but_no_id)
print(result)  # [<p class="text" id="p1">...</p>, <p class="text" id="p2">...</p>]

# 遞歸搜索控制
non_recursive = soup.find_all('p', recursive=False)
print(len(non_recursive))  # 0（因為 p 不是 soup 的直接子節(jié)點）

5.2 CSS 選擇器（select() 方法）

# 基礎(chǔ)選擇器
soup.select('p')              # 所有 p 標(biāo)簽
soup.select('.text')          # class="text" 的元素
soup.select('#p1')            # id="p1" 的元素
soup.select('div p')          # div 下的所有 p 標(biāo)簽
soup.select('div > p')        # div 的直接子 p 標(biāo)簽

# 組合選擇器
soup.select('p.text')         # class="text" 的 p 標(biāo)簽
soup.select('p#p1')           # id="p1" 的 p 標(biāo)簽
soup.select('p, span')        # p 或 span 標(biāo)簽

# 屬性選擇器
soup.select('[href]')         # 有 href 屬性的元素
soup.select('[ rel="external nofollow"  rel="external nofollow" ]')  # href 精確匹配
soup.select('[class~="text"]')  # class 包含 text

# 偽類選擇器
soup.select('p:first-child')  # 第一個 p 標(biāo)簽
soup.select('p:last-child')   # 最后一個 p 標(biāo)簽
soup.select('p:nth-of-type(2)')  # 第二個 p 標(biāo)簽

5.3 find() vs find_all() vs select() 對比

方法	返回值	語法	適用場景
find()	單個 Tag 對象	Python 字典語法	只需要第一個匹配項
find_all()	Tag 對象列表	Python 字典語法	需要所有匹配項，參數(shù)靈活
select()	Tag 對象列表	CSS 選擇器語法	熟悉 CSS，層級關(guān)系復(fù)雜

第六章：數(shù)據(jù)清洗和處理

6.1 提取文本內(nèi)容

# 基礎(chǔ)文本提取
print(soup.p.string)        # '段落1'
print(soup.p.get_text())    # '段落1'

# 提取所有文本（包括子標(biāo)簽）
print(soup.get_text())      # '標(biāo)題段落1段落2鏈接其他內(nèi)容'

# 提取時去除空白
print(soup.get_text(strip=True))  # '標(biāo)題段落1段落2鏈接其他內(nèi)容'

# 提取多個元素的文本
texts = [p.get_text() for p in soup.find_all('p')]
print(texts)  # ['段落1', '段落2']

6.2 提取屬性值

# 獲取單個屬性
link = soup.find('a')
print(link['href'])          # 'https://example.com'
print(link.get('href'))      # 'https://example.com'

# 獲取所有屬性
print(link.attrs)            # {'href': 'https://example.com'}

# 安全獲取（屬性不存在時返回 None）
print(link.get('target'))    # None
print(link.get('target', '_blank'))  # '_blank'（默認(rèn)值）

# 獲取所有鏈接
links = soup.find_all('a')
hrefs = [link.get('href') for link in links]
print(hrefs)

6.3 處理特殊內(nèi)容

# 處理 HTML 實體
html_with_entities = "<p>Price: &yen;100</p>"
soup = BeautifulSoup(html_with_entities, 'lxml')
print(soup.p.string)  # 'Price: ￥100'（自動解碼）

# 處理注釋
html_with_comment = "<p><!-- 這是注釋 -->正文</p>"
soup = BeautifulSoup(html_with_comment, 'lxml')
comment = soup.find(string=lambda text: isinstance(text, Comment))
print(comment)  # ' 這是注釋 '

# 提取腳本和樣式內(nèi)容
script_content = soup.find('script').string
style_content = soup.find('style').string

第七章：修改文檔樹

7.1 修改標(biāo)簽和屬性

html = "<p class='old'>原始內(nèi)容</p>"
soup = BeautifulSoup(html, 'lxml')

# 修改標(biāo)簽名稱
soup.p.name = 'div'
print(soup)  # <div class="old">原始內(nèi)容</div>

# 修改屬性
soup.div['class'] = 'new'
soup.div['id'] = 'modified'
print(soup.div)  # <div class="new" id="modified">原始內(nèi)容</div>

# 刪除屬性
del soup.div['class']
print(soup.div)  # <div id="modified">原始內(nèi)容</div>

7.2 修改文本內(nèi)容

# 修改文本
soup.div.string = '新內(nèi)容'
print(soup.div)  # <div id="modified">新內(nèi)容</div>

# 替換為新標(biāo)簽
new_tag = soup.new_tag('span', class_='highlight')
new_tag.string = '高亮內(nèi)容'
soup.div.string.replace_with(new_tag)
print(soup.div)  # <div id="modified"><span class="highlight">高亮內(nèi)容</span></div>

7.3 添加和刪除元素

# append() - 添加到末尾
soup.div.append('追加的文本')

# insert() - 指定位置插入
new_p = soup.new_tag('p')
new_p.string = '插入的段落'
soup.div.insert(0, new_p)  # 插入到第一個位置

# insert_before() / insert_after()
soup.div.insert_before(soup.new_tag('hr'))
soup.div.insert_after(soup.new_tag('hr'))

# clear() - 清空內(nèi)容
soup.div.clear()

# extract() - 移除并返回
removed = soup.div.extract()

# decompose() - 移除且不返回
soup.div.decompose()

# unwrap() - 移除標(biāo)簽但保留內(nèi)容
soup.span.unwrap()

第八章：實戰(zhàn)案例

每個網(wǎng)站的設(shè)計各有差異，這里只提供模式思路，具體還需通過實際場景修改

8.1 案例1：電商商品信息爬取

import requests
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime
import time

class ECommerceScraper:
    def __init__(self, base_url):
        self.base_url = base_url
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        self.products = []
    
    def fetch_page(self, url):
        """獲取頁面內(nèi)容"""
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            print(f"請求失敗: {e}")
            return None
    
    def parse_product(self, product_div):
        """解析單個商品信息"""
        try:
            # 提取商品名稱
            name_tag = product_div.find('h2', class_='product-name')
            name = name_tag.get_text(strip=True) if name_tag else 'N/A'
            
            # 提取價格
            price_tag = product_div.find('span', class_='price')
            price = price_tag.get_text(strip=True) if price_tag else 'N/A'
            
            # 提取評分
            rating_tag = product_div.find('div', class_='rating')
            rating = rating_tag.get('data-score', 'N/A') if rating_tag else 'N/A'
            
            # 提取評論數(shù)
            review_tag = product_div.find('span', class_='review-count')
            reviews = review_tag.get_text(strip=True) if review_tag else '0'
            
            # 提取商品鏈接
            link_tag = product_div.find('a', class_='product-link')
            link = link_tag.get('href', '') if link_tag else ''
            if link and not link.startswith('http'):
                link = self.base_url + link
            
            return {
                'name': name,
                'price': price,
                'rating': rating,
                'reviews': reviews,
                'link': link,
                'scraped_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
        except Exception as e:
            print(f"解析商品失敗: {e}")
            return None
    
    def scrape_category(self, category_url, max_pages=5):
        """爬取分類頁面"""
        for page in range(1, max_pages + 1):
            url = f"{category_url}?page={page}"
            print(f"正在爬取: {url}")
            
            html = self.fetch_page(url)
            if not html:
                break
            
            soup = BeautifulSoup(html, 'lxml')
            
            # 查找所有商品容器
            product_divs = soup.find_all('div', class_='product-item')
            
            if not product_divs:
                print("未找到商品，可能已到達(dá)最后一頁")
                break
            
            for product_div in product_divs:
                product = self.parse_product(product_div)
                if product:
                    self.products.append(product)
            
            print(f"第 {page} 頁: 找到 {len(product_divs)} 個商品")
            time.sleep(1)  # 避免請求過快
    
    def save_to_excel(self, filename='products.xlsx'):
        """保存到 Excel"""
        df = pd.DataFrame(self.products)
        df.to_excel(filename, index=False)
        print(f"數(shù)據(jù)已保存到 {filename}")
        print(f"共爬取 {len(self.products)} 個商品")
    
    def analyze_data(self):
        """數(shù)據(jù)分析"""
        if not self.products:
            print("沒有數(shù)據(jù)可分析")
            return
        
        df = pd.DataFrame(self.products)
        
        # 價格分析
        df['price_clean'] = df['price'].str.replace(r'[^\d.]', '', regex=True).astype(float)
        print(f"\n價格統(tǒng)計:")
        print(f"平均價格: ￥{df['price_clean'].mean():.2f}")
        print(f"最高價格: ￥{df['price_clean'].max():.2f}")
        print(f"最低價格: ￥{df['price_clean'].min():.2f}")
        
        # 評分分析
        df['rating'] = pd.to_numeric(df['rating'], errors='coerce')
        print(f"\n評分統(tǒng)計:")
        print(f"平均評分: {df['rating'].mean():.2f}")
        print(f"好評率(4.0+): {(df['rating'] >= 4.0).sum() / len(df) * 100:.1f}%")

# 使用示例
if __name__ == "__main__":
    scraper = ECommerceScraper('https://example-ecommerce.com')
    scraper.scrape_category('https://example-ecommerce.com/category/electronics', max_pages=3)
    scraper.save_to_excel()
    scraper.analyze_data()

8.2 案例2：新聞網(wǎng)站內(nèi)容監(jiān)控

import requests
from bs4 import BeautifulSoup
import sqlite3
from datetime import datetime
import hashlib

class NewsMonitor:
    def __init__(self, db_path='news.db'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """初始化數(shù)據(jù)庫"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS news (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                url TEXT UNIQUE NOT NULL,
                summary TEXT,
                publish_time TEXT,
                source TEXT,
                content_hash TEXT,
                scraped_at TEXT,
                is_new INTEGER DEFAULT 1
            )
        ''')
        conn.commit()
        conn.close()
    
    def get_article_hash(self, content):
        """生成內(nèi)容哈希（用于去重）"""
        return hashlib.md5(content.encode('utf-8')).hexdigest()
    
    def is_article_exists(self, content_hash):
        """檢查文章是否已存在"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT id FROM news WHERE content_hash = ?', (content_hash,))
        result = cursor.fetchone()
        conn.close()
        return result is not None
    
    def save_article(self, article):
        """保存文章到數(shù)據(jù)庫"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
                INSERT OR IGNORE INTO news 
                (title, url, summary, publish_time, source, content_hash, scraped_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                article['title'],
                article['url'],
                article['summary'],
                article['publish_time'],
                article['source'],
                article['content_hash'],
                article['scraped_at']
            ))
            conn.commit()
            return cursor.rowcount > 0
        except sqlite3.IntegrityError:
            return False
        finally:
            conn.close()
    
    def scrape_news(self, url, source_name):
        """爬取新聞列表"""
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.encoding = 'utf-8'
            soup = BeautifulSoup(response.text, 'lxml')
            
            # 查找新聞列表（根據(jù)實際網(wǎng)站結(jié)構(gòu)調(diào)整）
            news_items = soup.find_all('div', class_='news-item')
            
            new_articles = 0
            for item in news_items:
                try:
                    # 提取標(biāo)題
                    title_tag = item.find('h3', class_='news-title')
                    title = title_tag.get_text(strip=True) if title_tag else ''
                    
                    # 提取鏈接
                    link_tag = item.find('a')
                    article_url = link_tag.get('href', '') if link_tag else ''
                    if article_url and not article_url.startswith('http'):
                        article_url = url + article_url
                    
                    # 提取摘要
                    summary_tag = item.find('p', class_='summary')
                    summary = summary_tag.get_text(strip=True) if summary_tag else ''
                    
                    # 提取發(fā)布時間
                    time_tag = item.find('span', class_='publish-time')
                    publish_time = time_tag.get_text(strip=True) if time_tag else ''
                    
                    # 生成內(nèi)容哈希
                    content_hash = self.get_article_hash(title + summary)
                    
                    # 檢查是否已存在
                    if self.is_article_exists(content_hash):
                        continue
                    
                    # 保存文章
                    article = {
                        'title': title,
                        'url': article_url,
                        'summary': summary,
                        'publish_time': publish_time,
                        'source': source_name,
                        'content_hash': content_hash,
                        'scraped_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    }
                    
                    if self.save_article(article):
                        new_articles += 1
                        print(f"? {title}")
                
                except Exception as e:
                    print(f"解析文章失敗: {e}")
                    continue
            
            print(f"\n本次爬取: {len(news_items)} 篇, 新增: {new_articles} 篇")
            return new_articles
        
        except requests.RequestException as e:
            print(f"請求失敗: {e}")
            return 0
    
    def get_latest_news(self, limit=10):
        """獲取最新新聞"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            SELECT title, url, publish_time, source 
            FROM news 
            ORDER BY scraped_at DESC 
            LIMIT ?
        ''', (limit,))
        results = cursor.fetchall()
        conn.close()
        return results

# 使用示例
if __name__ == "__main__":
    monitor = NewsMonitor()
    
    # 監(jiān)控多個新聞源
    news_sources = [
        ('https://news.example.com/politics', '政治新聞'),
        ('https://news.example.com/tech', '科技新聞'),
        ('https://news.example.com/business', '商業(yè)新聞')
    ]
    
    for url, source in news_sources:
        print(f"\n爬取 {source}...")
        monitor.scrape_news(url, source)
    
    # 查看最新新聞
    print("\n=== 最新新聞 ===")
    latest = monitor.get_latest_news(5)
    for title, url, time, source in latest:
        print(f"[{source}] {time} - {title}")

8.3 案例3：股票數(shù)據(jù)實時監(jiān)控

import requests
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime
import time
import matplotlib.pyplot as plt

class StockMonitor:
    def __init__(self):
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        self.price_history = {}
    
    def get_stock_price(self, stock_code):
        """獲取股票價格"""
        url = f"https://finance.example.com/quote/{stock_code}"
        
        try:
            response = requests.get(url, headers=self.headers, timeout=5)
            soup = BeautifulSoup(response.text, 'lxml')
            
            # 根據(jù)實際網(wǎng)站結(jié)構(gòu)調(diào)整選擇器
            price_tag = soup.select_one('.stock-price .current')
            price = float(price_tag.get_text(strip=True).replace(',', '')) if price_tag else None
            
            change_tag = soup.select_one('.stock-price .change')
            change = change_tag.get_text(strip=True) if change_tag else '0.00'
            
            return {
                'code': stock_code,
                'price': price,
                'change': change,
                'time': datetime.now().strftime('%H:%M:%S')
            }
        except Exception as e:
            print(f"獲取 {stock_code} 價格失敗: {e}")
            return None
    
    def monitor_stocks(self, stock_codes, duration_minutes=5, interval_seconds=10):
        """監(jiān)控多只股票"""
        end_time = time.time() + duration_minutes * 60
        
        print(f"開始監(jiān)控 {len(stock_codes)} 只股票，持續(xù) {duration_minutes} 分鐘...")
        
        while time.time() < end_time:
            timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(f"\n[{timestamp}]")
            
            for code in stock_codes:
                data = self.get_stock_price(code)
                if data and data['price']:
                    print(f"[code]: ￥{data['price']:.2f} ({data['change']})")
                    
                    # 記錄歷史數(shù)據(jù)
                    if code not in self.price_history:
                        self.price_history[code] = []
                    self.price_history[code].append({
                        'time': data['time'],
                        'price': data['price']
                    })
            
            time.sleep(interval_seconds)
    
    def plot_price_trend(self):
        """繪制價格趨勢圖"""
        if not self.price_history:
            print("沒有數(shù)據(jù)可繪制")
            return
        
        plt.figure(figsize=(12, 6))
        
        for code, history in self.price_history.items():
            times = [h['time'] for h in history]
            prices = [h['price'] for h in history]
            plt.plot(times, prices, marker='o', label=code)
        
        plt.xlabel('Time')
        plt.ylabel('Price (￥)')
        plt.title('Stock Price Trend')
        plt.legend()
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.grid(True, alpha=0.3)
        plt.savefig('stock_trend.png')
        print("價格趨勢圖已保存為 stock_trend.png")
    
    def generate_report(self):
        """生成監(jiān)控報告"""
        report = []
        report.append("=" * 50)
        report.append("STOCK MONITORING REPORT")
        report.append("=" * 50)
        report.append(f"Report Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"Total Stocks Monitored: {len(self.price_history)}")
        report.append("")
        
        for code, history in self.price_history.items():
            if len(history) < 2:
                continue
            
            first_price = history[0]['price']
            last_price = history[-1]['price']
            change = last_price - first_price
            change_percent = (change / first_price) * 100
            
            report.append(f"Stock: [code]")
            report.append(f"  Start Price: ￥{first_price:.2f}")
            report.append(f"  End Price: ￥{last_price:.2f}")
            report.append(f"  Change: ￥{change:+.2f} ({change_percent:+.2f}%)")
            report.append(f"  High: ￥{max(h['price'] for h in history):.2f}")
            report.append(f"  Low: ￥{min(h['price'] for h in history):.2f}")
            report.append("")
        
        report_text = "\n".join(report)
        with open('stock_report.txt', 'w', encoding='utf-8') as f:
            f.write(report_text)
        
        print("監(jiān)控報告已保存為 stock_report.txt")
        print(report_text)

# 使用示例
if __name__ == "__main__":
    monitor = StockMonitor()
    
    # 監(jiān)控股票列表
    stocks = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN']
    
    # 開始監(jiān)控（監(jiān)控5分鐘，每10秒刷新一次）
    monitor.monitor_stocks(stocks, duration_minutes=5, interval_seconds=10)
    
    # 生成可視化圖表
    monitor.plot_price_trend()
    
    # 生成文本報告
    monitor.generate_report()

第九章：優(yōu)化技巧

9.1 代碼優(yōu)化技巧

from bs4 import BeautifulSoup
import time

# 反面案例：低效的多次查找
def bad_practice(html):
    soup = BeautifulSoup(html, 'lxml')
    
    # 每次都重新遍歷整個文檔樹
    for i in range(100):
        title = soup.find('h1')
        print(title.text)

# 正面案例：緩存查找結(jié)果
def good_practice(html):
    soup = BeautifulSoup(html, 'lxml')
    
    # 只查找一次，緩存結(jié)果
    title = soup.find('h1')
    title_text = title.text if title else ''
    
    for i in range(100):
        print(title_text)

# 性能對比
html = "<html><h1>Test</h1></html>"
start = time.time()
bad_practice(html)
print(f"反面案例耗時: {time.time() - start:.4f}s")

start = time.time()
good_practice(html)
print(f"正面案例耗時: {time.time() - start:.4f}s")

9.2 內(nèi)存優(yōu)化策略

from bs4 import BeautifulSoup
import gc

def process_large_html(html_chunks):
    """處理大文件的內(nèi)存優(yōu)化方法"""
    results = []
    
    for i, chunk in enumerate(html_chunks):
        # 創(chuàng)建局部作用域
        soup = BeautifulSoup(chunk, 'lxml')
        
        # 提取需要的數(shù)據(jù)
        data = extract_data(soup)
        results.append(data)
        
        # 顯式刪除大對象
        del soup
        
        # 定期觸發(fā)垃圾回收
        if i % 10 == 0:
            gc.collect()
    
    return results

def extract_data(soup):
    """提取數(shù)據(jù)的函數(shù)"""
    return {
        'title': soup.find('h1').text if soup.find('h1') else '',
        'content': soup.find('div', class_='content').text if soup.find('div', class_='content') else ''
    }

9.3 錯誤處理與重試機制

import requests
from bs4 import BeautifulSoup
import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=2):
    """重試裝飾器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt < max_retries - 1:
                        print(f"嘗試 {attempt + 1}/{max_retries} 失敗: {e}")
                        time.sleep(delay * (attempt + 1))  # 指數(shù)退避
                    else:
                        print(f"所有 {max_retries} 次嘗試均失敗")
                        raise
        return wrapper
    return decorator

class RobustScraper:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
    
    @retry_on_failure(max_retries=3, delay=2)
    def fetch_page(self, url):
        """帶重試的頁面獲取"""
        response = self.session.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    
    def parse_with_fallback(self, html, parsers=None):
        """帶降級策略的解析"""
        if parsers is None:
            parsers = ['lxml', 'html.parser', 'html5lib']
        
        for parser in parsers:
            try:
                soup = BeautifulSoup(html, parser)
                # 驗證解析是否成功
                if soup.find('body'):
                    print(f"? 使用 {parser} 解析成功")
                    return soup
            except Exception as e:
                print(f"? {parser} 解析失敗: {e}")
        
        raise ValueError("所有解析器均失敗")

9.4 反爬蟲策略應(yīng)對

import requests
from bs4 import BeautifulSoup
import random
import time

class AntiAntiCrawler:
    def __init__(self):
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
        ]
        
        self.proxies = [
            # 可配置代理池
        ]
    
    def get_random_headers(self):
        """隨機化請求頭"""
        return {
            'User-Agent': random.choice(self.user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        }
    
    def scrape_with_delay(self, urls, min_delay=1, max_delay=3):
        """帶隨機延遲的爬取"""
        results = []
        
        for url in urls:
            headers = self.get_random_headers()
            
            try:
                response = requests.get(url, headers=headers, timeout=10)
                soup = BeautifulSoup(response.text, 'lxml')
                results.append(soup)
                
                # 隨機延遲，模擬人類行為
                delay = random.uniform(min_delay, max_delay)
                time.sleep(delay)
                
            except Exception as e:
                print(f"爬取 {url} 失敗: {e}")
                continue
        
        return results

第十章：常見問題與解決方案

10.1 編碼問題

# 問題：中文亂碼
# 解決方案：
response = requests.get(url)
response.encoding = 'utf-8'  # 顯式指定編碼
soup = BeautifulSoup(response.text, 'lxml')

# 或者自動檢測編碼
from charset_normalizer import detect
encoding = detect(response.content)['encoding']
response.encoding = encoding

10.2 動態(tài)內(nèi)容加載

# 問題：JavaScript 渲染的內(nèi)容無法獲取
# 解決方案1：使用 Selenium
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
driver.get(url)
time.sleep(3)  # 等待頁面加載
html = driver.page_source
soup = BeautifulSoup(html, 'lxml')
driver.quit()

# 解決方案2：查找 API 接口
# 很多網(wǎng)站的數(shù)據(jù)來自后端 API，直接調(diào)用 API 更高效
api_url = "https://example.com/api/data"
response = requests.get(api_url, headers=headers)
data = response.json()  # 直接獲取 JSON 數(shù)據(jù)

10.3 復(fù)雜選擇器

# 問題：層級太深，選擇器復(fù)雜
# 解決方案：分步查找

# 不推薦：超長選擇器
result = soup.select('div.container > div.row > div.col-md-8 > ul.list > li.item > a.link')

# 推薦：分步查找
container = soup.find('div', class_='container')
if container:
    row = container.find('div', class_='row')
    if row:
        col = row.find('div', class_='col-md-8')
        if col:
            items = col.find_all('li', class_='item')
            for item in items:
                link = item.find('a', class_='link')
                if link:
                    print(link.get('href'))

10.4 性能瓶頸

# 問題：處理大量數(shù)據(jù)時速度慢
# 解決方案：

# 1. 使用更快的解析器
soup = BeautifulSoup(html, 'lxml')  # 而不是 'html.parser'

# 2. 限制搜索范圍
# 錯誤示例：搜索整個文檔
all_divs = soup.find_all('div')

# 正確示例：先定位容器，再在容器內(nèi)搜索
container = soup.find('div', id='main-content')
if container:
    all_divs = container.find_all('div')

# 3. 使用 CSS 選擇器（通常比 find_all 快）
items = soup.select('div.item')  # 比 soup.find_all('div', class_='item') 快

# 4. 避免在循環(huán)中重復(fù)查找
# ?
for i in range(1000):
    title = soup.find('h1').text

# ?
title = soup.find('h1').text
for i in range(1000):
    use_title(title)

第十一章：完整項目模板

"""
企業(yè)級數(shù)據(jù)爬蟲框架
功能：
- 配置化管理
- 自動重試
- 數(shù)據(jù)驗證
- 日志記錄
- 異常處理
- 數(shù)據(jù)導(dǎo)出
"""

import requests
from bs4 import BeautifulSoup
import pandas as pd
import logging
from datetime import datetime
import json
import os
from typing import List, Dict, Optional
import time

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('scraper.log'),
        logging.StreamHandler()
    ]
)

class EnterpriseScraper:
    """企業(yè)級爬蟲基類"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.base_url = config.get('base_url', '')
        self.headers = config.get('headers', {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
        self.timeout = config.get('timeout', 10)
        self.max_retries = config.get('max_retries', 3)
        self.delay = config.get('delay', 1)
        self.session = requests.Session()
        self.session.headers.update(self.headers)
        self.data = []
    
    def fetch(self, url: str, params: Optional[Dict] = None) -> Optional[str]:
        """獲取頁面內(nèi)容（帶重試）"""
        for attempt in range(self.max_retries):
            try:
                logging.info(f"請求: {url}")
                response = self.session.get(url, params=params, timeout=self.timeout)
                response.raise_for_status()
                
                # 檢查響應(yīng)內(nèi)容
                if not response.text or len(response.text) < 100:
                    logging.warning(f"響應(yīng)內(nèi)容過短: {len(response.text)}")
                    raise ValueError("響應(yīng)內(nèi)容異常")
                
                return response.text
                
            except Exception as e:
                logging.error(f"請求失敗 (嘗試 {attempt + 1}/{self.max_retries}): {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(self.delay * (attempt + 1))
                else:
                    logging.error(f"所有重試均失敗: {url}")
                    return None
        
        return None
    
    def parse(self, html: str) -> List[Dict]:
        """解析頁面（子類實現(xiàn)）"""
        raise NotImplementedError("子類必須實現(xiàn) parse 方法")
    
    def validate(self, item: Dict) -> bool:
        """驗證數(shù)據(jù)（子類實現(xiàn)）"""
        return True
    
    def scrape(self, urls: List[str]) -> List[Dict]:
        """爬取多個 URL"""
        all_data = []
        
        for i, url in enumerate(urls, 1):
            logging.info(f"處理 {i}/{len(urls)}: {url}")
            
            html = self.fetch(url)
            if not html:
                continue
            
            try:
                items = self.parse(html)
                logging.info(f"解析得到 {len(items)} 條數(shù)據(jù)")
                
                # 驗證和過濾
                valid_items = [item for item in items if self.validate(item)]
                logging.info(f"驗證通過 {len(valid_items)} 條數(shù)據(jù)")
                
                all_data.extend(valid_items)
                
                # 延遲避免被封
                if i < len(urls):
                    time.sleep(self.delay)
                    
            except Exception as e:
                logging.error(f"解析失敗: {e}")
                continue
        
        self.data = all_data
        return all_data
    
    def save_to_csv(self, filename: str = 'output.csv'):
        """保存為 CSV"""
        if not self.data:
            logging.warning("沒有數(shù)據(jù)可保存")
            return
        
        df = pd.DataFrame(self.data)
        df.to_csv(filename, index=False, encoding='utf-8-sig')
        logging.info(f"數(shù)據(jù)已保存到 {filename} ({len(self.data)} 條)")
    
    def save_to_excel(self, filename: str = 'output.xlsx'):
        """保存為 Excel"""
        if not self.data:
            logging.warning("沒有數(shù)據(jù)可保存")
            return
        
        df = pd.DataFrame(self.data)
        df.to_excel(filename, index=False)
        logging.info(f"數(shù)據(jù)已保存到 {filename} ({len(self.data)} 條)")
    
    def save_to_json(self, filename: str = 'output.json'):
        """保存為 JSON"""
        if not self.data:
            logging.warning("沒有數(shù)據(jù)可保存")
            return
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)
        logging.info(f"數(shù)據(jù)已保存到 {filename} ({len(self.data)} 條)")
    
    def generate_report(self):
        """生成統(tǒng)計報告"""
        if not self.data:
            logging.warning("沒有數(shù)據(jù)可分析")
            return
        
        df = pd.DataFrame(self.data)
        
        report = []
        report.append("=" * 60)
        report.append("SCRAPING REPORT")
        report.append("=" * 60)
        report.append(f"Report Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"Total Records: {len(self.data)}")
        report.append(f"Fields: {', '.join(df.columns)}")
        report.append("")
        
        # 統(tǒng)計信息
        for col in df.columns:
            report.append(f"Field: {col}")
            report.append(f"  Non-null: {df[col].notnull().sum()}/{len(df)}")
            report.append(f"  Unique: {df[col].nunique()}")
            
            # 數(shù)值字段統(tǒng)計
            if df[col].dtype in ['int64', 'float64']:
                report.append(f"  Mean: {df[col].mean():.2f}")
                report.append(f"  Min: {df[col].min():.2f}")
                report.append(f"  Max: {df[col].max():.2f}")
            
            report.append("")
        
        report_text = "\n".join(report)
        logging.info(report_text)
        
        # 保存報告
        with open('scraping_report.txt', 'w', encoding='utf-8') as f:
            f.write(report_text)
        
        logging.info("報告已保存到 scraping_report.txt")

# 使用示例
if __name__ == "__main__":
    # 配置
    config = {
        'base_url': 'https://example.com',
        'headers': {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9',
        },
        'timeout': 10,
        'max_retries': 3,
        'delay': 2
    }
    
    # 創(chuàng)建爬蟲實例（需要繼承并實現(xiàn) parse 和 validate 方法）
    # scraper = MyCustomScraper(config)
    # urls = ['https://example.com/page1', 'https://example.com/page2']
    # scraper.scrape(urls)
    # scraper.save_to_excel()
    # scraper.generate_report()
    
    logging.info("框架加載成功，請繼承 EnterpriseScraper 類實現(xiàn)具體爬蟲")

總結(jié)

核心要點回顧

解析器選擇：生產(chǎn)環(huán)境優(yōu)先使用 lxml，開發(fā)調(diào)試使用 html5lib
搜索方法：find/find_all 適合復(fù)雜條件，select 適合層級關(guān)系
性能優(yōu)化：緩存查找結(jié)果、限制搜索范圍、使用合適的選擇器
錯誤處理：實現(xiàn)重試機制、降級策略、完善的日志記錄
反爬應(yīng)對：隨機化請求頭、添加延遲、使用代理池

進(jìn)階學(xué)習(xí)路徑

深入理解 HTML/CSS：掌握更復(fù)雜的選擇器語法
學(xué)習(xí) XPath：與 lxml 配合使用，處理更復(fù)雜的文檔結(jié)構(gòu)
掌握 Selenium：處理 JavaScript 動態(tài)渲染的頁面
學(xué)習(xí) Scrapy 框架：構(gòu)建更強大的爬蟲系統(tǒng)
數(shù)據(jù)存儲優(yōu)化：學(xué)習(xí) MongoDB、Elasticsearch 等數(shù)據(jù)庫
分布式爬蟲：使用 Celery、Redis 構(gòu)建分布式系統(tǒng)

法律與道德提醒

遵守網(wǎng)站的 robots.txt 協(xié)議
尊重網(wǎng)站的使用條款
合理控制請求頻率，避免對服務(wù)器造成壓力
僅用于合法的數(shù)據(jù)分析目的
注意數(shù)據(jù)隱私和版權(quán)問題

以上就是一文詳解Python HTML/XML解析庫BeautifulSoup4的詳細(xì)內(nèi)容，更多關(guān)于Python HTML/XML解析庫BeautifulSoup4詳解的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线