python實(shí)現(xiàn)的B站直播錄制工具
項(xiàng)目地址:
https://github.com/Redlnn/blive_record
前言
- 作者: Red_lnn
- 不允許將本項(xiàng)目運(yùn)用于非法以及違反B站用戶協(xié)議的用途
- 僅支持單個(gè)主播,多個(gè)主播請(qǐng)復(fù)制多份并分開單獨(dú)啟動(dòng)
- 運(yùn)行時(shí)如要停止錄制并退出,請(qǐng)按鍵盤 Ctrl+C
- 如要修改錄制設(shè)置,請(qǐng)以純文本方式打開.py文件
- 利用 FFmpeg 直接抓取主播推送的流,無需打開瀏覽器
- 有新功能需求請(qǐng)直接提 Pull requests
- 建議錄制為 flv 格式(默認(rèn)),以防止意外中斷導(dǎo)致錄制文件損壞,若要進(jìn)行剪輯可使用 FFmpeg 轉(zhuǎn)換為 mp4 文件后再倒入到剪輯軟件(使用 FFmpeg 轉(zhuǎn)換 flv 為 mp4 : ffmpeg -i {input}.flv -c:v copy -c:a copy {output}.mp4)
使用方式
1.安裝 Python(>=3.7) 并設(shè)置環(huán)境變量
2.打開終端或命令行進(jìn)入本腳本所在目錄
3.通過 pip 安裝必須的第三方庫(kù)
Windows:
pip install -r requirements.txt
Linux:
python3 -m pip install -r requirements.txt
4.下載 ffmpeg 并正確設(shè)置環(huán)境變量(下載地址)
5.Windows 直接雙擊運(yùn)行start.bat
6.Linux 先運(yùn)行 chmod +x start.sh 再運(yùn)行 ./start.sh
主要代碼
blive_record.py
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
"""
*--------------------------------------*
B站直播錄播姬 By: Red_lnn
僅支持單個(gè)主播,多個(gè)主播請(qǐng)復(fù)制多份并分開單獨(dú)啟動(dòng)
運(yùn)行時(shí)如要停止錄制并退出,請(qǐng)按鍵盤 Ctrl+C
如要修改錄制設(shè)置,請(qǐng)以純文本方式打開.py文件
利用ffmpeg直接抓取主播推送的流,不需要打開瀏覽器
*--------------------------------------*
"""
# import ffmpy3 # noqa
import logging
import os
import signal
import sys
import threading
import time
import traceback
from json import loads
from logging import handlers
from subprocess import PIPE, Popen, STDOUT
import requests
from regex import match
# 導(dǎo)入配置
from config import * # noqa
record_status = False # 錄制狀態(tài),True為錄制中
kill_times = 0 # 嘗試強(qiáng)制結(jié)束FFmpeg的次數(shù)
logging.addLevelName(15, 'FFmpeg') # 自定義FFmpeg的日志級(jí)別
logger = logging.getLogger('Record')
logger.setLevel(logging.DEBUG)
fms = '[%(asctime)s %(levelname)s] %(message)s'
# datefmt = "%Y-%m-%d %H:%M:%S"
datefmt = "%H:%M:%S"
default_handler = logging.StreamHandler(sys.stdout)
if debug:
default_handler.setLevel(logging.DEBUG)
elif verbose:
default_handler.setLevel(15)
else:
default_handler.setLevel(logging.INFO)
default_handler.setFormatter(logging.Formatter(fms, datefmt=datefmt))
logger.addHandler(default_handler)
if save_log:
# file_handler = logging.FileHandler("debug.log", mode='w+', encoding='utf-8')
if not os.path.exists(os.path.join('logs')):
os.mkdir(os.path.join('logs'))
file_handler = handlers.TimedRotatingFileHandler(os.path.join('logs', 'debug.log'), 'midnight', encoding='utf-8')
if debug:
default_handler.setLevel(logging.DEBUG)
else:
default_handler.setLevel(15)
file_handler.setFormatter(logging.Formatter(fms, datefmt=datefmt))
logger.addHandler(file_handler)
def get_timestamp() -> int:
"""
獲取當(dāng)前時(shí)間戳
"""
return int(time.time())
def get_time() -> str:
"""
獲取格式化后的時(shí)間
"""
time_now = get_timestamp()
time_local = time.localtime(time_now)
dt = time.strftime("%Y%m%d_%H%M%S", time_local)
return dt
def record():
"""
錄制過程中要執(zhí)行的檢測(cè)與判斷
"""
global p, record_status, last_record_time, kill_times # noqa
while True:
line = p.stdout.readline().decode()
p.stdout.flush()
logger.log(15, line.rstrip())
if match('video:[0-9kmgB]* audio:[0-9kmgB]* subtitle:[0-9kmgB]*', line) or 'Exiting normally' in line:
record_status = False # 如果FFmpeg正常結(jié)束錄制則退出本循環(huán)
break
elif match('frame=[0-9]', line) or 'Opening' in line:
last_record_time = get_timestamp() # 獲取最后錄制的時(shí)間
elif 'Failed to read handshake response' in line:
time.sleep(5) # FFmpeg讀取m3u8流失敗,等個(gè)5s康康會(huì)不會(huì)恢復(fù)
continue
time_diff = get_timestamp() - last_record_time # 計(jì)算上次錄制到目前的時(shí)間差
if time_diff >= 65:
logger.error('最后一次錄制到目前已超65s,將嘗試發(fā)送終止信號(hào)')
logger.debug(f'間隔時(shí)間:{time_diff}s')
kill_times += 1
p.send_signal(signal.SIGTERM) # 若最后一次錄制到目前已超過65s,則認(rèn)為FFmpeg卡死,嘗試發(fā)送終止信號(hào)
time.sleep(0.5)
if kill_times >= 3:
logger.critical('由于無法結(jié)束FFmpeg進(jìn)程,將嘗試自我了結(jié)')
sys.exit(1)
if 'Immediate exit requested' in line:
logger.info('FFmpeg已被強(qiáng)制結(jié)束')
break
if p.poll() is not None: # 如果FFmpeg已退出但沒有被上一個(gè)判斷和本循環(huán)第一個(gè)判斷捕捉到,則當(dāng)作異常退出
logger.error('ffmpeg未正常退出,請(qǐng)檢查日志文件!')
record_status = False
break
def main():
global p, room_id, record_status, last_record_time, kill_times # noqa
while True:
record_status = False
while True:
logger.info('------------------------------')
logger.info(f'正在檢測(cè)直播間:{room_id}')
try:
room_info = requests.get(f'https://api.live.bilibili.com/room/v1/Room/get_info?room_id={room_id}',
timeout=5)
except (requests.exceptions.ReadTimeout, requests.exceptions.Timeout, requests.exceptions.ConnectTimeout):
logger.error(f'無法連接至B站API,等待{check_time}s后重新開始檢測(cè)')
time.sleep(check_time)
continue
live_status = loads(room_info.text)['data']['live_status']
if live_status == 1:
break
elif live_status == 0:
logger.info(f'沒有開播,等待{check_time}s重新開始檢測(cè)')
time.sleep(check_time)
if not os.path.exists(os.path.join('download')):
try:
os.mkdir(os.path.join('download'))
except: # noqa
logger.error(f'無法創(chuàng)建下載文件夾 ↓\n{traceback.format_exc()}')
sys.exit(1)
if os.path.isfile(os.path.join('download')):
logger.error('存在與下載文件夾同名的文件')
sys.exit(1)
logger.info('正在直播,準(zhǔn)備開始錄制')
m3u8_list = requests.get(
f'https://api.live.bilibili.com/xlive/web-room/v1/playUrl/playUrl?cid={room_id}&platform=h5&qn=10000')
m3u8_address = loads(m3u8_list.text)['data']['durl'][0]['url']
# 下面命令中的timeout單位為微秒,10000000us為10s(https://www.cnblogs.com/zhifa/p/12345376.html)
command = ['ffmpeg', '-rw_timeout', '10000000', '-timeout', '10000000', '-listen_timeout', '10000000',
'-headers',
'"Accept: */*? Accept-Encoding: gzip, deflate, br? Accept-Language: zh,zh-TW;q=0.9,en-US;q=0.8,en;'
f'q=0.7,zh-CN;q=0.6,ru;q=0.5? Origin: https://live.bilibili.com/{room_id}? '
'User-Agent: Mozilla/5.0 (Windows NT 10.0;Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Safari/537.36?"', '-i',
m3u8_address, '-c:v', 'copy', '-c:a', 'copy', '-bsf:a', 'aac_adtstoasc',
'-f', 'segment', '-segment_time', str(segment_time), '-segment_start_number', '1',
os.path.join('download', f'[{room_id}]_{get_time()}_part%03d.{file_extensions}'), '-y']
if debug:
logger.debug('FFmpeg命令如下 ↓')
command_str = ''
for _ in command:
command_str += _
logger.debug(command_str)
p = Popen(command, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False)
record_status = True
start_time = last_record_time = get_timestamp()
try:
t = threading.Thread(target=record)
t.start()
while True:
if not record_status:
break
if verbose or debug:
time.sleep(20)
logger.info(f'--==>>> 已錄制 {round((get_timestamp() - start_time) / 60, 2)} 分鐘 <<<==--')
else:
time.sleep(60)
logger.info(f'--==>>> 已錄制 {int((get_timestamp() - start_time) / 60)} 分鐘 <<<==--')
if not record_status:
break
except KeyboardInterrupt:
# p.send_signal(signal.CTRL_C_EVENT)
logger.info('停止錄制,等待ffmpeg退出后本程序會(huì)自動(dòng)退出')
logger.info('若長(zhǎng)時(shí)間卡住,請(qǐng)?jiān)俅伟聪耤trl+c (可能會(huì)損壞視頻文件)')
logger.info('Bye!')
sys.exit(0)
kill_times = 0
logger.info('FFmpeg已退出,重新開始檢測(cè)直播間')
# time.sleep(check_time)
if __name__ == '__main__':
logger.info('B站直播錄播姬 By: Red_lnn')
logger.info('如要停止錄制并退出,請(qǐng)按鍵盤 Ctrl+C')
logger.info('如要修改錄制設(shè)置,請(qǐng)以純文本方式打開.py文件')
logger.info('準(zhǔn)備開始錄制...')
time.sleep(0.3)
try:
main()
except KeyboardInterrupt:
logger.info('Bye!')
sys.exit(0)
config.py(配置文件)
#!/usr/bin/env python3 # -*- coding:utf-8 -*- """ *------------以下為可配置項(xiàng)-------------* """ # room_id = 1151716 # 萵苣某人 # room_id = 1857249 # Red_lnn room_id = 1151716 # 要錄制的B站直播間的直播間ID segment_time = 3600 # 錄播分段時(shí)長(zhǎng)(單位:秒) check_time = 60 # 開播檢測(cè)間隔(單位:秒) file_extensions = 'flv' # 錄制文件后綴名(文件格式) verbose = True # 是否打印ffmpeg輸出信息到控制臺(tái) debug = False # 是否顯示并保存調(diào)試信息(優(yōu)先級(jí)高于 verbose) save_log = True # 是否保存日志信息為文件,同一天多次啟動(dòng)本腳本會(huì)共用同一個(gè)日志文件,每天凌晨分割一次日志文件 """ *------------以上為可配置項(xiàng)-------------* """
以上就是python實(shí)現(xiàn)的B站直播錄播工具的詳細(xì)內(nèi)容,更多關(guān)于python B站直播錄播的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
解決python 輸出到csv 出現(xiàn)多空行的情況
這篇文章主要介紹了解決python 輸出到csv 出現(xiàn)多空行的情況,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2021-03-03
Python Flask全棧項(xiàng)目實(shí)戰(zhàn)構(gòu)建在線書店流程
這篇文章主要為大家介紹了Python Flask全流程全棧項(xiàng)目實(shí)戰(zhàn)之在線書店構(gòu)建實(shí)現(xiàn)過程,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-11-11
python Selenium實(shí)現(xiàn)付費(fèi)音樂批量下載的實(shí)現(xiàn)方法
這篇文章主要介紹了python Selenium實(shí)現(xiàn)付費(fèi)音樂批量下載的實(shí)現(xiàn)方法,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-01-01
Python遞歸遍歷列表及輸出的實(shí)現(xiàn)方法
這篇文章主要介紹了Python遞歸遍歷列表及輸出的實(shí)現(xiàn)方法,可實(shí)現(xiàn)遞歸遍歷列表中的每一項(xiàng),若是元祖則判斷是否為基本類型然后輸出,需要的朋友可以參考下2015-05-05
python 讀取攝像頭數(shù)據(jù)并保存的實(shí)例
今天小編就為大家分享一篇python 讀取攝像頭數(shù)據(jù)并保存的實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2018-08-08
詳解Python的Twisted框架中reactor事件管理器的用法
這篇文章主要介紹了詳解Python的Twisted框架中reactor事件管理器的用法,Twisted是一款高人氣的異步Python開發(fā)框架,需要的朋友可以參考下2016-05-05
Python操作Excel的10個(gè)必學(xué)腳本分享
Excel表格操作是我們?nèi)粘9ぷ髦胁豢苫蛉钡囊徊糠?而Python憑借其強(qiáng)大的數(shù)據(jù)處理能力,可以極大地提升我們操作Excel的效率,本文將介紹10個(gè)必學(xué)的Python腳本,需要的可以零基礎(chǔ)2025-03-03
Python自動(dòng)化辦公Excel模塊openpyxl原理及用法解析
這篇文章主要介紹了Python自動(dòng)化辦公Excel模塊openpyxl原理及用法解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-11-11

