简单的Bilibili直播弹幕发送实现

postLiveDanmaku.py

#!/usr/bin/env python
#coding:utf-8
# Author: Sg4Dylan --<sg4dylan#gmail.com>
# Update: 9/5/2017

import requests
import ast

def sessionPool(pool_connections=2, pool_maxsize=5, max_retries=5):
    session = requests.Session()
    adapter = requests.adapters.HTTPAdapter(
        pool_connections = pool_connections,
        pool_maxsize = pool_maxsize,
        max_retries = max_retries)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

def postLiveDanmaku(raw_cookie, room_id, message, session=None):

    danmaku = {}
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                      'AppleWebKit/537.36 (KHTML, like Gecko) '
                      'Chrome/53.0.2785.30 Safari/537.36',
        'Cookie': raw_cookie
    }
    api_url = "https://live.bilibili.com/msg/send"

    def typeCheck():
        if not raw_cookie or not room_id or not message:
            return False
        if not isinstance(raw_cookie, str):
            return False
        if not isinstance(room_id, str):
            return False
        if not isinstance(message, str):
            return False
        return True

    def cookieCheck():
        if not "SESSDATA" in raw_cookie:
            return False
        return True

    def genDanmaku():
        nonlocal danmaku
        danmaku_str = """{'color': '16777215',
                         'fontsize': '25',
                         'mode': '1',
                         'msg': '%s',
                         'rnd': '1504615858',
                         'roomid': '%s',}""" % \
                         (message, room_id)
        danmaku = ast.literal_eval(danmaku_str)
        print(danmaku)

    def sendDanmaku():
        if not typeCheck():
            return "Input error."
        if not cookieCheck():
            return "Cookie error."
        genDanmaku()
        if not session:
            sent_result = requests.post(api_url,
                                        data=danmaku,
                                        headers=headers)
            return sent_result.text
        sent_result = session.post(api_url,
                                   data=danmaku,
                                   headers=headers)
        return sent_result.text
    
    return sendDanmaku()

# Copy from Chrome -> Developer Tools -> Network -> Headers -> Request Headers -> Cookie
my_cookie = ""
# Enable sesion pool to send faster
danmaku_spool = sessionPool()
result = postLiveDanmaku(my_cookie, "456784", "nyaa", danmaku_spool)
print(result)