shop_live_scraper/scraper.py

#!/usr/bin/python3
# coding=utf-8
# -*- coding: utf-8 -*-

import random
import websocket
import redis
import requests
import threading
import sys
import json

from barrage import MessageDecode
from libs.proxy import Proxy

try:
    import thread
except ImportError:
    import _thread as thread

import time


def get_rand_str(len):
    charset = "bjectSymhasOwnProp0123456789ABCDEFGHIJKLMNQRTUVWXYZdfgiklquvxz"
    str = ''
    for _ in range(0, len):
        str += random.choice(charset)
    return str


def get_page_id():
    page_id = get_rand_str(16)
    page_id += "_"
    page_id += str(int(time.time() * 1000))
    return page_id


def get_websocket_info(live_stream_id, retry=0):
    param_json = {
        "operationName": "WebSocketInfoQuery",
        "variables": {
            "liveStreamId": live_stream_id
        },
        "query": "query WebSocketInfoQuery($liveStreamId: String) {\n  webSocketInfo(liveStreamId: $liveStreamId) {\n    token\n    webSocketUrls\n    __typename\n  }\n}\n"
    }

    user_agents = {
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3908.2 Safari/537.36',
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/82.0.4055.0 Safari/537.36',
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36',
    }

    cookie_list = [
        'web_8f5a71d99d502219d7ce1ca1ffec68cc; clientid=3; client_key=65890b29; Hm_lvt_86a27b7db2c5c0ae37fee4a8a35033ee=1595212029,1597053525; userId=1089887853; kuaishou.live.bfb1s=477cb0011daca84b36b3a4676857e5a1; didv=1603703368000; kpn=GAME_ZONE; kuaishou.live.web_st=ChRrdWFpc2hvdS5saXZlLndlYi5zdBKgATKQJmTG8mk6lH2pOJeVa-184Sq95qCmEB1lP_q3lItdj6a6gnYgCIcPkt1ZvnJlE3cAkBVIiK46xUEyJ1SZe2hatA7MGwchvIhNlvumZQWfwUHwDOKWRkpNOn-4c0wYnHmgituvkv2B2XJC1lnaLBJ57zZU6iKrtgtXyekfNrf37VW3n0cpGF8LfjxvfQibwYfxewf2uOhDQzKnCA-kLosaEhqtgBlt9k0TlHg5Y4Goo9D9kyIgPsuUj-GlXMBsywOViBgpl-tt7OVb051_AZAHST9ItzAoBTAB; kuaishou.live.web_ph=8f83f1ed0c54c288352d86b3a23fe996499e; userId=1089887853',
        'kuaishou.live.bfb1s=7206d814e5c089a58c910ed8bf52ace5; clientid=3; client_key=65890b29; sid=83cf0fb9ee15fde6b6987781; did=web_c280cf49619a100d8305b6f095ed5348; userId=1509370340; didv=1600053596890; WEBLOGGER_HTTP_SEQ_ID=18928; WEBLOGGER_INCREAMENT_ID_KEY=19033; Hm_lvt_86a27b7db2c5c0ae37fee4a8a35033ee=1601391911; Hm_lpvt_86a27b7db2c5c0ae37fee4a8a35033ee=1602073621; kpn=GAME_ZONE; kuaishou.live.web_st=ChRrdWFpc2hvdS5saXZlLndlYi5zdBKgAcSZ7bO7s4CskEWV8oXNhqBeAW7qebb6Gv8_pmwLpeDLgzlIl3Nwgd2qal6895okBZxhkFFXQLIRvSJss1ALy5H2k0B9JvRZc9RNm4FzKmZ4eWPx2B3f-RLyiz6CkNpEsjcgpCQ90pQyBJgc9nWyo3vJRJS9fl53t2hFW29cvcHJktjSh4VRl-JrZ6WI3r3Et1kGxyh0wGqW3Q3xWzVMmIkaEsvpGUru20c-iIt7T0W8MQrXwiIgMbhYaFs2XBsAXpVKQoE6xmrFlpUw3VfiZaze7OM6mkEoBTAB; kuaishou.live.web_ph=6248010c3bbf0923d41c9d3c231b2903007e; userId=1509370340',
        'did=web_03d225bf290f905c1d1a5b3810af3b07; didv=1607676125725; kuaishou.live.bfb1s=9b8f70844293bed778aade6e0a8f9942; clientid=3; client_key=65890b29; kpn=GAME_ZONE; Hm_lvt_86a27b7db2c5c0ae37fee4a8a35033ee=1607676148; Hm_lpvt_86a27b7db2c5c0ae37fee4a8a35033ee=1607676148; userId=2020851346; userId=2020851346; kuaishou.live.web_st=ChRrdWFpc2hvdS5saXZlLndlYi5zdBKgAR00HvTWTmSGTYkxB2Mny84aCe3Lu8qGgMyP7Bhp7Y8lHqkC4TPLNt8H71ZesLiX4hpu9eTzOCiCmzXp7bCHQxzYt6_POWrTVT9L-jgFL4ET3JmTqrjxudwbSZQP-icUJACHgXEpf9a_BXvPS1RrBnl82Gyg1LwLXxIo0uIdwhheHrAp3u4h4C6S5lqWvZ7aNX1nDk_5Jsyq8UqsuCXLNiMaEgCrAu8bFEUPixNgRvVq1Nb0ZSIgKyTMpEkPIRE47p3L_U0hfzZ2IIBI9YfiD5jI34L2zmAoBTAB; kuaishou.live.web_ph=2f965caadcde0572313c46a7f8de50843381'
    ]

    res = redis_connection.zrangebyscore(scrape_cookie_key, 0, '+inf', start=0, num=1, withscores=True)

    if len(res) == 0:
        cookie = 'kuaishou.live.bfb1s=9b8f70844293bed778aade6e0a8f9942; clientid=3; did=web_2ba8ea2ca07df85a9e193a75d443d128; client_key=65890b29; kpn=GAME_ZONE; userId=1509370340; userId=1509370340; kuaishou.live.web_st=ChRrdWFpc2hvdS5saXZlLndlYi5zdBKgAZIqgEWU0YHOwebl3XEBuu7odxIyvSspUq1v_kvLGCVI_eURJXwii1_AN0CwuxScDBKoK2y2HIghAZLxMVzJhR0wQ4IHJ0zi3TtLsnrbNlA9VnnSZpuHK2b1M2RxEiFEKBRTK0AwnRzN7UOvkKQTfOIxtALOFXqSPOzJ4R8ScI30o_CqmgWPbg7dw0Kt5sgpGFYcZMCA1vgkFhiph-O4cE8aEjGueioax06vmORaF3eBQr3cQSIgsBvfCInpIGIqrTgvvo3648essPww4pkcAKNHOgeNHjUoBTAB; kuaishou.live.web_ph=d77227e666b43027931968af443b14bd7b28'
    else:
        cookie, times = res[0]
        print('获取cookie ' + str(int(times)) + ' ' + cookie[0:30])

    headers = {
        'user-agent': user_agents.pop(),
        'Cookie': cookie,
        'Referer': 'https://live.kuaishou.com/u/3xse' + live_stream_id
    }

    proxy = Proxy.get()
    print('获取代理 ' + proxy)

    proxies = {
        "http": "http://" + proxy,
        "https": "http://" + proxy
    }

    try:
        # https://live.kuaishou.com/live_graphql
        r = requests.post(
            "https://live.kuaishou.com/m_graphql",
            json=param_json,
            headers=headers,
            proxies=proxies,
            timeout=30
        )

        webSocketInfo = r.json()['data']['webSocketInfo']
        token = webSocketInfo.get('token')

        if token is None:
            redis_connection.zincrby(scrape_cookie_key, 1, cookie)
            # Proxy.del_proxy(proxy)
            print('获取Token失败')
            # 重试10次还失败
            if retry > 10:
                return None, None

            time.sleep(2)
            retry = retry + 1
            return get_websocket_info(live_stream_id, retry)

        url = webSocketInfo['webSocketUrls'][0]

        return url, token

    except Exception as e:
        print('获取Token抛出异常：' + str(e))
        return get_websocket_info(live_stream_id, retry)


