split out diarization and download

Author: kendallwerts

.github/workflows/python-app.yml

@@ -46,4 +46,4 @@ jobs:
 
     - name: Run Diarization
       run: |
-        /opt/poetry-venv/bin/python -m flows.translate_meetings
+        /opt/poetry-venv/bin/python -m flows.transcribe_meetings

db/migrate_s3_path.py

@@ -0,0 +1,31 @@
+"""
+Migration script to add s3_path field to existing meeting records.
+"""
+
+from dyntastic import A
+from src.models.meeting import Meeting
+
+
+def migrate_s3_path():
+    """
+    Add s3_path field to all existing meeting records that don't have it.
+    """
+    print("Starting migration to add s3_path field to existing meetings...")
+
+    # Get all meetings
+    meetings = Meeting.scan()
+    updated_count = 0
+
+    for meeting in meetings:
+        # Check if s3_path field exists and is None
+        if not hasattr(meeting, "s3_path") or meeting.s3_path is None:
+            print(f"Updating meeting: {meeting.meeting} ({meeting.date})")
+            meeting.s3_path = None
+            meeting.save()
+            updated_count += 1
+
+    print(f"Migration complete. Updated {updated_count} meetings.")
+
+
+if __name__ == "__main__":
+    migrate_s3_path()

db/queries.py

@@ -4,7 +4,9 @@
 from src.models.meeting import Meeting
 
 
-def get_meetings(days: int = 7, video: Optional[bool] = None) -> List[Meeting]:
+def get_meetings(
+    days: int = 7, video: Optional[bool] = None, s3_path: Optional[bool] = None
+) -> List[Meeting]:
     """
     Get meetings that occurred in the past number of days from now.
     """
@@ -13,6 +15,11 @@ def get_meetings(days: int = 7, video: Optional[bool] = None) -> List[Meeting]:
     meetings = Meeting.scan(
         A.date >= target_date,
     )
-    meetings_list = [m for m in meetings if (video is None or bool(m.video) == video)]
+    meetings_list = [
+        m
+        for m in meetings
+        if (video is None or bool(m.video) == video)
+        and (s3_path is None or bool(m.s3_path) == s3_path)
+    ]
 
     return list(meetings_list)

flows/translate_meetings.py flows/download_meetings.pyflows/translate_meetings.py renamed to flows/download_meetings.py

@@ -1,21 +1,21 @@
 from prefect import flow
 
 from db.queries import get_meetings
-from tasks.diarize import diarize_meeting
+from tasks.diarize import download_video_and_put_in_s3
 from tasks.meetings import register_meetings
 
 
-@flow(log_prints=True)
-def translate_meetings():
+# @flow(log_prints=True)
+def download_meetings():
     new_meetings = register_meetings()
     print(f"Registered {len(new_meetings)} new meetings")
-    meetings_to_diarize = get_meetings(video=True)
-    print(f"Found {len(meetings_to_diarize)} meetings to diarize")
-    for meeting in meetings_to_diarize:
-        diarize_meeting(meeting)
+    meetings_to_download = get_meetings(days=7, video=True, s3_path=False)
+    print(f"Found {len(meetings_to_download)} meetings to download")
+    for meeting in meetings_to_download:
+        download_video_and_put_in_s3(meeting)
     # new_subtitled_video_pages = await create_subtitled_video_pages(new_transcribed_meetings)
     # new_translated_meetings = await translate_transcriptions(new_transcribed_meetings)
 
 
 if __name__ == "__main__":
-    translate_meetings()
+    download_meetings()

flows/transcribe_meetings.py

@@ -0,0 +1,16 @@
+from prefect import flow
+
+from db.queries import get_meetings
+from tasks.diarize import diarize_meeting
+
+
+@flow(log_prints=True)
+def transcribe_meetings():
+    meetings_to_diarize = get_meetings(video=True, s3_path=True)
+    print(f"Found {len(meetings_to_diarize)} meetings to diarize")
+    for meeting in meetings_to_diarize:
+        diarize_meeting(meeting)
+
+
+if __name__ == "__main__":
+    transcribe_meetings()

src/aws.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 
 import boto3
 from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError
@@ -52,3 +53,21 @@ def save_content_to_s3(content, bucket_name, s3_key, content_type):
     region = s3_client.meta.region_name
     url = f"https://{bucket_name}.s3.{region}.amazonaws.com/{s3_key}"
     return HttpUrl(url)
