Created
October 14, 2024 07:41
-
-
Save jamsea/78340fe7519ba0cd2eca154c6d6f6e3e to your computer and use it in GitHub Desktop.
Leave a Daily call in daily-python when only the bot participant remains
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import threading | |
| import time | |
| import requests | |
| from daily import CallClient, Daily, EventHandler | |
| # Get DAILY_API_KEY from the environment | |
| DAILY_API_KEY = os.getenv("DAILY_API_KEY") | |
| if not DAILY_API_KEY: | |
| raise EnvironmentError("DAILY_API_KEY not found in environment variables") | |
| class DailyEvents(EventHandler): | |
| def __init__(self): | |
| self.__client = CallClient(event_handler=self) | |
| self.__time = time.time() | |
| self.__start_event = threading.Event() | |
| self.__thread = threading.Thread(target=self.wait) | |
| self.__thread.start() | |
| def wait(self): | |
| print("waiting...") | |
| self.__start_event.wait() | |
| # Send a POST request to https://api.daily.co/v1 to create a room that expires in 1 hour | |
| def create_room(self): | |
| url = "https://api.daily.co/v1/rooms" | |
| headers = {"Authorization": "Bearer " + DAILY_API_KEY, "Content-Type": "application/json"} | |
| payload = {"properties": {"exp": int(self.__time) + 3600}} | |
| response = requests.post(url, headers=headers, json=payload) | |
| if response.status_code == 200: | |
| return response.json()["url"] | |
| else: | |
| response.raise_for_status() | |
| def create_meeting_token(self, room_url: str): | |
| url = "https://api.daily.co/v1/meeting-tokens" | |
| headers = {"Authorization": "Bearer " + DAILY_API_KEY, "Content-Type": "application/json"} | |
| payload = {"properties": {"user_name": "bot", "room_name": room_url.split("/")[-1]}} | |
| response = requests.post(url, headers=headers, json=payload) | |
| if response.status_code == 200: | |
| room_url = response.json()["token"] | |
| if not room_url or not isinstance(room_url, str): | |
| raise ValueError("room_url invalid") | |
| return room_url | |
| else: | |
| response.raise_for_status() | |
| def run(self): | |
| meeting_url = self.create_room() | |
| print("Room URL:", meeting_url) | |
| meeting_token = self.create_meeting_token(meeting_url) | |
| self.__client.join(meeting_url, meeting_token, completion=self.on_joined) | |
| def on_joined(self, join_data, error): | |
| print(join_data, error) | |
| def on_participant_joined(self, participant): | |
| print("on_participant_joined:", participant) | |
| def on_participant_left(self, participant, reason): | |
| print("on_participant_left:", participant, reason) | |
| present_participants = self.__client.participant_counts()["present"] | |
| if present_participants == 1: | |
| self.leave() | |
| def leave(self): | |
| self.__thread.join() | |
| self.__client.leave() | |
| self.__client.release() | |
| def main(): | |
| Daily.init() | |
| app = DailyEvents() | |
| try: | |
| app.run() | |
| except KeyboardInterrupt: | |
| print("Ctrl-C detected. Exiting!") | |
| finally: | |
| app.leave() | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment