经常在各种影视站看到图床切片的视频,研究了下。
首先看效果(新人练手项目,代码很烂,重在思路)
视频仅提供示例,随便下的一个视频
本文不提供相关接口,仅做技术复现!
前期准备
需要首先安装
ffmpeg
并配置到环境变量(使用命令行输入ffmpeg
可以正常打开的那种) gyan.dev/ffmpeg/builds 或 github.com/BtbN/FFmpeg-Builds/releases
目录结构
项目分为以下几部分
- 对 mp4 文件进行格式转换,转成单个 ts 文件
- 对 单个 ts 文件进行转换,转成 m3u8 文件和 ts 切片文件
- 对 ts 切片文件进行重构,使其具有
PNG
文件格式的文件头 - 上传图床并根据返回的结果重写之前 m3u8 文件。
一、文件转换部分
mp4
转 单个 ts
def video_to_ts(video_path):
ts_all = ".//ts_single//ts_single.ts"
cmd_str = f'ffmpeg -y -i {video_path} -vcodec copy -acodec copy -vbsf h264_mp4toannexb {ts_all}'
subprocess.run(cmd_str, encoding="utf-8", shell=True)
print(f'从 {video_path} 转换到 {cmd_str} 成功!')
二、生成 m3u8
文件
单个 ts
转 m3u8
def ts_to_m3u8(ts_single, singlg_time):
cmd_str = f'ffmpeg -i {ts_single} -c copy -map 0 -f segment -segment_list temp_playlist.m3u8 -segment_time {singlg_time} ./ts_all/%03d.ts'
subprocess.run(cmd_str, encoding="utf-8", shell=True)
print(f'从 {ts_single} 转换到 temp_playlist.m3u8 成功!')
三、ts
文件格式处理部分
首先重命名为 PNG
后缀
def rename_ts_to_png():
file_list = os.listdir("./ts_all")
for i in file_list:
if i.endswith(".ts"):
new_name = i.replace(".ts", ".png")
os.rename("./ts_all/" + i, "./ts_all/" + new_name)
print("rename_ts_to_png")
重写覆写 ts
文件头,使其拥有 PNG
文件头
def rewrite_ts_to_png():
file_list = os.listdir("./ts_all")
for i in file_list:
copyfile("PNG", "./ts_rewrite/" + i)
print("rewrite_ts_to_png")
file_list = os.listdir("./ts_rewrite")
for i in file_list:
if i.endswith(".png"):
bin_file = open("./ts_all/" + i, 'rb') # 打开二进制文件
# 合并文件
with open("./ts_rewrite/" + i, 'ab') as f:
f.write(bin_file.read())
bin_file.close()
return None
PNG 文件头 16 进制数据如下(放在项目目录命名为 PNG
)
89504E470D0A1A0A0000000D4948445200000320000003200803000000ECAEF65A0000001974455874536F6674776172650041646F626520496D616765526561647971C9653C00000006504C5445000000000000A567B9CF0000000174524E530040E6D866000002864944415478DAECC1010D000000C2A0F74F6D0E37A0
四、文件上传以及重写 M3U8
文件
文件上传
# 上传图片
def updateImage(filepath):
_upload_url = "******************************/api/content/file/upload"
files = {'media': open(filepath, 'rb')}
upload_res = requests.post(_upload_url, files=files)
print('正在上传:', filepath)
return upload_res.json()
提取返回的 url
def upload_list(_upload_url):
_file_dict = {}
file_list = os.listdir(_upload_url)
for i in file_list:
data_json = updateImage(_upload_url + i)
_file_dict[i] = data_json["***"]["*pic*"] + "\n"
return _file_dict
重写 m3u8 文件
def rewrite_m3u8(m3u8_file, file_dict):
print(file_dict)
with open(m3u8_file, 'r') as f:
lines = f.readlines()
for i in lines:
rename = i.replace("ts", "png").replace("\n", "")
if rename in file_dict.keys():
lines[lines.index(i)] = file_dict[rename]
print("正在替换:", i)
with open(m3u8_file, 'w') as f:
f.writelines(lines)
print("rewrite_m3u8")
五、串联以上代码的入口函数
if __name__ == '__main__':
video_to_ts("video.mp4")
ts_to_m3u8("./ts_single/ts_single.ts", "5")
rename_ts_to_png()
rewrite_ts_to_png()
upload_url = "./ts_rewrite/"
file_dict = upload_list(upload_url)
rewrite_m3u8("./temp_playlist.m3u8", file_dict)
print("main")
需要自行创建文件夹及准备文件
- ./ts_single/
- ./ts_all/
- ./ts_rewrite/
- ./video.mp4
- ./PNG
完整代码 (无上传接口版本)
# coding:utf-8
import subprocess
import requests
from shutil import copyfile
import os
def video_to_ts(video_path):
ts_all = ".//ts_single//ts_single.ts"
cmd_str = f'ffmpeg -y -i {video_path} -vcodec copy -acodec copy -vbsf h264_mp4toannexb {ts_all}'
subprocess.run(cmd_str, encoding="utf-8", shell=True)
print(f'从 {video_path} 转换到 {cmd_str} 成功!')
def ts_to_m3u8(ts_single, singlg_time):
cmd_str = f'ffmpeg -i {ts_single} -c copy -map 0 -f segment -segment_list temp_playlist.m3u8 -segment_time {singlg_time} ./ts_all/%03d.ts'
subprocess.run(cmd_str, encoding="utf-8", shell=True)
print(f'从 {ts_single} 转换到 temp_playlist.m3u8 成功!')
def rename_ts_to_png():
file_list = os.listdir("./ts_all")
for i in file_list:
if i.endswith(".ts"):
new_name = i.replace(".ts", ".png")
os.rename("./ts_all/" + i, "./ts_all/" + new_name)
print("rename_ts_to_png")
def rewrite_ts_to_png():
file_list = os.listdir("./ts_all")
for i in file_list:
copyfile("PNG", "./ts_rewrite/" + i)
print("rewrite_ts_to_png")
file_list = os.listdir("./ts_rewrite")
for i in file_list:
if i.endswith(".png"):
bin_file = open("./ts_all/" + i, 'rb') # 打开二进制文件
# 合并文件
with open("./ts_rewrite/" + i, 'ab') as f:
f.write(bin_file.read())
bin_file.close()
return None
# 上传图片
def updateImage(filepath):
_upload_url = "*************************************************"
files = {'media': open(filepath, 'rb')}
upload_res = requests.post(_upload_url, files=files)
print('正在上传:', filepath)
return upload_res.json()
def upload_list(_upload_url):
_file_dict = {}
file_list = os.listdir(_upload_url)
for i in file_list:
data_json = updateImage(_upload_url + i)
_file_dict[i] = data_json["****"]["******"] + "\n"
return _file_dict
def rewrite_m3u8(m3u8_file, file_dict):
print(file_dict)
with open(m3u8_file, 'r') as f:
lines = f.readlines()
for i in lines:
rename = i.replace("ts", "png").replace("\n", "")
if rename in file_dict.keys():
lines[lines.index(i)] = file_dict[rename]
print("正在替换:", i)
with open(m3u8_file, 'w') as f:
f.writelines(lines)
print("rewrite_m3u8")
if __name__ == '__main__':
video_to_ts("video.mp4")
ts_to_m3u8("./ts_single/ts_single.ts", "5")
rename_ts_to_png()
rewrite_ts_to_png()
upload_url = "./ts_rewrite/"
file_dict = upload_list(upload_url)
rewrite_m3u8("./temp_playlist.m3u8", file_dict)
print("main")