feat: (telegram) inline search songs from bili only get results from music type

core/inline.py

@@ -1,5 +1,6 @@
 from aiogram.enums import ParseMode
 from aiogram.types import InlineQuery, InlineQueryResultArticle, InputTextMessageContent
+from aiogram.utils.formatting import BlockQuote, Text
 
 
 async def handle_inline_query(query: InlineQuery):
@@ -106,7 +107,7 @@ async def handle_inline_query(query: InlineQuery):
         return
     """
     if query_text.startswith('anuo'):
-        main = query_text.replace("anuo", "").strip()
+        main = query_text.replace("anuo", "",1).strip()
         await query.answer(results=[
             InlineQueryResultArticle(
                 id="1",
@@ -119,6 +120,68 @@ async def handle_inline_query(query: InlineQuery):
             )
         ], cache_time=0)
         return
+    if query_text.startswith("b23"):
+        b23_query = query_text.replace("b23", "",1).strip()
+        b23_resp = None
+        import aiohttp
+        async with aiohttp.ClientSession() as session:
+            # 先访问 bilibili.com 获取 cookies
+            async with session.get('https://bilibili.com', headers={
+                "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/138.0.0.0"}) as response:
+                pass
+
+            # 使用获取的 cookies 请求搜索 API
+            params = {'keyword': b23_query}
+            async with session.get(
+                    'https://api.bilibili.com/x/web-interface/search/all/v2',
+                    params=params,
+                    headers={
+                        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/138.0.0.0"
+                    }
+            ) as response:
+                b23_resp = await response.json()
+        search_results = []
+        if b23_resp and b23_resp.get('data'):
+            # 假设我们只取第一个视频的结果
+            videos = next((item for item in b23_resp['data']['result'] if item.get('result_type') == 'video'), None)
+            if videos and videos.get('data'):
+                # 取前十个结果
+                for i, video in enumerate(videos['data'][:10]):
+                    title = video.get('title', '').replace('<em class="keyword">', '').replace('</em>', '')
+                    bvid = video.get('bvid', '')
+                    link = video.get('arcurl', '').replace('http://','https://',1)
+                    video_type = video.get('typename', '')
+                    author = video.get('author', '')
+                    play = video.get('play', 0)
+                    thumbnail = f"https:{video.get('pic')}"
+                    description = video.get('description', '')
+
+                    search_results.append(InlineQueryResultArticle(
+                        id=str(i + 1),
+                        title=title,
+                        thumbnail_url=thumbnail,
+                        input_message_content=InputTextMessageContent(
+                            message_text=f"<a href=\"{link}\">{title}</a>\n{video_type} | 作者：{author} | "
+                                         f"播放量：{play} {Text(BlockQuote(description)).as_html()}",
+                            parse_mode=ParseMode.HTML
+                        ),
+                        description=f"{bvid} | 作者：{author} | 播放量：{play}"
+                    ))
+        if b23_query and search_results:
+            await query.answer(results=search_results, cache_time=0)
+        else:
+            await query.answer(results=[
+                InlineQueryResultArticle(
+                    id="1",
+                    title="输入搜索内容",
+                    input_message_content=InputTextMessageContent(
+                        message_text="ta 好像想在 b 站搜索视频，但 ta 没有输入任何内容。",
+                        parse_mode=ParseMode.MARKDOWN
+                    ),
+                    description="请在 'b23' 后输入你想要搜索的内容。"
+                )
+            ], cache_time=0)
+        return
     if query_text.startswith("将军："):
         await query.answer(results=[
             InlineQueryResultArticle(

helpers/songs.py

@@ -1,6 +1,8 @@
 # 一个暂时性的办法用来存储歌曲信息
 import aiohttp
 
+from helpers.wbi import get_signed_params
+
 songs = {
     "将军的小曲,三太阳的小曲": "你若三冬 - 阿悠悠",
     "全斗焕的小曲,光州跑男的小曲,打成一片的小曲,无限制格斗的小曲,重拳的小曲,光州的小曲": "Shake and Sway",
@@ -37,18 +39,21 @@ async def fetch_from_b23_api(song_name):
             pass
 
         # 使用获取的 cookies 请求搜索 API
-        params = {'keyword': song_name}
+        params = {'keyword': song_name, 'search_type': 'video', 'duration': 1, 'order': 'click', 'tid': 3}
+        # 过一次 wbi 签名，防止被风控
+        signed_params = get_signed_params(params)
         async with session.get(
-                'https://api.bilibili.com/x/web-interface/search/all/v2',
-                params=params,
+                'https://api.bilibili.com/x/web-interface/wbi/search/type',
+                params=signed_params,
                 headers={
-                    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0"
+                    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0",
+                    "referer": "https://www.bilibili.com/",
                 }
         ) as response:
             resp = await response.json()
-    if resp and resp.get('data'):
+    if resp and resp.get('data').get('result'):
         # 假设我们只取第一个视频的结果
-        videos = next((item for item in resp['data']['result'] if item.get('result_type') == 'video'), None)
+        videos = next((item for item in resp['data']['result'] if item.get('type') == 'video'), None)
         first_result = videos['data'][0]
         title = first_result.get('title').replace('<em class="keyword">', '').replace('</em>', '') # 清理标题中的 HTML 标签
         link = first_result.get('arcurl')

helpers/wbi.py

@@ -0,0 +1,55 @@
+from functools import reduce
+from hashlib import md5
+import urllib.parse
+import time
+import requests
+
+mixinKeyEncTab = [
+    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
+    33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
+    61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
+    36, 20, 34, 44, 52
+]
+
+def get_mixin_key(orig: str):
+    '对 imgKey 和 subKey 进行字符顺序打乱编码'
+    return reduce(lambda s, i: s + orig[i], mixinKeyEncTab, '')[:32]
+
+def enc_wbi(params: dict, img_key: str, sub_key: str):
+    '为请求参数进行 wbi 签名'
+    mixin_key = get_mixin_key(img_key + sub_key)
+    curr_time = round(time.time())
+    params['wts'] = curr_time                                   # 添加 wts 字段
+    params = dict(sorted(params.items()))                       # 按照 key 重排参数
+    # 过滤 value 中的 "!'()*" 字符
+    params = {
+        k : ''.join(filter(lambda chr: chr not in "!'()*", str(v)))
+        for k, v
+        in params.items()
+    }
+    query = urllib.parse.urlencode(params)                      # 序列化参数
+    wbi_sign = md5((query + mixin_key).encode()).hexdigest()    # 计算 w_rid
+    params['w_rid'] = wbi_sign
+    return params
+
+def get_wbi_keys() -> tuple[str, str]:
+    '获取最新的 img_key 和 sub_key'
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0',
+        'Referer': 'https://www.bilibili.com/'
+    }
+    resp = requests.get('https://api.bilibili.com/x/web-interface/nav', headers=headers)
+    resp.raise_for_status()
+    json_content = resp.json()
+    img_url: str = json_content['data']['wbi_img']['img_url']
+    sub_url: str = json_content['data']['wbi_img']['sub_url']
+    img_key = img_url.rsplit('/', 1)[1].split('.')[0]
+    sub_key = sub_url.rsplit('/', 1)[1].split('.')[0]
+    return img_key, sub_key
+
+def get_signed_params(params):
+    img_key, sub_key = get_wbi_keys()
+
+    signed_params = enc_wbi(params=params, img_key=img_key, sub_key=sub_key)
+
+    return signed_params