蜜桃av色欲a片精品一区,麻豆aⅴ精品无码一区二区,亚洲人成网站在线播放影院在线,亚洲 素人 字幕 在线 最新

  douyinliveMan.py

發(fā)布日期：2025/11/20 19:19:16      瀏覽量：

#!/usr/bin/python
# coding:utf-8

# @FileName:    liveMan.py
# @Time:        2024/1/2 21:51
# @Author:      bubu
# @Project:     douyinLiveWebFetcher

import codecs
import gzip
import hashlib
import random
import re
import ssl
import string
import subprocess
import urllib.parse
from contextlib import contextmanager
from unittest.mock import patch

import requests
import websocket
from py_mini_racer import MiniRacer

from protobuf.douyin import *


@contextmanager
def patched_popen_encoding(encoding=’utf-8’):
    original_popen_init = subprocess.Popen.__init__
    
    def new_popen_init(self, *args, **kwargs):
        kwargs[’encoding’] = encoding
        original_popen_init(self, *args, **kwargs)
    
    with patch.object(subprocess.Popen, ’__init__’, new_popen_init):
        yield


def generateSignature(wss, script_file=’sign.js’):
    """
    出現(xiàn)gbk編碼問題則修改 python模塊subprocess.py的源碼中Popen類的__init__函數(shù)參數(shù)encoding值為 "utf-8"
    """
    params = ("live_id,aid,version_code,webcast_sdk_version,"
              "room_id,sub_room_id,sub_channel_id,did_rule,"
              "user_unique_id,device_platform,device_type,ac,"
              "identity").split(’,’)
    wss_params = urllib.parse.urlparse(wss).query.split(’&’)
    wss_maps = {i.split(’=’)[0]: i.split("=")[-1] for i in wss_params}
    tpl_params = [f"{i}={wss_maps.get(i, ’’)}" for i in params]
    param = ’,’.join(tpl_params)
    md5 = hashlib.md5()
    md5.update(param.encode())
    md5_param = md5.hexdigest()
    
    with codecs.open(script_file, ’r’, encoding=’utf8’) as f:
        script = f.read()
    
    ctx = MiniRacer()
    ctx.eval(script)
    
    try:
        signature = ctx.call("get_sign", md5_param)
        return signature
    except Exception as e:
        print(e)
    
    # 以下代碼對(duì)應(yīng)js腳本為sign_v0.js
    # context = execjs.compile(script)
    # with patched_popen_encoding(encoding=’utf-8’):
    #     ret = context.call(’getSign’, {’X-MS-STUB’: md5_param})
    # return ret.get(’X-Bogus’)


def generateMsToken(length=107):
    """
    產(chǎn)生請(qǐng)求頭部cookie中的msToken字段，其實(shí)為隨機(jī)的107位字符
    :param length:字符位數(shù)
    :return:msToken
    """
    random_str = ’’
    base_str = string.ascii_letters + string.digits + ’=_’
    _len = len(base_str) - 1
    for _ in range(length):
        random_str += base_str[random.randint(0, _len)]
    return random_str


