n8n自动视频剪辑工作流完全指南

# 安装Node.js（推荐使用nvm） curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh | bash nvm install 18 nvm use 18 # 全局安装n8n npm install -g n8n # 启动n8n n8n start

# Ubuntu/Debian sudo apt update sudo apt install ffmpeg # macOS（使用Homebrew） brew install ffmpeg # Windows（使用Chocolatey） choco install ffmpeg

# 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心库 pip install opencv-python pip install moviepy pip install whisper-openai pip install google-generativeai pip install requests pip install pillow

# 创建 .env 文件 cat > .env << EOF # Google Gemini API GEMINI_API_KEY=your_gemini_api_key_here # TTS服务配置 TTS_SERVICE=google # 或 azure, openai GOOGLE_TTS_KEY=your_google_tts_key AZURE_TTS_KEY=your_azure_key AZURE_TTS_REGION=your_region # 存储配置 STORAGE_TYPE=local # 或 oss, s3, gcs OSS_ACCESS_KEY=your_oss_key OSS_SECRET_KEY=your_oss_secret OSS_BUCKET=your_bucket_name # n8n配置 N8N_HOST=0.0.0.0 N8N_PORT=5678 N8N_PROTOCOL=http EOF

{ "name": "视频输入", "type": "n8n-nodes-base.webhook", "parameters": { "path": "video-input", "httpMethod": "POST", "responseMode": "onReceived", "options": { "rawBody": true } } }

{ "name": "Gemini视频分析", "type": "n8n-nodes-base.httpRequest", "parameters": { "url": "https://generativelanguage.googleapis.com/v1/models/gemini-pro-vision:generateContent", "authentication": "genericCredentialType", "genericAuthType": "httpHeaderAuth", "httpHeaderAuth": { "name": "x-goog-api-key", "value": "={{$env.GEMINI_API_KEY}}" }, "method": "POST", "body": { "contents": [{ "parts": [{ "text": "请分析这个视频的主要内容，包括：1.核心主题 2.关键场景描述 3.适合的解说角度 4.建议的分镜时间点。请以JSON格式返回结果。" }, { "inline_data": { "mime_type": "video/mp4", "data": "={{$json.videoBase64}}" } }] }] } } }

import subprocess import json from moviepy.editor import VideoFileClip import os def process_video_segments(video_path, segments_data): """根据AI分析结果切割视频片段""" segments = json.loads(segments_data) output_files = [] for i, segment in enumerate(segments['scenes']): start_time = segment['start_time'] end_time = segment['end_time'] output_path = f"segment_{i}_{start_time}_{end_time}.mp4" # 使用FFmpeg进行精确切割 cmd = [ 'ffmpeg', '-i', video_path, '-ss', str(start_time), '-to', str(end_time), '-c', 'copy', # 使用流拷贝以提高速度 '-avoid_negative_ts', 'make_zero', output_path ] subprocess.run(cmd, check=True) output_files.append(output_path) return output_files

# 视频分析提示词模板 你是一位专业的视频导演和编剧。请仔细分析这个视频，并按照以下要求返回JSON格式的分析结果： ## 分析要求： 1. **内容主题**：概括视频的核心主题和亮点 2. **情感基调**：识别视频的整体情感色彩（激励、轻松、严肃等） 3. **关键场景**：识别3-5个最具吸引力的场景片段 4. **解说角度**：建议最适合的解说风格和切入点 5. **目标受众**：分析适合的观众群体 ## 输出格式： { "theme": "视频核心主题", "emotion": "情感基调", "key_scenes": [ { "start_time": 开始时间(秒), "end_time": 结束时间(秒), "description": "场景描述", "importance_score": 重要性评分(1-10) } ], "narration_style": "建议的解说风格", "target_audience": "目标受众描述", "hook_points": ["吸引点1", "吸引点2", "吸引点3"] } 请确保分析准确、具体，为后续的自动化处理提供高质量的数据支撑。

