304 lines
10 KiB
Python
304 lines
10 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)
|
|
|
|
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
|
|
print("\n📄 Step 1/6: Extracting data from PDF...")
|
|
pdf_data = self.extract_pdf()
|
|
|
|
# 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...")
|
|
script = self.generate_script(pdf_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)
|
|
|
|
# Step 5: Build render props
|
|
print("\n📊 Step 5/6: Building render props...")
|
|
render_props = self.build_render_props(pdf_data, script, broll_assets)
|
|
|
|
# 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"""
|
|
# TODO: Implement backend.extractor.extract_pdf_data
|
|
print(" [MOCK] Extracting PDF data...")
|
|
return {
|
|
"client_name": "山田太郎",
|
|
"current_age": 35,
|
|
"retirement_age": 65,
|
|
"life_expectancy": 100,
|
|
"current_assets": 5000,
|
|
"annual_income": 600,
|
|
"annual_expenses": 450,
|
|
}
|
|
|
|
def extract_video(self) -> Dict[str, Any]:
|
|
"""Extract data from video using Whisper (if video provided)"""
|
|
if not self.config.get("video_path"):
|
|
print(" [SKIP] No video file provided")
|
|
return {}
|
|
|
|
# TODO: Implement backend.extractor.extract_video_data
|
|
print(" [MOCK] Extracting video timestamps and transcription...")
|
|
return {
|
|
"transcription": "こんにちは...",
|
|
"timestamps": [
|
|
{"time": 0.5, "text": "Introduction", "keywords": ["保険", "将来"]},
|
|
{"time": 15.2, "text": "Main topic", "keywords": ["資産", "計画"]},
|
|
],
|
|
}
|
|
|
|
def generate_script(self, pdf_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
|
|
|
|
try:
|
|
return generate_script(pdf_data, video_data)
|
|
except Exception as e:
|
|
print(f" ⚠️ Script generation failed: {e}")
|
|
print(" Using fallback script...")
|
|
return {
|
|
"segments": [
|
|
{
|
|
"id": 1,
|
|
"text": "これからあなたの将来設計について説明します。",
|
|
"duration": 3.5,
|
|
"type": "narration",
|
|
},
|
|
{
|
|
"id": 2,
|
|
"text": "現在の資産状況を見てみましょう。",
|
|
"duration": 2.8,
|
|
"type": "chart_intro",
|
|
},
|
|
],
|
|
"total_duration": 6.3,
|
|
}
|
|
|
|
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
|
|
|
|
try:
|
|
return generate_audio_for_script(script_data, voice="alloy")
|
|
except Exception as e:
|
|
print(f" ⚠️ Audio generation failed: {e}")
|
|
print(" Continuing without audio...")
|
|
return script_data
|
|
|
|
def fetch_broll(self, script: Dict) -> Dict[str, Any]:
|
|
"""Fetch B-Roll footage from Pexels API"""
|
|
# TODO: Implement backend.media_fetcher.fetch_broll
|
|
print(" [MOCK] Fetching B-Roll from Pexels...")
|
|
return {
|
|
"retirement": "https://example.com/retirement.mp4",
|
|
"family": "https://example.com/family.mp4",
|
|
}
|
|
|
|
def build_render_props(
|
|
self, pdf_data: Dict, script: Dict, broll: Dict
|
|
) -> 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)
|
|
|
|
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": [],
|
|
},
|
|
}
|
|
|
|
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}",
|
|
]
|
|
|
|
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",
|
|
)
|
|
|
|
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",
|
|
)
|
|
|
|
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:
|
|
if not args.pdf:
|
|
parser.error("--pdf is required when --config is not provided")
|
|
config["pdf_path"] = args.pdf
|
|
config["video_path"] = args.video
|
|
|
|
if args.output:
|
|
config["output_filename"] = args.output
|
|
|
|
config["dry_run"] = args.dry_run
|
|
|
|
# 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()
|