登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›爬取爱奇艺电影榜单信息

爬取爱奇艺电影榜单信息

懵诬哇 2025-6-22 17:18:57
实验介绍：

代码爬取了爱奇艺的多个榜单图片和榜单信息。
由于爱奇艺的数据格式，电影榜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:
"""爬虫工具类，提供常用工具方法"""
  1. @staticmethod
  2. def clean_filename(filename: str) -> str:
  3.     """清理文件名，移除不合法字符"""
  4.     # 使用正则表达式移除文件名中不合法的字符
  5.     return re.sub(r'[\\/*?:"<>|\'\n]', '', filename)
  7. @staticmethod
  8. def clean_tags(tags_str: str) -> List[str]:
  9.     """清理标签字符串，返回标签列表"""
  10.     # 如果标签字符串为空，返回空列表
  11.     if not tags_str:
  12.         return []
  13.     # 按逗号分割标签字符串，并去除前后空格
  14.     tags = [tag.strip() for tag in tags_str.split(',') if tag.strip()]
  15.     # 去除标签中的中文逗号和空格
  16.     return [tag.replace('，', '').replace(' ', '') for tag in tags]
  18. @staticmethod
  19. def create_dirs(paths: List[str]) -> None:
  20.     """创建目录，不存在则创建"""
  21.     # 遍历路径列表
  22.     for path in paths:
  23.         # 如果路径不存在，则创建该路径对应的目录
  24.         if not os.path.exists(path):
  25.             os.makedirs(path)
复制代码
网络请求类

class NetworkHandler:
"""网络请求处理类"""
  1. def __init__(self):
  2.     # 初始化请求头，设置用户代理
  3.     self.headers = {'User-Agent': Config.USER_AGENT}
  5. def check_robots(self, robots_path: str, url: str) -> bool:
  6.     """检查URL是否符合robots协议"""
  7.     # 创建RobotFileParser对象
  8.     rp = RobotFileParser()
  9.     try:
  10.         # 设置robots.txt文件的URL
  11.         rp.set_url(robots_path)
  12.         # 读取robots.txt文件
  13.         rp.read()
  14.         # 检查当前用户代理是否可以访问指定URL
  15.         return rp.can_fetch('*', url)
  16.     except Exception as e:
  17.         # 打印检查robots协议失败的错误信息
  18.         print(f"检查robots协议失败: {e}")
  19.         return False
  21. def fetch_text(self, url: str) -> Optional[str]:
  22.     """获取网页文本内容"""
  23.     try:
  24.         # 发送GET请求，设置请求头和超时时间
  25.         response = requests.get(url, headers=self.headers, timeout=10)
  26.         # 检查响应状态码，如果不是200，抛出异常
  27.         response.raise_for_status()
  28.         # 设置响应的编码为实际编码
  29.         response.encoding = response.apparent_encoding
  30.         # 返回响应的文本内容
  31.         return response.text
  32.     except Exception as e:
  33.         # 打印请求失败的错误信息
  34.         print(f"请求失败: {url}, 错误: {e}")
  35.         return None
  37. def fetch_binary(self, url: str) -> Optional[bytes]:
  38.     """获取二进制内容(如图像)"""
  39.     try:
  40.         # 发送GET请求，设置请求头和超时时间
  41.         response = requests.get(url, headers=self.headers, timeout=10)
  42.         # 检查响应状态码，如果不是200，抛出异常
  43.         response.raise_for_status()
  44.         # 返回响应的二进制内容
  45.         return response.content
  46.     except Exception as e:
  47.         # 打印下载失败的错误信息
  48.         print(f"下载失败: {url}, 错误: {e}")
  49.         return None
复制代码
解析类

