python+html+ffmpeg+openai兼容api实现ai自动视频剪辑软件bfwcutmaster代码-python代码-BFW代码库

代码语言：python
所属分类：其他
版本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 traceback
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, token}
EXPORT_STATUS_LOCK = threading.Lock()

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'}

WIN_CREATIONFLAGS = 0
if os.name == "nt":
    WIN_CREATIONFLAGS = getattr(subprocess, "CREATE_NO_WINDOW", 0)


# ==================================================================
# FFmpeg 底层执行封装（核心修复：同步等待 + 严格判定成功）
# ==================================================================
def explain_ffmpeg_exit_code(code):
    """把常见的异常退出码翻译成人类可读的中文提示"""
    if code is None:
        return "进程未正常结束"
    if code == 0:
        return "正常结束"
    try:
        uns = code & 0xFFFFFFFF
    except Exception:
        uns = code
    known = {
        0xC0000005: "内存访问越界（ACCESS_VIOLATION），通常是 ffmpeg 内部崩溃，常见原因：drawtext缺少字体文件、滤镜链参数非法、显卡/硬件加速驱动异常、ffmpeg版本存在bug",
        0xC00000FD: "栈溢出（STACK_OVERFLOW），滤镜链可能过于复杂或存在递归问题",
        0xC0000135: "缺少依赖的DLL",
        0xC0000409: "栈缓冲区溢出保护触发",
    }
    hint = known.get(uns)
    if hint:
        return f"退出码 {code} (0x{uns:08X}) — {hint}"
    if code < 0:
        return f"退出码 {code}，进程被信号终止"
    return f"退出码 {code}，请查看完整日志定位具体报错行（通常搜索关键字 Error / Invalid / No such）"


def run_subprocess_sync(cmd, timeout=None, cwd=None):
    """
    同步执行外部命令：完整等待进程结束后，再返回结果。
    返回: (success: bool, returncode: int, full_log: str)
    绝不会出现"命令还没跑完就判断"的情况。
    """
    try:
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            cwd=cwd,
            creationflags=WIN_CREATIONFLAGS
        )
    except FileNotFoundError:
        return False, -1, f"未找到可执行文件: {cmd[0]}，请确认已安装并加入系统 PATH 环境变量"
    except Exception as e:
        return False, -1, f"启动进程失败: {e}\n{traceback.format_exc()}"

    try:
        # communicate() 会阻塞直到进程完全退出，并读取全部输出，不会有"提前判断"的风险
        raw_output, _ = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            raw_output, _ = proc.communicate(timeout=5)
        except Exception:
            raw_output = b""
        log_text = (raw_output or b"").decode("utf-8", errors="ignore")
        return False, -1, log_text + "\n[进程执行超时，已被强制终止]"
    except Exception as e:
        try:
            proc.kill()
        except Exception:
            pass
        return False, -1, f"等待进程结束时发生异常: {e}\n{traceback.format_exc()}"

    returncode = proc.returncode  # 进程已100%结束，returncode 此时必然有效
    log_text = (raw_output or b"").decode("utf-8", errors="ignore")
    success = (returncode == 0)
    return success, returncode, log_text


def run_subprocess_streaming(cmd, on_line=None, timeout=None):
    """
    带实时进度回调的执行方式（用于导出时更新进度条），
    但同样保证：只有 proc.wait() 明确返回后，才判定成败。
    返回: (success, returncode, full_log)
    """
    try:
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            creationflags=WIN_CREATIONFLAGS
        )
    except FileNotFoundError:
        return False, -1, f"未找到可执行文件: {cmd[0]}，请确认已安装并加入系统 PATH 环境变量"
    except Exception as e:
        return False, -1, f"启动进程失败: {e}\n{traceback.format_exc()}"

    lines = []
    try:
        # 逐行读取用于进度回调（不会阻塞等待整个输出缓冲区）
        for raw_line in iter(proc.stdout.readline, b""):
            try:
                text_line = raw_line.decode("utf-8", errors="ignore")
            except Exception:
                text_line = ""
            lines.append(text_line)
            if on_line:
                try:
                    on_line(text_line)
                except Exception:
                    pass
        proc.stdout.close()
    except Exception as e:
        lines.append(f"\n[读取输出流时发生异常: {e}]\n{traceback.format_exc()}")

    # 关键：无论上面读流是否顺利，都必须显式等待进程彻底结束再取 returncode
    try:
        returncode = proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        returncode = proc.wait()
        lines.append("\n[进程执行超时，已被强制终止]")
    except Exception as e:
        try:
            proc.kill()
        except Exception:
            pass
        returncode = -1
        lines.append(f"\n[等待进程结束时发生异常: {e}]\n{traceback.format_exc()}")

    full_log = "".join(lines)
    success = (returncode == 0)
    return success, returncode, full_log


def ffprobe_info(path):
    cmd = ["ffprobe", "-v", "quiet", "-print_format", "json",
           "-show_format", "-show_streams", path]
    success, code, log = run_subprocess_sync(cmd, timeout=30)
    if not success:
        print(f"[ffprobe警告] 探测媒体信息失败 (退出码 {code}): {path}\n{log[:500]}")
        return {"duration": 0, "width": 0, "height": 0, "has_audio": False, "has_video": False}
    try:
        data = json.loads(log)
    except Exception as e:
        print(f"[ffprobe警告] 解析JSON失败: {e}")
        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
            cmd = ["ffmpeg", "-y", "-ss", str(t), "-i", src_path,
                   "-frames:v", "1", "-vf", "scale=160:-1", out_path]
        elif media_type == "image":
            cmd = ["ffmpeg", "-y", "-i", src_path,
                   "-vf", "scale=160:-1", "-frames:v", "1", out_path]
        else:
            return False
        success, code, log = run_subprocess_sync(cmd, timeout=30)
        if not success:
            print(f"[缩略图生成失败] {src_path} 退出码={code}\n{log[-500:]}")
            return False
        if not os.path.exists(out_path) or os.path.getsize(out_path) < 100:
            print(f"[缩略图生成失败] 输出文件未生成或过小: {out_path}")
            return False
        return True
    except Exception as e:
        print("生成缩略图异常:", e)
        return False


_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
 .........完整代码请登录后点击上方下载按钮下载查看