import os import openai import requests from dotenv import load_dotenv from flask import Flask, render_template, request, redirect, session, jsonify, flash import json import subprocess import threading import time import logging from transcriber import Transcriber from llm import LLM from weather import Weather from tts import TTS from pc_command import PcCommand from spotify_control import SpotifyControl from web_search import WebSearch # Cargar llaves del archivo .env load_dotenv() openai.api_key = os.getenv('OPENAI_API_KEY') elevenlabs_key = os.getenv('ELEVENLABS_API_KEY') SPOTIFY_CLIENT_ID = os.getenv('SPOTIFY_CLIENT_ID') SPOTIFY_CLIENT_SECRET = os.getenv('SPOTIFY_CLIENT_SECRET') SPOTIFY_REDIRECT_URI = "http://127.0.0.1:5000/callback" app = Flask(__name__) app.secret_key = "super_secret_key" # Configuración del logger para registrar comandos y respuestas logger = logging.getLogger("jarvis") logger.setLevel(logging.INFO) file_handler = logging.FileHandler("jarvis.log") formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") file_handler.setFormatter(formatter) logger.addHandler(file_handler) # Variable global para controlar el estado de la búsqueda global searching searching = False search_thread = None # Función para detener la búsqueda después de 30 segundos def stop_search_after_timeout(): global searching time.sleep(30) if searching: searching = False print("El tiempo de espera ha pasado. Jarvis ha dejado de buscar.") @app.route("/") def index(): return render_template("recorder.html") # Endpoint existente para audio (funcionalidad previa) @app.route("/audio", methods=["POST"]) def audio(): global searching, search_thread audio = request.files.get("audio") if not audio: return {"result": "error", "text": "No se recibió ningún archivo de audio"} audio_path = "temp_audio.wav" audio.save(audio_path) transcriber = Transcriber() try: text = transcriber.transcribe(audio_path) print(f"Texto transcrito: {text}") except Exception as e: return {"result": "error", "text": f"Error al transcribir audio: {str(e)}"} llm = LLM() function_name, args, message = llm.process_functions(text) args = args or {} print(f"Función detectada: {function_name}, Argumentos: {args}") if function_name is not None: try: if function_name == "get_weather": function_response = Weather().get(args.get("ubicacion", "")) function_response = json.dumps(function_response) elif function_name == "send_email": function_response = "Aún no implementado, pero pronto podrás enviar emails." elif function_name == "open_chrome": website = args.get("website", "https://www.google.com") PcCommand().open_chrome(website) function_response = f"Listo, abrí Chrome en {website}" elif function_name == "dominate_human_race": function_response = "No te creas. Suscríbete al canal!" elif function_name == "current_song": spotify = SpotifyControl() token = session.get("spotify_token") if not token: function_response = "Error: No estás autenticado con Spotify." else: spotify.token = token function_response = spotify.get_current_song() elif "JARVIS QUIERO QUE BUSQUES EN INTERNET" in text.upper(): query = text.upper().replace("JARVIS QUIERO QUE BUSQUES EN INTERNET", "").strip() if query: web_search = WebSearch() function_response = web_search.search(query) searching = True if search_thread is None or not search_thread.is_alive(): search_thread = threading.Thread(target=stop_search_after_timeout) search_thread.start() else: function_response = "No se proporcionó una consulta de búsqueda válida." elif "APÁGATE JARVIS" in text.upper(): searching = False function_response = "Jarvis se está apagando. Si me necesita, solo vuelva a activarme." else: function_response = "Comando no reconocido." final_response = llm.process_response(text, message, function_name, function_response) tts_file = TTS().process(final_response) return {"result": "ok", "text": final_response, "file": tts_file} except Exception as e: return {"result": "error", "text": f"Error al ejecutar función: {str(e)}"} else: final_response = "No tengo idea de lo que estás diciendo, Ringa Tech" tts_file = TTS().process(final_response) return {"result": "ok", "text": final_response, "file": tts_file} # NUEVO ENDPOINT: Procesar comandos naturales (todo en uno) @app.route("/process_command", methods=["POST"]) def process_command(): # Este endpoint recibe audio (form-data) y realiza: # 1) Transcripción del audio. # 2) Envío del texto a ChatGPT para obtener una respuesta natural. # 3) Ejecución de comandos (si es el caso) y generación de TTS. audio = request.files.get("audio") if not audio: return jsonify({"result": "error", "text": "No se recibió ningún archivo de audio"}), 400 # Guardar el audio temporalmente audio_path = "temp_audio.wav" audio.save(audio_path) # Transcribir el audio transcriber = Transcriber() try: user_text = transcriber.transcribe(audio_path) print(f"Texto transcrito: {user_text}") logger.info(f"Transcripción: {user_text}") except Exception as e: return jsonify({"result": "error", "text": f"Error al transcribir audio: {str(e)}"}), 500 # Enviar el texto a ChatGPT para obtener una respuesta natural response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": ( "Eres Jarvis, un asistente de voz conversacional que ejecuta órdenes del sistema y responde preguntas de manera detallada y natural. " "Cuando el usuario te hable, interpreta su intención. Si es una orden, tradúcela a una instrucción concreta (por ejemplo, 'abrir chrome en <URL>', 'buscar <consulta>' o 'apágate'). " "Si es una pregunta, responde de forma conversacional y precisa. " "Responde únicamente con la instrucción a ejecutar en caso de comandos, o con una respuesta conversacional si es una pregunta." )}, {"role": "user", "content": user_text} ] ) chat_output = response["choices"][0]["message"]["content"] logger.info(f"ChatGPT output: {chat_output}") # Procesar la respuesta: si es un comando, se ejecuta la acción; si no, se devuelve la respuesta conversacional. result = process_system_command(chat_output.lower()) logger.info(f"Processed command result: {result}") # Generar audio de respuesta con TTS tts_file = TTS().process(result) return jsonify({"result": "ok", "transcription": user_text, "text": result, "file": tts_file}) def process_system_command(command): # Interpretar el comando y ejecutar acciones concretas if "abrir chrome" in command: url = command.split("en", 1)[-1].strip() if "en" in command else "https://www.google.com" subprocess.Popen(["google-chrome", url]) return f"Chrome abierto en {url}, señor." elif "buscar" in command: query = command.split("buscar", 1)[1].strip() url = f"https://www.google.com/search?q={query}" subprocess.Popen(["google-chrome", url]) return f"Buscando {query}, señor." elif "apágate" in command: # Aquí se integra la lógica para apagar a Jarvis return "Apagando Jarvis, señor." else: # Devuelve la respuesta conversacional de ChatGPT return command @app.route("/login_spotify") def login_spotify(): spotify = SpotifyControl() auth_url = spotify.get_auth_url() return redirect(auth_url) @app.route("/callback") def callback(): code = request.args.get("code") if not code: return jsonify({"error": "No se recibió el código de autorización."}), 400 spotify = SpotifyControl() token_info = spotify.authenticate(code) if not token_info or "access_token" not in token_info: return jsonify({"error": "Error en la autenticación de Spotify.", "token_info": token_info}), 400 session["spotify_token"] = token_info["access_token"] return jsonify(token_info) @app.route("/current_song") def current_song(): spotify = SpotifyControl() token = session.get("spotify_token") if not token: return jsonify({"error": "No estás autenticado con Spotify."}) spotify.token = token return spotify.get_current_song() # Panel de control para visualizar registros y ajustar parámetros @app.route("/control") def control(): try: with open("jarvis.log", "r") as f: logs = f.read().splitlines()[-20:] except Exception as e: logs = [f"Error al leer el log: {str(e)}"] return render_template("control.html", logs=logs) @app.route("/update_config", methods=["POST"]) def update_config(): normal_duration = request.form.get("normal_duration", 10) confirmation_duration = request.form.get("confirmation_duration", 5) logger.info(f"Configuración actualizada: normal_duration={normal_duration}, confirmation_duration={confirmation_duration}") return redirect("/control") if __name__ == "__main__": app.run(debug=True)

Please keep input under 1000 characters