add voice recording snippets

Author: maxkahan

.env.dist

@@ -79,6 +79,7 @@ START_DATE='START_DATE'
 # Voice
 CONFERENCE_NAME='NAME_OF_YOUR_CONFERENCE'
 YOUR_SECOND_NUMBER='YOUR_SECOND_NUMBER'
+RECORDING_URL='RECORDING_URL'
 
 # Numbers
 COUNTRY_CODE='GB'

voice/get-recording.py

@@ -1,21 +1,25 @@
-#!/usr/bin/env python3
+from pprint import pprint
 import os
 from os.path import join, dirname
-from pprint import pprint
-import vonage
 from dotenv import load_dotenv
 
-dotenv_path = join(dirname(__file__), "../.env")
+dotenv_path = join(dirname(__file__), '../.env')
 load_dotenv(dotenv_path)
 
-VONAGE_APPLICATION_ID = os.environ.get("VONAGE_APPLICATION_ID")
-VONAGE_APPLICATION_PRIVATE_KEY_PATH = os.environ.get("VONAGE_APPLICATION_PRIVATE_KEY_PATH")
-
-client = vonage.Client(
-    application_id=VONAGE_APPLICATION_ID,
-    private_key=VONAGE_APPLICATION_PRIVATE_KEY_PATH,
+VONAGE_APPLICATION_ID = os.environ.get('VONAGE_APPLICATION_ID')
+VONAGE_APPLICATION_PRIVATE_KEY_PATH = os.environ.get(
+    'VONAGE_APPLICATION_PRIVATE_KEY_PATH'
 )
 
-response = client.voice.get_recording("RECORDING_URL")
-pprint(response)
+RECORDING_URL = os.environ.get('RECORDING_URL')
+
+from vonage import Auth, Vonage
+
+client = Vonage(
+    Auth(
+        application_id=VONAGE_APPLICATION_ID,
+        private_key=VONAGE_APPLICATION_PRIVATE_KEY_PATH,
+    )
+)
 
+client.voice.download_recording(RECORDING_URL, 'recording.mp3')

voice/make-outbound-call-ncco.py

@@ -4,16 +4,16 @@
 from os.path import join, dirname
 from dotenv import load_dotenv
 
-dotenv_path = join(dirname(__file__), "../.env")
+dotenv_path = join(dirname(__file__), '../.env')
 load_dotenv(dotenv_path)
 
-VONAGE_APPLICATION_ID = os.environ.get("VONAGE_APPLICATION_ID")
+VONAGE_APPLICATION_ID = os.environ.get('VONAGE_APPLICATION_ID')
 VONAGE_APPLICATION_PRIVATE_KEY_PATH = os.environ.get(
-    "VONAGE_APPLICATION_PRIVATE_KEY_PATH"
+    'VONAGE_APPLICATION_PRIVATE_KEY_PATH'
 )
 
-VONAGE_NUMBER = os.environ.get("VONAGE_NUMBER")
-TO_NUMBER = os.environ.get("TO_NUMBER")
+VONAGE_NUMBER = os.environ.get('VONAGE_NUMBER')
+TO_NUMBER = os.environ.get('TO_NUMBER')
 
 from vonage import Auth, Vonage
 from vonage_voice.models import CreateCallRequest, Phone, Talk, ToPhone

voice/record-a-call-with-split-audio.py

@@ -1,44 +1,41 @@
-#!/usr/bin/env python3
+import os
+from os.path import join, dirname
+from dotenv import load_dotenv
+from fastapi import FastAPI, Body
 from pprint import pprint
-from flask import Flask, request, jsonify
-
-app = Flask(__name__)
-
-
-@app.route("/webhooks/answer")
-def answer_call():
-    ncco = [
-        {
-            "action": "talk",
-            "text": "Hi, we will shortly forward your call. This call is recorded for quality assurance purposes."
-        },
-        {
-            "action": "record",
-            "split": "conversation",
-            "channels": 2,
-            "eventUrl": ["https://demo.ngrok.io/webhooks/recordings"]
-        },
-        {
-            "action": "connect",
-            "eventUrl": ["https://demo.ngrok.io/webhooks/event"],
-            "from": "VONAGE_NUMBER",
-            "endpoint": [
-                {
-                    "type": "phone",
-                    "number": "RECIPIENT_NUMBER"
-                }
-            ]
-        }
-    ]
-    return jsonify(ncco)
+from vonage_voice.models import Connect, NccoAction, PhoneEndpoint, Record, Talk
 
+dotenv_path = join(dirname(__file__), '../.env')
+load_dotenv(dotenv_path)
+
+VONAGE_NUMBER = os.environ.get('VONAGE_NUMBER')
+RECIPIENT_NUMBER = os.environ.get('RECIPIENT_NUMBER')
+
+app = FastAPI()
 
-@app.route("/webhooks/recordings", methods=['POST'])
-def recordings():
-    data = request.get_json()
-    pprint(data)
-    return "Webhook received"
 
+@app.get('/webhooks/answer')
+async def inbound_call():
+    ncco: list[NccoAction] = [
+        Talk(
+            text=f'Hi, we will shortly forward your call. This call is recorded for quality assurance purposes.'
+        ),
+        Record(
+            split='conversation',
+            channels=2,
+            eventUrl=['https://demo.ngrok.io/webhooks/recordings'],
+        ),
+        Connect(
+            endpoint=[PhoneEndpoint(number=RECIPIENT_NUMBER)],
+            from_=VONAGE_NUMBER,
+            eventUrl=['https://demo.ngrok.io/webhooks/event'],
+        ),
+    ]
+
+    return [step.model_dump(by_alias=True, exclude_none=True) for step in ncco]
 
-if __name__ == '__main__':
-    app.run(port=3000)
+
+@app.post('/webhooks/recordings')
+async def recordings(data: dict = Body(...)):
+    pprint(data)
+    return {'message': 'webhook received'}

voice/record-a-call.py

@@ -1,42 +1,37 @@
-#!/usr/bin/env python3
+import os
+from os.path import join, dirname
+from dotenv import load_dotenv
+from fastapi import FastAPI, Body
 from pprint import pprint
-from flask import Flask, request, jsonify
-
-app = Flask(__name__)
-
-
-@app.route("/webhooks/answer")
-def answer_call():
-    ncco = [
-        {
-            "action": "talk",
-            "text": "Hi, we will shortly forward your call. This call is recorded for quality assurance purposes."
-        },
-        {
-            "action": "record",
-            "eventUrl": ["https://demo.ngrok.io/webhooks/recordings"]
-        },
-        {
-            "action": "connect",
-            "eventUrl": ["https://demo.ngrok.io/webhooks/event"],
-            "from": "VONAGE_NUMBER",
-            "endpoint": [
-                {
-                    "type": "phone",
-                    "number": "RECIPIENT_NUMBER"
-                }
-            ]
-        }
-    ]
-    return jsonify(ncco)
+from vonage_voice.models import Connect, NccoAction, PhoneEndpoint, Record, Talk
 
+dotenv_path = join(dirname(__file__), '../.env')
+load_dotenv(dotenv_path)
+
+VONAGE_NUMBER = os.environ.get('VONAGE_NUMBER')
+RECIPIENT_NUMBER = os.environ.get('RECIPIENT_NUMBER')
+
+app = FastAPI()
 
-@app.route("/webhooks/recordings", methods=['POST'])
-def recordings():
-    data = request.get_json()
-    pprint(data)
-    return "webhook received"
 
+@app.get('/webhooks/answer')
+async def inbound_call():
+    ncco: list[NccoAction] = [
+        Talk(
+            text=f'Hi, we will shortly forward your call. This call is recorded for quality assurance purposes.'
+        ),
+        Record(eventUrl=['https://demo.ngrok.io/webhooks/recordings']),
+        Connect(
+            endpoint=[PhoneEndpoint(number=RECIPIENT_NUMBER)],
+            from_=VONAGE_NUMBER,
+            eventUrl=['https://demo.ngrok.io/webhooks/event'],
+        ),
+    ]
+
+    return [step.model_dump(by_alias=True, exclude_none=True) for step in ncco]
 
-if __name__ == '__main__':
-    app.run(port=3000)
+
+@app.post('/webhooks/recordings')
+async def recordings(data: dict = Body(...)):
+    pprint(data)
+    return {'message': 'webhook received'}

voice/record-a-conversation.py

@@ -1,31 +1,25 @@
-#!/usr/bin/env python3
-# `eventMethod` is a required workaround currently, otherwise `/webhooks/recordings` is never called.
+from fastapi import FastAPI, Body
 from pprint import pprint
-from flask import Flask, request, jsonify
 
-app = Flask(__name__)
+app = FastAPI()
 
 
-@app.route("/webhooks/answer")
-def answer_call():
+@app.get('/webhooks/answer')
+async def answer_call():
     ncco = [
         {
             "action": "conversation",
             "name": "CONF_NAME",
             "record": "true",
             "eventMethod": "POST",
-            "eventUrl": ["https://demo.ngrok.io/webhooks/recordings"]
+            "eventUrl": ["https://demo.ngrok.io/webhooks/recordings"],
         }
     ]
-    return jsonify(ncco)
 
-
-@app.route("/webhooks/recordings", methods=['POST'])
-def recordings():
-    data = request.get_json()
-    pprint(data)
-    return "Webhook received"
+    return ncco
 
 
-if __name__ == '__main__':
-    app.run(port=3000)
+@app.post('/webhooks/recordings')
+async def recordings(data: dict = Body(...)):
+    pprint(data)
+    return {'message': 'webhook received'}

voice/record-a-message.py

@@ -1,40 +1,27 @@
-#!/usr/bin/env python3
+from fastapi import FastAPI, Body, Request
 from pprint import pprint
-import http
-from flask import Flask, request, jsonify
+from vonage_voice.models import NccoAction, Record, Talk
 
-app = Flask(__name__)
+app = FastAPI()
 
 
-@app.route("/webhooks/answer")
-def answer_call():
-    for param_key, param_value in request.args.items():
-        print("{}: {}".format(param_key, param_value))
-    recording_webhook_url = request.url_root + "webhooks/recording"
-    ncco = [
-        {
-            "action": "talk",
-            "text": "Please leave a message after the tone, then press the hash key."
-        },
-        {
-            "action": "record",
-            "endOnKey": "#",
-            "beepStart": "true",
-            "eventUrl": [recording_webhook_url]
-        },
-        {
-            "action": "talk",
-            "text": "Thank you for your message."
-        }
+@app.get('/webhooks/answer')
+async def answer_call(request: Request):
+    print(request.base_url)
+    ncco: list[NccoAction] = [
+        Talk(text='Please leave a message after the tone, then press the hash key.'),
+        Record(
+            endOnKey='#',
+            beepStart=True,
+            eventUrl=[str(request.base_url) + '/webhooks/recordings'],
+        ),
+        Talk(text='Thank you for your message.'),
     ]
-    return jsonify(ncco)
 
+    return [step.model_dump(by_alias=True, exclude_none=True) for step in ncco]
 
-@app.route("/webhooks/recording", methods=['POST'])
-def recording_webhook():
-    pprint(request.get_json())
-    return ('', http.HTTPStatus.NO_CONTENT)
 
-
-if __name__ == '__main__':
-    app.run(port=3000)
+@app.post('/webhooks/recordings')
+async def recordings(data: dict = Body(...)):
+    pprint(data)
+    return {'message': 'webhook received'}