-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp_communication.py
89 lines (69 loc) · 2.7 KB
/
app_communication.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import requests
import json
import time
from api_secrets import API_KEY_ASSEMBLYAI, API_KEY_LISTENNOTES
import pprint
transcript_endpoint = 'https://api.assemblyai.com/v2/transcript'
headers_assemblyai = {
"authorization": API_KEY_ASSEMBLYAI,
"content-type": "application/json"
}
listennotes_episode_endpoint = 'https://listen-api.listennotes.com/api/v2/episodes'
headers_listennotes = {
'X-ListenAPI-Key': API_KEY_LISTENNOTES,
}
def get_episode_audio_url(episode_id):
url = listennotes_episode_endpoint + '/' + episode_id
response = requests.request('GET', url, headers=headers_listennotes)
data = response.json()
# pprint.pprint(data)
episode_title = data['title']
thumbnail = data['thumbnail']
podcast_title = data['podcast']['title']
audio_url = data['audio']
return audio_url, thumbnail, podcast_title, episode_title
def transcribe(audio_url, auto_chapters):
transcript_request = {
'audio_url': audio_url,
'auto_chapters': auto_chapters
}
transcript_response = requests.post(transcript_endpoint, json=transcript_request, headers=headers_assemblyai)
pprint.pprint(transcript_response.json())
return transcript_response.json()['id']
def poll(transcript_id):
polling_endpoint = transcript_endpoint + '/' + transcript_id
polling_response = requests.get(polling_endpoint, headers=headers_assemblyai)
return polling_response.json()
def get_transcription_result_url(url, auto_chapters):
transcribe_id = transcribe(url, auto_chapters)
while True:
data = poll(transcribe_id)
if data['status'] == 'completed':
return data, None
elif data['status'] == 'error':
return data, data['error']
print("waiting for 60 seconds")
time.sleep(60)
def save_transcript(episode_id):
audio_url, thumbnail, podcast_title, episode_title = get_episode_audio_url(episode_id)
data, error = get_transcription_result_url(audio_url, auto_chapters=True)
if data:
filename = episode_id + '.txt'
with open(filename, 'w') as f:
f.write(data['text'])
filename = episode_id + '_chapters.json'
with open(filename, 'w') as f:
chapters = data['chapters']
data = {'chapters': chapters}
data['audio_url']=audio_url
data['thumbnail']=thumbnail
data['podcast_title']=podcast_title
data['episode_title']=episode_title
# for key, value in kwargs.items():
# data[key] = value
json.dump(data, f, indent=4)
print('Transcript saved')
return True
elif error:
print("Error!!!", error)
return False