Merge pull request #21 from codefortulsa/change-from-s3-to-dynamoDB

feat: switch registry storage to DynamoDB via Dyntastic, add local fallback

Author: groovecoder

README.md

@@ -33,7 +33,9 @@ python -m ipykernel install --user --name=tgov-scraper --display-name="TGOV Scra
 jupyter notebook
 ```
 
-### Prefect flows
+### Prefect workflows
+We use prefect to organize code into workflows of data tasks.
+
 See https://docs.prefect.io/get-started
 
 ```bash
@@ -42,6 +44,42 @@ prefect server start                      # to start the persistent server
 python -m flows.translate_meetings        # to run a specific flow
 ```
 
+#### Data "registry" for workflows
+The prefect workflows use a "registry" to track meetings and urls to their data artifacts.
+E.g., `get_new_meetings` task adds each meeting to the registry with a `video_url`, `transcribe_videos` transcribes the video and adds `transcription_url`, etc.
+
+```mermaid
+gitGraph
+    branch registry
+    checkout registry
+    commit id: " "
+
+    branch get_new_meetings
+    checkout get_new_meetings
+    commit id: "get_new_meetings()"
+    checkout registry
+    merge get_new_meetings id: "video_url"
+
+    branch transcribe_videos
+    checkout transcribe_videos
+    commit id: "transcribe_videos()"
+    checkout registry
+    merge transcribe_videos id: "transcription_url"
+
+    branch create_subtitled_video_pages
+    checkout create_subtitled_video_pages
+    commit id: "create_subtitled_video_pages()"
+    checkout registry
+    merge create_subtitled_video_pages id: "subtitled_video_url"
+
+    branch translate_transcriptions
+    checkout translate_transcriptions
+    commit id: "translate_transcriptions(es)"
+    commit id: "translate_transcriptions(mya)"
+    checkout registry
+    merge translate_transcriptions id: "translated_transcription_urls"
+```
+
 ### Tests
 
 ```bash

db/create_tables.py

@@ -0,0 +1,9 @@
+from src.models.meeting import Meeting
+
+def create_tables():
+    print("Creating DynamoDB tables if they don't exist...")
+    Meeting.create_table(wait=True, billing_mode="PAY_PER_REQUEST")
+    print("All tables created!")
+
+if __name__ == "__main__":
+    create_tables()

pyproject.toml

@@ -27,6 +27,7 @@ aiofiles = "^24.1.0"
 faster-whisper = "^1.1.1"
 prefect = "^3.3.0"
 boto3 = "^1.37.24"
+dyntastic = "^0.18.0"
 
 
 [tool.poetry.group.dev.dependencies]

src/aws.py

@@ -3,8 +3,9 @@
 import boto3
 from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError
 
+
 def is_aws_configured():
-    required_vars = ['AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY']
+    required_vars = ['AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', 'AWS_DEFAULT_REGION']
     return all(var in os.environ for var in required_vars)
 
 
@@ -37,4 +38,3 @@ def upload_to_s3(file_path, bucket_name, s3_path):
     except (NoCredentialsError, PartialCredentialsError) as e:
         print(f"Failed to upload to S3: {str(e)}")
         return False
-

src/local_store.py

@@ -0,0 +1,20 @@
+import json
+import os
+from typing import Sequence
+from src.models.meeting import Meeting
+
+LOCAL_STORE_PATH = "data/meetings.json"
+
+def read_meetings() -> Sequence[Meeting]:
+    if not os.path.exists(LOCAL_STORE_PATH):
+        return []
+
+    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]):
+    os.makedirs(os.path.dirname(LOCAL_STORE_PATH), exist_ok=True)
+    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

@@ -15,15 +15,14 @@
 from selectolax.parser import HTMLParser
 
 from src.aws import is_aws_configured
-from src.models.utils import from_jsonl, to_jsonl
+from src.local_store import read_meetings, write_meetings
 
 from .models.meeting import Meeting
 
 BASE_URL = "https://tulsa-ok.granicus.com/ViewPublisher.php?view_id=4"
 TGOV_BUCKET_NAME = "tgov-meetings"
 MEETINGS_REGISTRY_PATH = "data/meetings.jsonl"
 
-
 async def fetch_page(url: str, session: aiohttp.ClientSession) -> str:
     """
     Fetch the HTML content of a page.
@@ -40,11 +39,9 @@ async def fetch_page(url: str, session: aiohttp.ClientSession) -> str:
             raise Exception(f"Failed to fetch {url}, status code: {response.status}")
         return await response.text()
 
-
 def clean_date(date: str) -> str:
     return re.sub(r"\s+", " ", date).strip()
 
-
 async def parse_meetings(html: str) -> List[Dict[str, str]]:
     """
     Parse the meeting data from the HTML content.
@@ -67,21 +64,18 @@ async def parse_meetings(html: str) -> List[Dict[str, str]]:
     # Process each table
     for table in tables:
         for row in table.css("tr.listingRow"):
