feat: Dynamic push support

starbot/core/bot.py

@@ -7,6 +7,7 @@ from graia.broadcast import Broadcast
 from loguru import logger
 
 from .datasource import DataSource
+from .dynamic import dynamic_spider
 from .server import http_init
 from ..exception import LiveException
 from ..exception.DataSourceException import DataSourceException
@@ -89,6 +90,9 @@ class StarBot:
             except LiveException as ex:
                 logger.error(ex.msg)
 
+        # 启动动态推送模块
+        asyncio.get_event_loop().create_task(dynamic_spider(self.__datasource))
+
         # 启动 HTTP API 服务
         if config.get("USE_HTTP_API"):
             asyncio.get_event_loop().create_task(http_init(self.__datasource))

starbot/core/dynamic.py

@@ -0,0 +1,60 @@
+import asyncio
+import time
+
+from aiohttp import ClientOSError
+from loguru import logger
+
+from .datasource import DataSource
+from ..exception import ResponseCodeException, DataSourceException
+from ..utils import config
+from ..utils.network import request
+from ..utils.utils import get_credential
+
+
+async def dynamic_spider(datasource: DataSource):
+    logger.success("动态推送模块已启动")
+
+    dynamic_url = "https://api.vc.bilibili.com/dynamic_svr/v1/dynamic_svr/dynamic_new?type_list=268435455"
+    credential = get_credential()
+    dynamic_interval = config.get("DYNAMIC_INTERVAL")
+
+    if dynamic_interval < 10:
+        logger.warning("当前动态推送抓取频率设置过小, 可能会造成动态抓取 API 访问被暂时封禁, 推荐将其设置为 10 以上的数值")
+
+    while True:
+        await asyncio.sleep(dynamic_interval)
+
+        latest_dynamic = {}
+        try:
+            latest_dynamic = await request("GET", dynamic_url, credential=credential)
+        except ResponseCodeException as ex:
+            if ex.code == -6:
+                continue
+            logger.error(f"动态推送任务抓取最新动态异常, HTTP 错误码: {ex.code} ({ex.msg})")
+        except ClientOSError:
+            continue
+        except Exception as ex:
+            logger.exception("动态推送任务抓取最新动态异常", ex)
+            continue
+
+        if "new_num" not in latest_dynamic:
+            continue
+
+        new_num = latest_dynamic["new_num"]
+        if new_num > 0:
+            logger.debug(f"检测到新动态个数: {new_num}")
+        for i in range(new_num):
+            try:
+                detail = latest_dynamic["cards"][i]
+            except IndexError:
+                break
+
+            if int(time.time()) - detail["desc"]["timestamp"] > dynamic_interval:
+                break
+
+            try:
+                up = datasource.get_up(detail["desc"]["uid"])
+            except DataSourceException:
+                continue
+
+            up.dispatch("DYNAMIC_UPDATE", detail)

starbot/core/model.py

@@ -26,7 +26,7 @@ class LiveOn(BaseModel):
     """
     开播推送内容模板。
     专用占位符：{uname} 主播昵称，{title} 直播间标题，{url} 直播间链接，{cover} 直播间封面图。
-    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片。
+    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片，{base64pic=base64字符串} base64图片。
     默认：""
     """
 
@@ -59,7 +59,7 @@ class LiveOff(BaseModel):
     """
     下播推送内容模板。
     专用占位符：{uname}主播昵称。
-    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片。
+    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片，{base64pic=base64字符串} base64图片。
     默认：""
     """
 
@@ -152,7 +152,7 @@ class DynamicUpdate(BaseModel):
     """
     动态推送内容模板。
     专用占位符：{uname}主播昵称，{action}动态操作类型（发表了新动态，转发了新动态，投稿了新视频...），{url}动态链接（若为发表视频、专栏等则为视频、专栏等对应的链接），{picture}动态图片。
-    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片。
+    通用占位符：{next} 消息分条，{atall} @全体成员，{at114514} @指定QQ号，{urlpic=链接} 网络图片，{pathpic=路径} 本地图片，{base64pic=base64字符串} base64图片。
     默认：""
     """
 

starbot/core/room.py

@@ -18,6 +18,7 @@ from .model import PushTarget
 from .user import User
 from ..exception import LiveException
 from ..utils import config, redis
+from ..utils.Painter import DynamicPicGenerator
 from ..utils.utils import get_credential, timestamp_format
 
 if typing.TYPE_CHECKING:
@@ -71,6 +72,9 @@ class Up(BaseModel):
     def inject_bot(self, bot):
         self.__bot = bot
 
+    def dispatch(self, name, data):
+        self.__room.dispatch(name, data)
+
     def __any_live_on_enabled(self):
         return any(map(lambda conf: conf.enabled, map(lambda group: group.live_on, self.targets)))
 
