网站 › 『编程语言区』 › 简单的python爬虫工具，B站视频爬虫

wuxiaoxiaowu 发表于 2023-11-13 10:10

简单的python爬虫工具，B站视频爬虫

分享一个我自己写的pythonB站视频爬虫，写的比较粗糙（大佬们可以帮我添加一些其他功能爬虫功能吗，如模拟登录，数据分析等；求求啦(>W<)）

当然网上一堆B站视频获取的工具，也不差我这个粗糙的python脚本，就是分享出来大家一起讨论学习，如果大家有什么好的想法和功能我们可以一起聊聊。

这里分享一个我自己用的B站视频下载的工具**(https://github.com/nilaoda/BBDown)**，很好用，作者也是在一直更新。

必要工具**(https://www.gyan.dev/ffmpeg/builds/)**，建议还是放在你的python项目目录下（我不知道为什么配置的环境变量没有生效）

这个如果想爬取高清视频就把自己的**cookie**加到**api_headers**。这里进度条加载有点问题，就是视频太小了进度条可能加载不完全，还有就是视频合成也有点问题，有时视频合成不了（我推测可能是特殊字符没有过滤完全，当然也可能是其他原因,大佬求解决(0V<)）
```
import argparse

import requests, re, sys, os, time
from contextlib import closing
from urllib import parse
from lxml import etree
import subprocess
from tqdm import tqdm


class BiliBili:
    def __init__(self, dirname):

      self.search_headers = {
            'authority': 'search.bilibili.com',
            'Accept': '*/*',
            'Referer': 'https://www.bilibili.com/',
            'Accept-Encoding': 'gzip, deflate, br',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                        'Chrome/118.0.0.0 Safari/537.36 Edg/118.0.2088.61',
      }

      self.video_headers = {
            'authority': 'www.bilibili.com',
            'Referer': 'https://www.bilibili.com/',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                        'Chrome/118.0.0.0 Safari/537.36'
      }

      self.api_headers = {
            'authority': 'api.bilibili.com',
            'Accept': '*/*',
            'Referer': 'https://www.bilibili.com/',
            'Accept-Encoding': 'gzip, deflate, br',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            # 'cookie':"",
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                        'Chrome/119.0.0.0 Safari/537.36'
      }
      self.sess = requests.Session()

      self.dir = dirname

    def downloader(self, data_url, title):
      """
      数据下载
      Parameters:
            data_url: 数据地址
            title: 标题
      """
      if self.dir not in os.listdir():
            os.mkdir(self.dir)
      size = 0
      with closing(self.sess.get(data_url, headers=self.video_headers, stream=True)) as response:
            chunk_size = 1000
            content_size = int(response.headers['content-length'])
            content_mb = content_size / 1000 / 1000
            if response.status_code == 200:
                sys.stdout.write('[开始下载]\n')
                sys.stdout.write('[文件大小]: %0.2f MB\n' % content_mb)
                video_name = os.path.join(self.dir, title)
                # 保存视频，并输出进度
                with tqdm(total=content_size, desc='[下载进度]',leave=False, ncols=100, unit='B',unit_scale=True) as pbar:
                  with open(video_name, 'wb') as file:
                        if content_mb < 3:
                            file.write(response.content)
                            for i in range(5):
                              pbar.update(content_size/5)
                        else:
                            for data in response.iter_content(chunk_size=chunk_size):
                              file.write(data)
                              pbar.update(len(data))
                              size += len(data)
                              file.flush()


                        sys.stdout.write('\n')
                        sys.stdout.write('[下载完成]' + '\r')
                        sys.stdout.flush()
                        if size / content_size == 1:
                            print('\n')
            else:
                print('~~~链接异常~~~'+'\r')
                time.sleep(1)

    def search_video(self, keyword, page=1):
      """
      搜索页视频信息
      Parameters:
            keyword: 关键词
            page: 页码
      Returns:
            videos
            titles:标题
            bvs: bv号
      """
      url = f'https://search.bilibili.com/all?keyword={parse.quote(keyword)}&page={page}&o=30'
      req = self.sess.get(url=url, headers=self.search_headers)
      html = etree.fromstring(req.text, etree.HTMLParser())
      bvs = html.xpath('//div[@class="bili-video-card__info--right"]/a/@href')[:3]
      titles = html.xpath('//div[@class="bili-video-card__info--right"]/a/h3/@title')[:3]
      videos = []
      for i, j in zip(titles, bvs):
            for c in u'´★☆❤◦\/:*?"<>|':
                i = i.replace(c, '')
            tmp =
            videos.append(tmp)
      # 输出搜索页面视频标题和视频url
      print(videos)
      return videos

    # titles, bvs
    def get_download_url(self, arcurl):
      """
      获取详情页数据信息
      Parameters:
            arcurl: 视频播放地址
      Returns:
            accept_description: 视频清晰度
            video_data: 视频地址
            audio_data: 音频地址
            title: 标题
      """
      xp = 'BV\d.{9}'
      if re.findall(xp, arcurl):
            bv = re.findall(xp, arcurl)
            url = f'https://api.bilibili.com/x/web-interface/view?bvid={bv}'# avid&cid
      else:
            print('视频BV号解析失败,请检查输入的bv号是否正确')
            exit(0)
      req1 = self.sess.get(url=url, headers=self.video_headers)
      ac_json = req1.json()
      avid = ac_json['data']['aid']
      cid = ac_json['data']['cid']
      url2 = f'https://api.bilibili.com/x/player/wbi/playurl?avid={avid}&cid={cid}&fnval=4048'# playurl
      title = ac_json['data']['title']
      req2 = self.sess.get(url=url2, headers=self.api_headers)
      playinfo_dict = req2.json()

      accept_description = playinfo_dict["data"]["accept_description"]# 视频清晰度
      # id = ["dash"]["video"]["id"]]
      audio_data = ["dash"]["audio"]["baseUrl"]]# 音频数据
      video_data = ["dash"]["video"]["baseUrl"]]
      # print(id)
      if not audio_data and not video_data:
            print('视频解析失败')
            exit(0)
      return

    def merge_data(self, dir, video_name):
      """
      视频合成
      Parameters:
            dir: 目录
            video_name: 视频名
      """
      time.sleep(0.1)
      if video_name+'_2' in os.listdir(self.dir):
            print( '合成视频已存在')
            exit(0)
      else:
            print('视频合成开始：', video_name)
            cmd = f"cd {dir} & ffmpeg -y -i {video_name}.mp4 -i {video_name}.mp3 -c:v copy -c:a aac -strict experimental -map 0:0 -map 1:0 {video_name}_2.mp4 && del {video_name}.mp4 {video_name}.mp3"
            subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            print('视频合成结束：', video_name+'\r')

    def search_downloader(self, keyword,page=1):
      """
      批量爬取搜索页视频
      Parameters:
            keyword: 关键词
            page: 页码
      """
      if self.dir not in os.listdir():
            os.mkdir(self.dir)
      for j in range(page):
            s_video = self.search_video(keyword, j+1)
            for i in range(len(s_video)):
                title = s_video
                arcurl = s_video
                if title not in os.listdir(self.dir):
                  videos_data = self.get_download_url(arcurl)
                  audio_data = self.get_download_url(arcurl)
                  if not videos_data or not audio_data:
                        print('第[ %d ]页:%s视频或音频解析失败，跳过下载:' % (1 + j, title))
                        continue# Skip video download if video or audio parsing fails

                  fname = title + '.mp4'
                  print('第[ %d ]页:视频[ %s ]下载中:' % (1 + j, fname))# 打印页码和指定下载视频
                  self.downloader(videos_data, fname)
                  print('视频下载完成!')

                  fname = title + '.mp3'
                  print('第[ %d ]页:音频[ %s ]下载中:' % (1 + j, fname))# 打印页码和指定下载视频
                  self.downloader(audio_data, fname)
                  print('音频下载完成!')

                  # 创建临时文本文件用于合并视频音频
                  try:
                        video_name = title
                        dirz = self.dir
                        self.merge_data(dirz, video_name)
                  except:
                        print('请安装FFmpeg,并配置环境变量 http://ffmpeg.org/')

    def a_video_download(self,bv):
      """
      单个视频爬取
      Parameters:
            bv: 关bv号
      """
      video_info = self.get_download_url(bv)
      title = video_info
      fname = "{0}.mp4".format(title)
      print('视频[ %s ]下载中:' % fname)# 打印页码和指定下载视频
      self.downloader(video_info, fname)
      print('视频下载完成!')
      fname = '{0}.mp3'.format(title)
      print('音频[ %s ]下载中:' % fname)# 打印页码和指定下载视频
      self.downloader(video_info, fname)
      print('音频下载完成!')
      self.merge_data(self.dir,video_info)


if __name__ == '__main__':
    if len(sys.argv) == 1:
      sys.argv.append('--help')
    parser = argparse.ArgumentParser()
    parser.add_argument('-d', '--dir', required=True, help='必要，下载路径')
    parser.add_argument('-bv', '--bvid', required=False, help='下载指定bv视频')
    parser.add_argument('-s', '--search', required=False, action='store_true', help='批量下载搜索页视频')
    parser.add_argument('-k', '--keyword', required=False, help='搜索关键词内容')
    parser.add_argument('-p', '--pages', required=False, help='需要下载页码数', type=int)
    args = parser.parse_args()

    B = BiliBili(args.dir)
    if args.search:
      if args.keyword and args.pages is None:
            print('请输入搜索关键词和页码')
            exit(0)
      B.search_downloader(args.keyword, args.pages)
    if args.bvid:
      if args.search or args.keyword or args.pages:
            print('下载单个视频请只输入BV号')
            exit(0)
      B.a_video_download(args.bvid)
    # return

    # B = BiliBili('猫')
    # url = 'https://www.bilibili.com/video/BV1Jy4y1K7yp/'
    # a=B.get_download_url(url)
    # B.downloader(a, a)

```

tanglikecuan 发表于 2023-11-13 14:04

class BiliBili:
    def __init__(self, dirname):
      #...
      self.cookie = None

    def login(self, username, password):
      login_url = 'https://accounts.bilibili.com/login/web'# B站登录页面的URL
      data = {
            'account': username,
            'password': password,
            'remember': 'true',
            'remember_me': 'on',
            'login_button': '登录',
            'service': 'pc',
            'app': 'web',
            'module': 'login',
            'from': '',
            'type': 'json',
            'ts': int(time.time() * 1000),
      }
      data = urllib.parse.urlencode(data).encode('utf-8')# 将数据编码为 bytes 类型
      headers = {
            'Content-Type': 'application/x-www-form-urlencoded',# 设置请求头中的 Content-Type
      }
      response = requests.post(login_url, data=data, headers=headers)# 发送 POST 请求进行登录
      if response.status_code == 200:# 如果状态码为200，表示请求成功
            self.cookie = response.cookies# 保存 cookie 以便后续请求使用
            print('登录成功！')
      else:
            print('登录失败！')
            print('错误信息：', response.text)# 如果请求失败，打印错误信息

需要引入 requests 和 urllib 库来处理 HTTP 请求和 URL 编码。同时，我们需要使用 re 库来处理正则表达式，以及 time 库来处理时间戳。
在 BiliBili 类的 __init__ 方法中，我们可以添加一个新的属性 cookie 来保存登录时获取的 cookie。同时，我们需要在 search_headers、video_headers 和 api_headers 中添加 cookie 字段。
接下来，我们需要添加一个方法 login 来模拟登录过程。你需要提供登录所需要的用户名和密码。

模拟登录可能会违反B站的服务条款。

sai609 发表于 2023-11-13 11:27

爬虫采集，批量下载，微信公众号原创文章，有啥办法实现？

wuxiaoxiaowu 发表于 2023-11-13 10:26

忘记改了，这个函数search_video下      bvs = html.xpath('//div[@class="bili-video-card__info--right"]/a/@href')[:3]，titles = html.xpath('//div[@class="bili-video-card__info--right"]/a/h3/@title')[:3]，这两个后面的索引改成[:30]。

yu520 发表于 2023-11-13 11:02

视屏最后总是报错是啥情况，打不开

tasty004 发表于 2023-11-13 11:15

有点看不懂,太难了

White551 发表于 2023-11-13 11:25

刚开始学Python，有些地方还不太懂，但是相信以后会理解的，大佬厉害

safew 发表于 2023-11-13 11:29

看了半懂，再看看{:1_908:}

zhang521054 发表于 2023-11-13 11:37

感谢分享 很有用啊:lol

blueshark0 发表于 2023-11-13 11:48

感谢大佬，学习下思路

wuxiaoxiaowu 发表于 2023-11-13 12:20

yu520 发表于 2023-11-13 11:02
视屏最后总是报错是啥情况，打不开

具体是什么情况，发来看看，看看能不能一起解决

查看完整版本: 简单的python爬虫工具，B站视频爬虫