实验介绍：
代码爬取了爱奇艺的多个榜单图片和榜单信息。
由于爱奇艺的数据格式，电影榜top100是获取json文件爬取的，电影榜单top25是通过HTML文件爬取的。
代码有数据结构的设计，模块划分。
代码如下：
import os
import json
import requests
from bs4 import BeautifulSoup
from urllib.robotparser import RobotFileParser
import re
from dataclasses import dataclass
from typing import List, Dict, Optional, Union, Generator
数据结构设计
@dataclass
class MovieInfo:
“””电影信息数据结构”””
rank: int # 排名
title: str # 标题
img_url: str # 图片URL
desc: str # 描述
tags: List[str] # 标签列表
source: str # 来源：Top25或Top100
配置类
class Config:
“””爬虫配置类”””
# 模拟浏览器的用户代理，用于请求网页时伪装身份
USER_AGENT = ‘Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.51 Safari/537.36’
# 爱奇艺网站的robots.txt文件路径，用于检查爬虫是否被允许访问特定URL
ROBOTS_TXT_PATHS = [‘https://www.iqiyi.com/robots.txt‘, ‘https://pcw-api.iqiyi.com/robots.txt‘]
# Top25榜单的基础URL
BASE_URL_25 = ‘https://www.iqiyi.com/ranks1/1/‘
# Top100榜单的API地址
BASE_URL_100 = ‘https://pcw-api.iqiyi.com/strategy/pcw/data/topRanksData‘
# 不同榜单类型及其对应的标签ID
RANK_TYPES = {
‘热播榜’: ‘0’,
‘飙升榜’: ‘-1’,
‘必看榜’: ‘-6’,
‘上新榜’: ‘-5’,
‘高分榜’: ‘-4’,
‘恐怖榜’: ‘7128547076428333’,
‘战争榜’: ‘4705204050526533’,
‘青春榜’: ‘8902937931540733’,
‘悬疑榜’: ‘5836257895783433’,
‘家庭榜’: ‘2375714428805633’,
‘奇幻榜’: ‘8035796650176933’,
‘动作榜’: ‘7086834452347833’,
}
工具类
class CrawlerUtils:
“””爬虫工具类，提供常用工具方法”””
@staticmethod
def clean_filename(filename: str) -> str:
"""清理文件名，移除不合法字符"""
# 使用正则表达式移除文件名中不合法的字符
return re.sub(r'[\\/*?:"<>|\'\n]', '', filename)
@staticmethod
def clean_tags(tags_str: str) -> List[str]:
"""清理标签字符串，返回标签列表"""
# 如果标签字符串为空，返回空列表
if not tags_str:
return []
# 按逗号分割标签字符串，并去除前后空格
tags = [tag.strip() for tag in tags_str.split(',') if tag.strip()]
# 去除标签中的中文逗号和空格
return [tag.replace('，', '').replace(' ', '') for tag in tags]
@staticmethod
def create_dirs(paths: List[str]) -> None:
"""创建目录，不存在则创建"""
# 遍历路径列表
for path in paths:
# 如果路径不存在，则创建该路径对应的目录
if not os.path.exists(path):
os.makedirs(path)
网络请求类
class NetworkHandler:
“””网络请求处理类”””
def __init__(self):
# 初始化请求头，设置用户代理
self.headers = {'User-Agent': Config.USER_AGENT}
def check_robots(self, robots_path: str, url: str) -> bool:
"""检查URL是否符合robots协议"""
# 创建RobotFileParser对象
rp = RobotFileParser()
try:
# 设置robots.txt文件的URL
rp.set_url(robots_path)
# 读取robots.txt文件
rp.read()
# 检查当前用户代理是否可以访问指定URL
return rp.can_fetch('*', url)
except Exception as e:
# 打印检查robots协议失败的错误信息
print(f"检查robots协议失败: {e}")
return False
def fetch_text(self, url: str) -> Optional[str]:
"""获取网页文本内容"""
try:
# 发送GET请求，设置请求头和超时时间
response = requests.get(url, headers=self.headers, timeout=10)
# 检查响应状态码，如果不是200，抛出异常
response.raise_for_status()
# 设置响应的编码为实际编码
response.encoding = response.apparent_encoding
# 返回响应的文本内容
return response.text
except Exception as e:
# 打印请求失败的错误信息
print(f"请求失败: {url}, 错误: {e}")
return None
def fetch_binary(self, url: str) -> Optional[bytes]:
"""获取二进制内容(如图像)"""
try:
# 发送GET请求，设置请求头和超时时间
response = requests.get(url, headers=self.headers, timeout=10)
# 检查响应状态码，如果不是200，抛出异常
response.raise_for_status()
# 返回响应的二进制内容
return response.content
except Exception as e:
# 打印下载失败的错误信息
print(f"下载失败: {url}, 错误: {e}")
return None
解析类
class Parser:
“””数据解析类，负责解析HTML和JSON数据”””
@staticmethod
def parse_top25(html: str) -> List[MovieInfo]:
"""解析Top25榜单数据"""
# 初始化电影信息列表
movies = []
# 使用BeautifulSoup解析HTML内容
soup = BeautifulSoup(html, 'lxml')
# 选择所有电影条目的链接元素
items = soup.select('div.rvi__list a')
# 遍历电影条目
for idx, item in enumerate(items, 1):
try:
# 获取电影标题
title = item.select('.rvi__tit1')[0].text.strip()
# 获取电影图片的URL
img_url = 'https:' + item.select('picture img')[0].attrs['src']
# 获取电影描述
desc = item.select('.rvi__des2')[0].text.strip()
# 获取电影标签，如果不存在则为空字符串
tags = item.select('.rvi__type1')[0].text.strip() if len(item.select('.rvi__type1')) > 0 else ''
# 创建电影信息对象并添加到列表中
movies.append(MovieInfo(
rank=idx,
title=title,
img_url=img_url,
desc=desc,
tags=CrawlerUtils.clean_tags(tags),
source='Top25'
))
except (IndexError, KeyError) as e:
# 打印解析Top25条目失败的错误信息
print(f"解析Top25条目失败: {e}")
continue
return movies
@staticmethod
def parse_top100(json_data: Dict) -> List[MovieInfo]:
"""解析Top100榜单数据"""
# 初始化电影信息列表
movies = []
try:
# 获取JSON数据中的电影内容
content = json_data['data']['formatData']['data']['content']
# 遍历电影内容
for rank, item in enumerate(content, 1):
# 获取电影标题，如果不存在则为'未知标题'
title = item.get('title', '未知标题')
# 获取电影图片的URL
img_url = item.get('img', '')
# 获取电影描述，如果不存在则为空字符串
desc = item.get('desc', '')
# 获取电影标签，将标签列表转换为逗号分隔的字符串
tags = ','.join(item.get('tags', []))
# 创建电影信息对象并添加到列表中
movies.append(MovieInfo(
rank=rank,
title=title,
img_url=img_url,
desc=desc,
tags=CrawlerUtils.clean_tags(tags),
source='Top100'
))
except (KeyError, TypeError) as e:
# 打印解析Top100数据失败的错误信息
print(f"解析Top100数据失败: {e}")
return movies
存储类
class DataStorage:
“””数据存储类，负责数据保存和图片下载”””
@staticmethod
def save_text_data(movies: List[MovieInfo], file_path: str) -> None:
"""保存文本数据到文件"""
# 以写入模式打开文件，指定编码为UTF-8
with open(file_path, 'w', encoding='utf-8') as f:
# 遍历电影信息列表
for movie in movies:
# 生成电影信息的文本行，修正标题格式
line = f"{movie.rank}. 标题:{movie.title} 图片URL:{movie.img_url} 描述:{movie.desc} 标签:{''.join(movie.tags)}\n"
# 将文本行写入文件
f.write(line)
@staticmethod
def download_image(movie: MovieInfo, image_dir: str, network_handler: NetworkHandler) -> None:
"""下载并保存单个图片"""
# 如果电影图片的URL为空，直接返回
if not movie.img_url:
return
# 获取电影图片的二进制数据
img_data = network_handler.fetch_binary(movie.img_url)
if img_data:
# 生成图片文件名，使用排名和电影标题
filename = f"{movie.rank:03d}_{CrawlerUtils.clean_filename(movie.title)}.jpg"
# 以二进制写入模式打开文件
with open(os.path.join(image_dir, filename), 'wb') as img_f:
# 将图片二进制数据写入文件
img_f.write(img_data)
主爬虫类：协调各组件工作
class IqiyiCrawler:
“””爱奇艺榜单爬虫主类”””
def __init__(self):
# 初始化网络请求处理对象
self.network = NetworkHandler()
# 初始化数据解析对象
self.parser = Parser()
# 初始化数据存储对象
self.storage = DataStorage()
def crawl_rank(self, rank_name: str, tag_id: str, save_path: str) -> None:
"""爬取单个榜单数据"""
# 打印开始处理的榜单名称
print(f"\n开始处理 {rank_name}")
# 创建目录结构
# 基础路径，包含榜单名称
base_path = os.path.join(save_path, 'data', rank_name)
# 文本数据保存路径
text_path = os.path.join(base_path, 'text')
# 图片保存路径
image_path = os.path.join(base_path, 'images')
# 创建目录
CrawlerUtils.create_dirs([text_path, image_path])
# 处理Top25数据（HTML方式）
# 构建Top25榜单的URL
url_25 = f"{Config.BASE_URL_25}{tag_id}"
# 检查URL是否符合robots协议
if self._check_robots_all(url_25):
# 打印开始获取Top25数据的信息
print(f"开始获取 {rank_name} Top25数据")
# 获取Top25榜单的HTML内容
html = self.network.fetch_text(url_25)
if html:
# 解析Top25榜单的HTML内容
top25_movies = self.parser.parse_top25(html)
if top25_movies:
# 保存Top25数据（仅文本）
self.storage.save_text_data(
top25_movies,
os.path.join(text_path, 'top25_info.txt')
)
# 打印Top25数据保存成功的信息
print(f"{rank_name} Top25排名信息已保存")
else:
# 打印Top25解析失败的信息
print(f"{rank_name} Top25解析失败，未获取到数据")
else:
# 打印Top25不满足robots协议的信息
print(f"{rank_name} 的Top25不满足robots协议，跳过")
# 处理Top100数据（JSON方式）
# 构建Top100榜单的URL列表
urls_100 = [
f"{Config.BASE_URL_100}?page_st={tag_id}&tag={tag_id}&category_id=1&date=&pg_num={i}"
for i in range(1, 5)
]
# 检查URL列表是否符合robots协议
if self._check_robots_all(urls_100):
# 打印开始获取Top100数据的信息
print(f"开始获取 {rank_name} Top100数据")
# 初始化Top100电影信息列表
top100_movies = []
# 遍历Top100榜单的URL列表
for url in urls_100:
# 获取URL的文本内容
response_text = self.network.fetch_text(url)
if response_text:
try:
# 将文本内容解析为JSON数据
response_json = json.loads(response_text)
# 解析JSON数据中的电影信息
page_movies = self.parser.parse_top100(response_json)
# 更新排名
for i, movie in enumerate(page_movies, 1):
movie.rank = (len(top100_movies) + i)
# 将当前页的电影信息添加到Top100电影信息列表中
top100_movies.extend(page_movies)
except json.JSONDecodeError:
# 打印解析JSON失败的信息
print(f"解析JSON失败：{url}")
if top100_movies:
# 保存Top100数据（仅文本）
self.storage.save_text_data(
top100_movies,
os.path.join(text_path, 'top100_info.txt')
)
# 打印Top100数据保存成功的信息
print(f"{rank_name} Top100排名信息已保存")
# 下载图片
print(f"开始下载 {rank_name} Top100图片")
# 遍历Top100电影信息列表
for movie in top100_movies:
# 下载并保存电影图片
self.storage.download_image(movie, image_path, self.network)
# 打印Top100图片下载完成的信息
print(f"{rank_name} Top100图片下载完成")
else:
# 打印Top100未获取到数据的信息
print(f"{rank_name} Top100未获取到数据")
else:
# 打印Top100不满足robots协议的信息
print(f"{rank_name} 的Top100不满足robots协议，跳过")
def _check_robots_all(self, urls: list) -> bool:
"""检查所有robots.txt路径"""
# 如果urls是字符串，将其转换为列表
if isinstance(urls, str):
urls = [urls]
# 遍历所有robots.txt路径
for robots_path in Config.ROBOTS_TXT_PATHS:
# 遍历所有URL
for url in urls:
# 检查URL是否符合robots协议
if not self.network.check_robots(robots_path, url):
return False
return True
def run(self):
"""运行爬虫"""
# 获取用户输入的保存路径
save_path = input("请输入要将data下载到的电脑路径（例如：C:/Users/YourName/Downloads）：")
# 遍历所有榜单类型
for rank_name, tag_id in Config.RANK_TYPES.items():
# 爬取单个榜单数据
self.crawl_rank(rank_name, tag_id, save_path)
if name == “main“:
# 创建爱奇艺爬虫对象
crawler = IqiyiCrawler()
# 运行爬虫
crawler.run()
京ICP备14020293号-4