crawl-kuaishou/utils/alive.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   alive.py
@Time    :   2019/05/26 12:14:04
@Author  :   Liuyuqi 
@Version :   1.0
@Contact :   liuyuqi.gov@msn.cn
@License :   (C)Copyright 2019
@Desc    :   保持心跳

https://live.kuaishou.com/rest/wd/live/liveStream/myfollow
Request URL: https://live.kuaishou.com/rest/wd/live/liveStream/myfollow
Request Method: POST
Status Code: 200 OK
Remote Address: 127.0.0.1:1080
Referrer Policy: unsafe-url

Accept: application/json
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Connection: keep-alive
Content-Length: 0
Cookie: did=web_4095924dbbb4d12df7a87725a16104b6; didv=1558790317127; Hm_lvt_86a27b7db2c5c0ae37fee4a8a35033ee=1558790317; kuaishou.live.bfb1s=9b8f70844293bed778aade6e0a8f9942; clientid=3; client_key=65890b29; Hm_lpvt_86a27b7db2c5c0ae37fee4a8a35033ee=1558793089; userId=1365365171; userId=1365365171; kuaishou.live.web_st=ChRrdWFpc2hvdS5saXZlLndlYi5zdBKgAZFHcXmdSz0ZqgeR8-TTgoi9QQRV_2I1NT-eK4Wr46MXF-NpYe4Fo00puYYJRRdqB4nJLluPiMpO7YGwHtxLbnttpwFVvU0_aelP9U7kgg9s1BVI9U1dGQsZ0_JLfL8137gJKr4CqwJaq-9LBhM5a_ATTdc5Te1meUd1st2HGg-Dy_g18QwLY84WirWBIaKm8R7r6Dd3fFKYm-gdtJYmMukaEk2hY_LIikBot7IUVtJ3ydB6KCIgx-OxXNSqmT0GdbC5rtc3vchNEkamnb0pAQf6txb5g3YoBTAB; kuaishou.live.web_ph=b1dec6f6601bf17d2cb5b72665242eea8524
Host: live.kuaishou.com
kpf: PC_WEB
kpn: GAME_ZONE
Origin: https://live.kuaishou.com
Referer: https://live.kuaishou.com/u/fengyeYS
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36

res:

