481 lines
18 KiB
Python
481 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Auto Life Plan Video Generator - Main Entry Point
|
|
|
|
This script orchestrates the entire video generation pipeline:
|
|
1. Parse PDF/Video inputs (Gemini Pro / Whisper)
|
|
2. Generate script (Gemini Pro)
|
|
3. Fetch B-Roll assets (Pexels API)
|
|
4. Build chart data (Simulation engine)
|
|
5. Generate render_props.json
|
|
6. Trigger Remotion render
|
|
|
|
Usage:
|
|
python main.py --pdf path/to/report.pdf
|
|
python main.py --pdf report.pdf --video interview.mp4
|
|
python main.py --config config.json
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
|
|
# Import backend modules (to be implemented)
|
|
# from backend.extractor import extract_pdf_data, extract_video_data
|
|
# from backend.scripter import generate_script
|
|
# from backend.media_fetcher import fetch_broll
|
|
# from backend.data_builder import build_chart_data
|
|
|
|
|
|
class VideoGeneratorPipeline:
|
|
"""Main orchestrator for video generation"""
|
|
|
|
def __init__(self, config: Dict[str, Any]):
|
|
self.config = config
|
|
self.project_root = Path(__file__).parent
|
|
self.assets_dir = self.project_root / "assets"
|
|
self.input_dir = self.assets_dir / "0_input"
|
|
self.processed_dir = self.assets_dir / "1_processed"
|
|
self.output_dir = self.assets_dir / "2_output"
|
|
|
|
# Ensure directories exist
|
|
for dir_path in [self.input_dir, self.processed_dir, self.output_dir]:
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Auto-discover input files if not specified
|
|
self._auto_discover_inputs()
|
|
|
|
def _auto_discover_inputs(self):
|
|
"""
|
|
Auto-discover input files from assets/0_input/ directory
|
|
Priority: PDF > PPTX > Video
|
|
"""
|
|
print("🔍 Scanning input directory for files...")
|
|
|
|
# Find PDF files
|
|
if not self.config.get("pdf_path"):
|
|
pdf_files = list(self.input_dir.glob("*.pdf"))
|
|
if pdf_files:
|
|
self.config["pdf_path"] = str(pdf_files[0])
|
|
print(f" ✓ Found PDF: {pdf_files[0].name}")
|
|
|
|
# Find PPTX files
|
|
if not self.config.get("ppt_path"):
|
|
ppt_files = list(self.input_dir.glob("*.pptx"))
|
|
# Filter out temporary files (starting with ~$)
|
|
ppt_files = [f for f in ppt_files if not f.name.startswith("~$")]
|
|
if ppt_files:
|
|
self.config["ppt_path"] = str(ppt_files[0])
|
|
print(f" ✓ Found PPT: {ppt_files[0].name}")
|
|
|
|
# Find video files
|
|
if not self.config.get("video_path"):
|
|
video_extensions = ["*.mp4", "*.mov", "*.avi", "*.mkv"]
|
|
for ext in video_extensions:
|
|
video_files = list(self.input_dir.glob(ext))
|
|
if video_files:
|
|
self.config["video_path"] = str(video_files[0])
|
|
print(f" ✓ Found Video: {video_files[0].name}")
|
|
break
|
|
|
|
# Summary
|
|
if not any([self.config.get("pdf_path"), self.config.get("ppt_path")]):
|
|
print(" ⚠️ No PDF or PPT files found in input directory")
|
|
|
|
print()
|
|
|
|
def run(self) -> str:
|
|
"""Execute the full pipeline and return output video path"""
|
|
print("🚀 Starting Auto Life Plan Video Generation Pipeline...")
|
|
|
|
# Step 1: Extract data from PDF or PPT
|
|
print("\n📄 Step 1/6: Extracting data from documents...")
|
|
pdf_data = self.extract_pdf()
|
|
ppt_data = self.extract_ppt()
|
|
|
|
# Step 2: Extract data from video (if provided)
|
|
print("\n🎥 Step 2/6: Processing video footage...")
|
|
video_data = self.extract_video()
|
|
|
|
# Step 3: Generate narration script
|
|
print("\n✍️ Step 3/6: Generating narration script...")
|
|
props_file = self.processed_dir / "render_props.json"
|
|
|
|
# FEATURE: Use existing script if skip_script is enabled to preserve manual edits
|
|
if self.config.get("skip_script") and props_file.exists():
|
|
print(f" ♻️ PRESERVING existing script from {props_file.name}")
|
|
with open(props_file, "r", encoding="utf-8") as f:
|
|
existing_props = json.load(f)
|
|
script = {"segments": existing_props.get("script", [])}
|
|
else:
|
|
script = self.generate_script(pdf_data, ppt_data, video_data)
|
|
|
|
# Step 3.5: Generate audio from script (NEW)
|
|
print("\n🎙️ Step 3.5/6: Generating audio narration...")
|
|
script = self.generate_audio(script)
|
|
|
|
# Step 4: Fetch B-Roll assets
|
|
print("\n🎬 Step 4/6: Fetching B-Roll assets...")
|
|
broll_assets = self.fetch_broll(script, video_data)
|
|
|
|
# Step 5: Build render props
|
|
print("\n📊 Step 5/6: Building render props...")
|
|
render_props = self.build_render_props(pdf_data, ppt_data, script, broll_assets, video_data)
|
|
|
|
# Save render_props.json
|
|
props_file = self.processed_dir / "render_props.json"
|
|
with open(props_file, "w", encoding="utf-8") as f:
|
|
json.dump(render_props, f, ensure_ascii=False, indent=2)
|
|
print(f" ✓ Saved render props to {props_file}")
|
|
|
|
# Step 6: Render video with Remotion
|
|
print("\n🎞️ Step 6/6: Rendering video with Remotion...")
|
|
output_path = self.render_video(props_file)
|
|
|
|
print(f"\n✅ Video generation complete! Output: {output_path}")
|
|
return str(output_path)
|
|
|
|
def extract_pdf(self) -> Dict[str, Any]:
|
|
"""Extract data from PDF using Gemini Pro"""
|
|
pdf_path = self.config.get("pdf_path")
|
|
|
|
if not pdf_path or not Path(pdf_path).exists():
|
|
print(" [SKIP] No PDF file found")
|
|
return {}
|
|
|
|
print(f" 📄 Processing PDF: {Path(pdf_path).name}")
|
|
|
|
try:
|
|
from backend.extractor import extract_pdf_data
|
|
pdf_data = extract_pdf_data(pdf_path)
|
|
print(f" ✓ Successfully extracted data for: {pdf_data.get('client_name', 'Unknown')}")
|
|
return pdf_data
|
|
except Exception as e:
|
|
print(f" ⚠️ PDF extraction failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {}
|
|
|
|
def extract_ppt(self) -> Dict[str, Any]:
|
|
"""Extract slides from PowerPoint presentation"""
|
|
ppt_path = self.config.get("ppt_path")
|
|
|
|
if not ppt_path or not Path(ppt_path).exists():
|
|
print(" [SKIP] No PPT file found")
|
|
return {}
|
|
|
|
print(f" 📊 Processing PPT: {Path(ppt_path).name}")
|
|
|
|
try:
|
|
from backend.ppt_extractor import extract_ppt_data
|
|
ppt_data = extract_ppt_data(ppt_path)
|
|
print(f" ✓ Successfully extracted {ppt_data.get('total_slides', 0)} slides")
|
|
return ppt_data
|
|
except Exception as e:
|
|
print(f" ⚠️ PPT extraction failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {}
|
|
|
|
def extract_video(self) -> Dict[str, Any]:
|
|
"""Extract data from video using Whisper (if video provided)"""
|
|
video_path = self.config.get("video_path")
|
|
|
|
if not video_path or not Path(video_path).exists():
|
|
print(" [SKIP] No video file provided or file not found")
|
|
return {}
|
|
|
|
try:
|
|
from backend.extractor import extract_video_data
|
|
video_data = extract_video_data(video_path)
|
|
|
|
# Optionally extract video clips
|
|
if self.config.get("extract_clips", False):
|
|
print(" ✂️ Extracting key video clips...")
|
|
from backend.video_slicer import extract_video_clips
|
|
clips = extract_video_clips(video_path, video_data, max_clips=3)
|
|
video_data["clips"] = clips
|
|
|
|
return video_data
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ Video extraction failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {}
|
|
|
|
def generate_script(self, pdf_data: Dict, ppt_data: Dict, video_data: Dict) -> Dict[str, Any]:
|
|
"""Generate narration script using Gemini Pro"""
|
|
print(" ✍️ Generating script with Gemini Pro...")
|
|
from backend.scripter import generate_script
|
|
|
|
# Merge PPT and PDF data
|
|
combined_data = {**pdf_data}
|
|
if ppt_data:
|
|
combined_data["presentation"] = ppt_data
|
|
|
|
try:
|
|
script = generate_script(combined_data, video_data)
|
|
print(f" ✓ Generated {len(script.get('segments', []))} script segments")
|
|
return script
|
|
except Exception as e:
|
|
print(f" ❌ Script generation failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise
|
|
|
|
def generate_audio(self, script_data: Dict) -> Dict[str, Any]:
|
|
"""Generate audio narration from script using OpenAI TTS"""
|
|
from backend.audio_generator import generate_audio_for_script
|
|
from backend.video_processor.utils import get_video_duration
|
|
|
|
try:
|
|
# 1. Generate TTS audio for non-video segments
|
|
script_data = generate_audio_for_script(script_data, voice="nova")
|
|
# voice options: "alloy", "echo", "fable", "onyx", "nova", "shimmer"
|
|
|
|
# 2. Sync duration for video segments with actual clip duration
|
|
print(" 🎬 Syncing durations for interview video segments...")
|
|
video_data = self.extract_video() # Get cached video data with clips
|
|
clips = video_data.get("clips", [])
|
|
|
|
for segment in script_data.get("segments", []):
|
|
if segment.get("visualType") == "video":
|
|
idx = segment.get("visualAssetIndex", -1)
|
|
if 0 <= idx < len(clips):
|
|
clip = clips[idx]
|
|
clip_path = clip.get("outputPath") or clip.get("filepath")
|
|
if clip_path and Path(clip_path).exists():
|
|
actual_duration = get_video_duration(clip_path)
|
|
print(f" ✓ Video Segment {segment['id']}: Updated duration to {actual_duration:.2f}s (Clip {idx})")
|
|
segment["duration"] = actual_duration
|
|
|
|
return script_data
|
|
except Exception as e:
|
|
print(f" ⚠️ Audio generation failed: {e}")
|
|
print(" Continuing without audio...")
|
|
return script_data
|
|
|
|
def fetch_broll(self, script: Dict, video_data: Dict = None) -> Dict[str, Any]:
|
|
"""Fetch B-Roll footage from Pexels API based on script context"""
|
|
try:
|
|
from backend.media_fetcher import MediaFetcher
|
|
|
|
# 1. Extract keywords explicitly requested in script segments
|
|
keywords = set()
|
|
for segment in script.get("segments", []):
|
|
# ENHANCED: Collect keywords from ANY segment that has one,
|
|
# as our 3-column layout can show B-Roll alongside PPT/Charts
|
|
if segment.get("brollKeyword"):
|
|
keywords.add(segment.get("brollKeyword"))
|
|
|
|
# 2. Add keywords from video data as fallback/supplement
|
|
if video_data and "keywords" in video_data:
|
|
for kw in video_data["keywords"][:3]:
|
|
keywords.add(kw)
|
|
|
|
# 3. Use default keywords if still empty
|
|
if not keywords:
|
|
keywords.update(["家族", "老後", "貯蓄"])
|
|
|
|
keyword_list = list(keywords)
|
|
print(f" 🎬 Fetching B-Roll for context: {', '.join(keyword_list)}")
|
|
|
|
fetcher = MediaFetcher()
|
|
broll_results = fetcher.fetch_broll(keyword_list, max_per_keyword=1)
|
|
|
|
# Convert to simple filepath mapping
|
|
broll_assets = {}
|
|
for keyword, videos in broll_results.items():
|
|
if videos:
|
|
# Store only the filename to be used with staticFile() in Remotion
|
|
broll_assets[keyword] = Path(videos[0]["filepath"]).name
|
|
|
|
return broll_assets
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ B-Roll fetching failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {}
|
|
|
|
def build_render_props(
|
|
self, pdf_data: Dict, ppt_data: Dict, script: Dict, broll: Dict, video_data: Dict = None
|
|
) -> Dict[str, Any]:
|
|
"""Build the complete render_props.json structure"""
|
|
print(" Building chart data and render props...")
|
|
|
|
# Use the financial simulator to build chart data that matches the
|
|
# TypeScript RenderProps contract (frontend/src/render-props-types.ts)
|
|
from backend.data_builder import build_chart_data
|
|
|
|
charts = build_chart_data(pdf_data) if pdf_data else {}
|
|
|
|
# Add video clips if available
|
|
video_clips = []
|
|
if video_data and "clips" in video_data:
|
|
video_clips = video_data["clips"]
|
|
|
|
# Add PPT slides if available
|
|
ppt_slides = []
|
|
if ppt_data and "slides" in ppt_data:
|
|
ppt_slides = [
|
|
{
|
|
"index": slide["index"],
|
|
"title": slide["title"],
|
|
"image_path": slide["image_path"].replace("\\", "/"),
|
|
}
|
|
for slide in ppt_data["slides"]
|
|
]
|
|
|
|
return {
|
|
"meta": {
|
|
"title": f"{pdf_data.get('client_name', 'お客様')}様の人生設計",
|
|
"currencyUnit": "万円",
|
|
"duration": 180, # 3 minutes
|
|
"clientName": pdf_data.get("client_name"),
|
|
},
|
|
"charts": charts,
|
|
"script": script["segments"],
|
|
"assets": {
|
|
"broll": broll,
|
|
"videoClips": video_clips,
|
|
"slides": ppt_slides,
|
|
},
|
|
}
|
|
|
|
def render_video(self, props_file: Path) -> Path:
|
|
"""Render video using Remotion"""
|
|
output_file = self.output_dir / "life_plan_video.mp4"
|
|
|
|
# If dry-run flag is set, skip rendering step
|
|
if self.config.get("dry_run"):
|
|
print(" [SKIP] Dry run enabled - not rendering video with Remotion")
|
|
return output_file
|
|
|
|
# Remotion render command
|
|
# Use "npx.cmd" on Windows to ensure subprocess can find it
|
|
npx_executable = "npx.cmd" if sys.platform == "win32" else "npx"
|
|
cmd = [
|
|
npx_executable,
|
|
"remotion",
|
|
"render",
|
|
"src/index.ts",
|
|
"MainVideo",
|
|
str(output_file),
|
|
f"--props={props_file}",
|
|
"--concurrency=1",
|
|
]
|
|
|
|
print(f" Running: {' '.join(cmd)}")
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True, cwd=self.project_root, shell=True)
|
|
except subprocess.CalledProcessError as e:
|
|
print(f" ❌ Remotion render failed: {e}")
|
|
sys.exit(1)
|
|
|
|
return output_file
|
|
|
|
|
|
def main():
|
|
"""CLI entry point"""
|
|
parser = argparse.ArgumentParser(
|
|
description="Auto Life Plan Video Generator",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
python main.py --pdf assets/0_input/report.pdf
|
|
python main.py --pdf report.pdf --video interview.mp4
|
|
python main.py --config config.json --output custom_output.mp4
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--pdf",
|
|
type=str,
|
|
help="Path to the Life Plan PDF report (optional if files exist in assets/0_input/)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--ppt",
|
|
type=str,
|
|
help="Path to the PowerPoint presentation (optional if files exist in assets/0_input/)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--video",
|
|
type=str,
|
|
help="Path to the interview video (optional)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--config",
|
|
type=str,
|
|
help="Path to configuration JSON file",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--output",
|
|
type=str,
|
|
help="Custom output filename (default: life_plan_video.mp4)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Generate render_props.json without rendering",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--extract-clips",
|
|
action="store_true",
|
|
help="Extract video clips from source video (optional)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--skip-script",
|
|
action="store_true",
|
|
help="Skip AI script generation and use existing script from render_props.json (preserves manual edits)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Build configuration
|
|
config = {}
|
|
|
|
if args.config:
|
|
with open(args.config, "r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
else:
|
|
# Only require --pdf if no files found in input directory
|
|
config["pdf_path"] = args.pdf
|
|
config["ppt_path"] = args.ppt
|
|
config["video_path"] = args.video
|
|
|
|
if args.output:
|
|
config["output_filename"] = args.output
|
|
|
|
config["dry_run"] = args.dry_run
|
|
config["extract_clips"] = args.extract_clips
|
|
config["skip_script"] = args.skip_script
|
|
|
|
# Run pipeline
|
|
pipeline = VideoGeneratorPipeline(config)
|
|
|
|
try:
|
|
output_path = pipeline.run()
|
|
print(f"\n🎉 Success! Video saved to: {output_path}")
|
|
except Exception as e:
|
|
print(f"\n❌ Pipeline failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|