Bilibili 关注up动态监控工具

Bilibili 关注动态监控工具

📖 简介

本脚本用于监控哔哩哔哩（Bilibili）用户关注的 UP 主动态，当有新动态发布时，会通过飞书机器人发送卡片消息通知。支持视频、图文、文字、转发、专栏等多种动态类型的监控。

✨ 功能特点

• 🔐 自动登录：支持扫码登录，自动处理 Cookie 的保存和验证
• 🎯 精确监控：支持白名单/黑名单模式，只监控指定 UP 主或排除指定 UP 主
• 📢 多类型支持：监控视频、图文、文字动态、转发、专栏等多种动态类型
• 📨 消息推送：通过飞书机器人发送包含动态信息的精美卡片消息
• 🔄 智能去重：自动识别并只推送新增动态，避免重复推送
• 🛡️ 重试机制：网络请求失败自动重试，提高稳定性
• ⏰ 定时检查：可配置检查间隔，默认 30 秒检查一次

🚀 快速开始

飞书设置

​


把这个复制下来，接下来有用。

1. 环境要求

• Python 3.7+
• 所需依赖库：requests, qrcode, schedule

2. 安装依赖

pip install requests qrcode schedule

3. 配置参数

编辑 bilibili_followed_dynamics.py 文件，修改以下配置：

3.1 配置 UP 主名单

# 要监控的UP主名单（精确匹配UP主名称）
TARGET_UPS = ["老番茄", "罗翔说刑法", "老师好我叫何同学"]

# True=白名单模式（只监控列表中的UP主）
# False=黑名单模式（监控除列表外的所有UP主）
USE_WHITELIST = True

3.2 配置监控的动态类型

# 可选类型：
# DYNAMIC_TYPE_AV – 视频
# DYNAMIC_TYPE_DRAW – 图文
# DYNAMIC_TYPE_WORD – 纯文字
# DYNAMIC_TYPE_FORWARD – 转发
# DYNAMIC_TYPE_ARTICLE – 专栏
# DYNAMIC_TYPE_PGC – 番剧
DYNAMIC_TYPES_TO_CAPTURE = ["DYNAMIC_TYPE_AV", "DYNAMIC_TYPE_DRAW", "DYNAMIC_TYPE_WORD"]

3.3 配置飞书 Webhook

在脚本中搜索 FEISHU_WEBHOOK，替换为你的飞书机器人 Webhook 地址：

FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-url"

或通过环境变量设置：

# Windows CMD
set FEISHU_WEBHOOK=你的飞书Webhook地址

# Windows PowerShell
$env:FEISHU_WEBHOOK="你的飞书Webhook地址"

# Linux/Mac
export FEISHU_WEBHOOK="你的飞书Webhook地址"

4. 运行脚本

python bilibili_followed_dynamics.py

📋 使用流程

首次运行

日常运行

• 脚本会按照设定的间隔（默认 30 秒）自动检查新动态
• 发现新动态时会通过飞书机器人发送卡片消息通知
• 动态 ID 会保存到 old_bvid.json 用于去重

📁 文件说明

⚙️ 高级配置

修改检查间隔

编辑脚本最后一部分：

# 每 30 秒检查一次
schedule.every(30).seconds.do(job)

# 或改为每 2 分钟检查一次
# schedule.every(2).minutes.do(job)

修改定时轮询间隔

扫码登录时的轮询间隔可在 _wait_for_qr_login 方法中修改：

time.sleep(5) # 当前为每 5 秒轮询一次

📝 飞书消息格式

脚本会发送以下格式的卡片消息：

错误通知：

• 标题：⚠️ 系统错误通知
• 内容：错误信息 + 时间戳
• 按钮：👉 扫码登录

动态更新：

• 标题：🔔 关注的 UP 更新动态啦！
• 内容：UP 主名称、发布时间、动态类型、标题/内容
• 按钮：👉 打开链接

🔧 常见问题

1. 飞书推送失败

• 检查 Webhook 地址是否正确
• 确认飞书机器人是否被限制发送频率
• 脚本内置了重试机制（最多 3 次），会自动重试

2. Cookie 失效

• 当 Cookie 失效时，脚本会自动生成新二维码
• 重新扫码登录即可，无需手动干预

3. 没有收到推送

• 检查 UP 主名称是否匹配（完全匹配）
• 检查动态类型配置是否包含目标类型
• 查看控制台日志，确认是否有新动态被检测到

