yulinfeng000/blive

控制台报错

Closed this issue · 15 comments

FishZe commented 
await self._ws.send_bytes(packman.pack(heartbeat(), Operation.HEARTBEAT))
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/client_ws.py", line 156, in send_bytes
    await self._writer.send(data, binary=True, compress=compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 688, in send
    await self._send_frame(message, WSMsgType.BINARY, compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 601, in _send_frame
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport
Job "BLiver.heartbeat (trigger: interval[0:00:30], next run at: 2022-02-18 00:14:06 CST)" raised an exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/executors/base_py3.py", line 30, in run_coroutine_job
    retval = await job.func(*job.args, **job.kwargs)
  File "/usr/local/lib/python3.8/dist-packages/blive/framework.py", line 147, in heartbeat
    await self._ws.send_bytes(packman.pack(heartbeat(), Operation.HEARTBEAT))
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/client_ws.py", line 156, in send_bytes
    await self._writer.send(data, binary=True, compress=compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 688, in send
    await self._send_frame(message, WSMsgType.BINARY, compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 601, in _send_frame
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport
Job "BLiver.heartbeat (trigger: interval[0:00:30], next run at: 2022-02-18 00:14:36 CST)" raised an exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/executors/base_py3.py", line 30, in run_coroutine_job
    retval = await job.func(*job.args, **job.kwargs)
  File "/usr/local/lib/python3.8/dist-packages/blive/framework.py", line 147, in heartbeat
    await self._ws.send_bytes(packman.pack(heartbeat(), Operation.HEARTBEAT))
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/client_ws.py", line 156, in send_bytes
    await self._writer.send(data, binary=True, compress=compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 688, in send
    await self._send_frame(message, WSMsgType.BINARY, compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 601, in _send_frame
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport
Job "BLiver.heartbeat (trigger: interval[0:00:30], next run at: 2022-02-18 00:15:06 CST)" raised an exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/executors/base_py3.py", line 30, in run_coroutine_job
    retval = await job.func(*job.args, **job.kwargs)
  File "/usr/local/lib/python3.8/dist-packages/blive/framework.py", line 147, in heartbeat
    await self._ws.send_bytes(packman.pack(heartbeat(), Operation.HEARTBEAT))
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/client_ws.py", line 156, in send_bytes
    await self._writer.send(data, binary=True, compress=compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 688, in send
    await self._send_frame(message, WSMsgType.BINARY, compress)
  File "/usr/local/lib/python3.8/dist-packages/aiohttp/http_websocket.py", line 601, in _send_frame
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport
^CTraceback (most recent call last):
  File "main.py", line 44, in <module>
    app.run()
  File "/usr/local/lib/python3.8/dist-packages/blive/framework.py", line 176, in run
    loop.run_forever()
  File "/usr/lib/python3.8/asyncio/base_events.py", line 570, in run_forever
    self._run_once()
  File "/usr/lib/python3.8/asyncio/base_events.py", line 1823, in _run_once
    event_list = self._selector.select(timeout)
  File "/usr/lib/python3.8/selectors.py", line 468, in select
    fd_event_list = self._selector.poll(timeout, max_ev)
FishZe commented

可能是由于直播间一段时间没有弹幕，导致超时

yulinfeng000 commented

可能B站因为直播间长时间没有弹幕关闭ws连接了

yulinfeng000 commented

请问能提供复现的条件吗

FishZe commented

使用的app.py，是一个小主播，除了直播的时候没有弹幕。
我使用screen挂在后台全天监测保存弹幕，在非直播时间报错，弹幕姬停止运行并不在重启
可以增加timeout时间，或者通过try不显示报错并自动重连？

yulinfeng000 commented

好的，这个需求我收到了,会跟进的，如果你有什么想法也欢迎提pr

yulinfeng000 commented

已增加断线重连机制，如果没有其他的问题的话我会在一周后关闭这个issue

FishZe commented

请教一下，我如果想同时监听好几个直播间弹幕的话，把app.py封装成class，类似这样可以吗？

from blive import BLiver, Events, BLiverCtx
from blive.msg import (
    DanMuMsg,
    HotRankChangeV2Msg,
    InteractWordMsg,
    SendGiftMsg,
    SuperChatMsg,
)


class bullet:
    app = BLiver(255)
    
    def __init__(self, uid):
        self.uid = uid
        self.app = BLiver(uid)
        print(uid)

    @app.on(Events.DANMU_MSG)
    async def listen(ctx: BLiverCtx):
        danmu = DanMuMsg(ctx.body)
        print(
            f"\n{danmu.sender['name']}({danmu.sender['medal']['medal_name']}:{danmu.sender['medal']['medal_level']}): \"{danmu.content}\"\n"
        )
    
    
    @app.on(Events.INTERACT_WORD)
    async def listen_join(ctx: BLiverCtx):
        join = InteractWordMsg(ctx.body)
        print(
            "欢迎",
            f"{join.user['name']} ({join.user['medal']['medal_name']}:{join.user['medal']['medal_level']})",
            "进入直播间",
        )
    
    
    @app.on(Events.SUPER_CHAT_MESSAGE)
    async def listen_sc(ctx: BLiverCtx):
        msg = SuperChatMsg(ctx.body)
        print(msg.sender)
        print(msg.content)
        print(msg.start_time)
        print(msg.time)
        print(msg.price)
    
    
    @app.on(Events.SEND_GIFT)
    async def listen_gift(ctx: BLiverCtx):
        msg = SendGiftMsg(ctx.body)
        print(f"{msg.sender['name']} 送出 {msg.gift['gift_name']}")
    
    
    @app.on(Events.HOT_RANK_CHANGED_V2)
    async def hot(ctx: BLiverCtx):
        msg = HotRankChangeV2Msg(ctx.body)
        print(
            f"恭喜 {ctx.bliver.uname} 在 {msg.area_name} 区 的 {msg.rank_desc} 榜单中获得第 {msg.rank} 名"
        )
    
    
x = bullet(5169315)
x.app.run()

但是这样似乎无法正常运行，可能是在初始化bliver时就已经开始监听了，无法在__init__中修改，有什么好的办法吗？

谢谢~

yulinfeng000 commented

给一个例子吧，其实有设计了这样的功能但是没有写在README里，如果你用着不错我会考虑写进README，例子里实现了同时监听2个直播间的弹幕，更多直播间可以类推，如果遇到性能问题建议安装uvloop库替换python的原生事件循环库

import asyncio
from blive import BLiver,Events,BLiverCtx
from blive.msg import DanMuMsg


# 定义弹幕事件handler
async def listen(ctx: BLiverCtx):
    danmu = DanMuMsg(ctx.body)
    print(
        f'\n{danmu.sender.name} ({danmu.sender.medal.medal_name}:{danmu.sender.medal.medal_level}): "{danmu.content}"\n'
    )


async def main():
    # 两个直播间
    ke = BLiver(605)
    azi = BLiver(510)

    # 注册handler
    ke.register_handler(Events.DANMU_MSG, listen)
    azi.register_handler(Events.DANMU_MSG,listen)

    # 以异步task的形式运行
    task1 = ke.run_as_task()
    task2 = azi.run_as_task()

    # await 两个任务
    await asyncio.gather(*[task1,task2])

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
FishZe commented

试了一下，感觉不错，建议加入到README
我还在尝试监听1、2号直播间的同时，加入监听3号直播间，或不影响1号直播间弹幕的同时，关闭2号直播间监听，但是我只能关闭重启，会导致其他直播间弹幕丢失，有什么好办法吗？

yulinfeng000 commented

以run_as_task的形式运行后，一个BLiver应用的生命周期都可以以 异步任务( asyncio.Task ) 的形式管理了。关于这方面请学习asyncio库和python异步io编程相关的知识，原谅我很难在issue里三言两语讲清楚，再给一个例子以供参考，例子实现了在运行中停止一个BLiver实例和启动一个BLiver实例。

我的想法是本项目应该尽可能简单（simple），灵活而强大。而不是臃肿和复杂，并且提供asyncio的Task对象很容易和其他asyncio生态融合（ fastapi，sanic ... ），你可以写出任何形式的代码，所以本项目不会提供更高级的封装类。

def main():
    async def stop_bliver(task):
         # 睡10秒后停止azi直播间
        await asyncio.sleep(10)
        print("停止一个直播间")
        task.cancel()

    async def stop_bliver_38():
        await asyncio.sleep(10)
        print("停止azi直播间")
        for task in asyncio.all_tasks():
            if task.get_name() == "azi":
                task.cancel()
    
    async def new_bliver():
        # 15秒后创建一个新直播间监听
        await asyncio.sleep(15)
        print("创建一个新直播间监听")
        dys = BLiver(7777)
        dys.run_as_task()
        


    loop = asyncio.get_event_loop()
    # 两个直播间
    ke = BLiver(605)
    azi = BLiver(510)

    # 注册handler
    ke.register_handler(Events.DANMU_MSG, listen)
    azi.register_handler(Events.DANMU_MSG, listen)

    
    task1 = ke.run_as_task()
    task2 = azi.run_as_task()

    # asyncio.Task.set_name 方法只在python3.8以上才有！！！想用名字管理task建议升级到python3.8
    # task1.set_name("ke")
    # task2.set_name("azi")

    loop.create_task(stop_bliver(task1))
    # loop.create_task(stop_bliver_38())
    loop.create_task(new_bliver())
    
    #只要run_forever，也可以不await，因为await是为了达到程序表现像同步代码的效果
    loop.run_forever() 


if __name__ == "__main__":
    main()
yulinfeng000 commented

和fastapi整合的例子

# main.py

from fastapi import FastAPI
from blive import BLiver,Events
from blive.msg import DanMuMsg

app = FastAPI()

BLIVER_POOL = {}

def create_bliver(roomid):
    # 定义弹幕事件handler
    async def listen(ctx):
        danmu = DanMuMsg(ctx.body)
        print(
            f'\n{danmu.sender.name} ({danmu.sender.medal.medal_name}:{danmu.sender.medal.medal_level}): "{danmu.content}"\n'
        )
    b = BLiver(roomid)
    b.register_handler(Events.DANMU_MSG,listen)
    return b


@app.get("/create")
async def create_new_bliver(roomid:int):
    room = BLIVER_POOL.get(roomid,None)
    if not room:
        b = create_bliver(roomid)
        BLIVER_POOL[roomid] = b.run_as_task()
    return {"msg":"创建一个新直播间弹幕监听成功"}


@app.get("/del")
async def rm_bliver(roomid:int):
    room = BLIVER_POOL.get(roomid,None)
    if room:
        room.cancel()
    return {"msg":"移除直播间弹幕监听成功"}


@app.get("/show")
async def show():
    return list(BLIVER_POOL.keys())

实现了请求 GET http://ip:port/create?roomid=xxx 创建新监听, 请求 GET http://ip:port/del?roomid=xxx 移除监听 和 GET http://ip:port/show 展示当前所有监听的房间

FishZe commented

和fastapi的这个脚本，连续开好几天之后，似乎create会失效，目前不知道原因

yulinfeng000 commented

新开个issue吧，顺便附上log之类的（如果有的话），感觉很难复现

FishZe commented

之前说的，create失效的原因找到了
我直接复制的readme中与fastapi结合部分的代码，其中，del部分没有从BLIVER_POOL中pop，再次开启监听时，程序会认为已开启，导致开启失败。

我简单修改了一下

# 创建直播间弹幕监听
@app.get("/create")
async def create_new_bliver(roomid: int):
    room = BLIVER_POOL.get(roomid, None)
    if room == None:
        b = create_bliver(roomid)
        BLIVER_POOL[roomid] = b.run_as_task() 
        log("INFO", f"开启直播间监听: {roomid}")
        return {"msg": f"创建直播间 {roomid} 弹幕监听成功", "status": True}
    return {"msg": f"直播间 {roomid} 已在监听中", "status": False}

# 删除直播间弹幕监听
@app.get("/del")
async def rm_bliver(roomid: int):
    global BLIVER_POOL
    room = BLIVER_POOL.get(roomid, None)
    if room != None:
        room.cancel()
        BLIVER_POOL.pop(roomid)
        return {"msg": f"删除直播间 {roomid} 弹幕监听成功", "status": True}
    return {"msg": f"直播间 {roomid} 不在弹幕监听中", "status": False}

# 已在监听中的直播间列表
@app.get("/show")
async def show():
    return list(BLIVER_POOL.keys())

# 查询直播间弹幕监听
@app.get("/query")
async def query(roomid: int):
    room = BLIVER_POOL.get(roomid, None)
    if room != None:
        return {"msg": f"直播间 {roomid} 弹幕正在监听中", "status": True}
    return {"msg": f"直播间 {roomid} 弹幕不在监听中", "status": False}
  • 👍1
yulinfeng000 commented

之前说的，create失效的原因找到了 我直接复制的readme中与fastapi结合部分的代码，其中，del部分没有从BLIVER_POOL中pop，再次开启监听时，程序会认为已开启，导致开启失败。

我简单修改了一下

# 创建直播间弹幕监听
@app.get("/create")
async def create_new_bliver(roomid: int):
    room = BLIVER_POOL.get(roomid, None)
    if room == None:
        b = create_bliver(roomid)
        BLIVER_POOL[roomid] = b.run_as_task() 
        log("INFO", f"开启直播间监听: {roomid}")
        return {"msg": f"创建直播间 {roomid} 弹幕监听成功", "status": True}
    return {"msg": f"直播间 {roomid} 已在监听中", "status": False}

# 删除直播间弹幕监听
@app.get("/del")
async def rm_bliver(roomid: int):
    global BLIVER_POOL
    room = BLIVER_POOL.get(roomid, None)
    if room != None:
        room.cancel()
        BLIVER_POOL.pop(roomid)
        return {"msg": f"删除直播间 {roomid} 弹幕监听成功", "status": True}
    return {"msg": f"直播间 {roomid} 不在弹幕监听中", "status": False}

# 已在监听中的直播间列表
@app.get("/show")
async def show():
    return list(BLIVER_POOL.keys())

# 查询直播间弹幕监听
@app.get("/query")
async def query(roomid: int):
    room = BLIVER_POOL.get(roomid, None)
    if room != None:
        return {"msg": f"直播间 {roomid} 弹幕正在监听中", "status": True}
    return {"msg": f"直播间 {roomid} 弹幕不在监听中", "status": False}

updated, thank you