抓取爱奇艺电影排行数据

实验阐述:

代码对爱奇艺的多个榜单图片以及榜单相关信息进行了抓取操作。由于爱奇艺的数据呈现特性,电影榜的前一百名是通过获取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

(0)
LomuLomu
上一篇 4小时前
下一篇 2小时前

相关推荐

  • 🚀 2025最新PyCharm永久激活教程|亲测可用至2099年(附激活码+破解补丁)

    还在为PyCharm的试用期到期而烦恼吗?😫 今天给大家带来一个超级实用的教程,手把手教你如何将PyCharm破解至2099年!这个方法适用于JetBrains全家桶(IDEA、PyCharm、DataGrip、Goland等),无论你是Windows、Mac还是Linux系统,统统都能搞定!💪 先看看成功破解后的效果吧~👇 🔍 准备工作 1. 下载PyCh…

    2025 年 6 月 2 日
    67000
  • 🚀 2025年最新IDEA永久破解教程 | 含最新激活码&全平台支持(Windows/Mac/Linux)

    作为一名Java开发者,IntelliJ IDEA无疑是提升开发效率的利器💪。这款功能强大的IDE不仅拥有智能代码补全、重构等核心功能,还有丰富的插件生态系统。但高昂的授权费用让不少开发者望而却步。今天,我将分享一套完整的IDEA永久激活方案,助你畅享2099年!(仅供学习,请支持正版) 🔍 前期准备 ⚠️ 重要提示:如果你之前安装过非官方版本或尝试过其他破…

    IDEA破解教程 2025 年 6 月 5 日
    71500
  • 2024 WebStorm最新激活码,WebStorm永久免费激活码2025-02-14 更新

    WebStorm 2024最新激活码 以下是最新的WebStorm激活码,更新时间:2025-02-14 🔑 激活码使用说明 1️⃣ 复制下方激活码 2️⃣ 打开 WebStorm 软件 3️⃣ 在菜单栏中选择 Help -> Register 4️⃣ 选择 Activation Code 5️⃣ 粘贴激活码,点击 Activate ⚠️ 必看!必看! 🔥 …

    2025 年 2 月 14 日
    37800
  • 数据类型与约束

    “`markdown title: 数据类型与约束date: 2024/12/10updated: 2024/12/10author: cmdragon excerpt:在数据库领域,数据类型与约束构成了数据存储结构的基石。正确选择数据类型不仅能够显著提升存储效率,还能增强数据的准确性与一致性。本文将深入探讨MySQL数据库中的数据类型、约束的功能以及它们…

    2024 年 12 月 24 日
    33600
  • GreatSQL temp文件占用时长分析

    GreatSQL temp文件占用时长分析 GreatSQL DBA在日常工作中可能会遇到这种情况,存在一个 InnoDB 引擎下的 temp_x.ibt 文件很大,但是却无法确定这个文件是什么时间由哪个连接建立的,难以支撑后续定位问题,今天这篇文章彻底讲明白这个问题。 现象:发现一个实例下面(4406端口对外提供服务的实例)temp文件很大,如下所示: `…

    2025 年 1 月 10 日
    32200

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

联系我们

400-800-8888

在线咨询: QQ交谈

邮件:admin@example.com

工作时间:周一至周五,9:30-18:30,节假日休息

关注微信