Move download code to new fancy client

scripts/lib/zoom.py

@@ -1,5 +1,9 @@
-from typing import Dict
+
 from functools import wraps
+from typing import Dict
+from urllib.parse import urlparse
+import os.path
+import requests
 from requests import Response
 from zoomus import ZoomClient
 
@@ -23,10 +27,18 @@ def __init__(self, response, message=None):
         self.message = message or self.data.pop('message', 'Zoom API error')
         self.code = self.data.pop('code')
         super().__init__(
-            f'{self.message} (code={self.code}, http_status={self.response.status_code}) '
+            f'{self.message} '
+            f'(code={self.code}, http_status={self.response.status_code}) '
             f'Check the docs for details: {ZOOM_DOCS_URL}.'
         )
 
+    @classmethod
+    def raise_for_response(cls, response: Response) -> Response:
+        if response.status_code >= 400:
+            raise cls(response)
+
+        return response
+
 
 class ZoomResponse:
     response: Response
@@ -55,10 +67,8 @@ def wrap_method_with_parsing(original):
     def wrapper(*args, **kwargs):
         result = original(*args, **kwargs)
         if isinstance(result, Response):
-            if result.status_code >= 400:
-                raise ZoomError(result)
-            else:
-                return ZoomResponse(result)
+            ZoomError.raise_for_response(result)
+            return ZoomResponse(result)
         else:
             return result
 
@@ -100,3 +110,30 @@ def __init__(self, *args, **kwargs):
 
         for component in self.components.values():
             wrap_component_with_parsing(component)
+
+    def get_file(self, url: str) -> Response:
+        # Note the token info in the client isn't really *public*, but it's
+        # not explicitly private, either. Use `config[]` syntax instead of
+        # `config.get()` so we get an exception if things have changed and
+        # this data is no longer available.
+        response = requests.get(url, stream=True, headers={
+            'Authorization': f'Bearer {self.config['token']}'
+        })
+        ZoomError.raise_for_response(response)
+        return response
+
+    def download_file(self, url: str, download_directory: str) -> str:
+        response = self.get_file(url)
+        resolved_url = response.url
+        filename = urlparse(resolved_url).path.split('/')[-1]
+        filepath = os.path.join(download_directory, filename)
+        if os.path.exists(filepath):
+            response.close()
+            return
+
+        with open(filepath, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=1024):
+                if chunk:  # filter out keep-alive new chunks
+                    f.write(chunk)
+
+        return filepath

scripts/upload_zoom_recordings.py

@@ -28,12 +28,10 @@
 from datetime import datetime
 import os
 import re
-import requests
 import subprocess
 import sys
 import tempfile
 from typing import Dict
-from urllib.parse import urlparse
 from lib.zoom import FancyZoom, ZoomError
 from lib.constants import VIDEO_CATEGORY_IDS, ZOOM_ROLES
 from lib.youtube import get_youtube_client, upload_video, add_video_to_playlist, validate_youtube_credentials
@@ -77,29 +75,6 @@ def pretty_date(date_string: str) -> str:
     return datetime.strptime(date_string, '%Y-%m-%dT%H:%M:%SZ').strftime('%b %-d, %Y')
 
 
-def download_zoom_file(client: FancyZoom, url: str, download_directory: str) -> str:
-    # Note the token info in the client isn't really *public*, but it's
-    # not explicitly private, either. Use `config[]` syntax instead of
-    # `config.get()` so we get an exception if things have changed and
-    # this data is no longer available.
-    r = requests.get(url, stream=True, headers={
-        'Authorization': f'Bearer {client.config['token']}'
-    })
-    r.raise_for_status()
-    resolved_url = r.url
-    filename = urlparse(resolved_url).path.split('/')[-1]
-    filepath = os.path.join(download_directory, filename)
-    if os.path.exists(filepath):
-        r.close()
-        return
-    with open(filepath, 'wb') as f:
-        for chunk in r.iter_content(chunk_size=1024):
-            if chunk:  # filter out keep-alive new chunks
-                f.write(chunk)
-
-    return filepath
-
-
 def meeting_had_no_participants(client: FancyZoom, meeting: Dict) -> bool:
     participants = client.past_meeting.get_participants(meeting_id=meeting['uuid'])['participants']
 
@@ -152,7 +127,7 @@ def main():
     with tempfile.TemporaryDirectory() as tmpdirname:
         print(f'Creating tmp dir: {tmpdirname}\n')
 
-        meetings = zoom.recording.list(user_id=zoom_user_id})['meetings']
+        meetings = zoom.recording.list(user_id=zoom_user_id)['meetings']
         meetings = sorted(meetings, key=lambda m: m['start_time'])
         # Filter recordings less than 1 minute
         meetings = filter(lambda m: m['duration'] > 1, meetings)
@@ -188,7 +163,7 @@ def main():
             for file in videos:
                 url = file['download_url']
                 print(f'    Download from {url}...')
-                filepath = download_zoom_file(zoom, url, tmpdirname)
+                filepath = zoom.download_file(url, tmpdirname)
 
                 if video_has_audio(filepath):
                     recording_date = fix_date(meeting['start_time'])