Skip to main content

脚本功能编写——bilibili机器人为例

2024-6-11 ~ 2024-6-13

基础脚本，后续的修改和更新不写

功能设计​

首先我们先了解整个bilibili监控的框架，个人的想法是这样的

|--main.py
|---scripts/
|  |bilibili/
|     |bilibili.py
|     |update.py
|     |bilibili.db

其中bilibili.py中存放的是获取b站更新内容，以及执行用户的指令来添加与删除。json文件存储用户的相关信息，保存的是用户的QQ号、用户在哪个群中添加的，UID。即"QQID","GroupID","bilibiliUID"

首先安装bilibili-api库与其他依赖(接口可能改动，请及时更新最新版)

pip install bilibili-api-python asyncio importlib

接着是想要实现的功能，目前是想了五个功能

不同群聊中对某UID账号的监控
/bili add [uid] 添加指定UID进行监控
/bili del [uid] 删除UID监控
/bili info 查询个人在当前群里所添加的监控
/bili at [on/off] 开启/关闭更新艾特添加人，添加新关注后需要再次开启
/bili user [uid] 查询指定UID

init_db.py​

既然要使用不同群聊，那么就要涉及到存储，这里使用sqlite3本地来存储,init_db.py如下

import sqlite3
import os

DATABASE_PATH = os.path.join(os.path.dirname(__file__), 'bilibili.db')

