start diarization flow

Author: kendallwerts

.github/workflows/python-app.yml

@@ -2,7 +2,7 @@ name: Build/run tgov
 
 on:
   push:
-    branches: [ "main", "deploy-lambda" ]
+    branches: [ "main", "deploy-lambda", "test-flows" ]
   pull_request:
     branches: [ "main" ]
 

Dockerfile

@@ -23,4 +23,4 @@ RUN poetry install
 ENV PYTHONPATH=/app
 ENV PATH="${POETRY_VENV}/bin:${PATH}"
 
-CMD ["python", "src/run_diarization.py"]
+CMD ["python", "-m", "flows.translate_meetings"]

db/queries.py

@@ -1,10 +1,10 @@
 from datetime import datetime, timedelta
-from typing import Sequence
+from typing import List, Optional
 from dyntastic import A
 from src.models.meeting import Meeting
 
 
-def get_meetings(days: int = 7) -> Sequence[Meeting]:
+def get_meetings(days: int = 7, video: Optional[bool] = None) -> List[Meeting]:
     """
     Get meetings that occurred in the past number of days from now.
     """
@@ -13,5 +13,6 @@ def get_meetings(days: int = 7) -> Sequence[Meeting]:
     meetings = Meeting.scan(
         A.date >= target_date,
     )
+    meetings_list = [m for m in meetings if (video is None or bool(m.video) == video)]
 
-    return list(meetings)
+    return list(meetings_list)

flows/add_subtitles.py

@@ -1,3 +1,4 @@
+import asyncio
 from prefect import flow
 from dyntastic import A
 from pydantic import HttpUrl
@@ -11,30 +12,41 @@
 
 
 @flow(log_prints=True)
-async def add_subtitles():
+def add_subtitles():
+    print("Getting meetings")
     meetings = get_meetings(days=90)
     meetings_with_transcripts = [
         meeting
         for meeting in meetings
         if hasattr(meeting, "transcripts") and meeting.transcripts is not None or []
     ]
+    print(f"Found {len(meetings_with_transcripts)} meetings with transcripts")
     for meeting in meetings_with_transcripts:
         for transcript_url in meeting.transcripts:
-            transcript_data = await async_get_json(transcript_url.encoded_string())
+            print(f"Processing {transcript_url}")
+            transcript_data = asyncio.run(
+                async_get_json(transcript_url.encoded_string())
+            )
+            print(f"Transcript data: {transcript_data}")
             transcript = Transcript.model_validate(transcript_data)
             language = transcript.language
             if f"{language}.vtt" in meeting.subtitles:
                 continue
-            track_content = await create_vtt_track(
-                transcript,
-                include_speaker_prefix=False,
+            print(f"Creating VTT track for {language}")
+            track_content = asyncio.run(
+                create_vtt_track(
+                    transcript,
+                    include_speaker_prefix=False,
+                )
             )
+            print(f"Saving VTT track to S3")
             result: HttpUrl = save_content_to_s3(
                 track_content,
                 "tgov-assets",
                 f"{meeting.filename()}/subtitles/{language}.vtt",
                 "text/vtt",
             )
+            print(f"VTT track saved to S3")
             if not meeting.subtitles:
                 meeting.subtitles = [result]
             else:
@@ -43,10 +55,11 @@ async def add_subtitles():
                     if result not in meeting.subtitles
                     else None
                 )
+            print(f"Saving meeting to DynamoDB")
             meeting.save()
+            print(f"Meeting saved to DynamoDB")
 
 
 if __name__ == "__main__":
-    import asyncio
-
-    asyncio.run(add_subtitles())
+    print("Starting add_subtitles")
+    add_subtitles()

flows/translate_meetings.py

@@ -1,15 +1,21 @@
 from prefect import flow
 
-from tasks.meetings import get_new_meetings
+from db.queries import get_meetings
+from tasks.diarize import diarize_meeting
+from tasks.meetings import register_meetings
 
 
 @flow(log_prints=True)
