# app.py (tu archivo de backend Python) import math 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 import googlemaps # Importar la librería de Google Maps from geopy import distance # O si prefieres una implementación más pura de Haversine sin importar toda geopy: # from math import radians, sin, cos, sqrt, atan2 app = Flask(__name__) CORS(app) # Dominio (patente del vehículo) para el cual se desea obtener información # DOMINIO = "AA494XW" # --- Configuración de Gemini --- #gemini_api_key = os.getenv("GEMINI_API_KEY") #gemini_api_key = "AIzaSyAx50JSlCAg5nUTKvXZh52WuvUdJmdkvdc" # ¡CAMBIA ESTO! es la mia api_key1 = "AIzaSyC-Kin27aZPxxMhHkpUPIlkzmdgPRkRqaw" #gemini y maps if not api_key1: raise ValueError("GEMINI_API_KEY de Gemini no encontrada. Por favor, configura la variable de entorno.") genai.configure(api_key=api_key1) GEMINI_MODEL_NAME = 'gemini-2.5-flash' # gemini_model = genai.GenerativeModel(GEMINI_MODEL_NAME) # Ya no se inicializa aquí, se hará con tools. # --- 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 api_key1: 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=api_key1) # --- 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*"(.*?)"', re.IGNORECASE) # --- Almacenamiento del historial de chat en memoria (solo para pruebas) --- chat_sessions = {} def haversine(coord1, coord2): R = 6371 # Earth radius in kilometers lat1, lon1 = math.radians(coord1[0]), math.radians(coord1[1]) lat2, lon2 = math.radians(coord2[0]), math.radians(coord2[1]) dlon = lon2 - lon1 dlat = lat2 - lat1 a = math.sin(dlat / 2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2)**2 c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) distance_km = R * c return distance_km # --- Definición de las herramientas (Functions) --- def calculate_total_distance(coordinates_json: str) -> float: """ Calcula la distancia total recorrida en kilómetros a partir de un string JSON de latitudes y longitudes. Args: coordinates_json (str): Un string JSON que representa un array de objetos, donde cada objeto tiene 'latitud' y 'longitud'. Ejemplo: '[{"latitud": -34.5, "longitud": -58.0}, {"latitud": -34.6, "longitud": -58.1}]' Returns: float: La distancia total recorrida en kilómetros. Devuelve 0.0 si el formato es incorrecto o hay menos de dos puntos. """ print(f"DEBUG: Received coordinates_json (type: {type(coordinates_json)}): '{coordinates_json}'") try: coordinates = json.loads(coordinates_json) except json.JSONDecodeError as e: print(f"Error: El string de entrada no es un JSON válido. Detalles: {e}") print(f"DEBUG: Problematic JSON string (first 100 chars): {coordinates_json[:100]}...") return 0.0 if not isinstance(coordinates, list) or len(coordinates) < 2: print("Advertencia: Se requieren al menos dos puntos en el array para calcular la distancia.") return 0.0 total_distance_km = 0.0 for i in range(len(coordinates) - 1): point1_data = coordinates[i] point2_data = coordinates[i+1] try: # Extraer latitud y longitud, asegurándose de que sean números lat1 = float(point1_data['latitud']) lon1 = float(point1_data['longitud']) lat2 = float(point2_data['latitud']) lon2 = float(point2_data['longitud']) except (KeyError, ValueError): print(f"Error: Punto {i} o {i+1} no tiene formato válido (se esperaba 'latitud' y 'longitud' numéricas).") return 0.0 # O podrías optar por saltar este punto y continuar, dependiendo del requerimiento point1 = (lat1, lon1) point2 = (lat2, lon2) # Calcula la distancia entre los dos puntos usando la fórmula del haversine # geopy.distance.haversine devuelve un objeto Distance, puedes acceder a .km distance_between_points = haversine(point1, point2).km total_distance_km += distance_between_points return total_distance_km 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, "calculate_total_distance": calculate_total_distance } # 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"], ), ), genai.protos.FunctionDeclaration( name="calculate_total_distance", description="Calcula la distancia total recorrida en kilómetros a partir de un string JSON de latitudes y longitudes.", parameters=genai.protos.Schema( type=genai.protos.Type.OBJECT, properties={ # ¡Cambia "direcciones" a "coordinates_json" aquí! "coordinates_json": genai.protos.Schema(type=genai.protos.Type.STRING, description="Un string JSON con el array de objetos de latitud y longitud. Ejemplo: '[{\"latitud\": -34.5, \"longitud\": -58.0}]'"), }, required=["coordinates_json"], ), ), ] ) ] # La instrucción de Gemini, actualizada para mencionar que puede usar herramientas gemini_instruction = ( "Como experto en logistica y 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. Tienes las funciones creadas para conectar con Google Maps en geocode_location,get_route_directions " "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." ) 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') session_id = data.get('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 # Recuperar o inicializar la sesión de chat con el modelo y las herramientas if session_id not in chat_sessions: print(f"Creando nueva sesión de chat para ID: {session_id}") # Importante: Pasar las herramientas al iniciar la sesión chat_sessions[session_id] = genai.GenerativeModel(GEMINI_MODEL_NAME, tools=tools_for_gemini).start_chat(history=[]) chat_session = chat_sessions[session_id] # --- Lógica de extracción de patente y llamada a Stopcar --- enriched_user_message = user_message extracted_patent = None stopcar_data = None 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() 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.") vehicle_info_str = json.dumps(stopcar_data['data'], indent=2) enriched_user_message = f"{user_message_clean}\n\n[INFORMACIÓN DEL VEHÍCULO - Patente {extracted_patent}: {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_user_message = f"{user_message_clean}\n\n[ERROR AL OBTENER INFORMACIÓN DEL VEHÍCULO - Patente {extracted_patent}: {error_message_stopcar}. Por favor, responde sin esta información del vehículo.]" # --- Añadir la instrucción inicial al historial si la sesión es nueva --- # Gemini funciona mejor si la instrucción inicial se le da como parte del historial de chat # en el primer turno (user message) if not chat_session.history: # Si el historial está vacío (primera interacción de la sesión) # Agrega la instrucción como un mensaje del usuario inicial chat_session.send_message(gemini_instruction) # Y luego, una respuesta simulada del modelo chat_session.send_message("Entendido. ¿En qué puedo ayudarte?") # --- Enviar el mensaje del usuario (posiblemente enriquecido) a Gemini --- # El modelo usará las herramientas si es necesario print(f"Enviando mensaje enriquecido a Gemini: {enriched_user_message}") response = chat_session.send_message(enriched_user_message) # --- Procesar las respuestas de Gemini, incluyendo llamadas a herramientas --- # Iterar sobre las partes de la respuesta para ver si hay llamadas a funciones response_text = "" def process_gemini_response(gemini_response_object): nonlocal response_text # Allow modifying response_text in outer scope for chunk in gemini_response_object: for part in chunk.parts: if part.function_call: function_name = part.function_call.name function_args = {k: v for k, v in part.function_call.args.items()} print(f"Gemini solicitó llamar a la función: {function_name} con argumentos: {function_args}") if function_name in available_tools: tool_function = available_tools[function_name] tool_response = tool_function(**function_args) print(f"Respuesta de la herramienta {function_name}: {tool_response}") response_from_tool_call = chat_session.send_message( genai.protos.Part( function_response=genai.protos.FunctionResponse( name=function_name, response=tool_response ) ) ) # Recursively process the response after the tool call process_gemini_response(response_from_tool_call) # IMPORTANT: If a function call was made, and this recursive call # finishes, we might still need to ensure response_text is set # from the *last* text response encountered in the recursion. # The 'return' in the else block helps, but if the final # step is still a function call, response_text might remain empty. else: response_text = f"Error: La función {function_name} no está disponible." print(response_text) return # Stop processing if tool is not available else: response_text = part.text print(f"Respuesta de texto de Gemini: {response_text}") # If we get text, we consider this the final text response for now return # Exit the function once text is found # Start processing the initial response process_gemini_response(response) # Si response_text sigue vacío (ej. si la primera respuesta fue una tool call y no hubo segunda respuesta de texto) if not response_text and response.text: response_text = response.text # Intenta obtener el texto directamente si disponible return jsonify({"reply": response_text}) 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 if "BlockedPromptException" in str(e) or "BlockedGenerationException" in str(e): return jsonify({"error": "La respuesta de Gemini fue bloqueada por razones de seguridad o contenido. Intenta reformular tu pregunta."}), 400 return jsonify({"error": str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5004, debug=True)