def on_message(ws, message):
    global redis_connection

    data = [m for m in message]

    messageDecode = MessageDecode(data)

    if messageDecode.decode():
        if messageDecode.message['payloadType'] == 310:  # 弹幕
            messageDecode.feed_decode()

            del messageDecode.message['compressionType'], messageDecode.message['payload']

            # if messageDecode.message.get('displayWatchingCount'):
            #     redis_connection.hset(stream_barrage_stat_hash_key, 'displayWatchingCount:' + ws.live_stream_id,
            #                           messageDecode.message['displayWatchingCount'])
                # print("观看人数：" + message.message['displayWatchingCount'])

            # if messageDecode.message.get('displayLikeCount'):
            #     redis_connection.hset(stream_barrage_stat_hash_key, 'displayLikeCount:' + ws.live_stream_id,
            #                           messageDecode.message['displayLikeCount'])
                # print("点赞人数：" + message.message['displayLikeCount'])

            messageDecode.message['live_stream_id'] = ws.live_stream_id
            redis_connection.lpush(stream_barrage_data_key, json.dumps(messageDecode.message))

            print(ws.live_stream_id + '_捕获弹幕信息')
            #
            # persistent_mysql = PersistentMysql()
            # persistent_mysql.live_stream_id = ws.live_stream_id
            # persistent_mysql.created_at = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
            # persistent_mysql.batch_insert(messageDecode.message)
            # print(ws.live_stream_id + "主动断开连接")
            # ws.close()
            # sys.stdout.flush()

            return True
        # elif messageDecode.message['payloadType'] == 381:  # PK
        #     messageDecode.pk_stat_decode()
        #     messageDecode.message['live_stream_id'] = ws.live_stream_id
        #     redis_connection.lpush('OnliveAdmin:barrageScraperPkStatData', json.dumps(messageDecode.message))
        #     print(messageDecode.message['pkPlayerStatistic'])
        #     sys.stdout.flush()
        #     return True

    # 如果有退出命令，则退出该线程
    if redis_connection.sismember(waiting_exit_stream_key, ws.live_stream_id):
        redis_connection.srem(waiting_exit_stream_key, ws.live_stream_id)
        print(ws.live_stream_id + "手动停止爬取")
        ws.close()