4. 二维码过期

• 二维码有效期约 5 分钟
• 超时后重新运行脚本即可

📜 注意事项

• 建议检查间隔不要低于 30 秒，避免频繁请求被 Bilibili 限制
• 确保飞书 Webhook 地址的安全性，不要泄露
• Cookie 文件包含敏感信息，请妥善保管
• 首次运行会推送所有符合条件的动态，之后只推送新增内容

⚠️ 免责声明：本工具仅供个人学习和研究使用，请勿用于商业用途或违反 Bilibili 服务条款的行为。

Code

import random
import requests, qrcode, time, re, json, os, tempfile, filecmp, shutil, schedule
from pathlib import Path
import requests.utils as ru
from datetime import datetime

# ———–配置区域———–
# 要监控的UP主名单（精确匹配UP主名称）
# 如果列表为空，则监控所有UP主
# 例如：TARGET_UPS = ["老番茄", "罗翔说刑法", "老师好我叫何同学"]
TARGET_UPS = ["老番茄", "罗翔说刑法", "老师好我叫何同学"]
# 如果设置为 True，则只监控 TARGET_UPS 中的UP主
# 如果设置为 False，则监控除 TARGET_UPS 之外的所有UP主（黑名单模式）
USE_WHITELIST = True # True=白名单模式，False=黑名单模式

# 监控的动态类型配置
# DYNAMIC_TYPE_AV: 视频, DYNAMIC_TYPE_DRAW: 图文, DYNAMIC_TYPE_WORD: 纯文字
# DYNAMIC_TYPE_FORWARD: 转发, DYNAMIC_TYPE_ARTICLE: 专栏, DYNAMIC_TYPE_PGC: 番剧
DYNAMIC_TYPES_TO_CAPTURE = ["DYNAMIC_TYPE_AV", "DYNAMIC_TYPE_DRAW", "DYNAMIC_TYPE_WORD", "DYNAMIC_TYPE_FORWARD", "DYNAMIC_TYPE_ARTICLE"]
#——————————-

HEADERS = {
'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
'Accept': '*/*',
'Host': 'passport.bilibili.com',
'Connection': 'keep-alive'
}

# ———–运行地址———–
BASE_DIR = Path(__file__).parent
OLD_BVID_FILE = BASE_DIR / 'old_bvid.json'
COOKIE_FILE = BASE_DIR / 'cookie.txt'
JSON_FILE = BASE_DIR / 'jsonAll.json'
SAVE_FILE = BASE_DIR / 'qr.png'
#↑↑服务器公网链接展示图片

session = requests.Session()

def saveNprint_qr_image(text: str, path: str) –> None:
os.makedirs(os.path.dirname(path), exist_ok=True)
img = qrcode.make(text)
img.save(path)
print("二维码已保存到", path)
qr = qrcode.QRCode(border=1)
qr.add_data(text)
qr.print_ascii(invert=True)

def send_feishu_card_error(error_str: str):
elements = []
# 添加错误信息
elements.append({
"tag": "div",
"text": {
"tag": "lark_md",
"content": (
f"**系统提示：** {error_str} \\n"
f"**时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} \\n"
)
}
})
elements.append({
"tag": "action",
"actions": [{
"tag": "button",
"text": {"tag": "plain_text", "content": "👉 扫码登录"},
"type": "primary",
"url": "https://www.bilibili.com/"
}]
})
elements.append({"tag": "hr"})

# 飞书 Webhook 地址 – 请替换为你的实际 Webhook
FEISHU_WEBHOOK = os.getenv('FEISHU_WEBHOOK', 'https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-url')

# 构造卡片消息
card = {
"msg_type": "interactive",
"card": {
"header": {
"title": {"tag": "plain_text", "content": "⚠️ 系统错误通知"},
"template": "red"
},
"elements": elements
}
}

# 发送请求
resp = requests.post(FEISHU_WEBHOOK, json=card, timeout=10)
print("飞书推送结果：", resp.json())

def send_feishu_card(items: list[dict]):
if not items:
return

elements = []
for item in items:
# 构建内容
content = f"**UP：**{item['name']} \\n" \\
f"**时间：**{item['pub_ts']} \\n" \\
f"**类型：**{item.get('type', '📝 动态')} \\n"

