有一天，当我坐在电脑前，思考着编程的奇妙之处时，一种冒险的感觉涌上心头。我想，为什么不用编程来探索互联网的深处，下载一些有趣的教材呢？于是，我开始了这次奇妙的编程之旅，带着好奇心和激情，我将向你展示如何使用Python编写一个多线程爬虫，用于下载各种教材的PDF文件。
准备工作
在我们踏上这次冒险之前，我们需要做一些准备工作。
初始化日志
首先，我们需要设置日志记录，以便随时跟踪我们的编程冒险。
import logging
def init_logging():
logging_format = '%(asctime)s\t[%(levelname)s]\t[%(name)s]\t[%(threadName)s]\t%(message)s'
logging.basicConfig(format=logging_format, level=logging.INFO)导入必要的库
为了让我们的编程冒险变得更加容易，我们需要使用requests库来进行网络请求和数据下载。如果你还没有安装这个库，请使用以下命令来安装它：
pip install requests设置下载目录和线程数量
这两个变量将在后面的代码中使用，你可以根据需要进行调整。
download_base_dir = 'd:/book3' # 设置下载文件保存的文件夹
thread_count = 8 # 设置线程数量开始编写爬虫
请求数据版本
首先，我们需要获取教材数据的版本信息，这将有助于我们确定要下载的文件。
def request_data_version() -> [str]:
# 目标URL
data_url = "https://s-file-1.ykt.cbern.com.cn/zxx/ndrs/resources/tch_material/version/data_version.json"
try:
logging.info("开始请求URL: %s", data_url)
response = requests.get(data_url)
logging.info("状态码 %s, 响应内容: %s", response.status_code, response.text)
if response.status_code == 200:
data = response.json()
return data["urls"].split(",")
else:
logging.warning("无法获得正确的响应，程序退出 -1。")
sys.exit(-1)
except requests.exceptions.RequestException as e:
logging.exception("请求数据版本失败", e)请求单个数据URL
接下来，我们将请求每个教材的数据URL，并解析教材信息。
def request_single_data_url(data_url: str) -> [dict]:
try:
logging.info("开始请求URL: %s", data_url)
response = requests.get(data_url)
logging.info("状态码 %s", response.status_code)
if response.status_code == 200:
data = response.json()
books = []
for d in data:
book_data = parse_book_data(d)
if len(book_data) != 0:
books.append(book_data)
else:
logging.warning("跳过此教材: %s", d['title'])
return books
except requests.exceptions.RequestException as e:
logging.exception("请求单个数据URL失败", e)解析教材数据
我们将解析每本教材的数据，包括教材名称和标签。
def parse_book_data(data: dict) -> dict:
book_id = data['id']
book_name = data['title']
book_tags = data['tag_list']
book_tags = sorted(book_tags, key=custom_sort_tag_list, reverse=True)
if '版' not in (book_tags[0]['tag_name']):
return {}
return {"id": book_id, "name": book_name, "dirs": get_dirs_from_tags(book_tags)}获取目录信息
我们将从标签中提取目录信息，以便稍后创建文件夹。
def get_dirs_from_tags(book_tags) -> str:
return f"{book_tags[0]['tag_name']}/{book_tags[1]['tag_name']}/{book_tags[2]['tag_name']}"自定义标签排序
为了确保我们获取的是正确的教材版本，我们需要自定义排序标签。
def custom_sort_tag_list(tag):
tag_name = tag['tag_name']
school_level, edition, grade, semester, course_name = '', '', '', '', ''
if "版" in tag_name:
edition = tag_name
elif "年级" in tag_name:
grade = tag_name
elif "册" in tag_name:
semester = tag_name
elif '小学' in tag_name or '初中' in tag_name or '高中' in tag_name:
school_level = tag_name
elif '教材' not in tag_name:
course_name = tag_name
return edition, school_level, grade, course_name, semester下载教材
最后，我们将下载每本教材的PDF文件。
def download_book(book):
file_dir = f"{download_base_dir}/{book['dirs']}"
file_name = f"{book['name']}.pdf"
file_full_path = os.path.join(file_dir, file_name)
logging.info("开始下载教材: %s/%s", file_dir, file_name)
try:
if not os.path.exists(file_dir):
os.makedirs(file_dir)
except FileExistsError:
logging.warning("文件夹 %s 已经存在", file_dir)
download_url = f"https://r1-ndr.ykt.cbern.com.cn/edu_product/esp/assets_document/{book['id']}.pkg/pdf.pdf"
response = requests.get(download_url)
if response.status_code == 200:
with open(os.path.join(file_dir, file_name), "wb") as file:
file.write(response.content)
logging.info("教材下载成功: %s", file_full_path)
else:
logging.warning("教材下载失败，状态码: %s", response.status_code)多线程下载
我们将使用多线程来提高下载速度。
def multi_thread_download
_books(shared_queue: queue.Queue):
try:
while True:
item = shared_queue.get_nowait()
download_book(item)
shared_queue.task_done()
except queue.Empty:
logging.warning("队列为空")请求所有教材
最后，我们将请求所有教材的信息并将其添加到共享队列中。
def request_all_books_to_shared_queue():
shared_queue = queue.Queue(maxsize=3000)
urls = request_data_version()
for url in urls:
books = request_single_data_url(url)
for book in books:
shared_queue.put(book)
return shared_queue执行下载
现在，我们已准备好执行下载任务。
if __name__ == '__main__':
init_logging()
# 下载文件保存的文件夹
download_base_dir = 'd:/book3'
# 线程数量
thread_count = 8
all_books_queue = request_all_books_to_shared_queue()
with concurrent.futures.ThreadPoolExecutor(max_workers=thread_count) as executor:
futures = [executor.submit(multi_thread_download_books, all_books_queue) for i in range(thread_count)]
concurrent.futures.wait(futures)
logging.info("完成下载!!!")结束语
通过这个教程，你已经学会了如何使用Python编写一个多线程爬虫，用于下载各种教材的PDF文件。这个项目不仅展示了编程的实际应用，还提供了深入探索网络资源的机会。希望你能充分利用这些技能，探索更多有趣的项目和挑战！
微信扫一扫 
