Spaces:
Running
Running
File size: 4,628 Bytes
e007232 f304a40 e007232 685f67f e007232 36582bd e007232 |
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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
import os
import gradio as gr
import torch
import whisper
from TTS.api import TTS
from torch.serialization import add_safe_globals
from TTS.tts.configs.xtts_config import XttsConfig
from TTS.tts.models.xtts import XttsAudioConfig, XttsArgs
from TTS.config.shared_configs import BaseDatasetConfig
from fetch_data import get_botpress_response
import requests, os, json
import warnings
import gradio as gr
import librosa
import torch
import numpy as np
from transformers import Wav2Vec2ForCTC, Wav2Vec2Processor
import assemblyai as aai
warnings.filterwarnings("ignore")
# Load model and processor
processor = Wav2Vec2Processor.from_pretrained("facebook/wav2vec2-base-960h")
model = Wav2Vec2ForCTC.from_pretrained("facebook/wav2vec2-base-960h")
def process_media(media_id, access_token, phone_no_id, phone_no,business_id):
"""Process media ID to send audio message."""
# Fetch media URL
media_url = get_media_url(media_id, access_token, phone_no_id)
if not media_url:
return "Failed to fetch media URL."
# Download the voice note
audio_path = download_voice_note(media_url, access_token)
if not audio_path:
return "Failed to download voice note."
# Transcribe audio
transcription = audio_transcribe(audio_path)
if not transcription:
return "Hi"
return transcription
def audio_transcribe(audio_path):
try:
# Load and preprocess audio
aai.settings.api_key = "c3db74329eea4ff68b95888dca1e2d72"
transcriber = aai.Transcriber()
transcript = transcriber.transcribe(audio_path)
return transcript.text
except Exception as e:
return f"Error: {str(e)}"
def get_media_url(media_id, access_token, phone_no_id):
"""Fetch media URL from Facebook Graph API."""
url = f"https://graph.facebook.com/v21.0/{media_id}?phone_number_id={phone_no_id}"
headers = {'Authorization': f'Bearer {access_token}'}
response = requests.get(url, headers=headers)
if response.status_code == 200:
return response.json().get('url', None)
return None
def download_voice_note(url, access_token):
"""Download a voice note and store it in the current directory."""
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get(url, headers=headers, stream=True)
if response.status_code == 200:
file_name = "voice_note.mp3"
file_path = os.path.join(os.getcwd(), file_name) # Save to current directory
with open(file_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"✅ Download complete: {file_path}")
return file_path
else:
print(f"❌ Failed to download file. Status code: {response.status_code}")
return None
def upload_audio(audio_path, access_token, phone_no_id):
"""Upload an audio file to WhatsApp Business API."""
url = f"https://graph.facebook.com/v21.0/{phone_no_id}/media"
payload = {'messaging_product': 'whatsapp'}
with open(audio_path, 'rb') as audio_file:
files = [('file', ('output.ogg', audio_file, 'audio/ogg'))]
headers = {'Authorization': f'Bearer {access_token}'}
response = requests.post(url, headers=headers, data=payload, files=files)
print(f"response is {response.text}")
return response.json().get('id', None)
def send_audio(phone_no, audio_id, access_token, phone_no_id):
"""Send an audio message via WhatsApp API."""
url = f"https://graph.facebook.com/v21.0/{phone_no_id}/messages"
payload = json.dumps({
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": phone_no,
"type": "audio",
"audio": {"id": audio_id}
})
headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {access_token}'}
try:
response = requests.post(url, headers=headers, data=payload)
response.raise_for_status()
print(response.text)
return response.json()
except requests.exceptions.RequestException as e:
return str(e)
# Gradio Interface
iface = gr.Interface(
fn=process_media,
inputs=[
gr.Textbox(label="Media ID"),
gr.Textbox(label="Access Token"),
gr.Textbox(label="Phone Number ID"),
gr.Textbox(label="Recipient Phone Number"),
gr.Textbox(label="Business ID")
],
outputs="text",
title="WhatsApp Audio Processor",
description="Enter Media ID, Access Token, Phone Number ID, and Recipient Phone Number to process and send audio."
)
iface.launch(debug=True) |