# 如果有标题或描述，添加到内容中
if item.get('title'):
content += f"**标题：**{item['title']} \\n"
if item.get('desc'):
content += f"**内容：**{item['desc']} \\n"

elements.append({
"tag": "div",
"text": {
"tag": "lark_md",
"content": content
}
})
elements.append({
"tag": "action",
"actions": [{
"tag": "button",
"text": {"tag": "plain_text", "content": "👉 打开链接"},
"type": "primary",
"url": item.get('link', 'https://www.bilibili.com/')
}]
})
elements.append({"tag": "hr"})

FEISHU_WEBHOOK = os.getenv('FEISHU_WEBHOOK', 'https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-url')
card = {
"msg_type": "interactive",
"card": {
"header": {
"title": {"tag": "plain_text", "content": "🔔 关注的 UP 更新动态啦！"},
"template": "blue"
},
"elements": elements
}
}

# 添加重试机制
max_retries = 3
for attempt in range(max_retries):
try:
resp = requests.post(FEISHU_WEBHOOK, json=card, timeout=10)
print(f"飞书推送结果：{resp.text}")
return
except requests.exceptions.ConnectionError as e:
print(f"飞书推送失败（第{attempt + 1}/{max_retries}次）: {e}")
if attempt < max_retries – 1:
time.sleep(2) # 等待2秒后重试
else:
print("飞书推送最终失败，跳过本次推送")

class session_cookie:

# cookie形式转换
def dict_cookie_to_header(self, dict_cookie_str: str) –> str:
# 1. 提取字典部分
m = re.search(r'\\{.*?\\}', dict_cookie_str, flags=re.S)
if not m:
raise ValueError('未找到字典部分')
cookie_dict = eval(m.group())
# 2. 第一段里所有“公共字段”的模板（除了下面 5 个会动态替换）
# 经过测试以上【xxxx】内容需要根据抓包去获得固定值（每个用户不同），
# 每次提交的5个实际值才是有效字段，
# 没有固定值却无法正常访问
template = (
"buvid3=xxxx; "
"b_nut=xxxx; "
"_uuid=xxxx; "
"header_theme_version=OPEN; "
"enable_web_push=DISABLE; "
"home_feed_column=4; "
"browser_resolution=xxx; "
"buvid4=xxxxx; "
"DedeUserID={DedeUserID}; "
"DedeUserID__ckMd5={DedeUserID__ckMd5}; "
"theme-tip-show=SHOWED; "
"rpdid=xxxx; "
"theme-avatar-tip-show=SHOWED; "
"CURRENT_QUALITY=80; "
"CURRENT_FNVAL=4048; "
"bsource=search_baidu; "
"fingerprint=xxxx; "
"buvid_fp_plain=undefined; "
"buvid_fp=xxxxx; "
"bili_ticket=xxxxx; "
"bili_ticket_expires=xxxx; "
"SESSDATA={SESSDATA}; "
"bili_jct={bili_jct}; "
"sid={sid}; "
"bp_t_offset_140462390=xxxxx; "
"b_lsid=xxxxx"
)
# 3. 把字典里的值填进去
header_cookie = template.format(**cookie_dict)
return f"Cookie: {header_cookie}"

def __init__(self):
self.sess = requests.Session()
self.sess.headers.update(HEADERS)
self.load_cookies()

def load_cookies(self):
if COOKIE_FILE.exists() and COOKIE_FILE.stat().st_size > 0:
try:
with open(COOKIE_FILE, 'r', encoding='utf-8') as f:
cookie_str = f.read().strip()
self.sess.headers['Cookie'] = self.dict_cookie_to_header(cookie_str)
print("已加载本地 Cookie")
except Exception as e:
print("Cookie 文件损坏，已删除，准备重新登录", e)
COOKIE_FILE.unlink(missing_ok=True)
else:
print("本地无 Cookie,准备登录")

def cookie_valid(self) –> bool:
try:
if not COOKIE_FILE.exists():
self._notify_and_save_qr("Cookie 文件不存在")
return False

url = "https://api.bilibili.com/x/space/myinfo"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'https://space.bilibili.com/',
'Cookie': self.dict_cookie_to_header(COOKIE_FILE.read_text(encoding='utf-8').strip())
}
r = requests.get(url, headers=headers, timeout=10)
data = r.json()
if data.get("code") == 0 and data.get("data", {}).get("mid"):
return True
except Exception as e:
print("Cookie 校验异常:", e)

