首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 編程語言 >  >> 更多程式語言 >> 更多程式設計知識

Python 爬蟲進階:抓取百度搜尋結果並提取連結完整實作

     編輯:更多程式設計知識

搜尋引擎結果頁面(SERP)抓取涉及反爬蟲應對、動態參數、分頁處理、結果解析等進階技巧。本文完整實作百度搜尋抓取:User-Agent 旋轉、分頁參數、正則提取標題連結、反爬應對、結果去重與儲存。注意:請遵守 robots.txt 與服務條款,控制請求頻率,僅供技術學習參考。

核心架構設計


import urllib.request
import urllib.parse
import urllib.error
import re
import random
import time
from typing import List, Tuple, Optional

class BaiduScraper:
    def __init__(self, delay_range=(1, 3), timeout=10):
        self.base_url = "https://www.baidu.com/s"
        self.delay_range = delay_range
        self.timeout = timeout
        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',
            # 更多現代 UA...
        ]
    
    def _build_request(self, params: dict) -> urllib.request.Request:
        url = self.base_url + "?" + urllib.parse.urlencode(params)
        headers = {
            'User-Agent': random.choice(self.user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-TW,zh;q=0.9,en;q=0.8',
            'Connection': 'keep-alive',
        }
        return urllib.request.Request(url, headers=headers)
    
    def _fetch(self, request: urllib.request.Request) -> Optional[str]:
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as resp:
                raw = resp.read()
                # 編碼偵測
                encoding = resp.headers.get_content_charset() or 'utf-8'
                return raw.decode(encoding, errors='ignore')
        except urllib.error.HTTPError as e:
            if e.code == 429:
                print("觸發頻率限制,需增加延遲或更換 IP")
            return None
        except Exception as e:
            print(f"請求失敗: {e}")
            return None
    
    def _parse_results(self, html: str) -> List[Tuple[str, str]]:
        """解析搜尋結果,回傳 [(標題, 連結), ...]"""
        results = []
        # 百度結果容器正則(需根據實際 HTML 調整)
        # 這裡使用簡化版,實際需分析當前頁面結構
        pattern = re.compile(
            r']*class="t"[^>]*>\s*]*href="([^"]*)"[^>]*>(.*?)',
            re.DOTALL | re.IGNORECASE
        )
        for match in pattern.finditer(html):
            url = match.group(1)
            title = re.sub(r'<[^>]+>', '', match.group(2)).strip()
            if url and title:
                results.append((title, url))
        return results
    
    def search(self, keyword: str, max_pages: int = 5) -> List[Tuple[str, str]]:
        all_results = []
        seen_urls = set()
        
        for page in range(max_pages):
            pn = page * 10  # 百度分頁參數
            params = {
                'wd': keyword,
                'pn': pn,
                'rn': 10,   # 每頁結果數
            }
            
            print(f"抓取第 {page+1} 頁: pn={pn}")
            request = self._build_request(params)
            html = self._fetch(request)
            
            if not html:
                print("獲取失敗,停止")
                break
            
            results = self._parse_results(html)
            if not results:
                print("無結果,可能被反爬或結構變更")
                break
            
            new_count = 0
            for title, url in results:
                # 去重
                if url not in seen_urls:
                    seen_urls.add(url)
                    all_results.append((title, url))
                    new_count += 1
            
            print(f"  新增 {new_count} 筆結果")
            if new_count == 0:
                print("  無新結果,結束")
                break
            
            # 禮貌延遲
            time.sleep(random.uniform(*self.delay_range))
        
        return all_results

# 使用範例
if __name__ == "__main__":
    scraper = BaiduScraper(delay_range=(2, 5))
    results = scraper.search("Python 教學", max_pages=3)
    for i, (title, url) in enumerate(results, 1):
        print(f"{i}. {title} -> {url}")

關鍵技術細節解析

1. 分頁參數

  • pn:起始結果索引(0, 10, 20...)
  • rn:每頁結果數(預設 10,可設 50)
  • wd:搜尋關鍵字(URL 編碼)

2. User-Agent 旋轉策略


# 維護 UA 池,定期更新
USER_AGENTS = [
    # Chrome Windows
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    # Chrome Mac
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    # Firefox
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
    # Safari
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15',
    # Edge
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
]
# 隨機選擇
import random
headers = {'User-Agent': random.choice(USER_AGENTS)}

3. 反爬蟲應對

  • IP 限制:使用代理 IP 池、降低頻率、分散時間
  • 驗證碼:偵測驗證碼頁面,人工介入或使用打碼服務
  • JS 渲染:百度大量使用 JS,需 selenium/playwright 或分析 API
  • Cookie 驗證:維持 Session、處理登入狀態

4. 結果解析穩健性


# 使用 BeautifulSoup 較正則穩健
from bs4 import BeautifulSoup

def parse_with_bs4(html: str):
    soup = BeautifulSoup(html, 'html.parser')
    results = []
    # 百度結果選擇器(需根據實際 DOM 調整)
    for item in soup.select('.result .t a'):
        title = item.get_text(strip=True)
        url = item.get('href')
        if title and url:
            results.append((title, url))
    return results

進階:即時解析真實連結(重導向追蹤)


def resolve_real_url(baidu_url: str, headers: dict) -> str:
    """追蹤百度跳轉連結到真實目標"""
    try:
        req = urllib.request.Request(baidu_url, headers=headers)
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.geturl()  # 最終真實 URL
    except Exception:
        return baidu_url  # 失敗回傳原連結

# 批次解析(可併發加速)
from concurrent.futures import ThreadPoolExecutor

def resolve_batch(urls: List[str], max_workers=5):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(resolve_real_url, url, headers): url for url in urls}
        for future in concurrent.futures.as_completed(futures):
            url = futures[future]
            try:
                real = future.result()
                print(f"{url} -> {real}")
            except Exception as e:
                print(f"{url} 解析失敗: {e}")

資料儲存與後續處理


import json
import csv
from datetime import datetime

def save_results(results: List[Tuple[str, str]], keyword: str):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    base = f"baidu_{keyword}_{timestamp}"
    
    # JSON
    with open(f"{base}.json", 'w', encoding='utf-8') as f:
        json.dump([
            {"rank": i, "title": t, "url": u}
            for i, (t, u) in enumerate(results, 1)
        ], f, ensure_ascii=False, indent=2)
    
    # CSV
    with open(f"{base}.csv", 'w', encoding='utf-8', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['Rank', 'Title', 'URL'])
        for i, (t, u) in enumerate(results, 1):
            writer.writerow([i, t, u])
    
    print(f"已儲存 {len(results)} 筆結果至 {base}.json/.csv")

法律與道德提醒

  1. 遵守 robots.txt:https://www.baidu.com/robots.txt
  2. 頻率控制:最小延遲 2-5 秒,避免觸發 429
  3. 資料用途:僅供個人學習研究,勿商業用途、勿大量重發布
  4. 官方 API:百度提供開放平台 API,正式專案請申請使用

總結

百度搜尋抓取涉及請求構建、反爬應對、HTML 解析、重導向追蹤、資料儲存等完整爬蟲流程。核心挑戰在於**反爬蟲機制**(IP、驗證碼、JS 渲染)與**頁面結構變動**。生產環境強烈建議:使用 requests+BeautifulSoup、代理 IP 池、selenium/playwright 處理 JS、申請官方 API。本文代碼為教學示範,實際部署需大幅增強穩健性、合規性與可維護性。

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved