Sapian
/
callflow_sian
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

files cleared

Browse Source
master
johan.ocampo 4 months ago
parent ef7917befc
commit b49542b2a3
2 changed files with 0 additions and 194 deletions
Show Stats Download Patch File Download Diff File

173
detectar.py
Unescape View File

@ -1,173 +0,0 @@
#!/usr/bin/env python3
import speech_recognition as sr
import re
import tts
import subprocess
import os
from asterisk.agi import AGI
import anthropic
import json
agi = AGI()
class CallHandler:
def __init__(self):
self.audio_path = "/var/lib/asterisk/sounds/grabacion_usuario.wav"
self.tab_id = self.get_tab_id()
def get_tab_id(self):
# Obtener el tab_id desde una variable de entorno AGI
# Asegúrate de pasar esta variable desde el Dialplan
tab_id = agi.env.get('TAB_ID', None)
if not tab_id:
agi.verbose("TAB_ID no fue proporcionado. Utilizando 'default_tab_id'.", 1)
return 'default_tab_id'
return tab_id
def grabar_llamada(self):
callerid = agi.env.get('agi_callerid', '')
uniqueid = agi.env.get('agi_uniqueid', '')
# Registrar la información de la llamada
agi.verbose(f"CallerID: {callerid}", 1)
agi.verbose(f"UniqueID: {uniqueid}", 1)
# Iniciar la grabación
agi.verbose("Iniciando grabación...", 1)
try:
result = agi.record_file(
filename='/var/lib/asterisk/sounds/grabacion_usuario',
format='wav',
timeout=4000, # Puedes ajustar el tiempo de espera según tus necesidades
offset=0,
beep=True,
silence=0,
escape_digits='#'
)
except Exception as e:
agi.verbose(f"Error al grabar: {e}", 1)
else:
agi.verbose(f"Grabación completada. Resultado: {result}", 1)
self.detectar_audio()
def detectar_audio(self):
recognizer = sr.Recognizer()
with sr.AudioFile(self.audio_path) as source:
audio_data = recognizer.record(source)
try:
text = recognizer.recognize_google(audio_data, language="es-ES")
agi.verbose(f"La pregunta es: {text}", 1)
# Llamar a la función consultar_claude con el texto reconocido
respuesta = self.consultar_claude(text)
ext = self.determine_extension(respuesta)
cleaned_text = re.sub(r'^(?:\*+|\d+)', '', respuesta).strip()
cleaned_text = cleaned_text.replace('%', '')
# Reproducir la respuesta generada por Claude
try:
tts.convert_text_to_speech(cleaned_text)
agi.stream_file('resp_claude')
except Exception as e:
agi.verbose(f"Error al convertir texto a voz: {e}", 1)
if ext is not None and ext != '':
try:
resultado_dial = agi.execute("EXEC Dial", f"Local/{str(ext)}@from-internal,20")
agi.verbose(f"Resultado del Dial: {resultado_dial}", 1)
except Exception as e:
agi.verbose(f"Error ejecutando Dial en AGI: {e}", 1)
else:
agi.verbose("La variable ext no tiene un valor válido.", 1)
agi.stream_file('algomas')
# Volver a grabar si es necesario
self.grabar_llamada()
except sr.RequestError as e:
agi.verbose(f"Error al conectarse con el servicio de reconocimiento de voz: {e}", 1)
except sr.UnknownValueError:
agi.verbose("No se pudo entender el audio", 1)
agi.stream_file('no_entendi')
self.grabar_llamada()
def determine_extension(self, input_str):
# Definir la ruta al archivo de condiciones basado en tab_id
conditions_file = f"/home/Flow_IA/IVR/tab_{self.tab_id}_conditions.json"
if not os.path.exists(conditions_file):
agi.verbose(f"No se encontró el archivo de condiciones para la pestaña {self.tab_id}", 1)
return None
try:
with open(conditions_file, 'r', encoding='utf-8') as f:
conditions_data = json.load(f)
# Iterar sobre todas las condiciones de node3
for node_cond in conditions_data.get('node3_conditions', []):
for cond in node_cond.get('conditions', []):
condition_str = cond.get('condition_string', '').strip().lower()
target_id = cond.get('target_node_id', '').strip()
if input_str.lower() == condition_str:
agi.verbose(f"Condición encontrada: {condition_str} -> Extensión: {target_id}", 1)
return target_id
except Exception as e:
agi.verbose(f"Error al leer o procesar el archivo de condiciones: {e}", 1)
# Si no se encuentra ninguna condición que coincida
agi.verbose("No se encontró una condición que coincida con la entrada del usuario.", 1)
return None
def consultar_claude(self, text):
# Leer extensiones (este comando parece específico para tu configuración)
command = """grep -E '^\[|^callerid=' /etc/asterisk/pjsip.endpoint.conf | awk '{
if ($0 ~ /^\[/) {
gsub(/[\[\]]/, "", $1);
ext = $1;
} else if ($0 ~ /^callerid=/) {
split($0, arr, "=");
callerid = arr[2];
printf "Extension: " ext ", CallerID: " callerid "; ";
}
}'"""
result = subprocess.run(command, shell=True, capture_output=True, text=True)
output = result.stdout.strip()
# Crear el cliente Anthropic pasando la API key
client = anthropic.Anthropic(api_key='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')
message = client.messages.create(
model="claude-3-5-haiku-20241022",
#model = "claude-3-5-sonnet-20241022",
max_tokens=200,
temperature=0,
system=(
"Vas a ser un agente de telefonía con las siguientes condiciones: "
f'"{output}"'
),
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": text
}
]
}
]
)
# Asumiendo que `message.content` es una lista de mensajes
respuesta_claude = "".join([msg.get('text', '') for msg in message.content])
respuesta_claude = respuesta_claude.replace('%', '')
agi.verbose(f"Respuesta de Claude: {respuesta_claude}", 1)
return respuesta_claude
if __name__ == "__main__":
agi.answer()
agi.stream_file('saludo_claude')
call_handler = CallHandler()
call_handler.grabar_llamada()

21
getIp.py
Unescape View File

@ -1,21 +0,0 @@
import socket
def obtener_ip_privada():
try:
# Crear un socket UDP temporal
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
# Conectarse a un host arbitrario en Internet
s.connect(("8.8.8.8", 80)) # Google Public DNS
# Obtener la dirección IP del socket
ip_privada = s.getsockname()[0]
return ip_privada
except Exception as e:
print(f"Error al obtener la IP privada: {e}")
return None
# Llamar a la función
mi_ip_privada = obtener_ip_privada()
if mi_ip_privada:
print(f"Tu IP privada es: {mi_ip_privada}")
else:
print("No se pudo determinar la IP privada.")
Write Preview
Loading…
Cancel
Save