# 失效 → 提醒 + 保存二维码
self._notify_and_save_qr("Cookie 已失效，需重新扫码登录")
return False

def _notify_and_save_qr(self, msg: str):
time.sleep(1)
# 飞书提醒
send_feishu_card_error(msg)
gen_url = 'https://passport.bilibili.com/x/passport-login/web/qrcode/generate'
resp = self.sess.get(gen_url).json()
login_url = re.search(r'(https?://[^\\s<]+)', resp['data']['url']).group(0)
saveNprint_qr_image(login_url, SAVE_FILE)

def save_cookies(self):
with open(COOKIE_FILE, 'w', encoding='utf-8') as f:
json.dump(ru.dict_from_cookiejar(self.sess.cookies), f, ensure_ascii=False)
print("Cookie 已保存到", COOKIE_FILE)

def getQrCode(self):
gen_url = 'https://passport.bilibili.com/x/passport-login/web/qrcode/generate'
resp = self.sess.get(gen_url).json()
self.qrcode_key = resp['data']['qrcode_key'] # 保存 qrcode_key
login_url = re.search(r'(https?://[^\\s<]+)', resp['data']['url']).group(0)
self._notify_and_save_qr(login_url)
print(login_url)
saveNprint_qr_image(login_url, SAVE_FILE)
print(login_url)
print('请使用哔哩哔哩 App 扫描二维码，qrcode_key =', self.qrcode_key)

def ensure_login(self):
if self.cookie_valid():
print("✅ Cookie 有效，已登录")
return True # 返回 True 表示登录成功

print("❌ Cookie 无效或未登录，开始扫码登录")
self._notify_and_save_qr("Cookie 已失效，需重新扫码登录")
return self._wait_for_qr_login() # 等待扫码成功

def _wait_for_qr_login(self) –> bool:
self.getQrCode() # 显示二维码
poll_url = 'https://passport.bilibili.com/x/passport-login/web/qrcode/poll'
max_attempts = 60 # 最多等待 5分钟 (60 * 5秒)
attempts = 0
while attempts < max_attempts:
time.sleep(5) # 每 5 秒轮询一次
attempts += 1
try:
poll_resp = self.sess.get(poll_url, params={'qrcode_key': self.qrcode_key}, timeout=10).json()
code = poll_resp['data']['code']
if code == 0: # 登录成功
print("🎉 扫码成功，登录完成")
self.save_cookies() # 保存 Cookie
return True
elif code == 86101: # 未扫描
if attempts % 6 == 0: # 每30秒打印一次
print(f"等待扫码中… (已等待 {attempts * 5} 秒)")
elif code == 86090: # 已扫描未确认
print("已扫描，等待确认…")
elif code in (86038, 86039): # 二维码过期 / 失效
print("二维码已过期，请重新运行脚本")
return False
else:
print("未知状态:", poll_resp)
return False
except Exception as e:
print(f"轮询异常: {e}")
continue
print("等待超时，请重新运行脚本")
return False

def compare_and_run(self, resp: dict) –> bool:
"""返回 True 表示有更新"""
with tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8') as tmp:
json.dump(resp, tmp, ensure_ascii=False, indent=2, sort_keys=True)
tmp_path = tmp.name

try:
if JSON_FILE.exists() and filecmp.cmp(tmp_path, JSON_FILE, shallow=False):
return False
else:
shutil.move(tmp_path, JSON_FILE)
return True
finally:
if os.path.exists(tmp_path):
os.unlink(tmp_path)

def get_followed_dynamic(self):
Url_followed_dynamics = 'https://api.bilibili.com/x/polymer/web-dynamic/v1/feed/all?type=all&page=1&features=itemOpusStyle'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Cache-Control': 'max-age=0',
'Referer': 'https://www.bilibili.com/',
'Host': 'api.bilibili.com'
}

# 添加重试机制
max_retries = 3
retry_delay = 5
resp = None

for attempt in range(max_retries):
try:
resp = self.sess.get(Url_followed_dynamics, headers=headers, timeout=30)
resp.raise_for_status()
resp = resp.json()
break
except requests.exceptions.ConnectionError as e:
print(f"连接失败（第{attempt + 1}/{max_retries}次）: {e}")
if attempt < max_retries – 1:
print(f"等待 {retry_delay} 秒后重试…")
time.sleep(retry_delay)
retry_delay *= 2 # 指数退避
else:
print("连接重试最终失败，跳过本次检查")
return
has_update = self.compare_and_run(resp)
if not JSON_FILE.exists():
print("首次运行，本地无旧数据，视为更新。")
# 写json
with open(JSON_FILE, 'w', encoding='utf-8') as f:
json.dump(resp, f, ensure_ascii=False)
time.sleep(1)