class DouyinLiveWebFetcher:
    
    def __init__(self, live_id):
        """
        直播間彈幕抓取對(duì)象
        :param live_id: 直播間的直播id，打開直播間web首頁的鏈接如：https://live.douyin.com/261378947940，
                        其中的261378947940即是live_id
        """
        self.__ttwid = None
        self.__room_id = None
        self.live_id = live_id
        self.live_url = "https://live.douyin.com/"
        self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " \
                          "Chrome/120.0.0.0 Safari/537.36"
        self.ball_game = None  # 添加一個(gè)屬性來存儲(chǔ)BallGame實(shí)例

    def set_ball_game(self, ball_game):
        self.ball_game = ball_game  # 設(shè)置BallGame實(shí)例

    def start(self):
        self._connectWebSocket()
    
    def stop(self):
        self.ws.close()
    
    @property
    def ttwid(self):
        """
        產(chǎn)生請(qǐng)求頭部cookie中的ttwid字段，訪問抖音網(wǎng)頁版直播間首頁可以獲取到響應(yīng)cookie中的ttwid
        :return: ttwid
        """
        if self.__ttwid:
            return self.__ttwid
        headers = {
            "User-Agent": self.user_agent,
        }
        try:
            response = requests.get(self.live_url, headers=headers)
            response.raise_for_status()
        except Exception as err:
            print("【X】Request the live url error: ", err)
        else:
            self.__ttwid = response.cookies.get(’ttwid’)
            return self.__ttwid
    
    @property
    def room_id(self):
        """
        根據(jù)直播間的地址獲取到真正的直播間roomId，有時(shí)會(huì)有錯(cuò)誤，可以重試請(qǐng)求解決
        :return:room_id
        """
        if self.__room_id:
            return self.__room_id
        url = self.live_url + self.live_id
        headers = {
            "User-Agent": self.user_agent,
            "cookie": f"ttwid={self.ttwid}&msToken={generateMsToken()}; __ac_nonce=0123407cc00a9e438deb4",
        }
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
        except Exception as err:
            print("【X】Request the live room url error: ", err)
        else:
            match = re.search(r’roomId\\":\\"(\d+)\\"’, response.text)
            if match is None or len(match.groups()) < 1:
                print("【X】No match found for roomId")
            
            self.__room_id = match.group(1)
            
            return self.__room_id

    def _connectWebSocket(self):
        """
        連接抖音直播間websocket服務(wù)器，請(qǐng)求直播間數(shù)據(jù)
        """
        wss = ("wss://webcast5-ws-web-hl.douyin.com/webcast/im/push/v2/?app_name=douyin_web"
               "&version_code=180800&webcast_sdk_version=1.0.14-beta.0"
               "&update_version_code=1.0.14-beta.0&compress=gzip&device_platform=web&cookie_enabled=true"
               "&screen_width=1536&screen_height=864&browser_language=zh-CN&browser_platform=Win32"
               "&browser_name=Mozilla"
               "&browser_version=5.0%20(Windows%20NT%2010.0;%20Win64;%20x64)%20AppleWebKit/537.36%20(KHTML,"
               "%20like%20Gecko)%20Chrome/126.0.0.0%20Safari/537.36"
               "&browser_online=true&tz_name=Asia/Shanghai"
               "&cursor=d-1_u-1_fh-7392091211001140287_t-1721106114633_r-1"
               f"&internal_ext=internal_src:dim|wss_push_room_id:{self.room_id}|wss_push_did:7319483754668557238"
               f"|first_req_ms:1721106114541|fetch_time:1721106114633|seq:1|wss_info:0-1721106114633-0-0|"
               f"wrds_v:7392094459690748497"
               f"&host=https://live.douyin.com&aid=6383&live_id=1&did_rule=3&endpoint=live_pc&support_wrds=1"
               f"&user_unique_id=7319483754668557238&im_path=/webcast/im/fetch/&identity=audience"
               f"&need_persist_msg_count=15&insert_task_id=&live_reason=&room_id={self.room_id}&heartbeatDuration=0")

        signature = generateSignature(wss)
        wss += f"&signature={signature}"

        headers = {
            "cookie": f"ttwid={self.ttwid}",
            ’user-agent’: self.user_agent,
        }
        self.ws = websocket.WebSocketApp(wss,
                                         header=headers,
                                         on_open=self._wsOnOpen,
                                         on_message=self._wsOnMessage,
                                         on_error=self._wsOnError,
                                         on_close=self._wsOnClose)
        try:
            # 禁用 SSL 驗(yàn)證
            self.ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
        except Exception:
            self.stop()
            raise
    
    def _wsOnOpen(self, ws):
        """
        連接建立成功
        """
        print("WebSocket connected.")
    
    def _wsOnMessage(self, ws, message):
        """
        接收到數(shù)據(jù)
        :param ws: websocket實(shí)例
        :param message: 數(shù)據(jù)
        """
        
        # 根據(jù)proto結(jié)構(gòu)體解析對(duì)象
        package = PushFrame().parse(message)
        response = Response().parse(gzip.decompress(package.payload))
        
        # 返回直播間服務(wù)器鏈接存活確認(rèn)消息，便于持續(xù)獲取數(shù)據(jù)
        if response.need_ack:
            ack = PushFrame(log_id=package.log_id,
                            payload_type=’ack’,
                            payload=response.internal_ext.encode(’utf-8’)
                            ).SerializeToString()
            ws.send(ack, websocket.ABNF.OPCODE_BINARY)
        
        # 根據(jù)消息類別解析消息體
        for msg in response.messages_list:
            method = msg.method
            try:
                {
                    ’WebcastChatMessage’: self._parseChatMsg,  # 聊天消息
                    ’WebcastGiftMessage’: self._parseGiftMsg,  # 禮物消息
                    ’WebcastLikeMessage’: self._parseLikeMsg,  # 點(diǎn)贊消息
                    ’WebcastMemberMessage’: self._parseMemberMsg,  # 進(jìn)入直播間消息
                    ’WebcastSocialMessage’: self._parseSocialMsg,  # 關(guān)注消息
                    ’WebcastRoomUserSeqMessage’: self._parseRoomUserSeqMsg,  # 直播間統(tǒng)計(jì)
                    ’WebcastFansclubMessage’: self._parseFansclubMsg,  # 粉絲團(tuán)消息
                    ’WebcastControlMessage’: self._parseControlMsg,  # 直播間狀態(tài)消息
                    ’WebcastEmojiChatMessage’: self._parseEmojiChatMsg,  # 聊天表情包消息
                    ’WebcastRoomStatsMessage’: self._parseRoomStatsMsg,  # 直播間統(tǒng)計(jì)信息
                    ’WebcastRoomMessage’: self._parseRoomMsg,  # 直播間信息
                    ’WebcastRoomRankMessage’: self._parseRankMsg,  # 直播間排行榜信息
                    ’WebcastRoomStreamAdaptationMessage’: self._parseRoomStreamAdaptationMsg,  # 直播間流配置
                }.get(method)(msg.payload)
            except Exception:
                pass
    
    def _wsOnError(self, ws, error):
        print("WebSocket error: ", error)
    
    def _wsOnClose(self, ws, *args):
        print("WebSocket connection closed.")
    
    def _parseChatMsg(self, payload):
        """聊天消息"""
        message = ChatMessage().parse(payload)
        user_name = message.user.nick_name
        user_id = message.user.id
        content = message.content
        print(f"【聊天msg】[{user_id}]{user_name}: {content}")
        if self.ball_game:
            if content == ’1’:
                self.ball_game.move_left(None)
            elif content == ’2’:
                self.ball_game.move_right(None)
    
    def _parseGiftMsg(self, payload):
        """禮物消息"""
        message = GiftMessage().parse(payload)
        user_name = message.user.nick_name
        gift_name = message.gift.name
        gift_cnt = message.combo_count
        print(f"【禮物msg】{user_name} 送出了 {gift_name}x{gift_cnt}")
    
    def _parseLikeMsg(self, payload):
        ’’’點(diǎn)贊消息’’’
        message = LikeMessage().parse(payload)
        user_name = message.user.nick_name
        count = message.count
        print(f"【點(diǎn)贊msg】{user_name} 點(diǎn)了{(lán)count}個(gè)贊")
    
    def _parseMemberMsg(self, payload):
        ’’’進(jìn)入直播間消息’’’
        message = MemberMessage().parse(payload)
        user_name = message.user.nick_name
        user_id = message.user.id
        gender = ["女", "男"][message.user.gender]
        print(f"【進(jìn)場(chǎng)msg】[{user_id}][{gender}]{user_name} 進(jìn)入了直播間")
    
    def _parseSocialMsg(self, payload):
        ’’’關(guān)注消息’’’
        message = SocialMessage().parse(payload)
        user_name = message.user.nick_name
        user_id = message.user.id
        print(f"【關(guān)注msg】[{user_id}]{user_name} 關(guān)注了主播")
    
    def _parseRoomUserSeqMsg(self, payload):
        ’’’直播間統(tǒng)計(jì)’’’
        message = RoomUserSeqMessage().parse(payload)
        current = message.total
        total = message.total_pv_for_anchor
        print(f"【統(tǒng)計(jì)msg】當(dāng)前觀看人數(shù): {current}, 累計(jì)觀看人數(shù): {total}")
    
    def _parseFansclubMsg(self, payload):
        ’’’粉絲團(tuán)消息’’’
        message = FansclubMessage().parse(payload)
        content = message.content
        print(f"【粉絲團(tuán)msg】 {content}")
    
    def _parseEmojiChatMsg(self, payload):
        ’’’聊天表情包消息’’’
        message = EmojiChatMessage().parse(payload)
        emoji_id = message.emoji_id
        user = message.user
        common = message.common
        default_content = message.default_content
        print(f"【聊天表情包id】 {emoji_id},user：{user},common:{common},default_content:{default_content}")
    
    def _parseRoomMsg(self, payload):
        message = RoomMessage().parse(payload)
        common = message.common
        room_id = common.room_id
        print(f"【直播間msg】直播間id:{room_id}")
    
    def _parseRoomStatsMsg(self, payload):
        message = RoomStatsMessage().parse(payload)
        display_long = message.display_long
        print(f"【直播間統(tǒng)計(jì)msg】{display_long}")
    
    def _parseRankMsg(self, payload):
        message = RoomRankMessage().parse(payload)
        ranks_list = message.ranks_list
        print(f"【直播間排行榜msg】{ranks_list}")
    
    def _parseControlMsg(self, payload):
        ’’’直播間狀態(tài)消息’’’
        message = ControlMessage().parse(payload)
        
        if message.status == 3:
            print("直播間已結(jié)束")
            self.stop()
    
    def _parseRoomStreamAdaptationMsg(self, payload):
        message = RoomStreamAdaptationMessage().parse(payload)
        adaptationType = message.adaptation_type
        print(f’直播間adaptation: {adaptationType}’)