好友
阅读权限10
听众
最后登录1970-1-1
|
楼主
一叶青
发表于 2022-6-1 10:40
本帖最后由 一叶青 于 2022-6-1 10:46 编辑
说明：半成品，暂时不能发送弹幕。
功能很简单，获取did并注册，到页面取到liveStreamId，进入api拉取弹幕
感谢大神的分享，才有了本段代码，
did获取关键代码来自： ermao @ https://www.52pojie.cn/thread-1266439-1-1.html
弹幕链接api来自： Harl02 （用户为其它博客用户，规避版规，不加入引用地址）
1.程序启动即自动开始注册did,成功后，‘进房’按钮可点击
2.点‘进房’开始获取弹幕
3.小bug,刚进入，就点‘进房’，左下log会有一个3秒自动重试提醒，右列表不能获取房间名
4.‘断开’可以停止获取弹幕
5.‘发送’按钮不能发送弹幕
6.目前只自动获取王者区第一页主播列表，获取其它直播间，请自行改程序。成品用户，可以复制其它主播ID放到‘主播ID’内，再‘进房’
7.本机环境 win7_sp1_x64 python_3.8 截止发贴时间测试通过,截图如下
8.代码自取，随意更改，无需署名，无需通知
9.侵删
10.其它
4个文件都下载了，再解压，没有密码
快手直播弹幕.part1.rar
(2.9 MB, 下载次数: 354)
快手直播弹幕.part2.rar
(2.9 MB, 下载次数: 291)
快手直播弹幕.part3.rar
(2.9 MB, 下载次数: 245)
]
快手直播弹幕.part4.rar
(1.19 MB, 下载次数: 224)
代码分两段。
主代码
[Python] 纯文本查看 复制代码 # -*- coding:utf-8 -*-
# @FileName :IDEKuaishou.py
# [url=home.php?mod=space&uid=238618]@Time[/url] :2022/5/31 9:58
from tkinter import *
import tkinter as tk
from tkinter import ttk
import tkinter.font as tkFont
import time
import IDEKuaishou3_kuaishou
import threading
LOG_LINE_NUM = 0
CHAT_LINE_NUM = 0
class App_Kuaishou:
def __init__(self,root):
self.InitIDE()
self.id_list = []
self.name_list = []
self.kuaishou = IDEKuaishou3_kuaishou.Kuaishou(self)
self.Show_liveroom()
self.threadStatus = False
self.anchor = ''
def InitIDE(self):
# setting title
root.title("快手直播-王者荣耀版")
# setting window size
width = 1070
height = 491
screenwidth = root.winfo_screenwidth()
screenheight = root.winfo_screenheight()
alignstr = '%dx%d+%d+%d' % (width, height, (screenwidth - width) / 2, (screenheight - height) / 2)
root.geometry(alignstr)
root.resizable(width=False, height=False)
ft = tkFont.Font(family='宋体', size=12)
GLabel_liveroom = tk.Label(root,font = ft,fg ="#333333" ,justify = "center" , text = "直播间" , relief = "flat" )
GLabel_liveroom.place(x=0, y=0, width=51, height=33)
self.Combobox_liveroom = ttk.Combobox(root, font = ft, justify = "center" ,state = 'readonly' )
self.Combobox_liveroom.bind("<<ComboboxSelected>>", self.Show_liveroom)
self.Combobox_liveroom.place(x=60, y=0, width=249, height=34)
self.GButton_enterroom = tk.Button(root,font = ft,bg ="#efefef",fg ="#000000",text = "进房",justify = "center", relief = "groove",
state = DISABLED,command = self.GButton_enterroom_command )
self.GButton_enterroom.place(x=310, y=0, width=87, height=69)
GLabel_did = tk.Label(root,font = ft,fg ="#333333" ,justify = "center" , text = "did" , relief = "flat" )
GLabel_did.place(x=0, y=40, width=43, height=31)
ft_did = tkFont.Font(family='宋体', size=9)
self.GLineEdit_did = tk.Entry(root,borderwidth = "1px",font = ft_did,fg ="#333333" ,justify = "left", relief = "groove")
self.GLineEdit_did.place(x=60, y=40, width=248, height=30)
GLabel_anchor = tk.Label(root,font = ft,fg ="#333333" ,justify = "center" , text = "主播ID" , relief = "flat" )
GLabel_anchor.place(x=0, y=80, width=53, height=30)
self.GLineEdit_anchorid = tk.Entry(root,borderwidth = "1px",font = ft,fg ="#333333" ,justify = "left", relief = "groove")
self.GLineEdit_anchorid.place(x=60, y=80, width=148, height=30)
self.GButton_disconnet = tk.Button(root,font = ft,bg ="#efefef",fg ="#000000",text = "断开",justify = "center", relief = "groove",
command = self.GButton_disconnet_command )
self.GButton_disconnet.place(x=210, y=80, width=74, height=30)
ft_onlinenumber = tkFont.Font(family='宋体', size=9)
self.GLabel_onlinenumber = tk.Label(root,font = ft_onlinenumber,fg ="#333333" ,justify = "center" , text = "在线人数" , relief = "flat" )
self.GLabel_onlinenumber.place(x=300, y=80, width=99, height=31)
self.GLineEdit_inputmsg = tk.Entry(root,borderwidth = "1px",font = ft,fg ="#333333" ,text = "cookie",justify = "left", relief = "groove")
self.GLineEdit_inputmsg.place(x=0, y=120, width=305, height=30) #输入消息
self.GButton_sendmsg = tk.Button(root,font = ft,bg ="#efefef",fg ="#000000",text = "发送",justify = "center", relief = "groove",
command = self.GButton_sendmsg_command )
self.GButton_sendmsg.place(x=310, y=120, width=85, height=30)
ft_txt = tkFont.Font(family='宋体', size=10)
self.GText_chatlist = tk.Text(root,borderwidth = "1px",font = ft_txt,fg ="#333333", relief = "groove")
self.GText_chatlist.place(x=400, y=0, width=667, height=484) #聊天显示
self.GText_log = tk.Text(root,borderwidth = "1px",font = ft_txt,fg ="#333333", relief = "groove")
self.GText_log.place(x=0, y=160, width=398, height=326)
def GButton_enterroom_command(self):
print("GButton_enterroom")
self.write_log_to_Text("GButton_enterroom")
self.GButton_enterroom['state'] = DISABLED
self.threadStatus = True
self.write_chat_to_Text('~~~~~~~~~~~~~进入[%s]聊天室~~~~~~~~~~~开始获取聊天记录'%self.anchor)
obj1 = threading.Thread(target=self.kuaishou.EnterRoom, args=(self.GLineEdit_anchorid.get(),))
obj1.start()
def GButton_disconnet_command(self):
print("弹幕线程终止")
self.threadStatus = False
self.write_log_to_Text("弹幕线程终止")
self.GButton_enterroom['state'] = ACTIVE
def GButton_sendmsg_command(self):
print("GButton_sendmsg")
self.write_log_to_Text("GButton_sendmsg")
self.write_chat_to_Text("GButton_sendmsg")
def Show_liveroom(self,*args):
self.GLineEdit_anchorid.delete(0,END)
self.GLineEdit_anchorid.insert(0,self.id_list[self.Combobox_liveroom.current()])
self.anchor = self.name_list[self.Combobox_liveroom.current()]
# 聊天内容打印
def write_chat_to_Text(self, logmsg):
global CHAT_LINE_NUM
logmsg_in = str(logmsg) + "\n" # 换行
if CHAT_LINE_NUM <= 24:
self.GText_chatlist.insert(END, logmsg_in)
CHAT_LINE_NUM = CHAT_LINE_NUM + 1
ctemp = logmsg.split('：')[0]
cstart = str(CHAT_LINE_NUM)+'.0'
cend = str(CHAT_LINE_NUM)+'.'+str(len(ctemp))
self.GText_chatlist.tag_add('tag1',cstart,cend) # 先为te1创建tag1标签，确定范围
self.GText_chatlist.tag_config('tag1',background='lightyellow',foreground='red',font='黑体 12') # 再为tag1标签进行设置
cstart = str(CHAT_LINE_NUM)+'.'+str(len(logmsg)-19)
cend = str(CHAT_LINE_NUM)+'.end'
self.GText_chatlist.tag_add('tag2',cstart,cend) # 先为te1创建tag1标签，确定范围
self.GText_chatlist.tag_config('tag2',foreground='DarkSlateBlue',font='黑体 10') # 再为tag1标签进行设置
else:
self.GText_chatlist.delete(1.0, 2.0)
self.GText_chatlist.insert(END, logmsg_in)
ctemp = logmsg.split('：')[0]
cstart = str(CHAT_LINE_NUM)+'.0'
cend = str(CHAT_LINE_NUM)+'.'+str(len(ctemp))
self.GText_chatlist.tag_add('tag1',cstart,cend) # 先为te1创建tag1标签，确定范围
self.GText_chatlist.tag_config('tag1',background='lightyellow',foreground='red',font='黑体 12') # 再为tag1标签进行设置
cstart = str(CHAT_LINE_NUM)+'.'+str(len(logmsg)-19)
cend = str(CHAT_LINE_NUM)+'.end'
self.GText_chatlist.tag_add('tag2',cstart,cend) # 先为te1创建tag1标签，确定范围
self.GText_chatlist.tag_config('tag2',foreground='DarkSlateBlue',font='黑体 10') # 再为tag1标签进行设置
# 获取当前时间
def get_current_time(self):
current_time = time.strftime('%H:%M:%S', time.localtime(time.time()))
return current_time
# 日志动态打印
def write_log_to_Text(self, logmsg):
global LOG_LINE_NUM
current_time = self.get_current_time()
logmsg_in = str(current_time) + " " + str(logmsg) + "\n" # 换行
if LOG_LINE_NUM <= 20:
self.GText_log.insert(END, logmsg_in)
LOG_LINE_NUM = LOG_LINE_NUM + 1
else:
self.GText_log.delete(1.0, 2.0)
self.GText_log.insert(END, logmsg_in)
if __name__ == "__main__":
root = tk.Tk()
app = App_Kuaishou(root)
root.mainloop()
[Python] 纯文本查看 复制代码 # -*- coding:utf-8 -*-
# @FileName :IDEKuaishou3_kuaishou.py
# @Time :2022/5/31 9:58
import random
import time
import datetime
import json
import warnings
import requests
import re
import threading
class Kuaishou():
def __init__(self,app):
self.app = app
self.did = ''
self.userId = ''
self.roomurl = ''
self.username = ''
self.session = requests.Session()
self.session.headers.update({'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36'} )
self.liveStreamId =''
self.canRun = False
self.GetRoomlist('https://live.kuaishou.com/cate/SYXX/1001')
def GetRoomlist(self,url):
self.app.write_log_to_Text('开始初始化')
req = self._requests('get', url,decode_level=3)
req_text = req.text
idlt = re.findall(r'\"id\":\"User:.*?\",\"typename\":\"',req_text)
namelt = re.findall(r'jpg\",\"name\":\".*?\",\"',req_text)
name_list =[]
wclt = re.findall(r'\",\"watchingCount\":\".*?\"\},\"LiveInfo:',req_text)
qualitylt = re.findall(r'\.playUrls\.0\":\{\"quality\":\".*?\",\"__typename\":\"',req_text)
watching = 0
current = 0
for i in range(len(idlt)):
id = idlt[i].split(':')[2].split('\"')[0]
self.app.id_list.append(id)
name = namelt[i].split('\"')[4]
self.app.name_list.append(name)
wc = wclt[i].split('\"')[4]
playUrls = qualitylt[i].split('\":\"')[2].split('\",\"')[0].replace('\\u002F','/')
# print('%s %s %s %s'%(id,name,wc,playUrls))
watchingCount = 0
if wc.find('万') == -1 :
watchingCount = int(wc)
else:
wc = wc.split('万')[0]
watchingCount = float(wc) * 10000
if watchingCount > watching :
watching = watchingCount
self.userId = id
self.roomurl = playUrls
self.username = name
current= i
name_list.append('%s %d'%(name,watchingCount))
# end for
self.app.Combobox_liveroom['values']=name_list
self.app.Combobox_liveroom.current(current)
self.did = req.cookies.get('did')
self.client_key = req.cookies.get('client_key')
cookies = requests.utils.dict_from_cookiejar(req.cookies)
cookies['ksliveShowClipTip'] = 'true'
cookies['needLoginToWatchHD'] = '1'
cookies = requests.utils.dict_from_cookiejar(req.cookies)
self.session.cookies = requests.utils.cookiejar_from_dict(cookies)
misc2dic = {"common":{"identity_package":{"device_id":"web_f5be63a19dc2798f459de63919d7635f","global_id":""},"app_package":{"language":"zh-CN","platform":10,"container":"WEB","product_name":"KS_GAME_LIVE_PC"},"device_package":{"os_version":"NT 6.1","model":"Windows","ua":"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36"},"need_encrypt":'false',"network_package":{"type":3},"h5_extra_attr":"{\"sdk_name\":\"webLogger\",\"sdk_version\":\"3.9.49\",\"sdk_bundle\":\"log.common.js\",\"app_version_name\":\"\",\"host_product\":\"\",\"resolution\":\"1600x900\",\"screen_with\":1600,\"screen_height\":900,\"device_pixel_ratio\":1,\"domain\":\"https://live.kuaishou.com\"}","global_attr":"{}"},"logs":[{"client_timestamp":1666666666666,"client_increment_id":1234,"session_id":"1eb20f88-51ac-4ecf-8dc3-ace5aefcae4f","time_zone":"GMT+08:00","event_package":{"task_event":{"type":1,"status":0,"operation_type":1,"operation_direction":0,"session_id":"1eb20f88-51ac-4ecf-8dc3-ace5aefcae4f","url_package":{"page":"GAME_DETAL_PAGE","identity":"5316c78e-f0b6-4be2-a076-c8f9d11ebc0a","page_type":2,"params":"{\"game_id\":1001,\"game_name\":\"王者荣耀\"}"},"element_package":{}}}}]}
misc2dic['common']['identity_package']['device_id'] = self.did
misc2dic['logs'][0]['client_timestamp'] = int(round(time.time() * 1000))
misc2dic['logs'][0]['client_increment_id'] = random.randint(1000,9999)
url2 = 'https://log-sdk.ksapisrv.com/rest/wd/common/log/collect/misc2?v=3.9.49&kpn=KS_GAME_LIVE_PC'
reqjson = self._requests('post', url2,json=misc2dic,decode_level=2)
if(reqjson['result']==1):
url3 = 'https://live.kuaishou.com/u/'+self.userId
req = self._requests('get', url3)
if(self.username in req):
self.app.GButton_enterroom['state'] = 'active'
self.canRun = True
self.app.GLineEdit_did.delete(0,'end')
self.app.GLineEdit_did.insert(0,self.did)
self.app.write_log_to_Text('初始化成功')
return
self.app.write_log_to_Text('初始化失败')
def EnterRoom(self,url):
url = 'https://live.kuaishou.com/u/'+url
self.app.write_log_to_Text("开始连接%s"%url)
i_count =0
while i_count <= 3:
i_count += 1
try:
html = self._requests('get',url) # print(html)
match = re.search(r'\"liveStreamId\":\".*?\"',html)
if not match:
self.app.write_log_to_Text("解析失败~~~~~~~~~~3秒后自动重试")
time.sleep(3)
if(i_count==3):
self.app.write_log_to_Text("失败了~~~~~~~~~~请手动重试")
self.app.GButton_enterroom['state'] = 'active'
return
continue
self.liveStreamId= eval(match.group(0).split(':')[1])
self.canRun = True
self.app.write_log_to_Text("解析成功")
self.app.write_chat_to_Text("房间地址：%s直播地址：%s "%(url,self.roomurl))
break
except:
self.app.write_log_to_Text("解析出错")
self.app.GButton_enterroom['state'] = 'active'
return
url = 'http://livev.m.chenzhongtech.com/wap/live/feed?liveStreamId='+ self.liveStreamId
count = 0
# while count < 50 and self.canRun and self.app.threadStatus :
while self.canRun and self.app.threadStatus :
count += 1
req = self._requests('get',url,decode_level=3)
html = ''
if req.status_code == 200:
html = req.json()
else:
break
dict = json.loads(html)
self.app.GLabel_onlinenumber['text'] = '在线人数：%s'%dict['currentWatchingCount']
list = dict['liveStreamFeeds']
self.app.write_log_to_Text('%d获取聊天记录 %d 条'%(count,len(list)))
if list:
for dic in list:
times = datetime.datetime.fromtimestamp(dic['time']//1000)
# print(('%s ： %s [%s]')%(dic['author']['userName'],dic['content'],times))
self.app.write_chat_to_Text(('%s ： %s %s')%(dic['author']['userName'],dic['content'],times))
time.sleep(5)
self.app.GButton_enterroom['state'] = 'active'
def _requests(self, method, url, decode_level=1, retry=0, timeout=15, **kwargs):
if method in ["get", "post"]:
for _ in range(retry + 1):
try:
warnings.filterwarnings('ignore')
response = getattr(self.session, method)(url, timeout=timeout,verify=False, **kwargs)
return response.text if decode_level == 1 else response.json() if decode_level == 2 else response
except Exception as e:
print(e)
pass
return None
最后，再次感谢： ermao Harl02
|
免费评分
-
查看全部评分
|
[复制链接]
|
发表于 2022-6-1 18:20
|楼主
收藏
淘帖
有用
分享到朋友圈
冒昧的问一下 就是这种进直播间 只需要请求一次就行了 还是什么