'''

import json
import queue
import threading
import time

import requests

# import pymysql
import utils.get_cookie as get_cookie

baseUrl = "https://live.kuaishou.com"
baseTokenUrl = "https://id.kuaishou.com"
# 房间号
# https://live.kuaishou.com/u/lol111112
room = baseUrl + "/u/Q1998227"
# baseUrl = "http://nginx.ok.yoqi.me:88"
# baseTokenUrl="http://nginx.ok.yoqi.me:88"

proxies = {"http": "http://120.26.110.59:8080"}
# sessionClient=requests.session()
sessionClients = []

baseHeader = {
    "Accept": "application/json",
    "Origin": baseUrl,
}
account_alive_count = 0
detail_url_queue = queue.Queue(maxsize=1000)


def login(userPhone, cookies, userAgent, threadid):
    """
    登录快手,执行一次即可。
    param :
    return:
    """
    url = baseTokenUrl + "/pass/kuaishou/login/passToken"
    mPartHeader = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Referer": room,
        "User-Agent": userAgent
    }
    mHeader = baseHeader.copy()
    mHeader.update(mPartHeader)
    proxies = []
    res = sessionClients[threadid].post(url=url, headers=mHeader, cookies=cookies, data="sid=kuaishou.live.web"
                                        # proxies=proxies
                                        )
    print(res.text)
    resText = json.loads(res.text)
    if resText["result"] != 1:
        print("thread {id}: User {userPhone}.passToken failure".format(id=threadid, userPhone=userPhone))
        return False
    else:
        url2 = baseUrl + "/user/login"
        mPartHeader = {
            "Content-Type": "application/json;charset=UTF-8",
            "Referer": room,
            "kpn": "GAME_ZONE",
            "kpf": "PC_WEB",
            "User-Agent": userAgent
        }
        mHeader2 = baseHeader.copy()
        mHeader2.update(mPartHeader)
        cookie_dict = {
            "kuaishou.live.web.at": resText["kuaishou.live.web.at"],
            "kuaishou.live.web_st": resText["kuaishou.live.web_st"],
        }
        cookies.update(cookie_dict)
        sessionClients[threadid].cookies.update(cookie_dict)
        mData = {"authToken": resText["kuaishou.live.web.at"], "sid": "kuaishou.live.web"}
        mData = json.dumps(mData).replace("", "")
        res2 = sessionClients[threadid].post(url=url2, headers=mHeader2, data=mData, cookies=cookies
                                             # proxies=proxies
                                             )
        # urllib.request.
        print(res2.text)
        if res2.text == "'Bad Request'":
            print("thread {id}: User {userPhone}.login failure".format(id=threadid, userPhone=userPhone))
            return False
        if json.loads(res2.text)["result"] != 1:
            print("thread {id}: User {userPhone}.login failure".format(id=threadid, userPhone=userPhone))
            return False
        else:
            print("thread {id}: User {userPhone}.login success".format(id=threadid, userPhone=userPhone))
            return True


def goToRoom(userPhone, cookies, userAgent, threadid):
    '''
    进入直播间
    :param userPhone:
    :param cookies:
    :param userAgent:
    :param threadid:
    :return:
    '''
    url = room
    mPartHeader = {
        "Content-Type": "application/json;charset=UTF-8",
        "Referer": room,
        "User-Agent": userAgent
    }
    mHeader = baseHeader.copy()
    mHeader.update(mPartHeader)
    # cookies = requests.utils.dict_from_cookiejar(session.cookies)
    # print(session)
    res = sessionClients[threadid].get(
        url=url, headers=mHeader,
        # proxies=proxies
    )
    print("进入直播间结果：" + res.text[0:200])


def firstcollect(userPhone, cookies, userAgent, threadid):
    url = baseUrl + ""
    '''
    session_id: yLZP77qeoyWVjfJV
    page_id: 1ReS9GIUNs-VyxWJ_1558897018534
    refer_page_id: eM4eCCCOFHxvnT70_1558896991974
    refer_show_id: 
    refer_url: https://live.kuaishou.com/u/Q1998227
    page_live_stream_id: 9PLaXTwT6vs
    url: https://live.kuaishou.com/u/Q1998227
    screen: 1366*768
    platform: Win32
    log_time: 1558897029472
    type: error
    errorDetail: Error: WebSocket timeout
    '''
    pass


def watchingFeed(userPhone, cookies, userAgent, threadid):
    url = baseUrl + "/wd/live/watchingFeed"
    data = {"liveStreamId": "9PLaXTwT6vs"}
    sessionClients[threadid].post(url=url, data=data)
    pass

def livefeed(userPhone, cookies, userAgent, threadid):
    '''

    :param userPhone:
    :param cookies:
    :param userAgent:
    :param threadid:
    :return:
    '''
    url = baseUrl + "/rest/wd/live/feed/v2"
    data = '{"liveStreamId":"9PLaXTwT6vs","pageId":"1ReS9GIUNs-VyxWJ_1558897018534","isHeartbeat":true}'
    data2='{"liveStreamId":"9PLaXTwT6vs","commentCursor":"syR2en5SdlY","giftCursor":"tQO_AlQBrO4","pageId":"1ReS9GIUNs-VyxWJ_1558897018534","isHeartbeat":false}'
    sessionClients[threadid].post(url=url, data=data)
    pass

def wssLogin():
    url = "wss://live-ws-pg-group3.kuaishou.com/websocket"

    pass


def myfollow(userPhone, cookies, userAgent, threadid):
    '''
    维持心跳，更新cookie。
    :param userPhone:
    :param cookies:
    :param userAgent:
    :param threadid:
    :return:
    '''
    url = baseUrl + "/rest/wd/live/liveStream/myfollow"
    mPartHeader = {
        "Content-Type": "application/json;charset=UTF-8",
        "Referer": room,
        "User-Agent": userAgent
    }
    mHeader = baseHeader.copy()
    mHeader.update(mPartHeader)
    # cookies = requests.utils.dict_from_cookiejar(session.cookies)
    # print(session)
    res = sessionClients[threadid].get(
        url=url, headers=mHeader, cookies=cookies,
        # proxies=proxies
    )
    print(res.text)

    if (not json.loads(res.text).get("result")) or json.loads(res.text)["result"] != 1:
        return False
    else:
        return True


def collect(userPhone, cookies, userAgent, threadid):
    {
        "kbytes_received": 5865,
        "kernel_version": "1.4.4",
        "session_id": "326fa539-4620-4892-a38f-42e0163c92d1",
        "tick_index": 18,
        "play_url": "https://tx-adaptive.pull.yximgs.com/gifshow/9PLaXTwT6vs.flv?txSecret=3d78c0935d21b9fc7a2cc869280bc1fc&txTime=5cec3114&stat=K5DJraklhpYTr46EqykzzDPJ3eereyUOENGTqe69syLjKCw6RVyHE3Gg0nSpfL/6&oidc=txhb",
        "domain": "tx-adaptive.pull.yximgs.com",
        "play_start_time": 1558896531540,
        "tick_start": 1558896701559,
        "tick_duration": 10002,
        "block_count": 0,
        "buffer_time": 0,
        "pause_duration": 0,
        "play_duration": 10002,
        "played_video_duration": 10002,
        "a_buf_len": 9880,
        "a_buf_slices": 1,
        "v_buf_len": 9903,
        "v_buf_slices": 1,
        "speed_chg_metric": {
            "0.9x": 259,
            "1.1x": 7214
        },
        "block_count_with_buffer": 0,
        "buffer_during_block": 0,
        "decoded_video_frames": 95,
        "dropped_video_frames": 4,
        "adaptive_multi_rate": 0,
        "play_pos": 158379,
        "v_data_rate": 7981,
        "a_data_rate": 114,
        "error_code": 0,
        "error_message": "",
        "v_bandwidth": 0,
        "buffer_queue_count": 0,
        "v_segment_count": 102,
        "a_segment_count": 102,
        "hidden": 0,
        "first_screen_total_duration": -1,
        "stream_duration": "null",
        "skip_cnt": 0,
        "skip_duration": 0,
        "demuxed_video_duration": 5916,
        "dropped_packet_duration": 0,
        "v_rec_delay": 9984,
        "v_render_delay": 22865,
        "a_render_delay": 22872,
        "stream_id": "9PLaXTwT6vs",
        "server_ip": "",
        "retry_cnt": 0,
        "cpu": 0
    }
    url = baseUrl + "/rest/wd/live/liveStream/myfollow"
    mPartHeader = {
        "Content-Type": "application/json;charset=UTF-8",
        "Referer": room,
        "User-Agent": userAgent
    }
    mHeader = baseHeader.copy()
    mHeader.update(mPartHeader)
    # cookies = requests.utils.dict_from_cookiejar(session.cookies)
    # print(session)
    res = sessionClients[threadid].get(
        url=url, headers=mHeader, cookies=cookies,
        # proxies=proxies
    )
    print(res.text)

    if (not json.loads(res.text).get("result")) or json.loads(res.text)["result"] != 1:
        return False
    else:
        return True


def run(userPhone, cookies, userAgent, threadid):
    '''
    执行多线程批量登录，维持在线
    :param userPhone: 用户手机号
    :param cookies: 用户初始化cookie
    :param userAgent: 用户agent
    :param threadid: 线程id
    :return:
    '''
    state = False  # 登录状态
    state = login(userPhone, cookies, userAgent, threadid)
    # 进入直播间
    goToRoom(userPhone, cookies, userAgent, threadid)
    while True:
        state = myfollow(userPhone, cookies, userAgent, threadid)
        if state == False:
            print("thread {id}: User {userPhone}. is alive".format(id=threadid, userPhone=userPhone))
            break
        else:
            print("thread {id}: User {userPhone}. is alive".format(id=threadid, userPhone=userPhone))
        time.sleep(10)


def logout():
    '''
    登出，如果需要暂停，则登出所有账号
    :return:
    '''
    pass


if __name__ == "__main__":
    start_time = time.time()
    mUser = get_cookie.getUser()
    account_alive_count = len(mUser["userPhone"])
    # print(account_alive_count)
    # exit()
    for i in range(account_alive_count):
        sessionClients.append(requests.session())  # 新建一个session
        threading.Thread(target=run,
                         args=(mUser["userPhone"][i], mUser["cookies"][i], mUser["userAgent"][i], i)).start()

    print("last time: {} s".format(time.time() - start_time))