用 Python + FFmpeg 写一个延时摄影合成脚本#

1
import argparse
2
import os
3
import re
4
import shutil
5
import subprocess
6
import sys
7
import tempfile
8
import threading
9
import time
10
from datetime import datetime
11
from pathlib import Path
12

13
BAR_WIDTH = 28
14

15

16
def probe_size(path):
17
    """用ffprobe读一张图的宽高（只读文件头，很快）"""
18
    try:
19
        r = subprocess.run(
20
            ["ffprobe", "-v", "error", "-select_streams", "v",
21
             "-show_entries", "stream=width,height", "-of", "csv=p=0", str(path)],
22
            capture_output=True, text=True, check=True)
23
        w, h = r.stdout.strip().split(",")[:2]
24
        return int(w), int(h)
25
    except Exception:
26
        return None
27

28

29
def parse_date(text):
30
    """从文件夹名开头解析日期，统一成 YYYYMMDD"""
31
    m = re.match(r"(\d{4})[-_.]?(\d{2})[-_.]?(\d{2})", text)
32
    if m:
33
        return f"{m.group(1)}{m.group(2)}{m.group(3)}"
34
    return None
35

36

37
def even(n):
38
    return int(n) - int(n) % 2
39

40

41
def build_video_filter(out_w, out_h):
42
    # 源JPG是yuvj422p全范围，只写-pix_fmt yuv420p不会改变range，
43
    # 所以显式指定偶数尺寸并转成有限范围(tv)的标准yuv420p
44
    return f"scale={out_w}:{out_h}:in_range=auto:out_range=tv,format=yuv420p"
45

46

47
def write_concat_list(path, images, img_dir, duration):
48
    with open(path, "w", encoding="utf-8") as f:
49
        for p in images:
50
            f.write(f"file '{os.path.relpath(p, img_dir)}'\n")
51
            f.write(f"duration {duration:.6f}\n")
52

53

54
def format_duration(seconds):
55
    seconds = max(0, int(seconds))
56
    h, rem = divmod(seconds, 3600)
57
    m, s = divmod(rem, 60)
58
    return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}"
59

60

61
def draw_progress(done, total, start):
62
    frac = 0.0 if total <= 0 else min(1.0, done / total)
63
    filled = int(frac * BAR_WIDTH)
64
    bar = "█" * filled + "░" * (BAR_WIDTH - filled)
65
    elapsed = time.time() - start
66
    eta = format_duration(elapsed * (total - done) / done) if done > 0 else "--:--"
67
    line = f"\r[{bar}] {frac * 100:5.1f}%  已用 {format_duration(elapsed)}  剩余 {eta}"
68
    sys.stdout.write(line + " " * max(0, 84 - len(line)))
69
    sys.stdout.flush()
70

71

72
def run_ffmpeg_batch(cmds, totals, cwd):
73
    """并行跑多个ffmpeg，汇总成一条进度条；返回 [(returncode, stderr文本), ...]"""
74
    total = sum(totals)
75
    frames = [0] * len(cmds)
76
    lock = threading.Lock()
77
    start = time.time()
78
    logs = []
79
    procs = []
80

81
    for cmd in cmds:
82
        fd, path = tempfile.mkstemp(prefix="tl_log_", suffix=".txt")
83
        fh = os.fdopen(fd, "wb")
84
        logs.append((fh, path))
85
        procs.append(subprocess.Popen(cmd, cwd=cwd, stdout=subprocess.PIPE,
86
                                      stderr=fh, text=True, bufsize=1))
87

88
    def reader(i, proc):
89
        for line in proc.stdout:
90
            if line.startswith("frame="):
91
                try:
92
                    frames[i] = int(line.strip().split("=", 1)[1])
93
                except ValueError:
94
                    continue
95
                with lock:
96
                    draw_progress(sum(frames), total, start)
97

98
    threads = [threading.Thread(target=reader, args=(i, p), daemon=True)
99
               for i, p in enumerate(procs)]
100
    for t in threads:
101
        t.start()
102
    codes = [p.wait() for p in procs]
103
    for t in threads:
104
        t.join()
105
    for fh, _ in logs:
106
        fh.close()
107
    draw_progress(total, total, start)
108
    sys.stdout.write("\n")
109
    sys.stdout.flush()
110

111
    results = []
112
    for code, (_, path) in zip(codes, logs):
113
        try:
114
            with open(path, "r", encoding="utf-8", errors="replace") as fh:
115
                err = fh.read().strip()
