forked from yjqiang/bili2.0
-
Notifications
You must be signed in to change notification settings - Fork 0
/
bili_console.py
234 lines (197 loc) · 8.53 KB
/
bili_console.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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
import bili_statistics
import printer
import asyncio
from typing import Optional
import notifier
from cmd import Cmd
import getopt
from tasks.utils import UtilsTask
from tasks.bili_console import (
PrintGiftbagsTask,
PrintMedalsTask,
PrintMainBiliDailyJobTask,
PrintLiveBiliDailyJobTask,
PrintMainBiliUserInfoTask,
PrintLiveBiliUserInfoTask,
PrintJudgeTask,
PrintCapsuleTask,
OpenCapsuleTask,
SendDanmuTask,
PrintUserStatusTask
)
from tasks.custom import SendLatiaoTask, BuyLatiaoTask, BuyMedalTask
class FuncCore:
def __init__(self, function, *args):
self.function = function
self.args = args
async def exec(self):
args = list(self.args)
# 递归
for i, arg in enumerate(args):
if isinstance(arg, FuncCore):
args[i] = await arg.exec()
if asyncio.iscoroutinefunction(self.function):
return await self.function(*args)
return self.function(*args)
def convert2int(orig) -> Optional[int]:
try:
return int(orig)
except (ValueError, TypeError):
return None
class BiliConsole(Cmd):
prompt = ''
def __init__(self, loop: asyncio.AbstractEventLoop, room_id, printer_danmu):
self.loop = loop
self.default_roomid = room_id
self._printer_danmu = printer_danmu
super().__init__()
@staticmethod
def guide_of_console():
print(' __________________ ')
print('| 欢迎使用本控制台 |')
print('| 1 输出本次统计数据 |')
print('| 2 查看目前拥有礼物的统计 |')
print('| 3 查看持有勋章状态 |')
print('| 4 检查主站今日任务的情况 |')
print('| 5 检查直播分站今日任务的情况 |')
print('| 6 获取主站个人的基本信息 |')
print('| 7 获取直播分站个人的基本信息 |')
print('| 8 检查风纪委今日自动投票的情况 |')
print('| 9 检查脚本判断的用户小黑屋情况 |')
print('|11 当前拥有的扭蛋币 |')
print('|12 开扭蛋币(一、十、百) |')
print('|13 直播间的长短号码的转化 |')
print('|14 发送弹幕 |')
print('|15 切换监听的直播间 |')
print('|16 控制弹幕的开关 |')
print('|21 赠指定总数的辣条到房间 |')
print('|22 银瓜子全部购买辣条并送到房间 |')
print('|23 购买勋章(使用银瓜子或者硬币)|')
print('  ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ̄ ')
def default(self, line):
self.guide_of_console()
def emptyline(self):
self.guide_of_console()
# pattern = '-u:-p:' u(user_id):0,1…;n(num);p(point)指roomid(烂命名因为-r不合适)
def parse(self, arg, pattern, default_u=0, set_roomid=False):
args = arg.split()
try:
opts, args = getopt.getopt(args, pattern)
except getopt.GetoptError:
return []
dict_results = {opt_name: opt_value for opt_name, opt_value in opts}
opt_names = pattern.split(':')[:-1]
results = []
for opt_name in opt_names:
opt_value = dict_results.get(opt_name)
if opt_name == '-u':
int_value = convert2int(opt_value)
if int_value is not None:
results.append(int_value)
else:
results.append(default_u)
# -2是一个灾难性的东西
# results.append(-2)
elif opt_name == '-n':
int_value = convert2int(opt_value)
if int_value is not None:
results.append(int_value)
else:
results.append(0)
elif opt_name == '-p':
int_value = convert2int(opt_value)
if int_value is not None:
room_id = int_value
else:
room_id = self.default_roomid
if set_roomid:
self.default_roomid = room_id
results.append(self.fetch_real_roomid(room_id))
else:
results.append(opt_value)
return results
def do_1(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(bili_statistics.print_statistics, user_id))
def do_2(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintGiftbagsTask, user_id))
def do_3(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintMedalsTask, user_id))
def do_4(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintMainBiliDailyJobTask, user_id))
def do_5(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintLiveBiliDailyJobTask, user_id))
def do_6(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintMainBiliUserInfoTask, user_id))
def do_7(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintLiveBiliUserInfoTask, user_id))
def do_8(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintJudgeTask, user_id))
def do_9(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintUserStatusTask, user_id))
def do_11(self, arg):
user_id, = self.parse(arg, '-u:')
self.exec_func_threads(
FuncCore(notifier.exec_task, PrintCapsuleTask, user_id))
def do_12(self, arg):
user_id, num_opened = self.parse(arg, '-u:-n:')
self.exec_func_threads(
FuncCore(notifier.exec_task, OpenCapsuleTask, user_id, num_opened))
def do_13(self, arg):
real_roomid, = self.parse(arg, '-p:')
self.exec_func_threads(
FuncCore(notifier.exec_func, UtilsTask.get_real_roomid, real_roomid))
def do_14(self, arg):
user_id, msg, real_roomid = self.parse(arg, '-u:-m:-p:')
self.exec_func_threads(
FuncCore(notifier.exec_task, SendDanmuTask, user_id, msg, real_roomid))
def do_15(self, arg):
real_roomid, = self.parse(arg, '-p:', set_roomid=True)
self.exec_func_threads(
FuncCore(self._printer_danmu.reset_roomid, real_roomid))
def do_16(self, arg):
ctrl, = self.parse(arg, '-c:')
if ctrl == 'T':
self.exec_func_threads(
FuncCore(printer.control_printer, True))
else:
self.exec_func_threads(
FuncCore(printer.control_printer, False))
def do_21(self, arg):
real_roomid, num_max = self.parse(arg, '-p:-n:')
self.exec_func_threads(
FuncCore(notifier.exec_task, SendLatiaoTask, real_roomid, num_max))
def do_22(self, arg):
real_roomid, num_wanted = self.parse(arg, '-p:-n:')
self.exec_func_threads(
FuncCore(notifier.exec_task, BuyLatiaoTask, real_roomid, num_wanted))
def do_23(self, arg):
user_id, coin_type, real_roomid = self.parse(arg, '-u:-c:-p:') # coin_type = 'silver' / 'metal'
self.exec_func_threads(
FuncCore(notifier.exec_task, BuyMedalTask, user_id, real_roomid, coin_type))
@staticmethod
def fetch_real_roomid(room_id):
return FuncCore(notifier.exec_func, UtilsTask.get_real_roomid, room_id)
# 直接执行,不需要user_id
def exec_func_threads(self, func_core: FuncCore):
asyncio.run_coroutine_threadsafe(self.exec_func(func_core), self.loop)
@staticmethod
async def exec_func(func_core: FuncCore):
await func_core.exec()