@@ -85,6 +89,9 @@ class Up(BaseModel):
             return any([self.__any_live_report_item_enabled(a) for a in attribute])
         return any(map(lambda t: t.live_report.enabled and t.live_report.__getattribute__(attribute), self.targets))
 
+    def __any_dynamic_update_enabled(self):
+        return any(map(lambda conf: conf.enabled, map(lambda group: group.dynamic_update, self.targets)))
+
     async def connect(self):
         """
         连接直播间
@@ -306,6 +313,48 @@ class Up(BaseModel):
                 await redis.hincrby(f"Room{type_mapping[guard_type]}Count", self.room_id, month)
                 await redis.zincrby(f"User{type_mapping[guard_type]}Count:{self.room_id}", uid, month)
 
+        if self.__any_dynamic_update_enabled():
+            @self.__room.on("DYNAMIC_UPDATE")
+            async def dynamic_update(event):
+                """
+                动态更新事件
+                """
+                logger.debug(f"{self.uname} (DYNAMIC_UPDATE): {event}")
+
+                dynamic_id = event["desc"]["dynamic_id"]
+                dynamic_type = event["desc"]["type"]
+                bvid = event['desc']['bvid'] if dynamic_type == 8 else ""
+                rid = event['desc']['rid'] if dynamic_type in (64, 256) else ""
+
+                action_map = {
+                    1: "转发了动态",
+                    2: "发表了新动态",
+                    4: "发表了新动态",
+                    8: "投稿了新视频",
+                    64: "投稿了新专栏",
+                    256: "投稿了新音频",
+                    2048: "发表了新动态"
+                }
+                url_map = {
+                    1: f"https://t.bilibili.com/{dynamic_id}",
+                    2: f"https://t.bilibili.com/{dynamic_id}",
+                    4: f"https://t.bilibili.com/{dynamic_id}",
+                    8: f"https://www.bilibili.com/video/{bvid}",
+                    64: f"https://www.bilibili.com/read/cv{rid}",
+                    256: f"https://www.bilibili.com/audio/au{rid}",
+                    2048: f"https://t.bilibili.com/{dynamic_id}"
+                }
+                base64str = await DynamicPicGenerator.generate(event)
+
+                # 推送动态消息
+                dynamic_update_args = {
+                    "{uname}": self.uname,
+                    "{action}": action_map.get(dynamic_type, "发表了新动态"),
+                    "{url}": url_map.get(dynamic_type, f"https://t.bilibili.com/{dynamic_id}"),
+                    "{picture}": "".join(["{base64pic=", base64str, "}"])
+                }
+                self.__bot.send_dynamic_update(self, dynamic_update_args)
+
     async def __accumulate_data(self):
         """
         累计直播间数据