116
        except OSError:
117
            err = ""
118
        try:
119
            os.unlink(path)
120
        except OSError:
121
            pass
122
        results.append((code, err))
123
    return results
124

125

126
def main():
127
    # Windows默认控制台是GBK，无法输出✅等表情，会直接崩溃，这里强制UTF-8
128
    if hasattr(sys.stdout, "reconfigure"):
129
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
130

131
    parser = argparse.ArgumentParser(description="用ffmpeg concat模式合成延时摄影（图片清单）")
132
    parser.add_argument("--dir", "-d", required=True, help="照片所在文件夹路径")
133
    parser.add_argument("--fps", "-f", type=float, default=25, help="输出视频帧率，默认25")
134
    parser.add_argument("--output", "-o", default=None,
135
                        help="输出mp4文件名，默认自动命名：yyyymmdd_timelapse_宽x高_编码器.mp4")
136
    parser.add_argument("--scale", "-s", default=None, help="缩放宽度上限，默认1920（-4k模式为3200）；填0保留原图尺寸")
137
    parser.add_argument("--crf", type=int, default=23, help="H.264 CRF画质，默认23，越大画质越低")
138
    parser.add_argument("--preset", "-p", default=None, help="x264预设：ultrafast最快 / veryfast / medium画质更好但更慢（默认veryfast，-4k模式为ultrafast）")
139
    parser.add_argument("--jobs", "-j", type=int, default=0, help="并行进程数，0=自动（约等于CPU物理核心数），1=单进程")
140
    parser.add_argument("-4k", "--4k", dest="four_k", action="store_true", help="4K优化模式：3200宽(Level 5.1) + ultrafast + 自动并行")
141
    args = parser.parse_args()
142

143
    # -4k 给出一组优化默认值；单独再用 -s / -p 仍可覆盖
144
    scale = args.scale if args.scale is not None else ("3200" if args.four_k else "1920")
145
    preset = args.preset if args.preset is not None else ("ultrafast" if args.four_k else "veryfast")
146

147
    img_dir = Path(args.dir)
148
    if not img_dir.exists():
149
        print(f"错误：目录不存在 {img_dir}")
150
        return
151

152
    # 搜索jpg，大小写兼容，不会重复
153
    img_list = []
154
    for ext in ("*.jpg", "*.JPG"):
155
        img_list.extend(img_dir.glob(ext))
156
    img_list = sorted(list(set(img_list))) #去重，排序
157
    if len(img_list) == 0:
158
        print("错误：文件夹内没有找到jpg图片！")
159
        return
160
    print(f"✅ 一共找到 {len(img_list)} 张图片")
161
    if args.four_k:
162
        print(f"✅ 4K优化模式：宽度 {scale}（Level 5.1）+ preset {preset} + 自动并行")
163

164
    # 显式算出偶数输出尺寸：这样文件名里的分辨率一定和实际画面一致
165
    scale_arg = str(scale).strip().lower()
166
    keep_native = scale_arg in ("0", "", "none", "original")
167
    src = probe_size(img_list[0])
168
    if src:
169
        sw, sh = src
170
        if keep_native:
171
            out_w = even(sw)
172
        else:
173
            try:
174
                target_w = int(float(scale_arg))
175
            except ValueError:
176
                print(f"错误：--scale 需要是数字或 0，收到 {scale_arg}")
177
                return
178
            out_w = even(min(target_w, sw))
179
        out_h = even(round(sh * out_w / sw))
180
        vf = build_video_filter(out_w, out_h)
181
        res_tag = f"{out_w}x{out_h}"
182
    else:
183
        print("⚠️ ffprobe读不到源尺寸，改用自适应缩放（文件名分辨率可能不精确）")
184
        if keep_native:
185
            vf = "scale=trunc(iw/2)*2:trunc(ih/2)*2:in_range=auto:out_range=tv,format=yuv420p"
186
            res_tag = "native"
187
        else:
188
            vf = f"scale='trunc(min({scale_arg},iw)/2)*2':-2:in_range=auto:out_range=tv,format=yuv420p"
189
            res_tag = f"{scale_arg}w"
190

191
    codec_tag = "h264"  # 目前固定用 libx264
192
    if args.output:
193
        out_path = img_dir / args.output
194
    else:
195
        date = parse_date(img_dir.name) or datetime.fromtimestamp(img_list[0].stat().st_mtime).strftime("%Y%m%d")
