1056 lines
39 KiB
Python
1056 lines
39 KiB
Python
from flask import Flask, render_template, request, jsonify, url_for
|
|
from flask_sock import Sock
|
|
from lib.config_manager import ConfigurationManager
|
|
from lib.launcher_manager import LauncherManager
|
|
import os
|
|
import json # Added import
|
|
from datetime import datetime
|
|
import time # Added for shutdown delay
|
|
import sys # Added for platform detection
|
|
import subprocess # Add this to the imports at the top
|
|
import shutil # For shutil.whichimport os
|
|
|
|
# --- Imports for System Tray Icon ---
|
|
import threading
|
|
import webbrowser
|
|
import sys
|
|
import requests # To send shutdown request
|
|
from PIL import Image
|
|
import pystray
|
|
|
|
app = Flask(
|
|
__name__, static_url_path="", static_folder="static", template_folder="templates"
|
|
)
|
|
sock = Sock(app)
|
|
config_manager = ConfigurationManager()
|
|
|
|
# Inicializar launcher manager
|
|
launcher_manager = LauncherManager(config_manager.data_path)
|
|
|
|
# Lista global para mantener las conexiones WebSocket activas
|
|
websocket_connections = set()
|
|
|
|
# --- Globals for Tray Icon ---
|
|
tray_icon = None
|
|
|
|
|
|
@sock.route("/ws")
|
|
def handle_websocket(ws):
|
|
try:
|
|
websocket_connections.add(ws)
|
|
while True:
|
|
message = ws.receive()
|
|
if message:
|
|
broadcast_message(message)
|
|
except Exception as e:
|
|
print(f"WebSocket error: {e}")
|
|
finally:
|
|
websocket_connections.remove(ws)
|
|
|
|
|
|
def broadcast_message(message):
|
|
"""Envía un mensaje a todas las conexiones WebSocket activas y guarda en log."""
|
|
dead_connections = set()
|
|
timestamp = datetime.now().strftime("[%H:%M:%S] ")
|
|
|
|
# Normalize input to a list of messages
|
|
if isinstance(message, list):
|
|
messages = message
|
|
else:
|
|
# Si es un solo mensaje, dividirlo en líneas
|
|
messages = [line.strip() for line in message.splitlines() if line.strip()]
|
|
|
|
# Procesar cada mensaje
|
|
for raw_msg in messages:
|
|
# Limpiar timestamps duplicados al inicio del mensaje
|
|
while raw_msg.startswith("[") and "]" in raw_msg:
|
|
try:
|
|
closing_bracket = raw_msg.index("]") + 1
|
|
if raw_msg[1 : closing_bracket - 1].replace(":", "").isdigit():
|
|
raw_msg = raw_msg[closing_bracket:].strip() # Update raw_msg itself
|
|
else:
|
|
break
|
|
except:
|
|
break
|
|
|
|
# Log the raw message using the config_manager's logger
|
|
# The logger will handle its own timestamping for the file.
|
|
config_manager.append_log(raw_msg)
|
|
|
|
# Format message with timestamp *for WebSocket broadcast*
|
|
formatted_msg_for_ws = f"{timestamp}{raw_msg}"
|
|
|
|
# Enviar a todos los clientes WebSocket
|
|
for ws in list(websocket_connections):
|
|
try:
|
|
if ws.connected: # Check if ws is still connected before sending
|
|
ws.send(
|
|
f"{formatted_msg_for_ws}\n"
|
|
) # Use the correct variable name here
|
|
except Exception:
|
|
dead_connections.add(ws) # Collect dead connections
|
|
|
|
# Limpiar conexiones muertas
|
|
websocket_connections.difference_update(dead_connections)
|
|
|
|
|
|
@app.route("/api/execute_script", methods=["POST"])
|
|
def execute_script():
|
|
try:
|
|
script_group = request.json["group"]
|
|
script_name = request.json["script"]
|
|
|
|
# Ejecutar el script y obtener resultado
|
|
result = config_manager.execute_script(
|
|
script_group, script_name, broadcast_message
|
|
)
|
|
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
error_msg = f"Error ejecutando script: {str(e)}"
|
|
broadcast_message(error_msg)
|
|
return jsonify({"error": error_msg})
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
script_groups = config_manager.get_script_groups()
|
|
# Para ordenar una lista de diccionarios, necesitamos especificar una clave.
|
|
# Asumimos que cada diccionario tiene una clave 'name' por la cual ordenar.
|
|
sorted_script_groups = sorted(script_groups, key=lambda group: group['name'])
|
|
return render_template("index.html", script_groups=sorted_script_groups)
|
|
|
|
|
|
@app.route("/api/config/<level>", methods=["GET", "POST"])
|
|
def handle_config(level):
|
|
group = request.args.get("group")
|
|
if request.method == "GET":
|
|
try:
|
|
return jsonify(config_manager.get_config(level, group))
|
|
except FileNotFoundError:
|
|
return jsonify({})
|
|
else:
|
|
data = request.json
|
|
config_manager.update_config(level, data, group)
|
|
return jsonify({"status": "success"})
|
|
|
|
|
|
@app.route("/api/schema/<level>", methods=["GET", "POST"])
|
|
def handle_schema(level):
|
|
group = request.args.get("group")
|
|
if request.method == "GET":
|
|
return jsonify(config_manager.get_schema(level, group))
|
|
else:
|
|
data = request.json
|
|
config_manager.update_schema(level, data, group)
|
|
return jsonify({"status": "success"})
|
|
|
|
|
|
@app.route("/api/scripts/<group>")
|
|
def get_scripts(group):
|
|
# list_scripts ahora devuelve detalles y filtra los ocultos
|
|
scripts = config_manager.list_scripts(group)
|
|
# El frontend espera 'name' y 'description', mapeamos desde 'display_name' y 'short_description'
|
|
return jsonify(
|
|
[
|
|
{
|
|
"name": s["display_name"],
|
|
"description": s["short_description"],
|
|
"filename": s["filename"],
|
|
"long_description": s["long_description"],
|
|
}
|
|
for s in scripts
|
|
]
|
|
)
|
|
|
|
|
|
@app.route("/api/working-directory", methods=["POST"])
|
|
def set_working_directory():
|
|
data = request.json
|
|
if not data:
|
|
return jsonify({"status": "error", "message": "No data provided"})
|
|
|
|
path = data.get("path")
|
|
group = data.get("group")
|
|
|
|
if not path or not group:
|
|
return jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"Missing required fields. Path: {path}, Group: {group}",
|
|
}
|
|
)
|
|
|
|
print(f"Setting working directory - Path: {path}, Group: {group}") # Debug line
|
|
return jsonify(config_manager.set_work_dir(group, path))
|
|
|
|
|
|
@app.route("/api/working-directory/<group>", methods=["GET"])
|
|
def get_working_directory(group):
|
|
path = config_manager.get_work_dir(group)
|
|
return jsonify({"path": path, "status": "success" if path else "not_set"})
|
|
|
|
|
|
@app.route("/api/browse-directories")
|
|
def browse_directories():
|
|
import tkinter as tk
|
|
from tkinter import filedialog
|
|
|
|
# Obtener el directorio inicial
|
|
current_dir = request.args.get("current_path")
|
|
if not current_dir or not os.path.exists(current_dir):
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
# Crear y configurar la ventana principal de tkinter
|
|
root = tk.Tk()
|
|
root.attributes("-topmost", True) # Mantener la ventana siempre arriba
|
|
root.withdraw()
|
|
|
|
# Abrir el diálogo de selección de directorio
|
|
directory = filedialog.askdirectory(
|
|
initialdir=current_dir, title="Seleccionar Directorio de Trabajo"
|
|
)
|
|
|
|
# Destruir la ventana de tkinter
|
|
root.destroy()
|
|
|
|
if directory:
|
|
return jsonify({"status": "success", "path": directory})
|
|
return jsonify({"status": "cancelled"})
|
|
|
|
|
|
@app.route("/api/logs", methods=["GET", "DELETE"])
|
|
def handle_logs():
|
|
if request.method == "GET":
|
|
return jsonify({"logs": config_manager.read_log()})
|
|
else: # DELETE
|
|
success = config_manager.clear_log()
|
|
return jsonify({"status": "success" if success else "error"})
|
|
|
|
|
|
@app.route("/api/group-description/<group>", methods=["GET", "POST"])
|
|
def handle_group_description(group):
|
|
if request.method == "GET":
|
|
try:
|
|
details = config_manager.get_group_details(group)
|
|
if "error" in details:
|
|
return jsonify(details), 404 # Group not found
|
|
return jsonify(details)
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
else: # POST
|
|
try:
|
|
data = request.json
|
|
result = config_manager.update_group_description(group, data)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/script-details/<group>/<script_filename>", methods=["GET", "POST"])
|
|
def handle_script_details(group, script_filename):
|
|
if request.method == "GET":
|
|
try:
|
|
details = config_manager.get_script_details(group, script_filename)
|
|
return jsonify(details)
|
|
except Exception as e:
|
|
print(f"Error getting script details for {group}/{script_filename}: {e}")
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
else: # POST
|
|
try:
|
|
data = request.json
|
|
result = config_manager.update_script_details(group, script_filename, data)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
print(f"Error updating script details for {group}/{script_filename}: {e}")
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/directory-history/<group>")
|
|
def get_directory_history(group):
|
|
history = config_manager.get_directory_history(group)
|
|
return jsonify(history)
|
|
|
|
|
|
@app.route("/api/open-vscode/<group>", methods=["POST"])
|
|
def open_group_in_vscode(group):
|
|
try:
|
|
# Get the full path to the script group directory
|
|
script_group_path = os.path.join(config_manager.script_groups_path, group)
|
|
|
|
if not os.path.isdir(script_group_path):
|
|
return (
|
|
jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"Directorio del grupo '{group}' no encontrado",
|
|
}
|
|
),
|
|
404,
|
|
)
|
|
|
|
# VS Code executable path
|
|
vscode_path = (
|
|
r"C:\Users\migue\AppData\Local\Programs\Microsoft VS Code\Code.exe"
|
|
)
|
|
|
|
# Check if the file exists
|
|
if not os.path.isfile(vscode_path):
|
|
return (
|
|
jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"VS Code no encontrado en: {vscode_path}",
|
|
}
|
|
),
|
|
404,
|
|
)
|
|
|
|
print(f"Launching VS Code from: {vscode_path}")
|
|
print(f"Opening directory: {script_group_path}")
|
|
|
|
# Try with shell=True which can help with Windows path issues
|
|
process = subprocess.Popen(f'"{vscode_path}" "{script_group_path}"', shell=True)
|
|
|
|
# Log the process ID for debugging
|
|
print(f"Process started with PID: {process.pid}")
|
|
|
|
return jsonify(
|
|
{"status": "success", "message": f"VS Code abierto en: {script_group_path}"}
|
|
)
|
|
except Exception as e:
|
|
print(f"Error opening VS Code for group '{group}': {str(e)}")
|
|
return (
|
|
jsonify(
|
|
{"status": "error", "message": f"Error al abrir VS Code: {str(e)}"}
|
|
),
|
|
500,
|
|
)
|
|
|
|
|
|
@app.route("/api/open-miniconda", methods=["POST"])
|
|
def open_miniconda_console():
|
|
try:
|
|
# Path to the Miniconda installation
|
|
miniconda_path = r"C:\Users\migue\miniconda3"
|
|
|
|
# Check if directory exists
|
|
if not os.path.isdir(miniconda_path):
|
|
return (
|
|
jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"Miniconda no encontrado en: {miniconda_path}",
|
|
}
|
|
),
|
|
404,
|
|
)
|
|
|
|
# Path to the activate script
|
|
activate_path = os.path.join(miniconda_path, "Scripts", "activate.bat")
|
|
|
|
if not os.path.isfile(activate_path):
|
|
return (
|
|
jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"Script de activación no encontrado en: {activate_path}",
|
|
}
|
|
),
|
|
404,
|
|
)
|
|
|
|
print(f"Opening Miniconda Console from: {miniconda_path}")
|
|
|
|
# Use subprocess with CREATE_NEW_CONSOLE flag to ensure the window appears
|
|
# Start the Windows command processor and tell it to run the activate batch file
|
|
startupinfo = subprocess.STARTUPINFO()
|
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
startupinfo.wShowWindow = 1 # SW_SHOWNORMAL
|
|
|
|
# Run cmd.exe with specific flags to show a window
|
|
process = subprocess.Popen(
|
|
[
|
|
"cmd.exe",
|
|
"/K",
|
|
f"echo Activating Miniconda environment... && "
|
|
f'"{activate_path}" && '
|
|
f"echo Miniconda activated successfully. && "
|
|
f"cd /d {miniconda_path}",
|
|
],
|
|
creationflags=subprocess.CREATE_NEW_CONSOLE,
|
|
startupinfo=startupinfo,
|
|
)
|
|
|
|
print(f"Miniconda Console process started with PID: {process.pid}")
|
|
|
|
return jsonify(
|
|
{"status": "success", "message": "Miniconda Console abierta correctamente"}
|
|
)
|
|
except Exception as e:
|
|
print(f"Error opening Miniconda Console: {str(e)}")
|
|
return (
|
|
jsonify(
|
|
{
|
|
"status": "error",
|
|
"message": f"Error al abrir Miniconda Console: {str(e)}",
|
|
}
|
|
),
|
|
500,
|
|
)
|
|
|
|
|
|
@app.route("/api/open-explorer", methods=["POST"])
|
|
def open_explorer_route():
|
|
data = request.json
|
|
frontend_path = data.get("path")
|
|
group = data.get("group")
|
|
|
|
if not group:
|
|
return jsonify({"status": "error", "message": "Grupo no proporcionado."}), 400
|
|
if not frontend_path:
|
|
return jsonify({"status": "error", "message": "Ruta no proporcionada."}), 400
|
|
|
|
# Obtener el directorio de trabajo configurado y normalizado para el grupo
|
|
configured_work_dir_raw = config_manager.get_work_dir(group)
|
|
if not configured_work_dir_raw:
|
|
return jsonify({"status": "error", "message": f"No hay directorio de trabajo configurado para el grupo '{group}'."}), 404
|
|
|
|
configured_work_dir_abs = os.path.abspath(configured_work_dir_raw)
|
|
frontend_path_abs = os.path.abspath(frontend_path)
|
|
|
|
# Validar que la ruta del frontend coincide con la ruta configurada
|
|
if configured_work_dir_abs != frontend_path_abs:
|
|
print(f"Intento de acceso no válido: Grupo '{group}', Frontend Path '{frontend_path_abs}', Configured Path '{configured_work_dir_abs}'")
|
|
return jsonify({"status": "error", "message": "La ruta proporcionada no coincide con el directorio de trabajo seguro para este grupo."}), 403
|
|
|
|
# Validar que la ruta (ahora sabemos que es la configurada) realmente existe
|
|
if not os.path.isdir(configured_work_dir_abs):
|
|
return jsonify({"status": "error", "message": f"El directorio de trabajo configurado '{configured_work_dir_abs}' no es un directorio válido o no existe."}), 400
|
|
|
|
try:
|
|
if sys.platform == "win32":
|
|
os.startfile(configured_work_dir_abs)
|
|
elif sys.platform == "darwin": # macOS
|
|
subprocess.Popen(["open", configured_work_dir_abs])
|
|
else: # linux variants
|
|
subprocess.Popen(["xdg-open", configured_work_dir_abs])
|
|
|
|
return jsonify({"status": "success", "message": f"Abriendo '{configured_work_dir_abs}' en el explorador."})
|
|
except Exception as e:
|
|
error_msg = f"Error al abrir el explorador en '{configured_work_dir_abs}': {str(e)}"
|
|
print(error_msg)
|
|
return jsonify({"status": "error", "message": error_msg}), 500
|
|
|
|
|
|
# --- System Tray Icon Functions ---
|
|
|
|
|
|
def run_flask():
|
|
"""Runs the Flask app."""
|
|
print("Starting Flask server on http://127.0.0.1:5000/")
|
|
try:
|
|
# use_reloader=False is important when running in a thread
|
|
# For production, consider using waitress or gunicorn instead of app.run
|
|
app.run(host="127.0.0.1", port=5000, debug=True, use_reloader=False)
|
|
except Exception as e:
|
|
print(f"Error running Flask app: {e}")
|
|
# Optionally try to stop the tray icon if Flask fails critically
|
|
if tray_icon:
|
|
print("Attempting to stop tray icon due to Flask error.")
|
|
tray_icon.stop()
|
|
|
|
|
|
def open_app_browser(icon, item):
|
|
"""Callback function to open the browser."""
|
|
print("Opening application in browser...")
|
|
webbrowser.open("http://127.0.0.1:5000/")
|
|
|
|
|
|
def shutdown_flask_server():
|
|
"""Attempts to gracefully shut down the Werkzeug server."""
|
|
try:
|
|
# This requires the development server (werkzeug)
|
|
# Send a request to a special shutdown route
|
|
requests.post("http://127.0.0.1:5000/_shutdown", timeout=1)
|
|
except Exception as e:
|
|
print(f"Could not send shutdown request to Flask server: {e}")
|
|
print("Flask server might need to be closed manually.")
|
|
|
|
|
|
def stop_icon_thread():
|
|
"""Helper function to stop the icon after a delay, allowing HTTP response."""
|
|
time.sleep(0.1) # Small delay to allow the HTTP response to be sent
|
|
if tray_icon:
|
|
print("Stopping tray icon from shutdown route...")
|
|
tray_icon.stop()
|
|
else:
|
|
print("Tray icon not available to stop.")
|
|
# As a last resort if the icon isn't running for some reason
|
|
# print("Attempting os._exit(0) as fallback.")
|
|
# os._exit(0) # Force exit - use with caution
|
|
|
|
|
|
@app.route("/_shutdown", methods=["POST"])
|
|
def shutdown_route():
|
|
"""Internal route to shut down the application via the tray icon."""
|
|
print("Shutdown endpoint called.")
|
|
# Stop the main application thread by stopping the tray icon.
|
|
# Do this in a separate thread to allow the HTTP response to return first.
|
|
stopper = threading.Thread(target=stop_icon_thread, daemon=True)
|
|
stopper.start()
|
|
print("Shutdown signal sent to tray icon thread.")
|
|
return jsonify(status="success", message="Application shutdown initiated..."), 200
|
|
|
|
|
|
def exit_application(icon, item):
|
|
"""Callback function to exit the application."""
|
|
print("Exit requested via tray menu.")
|
|
# Just stop the icon. This will end the main thread, and the daemon Flask thread will exit.
|
|
print("Stopping tray icon...")
|
|
if icon: # pystray passes the icon object
|
|
icon.stop()
|
|
elif tray_icon: # Fallback just in case
|
|
tray_icon.stop()
|
|
|
|
|
|
# === LAUNCHER GUI APIs ===
|
|
|
|
@app.route("/api/launcher-groups", methods=["GET", "POST"])
|
|
def handle_launcher_groups():
|
|
"""Gestionar grupos de launcher (GET: obtener, POST: crear)"""
|
|
if request.method == "GET":
|
|
try:
|
|
groups = launcher_manager.get_launcher_groups()
|
|
return jsonify(groups)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
else: # POST
|
|
try:
|
|
data = request.json
|
|
result = launcher_manager.add_launcher_group(data)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-groups/<group_id>", methods=["GET", "PUT", "DELETE"])
|
|
def handle_launcher_group(group_id):
|
|
"""Gestionar grupo específico (GET: obtener, PUT: actualizar, DELETE: eliminar)"""
|
|
if request.method == "GET":
|
|
try:
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({"error": "Group not found"}), 404
|
|
return jsonify(group)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
elif request.method == "PUT":
|
|
try:
|
|
data = request.json
|
|
result = launcher_manager.update_launcher_group(group_id, data)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
else: # DELETE
|
|
try:
|
|
result = launcher_manager.delete_launcher_group(group_id)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-scripts/<group_id>")
|
|
def get_launcher_scripts(group_id):
|
|
"""Obtener scripts de un grupo del launcher"""
|
|
try:
|
|
scripts = launcher_manager.get_group_scripts(group_id)
|
|
return jsonify(scripts)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-scripts-all/<group_id>")
|
|
def get_all_launcher_scripts(group_id):
|
|
"""Obtener TODOS los scripts de un grupo (incluyendo ocultos) para gestión"""
|
|
try:
|
|
scripts = launcher_manager.get_all_group_scripts(group_id)
|
|
return jsonify(scripts)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-script-metadata/<group_id>/<script_name>", methods=["GET", "POST"])
|
|
def handle_launcher_script_metadata(group_id, script_name):
|
|
"""Gestionar metadatos de un script específico"""
|
|
if request.method == "GET":
|
|
try:
|
|
metadata = launcher_manager.get_script_metadata(group_id, script_name)
|
|
return jsonify(metadata)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
else: # POST
|
|
try:
|
|
data = request.json
|
|
result = launcher_manager.update_script_metadata(group_id, script_name, data)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/python-environments")
|
|
def get_python_environments():
|
|
"""Obtener entornos de Python disponibles"""
|
|
try:
|
|
envs = launcher_manager.get_available_python_envs()
|
|
return jsonify(envs)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/execute-gui-script", methods=["POST"])
|
|
def execute_gui_script():
|
|
"""Ejecutar script GUI con argumentos opcionales"""
|
|
try:
|
|
data = request.json
|
|
group_id = data["group_id"]
|
|
script_name = data["script_name"]
|
|
script_args = data.get("args", [])
|
|
working_dir = data.get("working_dir", None)
|
|
use_pythonw = data.get("use_pythonw", False) # Por defecto python.exe para logging
|
|
|
|
result = launcher_manager.execute_gui_script(
|
|
group_id, script_name, script_args, broadcast_message, working_dir, use_pythonw
|
|
)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
error_msg = f"Error ejecutando script GUI: {str(e)}"
|
|
broadcast_message(error_msg)
|
|
return jsonify({"error": error_msg}), 500
|
|
|
|
@app.route("/api/launcher-favorites", methods=["GET", "POST"])
|
|
def handle_launcher_favorites():
|
|
"""Gestionar favoritos del launcher"""
|
|
if request.method == "GET":
|
|
try:
|
|
favorites = launcher_manager.get_favorites()
|
|
return jsonify({"favorites": favorites})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
else: # POST
|
|
try:
|
|
data = request.json
|
|
group_id = data["group_id"]
|
|
script_name = data["script_name"]
|
|
result = launcher_manager.toggle_favorite(group_id, script_name)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-history", methods=["GET", "DELETE"])
|
|
def handle_launcher_history():
|
|
"""Gestionar historial del launcher"""
|
|
if request.method == "GET":
|
|
try:
|
|
history = launcher_manager.get_history()
|
|
return jsonify({"history": history})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
else: # DELETE
|
|
try:
|
|
result = launcher_manager.clear_history()
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-categories")
|
|
def get_launcher_categories():
|
|
"""Obtener categorías disponibles del launcher"""
|
|
try:
|
|
categories = launcher_manager.get_categories()
|
|
return jsonify(categories)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/group-icon/<launcher_type>/<group_id>")
|
|
def get_group_icon(launcher_type, group_id):
|
|
"""Obtener icono de un grupo (config o launcher)"""
|
|
try:
|
|
if launcher_type == "launcher":
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({"error": "Group not found"}), 404
|
|
|
|
icon_path = os.path.join(group["directory"], "icon.ico")
|
|
if os.path.exists(icon_path):
|
|
from flask import send_file
|
|
return send_file(icon_path, mimetype='image/x-icon')
|
|
|
|
elif launcher_type == "config":
|
|
group_path = os.path.join(config_manager.script_groups_path, group_id)
|
|
icon_path = os.path.join(group_path, "icon.ico")
|
|
if os.path.exists(icon_path):
|
|
from flask import send_file
|
|
return send_file(icon_path, mimetype='image/x-icon')
|
|
|
|
# Icono por defecto - devolver datos para que el frontend genere el icono
|
|
return jsonify({"type": "default", "icon": "📁"})
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
# Nuevas APIs para gestión de procesos y Markdown
|
|
|
|
@app.route("/api/launcher-process-focus/<int:pid>", methods=["POST"])
|
|
def focus_launcher_process(pid):
|
|
"""Activar foco de un proceso"""
|
|
try:
|
|
result = launcher_manager.focus_process(pid)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-process-terminate/<int:pid>", methods=["POST"])
|
|
def terminate_launcher_process(pid):
|
|
"""Cerrar un proceso"""
|
|
try:
|
|
result = launcher_manager.terminate_process(pid)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-running-processes")
|
|
def get_launcher_running_processes():
|
|
"""Obtener procesos en ejecución"""
|
|
try:
|
|
processes = launcher_manager.get_running_processes()
|
|
return jsonify({"processes": processes})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@app.route("/api/launcher-open-vscode/<group_id>", methods=["POST"])
|
|
def open_launcher_group_in_vscode(group_id):
|
|
"""Abrir grupo del launcher en VS Code"""
|
|
try:
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({"status": "error", "message": "Grupo no encontrado"}), 404
|
|
|
|
script_group_path = group["directory"]
|
|
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo '{group['name']}' no encontrado"
|
|
}), 404
|
|
|
|
# VS Code executable path
|
|
vscode_path = r"C:\Users\migue\AppData\Local\Programs\Microsoft VS Code\Code.exe"
|
|
|
|
if not os.path.isfile(vscode_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"VS Code no encontrado en: {vscode_path}"
|
|
}), 404
|
|
|
|
print(f"Launching VS Code for launcher group: {group['name']}")
|
|
print(f"Opening directory: {script_group_path}")
|
|
|
|
process = subprocess.Popen(f'"{vscode_path}" "{script_group_path}"', shell=True)
|
|
print(f"Process started with PID: {process.pid}")
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": f"VS Code abierto en: {script_group_path}"
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"Error opening VS Code for launcher group '{group_id}': {str(e)}")
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Error al abrir VS Code: {str(e)}"
|
|
}), 500
|
|
|
|
@app.route("/api/launcher-markdown/<group_id>")
|
|
def get_launcher_markdown_files(group_id):
|
|
"""Obtener archivos Markdown de un grupo"""
|
|
try:
|
|
markdown_files = launcher_manager.get_markdown_files(group_id)
|
|
return jsonify({"files": markdown_files})
|
|
except Exception as e:
|
|
print(f"Error getting markdown files for group {group_id}: {e}")
|
|
# Devolver lista vacía en lugar de error para no interferir con scripts
|
|
return jsonify({"files": []})
|
|
|
|
@app.route("/api/launcher-markdown-content/<group_id>/<path:relative_path>")
|
|
def get_launcher_markdown_content(group_id, relative_path):
|
|
"""Obtener contenido de un archivo Markdown"""
|
|
try:
|
|
result = launcher_manager.read_markdown_file(group_id, relative_path)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
# --- Global Error Handler (for debugging unhandled exceptions) ---
|
|
@app.errorhandler(Exception)
|
|
def handle_unhandled_exception(e):
|
|
# Log the error with traceback
|
|
app.logger.error('Unhandled Exception: %s', e, exc_info=True)
|
|
# Return a JSON response for API calls, or HTML for others
|
|
if request.path.startswith('/api/'):
|
|
return jsonify({"status": "error", "message": f"Internal Server Error: {str(e)}"}), 500
|
|
else:
|
|
return "<h1>Internal Server Error</h1><p>An unhandled error occurred.</p>", 500
|
|
|
|
|
|
# === FIN LAUNCHER GUI APIs ===
|
|
|
|
# --- Helper function to find VS Code ---
|
|
def find_vscode_executable():
|
|
"""Intenta encontrar el ejecutable de VS Code en ubicaciones comunes y en el PATH."""
|
|
# Comprobar la variable de entorno VSCODE_PATH primero (si la defines)
|
|
vscode_env_path = os.getenv("VSCODE_PATH")
|
|
if vscode_env_path and os.path.isfile(vscode_env_path):
|
|
return vscode_env_path
|
|
|
|
common_paths = []
|
|
local_app_data = os.getenv('LOCALAPPDATA')
|
|
if local_app_data:
|
|
common_paths.append(os.path.join(local_app_data, r"Programs\Microsoft VS Code\Code.exe"))
|
|
|
|
common_paths.extend([
|
|
r"C:\Program Files\Microsoft VS Code\Code.exe",
|
|
r"C:\Program Files (x86)\Microsoft VS Code\Code.exe",
|
|
])
|
|
for path in common_paths:
|
|
if os.path.isfile(path):
|
|
return path
|
|
return shutil.which("code") # Busca 'code' en el PATH
|
|
|
|
@app.route("/api/open-editor/<editor>/<group_system>/<group_id>", methods=["POST"])
|
|
def open_group_in_editor(editor, group_system, group_id):
|
|
"""Ruta unificada para abrir grupos en diferentes editores"""
|
|
try:
|
|
# Validar editor
|
|
if editor not in ['vscode', 'cursor']:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Editor '{editor}' no soportado. Usar 'vscode' o 'cursor'"
|
|
}), 400
|
|
|
|
# Determinar directorio según el sistema
|
|
if group_system == 'config':
|
|
script_group_path = os.path.join(config_manager.script_groups_path, group_id)
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo config '{group_id}' no encontrado"
|
|
}), 404
|
|
elif group_system == 'launcher':
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Grupo launcher '{group_id}' no encontrado"
|
|
}), 404
|
|
script_group_path = group["directory"]
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo launcher '{group['name']}' no encontrado"
|
|
}), 404
|
|
else:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Sistema de grupo '{group_system}' no válido. Usar 'config' o 'launcher'"
|
|
}), 400
|
|
|
|
# Definir rutas de ejecutables
|
|
if editor == 'vscode':
|
|
editor_path = r"C:\Users\migue\AppData\Local\Programs\Microsoft VS Code\Code.exe"
|
|
editor_name = "VS Code"
|
|
elif editor == 'cursor':
|
|
# Rutas comunes donde se instala Cursor
|
|
possible_cursor_paths = [
|
|
r"C:\Users\migue\AppData\Local\Programs\cursor\Cursor.exe",
|
|
r"C:\Program Files\Cursor\Cursor.exe",
|
|
r"C:\Program Files (x86)\Cursor\Cursor.exe"
|
|
]
|
|
editor_path = None
|
|
for path in possible_cursor_paths:
|
|
if os.path.isfile(path):
|
|
editor_path = path
|
|
break
|
|
|
|
if not editor_path:
|
|
# Intentar buscar en PATH
|
|
editor_path = shutil.which("cursor")
|
|
|
|
if not editor_path:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Cursor no encontrado. Intenté en: {', '.join(possible_cursor_paths)}"
|
|
}), 404
|
|
editor_name = "Cursor"
|
|
|
|
# Verificar que el ejecutable existe
|
|
if not os.path.isfile(editor_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"{editor_name} no encontrado en: {editor_path}"
|
|
}), 404
|
|
|
|
print(f"Launching {editor_name} from: {editor_path}")
|
|
print(f"Opening directory: {script_group_path}")
|
|
|
|
# Ejecutar el editor
|
|
process = subprocess.Popen(f'"{editor_path}" "{script_group_path}"', shell=True)
|
|
print(f"{editor_name} process started with PID: {process.pid}")
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": f"{editor_name} abierto en: {script_group_path}"
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"Error opening {editor} for {group_system} group '{group_id}': {str(e)}")
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Error al abrir {editor}: {str(e)}"
|
|
}), 500
|
|
|
|
@app.route("/api/open-group-folder/<group_system>/<group_id>", methods=["POST"])
|
|
def open_group_folder(group_system, group_id):
|
|
"""Abrir carpeta de un grupo en el explorador de archivos"""
|
|
try:
|
|
# Determinar directorio según el sistema
|
|
if group_system == 'config':
|
|
script_group_path = os.path.join(config_manager.script_groups_path, group_id)
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo config '{group_id}' no encontrado"
|
|
}), 404
|
|
elif group_system == 'launcher':
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Grupo launcher '{group_id}' no encontrado"
|
|
}), 404
|
|
script_group_path = group["directory"]
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo launcher '{group['name']}' no encontrado"
|
|
}), 404
|
|
else:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Sistema de grupo '{group_system}' no válido. Usar 'config' o 'launcher'"
|
|
}), 400
|
|
|
|
# Abrir en el explorador según el sistema operativo
|
|
try:
|
|
if sys.platform == "win32":
|
|
os.startfile(script_group_path)
|
|
elif sys.platform == "darwin": # macOS
|
|
subprocess.Popen(["open", script_group_path])
|
|
else: # linux variants
|
|
subprocess.Popen(["xdg-open", script_group_path])
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": f"Abriendo '{script_group_path}' en el explorador",
|
|
"path": script_group_path
|
|
})
|
|
except Exception as e:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Error al abrir el explorador en '{script_group_path}': {str(e)}"
|
|
}), 500
|
|
|
|
except Exception as e:
|
|
print(f"Error opening folder for {group_system} group '{group_id}': {str(e)}")
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Error al abrir carpeta: {str(e)}"
|
|
}), 500
|
|
|
|
@app.route("/api/get-group-path/<group_system>/<group_id>", methods=["GET"])
|
|
def get_group_path(group_system, group_id):
|
|
"""Obtener el path completo de un grupo de scripts"""
|
|
try:
|
|
# Determinar directorio según el sistema
|
|
if group_system == 'config':
|
|
script_group_path = os.path.join(config_manager.script_groups_path, group_id)
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo config '{group_id}' no encontrado"
|
|
}), 404
|
|
elif group_system == 'launcher':
|
|
group = launcher_manager.get_launcher_group(group_id)
|
|
if not group:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Grupo launcher '{group_id}' no encontrado"
|
|
}), 404
|
|
script_group_path = group["directory"]
|
|
if not os.path.isdir(script_group_path):
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Directorio del grupo launcher '{group['name']}' no encontrado"
|
|
}), 404
|
|
else:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Sistema de grupo '{group_system}' no válido. Usar 'config' o 'launcher'"
|
|
}), 400
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"path": script_group_path
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"Error getting path for {group_system} group '{group_id}': {str(e)}")
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"Error al obtener path: {str(e)}"
|
|
}), 500
|
|
|
|
if __name__ == "__main__":
|
|
# --- Start Flask in a background thread ---
|
|
flask_thread = threading.Thread(target=run_flask, daemon=True)
|
|
flask_thread.start()
|
|
|
|
# --- Setup and run the system tray icon ---
|
|
icon_path = (
|
|
r"d:\Proyectos\Scripts\ParamManagerScripts\icon.png" # Use absolute path
|
|
)
|
|
try:
|
|
image = Image.open(icon_path)
|
|
menu = pystray.Menu(
|
|
pystray.MenuItem("Abrir ParamManager", open_app_browser, default=True),
|
|
pystray.MenuItem("Salir", exit_application),
|
|
)
|
|
tray_icon = pystray.Icon("ParamManager", image, "ParamManager", menu)
|
|
print("Starting system tray icon...")
|
|
tray_icon.run() # This blocks the main thread until icon.stop() is called
|
|
except FileNotFoundError:
|
|
print(
|
|
f"Error: Icono no encontrado en '{icon_path}'. El icono de notificación no se iniciará.",
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
"La aplicación Flask seguirá ejecutándose en segundo plano. Presiona Ctrl+C para detenerla si es necesario."
|
|
)
|
|
# Keep the main thread alive so the Flask thread doesn't exit immediately
|
|
# This allows Flask to continue running even without the tray icon.
|
|
try:
|
|
while flask_thread.is_alive():
|
|
flask_thread.join(timeout=1.0) # Wait indefinitely
|
|
except KeyboardInterrupt:
|
|
print("\nCtrl+C detectado. Intentando detener Flask...")
|
|
shutdown_flask_server() # Try to shutdown Flask on Ctrl+C too
|
|
print("Saliendo.")
|
|
except Exception as e:
|
|
print(f"Error al iniciar el icono de notificación: {e}", file=sys.stderr)
|
|
|
|
print("Aplicación finalizada.")
|