-            cells = row.css("td")
             name_cells = row.css('td.listItem[headers^="Name"]')
             meeting_name = name_cells[0].text().strip() if name_cells else "Unknown"
 
             date_cells = row.css('td.listItem[headers^="Date"]')
             raw_date = clean_date(date_cells[0].text().strip()) if date_cells else "Unknown"
             meeting_date = raw_date.split("-")[0].strip() if "-" in raw_date else raw_date
 
-
             duration_cells = row.css('td.listItem[headers^="Duration"]')
             duration_str = duration_cells[0].text().strip() if duration_cells else "Unknown"
             minutes = duration_to_minutes(duration_str)
             meeting_duration = f"{minutes // 60}:{minutes % 60:02d}" if minutes is not None else "Unknown"
 
-
             meeting_data = {
                 "meeting": meeting_name,
                 "date": meeting_date,
@@ -131,7 +125,6 @@ async def parse_meetings(html: str) -> List[Dict[str, str]]:
 
     return meetings
 
-
 async def get_tgov_meetings() -> Sequence[Meeting]:
     """
     Fetch and parse meeting data from the Government Access Television website.
@@ -147,7 +140,6 @@ async def get_tgov_meetings() -> Sequence[Meeting]:
         meetings = [Meeting(**meeting_dict) for meeting_dict in meeting_dicts]
         return meetings
 
-
 def duration_to_minutes(duration):
     if not duration or pd.isna(duration):
         return None
@@ -172,43 +164,25 @@ def duration_to_minutes(duration):
     except:
         return None
 
-
 def get_registry_meetings() -> Sequence[Meeting]:
     if is_aws_configured():
-        print(f'Getting registry from AWS S3 bucket: {TGOV_BUCKET_NAME}, path: {MEETINGS_REGISTRY_PATH}')
-        import boto3
-        from botocore.exceptions import ClientError
-        s3 = boto3.client('s3')
-        try:
-            registry_response = s3.get_object(Bucket=TGOV_BUCKET_NAME, Key=MEETINGS_REGISTRY_PATH)
-            registry_body = registry_response['Body'].read().decode('utf-8')
-            return from_jsonl(registry_body, Meeting)
-        except ClientError as e:
-            if e.response['Error']['Code'] == 'NoSuchKey':
-                print('No registry file found on S3. Returning empty list.')
-
-    return []
-
+        print(f'Getting registry from DynamoDB.')
+        return list(Meeting.scan())
+    else:
+        print(f'Getting registry from local store')
+        return read_meetings()
 
 def write_registry_meetings(meetings: Sequence[Meeting]) -> Sequence[Meeting]:
-    jsonl_str = to_jsonl(meetings)
-
     if is_aws_configured():
-        print(f'Writing registry to AWS S3 bucket: {TGOV_BUCKET_NAME}, path: {MEETINGS_REGISTRY_PATH}')
-        import boto3
-        from botocore.exceptions import ClientError
-        s3 = boto3.client('s3')
-
-        try:
-            s3.put_object(
-                    Bucket=TGOV_BUCKET_NAME,
-                    Key=MEETINGS_REGISTRY_PATH,
-                    Body=jsonl_str,
-                    ContentType='application/x-ndjson'
-                    )
-            print(f'Wrote {len(meetings)} meetings to S3.')
-        except ClientError as e:
-            print(f"Failed to write to S3: {e}")
-            raise
+        print(f'Writing registry to DynamoDB.')
+        with Meeting.batch_writer():
+            for meeting in meetings:
+                if meeting.clip_id:
+                    meeting.save()
+                else:
+                    print(f'Skipping meeting with missing clip_id: {meeting}')
+    else:
+        print(f'Writing registry to local store')
+        write_meetings(meetings)
 
     return meetings

src/models/meeting.py

@@ -4,20 +4,24 @@
 
 from typing import Optional
 
+from dyntastic import Dyntastic
 from pydantic import BaseModel, Field, HttpUrl
 
 
-class Meeting(BaseModel):
+class Meeting(Dyntastic):
     """
     Model representing a government meeting
     """
 
+    __table_name__ = "tgov-meeting"
+    __hash_key__ = "clip_id"
+
+    clip_id: Optional[str] = Field(None, description="Granicus clip ID")
     meeting: str = Field(description="Name of the meeting")
     date: str = Field(description="Date and time of the meeting")
     duration: str = Field(description="Duration of the meeting")
     agenda: Optional[HttpUrl] = Field(None, description="URL to the meeting agenda")
     video: Optional[HttpUrl] = Field(None, description="URL to the meeting video")
-    clip_id: Optional[str] = Field(None, description="Granicus clip ID")
 
     def __str__(self) -> str:
         """String representation of the meeting"""

tasks/meetings.py

@@ -10,7 +10,6 @@ async def get_new_meetings():
     # TODO: accept max_limit parameter
     tgov_meetings: Sequence[Meeting] = await get_tgov_meetings()
     print(f"Got {len(tgov_meetings)} tgov meetings.")
-    tgov_clip_ids = [tm.clip_id for tm in tgov_meetings]
     # print(f"tgov_clip_ids: {tgov_clip_ids}")
 
     registry_meetings: Sequence[Meeting] = get_registry_meetings()