import tempfile from typing import Sequence import streamlit as st from ..adapters.aws.s3 import S3Client from ..adapters.opus_clip.enums import LayoutAspectRatio, ClipDuration, ProjectStage from .enums import Page from ..assets.models import Asset from ..assets.services import AssetService from ..opus_clip.models import OpusClipProject from ..opus_clip.services import OpusClipService, ClipGenerationRequest def render_clip_it_page( opus_service: OpusClipService, asset_service: AssetService, s3_client: S3Client ): asset: Asset = st.session_state.asset artist: str = st.session_state.artist_name or "" projects: Sequence[OpusClipProject] = st.session_state.projects """Renders the 'Clip It' page for a selected song with multiple sections.""" st.title(f"✂️ Clip It: {asset.asset_meta_data.asset_name}") st.subheader(f"by {artist}") if st.button("⬅️ Back to Search", use_container_width=True): st.session_state.page = Page.ARTIST_SEARCH st.rerun() st.markdown("---") # --- Section 1: Original Video --- st.subheader("Original Video") st.video(asset.s3_key) st.markdown("---") if st.button("⬅️ Refresh Clips", use_container_width=True): for project in projects: if project.status != ProjectStage.COMPLETE: opus_service.update_clip_project(project) if project.status == ProjectStage.COMPLETE: for clip in project.clips: with tempfile.TemporaryDirectory() as temp_dir: local_video_path = asset_service.download_video( url=clip.s3_key, name=clip.id, target_directory=temp_dir ) s3_key = f"clips/{asset.global_participant_id}/{project.id}/{clip.id}.mp4" print(local_video_path) print(s3_key) s3_client.upload_file(local_video_path, s3_key) # --- Section 2: Existing Clips --- render_clips(projects=projects, global_participant_id=asset.global_participant_id) st.markdown("---") # --- Section 3: Generate New Clip --- st.subheader("Generate New Clip") with st.form(key="clip_form"): title = st.text_input("Title", placeholder=asset.asset_meta_data.asset_name) description = st.text_input("Description") layout = st.selectbox("Clip Layout", options=LayoutAspectRatio.list()) duration = st.selectbox("Clip Duration (seconds)", options=ClipDuration.list()) prompt = st.text_area("Clip Generation Prompt") # Submit button for the form submitted = st.form_submit_button("Generate Clip") if submitted: st.success("Clip generation started!") st.write("Captured Data:") project = opus_service.create_clip_project( asset=asset, request=ClipGenerationRequest( title=title.strip(), description=description.strip(), layout=LayoutAspectRatio(layout), duration=ClipDuration(duration), prompt=prompt.strip(), ), ) st.json(project.__dict__) def render_clips( projects: Sequence[OpusClipProject], global_participant_id: str ) -> None: st.subheader("Existing Clips") for project_key, project in enumerate(projects): st.subheader(f"Project #{project_key + 1}: {project.name}") st.markdown( f"**Created at (UTC):** {project.created_at.strftime('%Y-%B-%d %H:%M:%S')}" ) st.markdown(f"**Status:** {project.status}") st.markdown(f"**Description:** {project.description}") st.markdown( f"**Prompt:** {project.parameters.get('curation_pref', {}).get('custom_prompt', 'None')}" ) skip = False sorted_clips = sorted(project.clips, key=lambda x: x.external_id) for key, clip in enumerate(sorted_clips): if not skip: col1, col2 = st.columns(2, border=True) url = clip.url(global_participant_id) try: clip_2 = project.clips[key + 1] url_2 = clip_2.url(global_participant_id) skip = True except IndexError: clip_2 = None skip = False url_2 = None with col1: st.markdown(f"**CLIP #{key + 1}**") st.video(url) st.link_button("Download Clip", url, use_container_width=True) with col2: if clip_2 and url_2: st.markdown( f"**CLIP #{key + 2}**", ) st.video(url_2) st.link_button("Download Clip", url_2, use_container_width=True) else: skip = False st.markdown("---")