Project Gutenberg近30天热门电子书爬取与文本处理问题求助

算力阿峰

2026-8-18

需求与问题

需求如下：

爬取并下载Project Gutenberg近30天下载量前k的TXT格式公版电子书
记录前20本的书名及下载地址
对下载的书籍去除首尾页眉页脚（页眉以*** START OF THIS PROJECT GUTENBERG EBOOK 书名结尾，页脚以*** END OF THIS PROJECT GUTENBERG EBOOK 书名开头），提取正文
对正文进行分词与词形还原，构建统一词汇表并输出词频前100的词汇

现有代码无法正确获取近30天的目标书籍，存在语法错误、定位错误、URL构造错误等问题，以下是调试修复后的完整实现：

调试后的完整代码

import re
import os
from os.path import join
from os import makedirs
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from bs4 import BeautifulSoup
import nltk
from nltk.tokenize import word_tokenize
from nltk.stem import WordNetLemmatizer
from nltk.corpus import stopwords
from collections import Counter

# 下载页面内容，返回文本或None
def download_url(urlpath):
    try:
        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'
        }
        response = requests.get(urlpath, headers=headers, timeout=5)
        response.encoding = 'utf-8'
        return response.text
    except Exception as e:
        print(f"下载失败 {urlpath}: {str(e)}")
        return None

# 提取近30天下载量前k的书籍信息（书名、ID、下载地址）
def get_top_books(url, k=20):
    html = download_url(url)
    if not html:
        return []
    soup = BeautifulSoup(html, 'html.parser')
    # 定位近30天下载量的区块
    last30_section = soup.find(id='books-last30')
    if not last30_section:
        print("未找到近30天下载量的书籍区块")
        return []
    # 获取书籍列表
    book_items = last30_section.find('ol').find_all('a')[:k]
    books_info = []
    base_url = 'https://www.gutenberg.org'
    for item in book_items:
        book_name = item.text.strip()
        # 提取书籍ID
        book_id_match = re.match(r'/ebooks/(\d+)', item.get('href'))
        if not book_id_match:
            continue
        book_id = book_id_match.group(1)
        # 构造TXT下载地址（优先用UTF-8格式）
        txt_url = f"{base_url}/ebooks/{book_id}.txt.utf-8"
        # 验证地址是否有效
        test_response = requests.head(txt_url, timeout=3)
        if test_response.status_code != 200:
            txt_url = f"{base_url}/files/{book_id}/{book_id}-0.txt"
        books_info.append({
            'name': book_name,
            'id': book_id,
            'url': txt_url
        })
    return books_info

# 去除书籍的页眉页脚，提取正文
def clean_book_content(content, book_name):
    # 匹配页眉结束标记
    start_pattern = re.compile(rf'\*\*\* START OF THIS PROJECT GUTENBERG EBOOK {re.escape(book_name)} \*\*\*', re.IGNORECASE)
    # 匹配页脚开始标记
    end_pattern = re.compile(rf'\*\*\* END OF THIS PROJECT GUTENBERG EBOOK {re.escape(book_name)} \*\*\*', re.IGNORECASE)
    
    start_match = start_pattern.search(content)
    end_match = end_pattern.search(content)
    
    if start_match and end_match:
        return content[start_match.end():end_match.start()].strip()
    elif start_match:
        return content[start_match.end():].strip()
    elif end_match:
        return content[:end_match.start()].strip()
    else:
        return content.strip()

# 下载单本书籍并保存
def download_and_save_book(book_info, save_path):
    book_name = book_info['name']
    book_id = book_info['id']
    txt_url = book_info['url']
    
    content = download_url(txt_url)
    if not content:
        return f"下载失败：{book_name}"
    
    # 清理内容
    cleaned_content = clean_book_content(content, book_name)
    
    # 保存文件
    save_file = join(save_path, f"{book_id}_{re.sub(r'[\\/*?:"<>|]', '_', book_name)}.txt")
    with open(save_file, 'w', encoding='utf-8') as f:
        f.write(cleaned_content)
    return f"已保存：{book_name} -> {save_file}"

