四、小電視自動抽獎
阿新 • • 發佈：2018-12-12
自動參加小電視抽獎 自動領取艦長、提督獎勵 執行一天大概能獲取1800包辣條 However. 每次凌晨B站後臺都會檢測賬號指令碼，一次搶太多很容易被封號一週。 可能是依據最大的連續領獎次數判斷的吧
一、抽獎流程
1. 分析頁面可知
每當有人贈送小電視之類的禮物時
系統就會發送廣播彈幕（SYS_MSG）
2. 點選廣播跳轉到抽獎房間
F12清空網路請求
接著點選抽獎
就找到了抽獎請求
POST https://api.live.bilibili.com/gift/v3/smalltv/join引數
響應
- 分析post引數 roomid系統廣播裡就有 raffleId是抽獎關鍵引數 type 固定為 Gift csrf_token 來自於cookie值bili_jct （可為空） visit_id （可為空）
- 重新整理抽獎頁面 F12過濾得到XHR 請求 逐個分析返回值和url請求 發現raffleId請求
GET https://api.live.bilibili.com/gift/v3/smalltv/check
data={
'roomid':roomid
}響應
分析結果
- 小電視抽獎和月色真美、摩天大樓 獲取raffleId的請求略有不同
https://api.live.bilibili.com/gift/v3/smalltv/check
https://api.live.bilibili.com/gift/v2/smalltv/check參加抽獎的請求相同 二、程式碼實現 api.py 新增Gift類
class Gift(): def check_smalltv(self,s,roomid,ver='v2'): url='https://api.live.bilibili.com/gift/'+ver+'/smalltv/check' data={'roomid': roomid} return ajax(s,url,'GET',data) def join_smalltv(self,s,roomid,raffleId,csrf_token,visit_id='', type = 'Gift'): url= 'https://api.live.bilibili.com/gift/v3/smalltv/join' data={ 'roomid': roomid, 'raffleId': raffleId, 'type': type, 'csrf_token': csrf_token, 'visit_id': visit_id } return ajax(s,url,'POST',data) def check_guard(self,s,roomid,_type='_guard'): url='https://api.live.bilibili.com/lottery/v1/Lottery/check'+_type+'?roomid='+str(roomid) return ajax(s,url,'GET'); def join_guard(self,s,roomid,id,csrf_token,type='guard',visit_id=None): url= 'https://api.live.bilibili.com/lottery/v2/Lottery/join' data= { 'roomid': roomid, 'id': id, 'type': type, 'csrf_token': csrf_token, 'visit_id': visit_id } return ajax(s,url,'POST',data)
server.py新增抽獎類
class Join: def __init__(self,Info,s): self.Info = Info self.check_roomids=[Info['roomid']] self.raffleId=0 self.hour_run=Timer(7.2e3,self.each_hour_run) self.s=s self.num=1 def join_smalltv(self,obj): if self.num==50: self.num=0 print('[跳過抽獎]['+obj['title']+']已跳過抽獎(roomid=' + str(obj['roomid'])+ ',raffleId=' + str(obj['raffleId'])+ ')') return self.num+=1 res=API.Gift.join_smalltv(self.s,obj['roomid'], obj['raffleId'],self.Info['csrf_token'],self.Info['visit_id']) code=res['code'] if code==0:Logger.info('[自動抽獎]['+obj['title']+']已參加抽獎(roomid=' + str(obj['roomid'])+ ',raffleId=' + str(obj['raffleId'])+ ')') elif code==400:Logger.warning('[自動抽獎][禮物抽獎]訪問被拒絕，您的帳號可能已經被封禁，已停止') elif code==65531:Logger.info('[自動抽獎][禮物抽獎]已參加抽獎(roomid=' + str(obj['roomid'])+ ',raffleId=' + str(obj['raffleId'])+ ')') else:Logger.error('[自動抽獎][禮物抽獎]已參加抽獎(roomid=' + str(obj['roomid'])+ ',raffleId=' + str(obj['raffleId'])+ ')'+res['msg']) def push_and_check_roomid(self,roomid): self.push_roomid(roomid) for roomid in self.check_roomids:self.check_and_join_guard(roomid) def push_roomid(self,roomid): if roomid in self.check_roomids:return self.check_roomids.append(roomid) def each_hour_run(self): res=API.Room.room_rank(self.s,"master_realtime_hour",'areaid_realtime_hour') for room in res['data']['list']:self.push_roomid(room['roomid']) res=API.Room.room_rank(self.s,"week_star_master",'week') for room in res['data']['list']:self.push_roomid(room['roomid']) def check_and_join_smalltv(self,roomid): res=API.Gift.check_smalltv(self.s,roomid,'v2') if res['code']==-400:return for data in res['data']: if data['raffleId']<=self.raffleId:continue self.raffleId=data['raffleId'] self.join_smalltv({'roomid':roomid,'raffleId':self.raffleId,'title':data['title']}) if len(res['data'])==0: res=API.Gift.check_smalltv(self.s,roomid,'v3') for data in res['data']['list']: if data['raffleId']<=self.raffleId:continue self.raffleId=data['raffleId'] self.join_smalltv({'roomid':roomid,'raffleId':self.raffleId,'title':data['title']}) self.push_and_check_roomid(roomid) def join_guard(self,obj): res=API.Gift.join_guard(self.s,obj['roomid'],obj['id'],self.Info['csrf_token']) print(obj['roomid']) Logger.info(res['data']['message']) def check_and_join_guard(self,roomid): res=API.Gift.check_guard(self.s,roomid) if len(res['data'])==0:return for guard in res['data']:self.join_guard({'id':guard['id'],'roomid':roomid}) res=API.Gift.check_guard(self.s,roomid,'') if len(res['data']['guard'])==0:return for guard in res['data']['guard']:self.join_guard({'id':guard['id'],'roomid':roomid})
utils.py新增日誌類
@singleton
class Logger:
def __init__(self):
handlers = {
logging.NOTSET: "logs/notset.logs",
logging.DEBUG: "logs/debug.logs",
logging.INFO: "logs/info.logs",
logging.WARNING: "logs/warning.logs",
logging.ERROR: "logs/error.logs",
logging.CRITICAL: "logs/critical.logs"
}
self.__loggers = {}
logLevels = handlers.keys()
fmt = logging.Formatter('%(asctime)s [%(levelname)s]: %(message)s')
for level in logLevels: #建立logger
logger = logging.getLogger(str(level))
logger.setLevel(level)
#建立hander用於寫日日誌檔案
log_path = os.path.abspath(handlers[level])
fh = logging.FileHandler(log_path) #定義日誌的輸出格式
fh.setFormatter(fmt)
fh.setLevel(level) #給logger新增hander
logger.addHandler(fh)
self.__loggers.update({level: logger})
def info(self, message):
self.__loggers[logging.INFO].info(message)
def error(self, message):
self.__loggers[logging.ERROR].error(message)
def warning(self, message):
self.__loggers[logging.WARNING].warning(message)
def debug(self, message):
self.__loggers[logging.DEBUG].debug(message)
def critical(self, message):
self.__loggers[logging.CRITICAL].critical(message)utils.py新增 儲存配置檔案函式
def save_py(moudle):
keys,py=dir(moudle),''
for key in keys:
if '__' in key:continue
py+=key+'='+json.dumps(getattr(moudle,key),indent=4, separators=(',', ':'))+'\n'
with open(moudle.__name__+".py",'w') as f:f.write(py)新增配置檔案 config.py
headers={
"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:62.0) Gecko/20100101 Firefox/62.0",
"Accept":"application/json, text/plain, */*",
"Accept-Language":"zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
"Accept-Encoding":"gzip, deflate, br",
"Referer":"https://live.bilibili.com/",
"Origin":"https://live.bilibili.com",
"Connection":"keep-alive"
}
msg_info={
"reject_msg":[
"SEND_GIFT",
"WELCOME",
"WELCOME_GUARD",
"COMBO_SEND",
"ENTRY_EFFECT",
"NOTICE_MSG",
"DANMU_MSG",
"ROOM_RANK",
"WISH_BOTTLE",
"COMBO_END"
],
"receive_msg":[
"SYS_MSG",
"SPECIAL_GIFT"
]
}
post_info={
"csrf_token":"",
"visit_id":"",
"roomid":392351,
"uid":380741418
}
ws_info={
"uid":380741418,
"roomid":392351,
"protover":1,
"platform":"web",
"clientver":"1.4.0"
}
主程式
import os
os.chdir('/data/jupyter/root/軟體/Bilibili')
from jquery import http,session
from servers import Join,Login
from DanmuWS import DanmuWebSocket
from config import *
from utils import save_py
import config
%matplotlib inline
#save_py(config)
s=session(headers,'config/2685054765.txt')
login=Login(s)
join=Join(post_info,s)
ws=None
while not login.isLogin():
login.get_vdcode()
login.loop_vdcode()
ws_info['uid']=post_info['uid']=login.info['uid']
def oncmd(data):
cmd=data["cmd"]
if cmd in msg_info['reject_msg']:return
if cmd=="SYS_MSG":
if "real_roomid" in data:join.check_and_join_smalltv(data["real_roomid"])
def onlogin(data):
print("login success")
def onreconnect(code,data=None):
global ws
ws=data['dws']
room_num=0
def onheartbeat(num):
if num>3:return
close()
if len(join.check_roomids)==0:
global room_num
join.each_hour_run()
room_num+=1
if room_num>3:return join.hour_run.cancel()
ws_info['roomid']=post_info['roomid']=join.check_roomids.pop()
main()
def main():
global ws
try:
print(ws_info['roomid'])
ws = DanmuWebSocket(ws_info,'wss://broadcastlv.chat.bilibili.com/sub')
ws.connect()
ws.bind(onreconnect,onlogin,onheartbeat,oncmd)
#ws.run_forever()
except:
close()
def close():
ws.close()
save_py(config)
s.save()
main()