Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,11 @@ uv run python media_gen/video_regen_pipeline.py --video-path video.mp4 --user-in
- `--output-folder`: Output directory (default: `~/Downloads`)
- `--screenshot-interval`: Seconds between screenshots (default: `2.0`)

**Output:**
- Individual generated videos are saved in `~/Downloads/video_regen_[timestamp]/generated_videos/`
- Final concatenated video is saved in `~/Downloads/video_regen_[timestamp]_[session_id].mp4`
- The pipeline provides clear logging showing the exact file locations

## Development

### Adding Dependencies
Expand Down
2 changes: 1 addition & 1 deletion media_gen/image_regen_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@

Example:
python image_regen_pipeline.py --image-path \
./examples/media-gen/integration_tests/test_image.png \
media_gen/test_scripts/test_image.png \
--user-interests "Users like cute cats and capybara"
"""

Expand Down
50 changes: 43 additions & 7 deletions media_gen/video_regen_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@

Example:
python video_regen_pipeline.py --video-path \
./examples/media-gen/integration_tests/test_video.mp4 \
media_gen/test_scripts/test_video.mp4 \
--user-interests "Users like cinematic style with dramatic lighting"
"""

Expand Down Expand Up @@ -458,12 +458,25 @@ def _prepare_video_concatenation(self, tool_input: Dict[str, Any]) -> Dict[str,
output_folder = tool_input.get("output_folder", "~/Downloads")
output_format = tool_input.get("output_format", "mp4")

# Save concatenated video directly in the main output folder, not a subfolder
concatenated_video_path = f"{output_folder}/final_concatenated_video.{output_format}"
# Save concatenated video in the Downloads folder with a descriptive name
# output_folder is: ~/Downloads/video_regen_123/generated_videos
# parent is: ~/Downloads/video_regen_123 (session folder)
# grandparent is: ~/Downloads (base folder)
session_folder = os.path.dirname(output_folder) # video_regen_123 folder
base_folder = os.path.dirname(session_folder) # Downloads folder

# Create a descriptive filename with timestamp
from datetime import datetime

timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
session_id = os.path.basename(session_folder) # video_regen_123
concatenated_video_path = f"{base_folder}/video_regen_{timestamp}_{session_id}.{output_format}"

# Display concatenation info
print("\n🎬 VIDEO CONCATENATION STEP")
print(f"📁 Output folder: {output_folder}")
print(f"📁 Generated videos folder: {output_folder}")
print(f"📁 Final video output: {concatenated_video_path}")
print(f"📂 Absolute path: {os.path.abspath(concatenated_video_path)}")
print(f"🎞️ Concatenating {len(generated_videos)} videos:")

for i, video_path in enumerate(generated_videos):
Expand All @@ -483,6 +496,7 @@ def _prepare_video_concatenation(self, tool_input: Dict[str, Any]) -> Dict[str,
"output_folder": output_folder,
"output_path": concatenated_video_path,
"output_format": output_format,
"base_folder": base_folder,
}

def _extract_concatenation_result(self, tool_output: Dict[str, Any]) -> Dict[str, Any]:
Expand All @@ -495,6 +509,7 @@ def _extract_concatenation_result(self, tool_output: Dict[str, Any]) -> Dict[str
if concatenated_video_path and os.path.exists(concatenated_video_path):
file_size = os.path.getsize(concatenated_video_path)
print(f"✅ Concatenated video created: {os.path.basename(concatenated_video_path)}")
print(f"📂 Full path: {os.path.abspath(concatenated_video_path)}")
print(f"📏 File size: {file_size:,} bytes ({file_size / 1024 / 1024:.1f} MB)")

if concatenation_info:
Expand Down Expand Up @@ -535,15 +550,17 @@ def _execute_video_concatenation(self, step_input: Dict[str, Any]) -> Dict[str,
video_paths = step_input.get("video_paths", [])
output_folder = step_input.get("output_folder", "~/Downloads")
output_path = step_input.get("output_path", "")
base_folder = step_input.get("base_folder", os.path.dirname(os.path.dirname(output_folder)))

if not video_paths:
return {
"concatenated_video_path": "",
"concatenation_info": {"success": False, "error_message": "No videos to concatenate"},
}

# Create output directory
# Create output directories
os.makedirs(output_folder, exist_ok=True)
os.makedirs(base_folder, exist_ok=True)

# Create a temporary folder with the videos for concatenation
import shutil
Expand Down Expand Up @@ -725,10 +742,10 @@ def main():
parser = argparse.ArgumentParser(description="Regenerate a video based on user interests")
parser.add_argument(
"--video-path",
default="./examples/media-gen/integration_tests/test_video.mp4",
default="media_gen/test_scripts/test_video.mp4",
help=(
"Path to the original video file (supports ~ for home directory). "
"Default: ./examples/media-gen/integration_tests/test_video.mp4"
"Default: media_gen/test_scripts/test_video.mp4"
),
)
parser.add_argument(
Expand Down Expand Up @@ -871,6 +888,7 @@ def main():
if concatenated_video_path and os.path.exists(concatenated_video_path):
print("\n🎬 FINAL CONCATENATED VIDEO:")
print(f"📁 File: {concatenated_video_path}")
print(f"📂 Absolute path: {os.path.abspath(concatenated_video_path)}")

# Show relative path if it's in Downloads
downloads_path = os.path.expanduser("~/Downloads")
Expand All @@ -892,6 +910,14 @@ def main():
# Show file size
file_size = os.path.getsize(concatenated_video_path)
print(f"📏 File size: {file_size:,} bytes ({file_size / 1024 / 1024:.1f} MB)")

# Show command to open file location
print(f"🔍 To open file location: open -R '{os.path.abspath(concatenated_video_path)}'")

# Check if old format file exists and mention it
old_file_path = os.path.expanduser("~/final_concatenated_video.mp4")
if os.path.exists(old_file_path):
print(f"📝 Note: Previous video found at: {old_file_path}")
elif concatenation_info and not concatenation_info.get("success", True):
print(f"\n❌ Video concatenation failed: {concatenation_info.get('error_message', 'Unknown error')}")
else:
Expand Down Expand Up @@ -963,6 +989,16 @@ def main():
for error in video_errors:
print(f" Video: {error}")

# Final summary with prominent file location
if concatenated_video_path and os.path.exists(concatenated_video_path):
print("\n" + "=" * 60)
print("🎬 YOUR FINAL VIDEO IS READY!")
print("=" * 60)
print(f"📁 Location: {os.path.abspath(concatenated_video_path)}")
print(f"📂 Filename: {os.path.basename(concatenated_video_path)}")
print(f"🔍 Open in Finder: open -R '{os.path.abspath(concatenated_video_path)}'")
print("=" * 60)

print("=" * 60)

except Exception as e:
Expand Down