Live stream monitor module
An easy template to help you build a Live monitor which will loop check the live room status.
Why not support websocket?
Although using websocket to establish the live room is an efficient method, it is hard to find universal parts among many live platforms, and some platforms don't even support websocket, so I decided to use loop check method to monitor. If you know how to use websocket to monitor a specific live platform, I think the best way is to develop your dedicate tool.
Monitor processes flv stream by frames from v1.3, this method will probably reduce live video damages like video garbled due to a new video configuration followed by the old one.
Monitor downloads hls(*.m3u8) by using external FFmpeg from v1.5, please make sure FFmpeg binary file named
ffmpeg. * with your python script under the same folder.
Monitor corrects timestamp in header of flv frames while downloading flv stream from v1.6.7.
PyLiveRecorder | __init__.py | +---Core | Bilibili.py | HuaJiao.py | HuYa.py | YiZhiBo.py | LongZhu.py | DouYu.py | YY.py | __init__.py | \---NoticeWare Bilibili.py __init__.py
You can use pip to install this module from Pypi:
pip install pyliverecorder
Or use pip to install .whl file downloaded from releases.
For example, we can use built-in StreamPickers and Noticewares to build a Live monitor to download continuous stream like this:
from PyLiveRecorder import Monitor from PyLiveRecorder.Core.Bilibili import StreamPicker from PyLiveRecorder.NoticeWare.Bilibili import Notice sp = StreamPicker("roomId here") m = Monitor(sp, gap = 60, # loop check the live room per minute total_try = 5, # try StreamPicker for 5 times NoticeWares = [Notice("Bilibili session here")]) m.start() while True: cmd = input() # stop the monitor and exit if cmd == "stop": m.stop() break # check information about monitor (include streampicker and noticewares used) elif cmd == "minfo": print(m.getInfo()) # output name of streampicker elif cmd == "spname": print(sp.getName())
Or another Live monitor to download HLS like this:
from PyLiveRecorder import Monitor from PyLiveRecorder.Core.HuYa import StreamPicker_HLS sp = StreamPicker_HLS("roomId here") m = Monitor(sp, gap = 60, # loop check the live room per minute ) m.start() while True: cmd = input() # stop the monitor and exit if cmd == "stop": m.stop() break # check information about monitor (include streampicker and noticewares used) elif cmd == "minfo": print(m.getInfo()) # output name of streampicker elif cmd == "spname": print(sp.getName())
Also we can develop our own StreamPicker like this: (4 functions below are necessary)
class StreamPicker: def __init__(self, RoomId, skip_verify = False): ''' initialize xxx streampicker with RoomId RoomId: xxx-Live room Id skip_verify: skip verification while initializing ''' self.__name = "xxx" self.__RoomId = RoomId pass if not skip_verify: self.__verify() def getName(self): ''' return name of this ''' return self.__name def __verify(self): ''' verify if Id or this module is valid ''' try: self.getStreamURL() except: print("Input Id or this module is invalid!") exit() def getStreamURL(self): ''' try to check live status OnAir return: [True, roomid, checktime, roomurl, fullname of output file, nickname, continuous stream url] Not OnAir return: [False, roomid, checktime, None, None, None, None] ''' pass if True: return [False, self.__RoomId, checktime, None, None, None, None] else: return [True, self.__RoomId, checktime, roomurl, filename, nickname, url]
And more, a noticeware: (4 functions below are necessary)
class Notice: def __init__(self, ...): ''' initialize xxx NoticeWare ''' # name of noticeware self.__name = "xxx" pass def getName(self): ''' return name of this ''' return self.__name def send(self, nickname, roomurl): ''' send notice to xxx nickname: nickname of performer roomurl: link to live room ''' pass def remove(self): ''' delete notice from xxx ''' pass