196
        out_path = img_dir / f"{date}_timelapse_{res_tag}_{codec_tag}.mp4"
197
    print(f"✅ 输出文件：{out_path.name}")
198

199
    duration = 1.0 / args.fps
200

201
    # MJPEG解码是单线程的（实测 -threads 无效），所以提速只能靠"多个ffmpeg进程分片并行"
202
    cpus = os.cpu_count() or 2
203
    jobs = args.jobs if args.jobs > 0 else max(1, cpus // 2)
204
    jobs = max(1, min(jobs, len(img_list)))
205
    enc_threads = max(1, cpus // jobs)
206
    enc = ["-c:v", "libx264", "-preset", preset, "-crf", str(args.crf), "-threads", str(enc_threads)]
207

208
    def base_cmd(in_list):
209
        # -loglevel error 关掉日志，-progress pipe:1 输出机器可读进度
210
        return ["ffmpeg", "-hide_banner", "-loglevel", "error", "-nostats", "-progress", "pipe:1",
211
                "-f", "concat", "-safe", "0", "-i", str(in_list), "-vf", vf] + enc + ["-r", str(args.fps)]
212

213
    # 单进程：直接输出
214
    if jobs == 1:
215
        list_path = img_dir / "ffmpeg_list.txt"
216
        write_concat_list(list_path, img_list, img_dir, duration)
217
        cmd = base_cmd(list_path) + ["-pix_fmt", "yuv420p", "-movflags", "+faststart", "-y", str(out_path)]
218
        print("\n开始渲染…")
219
        code, err = run_ffmpeg_batch([cmd], [len(img_list)], str(img_dir))[0]
220
        list_path.unlink(missing_ok=True)
221
        if code == 0:
222
            print("✅ 渲染完成！")
223
        else:
224
            print(f"❌ ffmpeg执行失败，返回码：{code}")
225
            if err:
226
                print(err)
227
        print("✅ 临时清单文件已清理")
228
        return
229

230
    # 多进程：按顺序切成 jobs 个分片并行渲染，再用 -c copy 无损拼接
231
    chunk = (len(img_list) + jobs - 1) // jobs
232
    chunks = [img_list[i:i + chunk] for i in range(0, len(img_list), chunk)]
233
    print(f"✅ 并行渲染：{len(chunks)} 个分片，每个进程 {enc_threads} 线程（共 {cpus} 线程）")
234
    print("\n开始渲染…")
235

236
    tmpdir = Path(tempfile.mkdtemp(prefix="timelapse_"))
237
    seg_lists = []
238
    seg_files = []
239
    try:
240
        cmds = []
241
        for i, part in enumerate(chunks):
242
            lp = img_dir / f"ffmpeg_list_{i:03d}.txt"
243
            write_concat_list(lp, part, img_dir, duration)
244
            seg_lists.append(lp)
245
            seg = tmpdir / f"seg_{i:03d}.mp4"
246
            seg_files.append(seg)
247
            cmds.append(base_cmd(lp) + ["-pix_fmt", "yuv420p", "-y", str(seg)])
248

249
        results = run_ffmpeg_batch(cmds, [len(c) for c in chunks], str(img_dir))
250
        failed = [(code, err) for code, err in results if code != 0]
251
        if failed:
252
            print("❌ 有分片渲染失败，已中止")
253
            for code, err in failed:
254
                print(f"返回码 {code}: {err}")
255
            return
256

257
        print("✅ 分片渲染完成，开始拼接…")
258
        seg_list = tmpdir / "segs.txt"
259
        with open(seg_list, "w", encoding="utf-8") as f:
260
            for s in seg_files:
261
                f.write(f"file '{s.as_posix()}'\n")
262

263
        cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-nostats",
264
               "-f", "concat", "-safe", "0", "-i", str(seg_list),
265
               "-c", "copy", "-movflags", "+faststart", "-y", str(out_path)]
266
        ret = subprocess.run(cmd, cwd=str(img_dir), capture_output=True, text=True)
267
        if ret.returncode == 0:
268
            print("✅ 渲染完成！")
269
        else:
270
            print(f"❌ 拼接失败，返回码：{ret.returncode}")
271
            if ret.stderr.strip():
272
                print(ret.stderr.strip())
273
    finally:
274
        for lp in seg_lists:
275
            lp.unlink(missing_ok=True)
276
        shutil.rmtree(tmpdir, ignore_errors=True)
277
        print("✅ 临时文件已清理")
278

279

280
if __name__ == "__main__":
281
    main()