import requests import base64 from typing import Dict, Any class TTSProcessor: def __init__(self, service_type="google"): self.service_type = service_type def generate_audio(self, text: str, voice_config: Dict[str, Any]) -> str: """生成TTS音频并返回base64编码""" if self.service_type == "google": return self._google_tts(text, voice_config) elif self.service_type == "azure": return self._azure_tts(text, voice_config) else: raise ValueError(f"不支持的TTS服务: {self.service_type}") def _google_tts(self, text: str, config: Dict[str, Any]) -> str: """Google Cloud TTS实现""" url = "https://texttospeech.googleapis.com/v1/text:synthesize" payload = { "input": {"text": text}, "voice": { "languageCode": config.get("language", "zh-CN"), "name": config.get("voice_name", "zh-CN-Standard-A"), "ssmlGender": config.get("gender", "NEUTRAL") }, "audioConfig": { "audioEncoding": "MP3", "speakingRate": config.get("speed", 1.0), "pitch": config.get("pitch", 0.0) } } headers = { "Authorization": f"Bearer {os.getenv('GOOGLE_TTS_KEY')}", "Content-Type": "application/json" } response = requests.post(url, json=payload, headers=headers) response.raise_for_status() return response.json()["audioContent"] # 使用示例 tts = TTSProcessor("google") audio_base64 = tts.generate_audio( "这是一个精彩的视频片段，展现了令人震撼的画面。", { "language": "zh-CN", "voice_name": "zh-CN-Neural2-A", "speed": 1.1, "pitch": 2.0 } )

from moviepy.editor import VideoFileClip, AudioFileClip, CompositeAudioClip import librosa def sync_audio_video(video_path: str, audio_path: str, target_duration: float) -> str: """音画同步处理""" # 加载视频和音频 video = VideoFileClip(video_path) audio = AudioFileClip(audio_path) video_duration = video.duration audio_duration = audio.duration # 计算速度调整比例 if audio_duration > target_duration: # 音频太长，需要加速 speed_factor = audio_duration / target_duration adjusted_audio = audio.fx(lambda clip: clip.speedx(speed_factor)) else: # 音频太短，需要减速或添加停顿 speed_factor = audio_duration / target_duration if speed_factor > 0.7: # 减速不要过度 adjusted_audio = audio.fx(lambda clip: clip.speedx(speed_factor)) else: # 添加静音间隔 silence_duration = target_duration - audio_duration adjusted_audio = CompositeAudioClip([ audio, AudioFileClip("silence.mp3", duration=silence_duration) ]) # 如果需要，同时调整视频速度 if abs(speed_factor - 1.0) > 0.3: # 速度差异较大时调整视频 video_speed_factor = min(speed_factor * 0.8, 1.2) # 限制视频变速范围 adjusted_video = video.fx(lambda clip: clip.speedx(video_speed_factor)) else: adjusted_video = video # 合成最终视频 final_video = adjusted_video.set_audio(adjusted_audio) output_path = f"synced_{os.path.basename(video_path)}" final_video.write_videofile(output_path, audio_codec='aac') return output_path

class MultiLanguageProcessor: def __init__(self): self.supported_languages = { 'zh-CN': {'name': '简体中文', 'voice': 'zh-CN-Neural2-A'}, 'en-US': {'name': '美式英语', 'voice': 'en-US-Neural2-A'}, 'ja-JP': {'name': '日语', 'voice': 'ja-JP-Neural2-A'}, 'ko-KR': {'name': '韩语', 'voice': 'ko-KR-Neural2-A'}, 'es-ES': {'name': '西班牙语', 'voice': 'es-ES-Neural2-A'} } def translate_script(self, text: str, target_lang: str) -> str: """翻译解说文本""" url = "https://generativelanguage.googleapis.com/v1/models/gemini-pro:generateContent" prompt = f""" 请将以下中文解说词翻译成{self.supported_languages[target_lang]['name']}， 保持原有的语调和风格，确保翻译自然流畅： 原文：{text} 翻译要求： 1. 保持原文的情感基调 2. 符合目标语言的表达习惯 3. 适合短视频解说的语言风格 4. 确保翻译准确且吸引人 """ # API调用逻辑... return translated_text def generate_multilingual_videos(self, video_path: str, script: str, languages: list): """生成多语言版本视频""" results = [] for lang in languages: # 翻译脚本 translated_script = self.translate_script(script, lang) # 生成对应语言的TTS音频 tts_config = { 'language': lang, 'voice_name': self.supported_languages[lang]['voice'], 'speed': 1.0 } audio_base64 = self.generate_tts_audio(translated_script, tts_config) # 合成最终视频 output_path = f"{lang}_{os.path.basename(video_path)}" final_video = self.combine_video_audio(video_path, audio_base64, output_path) results.append({ 'language': lang, 'video_path': final_video, 'script': translated_script }) return results

