# app.py (tu archivo de backend Python) from flask import Flask, request, jsonify from flask_cors import CORS import google.generativeai as genai import os import requests import json import re import uuid # Para generar IDs de sesión únicos import googlemaps # Importar la librería de Google Maps app = Flask(__name__) CORS(app) # --- Configuración de Gemini --- gemini_api_key = "AIzaSyAx50JSlCAg5nUTKvXZh52WuvUdJmdkvdc" # ¡CAMBIA ESTO! if not gemini_api_key: raise ValueError("GEMINI_API_KEY no encontrada. Por favor, configura la variable de entorno.") # --- Configuración de Google Maps API --- # ¡IMPORTANTE! Usa una API Key DIFERENTE y RESTRINGIDA para Google Maps Maps_API_KEY = os.getenv("Maps_API_KEY") if not Maps_API_KEY: raise ValueError("Maps_API_KEY no encontrada. Configura la variable de entorno y habilita las APIs de Geocoding y Directions en GCP.") gmaps = googlemaps.Client(key=Maps_API_KEY) genai.configure(api_key=gemini_api_key) GEMINI_MODEL_NAME = 'gemini-2.5-flash-lite-preview-06-17' # O el que te haya listado el comando gemini_model = genai.GenerativeModel(GEMINI_MODEL_NAME) # --- Configuración de Stopcar API --- STOPCAR_API_URL = "https://gps.divisiongps.com.ar/index.php?r=api/integrador" STOPCAR_TOKEN = "cab8fa81616e5c3a2f17b25699149732" # ¡Reemplaza con tu token real! PATENT_REGEX = re.compile(r'PATENTE\s*=\s*"(.*?)"') # --- Almacenamiento del historial de chat en memoria (solo para pruebas) --- # Clave: session_id, Valor: objeto de chat de Gemini (genai.GenerativeModel.start_chat()) chat_sessions = {} def geocode_location(lat: float, lng: float): """ Geocodifica una latitud y longitud para obtener una dirección legible y un enlace de Google Maps. Args: lat (float): Latitud. lng (float): Longitud. Returns: dict: Un diccionario con la dirección formateada, ciudad, código postal y un enlace de Google Maps. """ print(f"Llamando a geocode_location para lat={lat}, lng={lng}") try: # Realizar la geocodificación inversa reverse_geocode_result = gmaps.reverse_geocode((lat, lng)) if reverse_geocode_result: address = reverse_geocode_result[0].get('formatted_address', 'Dirección no encontrada') # Intentar extraer componentes más específicos si es necesario city = next((c['long_name'] for c in reverse_geocode_result[0]['address_components'] if 'locality' in c['types']), 'N/A') postal_code = next((c['long_name'] for c in reverse_geocode_result[0]['address_components'] if 'postal_code' in c['types']), 'N/A') map_link = f"https://www.google.com/maps/search/?api=1&query={lat},{lng}" return { "address": address, "city": city, "postal_code": postal_code, "map_link": map_link } else: return {"error": "No se encontraron resultados de geocodificación para las coordenadas proporcionadas."} except Exception as e: print(f"Error en geocode_location: {e}") return {"error": f"Error al geocodificar la ubicación: {str(e)}"} def get_route_directions(origin: str, destination: str): """ Obtiene las direcciones para una ruta entre un origen y un destino. Args: origin (str): El punto de partida (puede ser una dirección, lat/lng, o el nombre de un lugar). destination (str): El punto de destino (puede ser una dirección, lat/lng, o el nombre de un lugar). Returns: dict: Un diccionario con los pasos de la ruta y la URL de Google Maps para la ruta. """ print(f"Llamando a get_route_directions para origen='{origin}', destino='{destination}'") try: directions_result = gmaps.directions(origin, destination, mode="driving") if directions_result: route_summary = { "distance": directions_result[0]['legs'][0]['distance']['text'], "duration": directions_result[0]['legs'][0]['duration']['text'], "steps": [] } for step in directions_result[0]['legs'][0]['steps']: route_summary['steps'].append(step['html_instructions']) # O step['maneuver'] # Enlace directo a Google Maps para la ruta origin_enc = requests.utils.quote(origin) destination_enc = requests.utils.quote(destination) map_route_link = f"https://www.google.com/maps/dir/{origin_enc}/{destination_enc}" route_summary['map_route_link'] = map_route_link return route_summary else: return {"error": "No se encontraron direcciones para la ruta especificada."} except Exception as e: print(f"Error en get_route_directions: {e}") return {"error": f"Error al obtener las direcciones de la ruta: {str(e)}"} # Mapeo de nombres de funciones a sus implementaciones reales available_tools = { "geocode_location": geocode_location, "get_route_directions": get_route_directions } # Definir las herramientas para Gemini # Esto le dice a Gemini qué funciones están disponibles y cómo usarlas tools_for_gemini = genai.GenerativeModel(GEMINI_MODEL_NAME).tools = [ genai.protos.Tool( function_declarations=[ genai.protos.FunctionDeclaration( name="geocode_location", description="Geocodifica una latitud y longitud para obtener una dirección legible y un enlace de Google Maps.", parameters=genai.protos.Schema( type=genai.protos.Type.OBJECT, properties={ "lat": genai.protos.Schema(type=genai.protos.Type.NUMBER, description="La latitud."), "lng": genai.protos.Schema(type=genai.protos.Type.NUMBER, description="La longitud."), }, required=["lat", "lng"], ), ), genai.protos.FunctionDeclaration( name="get_route_directions", description="Obtiene las direcciones y el tiempo de viaje para una ruta entre un origen y un destino.", parameters=genai.protos.Schema( type=genai.protos.Type.OBJECT, properties={ "origin": genai.protos.Schema(type=genai.protos.Type.STRING, description="El punto de partida de la ruta (puede ser una dirección, lat/lng, o el nombre de un lugar)."), "destination": genai.protos.Schema(type=genai.protos.Type.STRING, description="El punto de destino de la ruta (puede ser una dirección, lat/lng, o el nombre de un lugar)."), }, required=["origin", "destination"], ), ), ] ) ] def get_vehicle_info_stopcar(token: str, dominio: str): """ Realiza una solicitud HTTP POST a la API de Stopcar para obtener información de un vehículo específico. """ payload = { "token": token, "dominio": dominio } headers = { "Content-Type": "application/json" } print(f"Intentando conectar a la API de Stopcar en: {STOPCAR_API_URL}") print(f"Enviando payload a Stopcar: {json.dumps(payload)}") try: response = requests.post(STOPCAR_API_URL, headers=headers, data=json.dumps(payload), timeout=10) response.raise_for_status() data = response.json() return data except requests.exceptions.RequestException as e: print(f"Error al conectar con Stopcar: {e}") if hasattr(e, 'response') and e.response is not None: print(f"Respuesta de Stopcar: {e.response.text}") return {"error": f"No se pudo obtener información del vehículo desde Stopcar: {str(e)}"} except json.JSONDecodeError as e: print(f"Error al decodificar JSON de la respuesta de Stopcar: {e}") if 'response' in locals() and response is not None: print(f"Contenido de la respuesta de Stopcar que causó el error: {response.text}") return {"error": f"Error al procesar la respuesta de Stopcar: {str(e)}"} @app.route('/chat', methods=['POST']) def chat_endpoint(): try: data = request.json user_message = data.get('message') print("este es el mensaje dle usuario: "+user_message) session_id = data.get('session_id') # Obtener el ID de sesión del frontend print("session: "+session_id) if not user_message: return jsonify({"error": "No message provided"}), 400 if not session_id: return jsonify({"error": "Session ID not provided"}), 400 # Obtener o crear la sesión de chat para este usuario if session_id not in chat_sessions: print(f"Creando nueva sesión de chat para ID: {session_id}") chat_sessions[session_id] = gemini_model.start_chat(history=[]) chat_session = chat_sessions[session_id] enriched_prompt_parts = [] extracted_patent = None # 1. Intentar extraer la patente del mensaje del usuario match = PATENT_REGEX.search(user_message) if match: extracted_patent = match.group(1).strip().upper() print(f"Patente detectada en el mensaje: {extracted_patent}") user_message_clean = PATENT_REGEX.sub('', user_message).strip() # Añadir una instrucción inicial para Gemini si se detecta la patente enriched_prompt_parts.append(f"El usuario pregunta: '{user_message_clean}'.") # 2. Llamar a la API de Stopcar stopcar_data = get_vehicle_info_stopcar(STOPCAR_TOKEN, extracted_patent) if stopcar_data and stopcar_data.get('status') == 200: print("Datos de Stopcar obtenidos exitosamente.") # Formatear los datos de Stopcar para incluirlos en el prompt de Gemini # Es crucial que estos datos sean claros y concisos para Gemini vehicle_info_str = json.dumps(stopcar_data['data'], indent=2) enriched_prompt_parts.append(f"Información del vehículo {extracted_patent} de Stopcar: {vehicle_info_str}.") else: error_message_stopcar = stopcar_data.get('error', 'Error desconocido al obtener datos de Stopcar.') print(f"Error al obtener datos de Stopcar: {error_message_stopcar}") enriched_prompt_parts.append(f"Hubo un problema al obtener datos del vehículo {extracted_patent} de Stopcar: {error_message_stopcar}.") enriched_prompt_parts.append(f"Por favor, responde al usuario basándote en su pregunta original y el contexto de la conversación, sin datos del vehículo.") # Construye el prompt final para Gemini # Estas son las instrucciones generales que siempre queremos que Gemini considere # La instrucción de Gemini, actualizada para mencionar que puede usar herramientas gemini_instruction = ( "Eres un operador de atención al cliente de una empresa de rastreo vehicular. " "Responde de forma concisa y amigable. " "Si se proporciona información de un vehículo, úsala para responder las preguntas relacionadas. " "Puedes usar herramientas para geocodificar ubicaciones o obtener rutas. " "Si la información tiene latitud y longitud, puedes geocodificarla con Google Maps. " "Si te piden una ruta, puedes calcularla usando las direcciones dadas o la última ubicación conocida del vehículo. " "Verifica si, según la velocidad o el campo ignicion, el vehículo estaba encendido. " "Interpreta los datos clave-valor. " "No muestres más información que la pedida. Responde en castellano." ) # Combinar la instrucción general con el mensaje del usuario y los datos de Stopcar final_gemini_prompt = gemini_instruction + "\n\n" + "\n".join(enriched_prompt_parts) else: # Si no se encontró una patente, solo usa el mensaje original del usuario gemini_instruction = "Eres un operador de atención al cliente de una empresa de rastreo el cliente no tiene la patente pero quiere informacion basica sobre el siguiente texto" enriched_prompt_parts.append(user_message) # Estas son las instrucciones generales que siempre queremos que Gemini considere # Combinar la instrucción general con el mensaje del usuario y los datos de Stopcar final_gemini_prompt = gemini_instruction + "\n\n" + "\n".join(enriched_prompt_parts) print(f"Prompt final enviado a Gemini:\n{final_gemini_prompt}") # Envía el mensaje a la sesión de chat de Gemini # Gemini automáticamente usa el historial de la sesión para el contexto response_gemini = chat_session.send_message(final_gemini_prompt) gemini_content = response_gemini.text return jsonify({"reply": gemini_content}) except Exception as e: print(f"Error en el backend: {e}") if "API key not valid" in str(e): return jsonify({"error": "Error de autenticación con Gemini. Revisa tu API Key."}), 500 return jsonify({"error": str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5003, debug=True)