# 批量下载书籍并记录前20本信息
def batch_download_top_books(url, save_path, k=20):
    # 创建保存目录
    makedirs(save_path, exist_ok=True)
    
    # 获取前k本的书籍信息
    books_info = get_top_books(url, k)
    if not books_info:
        print("未获取到有效书籍信息")
        return
    
    # 记录前20本的书名和地址
    with open(join(save_path, 'top20_books.txt'), 'w', encoding='utf-8') as f:
        for idx, info in enumerate(books_info[:20], 1):
            f.write(f"{idx}. 《{info['name']}》 - 下载地址：{info['url']}\n")
    print("已记录前20本书籍信息到top20_books.txt")
    
    # 多线程下载书籍
    with ThreadPoolExecutor(max_workers=5) as exe:
        futures = [exe.submit(download_and_save_book, info, save_path) for info in books_info]
        for future in as_completed(futures):
            print(future.result())

# 处理书籍文本：分词+词形还原
def process_book_text(text):
    # 分词并转为小写
    tokens = word_tokenize(text.lower())
    # 加载英文停用词
    stop_words = set(stopwords.words('english'))
    # 过滤非字母字符和停用词
    filtered_tokens = [token for token in tokens if token.isalpha() and token not in stop_words]
    # 词形还原
    lemmatizer = WordNetLemmatizer()
    lemmatized_tokens = [lemmatizer.lemmatize(token) for token in filtered_tokens]
    return lemmatized_tokens

# 构建词汇表并输出前100词
def build_vocabulary(books_dir, top_n=100):
    word_counter = Counter()
    # 遍历所有书籍文件
    for filename in os.listdir(books_dir):
        if filename.endswith('.txt'):
            file_path = join(books_dir, filename)
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    text = f.read()
                    tokens = process_book_text(text)
                    word_counter.update(tokens)
            except Exception as e:
                print(f"处理文件失败 {filename}: {str(e)}")
                continue
    # 输出前top_n个高频词汇
    print("\n词汇表前100词（按词频排序）：")
    for word, count in word_counter.most_common(top_n):
        print(f"{word}: {count}")
    # 保存到文件
    with open(join(books_dir, 'vocabulary_top100.txt'), 'w', encoding='utf-8') as f:
        for word, count in word_counter.most_common(top_n):
            f.write(f"{word}: {count}\n")
    print("已保存词汇表到vocabulary_top100.txt")

# 主程序入口
if __name__ == "__main__":
    # 下载前20本近30天热门书籍
    TOP_BOOKS_URL = 'https://www.gutenberg.org/browse/scores/top'
    SAVE_DIR = 'gutenberg_books'
    batch_download_top_books(TOP_BOOKS_URL, SAVE_DIR, k=20)
    
    # 下载nltk必要语料（首次运行需要）
    nltk.download('punkt', quiet=True)
    nltk.download('wordnet', quiet=True)
    nltk.download('stopwords', quiet=True)
    
    # 构建词汇表并输出前100词
    build_vocabulary(SAVE_DIR)

关键修改说明

修复语法错误：修正原代码中download_top_100函数的缩进问题，补充缺失的变量定义，移除冗余代码。
精准定位目标书籍：通过页面id="books-last30"元素定位近30天下载量区块，避免依赖固定索引的<ol>标签（页面结构变化会导致失效）。
修正TXT地址构造：优先使用通用UTF-8格式的书籍地址，无效时回退到传统文件路径，提升下载成功率。
添加书籍信息记录：自动将前20本书的书名和下载地址保存到top20_books.txt。
实现页眉页脚清理：通过正则匹配精准移除Project Gutenberg的首尾标识内容，只保留正文。
优化多线程下载：限制线程池大小，避免请求过于频繁被服务器限制。

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