python：使用moviepy合并m3u8格式的视频

生信菜鸟团

发布于 2022-04-08 17:34:05

1.8K0

发布于 2022-04-08 17:34:05

文章被收录于专栏：生信菜鸟团

视频网站或者客户端缓存下来的文件很多时候都是m3u8格式的文件，也就是拆成了很多段的视频，一个m3u8 文件实质是一个播放列表（playlist），其可能是一个媒体播放列表（Media Playlist）或者是一个主列表（Master Playlist）。

当 m3u8 文件作为媒体播放列表（Meida Playlist）时，其内部信息记录的是一系列媒体片段资源，顺序播放该片段资源，即可完整展示多媒体资源。

如果要本地查看的话，可以考虑将其合并为一个mp4格式的文件，使用moviepy可以很轻松做到。

如下图所示所有视频文件放置到data文件夹，每一个都是m3u8格式的视频文件（确切的说.m3u8文件就是刚才所说的播放列表文件），打开后可以发现实际的视频文件被拆分成了子文件夹，每个子文件下是一些ts格式的视频小片段。

新建一个module，构建两个工具函数。

第一个函数是将sorted函数封装一下，主要用于正确排序视频文件的顺序，按照数字顺序1, 2, 3…排序，而不是字符顺序1, 10, 11…排序，不然会导致合并的视频是错序的。

### utils.py
import os
import glob
import moviepy.editor as me

# sort by 1, 2, 3.. not 1, 10, 11, ..
def sortByNumSuffix(all_file):
    # 'data\\j0033fufnxu.322012.hls\\j0033fufnxu.322012.hls_0_29\\0.ts
    return sorted(
        all_file,
        key=lambda x: int(x.split("\\")[-1].split(".")[0])
    )

第二个函数是合并视频文件的工具函数

### utils.py
# combined movie
def combineVideo(
    mov, 
    dat_dir = ".", 
    fps = 24,
    out_dir = "out",
    out_name = None,
    skip = 0
):  
    # out name 
    if not os.path.exists(out_dir):
        os.mkdir(out_dir)
    if out_name == None: out_name = mov

    # 拿到所有分段video的正确顺序的路径
    subMovDir = sorted(glob.glob(f"{dat_dir}\\{mov}\\{mov}*"))
    if len(subMovDir) < 1: 
        print("No files exist..")
        return None

    subMovDetail = [glob.glob(f"{d}/*") for d in subMovDir]

    subMovDetailSorted = [sortByNumSuffix(x) for x in subMovDetail]
    subMovDetailSortedFlatten = [j for i in subMovDetailSorted for j in i]

    # 是否跳过前面的一部分video不合并
    skip = int(skip)
    if skip > 0:
        subMovDetailSortedFlatten = subMovDetailSortedFlatten[skip: ]
    # moviepy 合并
    movReadList = [me.VideoFileClip(m) for m in subMovDetailSortedFlatten]
    finalMov = me.concatenate_videoclips(movReadList)
    finalMov.write_videofile(f"{out_dir}\\{out_name}.mp4", fps = fps, remove_temp = True)

这个module可以附加两个简单测试：

### utils.py
if __name__ == "__main__":
    # test for sortByNumSuffix
    try:
        sortByNumSuffix([
            'data\\j0033fufnxu.322012.hls\\j0033fufnxu.322012.hls_0_29\\0.ts', 
            'data\\j0033fufnxu.322012.hls\\j0033fufnxu.322012.hls_0_29\\1.ts'
        ])
    except:
        print("sortByNumSuffix test fail...")
    else:
        print("sortByNumSuffix test pass.")

    # test for combineVideo
    try:
        combineVideo('j0033fufnxu.322012.hls', dat_dir = "data")
    except:
        print("combineVideo test fail...")
    else:
        print("combineVideo test fail...")

另写一个main.py，用于调用utils模块来完成video合并。这里在获取全部movie名称时，调用了系统命令“ls -l”来完成，这样获得的movie名称列表就是按照视频缓存时的创建时间的排序。定义一个log文件，如果有合并失败的文件则将其写入到log.txt中。

### main.py
import pandas as pd
import os
import subprocess as sp
import utils

# params set
dat_dir  = "data"
out_dir  = "out"
log_file = "log.txt"
skip     = 10

# all movie
movies = sp.check_output(f"ls -t {dat_dir}").splitlines()
movies = [x.decode() for x in movies]

# if exist log file, only combined failed movie
if os.path.exists(log_file):
    log = pd.read_table(log_file)
    fail = log["file"]
    movies = fail

logDataFrame = pd.DataFrame(columns=["file", "exception"])

# export video and log error
with open(log_file, "a") as f:
    for m in movies:
        try:
            utils.combineVideo(m, dat_dir = dat_dir, out_dir = out_dir, skip = skip)
        except Exception as e:
            logDataFrame.append({'file':m, 'exception':e}, ignore_index=True)

    logDataFrame.to_string(log_file)

整体的目录结构如下，data下放置的所有的待合并文件，合并完成的文件在out文件中。