如何用Python抓取维基百科巴伐利亚城镇链接及Infobox数据

一、精准抓取巴伐利亚城镇页面链接的修正方案 #

原代码存在两个核心问题：

1. 请求的是英文维基的错误链接（en.wikipedia.org），但目标页面是德文维基
2. 遍历所有<a>标签会抓取导航、脚注等无关链接，需限定到城镇列表的容器内

修正后的链接抓取代码：

from bs4 import BeautifulSoup
import requests

def fetch_city_links(list_url):
    # 添加请求头模拟浏览器，避免被反爬拦截
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    response = requests.get(list_url, headers=headers)
    if response.status_code != 200:
        print(f"页面请求失败: {list_url}")
        return []

    soup = BeautifulSoup(response.content, 'html.parser')
    # 定位到城镇列表所在的column-multiple类容器
    list_containers = soup.find_all('div', class_='column-multiple')
    city_links = []

    for container in list_containers:
        # 只抓取列表项里的链接，排除特殊页面
        for li in container.find_all('li'):
            a_tag = li.find('a', href=True)
            if a_tag and '/wiki/' in a_tag['href'] and not a_tag['href'].startswith('/wiki/Spezial:'):
                full_url = 'https://de.wikipedia.org' + a_tag['href']
                city_links.append(full_url)
    
    # 去重避免重复链接
    return list(set(city_links))

二、你的全流程代码可行性评估与优化建议 #

代码可行性 #

你的整合代码整体逻辑可行，已经实现了从链接抓取、Infobox提取到CSV保存的完整流程，核心功能正常：

• 正确定位了城镇列表的容器
• 准确提取Infobox的表头和内容
• 将数据整理为DataFrame并导出CSV

关键优化点 #

为提升稳定性、效率和数据质量，建议做以下改进：

1. 添加请求头与访问延迟
   维基百科有反爬机制，频繁请求会被拦截，需添加浏览器请求头并控制请求速率：

import time

def scrape_infobox(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()  # 抛出HTTP错误
    except requests.exceptions.RequestException as e:
        print(f"请求页面出错: {url} - {str(e)}")
        return None
    # 添加1秒延迟，避免请求过快
    time.sleep(1)
    # 后续代码不变...

2. 清理Infobox数据中的冗余内容
   提取的文本可能包含引用标记（如[1]）、多余空格或换行，可加入清理逻辑：

import re

def clean_text(text):
    # 移除引用标记
    text = re.sub(r'\[\d+\]', '', text)
    # 替换多个空格为单个空格
    text = re.sub(r'\s+', ' ', text).strip()
    return text

# 在scrape_infobox函数中使用：
data[clean_text(header.get_text(" ", strip=True))] = clean_text(value.get_text(" ", strip=True))

3. 处理Infobox的合并行/表头
   部分维基页面的Infobox可能有合并的表头（如"Basisdaten"），需跳过这类无效行：

for row in infobox.find_all('tr'):
    # 跳过合并的大表头
    if row.find('th', colspan=True):
        continue
    header = row.find('th')
    value = row.find('td')
    if header and value:
        # 清理后存入数据
        clean_header = clean_text(header.get_text(" ", strip=True))
        clean_value = clean_text(value.get_text(" ", strip=True))
        data[clean_header] = clean_value

4. 可选：多线程加速抓取
   如果城镇数量较多，单线程抓取速度慢，可使用线程池并行处理，但需控制线程数量（建议5-10个）：

from concurrent.futures import ThreadPoolExecutor

def main():
    list_url = 'https://de.wikipedia.org/wiki/Liste_der_St%C3%A4dte_und_Gemeinden_in_Bayern'
    city_links = fetch_city_links(list_url)
    all_data = []

    # 使用8个线程并行抓取
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = [executor.submit(scrape_infobox, link) for link in city_links]
        for future in futures:
            result = future.result()
            if result:
                all_data.append(result)
    
    df = pd.DataFrame(all_data)
    df.to_csv('wikipedia_infoboxes.csv', index=False)

5. 保存中间数据
   抓取过程中如果中断，之前的数据会丢失，可定期保存临时结果：

# 在main函数的循环中，每抓取10个页面保存一次
for i, link in enumerate(city_links):
    print(f"抓取第{i+1}/{len(city_links)}个页面: {link}")
    infobox_data = scrape_infobox(link)
    if infobox_data:
        all_data.append(infobox_data)
    # 每10条数据保存一次临时文件
    if (i+1) % 10 == 0:
        pd.DataFrame(all_data).to_csv('wikipedia_infoboxes_temp.csv', index=False)

三、最终优化后的完整代码 #

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import re
from concurrent.futures import ThreadPoolExecutor

def fetch_city_links(list_url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    try:
        response = requests.get(list_url, headers=headers, timeout=10)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"页面请求失败: {list_url} - {str(e)}")
        return []

    soup = BeautifulSoup(response.content, 'html.parser')
    list_containers = soup.find_all('div', class_='column-multiple')
    city_links = []

    for container in list_containers:
        for li in container.find_all('li'):
            a_tag = li.find('a', href=True)
            if a_tag and '/wiki/' in a_tag['href'] and not a_tag['href'].startswith('/wiki/Spezial:'):
                full_url = 'https://de.wikipedia.org' + a_tag['href']
                city_links.append(full_url)
    
    return list(set(city_links))

def clean_text(text):
    text = re.sub(r'\[\d+\]', '', text)
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def scrape_infobox(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"请求页面出错: {url} - {str(e)}")
        return None
    
    time.sleep(1)
    soup = BeautifulSoup(response.content, 'html.parser')
    infobox = soup.find('table', {'class': 'infobox'})

    if not infobox:
        print(f"未找到Infobox: {url}")
        return None

    data = {'URL': url}
    for row in infobox.find_all('tr'):
        if row.find('th', colspan=True):
            continue
        header = row.find('th')
        value = row.find('td')
        if header and value:
            clean_header = clean_text(header.get_text(" ", strip=True))
            clean_value = clean_text(value.get_text(" ", strip=True))
            data[clean_header] = clean_value

    return data

def main():
    list_url = 'https://de.wikipedia.org/wiki/Liste_der_St%C3%A4dte_und_Gemeinden_in_Bayern'
    city_links = fetch_city_links(list_url)
    print(f"共找到{len(city_links)}个城镇链接")

    all_data = []
    # 控制线程数量，避免触发反爬
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = [executor.submit(scrape_infobox, link) for link in city_links]
        for i, future in enumerate(futures):
            result = future.result()
            if result:
                all_data.append(result)
            # 每10个页面保存一次临时数据
            if (i+1) % 10 == 0:
                print(f"已完成{i+1}个页面抓取，临时保存数据...")
                pd.DataFrame(all_data).to_csv('wikipedia_infoboxes_temp.csv', index=False)

    print("抓取完成，保存最终数据...")
    df = pd.DataFrame(all_data)
    df.to_csv('wikipedia_infoboxes.csv', index=False)
    print("数据已保存到wikipedia_infoboxes.csv")

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者zero