# 读json
with open(JSON_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)

items = data.get('data', {}).get('items', [])

dynamics = []
for item in items:
item_type = item.get('type')
# 检查是否是需要监控的动态类型
if item_type not in DYNAMIC_TYPES_TO_CAPTURE:
continue

up_name = item['modules']['module_author']['name']

# UP主过滤逻辑
if TARGET_UPS: # 如果配置了UP主列表
if USE_WHITELIST: # 白名单模式：只推送目标UP
if up_name not in TARGET_UPS:
continue
else: # 黑名单模式：排除目标UP
if up_name in TARGET_UPS:
continue

# 获取动态ID作为唯一标识
dynamic_id = item.get('id_str', str(item.get('id', '')))

# 根据类型提取内容
title = ""
desc = ""
link = ""
type_name = ""

if item_type == 'DYNAMIC_TYPE_AV': # 视频
archive = item.get('modules', {}).get('module_dynamic', {}).get('major', {}).get('archive', {})
title = archive.get('title', '无标题')
bvid = archive.get('bvid', '')
link = f"https://www.bilibili.com/video/{bvid}"
type_name = "📹 视频"
elif item_type == 'DYNAMIC_TYPE_DRAW': # 图文
desc = item.get('modules', {}).get('module_dynamic', {}).get('desc', '图文动态')
title = "图文动态"
type_name = "🖼 图文"
link = f"https://t.bilibili.com/{dynamic_id}"
elif item_type == 'DYNAMIC_TYPE_WORD': # 纯文字
desc = item.get('modules', {}).get('module_dynamic', {}).get('desc', '文字动态')
title = "文字动态"
type_name = "📝 文字"
link = f"https://t.bilibili.com/{dynamic_id}"
elif item_type == 'DYNAMIC_TYPE_FORWARD': # 转发
major = item.get('modules', {}).get('module_dynamic', {}).get('major') or {}
orig = major.get('forward') if major else None
orig_author = orig.get('author', {}).get('name', '未知') if orig else '未知'
desc = f"转发 @{orig_author} 的动态"
title = "转发动态"
type_name = "🔄 转发"
link = f"https://t.bilibili.com/{dynamic_id}"
elif item_type == 'DYNAMIC_TYPE_ARTICLE': # 专栏
article = item.get('modules', {}).get('module_dynamic', {}).get('major', {}).get('article', {})
title = article.get('title', '专栏文章')
desc = article.get('desc', '')
link = f"https://www.bilibili.com/read/cv{dynamic_id}"
type_name = "📄 专栏"

dynamics.append({
'name': up_name,
'pub_ts': datetime.fromtimestamp(item['modules']['module_author']['pub_ts']).strftime('%Y-%m-%d %H:%M:%S'),
'title': title,
'desc': desc,
'link': link,
'type': type_name,
'id': dynamic_id
})

# 读取旧动态ID列表
try:
with OLD_BVID_FILE.open(encoding='utf-8') as f:
content = f.read().strip()
old_ids = set(json.loads(content) if content else [])
except (FileNotFoundError, json.JSONDecodeError):
old_ids = set()

new_dynamics = [d for d in dynamics if d['id'] not in old_ids]

if new_dynamics:
send_feishu_card(new_dynamics)
# 保存本轮全部动态ID供下次差分
json.dump([d['id'] for d in dynamics], OLD_BVID_FILE.open('w', encoding='utf-8'))
else:
print("本次无新增动态，不推送")

def job():
bililogin = session_cookie()
if bililogin.ensure_login(): # 等待登录成功
print(f"[{datetime.now():%H:%M:%S}] 开始抓取…")
bililogin.get_followed_dynamic()
else:
print("登录失败，无法继续抓取")

schedule.every(30).seconds.do(job) # 每60秒检查一次（建议不要低于30秒，避免被封）

print("🚀 程序启动，立即执行第一次检查…")
job() # 立即执行第一次检查

while True:
schedule.run_pending()
time.sleep(1)

成功运行

​​