Skip to main content

四:使用机器人管理群

main.py
import re
from functools import lru_cache

import httpx

from mirai import Mirai, WebSocketAdapter, FriendMessage, GroupMessage, At, Plain
from mirai.exceptions import ApiError

@lru_cache()
def get_access_token(api_key, secret_key):
url = 'https://aip.baidubce.com/oauth/2.0/token'
params = {
'grant_type': 'client_credentials',
'client_id': api_key,
'client_secret': secret_key,
}
resp = httpx.get(url, params=params)
resp.raise_for_status()
return resp.json()['access_token']

BD_API_KEY = '你的 API KEY'
SECRET_KEY = '你的 SECRET KEY'

async def text_censor(s: str) -> bool:
"""文本审核。返回 True 表示文本无问题。"""
access_token = get_access_token(BD_API_KEY, SECRET_KEY)
url = "https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined"
async with httpx.AsyncClient() as client:
try:
resp = await client.post(
url, params={'access_token': access_token}, data={'text': s}
)
resp.raise_for_status()
return resp.json()['conclusionType'] != 2
except (httpx.NetworkError, httpx.HTTPStatusError, KeyError):
return True

if __name__ == '__main__':
bot = Mirai(
qq=12345678, # 改成你的机器人的 QQ 号
adapter=WebSocketAdapter(
verify_key='yirimirai', host='localhost', port=8080
)
)

@bot.on(FriendMessage)
def on_friend_message(event: FriendMessage):
if str(event.message_chain) == '你好':
return bot.send(event, 'Hello, World!')

@bot.on(GroupMessage)
def on_group_message(event: GroupMessage):
if At(bot.qq) in event.message_chain:
return bot.send(event, [At(event.sender.id), '你在叫我吗?'])

@bot.on(GroupMessage)
async def text_censor_event(event: GroupMessage):
msg = "".join(map(str, event.message_chain[Plain]))
if not await text_censor(msg):
try:
await bot.recall(event.message_chain.message_id)
except ApiError:
pass
await bot.send(event, f'{event.sender.member_name},不能说奇怪的话哟~')
await bot.send_friend_message(12345678, f'检测到异常发言: {repr(event.sender)}') # 改成你的 QQ 号

bot.run()