+
+
+def get_video_from_s3(bucket_name, s3_path):
+    try:
+        # Create output directory if it doesn't exist
+        output_dir = Path("data/video")
+        output_dir.mkdir(parents=True, exist_ok=True)
+
+        # Define output path
+        output_path = output_dir / Path(s3_path).name
+
+        # Download file from S3
+        s3_client.download_file(bucket_name, s3_path, str(output_path))
+        print(f"Downloaded {s3_path} from S3 to {output_path}")
+        return output_path
+    except ClientError as e:
+        print(f"Failed to get video from S3: {str(e)}")
+        return None

src/models/meeting.py

@@ -5,7 +5,7 @@
 from typing import Optional
 
 from dyntastic import Dyntastic
-from pydantic import BaseModel, Field, HttpUrl
+from pydantic import BaseModel, Field, HttpUrl, ConfigDict
 from datetime import datetime
 from typing import List
 
@@ -18,6 +18,8 @@ class Meeting(Dyntastic):
     __table_name__ = "tgov-meeting"
     __hash_key__ = "clip_id"
 
+    model_config = ConfigDict(extra="ignore")
+
     clip_id: Optional[str] = Field(None, description="Granicus clip ID")
     meeting: str = Field(description="Name of the meeting")
     date: datetime = Field(description="Date and time of the meeting")
@@ -30,6 +32,9 @@ class Meeting(Dyntastic):
     subtitles: Optional[List[HttpUrl]] = Field(
         None, description="URLs to the meeting subtitle tracks"
     )
+    s3_path: Optional[str] = Field(
+        default=None, description="S3 path to the meeting video"
+    )
 
     def __str__(self) -> str:
         """String representation of the meeting"""

src/run_diarization.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 
 from src.aws import save_content_to_s3
-from src.models.meeting import GranicusPlayerPage
+from src.models.meeting import GranicusPlayerPage, Meeting
 from src.granicus import get_video_player
 from src.videos import download_file, transcribe_video_with_diarization
 
@@ -38,7 +38,7 @@ def download_video(file_name: str, video_url: str):
     return video_file
 
 
-def run_diarization(video_file: Path):
+def run_diarization(video_file: Path, meeting: Meeting):
     transcription_dir = Path("data/transcripts")
 
     transcription = asyncio.run(
@@ -51,6 +51,9 @@ def run_diarization(video_file: Path):
         f"{FOLDER_NAME}/{video_file.name}.json",
         "application/json",
     )
+    meeting.transcripts = [f"{FOLDER_NAME}/{video_file.name}.json"]
+    meeting.save()
+
     print(transcription)
 
 

src/videos.py

@@ -79,8 +79,6 @@ def download_file(url: str, output_path: Path):
                         )
 
     print(f"Download complete: {url}")
-    # Add to S3
-    upload_to_s3(output_path, BUCKET_NAME, f"{FOLDER_NAME}/{output_path.name}")
     return output_path
 
 

tasks/diarize.py

@@ -1,14 +1,35 @@
+import os
+from src.aws import get_video_from_s3, upload_to_s3
 from src.run_diarization import download_video, run_diarization
 from prefect import task
 
 from src.models.meeting import Meeting
 
 
+BUCKET_NAME = os.getenv("S3_BUCKET")
+FOLDER_NAME = "videos"
+
+
+# @task
+def download_video_and_put_in_s3(meeting: Meeting):
+    video_file = download_video(f"{meeting.meeting}_{meeting.date}", str(meeting.video))
+    if video_file:
+        print(f"Uploading video to S3: {video_file}")
+        s3_path = f"{FOLDER_NAME}/{video_file.name}"
+        upload_to_s3(video_file, BUCKET_NAME, f"{FOLDER_NAME}/{video_file.name}")
+        print(f"Uploaded video to S3: {s3_path}")
+        print("Saving meeting.")
+        meeting.s3_path = s3_path
+        meeting.save()
+    else:
+        print("Video file not found")
+
+
 @task
 def diarize_meeting(meeting: Meeting):
-    video_file = download_video(f"{meeting.meeting}_{meeting.date}", str(meeting.video))
+    video_file = get_video_from_s3(BUCKET_NAME, meeting.s3_path)
     if video_file:
-        run_diarization(video_file)
+        run_diarization(video_file, meeting)
     else:
         print("Video file not found")
     # TODO: Update meeting with transcript location