开播提醒推送的图片改为使用直播间封面 (#249)

*  为bilibili直播添加下播提醒,添加直播分区信息

🚸 开播提醒推送的图片改为使用直播间封面

* ♻️ 使用Enum

* 🐛 修复bilibili发送视频动态时不推送视频标题的问题

* 🎨 调整LiveAction的注释和jaccard函数代码位置

* 🚨 修改RawPost类型定义,优化bililive类型显示
This commit is contained in:
AzideCupric 2023-05-14 01:41:25 +08:00 committed by GitHub
parent 1b077792aa
commit 24b6d60d69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 228 additions and 87 deletions

View File

@ -1,22 +1,22 @@
import json
import re
from copy import deepcopy
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Optional
from enum import Enum, unique
from typing import Any, Literal, Optional
from httpx import AsyncClient
from nonebot.log import logger
from pydantic import BaseModel, Field
from typing_extensions import Self
from ..post import Post
from ..types import ApiError, Category, RawPost, Tag, Target
from ..utils import SchedulerConfig
from ..utils import SchedulerConfig, jaccard_text_similarity
from .platform import CategoryNotRecognize, CategoryNotSupport, NewMessage, StatusChange
class BilibiliSchedConf(SchedulerConfig):
name = "bilibili.com"
schedule_type = "interval"
schedule_setting = {"seconds": 10}
@ -51,7 +51,6 @@ class BilibiliSchedConf(SchedulerConfig):
class Bilibili(NewMessage):
categories = {
1: "一般动态",
2: "专栏文章",
@ -148,7 +147,25 @@ class Bilibili(NewMessage):
pic = card["image_urls"]
elif post_type == 3:
# 视频
text = card["dynamic"]
dynamic = card.get("dynamic", "")
title = card["title"]
desc = card.get("desc", "")
if jaccard_text_similarity(desc, dynamic) > 0.8:
# 如果视频简介和动态内容相似,就只保留长的那个
if len(dynamic) > len(desc):
text = f"{dynamic}\n=================\n{title}"
else:
text = f"{title}\n\n{desc}"
else:
# 否则就把两个拼起来
text = f"""
{dynamic}
\n=================\n
{title}\n\n
{desc}
"""
pic = [card["pic"]]
elif post_type == 4:
# 纯文字
@ -191,21 +208,16 @@ class Bilibili(NewMessage):
text = card_content["item"]["content"]
orig_type = card_content["item"]["orig_type"]
orig = json.loads(card_content["origin"])
orig_text, _ = self._get_info(self._do_get_category(orig_type), orig)
orig_text, pic = self._get_info(self._do_get_category(orig_type), orig)
text += "\n--------------\n"
text += orig_text
pic = []
else:
raise CategoryNotSupport(post_type)
return Post("bilibili", text=text, url=url, pics=pic, target_name=target_name)
class Bilibililive(StatusChange):
# Author : Sichongzou
# Date : 2022-5-18 8:54
# Description : bilibili开播提醒
# E-mail : 1557157806@qq.com
categories = {1: "开播提醒", 2: "标题更新提醒"}
categories = {1: "开播提醒", 2: "标题更新提醒", 3: "下播提醒"}
platform_name = "bilibili-live"
enable_tag = False
enabled = True
@ -214,36 +226,64 @@ class Bilibililive(StatusChange):
name = "Bilibili直播"
has_target = True
@dataclass
class Info:
uname: str
live_status: int
room_id: str
@unique
class LiveStatus(Enum):
# 直播状态
# 0: 未开播
# 1: 正在直播
# 2: 轮播中
OFF = 0
ON = 1
CYCLE = 2
@unique
class LiveAction(Enum):
# 当前直播行为,由新旧直播状态对比决定
# on: 正在直播
# off: 未开播
# turn_on: 状态变更为正在直播
# turn_off: 状态变更为未开播
# title_update: 标题更新
TURN_ON = "turn_on"
TURN_OFF = "turn_off"
ON = "on"
OFF = "off"
TITLE_UPDATE = "title_update"
class Info(BaseModel):
title: str
cover_from_user: str
keyframe: str
category: Category = field(default=Category(0))
room_id: int # 直播间号
uid: int # 主播uid
live_time: int # 开播时间
live_status: "Bilibililive.LiveStatus"
area_name: str = Field(alias="area_v2_name") # 新版分区名
uname: str # 主播名
face: str # 头像url
cover: str = Field(alias="cover_from_user") # 封面url
keyframe: str # 关键帧url可能会有延迟
category: Category = Field(default=Category(0))
def __init__(self, raw_info: dict):
self.__dict__.update(raw_info)
def is_live_turn_on(self, old_info: Self) -> bool:
# 使用 & 判断直播开始
# live_status:
# 0:关播
# 1:直播中
# 2:轮播中
if self.live_status == 1 and old_info.live_status != self.live_status:
return True
return False
def is_title_update(self, old_info: Self) -> bool:
# 使用 ^ 判断直播时标题改变
if self.live_status == 1 and old_info.title != self.title:
return True
return False
def get_live_action(self, old_info: Self) -> "Bilibililive.LiveAction":
status = Bilibililive.LiveStatus
action = Bilibililive.LiveAction
if (
old_info.live_status in [status.OFF, status.CYCLE]
and self.live_status == status.ON
):
return action.TURN_ON
elif old_info.live_status == status.ON and self.live_status in [
status.OFF,
status.CYCLE,
]:
return action.TURN_OFF
elif old_info.live_status == status.ON and self.live_status == status.ON:
if old_info.title != self.title:
# 开播时通常会改标题,避免短时间推送两次
return action.TITLE_UPDATE
else:
return action.ON
else:
return action.OFF
@classmethod
async def get_target_name(
@ -259,47 +299,56 @@ class Bilibililive(StatusChange):
async def get_status(self, target: Target) -> Info:
params = {"uids[]": target}
# from https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/live/info.md#%E6%89%B9%E9%87%8F%E6%9F%A5%E8%AF%A2%E7%9B%B4%E6%92%AD%E9%97%B4%E7%8A%B6%E6%80%81
# https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/docs/live/info.md#批量查询直播间状态
res = await self.client.get(
"https://api.live.bilibili.com/room/v1/Room/get_status_info_by_uids",
params=params,
timeout=4.0,
)
res_dict = json.loads(res.text)
res_dict = res.json()
if res_dict["code"] == 0:
data = res_dict["data"][target]
info = self.Info(data)
self.Info.update_forward_refs()
info = self.Info.parse_obj(data)
return info
else:
raise self.FetchError()
def compare_status(
self, target: Target, old_status: Info, new_status: Info
self, _: Target, old_status: Info, new_status: Info
) -> list[RawPost]:
if new_status.is_live_turn_on(old_status):
# 判断开播 运算符左右有顺序要求
current_status = deepcopy(new_status)
current_status.category = Category(1)
return [current_status]
elif new_status.is_title_update(old_status):
# 判断直播时直播间标题变更 运算符左右有顺序要求
current_status = deepcopy(new_status)
current_status.category = Category(2)
return [current_status]
else:
return []
action = Bilibililive.LiveAction
match new_status.get_live_action(old_status):
case action.TURN_ON:
current_status = deepcopy(new_status)
current_status.category = Category(1)
return [current_status]
case action.TITLE_UPDATE:
current_status = deepcopy(new_status)
current_status.category = Category(2)
return [current_status]
case action.TURN_OFF:
current_status = deepcopy(new_status)
current_status.category = Category(3)
return [current_status]
case _:
return []
def get_category(self, status: Info) -> Category:
assert status.category != Category(0)
return status.category
async def parse(self, raw_info: Info) -> Post:
url = "https://live.bilibili.com/{}".format(raw_info.room_id)
pic = [raw_info.keyframe]
target_name = raw_info.uname
title = raw_info.title
async def parse(self, raw_post: Info) -> Post:
url = "https://live.bilibili.com/{}".format(raw_post.room_id)
pic = (
[raw_post.cover]
if raw_post.category == Category(1)
else [raw_post.keyframe]
)
title = f"[{self.categories[raw_post.category].rstrip('提醒')}] {raw_post.title}"
target_name = f"{raw_post.uname} {raw_post.area_name}"
return Post(
self.name,
text=title,
@ -311,7 +360,6 @@ class Bilibililive(StatusChange):
class BilibiliBangumi(StatusChange):
categories = {}
platform_name = "bilibili-bangumi"
enable_tag = False

View File

@ -5,7 +5,7 @@ from typing import Any, Literal, NamedTuple, NewType
from httpx import URL
from pydantic import BaseModel
RawPost = NewType("RawPost", Any)
RawPost = Any
Target = NewType("Target", str)
Category = int
Tag = str

View File

@ -98,3 +98,14 @@ if plugin_config.bison_filter_log:
if config.log_level is None
else config.log_level
)
def jaccard_text_similarity(str1: str, str2: str) -> float:
"""
计算两个字符串(基于字符)
[Jaccard相似系数](https://zh.wikipedia.org/wiki/雅卡尔指数)
是否达到阈值
"""
set1 = set(str1)
set2 = set(str2)
return len(set1 & set2) / len(set1 | set2)

View File

@ -32,7 +32,7 @@ async def test_video_forward(bilibili, bing_dy_list):
post = await bilibili.parse(bing_dy_list[1])
assert (
post.text
== "答案揭晓:宿舍!来看看投票结果\nhttps://t.bilibili.com/568093580488553786\n--------------\n#可露希尔的秘密档案# \n11来宿舍休息一下吧 \n档案来源lambda:\\罗德岛内务\\秘密档案 \n发布时间9/12 1:00 P.M. \n档案类型:可见 \n档案描述:今天请了病假在宿舍休息。很舒适。 \n提供者:赫默"
== "答案揭晓:宿舍!来看看投票结果\nhttps://t.bilibili.com/568093580488553786\n--------------\n#可露希尔的秘密档案# \n11来宿舍休息一下吧 \n档案来源lambda:\\罗德岛内务\\秘密档案 \n发布时间9/12 1:00 P.M. \n档案类型:可见 \n档案描述:今天请了病假在宿舍休息。很舒适。 \n提供者:赫默\n=================\n《可露希尔的秘密档案》11话来宿舍休息一下吧"
)

View File

@ -15,7 +15,7 @@ def bili_live(app: App):
@pytest.fixture
def dummy_only_status_user_subinfo(app: App):
def dummy_only_open_user_subinfo(app: App):
from nonebot_bison.types import User, UserSubInfo
user = User(123, "group")
@ -24,9 +24,7 @@ def dummy_only_status_user_subinfo(app: App):
@pytest.mark.asyncio
@respx.mock
async def test_fetch_bililive_only_status_change(
bili_live, dummy_only_status_user_subinfo
):
async def test_fetch_bililive_only_live_open(bili_live, dummy_only_open_user_subinfo):
mock_bili_live_status = get_json("bili_live_status.json")
bili_live_router = respx.get(
@ -38,28 +36,34 @@ async def test_fetch_bililive_only_status_change(
bilibili_main_page_router.mock(return_value=Response(200))
target = "13164144"
res = await bili_live.fetch_new_post(target, [dummy_only_status_user_subinfo])
res = await bili_live.fetch_new_post(target, [dummy_only_open_user_subinfo])
assert bili_live_router.call_count == 1
assert len(res) == 0
# 直播状态更新
# 直播状态更新-上播
mock_bili_live_status["data"][target]["live_status"] = 1
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res2 = await bili_live.fetch_new_post(target, [dummy_only_status_user_subinfo])
res2 = await bili_live.fetch_new_post(target, [dummy_only_open_user_subinfo])
post = res2[0][1][0]
assert post.target_type == "Bilibili直播"
assert post.text == "【Zc】从0挑战到15肉鸽目前10难度"
assert post.text == "[开播] 【Zc】从0挑战到15肉鸽目前10难度"
assert post.url == "https://live.bilibili.com/3044248"
assert post.target_name == "魔法Zc目录"
assert post.target_name == "魔法Zc目录 其他单机"
assert post.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
"https://i0.hdslb.com/bfs/live/new_room_cover/fd357f0f3cbbb48e9acfbcda616b946c2454c56c.jpg"
]
assert post.compress == True
# 标题变更
mock_bili_live_status["data"][target]["title"] = "【Zc】从0挑战到15肉鸽目前11难度"
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res3 = await bili_live.fetch_new_post(target, [dummy_only_status_user_subinfo])
res3 = await bili_live.fetch_new_post(target, [dummy_only_open_user_subinfo])
assert bili_live_router.call_count == 3
assert len(res3[0][1]) == 0
# 直播状态更新-下播
mock_bili_live_status["data"][target]["live_status"] = 0
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res4 = await bili_live.fetch_new_post(target, [dummy_only_open_user_subinfo])
assert bili_live_router.call_count == 4
assert len(res4[0][1]) == 0
@pytest.fixture
@ -95,7 +99,7 @@ async def test_fetch_bililive_only_title_change(
res0 = await bili_live.fetch_new_post(target, [dummy_only_title_user_subinfo])
assert bili_live_router.call_count == 2
assert len(res0) == 0
# 直播状态更新
# 直播状态更新-上播
mock_bili_live_status["data"][target]["live_status"] = 1
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res2 = await bili_live.fetch_new_post(target, [dummy_only_title_user_subinfo])
@ -107,9 +111,74 @@ async def test_fetch_bililive_only_title_change(
res3 = await bili_live.fetch_new_post(target, [dummy_only_title_user_subinfo])
post = res3[0][1][0]
assert post.target_type == "Bilibili直播"
assert post.text == "【Zc】从0挑战到15肉鸽目前12难度"
assert post.text == "[标题更新] 【Zc】从0挑战到15肉鸽目前12难度"
assert post.url == "https://live.bilibili.com/3044248"
assert post.target_name == "魔法Zc目录"
assert post.target_name == "魔法Zc目录 其他单机"
assert post.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
]
assert post.compress == True
# 直播状态更新-下播
mock_bili_live_status["data"][target]["live_status"] = 0
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res4 = await bili_live.fetch_new_post(target, [dummy_only_title_user_subinfo])
assert bili_live_router.call_count == 5
assert len(res4[0][1]) == 0
@pytest.fixture
def dummy_only_close_user_subinfo(app: App):
from nonebot_bison.types import User, UserSubInfo
user = User(123, "group")
return UserSubInfo(user=user, categories=[3], tags=[])
@pytest.mark.asyncio
@respx.mock
async def test_fetch_bililive_only_close(bili_live, dummy_only_close_user_subinfo):
mock_bili_live_status = get_json("bili_live_status.json")
target = "13164144"
bili_live_router = respx.get(
"https://api.live.bilibili.com/room/v1/Room/get_status_info_by_uids?uids[]=13164144"
)
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
bilibili_main_page_router = respx.get("https://www.bilibili.com/")
bilibili_main_page_router.mock(return_value=Response(200))
res = await bili_live.fetch_new_post(target, [dummy_only_close_user_subinfo])
assert bili_live_router.call_count == 1
assert len(res) == 0
# 未开播前标题变更
mock_bili_live_status["data"][target]["title"] = "【Zc】从0挑战到15肉鸽目前11难度"
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res0 = await bili_live.fetch_new_post(target, [dummy_only_close_user_subinfo])
assert bili_live_router.call_count == 2
assert len(res0) == 0
# 直播状态更新-上播
mock_bili_live_status["data"][target]["live_status"] = 1
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res2 = await bili_live.fetch_new_post(target, [dummy_only_close_user_subinfo])
assert bili_live_router.call_count == 3
assert len(res2[0][1]) == 0
# 标题变更
mock_bili_live_status["data"][target]["title"] = "【Zc】从0挑战到15肉鸽目前12难度"
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res3 = await bili_live.fetch_new_post(target, [dummy_only_close_user_subinfo])
assert bili_live_router.call_count == 4
assert len(res3[0][1]) == 0
# 直播状态更新-下播
mock_bili_live_status["data"][target]["live_status"] = 0
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res4 = await bili_live.fetch_new_post(target, [dummy_only_close_user_subinfo])
assert bili_live_router.call_count == 5
post = res4[0][1][0]
assert post.target_type == "Bilibili直播"
assert post.text == "[下播] 【Zc】从0挑战到15肉鸽目前12难度"
assert post.url == "https://live.bilibili.com/3044248"
assert post.target_name == "魔法Zc目录 其他单机"
assert post.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
]
@ -121,7 +190,7 @@ def dummy_bililive_user_subinfo(app: App):
from nonebot_bison.types import User, UserSubInfo
user = User(123, "group")
return UserSubInfo(user=user, categories=[1, 2], tags=[])
return UserSubInfo(user=user, categories=[1, 2, 3], tags=[])
@pytest.mark.asyncio
@ -147,17 +216,17 @@ async def test_fetch_bililive_combo(bili_live, dummy_bililive_user_subinfo):
res0 = await bili_live.fetch_new_post(target, [dummy_bililive_user_subinfo])
assert bili_live_router.call_count == 2
assert len(res0) == 0
# 直播状态更新
# 直播状态更新-上播
mock_bili_live_status["data"][target]["live_status"] = 1
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res2 = await bili_live.fetch_new_post(target, [dummy_bililive_user_subinfo])
post2 = res2[0][1][0]
assert post2.target_type == "Bilibili直播"
assert post2.text == "【Zc】从0挑战到15肉鸽目前11难度"
assert post2.text == "[开播] 【Zc】从0挑战到15肉鸽目前11难度"
assert post2.url == "https://live.bilibili.com/3044248"
assert post2.target_name == "魔法Zc目录"
assert post2.target_name == "魔法Zc目录 其他单机"
assert post2.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
"https://i0.hdslb.com/bfs/live/new_room_cover/fd357f0f3cbbb48e9acfbcda616b946c2454c56c.jpg"
]
assert post2.compress == True
# 标题变更
@ -166,10 +235,23 @@ async def test_fetch_bililive_combo(bili_live, dummy_bililive_user_subinfo):
res3 = await bili_live.fetch_new_post(target, [dummy_bililive_user_subinfo])
post3 = res3[0][1][0]
assert post3.target_type == "Bilibili直播"
assert post3.text == "【Zc】从0挑战到15肉鸽目前12难度"
assert post3.text == "[标题更新] 【Zc】从0挑战到15肉鸽目前12难度"
assert post3.url == "https://live.bilibili.com/3044248"
assert post3.target_name == "魔法Zc目录"
assert post3.target_name == "魔法Zc目录 其他单机"
assert post3.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
]
assert post3.compress == True
# 直播状态更新-下播
mock_bili_live_status["data"][target]["live_status"] = 0
bili_live_router.mock(return_value=Response(200, json=mock_bili_live_status))
res4 = await bili_live.fetch_new_post(target, [dummy_bililive_user_subinfo])
post4 = res4[0][1][0]
assert post4.target_type == "Bilibili直播"
assert post4.text == "[下播] 【Zc】从0挑战到15肉鸽目前12难度"
assert post4.url == "https://live.bilibili.com/3044248"
assert post4.target_name == "魔法Zc目录 其他单机"
assert post4.pics == [
"https://i0.hdslb.com/bfs/live-key-frame/keyframe10170435000003044248mwowx0.jpg"
]
assert post4.compress == True