@@ -403,8 +452,8 @@ class Up(BaseModel):
         hour, minute = divmod(minute, 60)
 
         live_report_param.update({
-            "start_time": timestamp_format(start_time),
-            "end_time": timestamp_format(end_time),
+            "start_time": timestamp_format(start_time, "%m/%d %H:%M:%S"),
+            "end_time": timestamp_format(end_time, "%m/%d %H:%M:%S"),
             "hour": hour,
             "minute": minute,
             "second": second
@@ -490,7 +539,7 @@ class Up(BaseModel):
                 io = BytesIO()
                 word_cloud = WordCloud(width=900,
                                        height=450,
-                                       font_path=f"{font_base_path}/font/{config.get('DANMU_CLOUD_FONT')}",
+                                       font_path=f"{font_base_path}/resource/{config.get('DANMU_CLOUD_FONT')}",
                                        background_color=config.get("DANMU_CLOUD_BACKGROUND_COLOR"),
                                        max_font_size=config.get("DANMU_CLOUD_MAX_FONT_SIZE"),
                                        max_words=config.get("DANMU_CLOUD_MAX_WORDS"))

starbot/utils/config.py

@@ -46,14 +46,17 @@ SIMPLE_CONFIG = {
     # 视为主播网络波动断线重连时，需发送的额外提示消息
     "UP_DISCONNECT_CONNECT_MESSAGE": "检测到下播后短时间内重新开播,可能是由于主播网络波动引起,本次开播不再重复通知",
 
-    # 绘图器普通字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
+    # 动态推送抓取频率和视为新动态的时间间隔，单位：秒
+    "DYNAMIC_INTERVAL": 10,
+
+    # 绘图器普通字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
     "PAINTER_NORMAL_FONT": "normal.ttf",
-    # 绘图器粗体字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 bold.ttf 为您的字体文件名
+    # 绘图器粗体字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 bold.ttf 为您的字体文件名
     "PAINTER_BOLD_FONT": "bold.ttf",
     # 绘图器自适应不覆盖已绘制图形的间距，单位：像素
     "PAINTER_AUTO_SIZE_BY_LIMIT_MARGIN": 10,
 
-    # 弹幕词云字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
+    # 弹幕词云字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
     "DANMU_CLOUD_FONT": "normal.ttf",
     # 弹幕词云图片背景色
     "DANMU_CLOUD_BACKGROUND_COLOR": "white",
@@ -135,14 +138,17 @@ FULL_CONFIG = {
     # 视为主播网络波动断线重连时，需发送的额外提示消息
     "UP_DISCONNECT_CONNECT_MESSAGE": "检测到下播后短时间内重新开播,可能是由于主播网络波动引起,本次开播不再重复通知",
 
-    # 绘图器普通字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
+    # 动态推送抓取频率和视为新动态的时间间隔，单位：秒
+    "DYNAMIC_INTERVAL": 10,
+
+    # 绘图器普通字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
     "PAINTER_NORMAL_FONT": "normal.ttf",
-    # 绘图器粗体字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 bold.ttf 为您的字体文件名
+    # 绘图器粗体字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 bold.ttf 为您的字体文件名
     "PAINTER_BOLD_FONT": "bold.ttf",
     # 绘图器自适应不覆盖已绘制图形的间距，单位：像素
     "PAINTER_AUTO_SIZE_BY_LIMIT_MARGIN": 10,
 
-    # 弹幕词云字体路径，如需自定义，请将字体放入 font 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
+    # 弹幕词云字体路径，如需自定义，请将字体放入 resource 文件夹中后，修改配置中的 normal.ttf 为您的字体文件名
     "DANMU_CLOUD_FONT": "normal.ttf",
     # 弹幕词云图片背景色
     "DANMU_CLOUD_BACKGROUND_COLOR": "white",

starbot/utils/utils.py

@@ -5,10 +5,14 @@
 import json
 import os
 import time
-from typing import Dict
+from io import BytesIO
+from typing import List, Dict, Optional, Any
+
+from PIL import Image, ImageDraw
 
 from . import config
 from .Credential import Credential
+from .network import get_session
 
 
 def get_api(field: str) -> Dict:
@@ -40,14 +44,104 @@ def get_credential() -> Credential:
     return Credential(sessdata, bili_jct, buvid3)
 
 
-def timestamp_format(timestamp: int) -> str:
+def timestamp_format(timestamp: int, format_str: str) -> str:
     """
     时间戳格式化为形如 11/04 00:00:00 的字符串形式
 
     Args:
         timestamp: 时间戳
+        format_str: 格式化字符串
 
     Returns:
         格式化后的字符串
     """
-    return time.strftime("%m/%d %H:%M:%S", time.localtime(timestamp))
+    return time.strftime(format_str, time.localtime(timestamp))
+
+
+async def open_url_image(url: str) -> Optional[Image.Image]:
+    """
+    读取网络图片
+
+    Args:
+        url: 图片 URL
+
+    Returns:
+        读取到的图片，URL 为空时返回 None
+    """
+    if not url:
+        return None
+
+    response = await get_session().get(url)
+    image_data = await response.read()
+    image = Image.open(BytesIO(image_data))
+    response.close()
+
+    return image
+
+
+def split_list(lst: List[Any], n: int) -> List[List[Any]]:
+    """
+    将传入列表划分为若干子列表，每个子列表包含 n 个元素
+
+    Args:
+        lst: 要划分的列表
+        n: 每个子列表包含的元素数量
+
+    Returns:
+        划分后的若干子列表组成的列表
+    """
+    sub_lists = []
+    for i in range(0, len(lst), n):
+        sub_lists.append(lst[i:i+n])
+    return sub_lists
+
+
+def limit_str_length(origin_str: str, limit: int) -> str:
+    """
+    限制字符串最大长度，将超出长度的部分截去，并添加 "..."，未超出长度则返回原字符串
+
+    Args:
+        origin_str: 原字符串
+        limit: 要限制的最大长度
+
+    Returns:
+        处理后的字符串
+    """
+    return f"{origin_str[:limit]}..." if len(origin_str) > limit else origin_str
+
+
+def mask_round(img: Image.Image) -> Image.Image:
+    """
+    将图片转换为圆形
+
+    Args:
+        img: 原图片
+
+    Returns:
+        圆形图片
+    """
+    mask = Image.new("L", img.size)
+    mask_draw = ImageDraw.Draw(mask)
+    img_width, img_height = img.size
+    mask_draw.ellipse((0, 0, img_width, img_height), fill=255)
+    img.putalpha(mask)
+    return img
+
+
+def mask_rounded_rectangle(img: Image.Image, radius: int = 10) -> Image.Image:
+    """
+    对指定图片覆盖圆角矩形蒙版，使得图片圆角化
+
+    Args:
+        img: 原图片
+        radius: 圆角半径。默认：10
+
+    Returns:
+        圆角化的图片
+    """
+    mask = Image.new("L", img.size)
+    mask_draw = ImageDraw.Draw(mask)
+    img_width, img_height = img.size
+    mask_draw.rounded_rectangle((0, 0, img_width, img_height), radius, 255)
+    img.putalpha(mask)
+    return img