代码语言：python

所属分类：其他

版本2(最新)：python+html+ffmpeg+openai兼容api实现ai自动视频剪辑软件bfwcutmaster代码

版本1(旧版)：python+ffmpeg实现类似剪映的可视化剪辑软件代码bfwcutmaster

代码描述：python+html+ffmpeg+openai兼容api实现ai自动视频剪辑软件bfwcutmaster代码，增加了可视化元素缩放改变大小和拖拽移动位置，增加了接入openai的兼容api实现自动剪辑素材到时间线的功能。

代码标签： python html ffmpeg openai 兼容 api 实现 ai 自动 视频 剪辑 软件

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Python + FFmpeg 可视化时间轴视频剪辑器（AI剪辑助手增强版）
运行: python video_editor.py
依赖: pip install flask ; 系统需安装 ffmpeg / ffprobe 并加入 PATH
"""

import os
import re
import json
import uuid
import base64
import shutil
import subprocess
import threading
import time
import urllib.request
import urllib.error
from datetime import datetime

from flask import Flask, request, jsonify, send_from_directory, Response

# ------------------------------------------------------------------
# 目录初始化
# ------------------------------------------------------------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
WORKSPACE = os.path.join(BASE_DIR, "workspace")
PROJECTS_DIR = os.path.join(WORKSPACE, "projects")
MEDIA_DIR = os.path.join(WORKSPACE, "media")
EXPORTS_DIR = os.path.join(WORKSPACE, "exports")
THUMBS_DIR = os.path.join(WORKSPACE, "thumbnails")
CONFIG_PATH = os.path.join(WORKSPACE, "config.json")

for d in (PROJECTS_DIR, MEDIA_DIR, EXPORTS_DIR, THUMBS_DIR):
    os.makedirs(d, exist_ok=True)

app = Flask(__name__)
EXPORT_STATUS = {}   # project_id -> {status, progress, output/message}

ALLOWED_VIDEO = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.flv', '.m4v'}
ALLOWED_AUDIO = {'.mp3', '.wav', '.aac', '.flac', '.m4a', '.ogg'}
ALLOWED_IMAGE = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}


# ==================================================================
# FFmpeg 工具函数
# ==================================================================
def run_cmd(cmd, timeout=None):
    return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)


def ffprobe_info(path):
    cmd = ["ffprobe", "-v", "quiet", "-print_format", "json",
           "-show_format", "-show_streams", path]
    try:
        out = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
        data = json.loads(out.decode("utf-8", errors="ignore"))
    except Exception:
        return {"duration": 0, "width": 0, "height": 0, "has_audio": False, "has_video": False}

    duration = float(data.get("format", {}).get("duration", 0) or 0)
    width = height = 0
    has_audio = has_video = False
    for s in data.get("streams", []):
        if s.get("codec_type") == "video" and not has_video:
            width = int(s.get("width") or 0)
            height = int(s.get("height") or 0)
            has_video = True
            if not duration:
                duration = float(s.get("duration", 0) or 0)
        if s.get("codec_type") == "audio":
            has_audio = True
    return {"duration": duration, "width": width, "height": height,
            "has_audio": has_audio, "has_video": has_video}


def detect_media_type(filename):
    ext = os.path.splitext(filename)[1].lower()
    if ext in ALLOWED_VIDEO:
        return "video"
    if ext in ALLOWED_AUDIO:
        return "audio"
    if ext in ALLOWED_IMAGE:
        return "image"
    return "unknown"


def generate_thumbnail(src_path, out_path, media_type, duration=0):
    try:
        if media_type == "video":
            t = min(1.0, duration / 2) if duration else 0.3
            run_cmd(["ffmpeg", "-y", "-ss", str(t), "-i", src_path,
                      "-frames:v", "1", "-vf", "scale=160:-1", out_path], timeout=30)
        elif media_type == "image":
            run_cmd(["ffmpeg", "-y", "-i", src_path,
                      "-vf", "scale=160:-1", "-frames:v", "1", out_path], timeout=30)
    except Exception as e:
        print("生成缩略图失败:", e)


_CACHED_FONT = None
def find_system_font():
    """自动探测一个可用的中文字体文件，避免 Windows 下 drawtext 因缺少 fontconfig 崩溃"""
    global _CACHED_FONT
    if _CACHED_FONT is not None:
        return _CACHED_FONT
    candidates = []
    if os.name == "nt":
        win_fonts = os.path.join(os.environ.get("WINDIR", "C:/Windows"), "Fonts")
        candidates += [
            os.path.join(win_fonts, "msyh.ttc"),
            os.path.join(win_fonts, "msyhbd.ttc"),
            os.path.join(win_fonts, "simhei.ttf"),
            os.path.join(win_fonts, "simsun.ttc"),
            os.path.join(win_fonts, "arial.ttf"),
        ]
    elif os.sys.platform == "darwin":
        candidates += [
            "/System/Library/Fonts/PingFang.ttc",
            "/System/Library/Fonts/STHeiti Light.ttc",
            "/Library/Fonts/Arial Unicode.ttf",
        ]
    else:
        candidates += [
            "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc",
            "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
            "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
            "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
            "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
        ]
    for c in candidates:
        if os.path.exists(c):
            _CACHED_FONT = c
            return c
    _CACHED_FONT = ""
    return ""


def escape_fontpath(path):
    path = path.replace("\\", "/")
    path = path.replace(":", "\\:")
    return path


# ==================================================================
# 项目管理
# ==================================================================
def project_path(pid):
    return os.path.join(PROJECTS_DIR, f"{pid}.json")


def load_project(pid):
    p = project_path(pid)
    if not os.path.exists(p):
        return None
    with open(p, "r", encoding="utf-8") as f:
        return json.load(f)


def save_project(pid, data):
    with open(project_path(pid), "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def list_projects():
    result = []
    for fn in os.listdir(PROJECTS_DIR):
        if fn.endswith(".json"):
            try:
                with open(os.path.join(PROJECTS_DIR, fn), "r", encoding="utf-8") as f:
                    d = json.load(f)
                result.append({
                    "id": d.get("id"),
                    "name": d.get("name"),
                    "created": d.get("created"),
                    "modified": d.get("modified"),
                    "resolution": d.get("resolution"),
                })
            except Exception:
                continue
    result.sort(key=lambda x: x.get("modified", ""), reverse=True)
    return result


def default_project(name):
    pid = uuid.uuid4().hex[:12]
    now = datetime.now().isoformat()
    return {
        "id": pid,
        "name": name,
        "created": now,
        "modified": now,
        "resolution": {"width": 1280, "height": 720},
        "fps": 30,
        "tracks": [
            {"id": uuid.uuid4().hex[:8], "type": "video", "name": "视频轨道1",
             "clips": [], "muted": False, "hidden": False},
            {"id": uuid.uuid4().hex[:8], "type": "audio", "name": "音频轨道1",
             "clips": [], "muted": False, "hidden": False},
            {"id": uuid.uuid4().hex[:8], "type": "subtitle", "name": "字幕轨道1",
             "clips": [], "muted": False, "hidden": False},
        ],
        "media": [],
        "aiChatHistory": []
    }


# ==================================================================
# 导出引擎（构建 ffmpeg filter_complex 并执行）
# ==================================================================
def escape_drawtext(text):
    text = text.replace("\\", "\\\\")
    text = text.replace(":", "\\:")
    text = text.replace("'", "\u2019")
    text = text.replace("\n", " ")
    return text


def build_atempo_chain(speed):
    filters = []
    s = speed if speed and speed > 0 else 1.0
    while s > 2.0:
        filters.append("atempo=2.0")
        s /= 2.0
    while s < 0.5:
        filters.append("atempo=0.5")
        s /= 0.5
    filters.append(f"atempo={s:.4f}")
    return filters


def compute_total_duration(project):
    end = 0.0
    for track in project["tracks"]:
        for clip in track["clips"]:
            e = clip.get("trackStart", 0) + clip.get("duration", 0)
            if e > end:
                end = e
    return max(end, 1.0)


def build_export(project, output_path):
    W = project["resolution"]["width"]
    H = project["resolution"]["height"]
    fps = project.get("fps", 30)
    total_dur = compute_total_duration(project)
    media_lookup = {m["id"]: m for m in project["media"]}

    input_args = []
    idx_counter = [0]

    def add_input(media, extra_pre=None):
        args = []
        if extra_pre:
            args += extra_pre
        args += ["-i", media["path"]]
        input_args.extend(args)
        i = idx_counter[0]
        idx_counter[0] += 1
        return i

    video_tracks = [t for t in project["tracks"] if t["type"] in ("video", "overlay", "sticker")]
    audio_tracks = [t for t in project["tracks"] if t["type"] == "audio"]
    subtitle_tracks = [t for t in project["tracks"] if t["type"] == "subtitle"]

    filter_lines = []
    filter_lines.append(f"color=c=black:s={W}x{H}:d={total_dur:.3f}:r={fps}[vbase0]")
    base_label = "vbase0"
    v_counter = 1
    audio_labels = []
    a_counter = 1

    # ---- 视频 / 叠加(画中画/贴纸) 轨道，按轨道顺序从下到上叠加 ----
    for track in video_tracks:
        if track.get("hidden"):
            continue
        for clip in track["clips"]:
            media = media_lookup.get(clip.get("mediaId"))
            if not media:
                continue
            speed = clip.get("speed", 1.0) or 1.0
            src_in = float(clip.get("sourceIn", 0))
            src_out = float(clip.get("sourceOut", media.get("duration", 5)))
            track_start = float(clip.get("trackStart", 0))
            duration = float(clip.get("duration", max(0.1, (src_out - src_in) / speed)))
            pos = clip.get("position") or {}
            x = pos.get("x", 0)
            y = pos.get("y", 0)
            w = int(pos.get("width", W))
            h = int(pos.get("height", H))
            opacity = float(clip.get("opacity", 1.0))

            if media["type"] == "image":
                iidx = add_input(media, extra_pre=["-loop", "1"])
                seg_len = duration * speed
                vf = (f"[{iidx}:v]trim=start=0:end={seg_len:.3f},setpts=PTS-STARTPTS,"
                      f"setpts=PTS/{speed}+{track_start:.3f}/TB,"
                      f"scale={w}:{h},format=rgba")
            else:
                iidx = add_input(media)
                vf = (f"[{iidx}:v]trim=start={src_in:.3f}:end={src_out:.3f},setpts=PTS-STARTPTS,"
                      f"setpts=PTS/{speed}+{track_start:.3f}/TB,"
                      f"scale={w}:{h}")

            label_v = f"v{v_counter}"
            if opacity < 0.999:
                vf += f",format=rgba,colorchannelmixer=aa={opacity:.3f}"
            vf += f"[{label_v}]"
            filter_lin.........完整代码请登录后点击上方下载按钮下载查看

网友评论0

取消

评论提交中...

提交评论