class Parser:
"""数据解析类，负责解析HTML和JSON数据"""
  1. @staticmethod
  2. def parse_top25(html: str) -> List[MovieInfo]:
  3.     """解析Top25榜单数据"""
  4.     # 初始化电影信息列表
  5.     movies = []
  6.     # 使用BeautifulSoup解析HTML内容
  7.     soup = BeautifulSoup(html, 'lxml')
  8.     # 选择所有电影条目的链接元素
  9.     items = soup.select('div.rvi__list a')
  11.     # 遍历电影条目
  12.     for idx, item in enumerate(items, 1):
  13.         try:
  14.             # 获取电影标题
  15.             title = item.select('.rvi__tit1')[0].text.strip()
  16.             # 获取电影图片的URL
  17.             img_url = 'https:' + item.select('picture img')[0].attrs['src']
  18.             # 获取电影描述
  19.             desc = item.select('.rvi__des2')[0].text.strip()
  20.             # 获取电影标签，如果不存在则为空字符串
  21.             tags = item.select('.rvi__type1')[0].text.strip() if len(item.select('.rvi__type1')) > 0 else ''
  23.             # 创建电影信息对象并添加到列表中
  24.             movies.append(MovieInfo(
  25.                 rank=idx,
  26.                 title=title,
  27.                 img_url=img_url,
  28.                 desc=desc,
  29.                 tags=CrawlerUtils.clean_tags(tags),
  30.                 source='Top25'
  31.             ))
  32.         except (IndexError, KeyError) as e:
  33.             # 打印解析Top25条目失败的错误信息
  34.             print(f"解析Top25条目失败: {e}")
  35.             continue
  36.     return movies
  38. @staticmethod
  39. def parse_top100(json_data: Dict) -> List[MovieInfo]:
  40.     """解析Top100榜单数据"""
  41.     # 初始化电影信息列表
  42.     movies = []
  43.     try:
  44.         # 获取JSON数据中的电影内容
  45.         content = json_data['data']['formatData']['data']['content']
  46.         # 遍历电影内容
  47.         for rank, item in enumerate(content, 1):
  48.             # 获取电影标题，如果不存在则为'未知标题'
  49.             title = item.get('title', '未知标题')
  50.             # 获取电影图片的URL
  51.             img_url = item.get('img', '')
  52.             # 获取电影描述，如果不存在则为空字符串
  53.             desc = item.get('desc', '')
  54.             # 获取电影标签，将标签列表转换为逗号分隔的字符串
  55.             tags = ','.join(item.get('tags', []))
  57.             # 创建电影信息对象并添加到列表中
  58.             movies.append(MovieInfo(
  59.                 rank=rank,
  60.                 title=title,
  61.                 img_url=img_url,
  62.                 desc=desc,
  63.                 tags=CrawlerUtils.clean_tags(tags),
  64.                 source='Top100'
  65.             ))
  66.     except (KeyError, TypeError) as e:
  67.         # 打印解析Top100数据失败的错误信息
  68.         print(f"解析Top100数据失败: {e}")
  69.     return movies
复制代码
存储类

class DataStorage:
"""数据存储类，负责数据保存和图片下载"""
  1. @staticmethod
  2. def save_text_data(movies: List[MovieInfo], file_path: str) -> None:
  3.     """保存文本数据到文件"""
  4.     # 以写入模式打开文件，指定编码为UTF-8
  5.     with open(file_path, 'w', encoding='utf-8') as f:
  6.         # 遍历电影信息列表
  7.         for movie in movies:
  8.             # 生成电影信息的文本行，修正标题格式
  9.             line = f"{movie.rank}. 标题:{movie.title} 图片URL:{movie.img_url} 描述:{movie.desc} 标签:{''.join(movie.tags)}\n"
  10.             # 将文本行写入文件
  11.             f.write(line)
  13. @staticmethod
  14. def download_image(movie: MovieInfo, image_dir: str, network_handler: NetworkHandler) -> None:
  15.     """下载并保存单个图片"""
  16.     # 如果电影图片的URL为空，直接返回
  17.     if not movie.img_url:
  18.         return
  20.     # 获取电影图片的二进制数据
  21.     img_data = network_handler.fetch_binary(movie.img_url)
  22.     if img_data:
  23.         # 生成图片文件名，使用排名和电影标题
  24.         filename = f"{movie.rank:03d}_{CrawlerUtils.clean_filename(movie.title)}.jpg"
  25.         # 以二进制写入模式打开文件
  26.         with open(os.path.join(image_dir, filename), 'wb') as img_f:
  27.             # 将图片二进制数据写入文件
  28.             img_f.write(img_data)
复制代码
主爬虫类：协调各组件工作

