微信直播间评论的抓取
想做个直播号，内容是和粉丝互动，根据粉丝的礼物，自动实现一些功能。比如一个爱心，水枪就射击一次。网上找了一份比较好的资料，做了一些修改 适配。
#!python
import requests
import time
import qrcode
import uuid
import json
import os
import base64
from threading import Thread
terminate_flag=False
uid = uuid.uuid4().hex
session = requests.Session()
finderUsername=''
authKey=""
X_Wechat_Uin=""
liveObjectId=""
liveId = ""
live_description = ""
liveCookies = ""
def setcoockis(response):
# 获取返回的cookie值
cookies = response.cookies
# 打印cookie值
for cookie in cookies:
print((cookie.name, cookie.value))
def generate_timestamp(length=10):
current_time = time.time()
if length == 10:
timestamp = int(current_time)
elif length == 13:
timestamp = int(current_time * 1000)
else:
raise ValueError("Invalid timestamp length. Must be 10 or 13.")
return str(timestamp)
#获取登录二维码
def getrcode():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/auth/auth_login_code"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/login-for-iframe?dark_mode=true&host_type=1",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": "0000000000",
"finger-print-device-id": uid,
"sec-ch-ua": '"Google Chrome";v="117", "Not;A=Brand";v="8", "Chromium";v="117"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"'
}
data = {
"timestamp": str(int(time.time() * 1000)), # 使用13位时间戳
"_log_finder_uin": "",
"_log_finder_id": "",
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
response = session.post(url, headers=headers, json=data)
#setcoockis(response)
redata = response.json()
# print(f'getrcode_errMesg:{redata["errMsg"]}')
if 'token' in redata['data']:
return redata['data']['token']
return ''
# 通过微信扫码登录
def request_qrcode(retoken):
url = f"https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/auth/auth_login_status?token={retoken}×tamp={generate_timestamp(13)}&_log_finder_uin=&_log_finder_id=&scene=7&reqScene=7"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/login-for-iframe?dark_mode=true&host_type=1",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": "0000000000",
"finger-print-device-id": uid,
"sec-ch-ua": '"Google Chrome";v="117", "Not;A=Brand";v="8", "Chromium";v="117"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"'
}
count = 0
status = 0
acctStatus = 0
while count < 200:
response = session.post(url, headers=headers)
#setcoockis(response)
rejson = response.json()
if rejson['errCode'] == 0:
# 处理返回的数据
# ...
status = rejson['data']['status']
acctStatus = rejson['data']['acctStatus']
if status == 0 and acctStatus == 0:
print('请使用微信扫码登录！')
elif status == 5 and acctStatus == 1:
print('已扫码请在手机上点击确认登录！')
elif status == 1 and acctStatus == 1:
print('已成功登录！')
break
elif status == 3 and acctStatus == 0:
print('已成功登录！')
break
else:
print('超时或网络异常已退出')
break
count += 1
time.sleep(1)
else:
print("请求失败")
break
if count >= 200:
print("二维码已超时")
if status == 1 & acctStatus == 1:
return True
return False
# 获取直播博主的信息，如finderUsername等
def auth_data():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/auth/auth_data"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/login",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": "0000000000",
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": "",
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
response = session.post(url, headers=headers, json=data)
#setcoockis(response)
rejson = response.json()
#print(response.json())
if rejson['errCode'] == 0:
global finderUsername
finderUsername = rejson['data']['finderUser']['finderUsername']
return True
else:
print(("登录异常："+rejson['errMsg']))
return False
# 获取帮助加载的参数信息，如authKey、X_Wechat_Uin
def helper_upload_params():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/helper/helper_upload_params"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/login",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": "0000000000",
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
response = session.post(url, headers=headers, json=data)
rejson=response.json()
if rejson['errCode'] == 0:
global authKey
authKey = rejson['data']['authKey']
global X_Wechat_Uin
X_Wechat_Uin = str(rejson['data']['uin'])
return True
else:
return False
# 判断直播间状态，获取直播间id、描述、直播间对象id
def check_live_status():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/check_live_status"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/home",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
#print("check_live_status")
try:
response = session.post(url, headers=headers, json=data, timeout=30)
rejson = response.json()
if rejson['errCode'] == 0:
global liveId
liveId = rejson['data']['liveId']
global live_description
live_description = rejson['data']['description']
global liveObjectId
liveObjectId = rejson['data']['liveObjectId']
#print("check_live_status end")
if rejson['data']['status'] == 1:
print(f'直播间【{live_description}】状态正常')
else:
print(f'直播间【{live_description}】状态={str(rejson["data"]["status"])}')
return True
else:
return False
except requests.exceptions.Timeout:
print("check_live_status请求超时了")
return True # 超时了？False
# 判断live-info此链接访问是否正常，如果正常为true，否则为false
def get_live_info():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/get_live_info"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/liveBuild",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"liveObjectId": liveObjectId,
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
#print('get_live_info')
try:
response = session.post(url, headers=headers, json=data, timeout=30)
rejson = response.json()
if rejson['errCode'] == 0:
return True
else:
print(f"get_live_info异常：{rejson}")
return False
except requests.exceptions.Timeout:
print("get_live_info请求超时了")
return True # 超时了？False
#获取msg消息刷新cookie
def join_live():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/join_live"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/liveBuild",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"objectId": liveObjectId,
"finderUsername": finderUsername,
"liveId": liveId,
"timestamp": str(int(time.time() * 1000)), # 使用当前的时间戳
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
response = session.post(url, headers=headers, json=data)
rejson = response.json()
if rejson['errCode'] == 0:
global liveCookies
liveCookies = rejson['data']['liveCookies']
return True
else:
print(f"join_live异常：{rejson}")
return False
#获取最新在线人员信息
def a_online_member():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/online_member"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/liveBuild",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"objectId": liveObjectId,
"finderUsername": finderUsername,
"clearRecentRewardHistory": True,
"liveId": liveId,
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
#print(f'online_member',data)
#print('online_member')
try:
response = session.post(url, headers=headers, json=data, timeout=30)
rejson = response.json()
#print(rejson)
if rejson['errCode'] == 0:
json_str = json.dumps(rejson)
# 将 JSON 字符串写入本地文件
with open("online_member.json", "w") as file:
file.write(json_str)
return True
else:
print(f"online_member异常：{rejson}")
return False
except requests.exceptions.Timeout:
print("online_member请求超时")
return True # 超时了？False
# 获取直播的cookies，并将获取的data传递给downmsg()
def msg():
global liveCookies
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/msg"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/liveBuild",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"objectId": liveObjectId,
"finderUsername": finderUsername,
"liveCookies": liveCookies,
"liveId": liveId,
"longpollingScene": 0,
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
try:
response = session.post(url, json=data, headers=headers, timeout=30)
rejson = response.json()
if rejson['errCode'] == 0:
#对本次的消息进行解析
liveCookies = rejson['data']['liveCookies']
#print("liveCookies",liveCookies)
downmsg(rejson['data'])
return True
else:
return False
except requests.exceptions.Timeout:
print("msg请求超时了")
return True
# 直播间激励
def reward_gains():
url = "https://channels.weixin.qq.com/cgi-bin/mmfinderassistant-bin/live/reward_gains"
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Content-Type": "application/json",
"Origin": "https://channels.weixin.qq.com",
"Referer": "https://channels.weixin.qq.com/platform/live/liveBuild",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
"X-WECHAT-UIN": X_Wechat_Uin,
"finger-print-device-id": uid,
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\""
}
data = {
"objectId": liveObjectId,
"finderUsername": finderUsername,
"clearRecentRewardHistory": True,
"liveId": liveId,
"timestamp": generate_timestamp(13),
"_log_finder_uin": "",
"_log_finder_id": finderUsername,
"rawKeyBuff": None,
"pluginSessionId": None,
"scene": 7,
"reqScene": 7
}
#print("reward_gains")
try:
response = session.post(url, headers=headers, json=data, timeout=30)
rejson=response.json()
if rejson['errCode'] == 0:
return True
else:
print('reward_gains_err:')
print(rejson)
return False
except requests.exceptions.Timeout:
print("reward_gains请求超时了")
return True
# 解析获取到的msg，将获取的msg写入指定文件内
def downmsg(rejson):
#######解析数据########
newmsg = []
for member in rejson['msgList']:
type = member['type']
# 系统通知，新人来了
if type == 10005:
nickname = member['nickname']
content = member['content']
newmsg.append({'nickname': nickname, 'msgType': type, 'content': content})
# 有人留言
if type == 1:
nickname = member['nickname']
content = member['content']
newmsg.append({'nickname': nickname, 'msgType': type, 'content': content})
for member in rejson['appMsgList']:
type = member['msgType']
# 礼物？？？
if type == 20009:
nickname = member['fromUserContact']['contact']['nickname']
base64_string = member['payload']
decoded_string = base64.b64decode(base64_string).decode('utf-8')
gift_info = json.loads(decoded_string)
newmsg.append({'nickname': nickname, 'msgType': type, 'gift_info': gift_info})
######################
# 将数据写入文件
# 检测当前目录下是否存在msglist目录，如果不存在则创建
# if not os.path.exists("msglist"):
# os.makedirs("msglist")
# 生成文件路径
# file_path = os.path.join("msglist", f"{generate_timestamp(13)}.json")
print(f"{'-'*26}newmsg{'-'*26}")
if len(newmsg) > 0:
# 将数据保存到文件中
# with open(file_path, "w") as file:
# json.dump(newmsg, file)
# print(f"newmsg: {newmsg}")
dnewmsg = newmsg[0]
for m in dnewmsg:
if 'gift_info' == m:
name = dnewmsg['gift_info']['reward_gift']['name']
price = dnewmsg['gift_info']['reward_gift']['price']
print(f'name:{name}, price: {price}')
else:
print(f'{m}:{dnewmsg[m]}, ')
# 根据获取的msg判断直播实时状态，并在指定时间间隔进行睡眠操作
def getmsg():
count = 0
global terminate_flag
while not terminate_flag:
count += 1
#print("当前时间：", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
if get_live_info() and msg():# and gift_enum_list():
time.sleep(0.1)
else:
# print('1 break')
break
if count % 3 == 0:
if check_live_status() and a_online_member() and reward_gains():
time.sleep(0.1)
else:
# print('2 break')
break
if __name__ == '__main__':
t1 = Thread(target=getmsg)
# 获取登录二维码 目的是：拿到token
retoken = getrcode()
print(retoken)
rehttp = f'https://channels.weixin.qq.com/mobile/confirm_login.html?token={retoken}'
# 展示获取的二维码
qr = qrcode.QRCode()
qr.border = 1
qr.add_data(rehttp)
qr.make()
print('请使用微信扫码登录！')
qr.print_tty()
# 获取二维码以及前期一系列准备工作。
if request_qrcode(retoken) and auth_data() and helper_upload_params() and check_live_status() and get_live_info() and join_live() and a_online_member():
print("加载成功，开启消息获取线程。获取实时弹幕消息。")
t1.start()
while True:
time.sleep(1)
user_input = eval(input("等待输入指令："))
print(("用户输入的指令是：" + user_input))
if user_input == "exit":
terminate_flag = True
t1.join()
break
