实验阐述:
代码对爱奇艺的多个榜单图片以及榜单相关信息进行了抓取操作。由于爱奇艺的数据呈现特性,电影榜的前一百名是通过获取json文件来开展爬取工作的,而电影榜单的前二十五名则是从HTML文件中进行爬取获取。代码具备数据结构的规划以及模块的划分情况。
代码示例:
```python
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 MovieInformation:
"""电影信息的数据结构"""
ranking: int # 排名
film_title: str # 标题
pic_url: str # 图片URL
movie_desc: str # 描述
label_list: List[str] # 标签列表
data_source: str # 来源:Top25或Top100
配置类
class CralwerConfig:
"""爬虫的配置类"""
# 模拟浏览器的用户代理,用于在请求网页时伪装身份
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 CrawlerHelper:
"""爬虫的工具类,提供常用的工具方法"""
@staticmethod
def purify_filename(file_name: str) -> str:
"""清理文件名,移除不合法的字符"""
# 使用正则表达式移除文件名中不合法的字符
return re.sub(r'[\\/*?:"<>|\'\n]', '', file_name)
@staticmethod
def purify_tags(tag_string: str) -> List[str]:
"""清理标签字符串,返回标签列表"""
# 如果标签字符串为空,返回空列表
if not tag_string:
return []
# 按逗号分割标签字符串,并去除前后空格
tags = [tag.strip() for tag in tag_string.split(',') if tag.strip()]
# 去除标签中的中文逗号和空格
return [tag.replace(',', '').replace(' ', '') for tag in tags]
@staticmethod
def build_directories(paths: List[str]) -> None:
"""创建目录,如果不存在则进行创建"""
# 遍历路径列表
for path in paths:
# 如果路径不存在,则创建该路径对应的目录
if not os.path.exists(path):
os.makedirs(path)
网络请求类
class NetworkAccess:
"""网络请求处理类"""
def __init__(self):
# 初始化请求头,设置用户代理
self.headers = {'User-Agent': CralwerConfig.USER_AGENT}
def verify_robots(self, robots_address: str, web_url: str) -> bool:
"""检查URL是否符合robots协议"""
# 创建RobotFileParser对象
rp = RobotFileParser()
try:
# 设置robots.txt文件的URL
rp.set_url(robots_address)
# 读取robots.txt文件
rp.read()
# 检查当前用户代理是否可以访问指定URL
return rp.can_fetch('*', web_url)
except Exception as e:
# 打印检查robots协议失败的错误信息
print(f"检查robots协议失败: {e}")
return False
def obtain_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 obtain_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 Analyzer:
"""数据解析类,负责解析HTML和JSON数据"""
@staticmethod
def analyze_top25(html_content: str) -> List[MovieInformation]:
"""解析Top25榜单数据"""
# 初始化电影信息列表
films = []
# 使用BeautifulSoup解析HTML内容
soup = BeautifulSoup(html_content, 'lxml')
# 选择所有电影条目的链接元素
entries = soup.select('div.rvi__list a')
# 遍历电影条目
for index, entry in enumerate(entries, 1):
try:
# 获取电影标题
title = entry.select('.rvi__tit1')[0].text.strip()
# 获取电影图片的URL
pic_url = 'https:' + entry.select('picture img')[0].attrs['src']
# 获取电影描述
desc = entry.select('.rvi__des2')[0].text.strip()
# 获取电影标签,如果不存在则为空字符串
tags = entry.select('.rvi__type1')[0].text.strip() if len(entry.select('.rvi__type1')) > 0 else ''
# 创建电影信息对象并添加到列表中
films.append(MovieInformation(
ranking=index,
film_title=title,
pic_url=pic_url,
movie_desc=desc,
label_list=CrawlerHelper.purify_tags(tags),
data_source='Top25'
))
except (IndexError, KeyError) as e:
# 打印解析Top25条目失败的错误信息
print(f"解析Top25条目失败: {e}")
continue
return films
@staticmethod
def analyze_top100(json_data: Dict) -> List[MovieInformation]:
"""解析Top100榜单数据"""
# 初始化电影信息列表
films = []
try:
# 获取JSON数据中的电影内容
content = json_data['data']['formatData']['data']['content']
# 遍历电影内容
for rank, item in enumerate(content, 1):
# 获取电影标题,如果不存在则为'未知标题'
title = item.get('title', '未知标题')
# 获取电影图片的URL
pic_url = item.get('img', '')
# 获取电影描述,如果不存在则为空字符串
desc = item.get('desc', '')
# 获取电影标签,将标签列表转换为逗号分隔的字符串
tags = ','.join(item.get('tags', []))
# 创建电影信息对象并添加到列表中
films.append(MovieInformation(
ranking=rank,
film_title=title,
pic_url=pic_url,
movie_desc=desc,
label_list=CrawlerHelper.purify_tags(tags),
data_source='Top100'
))
except (KeyError, TypeError) as e:
# 打印解析Top100数据失败的错误信息
print(f"解析Top100数据失败: {e}")
return films
存储类
class DataSaver:
"""数据存储类,负责数据保存和图片下载"""
@staticmethod
def save_text_data(movie_list: List[MovieInformation], file_path: str) -> None:
"""保存文本数据到文件"""
# 以写入模式打开文件,指定编码为UTF-8
with open(file_path, 'w', encoding='utf-8') as f:
# 遍历电影信息列表
for movie in movie_list:
# 生成电影信息的文本行,修正标题格式
line = f"{movie.ranking}. 标题:{movie.film_title} 图片URL:{movie.pic_url} 描述:{movie.movie_desc} 标签:{''.join(movie.label_list)}\n"
# 将文本行写入文件
f.write(line)
@staticmethod
def download_picture(movie: MovieInformation, image_directory: str, network_operate: NetworkAccess) -> None:
"""下载并保存单个图片"""
# 如果电影图片的URL为空,直接返回
if not movie.pic_url:
return
# 获取电影图片的二进制数据
img_content = network_operate.obtain_binary(movie.pic_url)
if img_content:
# 生成图片文件名,使用排名和电影标题
filename = f"{movie.ranking:03d}_{CrawlerHelper.purify_filename(movie.film_title)}.jpg"
# 以二进制写入模式打开文件
with open(os.path.join(image_directory, filename), 'wb') as img_f:
# 将图片二进制数据写入文件
img_f.write(img_content)
主爬虫类:协调各组件工作
class IqiyiSpider:
"""爱奇艺榜单爬虫主类"""
def __init__(self):
# 初始化网络请求处理对象
self.net_operation = NetworkAccess()
# 初始化数据解析对象
self.data_analyzer = Analyzer()
# 初始化数据存储对象
self.data_storer = DataSaver()
def crawl_list(self, list_name: str, tag_identifier: str, save_location: str) -> None:
"""爬取单个榜单数据"""
# 打印开始处理的榜单名称
print(f"\n开始处理 {list_name}")
# 创建目录结构
# 基础路径,包含榜单名称
basic_path = os.path.join(save_location, 'data', list_name)
# 文本数据保存路径
text_path = os.path.join(basic_path, 'text')
# 图片保存路径
image_path = os.path.join(basic_path, 'images')
# 创建目录
CrawlerHelper.build_directories([text_path, image_path])
# 处理Top25数据(HTML方式)
# 构建Top25榜单的URL
url_25 = f"{CralwerConfig.BASE_URL_25}{tag_identifier}"
# 检查URL是否符合robots协议
if self._verify_all_robots(url_25):
# 打印开始获取Top25数据的信息
print(f"开始获取 {list_name} Top25数据")
# 获取Top25榜单的HTML内容
html_content = self.net_operation.obtain_text(url_25)
if html_content:
# 解析Top25榜单的HTML内容
top25_movies = self.data_analyzer.analyze_top25(html_content)
if top25_movies:
# 保存Top25数据(仅文本)
self.data_storer.save_text_data(
top25_movies,
os.path.join(text_path, 'top25_info.txt')
)
# 打印Top25数据保存成功的信息
print(f"{list_name} Top25排名信息已保存")
else:
# 打印Top25解析失败的信息
print(f"{list_name} Top25解析失败,未获取到数据")
else:
# 打印Top25不满足robots协议的信息
print(f"{list_name} 的Top25不满足robots协议,跳过")
# 处理Top100数据(JSON方式)
# 构建Top100榜单的URL列表
urls_100 = [
f"{CralwerConfig.BASE_URL_100}?page_st={tag_identifier}&tag={tag_identifier}&category_id=1&date=&pg_num={i}"
for i in range(1, 5)
]
# 检查URL列表是否符合robots协议
if self._verify_all_robots(urls_100):
# 打印开始获取Top100数据的信息
print(f"开始获取 {list_name} Top100数据")
# 初始化Top100电影信息列表
top100_movies = []
# 遍历Top100榜单的URL列表
for url in urls_100:
# 获取URL的文本内容
response_text = self.net_operation.obtain_text(url)
if response_text:
try:
# 将文本内容解析为JSON数据
response_json = json.loads(response_text)
# 解析JSON数据中的电影信息
page_movies = self.data_analyzer.analyze_top100(response_json)
# 更新排名
for i, movie in enumerate(page_movies, 1):
movie.ranking = (len(top100_movies) + i)
# 将当前页的电影信息添加到Top100电影信息列表中
top100_movies.extend(page_movies)
except json.JSONDecodeError:
# 打印解析JSON失败的信息
print(f"解析JSON失败:{url}")
if top100_movies:
# 保存Top100数据(仅文本)
self.data_storer.save_text_data(
top100_movies,
os.path.join(text_path, 'top100_info.txt')
)
# 打印Top100数据保存成功的信息
print(f"{list_name} Top100排名信息已保存")
# 下载图片
print(f"开始下载 {list_name} Top100图片")
# 遍历Top100电影信息列表
for movie in top100_movies:
# 下载并保存电影图片
self.data_storer.download_picture(movie, image_path, self.net_operation)
# 打印Top100图片下载完成的信息
print(f"{list_name} Top100图片下载完成")
else:
# 打印Top100未获取到数据的信息
print(f"{list_name} Top100未获取到数据")
else:
# 打印Top100不满足robots协议的信息
print(f"{list_name} 的Top100不满足robots协议,跳过")
def _verify_all_robots(self, urls: list) -> bool:
"""检查所有robots.txt路径"""
# 如果urls是字符串,将其转换为列表
if isinstance(urls, str):
urls = [urls]
# 遍历所有robots.txt路径
for robots_address in CralwerConfig.ROBOTS_TXT_PATHS:
# 遍历所有URL
for url in urls:
# 检查URL是否符合robots协议
if not self.net_operation.verify_robots(robots_address, url):
return False
return True
def run_spider(self):
"""运行爬虫"""
# 获取用户输入的保存路径
save_path = input("请输入要将data下载到的电脑路径(例如:C:/Users/YourName/Downloads):")
# 遍历所有榜单类型
for list_name, tag_id in CralwerConfig.RANK_TYPES.items():
# 爬取单个榜单数据
self.crawl_list(list_name, tag_id, save_path)
if name == "main":
创建爱奇艺爬虫对象
spider = IqiyiSpider()
文章整理自互联网,只做测试使用。发布者:Lomu,转转请注明出处:https://www.it1024doc.com/12908.html