class IqiyiCrawler:
"""爱奇艺榜单爬虫主类"""
  1. def __init__(self):
  2.     # 初始化网络请求处理对象
  3.     self.network = NetworkHandler()
  4.     # 初始化数据解析对象
  5.     self.parser = Parser()
  6.     # 初始化数据存储对象
  7.     self.storage = DataStorage()
  9. def crawl_rank(self, rank_name: str, tag_id: str, save_path: str) -> None:
  10.     """爬取单个榜单数据"""
  11.     # 打印开始处理的榜单名称
  12.     print(f"\n开始处理 {rank_name}")
  14.     # 创建目录结构
  15.     # 基础路径，包含榜单名称
  16.     base_path = os.path.join(save_path, 'data', rank_name)
  17.     # 文本数据保存路径
  18.     text_path = os.path.join(base_path, 'text')
  19.     # 图片保存路径
  20.     image_path = os.path.join(base_path, 'images')
  21.     # 创建目录
  22.     CrawlerUtils.create_dirs([text_path, image_path])
  24.     # 处理Top25数据（HTML方式）
  25.     # 构建Top25榜单的URL
  26.     url_25 = f"{Config.BASE_URL_25}{tag_id}"
  27.     # 检查URL是否符合robots协议
  28.     if self._check_robots_all(url_25):
  29.         # 打印开始获取Top25数据的信息
  30.         print(f"开始获取 {rank_name} Top25数据")
  31.         # 获取Top25榜单的HTML内容
  32.         html = self.network.fetch_text(url_25)
  33.         if html:
  34.             # 解析Top25榜单的HTML内容
  35.             top25_movies = self.parser.parse_top25(html)
  36.             if top25_movies:
  37.                 # 保存Top25数据（仅文本）
  38.                 self.storage.save_text_data(
  39.                     top25_movies,
  40.                     os.path.join(text_path, 'top25_info.txt')
  41.                 )
  42.                 # 打印Top25数据保存成功的信息
  43.                 print(f"{rank_name} Top25排名信息已保存")
  44.             else:
  45.                 # 打印Top25解析失败的信息
  46.                 print(f"{rank_name} Top25解析失败，未获取到数据")
  47.     else:
  48.         # 打印Top25不满足robots协议的信息
  49.         print(f"{rank_name} 的Top25不满足robots协议，跳过")
  51.     # 处理Top100数据（JSON方式）
  52.     # 构建Top100榜单的URL列表
  53.     urls_100 = [
  54.         f"{Config.BASE_URL_100}?page_st={tag_id}&tag={tag_id}&category_id=1&date=&pg_num={i}"
  55.         for i in range(1, 5)
  56.     ]
  57.     # 检查URL列表是否符合robots协议
  58.     if self._check_robots_all(urls_100):
  59.         # 打印开始获取Top100数据的信息
  60.         print(f"开始获取 {rank_name} Top100数据")
  61.         # 初始化Top100电影信息列表
  62.         top100_movies = []
  64.         # 遍历Top100榜单的URL列表
  65.         for url in urls_100:
  66.             # 获取URL的文本内容
  67.             response_text = self.network.fetch_text(url)
  68.             if response_text:
  69.                 try:
  70.                     # 将文本内容解析为JSON数据
  71.                     response_json = json.loads(response_text)
  72.                     # 解析JSON数据中的电影信息
  73.                     page_movies = self.parser.parse_top100(response_json)
  74.                     # 更新排名
  75.                     for i, movie in enumerate(page_movies, 1):
  76.                         movie.rank = (len(top100_movies) + i)
  77.                     # 将当前页的电影信息添加到Top100电影信息列表中
  78.                     top100_movies.extend(page_movies)
  79.                 except json.JSONDecodeError:
  80.                     # 打印解析JSON失败的信息
  81.                     print(f"解析JSON失败：{url}")
  83.         if top100_movies:
  84.             # 保存Top100数据（仅文本）
  85.             self.storage.save_text_data(
  86.                 top100_movies,
  87.                 os.path.join(text_path, 'top100_info.txt')
  88.             )
  89.             # 打印Top100数据保存成功的信息
  90.             print(f"{rank_name} Top100排名信息已保存")
  92.             # 下载图片
  93.             print(f"开始下载 {rank_name} Top100图片")
  94.             # 遍历Top100电影信息列表
  95.             for movie in top100_movies:
  96.                 # 下载并保存电影图片
  97.                 self.storage.download_image(movie, image_path, self.network)
  98.             # 打印Top100图片下载完成的信息
  99.             print(f"{rank_name} Top100图片下载完成")
  100.         else:
  101.             # 打印Top100未获取到数据的信息
  102.             print(f"{rank_name} Top100未获取到数据")
  103.     else:
  104.         # 打印Top100不满足robots协议的信息
  105.         print(f"{rank_name} 的Top100不满足robots协议，跳过")
  107. def _check_robots_all(self, urls: list) -> bool:
  108.     """检查所有robots.txt路径"""
  109.     # 如果urls是字符串，将其转换为列表
  110.     if isinstance(urls, str):
  111.         urls = [urls]
  113.     # 遍历所有robots.txt路径
  114.     for robots_path in Config.ROBOTS_TXT_PATHS:
  115.         # 遍历所有URL
  116.         for url in urls:
  117.             # 检查URL是否符合robots协议
  118.             if not self.network.check_robots(robots_path, url):
  119.                 return False
  120.     return True
  122. def run(self):
  123.     """运行爬虫"""
  124.     # 获取用户输入的保存路径
  125.     save_path = input("请输入要将data下载到的电脑路径（例如：C:/Users/YourName/Downloads）：")
  127.     # 遍历所有榜单类型
  128.     for rank_name, tag_id in Config.RANK_TYPES.items():
  129.         # 爬取单个榜单数据
  130.         self.crawl_rank(rank_name, tag_id, save_path)
复制代码
if name == "main":
# 创建爱奇艺爬虫对象
crawler = IqiyiCrawler()
# 运行爬虫
crawler.run()

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
懵诬哇
2025-6-22 17:18:57

0

粉丝关注

19

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