为什么要保存 B 站视频的播放记录呢? 因为 B 站的历史记录,最多保存 3 个月,超过 3 个月自动清除。所以我专门写一个脚本,将历史记录导出,保存在数据库中,一来是是以备以后需要时能够找到,二来也方便对这些视频按自己的习惯进行分类和做备注。
B 站的历史记录,后端以 Web API 的方式将数据提交到前台,前后端是分离的。在浏览器中滚动条向下滑动的时候,动态提交 HTTP 请求,数据分批渲染。首先,我们需要学会如何查看 Web API。以 Chrome 浏览器为例。进入 B 站,点击右上角 「历史记录」按钮,然后按下 F12,调出开发者调试工具,切换到 Network 页签。因为历史记录是动态加载的,所以再点击筛选区的 「XHR」(XmlHttpRequest,其实就是 HTTP Request)。点击后,下面显示的都是与 xhr 相关的内容。
点击第二行(cursor***),右边出现多个页签,在 Headers 页签中,重要的有 Request URL 和 Request Headers 区域的 Cookie。将 Cookie 的内容拷贝到文本文件,比如将文件名命名为 cookie.txt。在 Chrome 浏览器中,默认情况下,右键不出现拷贝菜单,需要三次点击,选中整个字段,此时右键菜单有拷贝菜单出现。
切换到 Preview 页签,可以看出,B 站每次从后台返回 20 笔记录。我们可以通过 Preview 和 Response 来了解返回信息的数据结构。
向下滚动历史记录,左边出现更多的动态的请求内容,这些请求的 url path 主要的差异在 cursor 后面的 max 和 view_at 参数不同,max 和 view_at 来自上一次 response。换一个角度来说,每次服务端的 response,除本次 20 笔历史记录外,还同时返回下一次请求的 max 参数(表示最大的目标 ID号)和 view_at (查看的时间戳)参数。下图展示了刚才所述内容:
当所有的历史记录完毕,最后一次请求的响应,cursor 的 max = 0, view_at = 0, ps = 0。基于了解的这些信息,接下来可以通过 Python 代码来获取历史记录了。本次实现两个功能:
- 获取历史记录
- 保存到数据库
获取 B 站历史记录
编写一个从 url 返回 json 数据的函数。在该函数中,request header 参数有两个作用:提供客户端 cookie,以及将请求伪装为浏览器。
def get_response_json(url, req_headers):
"""根据url获取json格式的response文本"""
resp = requests.get(url, headers=req_headers)
return json.loads(resp.text)
从 B 站历史记录的 api 获取 json 格式数据,刚刚提到的客户端 cookie,作用是提供客户端的登录信息,cookie 信息在 request header 中。通过面向对象的方式封装代码。将 cookie 和 header 的加载放在 _init_ 方法中。BiliHistory 类对外提供两个方法:
- get_all_history(): 获取所有的浏览历史,list 类型。每一个元素为 dict 类型
- save_db(): 将历史记录保存至 sqlite 数据库 (history.db3,硬编码)
BiliHistory 类的完整代码如下:
class BiliHistory(object):
def __init__(self, cookie_file):
self.base_url = "https://api.bilibili.com/x/web-interface/history/cursor"
self.cookie = self._get_cookie_content(cookie_file)
self.request_headers = self._set_req_headers()
def _get_cookie_content(self, cookie_file):
"""从cookies.txt中读取cookie"""
with open(cookie_file, 'r') as fp:
cookies = fp.read()
return cookies
def _set_req_headers(self):
"""设置请求头:1)模拟浏览器;2)提供cookie"""
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
"Connection": "keep-alive",
"Cookie": self.cookie,
"Host": "api.bilibili.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " \
"Chrome/88.0.4324.96 Safari/537.36 Edg/88.0.705.50"
}
return headers
def _get_history(self, max, view_at, business):
"""根据url以及查询字符串中三个参数,获取20个历史记录,
history_list: 为包含dict的列表,每一个list是一个历史记录
cursor: 下一个请求的cursor信息
"""
url = self.base_url + f"?max={max}&view_at={view_at}&business={business}"
resp = get_response_json(url, self.request_headers)
history_list = resp.get("data").get("list")
next_cursor = resp.get("data").get("cursor")
return history_list, next_cursor
def get_all_history(self):
"""获取所有的的浏览历史记录"""
histories = []
max = 0
view_at = 0
business = ''
ps = 20
while(ps!=0): # ps为0表示后面没有记录
history, cursor = self._get_history(max, view_at, business)
max = cursor.get("max")
view_at = cursor.get("view_at")
business = cursor.get("business")
ps = cursor.get("ps")
for item in history:
histories.append(item)
time.sleep(0.1)
return histories
def save_db(self):
"""保存到sqlite3数据库"""
histories = self.get_all_history()
for item in histories:
history = item.get("history")
view_time = item.get("view_at")
# 如果记录不在数据库中,则新增记录
if is_url_exists(view_time) == False:
url_content = {
"title": item.get("title"),
"business": history.get("business"),
"bvid": history.get("bvid"),
"cid": history.get("cid"),
"epid": history.get("epid"),
"oid": history.get("oid"),
"page": history.get("page"),
"part": history.get("part"),
"dt": history.get("dt"),
"author_name": item.get("author_name"),
"videos": item.get("videos"),
"is_fav": item.get("is_fav"),
"tag_name": item.get("tag_name"),
"view_at": item.get("view_at"),
"progress": item.get("progress"),
"show_title": item.get("show_title"),
"cover": item.get("cover"),
"uri": item.get("uri")
}
create_url_info(url_content)
每次获取 20 条记录,在每次 Http 请求后,暂停 0.2 秒钟。
数据保存到数据库
使用 sqlalchemy ORM 的数据创建和查询功能。sqlalchemy 的用法本篇不讲述,只提供数据库操作的代码。有需要的小伙伴请参考我的博客:SQLAlchemy简明教程。
数据库操作的相关代码如下:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from model import BiliHistory
engine = create_engine("sqlite:///history.db3", echo=False)
session = sessionmaker(bind=engine)()
def is_url_exists(view_time):
"""根据浏览时间判断记录是否存在"""
item = session.query(BiliHistory).filter(BiliHistory.view_at==view_time).first()
return item != None
def create_url_info(url):
url = BiliHistory(
title = url.get("title"),
business = url.get("business"),
bvid = url.get("bvid"),
cid = url.get("cid"),
epid = url.get("epid"),
oid = url.get("oid"),
page = url.get("page"),
part = url.get("part"),
dt = url.get("dt"),
author_name = url.get("author_name"),
videos = url.get("videos"),
is_fav = url.get("is_fav"),
tag_name = url.get("tag_name"),
view_at = url.get("view_at"),
progress = url.get("progress"),
show_title = url.get("show_title"),
cover =url.get("cover"),
uri = url.get("uri")
)
session.add(url)
session.commit()
session.close()