def initialize_db():
    conn = sqlite3.connect(DATABASE_PATH)
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS monitoring (
        group_id TEXT,
        qq_id TEXT,
        bilibili_uid TEXT,
        at BOOLEAN,
        last_checked INTEGER,
        PRIMARY KEY (group_id, qq_id, bilibili_uid)
    )
    ''')
    conn.commit()
    conn.close()

if __name__ == '__main__':
    initialize_db()
    print("Database initialized.")

main.py​

接着呢要编写main.py，根据需求要实现以下操作：

  • 发送消息
  • 接收群消息事件
  • 执行bilibili.py
  • 执行update.py

其中，发送消息和接收群消息事件在之前就已经实现了，这里可以照搬

对于bilibili.py，查询bilibili-api库后发现需要异步操作

asyncio.run(handle_bili_command(message, user_id, group_id, send_group_message))

接着是最麻烦的update.py，需要长期刷新并检测更新，总之最后得到的main.py脚本如下

from flask import Flask, request
import requests
import asyncio
import logging
import time
import os
from concurrent.futures import ThreadPoolExecutor
import signal
import sys
from scripts.bilibili.bilibili import handle_bili_command
from scripts.bilibili.update import main as update_main

app = Flask(__name__)

# NapCatQQ API的基础URL
base_url = ""
access_token = ""

# 配置日志记录
log_path = os.path.join(os.path.dirname(__file__), 'log.txt')
logging.basicConfig(filename=log_path, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def send_group_message(group_id, message):
    url = f"{base_url}/send_group_msg"
    params = {
        "group_id": group_id,
        "message": message,
        "access_token": access_token
    }
    requests.get(url, params=params)

@app.route('/', methods=['POST'])
def receive_event():
    data = request.json
    logging.info(f"Received event: {data}")
    bili_data = ''
    # 检查是否是群消息事件
    if data['post_type'] == 'message' and data['message_type'] == 'group':
        group_id = str(data['group_id'])
        message_objects = data['message']
        message = ''.join([m['data']['text'] for m in message_objects if m['type'] == 'text'])
        user_id = str(data['user_id'])

        logging.info(f"Processed message: {message}")
        if message == '/help' or message == 'help' or message == '帮助':
            send_group_message(group_id, open('help.txt', 'r').read())
        elif message.startswith('/bili'):
            asyncio.run(handle_bili_command(message, user_id, group_id, send_group_message))
        elif(message == 'test1'):
            send_group_message(group_id,f'[CQ:json,data={bili_data}]')
    return "OK", 200

def run_flask():
    app.run(host='0.0.0.0', port=7777)

async def run_asyncio(send_group_message):
    loop = asyncio.get_event_loop()

    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(getattr(signal, signame), lambda: asyncio.ensure_future(shutdown(loop)))

    await update_main(send_group_message)

async def shutdown(loop):
    logging.info("Shutdown initiated.")
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    loop.stop()

if __name__ == '__main__':
    print('main:', time.time())
    with ThreadPoolExecutor() as executor:
        executor.submit(run_flask)

        try:
            asyncio.run(run_asyncio(send_group_message))
        except KeyboardInterrupt:
            pass

bilibili.py​

然后呢是对bilibili.py的编写，主要分为获取视频信息、添加用户、删除用户、查询用户、开关at、处理传入的命令

然后坑点在于-352报错：接口返回错误代码：-352，信息：风控校验失败。

这里是由于B站的检测，需要设备验证码否则会风控https://github.com/Nemo2011/bilibili-api/issues/691

然后在该库描述中写了如何获取所需信息https://nemo2011.github.io/bilibili-api/#/get-credential

使用方法为：

# 实例化 Credential 类
credential = Credential(sessdata=SESSDATA, bili_jct=BILI_JCT, buvid3=BUVID3)
# 实例化 Video 类
v = video.Video(bvid="BVxxxxxxxx", credential=credential)

同时在连接到bilibili.db时需要使用绝对路径，否则会在/app下进行操作，因为update和bilibili与main.py不在同一目录下

因此可以写出bilibili.py脚本

import sqlite3
import os
import asyncio
import datetime
from bilibili_api import user, Credential

DATABASE_PATH = os.path.join(os.path.dirname(__file__), 'bilibili.db')

# 连接到数据库
def connect_db():
    return sqlite3.connect(DATABASE_PATH)

# 获取用户视频信息
async def get_latest_video(uid):
    credential = Credential(
        sessdata="",
        bili_jct="", buvid3="",
        dedeuserid=""
    )
    u = user.User(uid, credential=credential)
    videos = await u.get_videos()
    return videos if videos else None

# 添加用户
async def add_user(qq_id, group_id, bilibili_uid):
    conn = connect_db()
    cursor = conn.cursor()

    videos = await get_latest_video(bilibili_uid)
    last_checked = videos['list']['vlist'][0]['created'] if videos else None

    cursor.execute('''
    INSERT OR IGNORE INTO monitoring (group_id, qq_id, bilibili_uid, at, last_checked)
    VALUES (?, ?, ?, ?, ?)
    ''', (group_id, qq_id, bilibili_uid, False, last_checked))

    conn.commit()
    conn.close()
    return True

# 删除用户
def remove_user(qq_id, group_id, bilibili_uid):
    conn = connect_db()
    cursor = conn.cursor()

    cursor.execute('''
    DELETE FROM monitoring
    WHERE group_id = ? AND qq_id = ? AND bilibili_uid = ?
    ''', (group_id, qq_id, bilibili_uid))

    conn.commit()
    conn.close()
    return True

# 查询用户添加的监控
def get_user_info(qq_id, group_id):
    conn = connect_db()
    cursor = conn.cursor()

    cursor.execute('''
    SELECT bilibili_uid, at FROM monitoring
    WHERE group_id = ? AND qq_id = ?
    ''', (group_id, qq_id))

    rows = cursor.fetchall()
    conn.close()

    if rows:
        uids = [row[0] for row in rows]
        at_status = "开启" if rows[0][1] else "关闭"
        return f"你在本群中添加的监控UID为: {', '.join(uids)}\n艾特功能: {at_status}"
    return "你在本群中没有添加任何监控"

# 设置艾特功能
def set_at_function(qq_id, group_id, on_off):
    conn = connect_db()
    cursor = conn.cursor()

    cursor.execute('''
    UPDATE monitoring
    SET at = ?
    WHERE group_id = ? AND qq_id = ?
    ''', (on_off.lower() == "on", group_id, qq_id))

    conn.commit()
    conn.close()
    return True

# 处理Bili命令
async def handle_bili_command(command, qq_id, group_id, send_group_message):
    parts = command.split()
    if len(parts) < 2:
        send_group_message(group_id, "命令格式错误,使用/help查看使用帮助")
        return

    action = parts[1]

    if action == "add" and len(parts) == 3:
        bilibili_uid = parts[2]
        if await add_user(qq_id, group_id, bilibili_uid):
            send_group_message(group_id, f"已添加B站UID: {bilibili_uid}")
        else:
            send_group_message(group_id, "你已经添加过该用户了哦")
    elif action == "del" and len(parts) == 3:
        bilibili_uid = parts[2]
        if remove_user(qq_id, group_id, bilibili_uid):
            send_group_message(group_id, f"已删除B站UID: {bilibili_uid}")
        else:
            send_group_message(group_id, "删除失败，你没有添加过该用户")
    elif action == "at" and len(parts) == 3:
        on_off = parts[2]
        if set_at_function(qq_id, group_id, on_off):
            send_group_message(group_id, f"艾特功能已{'开启' if on_off.lower() == 'on' else '关闭'}")
        else:
            send_group_message(group_id, "设置艾特功能失败，请先添加用户")
    elif action == "info" and len(parts) == 2:
        user_info = get_user_info(qq_id, group_id)
        send_group_message(group_id, user_info)
    elif action == "user" and len(parts) == 3:
        bilibili_uid = parts[2]
        videos = await get_latest_video(bilibili_uid)
        if videos:
            latest_video = videos['list']['vlist'][0]
            created_datetime = datetime.datetime.fromtimestamp(latest_video['created']+28800).strftime('%Y-%m-%d %H:%M:%S')
            video_pic = latest_video['pic']
            send_group_message(group_id,
                               f"用户：{latest_video['author']}\n最新视频: {latest_video['title']}\n链接: https://www.bilibili.com/video/{latest_video['bvid']}\n更新时间：{created_datetime}\n[CQ:image,file={video_pic}]")
        else:
            send_group_message(group_id, f"无法获取用户 {bilibili_uid} 信息")
    else:
        send_group_message(group_id, "命令格式错误")

update.py​

update就是检测更新，处理与上面差不多，不过由于需要一直刷新，这里与main.py有一定的，emmm，小冲突，总之已经解决了，脚本如下

import asyncio
import sqlite3
import os
import datetime
import logging
import time
from bilibili_api import user, Credential

DATABASE_PATH = os.path.join(os.path.dirname(__file__), 'bilibili.db')

log_path = os.path.join(os.path.dirname(__file__), 'log.txt')
logging.basicConfig(filename=log_path, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')  

def connect_db():
    return sqlite3.connect(DATABASE_PATH)

async def get_latest_video(uid):
    credential = Credential(
        sessdata="",
        bili_jct="", buvid3="",
        dedeuserid=""
    )
    u = user.User(uid, credential=credential)
    videos = await u.get_videos()
    return videos if videos else None

# 定期检查更新
async def check_updates(send_group_message):
    try:
        conn = connect_db()
        cursor = conn.cursor()

        cursor.execute('SELECT DISTINCT group_id, qq_id, bilibili_uid FROM monitoring')
        rows = cursor.fetchall()

        for group_id, qq_id, uid in rows:
            videos = await get_latest_video(uid)
            if videos:
                latest_video = videos['list']['vlist'][0]
                latest_timestamp = latest_video['created']
                cursor.execute('SELECT last_checked FROM monitoring WHERE group_id = ? AND qq_id = ? AND bilibili_uid = ?', (group_id, qq_id, uid))
                last_checked = cursor.fetchone()

                if last_checked is None or last_checked[0] is None or latest_timestamp > last_checked[0]:
                    at_status = cursor.execute('SELECT at FROM monitoring WHERE group_id = ? AND qq_id = ? AND bilibili_uid = ?', (group_id, qq_id, uid)).fetchone()[0]
                    at_text = f"[CQ:at,qq={qq_id}]" if at_status else ""
                    created_datetime = datetime.datetime.fromtimestamp(latest_timestamp+28800).strftime('%Y-%m-%d %H:%M:%S')
                    video_pic = latest_video['pic']
                    message = f"{at_text} 用户 {latest_video['author']}\n有一个新视频: {latest_video['title']}\n链接: https://www.bilibili.com/video/{latest_video['bvid']}\n更新时间：{created_datetime}\n[CQ:image,file={video_pic}]"
                    send_group_message(group_id, message)

                    cursor.execute('UPDATE monitoring SET last_checked = ? WHERE group_id = ? AND qq_id = ? AND bilibili_uid = ?', (latest_timestamp, group_id, qq_id, uid))

        conn.commit()
        conn.close()
    except :
        logging.info('wrong-update:', time.time())

async def main(send_group_message):
    while True:
        await check_updates(send_group_message)
        await asyncio.sleep(45)  # 45秒检测一次

if __name__ == "__main__":
    from scripts.bilibili.bilibili import send_group_message 
    asyncio.run(main(send_group_message))

help.txt​

这个就是自己写帮助手册啦，例如：

不同群聊中对某UID账号的监控
/bili add [uid] 添加指定UID进行监控
/bili del [uid] 删除UID监控
/bili info 查询个人在当前群里所添加的监控
/bili at [on/off] 开启/关闭更新艾特添加人，添加新关注后需要再次开启
/bili user [uid] 查询指定UID

然后运行python main.py即可，最终app的结构如下

.
├── help.txt
├── main.py
└── scripts
    └── bilibili
        ├── __pycache__
        │   ├── bilibili.cpython-310.pyc
        │   └── update.cpython-310.pyc
        ├── bilibili.db
        ├── bilibili.py
        ├── initialize_db.py
        ├── log.txt
        └── update.py