likes
comments
collection
share

数据采集-作业-使用python爬取弹幕一个基础的视频弹幕的实现方法，一个实现获取视频弹幕的方法，需要用到aiohttp

作者站长头像
站长
· 阅读数 38

前言

今天有一个好兄弟说，他开学了，在学到数据采集的课程的时候，老师布置了一个作业，爬取b站视频的弹幕。对此，他感觉到了非常困扰。

他说，“爬取小说我能够理解，小说是文章，能够直接看到，我知道该如何爬取，但是弹幕都在视频里面，这要怎么爬呢？是下载视频吗？我不明白啊！”

可以看出，好兄弟确实是遇到了一些困惑，他可能不太了解视频网站弹幕的实现。其实，当然，弹幕也是可以爬取的，因为，弹幕也是由视频网站服务器所提供的，自然也是可以爬取到的。

弹幕原理

如同我们经常用到的那样，弹幕就是在视频/直播中，显示一条文字。因此，只要将文字绘制出来，覆盖在视频上方即可。一个常见的实现是，创建一个Canvas的弹幕系统，绘制弹幕，并且通过动画效果，控制弹幕的移动。当然，简单的覆盖效果不理想，也要通过一些算法，来调整显示的方式，比如说必须考虑到弹幕之间不应该互相覆盖，或者如果弹幕太多，屏幕无法显示（对于分辨率低，屏幕小的设备，这尤其重要），需要限制弹幕池的大小。而一些高级的视频网站中，甚至可能检测视频中的关键人物，避免弹幕遮挡到人物。

弹幕通常需要包括几条重要的信息，比如说用户id，发送时间，弹幕出现的位置，弹幕的内容等。在实际使用中，弹幕不仅仅要考虑到如何绘制和显示，也要考虑到如何发送，接受和存储。并且在存储的过程中需要考虑到安全性，例如，需要对用户的输入进行转义，防止xss的注入。

对于弹幕爬取，在实际使用中，往往会有一个专门的弹幕接口，通过这个接口，就可以得到所有的弹幕信息。如果是实时弹幕，或者弹幕数量特别多，往往会使用轮询请求的方式，例如每隔几秒钟，就请求一次新的弹幕信息。也就是说，爬取弹幕其实不需要下载视频，只需要请求这个弹幕接口就可以了。

弹幕实现

如果我们需要制作一个弹幕视频，一个非常快速的方式，就是使用danmaku的js库，这是一个专门的弹幕库，通过这个库可以轻松的发送弹幕。

首先，我们需要加载这个danmaku的js库

<script src="https://cdn.jsdelivr.net/npm/danmaku/dist/danmaku.min.js"></script>

然后创建一个danmaku，这通常需要指定一个容器

var danmaku = new Danmaku({
            container: document.getElementById('my-video-container'),
            media: document.getElementById('my-video'),
            comments: []
        });

然后，就可以通过emit向屏幕发送弹幕，设置时间，大小，颜色等

danmaku.emit({
  text: '大佬6666666',
  style: {
    fontSize: '24px',
    color: '#fff',
    border: 'none',
    backgroundColor: 'rgba(0, 0, 0, 0.5)'
  },
  time: 10
});

完整的一个html页面可能像这样：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>视频弹幕网站</title>
    <script src="https://cdn.jsdelivr.net/npm/danmaku/dist/danmaku.min.js"></script>

</head>
<body>
    <div id="my-video-container" style="width:640px;height:360px;position:relative;">
      <video id="my-video" src="你的视频.mp4" style="position:absolute;" controls></video>
    </div>

    <script>
        var danmaku = new Danmaku({
            container: document.getElementById('my-video-container'),
            media: document.getElementById('my-video'),
            comments: []
        });


        danmaku.emit({
            text: '大佬66666',
            style: {
                fontSize: '24px',
                color: '#FFFFFF',
                border: 'none',
                backgroundColor: 'rgba(0, 0, 0, 0.5)'
            },
            time: 10
        });

    </script>
</body>
</html>

这是关于视频弹幕的一种方法，如果是直播弹幕的话，还会有所不同，关于danmaku的更多了解，可以访问该库的github地址：github.com/weizhenye/D…

爬取弹幕

回到我们刚刚的问题上来，如果爬取b站的弹幕，同样需要使用b站弹幕的接口，然后通过这个接口进行请求，得到结果即可，这个并不难做到，下面就是一个爬取b站弹幕的实现。

import sys
import json
import aiohttp
import asyncio
import time
from parsel import Selector


class BiliDanmaku:

    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

    @staticmethod
    def parse_p(data):
        fields = data.split(',')
        appear_time = float(fields[0])

        send_time_epoch = int(fields[4])
        send_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(send_time_epoch))

        return {
            "视频内弹幕出现时间": appear_time,
            "弹幕发送时间": send_time,
        }

    def __init__(self, bvid, agent="bilibili client"):
        self.bvid = bvid
        self.headers = {"user-agent": agent}
        self.base_url = "https://api.bilibili.com"


    async def initialize(self):
        url = "https://api.bilibili.com/x/web-interface/view"
        params = {"bvid": self.bvid}
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params, headers=self.headers) as response:
                data = await response.json()
                self.cid = data.get("data", {}).get("cid")

    async def get_danmaku(self):
        await self.initialize()
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{self.base_url}/x/v1/dm/list.so", headers=self.headers, params={"oid": self.cid}) as response:
                r = await response.text()
                s = Selector(r)
                d = s.xpath("//d")
                danmaku_list = []
                for i in d:
                    p = i.xpath("./@p").get()
                    text = i.xpath("./text()").get()
                    info = self.parse_p(p)
                    info["弹幕内容"] = text
                    danmaku_list.append(info)
        return danmaku_list


async def __crawl_danmaku(bvid):
    b = BiliDanmaku(bvid)
    danmaku = await b.get_danmaku()
    return danmaku

def crawl_danmaku(bvid):
    return asyncio.run(__crawl_danmaku(bvid))

def save_danmaku(bvid):
    try:
        danmaku_list = crawl_danmaku(bvid)
        with open("danmaku.json", "w", encoding="utf-8") as file:
            for item in danmaku_list:
                file.write(json.dumps(item, ensure_ascii=False) + "\n")
            print("弹幕保存成功")
    except Exception as e:
        print(f"保存弹幕时发生错误: {e}")

if __name__ == "__main__":
    save_danmaku("BV1EcHgezEyF") # 输入视频BV号

运行后会在同目录下生成danmuku.json，该视频的所有弹幕都保存在里面了！

转载自:https://juejin.cn/post/7410289323739168795
评论
请登录