手把手教你用Python爬取‘可可英语’《现代大学英语精读》原文与音频(以第六单元A篇为例)
Python自动化抓取可可英语教材资源的完整实战指南
在语言学习过程中,获取高质量的原文与配套音频往往需要反复访问不同平台,手动下载效率低下。本文将展示如何用Python构建自动化工具,从可可英语网站抓取《现代大学英语精读》教材内容,实现学习资源的批量获取与系统化管理。
1. 环境准备与目标分析
可可英语网站采用常规的HTML结构展示教材内容,每个单元通常包含课文原文、翻译和音频文件。通过开发者工具分析页面请求,我们发现音频文件以MP3格式直接托管在网站服务器上,而文本内容则嵌入在特定CSS样式的div标签中。
需要安装的核心Python库:
pip install requests beautifulsoup4 pandas典型页面结构特征:
- 课文正文位于
<div class="article-content">标签内 - 音频文件URL隐藏在
<audio>标签的src属性中 - 分页内容通过URL参数控制(如
.../56861_2.shtml)
2. 基础爬虫框架搭建
我们首先构建一个能够处理网站反爬机制的基础请求模块。可可英语采用了基础的User-Agent验证和请求频率限制:
import requests from bs4 import BeautifulSoup import time headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': 'http://www.kekenet.com/' } def get_page(url): try: response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() response.encoding = 'utf-8' return response.text except Exception as e: print(f"请求失败: {e}") return None针对音频文件的下载需要单独处理:
def download_audio(audio_url, save_path): audio_data = requests.get(audio_url, headers=headers, stream=True) with open(save_path, 'wb') as f: for chunk in audio_data.iter_content(chunk_size=1024): if chunk: f.write(chunk)3. 内容解析与数据提取
使用BeautifulSoup解析HTML内容时,需要特别注意文本的清洗和格式化。以下是提取课文正文和翻译的示例代码:
def extract_content(html): soup = BeautifulSoup(html, 'html.parser') content = {'text': '', 'translation': ''} # 提取英文原文 en_paragraphs = soup.select('div.article-content p[style="text-align: left;"]') content['text'] = '\n'.join([p.get_text(strip=True) for p in en_paragraphs]) # 提取中文翻译 cn_paragraphs = soup.select('div.article-content p[style="text-align: left;"] + p') content['translation'] = '\n'.join([p.get_text(strip=True) for p in cn_paragraphs]) # 提取音频链接 audio_tag = soup.find('audio') if audio_tag: content['audio_url'] = audio_tag['src'] return content4. 分页处理与批量下载
教材内容通常分布在多个页面,我们需要实现自动翻页检测和内容合并:
def crawl_unit(base_url, unit_name): all_content = {'text': [], 'translation': []} audio_urls = [] page = 1 while True: if page == 1: url = f"{base_url}.shtml" else: url = f"{base_url}_{page}.shtml" html = get_page(url) if not html or "404 Not Found" in html: break content = extract_content(html) if content['text']: all_content['text'].append(content['text']) if content['translation']: all_content['translation'].append(content['translation']) if 'audio_url' in content: audio_urls.append(content['audio_url']) page += 1 time.sleep(1) # 礼貌性延迟 # 合并分页内容 result = { 'unit': unit_name, 'text': '\n\n'.join(all_content['text']), 'translation': '\n\n'.join(all_content['translation']), 'audio_url': audio_urls[0] if audio_urls else None } return result5. 数据存储与管理系统
将抓取的内容结构化存储到本地文件系统,便于后续学习使用:
import os import json from datetime import datetime def save_content(content, base_dir='output'): unit_dir = os.path.join(base_dir, content['unit']) os.makedirs(unit_dir, exist_ok=True) # 保存文本 with open(os.path.join(unit_dir, 'text.txt'), 'w', encoding='utf-8') as f: f.write(content['text']) # 保存翻译 with open(os.path.join(unit_dir, 'translation.txt'), 'w', encoding='utf-8') as f: f.write(content['translation']) # 保存元数据 meta = { 'unit': content['unit'], 'crawl_time': datetime.now().isoformat(), 'audio_url': content['audio_url'] } with open(os.path.join(unit_dir, 'meta.json'), 'w', encoding='utf-8') as f: json.dump(meta, f, ensure_ascii=False, indent=2) # 下载音频 if content['audio_url']: audio_path = os.path.join(unit_dir, 'audio.mp3') download_audio(content['audio_url'], audio_path)6. 完整工作流与异常处理
将各个模块组合成完整的工作流程,并添加健壮的异常处理机制:
def main(): base_url = "http://www.kekenet.com/daxue/201810/56861" unit_name = "Unit6A-TheTelephone" try: print(f"开始抓取单元: {unit_name}") content = crawl_unit(base_url, unit_name) if not content['text']: raise ValueError("未获取到有效文本内容") save_content(content) print(f"成功保存单元内容到: output/{unit_name}") except Exception as e: print(f"处理单元 {unit_name} 时出错: {str(e)}") # 可以添加重试逻辑或记录错误日志 if __name__ == "__main__": main()7. 高级技巧与优化方案
对于需要大规模抓取的情况,我们可以进一步优化:
代理IP池集成
proxies = { 'http': 'http://user:pass@proxy_ip:port', 'https': 'https://user:pass@proxy_ip:port' } response = requests.get(url, headers=headers, proxies=proxies)异步请求加速
import aiohttp import asyncio async def fetch_page(session, url): async with session.get(url) as response: return await response.text() async def main_async(urls): async with aiohttp.ClientSession(headers=headers) as session: tasks = [fetch_page(session, url) for url in urls] return await asyncio.gather(*tasks)自动化监控与更新
import schedule import time def job(): # 检查内容更新的逻辑 pass schedule.every().day.at("02:00").do(job) while True: schedule.run_pending() time.sleep(60)8. 实际应用与学习整合
将抓取的内容整合到学习流程中:
- 使用Python文本处理库生成单词本
from collections import Counter import re def generate_vocab(text): words = re.findall(r'\b[a-zA-Z]+\b', text.lower()) return Counter(words).most_common(50)- 创建自动化复习提醒系统
import sqlite3 def create_review_system(): conn = sqlite3.connect('vocab.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS vocab (word text, frequency integer, last_reviewed text)''') conn.commit() conn.close()这套工具在实际使用中显著提升了我的学习效率,特别是在准备专业英语考试时,能够快速构建个性化的学习资料库。最实用的功能是将课文与音频自动匹配,方便进行听力训练。
