refactor: factor out Zoom, Jitsi, and w2g code into a package

Author: sloria

bot.py

@@ -1,11 +1,9 @@
 import asyncio
-import base64
 import datetime as dt
 import difflib
 import logging
 import random
 import re
-import uuid
 from contextlib import suppress
 from typing import Optional
 from urllib.parse import quote_plus
@@ -22,6 +20,7 @@
 import handshapes
 import cuteid
 import catchphrase
+import meetings
 
 # -----------------------------------------------------------------------------
 
@@ -370,26 +369,6 @@ async def idiom_command(ctx, spoiler: Optional[str]):
 # -----------------------------------------------------------------------------
 
 
-async def create_meeting():
-    async with aiohttp.ClientSession() as client:
-        resp = await client.post(
-            f"https://api.zoom.us/v2/users/{ZOOM_USER_ID}/meetings",
-            json={
-                "type": 1,
-                "topic": "PRACTICE",
-                "settings": {
-                    "host_video": False,
-                    "participant_video": False,
-                    "mute_upon_entry": True,
-                    "waiting_room": False,
-                },
-            },
-            headers={"Authorization": f"Bearer {ZOOM_JWT}"},
-        )
-    resp.raise_for_status()
-    return await resp.json()
-
-
 ZOOM_TEMPLATE = """**Join URL**: {join_url}
 **Passcode**: {passcode}
 🚀 This meeting is happening now. Go practice!
@@ -406,15 +385,25 @@ async def zoom_command(ctx: Context):
     message = await ctx.send("Creating meeting...")
     logger.info("creating zoom meeting")
     try:
-        data = await create_meeting()
+        meeting = await meetings.create_zoom(
+            token=ZOOM_JWT,
+            user_id=ZOOM_USER_ID,
+            topic="PRACTICE",
+            settings={
+                "host_video": False,
+                "participant_video": False,
+                "mute_upon_entry": True,
+                "waiting_room": False,
+            },
+        )
     except Exception:
         logger.exception("could not create Zoom meeting")
         await message.edit(
             content="🚨 _Could not create Zoom meeting. That's embarrassing._"
         )
     else:
         content = ZOOM_TEMPLATE.format(
-            join_url=data["join_url"], passcode=data["password"]
+            join_url=meeting.join_url, passcode=meeting.passcode
         )
         await message.edit(content=content, suppress=True)
 
@@ -441,13 +430,9 @@ async def zoom_error(ctx, error):
 MEET_CLOSED_MESSAGE = "✨ _Jitsi Meet ended_"
 
 
-def pretty_uuid() -> str:
-    return base64.urlsafe_b64encode(uuid.uuid4().bytes).decode().replace("=", "")
-
-
 @bot.command(name="meet", aliases=("jitsi",), help="Start a Jitsi Meet meeting")
 async def meet_command(ctx: Context):
-    join_url = f"https://meet.jit.si/{pretty_uuid()}"
+    join_url = meetings.create_jitsi_meet()
     content = MEET_TEMPLATE.format(join_url=join_url)
     logger.info("sending jitsi meet info")
     message = await ctx.send(content=content)
@@ -458,19 +443,6 @@ async def meet_command(ctx: Context):
 # -----------------------------------------------------------------------------
 
 
-async def create_watch2gether(video_url: Optional[str]) -> str:
-    async with aiohttp.ClientSession() as client:
-        payload = {"api_key": WATCH2GETHER_API_KEY}
-        if payload:
-            payload["share"] = video_url
-        resp = await client.post("https://w2g.tv/rooms/create.json", json=payload)
-    resp.raise_for_status()
-    data = await resp.json()
-    stream_key = data["streamkey"]
-    url = f"https://w2g.tv/rooms/{stream_key}"
-    return url
-
-
 WATCH2GETHER_HELP = """Create a new watch2gether room
 
 You can optionally pass a URL to use for the first video.
@@ -506,7 +478,7 @@ async def watch2gether_command(ctx: Context, video_url: str = None):
     message = await ctx.send("Creating watch2gether room...")
     logger.info("creating watch2gether meeting")
     try:
-        url = await create_watch2gether(video_url)
+        url = await meetings.create_watch2gether(WATCH2GETHER_API_KEY, video_url)
     except Exception:
         logger.exception("could not create watch2gether room")
         await message.edit(

lib/meetings/__init__.py

@@ -0,0 +1,50 @@
+import base64
+import uuid
+from typing import NamedTuple
+
+import aiohttp
+
+
+class ZoomMeeting(NamedTuple):
+    join_url: str
+    passcode: str
+
+
+async def create_zoom(
+    *, token: str, user_id: str, topic: str, settings: dict
+) -> ZoomMeeting:
+    """Create and return a Zoom meeting via the Zoom API."""
+    async with aiohttp.ClientSession() as client:
+        resp = await client.post(
+            f"https://api.zoom.us/v2/users/{user_id}/meetings",
+            json={
+                "type": 1,
+                "topic": topic,
+                "settings": settings,
+            },
+            headers={"Authorization": f"Bearer {token}"},
+        )
+    resp.raise_for_status()
+    data = await resp.json()
+    return ZoomMeeting(join_url=data["join_url"], passcode=data["password"])
+
+
+def _pretty_uuid() -> str:
+    return base64.urlsafe_b64encode(uuid.uuid4().bytes).decode().replace("=", "")
+
+
+def create_jitsi_meet() -> str:
+    """Return a Jitsi Meet URL with a unique ID."""
+    return f"https://meet.jit.si/{_pretty_uuid()}"
+
+
+async def create_watch2gether(api_key: str, video_url: str = None) -> str:
+    """Create and return a watch2gether URL via the watch2gether API."""
+    async with aiohttp.ClientSession() as client:
+        payload = {"api_key": api_key, "video_url": video_url}
+        resp = await client.post("https://w2g.tv/rooms/create.json", json=payload)
+    resp.raise_for_status()
+    data = await resp.json()
+    stream_key = data["streamkey"]
+    url = f"https://w2g.tv/rooms/{stream_key}"
+    return url