-
Notifications
You must be signed in to change notification settings - Fork 0
/
bilibiliClient.py
executable file
·152 lines (120 loc) · 6.14 KB
/
bilibiliClient.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
#encoding:utf-8
import asyncio
import http.cookies
import random
from typing import *
import var_set
import time
import post_dm
import aiohttp
import blivedm
import blivedm.models.web as web_models
import re
rule = r'ATA=(.*?);'
# 直播间ID的取值看直播间URL
TEST_ROOM_IDS = [
var_set.roomid,
]
# 这里填一个已登录账号的cookie。不填cookie也可以连接,但是收到弹幕的用户名会打码,UID会变成0
#SESSDATA = var_set.cookie[var_set.cookie.index('ATA=')+4:var_set.cookie.index('; bi')]
SESSDATA = re.findall(rule , var_set.cookie)[0]
session: Optional[aiohttp.ClientSession] = None
async def main():
init_session()
try:
await run_single_client()
await run_multi_clients()
finally:
await session.close()
def init_session():
cookies = http.cookies.SimpleCookie()
cookies['SESSDATA'] = SESSDATA
cookies['SESSDATA']['domain'] = 'bilibili.com'
global session
session = aiohttp.ClientSession()
session.cookie_jar.update_cookies(cookies)
async def run_single_client():
"""
演示监听一个直播间
"""
room_id = random.choice(TEST_ROOM_IDS)
client = blivedm.BLiveClient(room_id, session=session)
handler = MyHandler()
client.set_handler(handler)
client.start()
try:
# 演示5秒后停止
await asyncio.sleep(5)
client.stop()
await client.join()
finally:
await client.stop_and_close()
async def run_multi_clients():
"""
演示同时监听多个直播间
"""
clients = [blivedm.BLiveClient(room_id, session=session) for room_id in TEST_ROOM_IDS]
handler = MyHandler()
for client in clients:
client.set_handler(handler)
client.start()
try:
await asyncio.gather(*(
client.join() for client in clients
))
finally:
await asyncio.gather(*(
client.stop_and_close() for client in clients
))
class MyHandler(blivedm.BaseHandler):
# # 演示如何添加自定义回调
_CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
#async def __on_vip_enter(self,client: blivedm.BLiveClient,command):
# print(command['cmd'])
# if(command['cmd'] == 'LIVE'):
# print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' 正在直播中……')
# if(command['cmd'] == 'INTERACT_WORD'):
# print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' 听众['+command['data']['uname']+'] 进入直播……')
# post_dm.send_dm_long('欢迎【'+command['data']['uname']+'】进入直播间!')
#_CMD_CALLBACK_DICT['LIVE'] = __on_vip_enter
#_CMD_CALLBACK_DICT['INTERACT_WORD'] = __on_vip_enter
#_CMD_CALLBACK_DICT['ENTRY_EFFECT_MUST_RECEIVE'] = __on_vip_enter
#
# # 入场消息回调
def __interact_word_callback(self, client: blivedm.BLiveClient, command: dict):
#print(f"[{client.room_id}] INTERACT_WORD: self_type={type(self).__name__}, room_id={client.room_id},"
# f" uname={command['data']['uname']}")
#print(command['cmd'])
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' [log]听众['+command['data']['uname']+'] 进入直播……')
post_dm.send_dm_long('欢迎【'+command['data']['uname']+'】进入直播间!欢迎点播歌曲!')
if(post_dm.clean_files()): #检查空间是否在设定值以内,并自动删除多余视频缓存
post_dm.send_dm_long('树莓存储空间已爆炸,请联系up')
time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' [log]树莓存储空间已到设置值!'
return
_CMD_CALLBACK_DICT['INTERACT_WORD'] = __interact_word_callback # noqa
def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
#print(f'[{client.room_id}] 心跳')
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+f' [log]当前人气值:{message.popularity}')
def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage):
#print(f'[{client.room_id}][ {message.uid}] {message.uname}:{message.msg}')
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+f' [log]{message.uname}:{message.msg}')
post_dm.pick_msg(message.msg,message.uname,message.uid)
def _on_gift(self, client: blivedm.BLiveClient, message: web_models.GiftMessage):
#print(f'[{client.room_id}] {message.uname} 赠送{message.gift_name}x{message.num}'
# f' ({message.coin_type}瓜子x{message.total_coin})')
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+f' [log]{message.uname} 赠送{message.gift_name}x{message.num}'
f' ({message.coin_type}瓜子x{message.total_coin})')
post_dm.send_dm_long('感谢['+message.uname+']赠送的['+message.gift_name+']'+str(message.num)+'个!')
git_count = post_dm.get_coin(message.uname)
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' [log]观众['+message.uname+']已有'+str(git_count)+message.coin_type+'瓜子!')
post_dm.give_coin(message.uname,message.num*message.total_coin)
git_count = post_dm.get_coin(message.uname)
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+' [log]观众['+message.uname+']现有'+str(git_count)+message.coin_type+'瓜子!')
def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
#print(f'[{client.room_id}] {message.username} 购买{message.gift_name}')
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+f' [log]{message.username} 购买{message.gift_name}')
def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage):
#print(f'[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}:{message.message}')
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+f' [log]醒目留言 ¥{message.price} {message.uname}:{message.message}')
if __name__ == '__main__':
asyncio.run(main())