class QualityController: def __init__(self): self.quality_thresholds = { 'video_resolution': (720, 1280), # 最低分辨率要求 'video_duration': (15, 60), # 视频时长范围 'audio_quality': 128, # 音频比特率(kbps) 'content_score': 7.0 # 内容质量评分阈值 } def validate_video_quality(self, video_path: str) -> dict: """视频质量检测""" video = VideoFileClip(video_path) quality_report = { 'resolution': video.size, 'duration': video.duration, 'fps': video.fps, 'has_audio': video.audio is not None, 'file_size': os.path.getsize(video_path), 'quality_score': 0 } # 分辨率检查 if quality_report['resolution'][1] >= self.quality_thresholds['video_resolution'][1]: quality_report['quality_score'] += 3 # 时长检查 duration = quality_report['duration'] if self.quality_thresholds['video_duration'][0] <= duration <= self.quality_thresholds['video_duration'][1]: quality_report['quality_score'] += 3 # 音频检查 if quality_report['has_audio']: quality_report['quality_score'] += 2 # 文件大小合理性检查 size_per_second = quality_report['file_size'] / duration if 500000 <= size_per_second <= 2000000: # 500KB-2MB per second quality_report['quality_score'] += 2 quality_report['passed'] = quality_report['quality_score'] >= 7 return quality_report def content_compliance_check(self, script: str) -> dict: """内容合规性检测""" # 敏感词检测 sensitive_words = self.load_sensitive_words() detected_words = [] for word in sensitive_words: if word in script: detected_words.append(word) # AI内容适宜性评估 suitability_score = self.assess_content_suitability(script) return { 'has_sensitive_content': len(detected_words) > 0, 'detected_words': detected_words, 'suitability_score': suitability_score, 'approved': len(detected_words) == 0 and suitability_score > 0.8 }

import logging import psutil import time from datetime import datetime class SystemMonitor: def __init__(self): # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('video_processing.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def log_system_status(self): """记录系统状态""" cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() disk = psutil.disk_usage('/') status = { 'timestamp': datetime.now().isoformat(), 'cpu_usage': cpu_percent, 'memory_usage': memory.percent, 'memory_available': memory.available / (1024**3), # GB 'disk_usage': disk.percent, 'disk_free': disk.free / (1024**3) # GB } self.logger.info(f"系统状态: {status}") # 警告阈值检查 if cpu_percent > 90: self.logger.warning(f"CPU使用率过高: {cpu_percent}%") if memory.percent > 85: self.logger.warning(f"内存使用率过高: {memory.percent}%") if disk.percent > 90: self.logger.warning(f"磁盘使用率过高: {disk.percent}%") return status def log_processing_metrics(self, video_path: str, processing_time: float, success: bool): """记录处理指标""" file_size = os.path.getsize(video_path) / (1024**2) # MB metrics = { 'video_path': video_path, 'file_size_mb': file_size, 'processing_time': processing_time, 'processing_speed': file_size / processing_time, # MB/s 'success': success, 'timestamp': datetime.now().isoformat() } self.logger.info(f"处理指标: {metrics}") return metrics

{ "原视频": { "时长": "5分钟", "内容": "智能手表完整功能演示", "分辨率": "1920x1080" }, "自动生成版本": [ { "平台": "TikTok/抖音", "时长": "60秒", "比例": "9:16竖屏", "重点": "外观设计、核心功能", "风格": "年轻化、动感" }, { "平台": "微博", "时长": "90秒", "比例": "16:9横屏", "重点": "技术参数、使用场景", "风格": "专业、详细" }, { "平台": "小红书", "时长": "45秒", "比例": "3:4", "重点": "时尚搭配、生活方式", "风格": "精致、生活化" }, { "平台": "B站", "时长": "3分钟", "比例": "16:9横屏", "重点": "深度测评、对比分析", "风格": "详细、客观" } ] }