def on_error(ws, error):
    redis_connection.zrem(scraping_stream_key, ws.live_stream_id)
    print(ws.live_stream_id + error)
    sys.stdout.flush()


def on_close(ws):
    redis_connection.zrem(scraping_stream_key, ws.live_stream_id)
    redis_connection.lpush(live_ended_key, ws.live_stream_id)
    print(ws.live_stream_id + " 直播已结束 " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
    sys.stdout.flush()


def on_open(ws):
    part1 = [0x08, 0xC8, 0x01, 0x1A, 0xDC, 0x01, 0x0A, 0xAC, 0x01]  # 不变的头

    part2 = [ord(c) for c in ws.token]
    part3 = [0x12, 0x0B]  #

    part4 = [ord(c) for c in ws.live_stream_id]
    part5 = [0x3A, 0x1E]
    page_id = get_page_id()
    part6 = [ord(c) for c in page_id]

    d = part1 + part2 + part3 + part4 + part5 + part6
    ws.send(d, websocket.ABNF.OPCODE_BINARY)

    def run():
        while True:
            time.sleep(20)

            if ws.live_stream_id not in get_active_thread_set():
                print('停止心跳：' + ws.live_stream_id)
                sys.exit(0)

            print('心跳中：' + ws.live_stream_id)
            # 发送心跳-当前时间戳-毫秒
            head = [0x08, 0x01, 0x1A, 0x07, 0x08]
            timestamp = int(time.time() * 1000)
            time_arr = MessageDecode.hex_(timestamp)
            heartbeat = head + time_arr
            ws.send(heartbeat, websocket.ABNF.OPCODE_BINARY)

    thread.start_new_thread(run, ())


def scrape(live_stream_id):
    print("%s 开始执行" % live_stream_id)

    # 记录当前任务正在运行
    redis_connection.zadd(scraping_stream_key, {live_stream_id: int(time.time())})

    websocket.enableTrace(False)

    tokenInfo = redis_connection.hget(scrape_live_token_hash_key, live_stream_id)

    if (tokenInfo is None) or (tokenInfo == ''):

        url, token = get_websocket_info(live_stream_id)

        if url is None:
            return False

        redis_connection.hset(scrape_live_token_hash_key, live_stream_id, json.dumps({
            'url': url,
            'token': token,
        }))

        print(live_stream_id + " 获取Url成功：\n" + url)
        print(token)
    else:
        tokenInfoObj = json.loads(tokenInfo)
        url = tokenInfoObj.get('url')
        token = tokenInfoObj.get('token')

        print(live_stream_id + " 缓存获取Url成功：\n" + url)
        print(token)

    sys.stdout.flush()

    try:
        ws = websocket.WebSocketApp(
            url,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )

        ws.on_open = on_open
        ws.live_stream_id = live_stream_id
        ws.token = token
        ws.run_forever(skip_utf8_validation=True)
    except Exception as e:
        print(e)
        ws.close()


# 如果爬取的线程过多，则暂时等待
def check_threading_count():
    can_scrape_max_threading = redis_connection.get(can_scrape_max_threading_key)

    if can_scrape_max_threading is None:
        can_scrape_max_threading = 0

    # 如果爬取的线程过多，则暂时等待
    if threading.active_count() > int(can_scrape_max_threading):
        print('### 爬取直播弹幕任务过多，请等待...... ###')

        # 将标志位设为0， 暂时禁止添加爬取任务
        redis_connection.set(can_scrape_sign_key, 0)
        time.sleep(0.6)
        return False

    # 将标志位设为1， 允许添加爬取任务
    redis_connection.set(can_scrape_sign_key, 1)
    return True


# 清理脏数据
def clear_scraping_stream():
    # 获取Redis中记录的运行中线程列表
    active_list = redis_connection.zrange(scraping_stream_key, 0, -1)
    for live_stream_id in active_list:
        if live_stream_id not in get_active_thread_set():
            redis_connection.zrem(scraping_stream_key, live_stream_id)

    # 获取Redis中记录的待停止线程列表
    waiting_exit_stream_set = redis_connection.smembers(waiting_exit_stream_key)
    for waiting_exit_stream_id in waiting_exit_stream_set:
        if waiting_exit_stream_id not in active_list:
            redis_connection.srem(waiting_exit_stream_key, waiting_exit_stream_id)


# 清理Redis中记录的运行中线程列表
def clear_up_scraping_stream():
    # 获取Redis中记录的运行中线程列表
    # 如果运行中的线程列表中有数据，则先爬取
    active_list = redis_connection.zrange(scraping_stream_key, 0, -1)
    for live_stream_id in active_list:
        redis_connection.zrem(scraping_stream_key, live_stream_id)
        redis_connection.rpush(waiting_scrape_stream_key, live_stream_id)


def get_active_thread_set():
    # 待爬取的直播ID集合
    active_thread_set = set()
    for active_thread in threading.enumerate():  # 线程列表
        if active_thread.getName() != 'MainThread':
            active_thread_set.add(active_thread.getName())

    return active_thread_set


if __name__ == "__main__":

    print("主方法开始执行")

    # redis_connection = redis.StrictRedis(host='127.0.0.1', port=6379, decode_responses=True, db=1)
    redis_connection = redis.Redis(
        host='r-2ze28bdb7389a8a4.redis.rds.aliyuncs.com',
        port=6379,
        password='Zhuaduoduo2017',
        decode_responses=True
    )

    # 待爬取直播间ID List
    # waiting_scrape_stream_key = 'barrageScraperWaitingStreamIdList'
    waiting_scrape_stream_key = 'BrandLiveData.BarrageScraperWaitingStreamIdList'
    # 爬取中直播间ID Sort Set
    # scraping_stream_key = 'barrageScraperScrapingStreamIdSortSet'
    scraping_stream_key = 'BrandLiveData.BarrageScraperScrapingStreamIdSortSet'
    # cookie
    scrape_cookie_key = 'barrageScraperCookieSortSet'
    # token
    scrape_live_token_hash_key = 'BrandLiveData.BarrageScraperLiveTokenHash'
    # 待关闭直播间ID Set
    waiting_exit_stream_key = 'BrandLiveData.WaitingExitStreamIdSet'
    # 最大爬取线程数 string
    can_scrape_max_threading_key = 'BrandLiveData.BarrageScrapeMaxThreadingKey'
    # 是否可爬取标志位（当爬取线程数已经达到最大时置为0）
    can_scrape_sign_key = 'BrandLiveData.BarrageScrapeCanScrapeSignKey'
    # 记录获取的直播间观看人数和点赞数
    stream_barrage_stat_hash_key = 'BrandLiveData.BarrageScrapeLiveStreamStatHash'
    # 记录弹幕接口中获取到的数据
    stream_barrage_data_key = 'BrandLiveData.BarrageScraperBarrageData'
    # 记录直播结束的直播间信息
    live_ended_key = 'BrandLiveData.LiveEndedKey'
    # 先清理Redis中记录的运行中线程列表，一般是由于异常退出造成的
    clear_up_scraping_stream()

    while True:
        # 判断爬取线程是否超过设定的最大值
        if check_threading_count() is False:
            continue

        # 从 待爬取直播间ID 列表中获取一个 直播间ID
        stream_id = redis_connection.rpop(waiting_scrape_stream_key)

        # 如果没有待爬取的直播，则等一秒，循环
        if stream_id is None:
            # print("### 等待爬取直播弹幕...... ###")
            time.sleep(0.6)
            continue

        # 判断得到的直播间ID在爬取中
        # 判断线程中是否已经有该任务，有则跳过
        if stream_id in get_active_thread_set():
            print("%s 已经在执行" % stream_id)
            continue

        task = threading.Thread(target=scrape, args=(stream_id,), name=stream_id, daemon=False)
        task.start()  # 准备就绪,等待cpu执行

        # 清理脏数据
        clear_scraping_stream()
        time.sleep(1)