-async def translate_meetings():
-    new_meetings = await get_new_meetings()
-    # new_transcribed_meetings = await transcribe_videos(new_meetings)
+def translate_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)
     # new_subtitled_video_pages = await create_subtitled_video_pages(new_transcribed_meetings)
     # new_translated_meetings = await translate_transcriptions(new_transcribed_meetings)
 
+
 if __name__ == "__main__":
-    import asyncio
-    asyncio.run(translate_meetings())
+    translate_meetings()

src/aws.py

@@ -7,7 +7,8 @@
 
 def is_aws_configured():
     required_vars = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_DEFAULT_REGION"]
-    return all(var in os.environ for var in required_vars)
+    return True
+    # return all(var in os.environ for var in required_vars)
 
 
 s3_client = boto3.client("s3")

src/local_store.py

@@ -1,20 +1,22 @@
 import json
 import os
-from typing import Sequence
+from typing import List
 from src.models.meeting import Meeting
 
 LOCAL_STORE_PATH = "data/meetings.json"
 
-def read_meetings() -> Sequence[Meeting]:
+
+def read_meetings() -> List[Meeting]:
     if not os.path.exists(LOCAL_STORE_PATH):
         return []
 
-    with open(LOCAL_STORE_PATH, 'r') as f:
+    with open(LOCAL_STORE_PATH, "r") as f:
         data = json.load(f)
         return [Meeting(**meeting) for meeting in data]
 
-def write_meetings(meetings: Sequence[Meeting]):
+
+def write_meetings(meetings: List[Meeting]):
     os.makedirs(os.path.dirname(LOCAL_STORE_PATH), exist_ok=True)
-    with open(LOCAL_STORE_PATH, 'w') as f:
+    with open(LOCAL_STORE_PATH, "w") as f:
         json_data = [meeting.model_dump_json() for meeting in meetings]
         json.dump(json_data, f)

src/meetings.py

@@ -7,7 +7,7 @@
 """
 
 import re
-from typing import Dict, List, Sequence
+from typing import List
 from urllib.parse import urljoin
 
 import aiohttp
@@ -127,7 +127,7 @@ async def parse_meetings(html: str) -> List[Meeting]:
     return meetings
 
 
-async def get_tgov_meetings() -> Sequence[Meeting]:
+async def get_tgov_meetings() -> List[Meeting]:
     """
     Fetch and parse meeting data from the Government Access Television website.
 
@@ -165,7 +165,7 @@ def duration_to_minutes(duration):
         return None
 
 
-def get_registry_meetings() -> Sequence[Meeting]:
+def get_registry_meetings() -> List[Meeting]:
     if is_aws_configured():
         print(f"Getting registry from DynamoDB.")
         return list(Meeting.scan())
@@ -174,7 +174,7 @@ def get_registry_meetings() -> Sequence[Meeting]:
         return read_meetings()
 
 
-def write_registry_meetings(meetings: Sequence[Meeting]) -> Sequence[Meeting]:
+def write_registry_meetings(meetings: List[Meeting]) -> List[Meeting]:
     if is_aws_configured():
         print(f"Writing registry to DynamoDB.")
         with Meeting.batch_writer():

src/models/meeting.py

@@ -7,7 +7,7 @@
 from dyntastic import Dyntastic
 from pydantic import BaseModel, Field, HttpUrl
 from datetime import datetime
-from typing import Sequence, List
+from typing import List
 
 
 def clean_filename(meeting_name: str) -> str:

src/models/utils.py

@@ -34,4 +34,7 @@ def from_jsonl(jsonl_str: str, model_class: Type[T]) -> Sequence[T]:
     Returns:
         A list of instances of the specified Pydantic model class.
     """
-    return [model_class.model_validate(json.loads(line)) for line in jsonl_str.strip().splitlines()]
+    return [
+        model_class.model_validate(json.loads(line))
+        for line in jsonl_str.strip().splitlines()
+    ]