#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Script de déploiement CLI standalone. Utilisable après un git clone, sans besoin d'un agent XMPP actif. Usage : python3 deploy.py # déploiement interactif (local ou SSH) """ import sys import re import getpass import json import subprocess from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from deployer import deploy_agent, load_catalog COLORS = { "reset" : "\033[0m", "bold" : "\033[1m", "green" : "\033[32m", "yellow": "\033[33m", "red" : "\033[31m", "cyan" : "\033[36m", } def c(color: str, text: str) -> str: return COLORS.get(color, "") + text + COLORS["reset"] # ── Helpers ─────────────────────────────────────────────────────────────── def choose_agent(catalog: dict) -> str: print(c("cyan", "\nAgents disponibles :")) agents = list(catalog.items()) for i, (name, info) in enumerate(agents, 1): print(" {}. {} — {}".format(i, c("bold", name), info["description"])) print() while True: choice = input("Choisissez un agent (numéro ou nom) : ").strip() if choice.isdigit(): idx = int(choice) - 1 if 0 <= idx < len(agents): return agents[idx][0] elif choice in catalog: return choice print(c("red", "Choix invalide, réessayez.")) def choose_name(agent_type: str) -> str: print(c("cyan", "\nNom de l'agent :")) print(" Ce nom sera utilisé dans vos commandes (ex: \"mets à jour trouducul\"),") print(" comme identifiant MQTT et nom du service systemd.") print(" Lettres, chiffres, tirets et underscores uniquement.") while True: name = input(" Nom [{}] : ".format(agent_type)).strip().lower() or agent_type if re.match(r'^[a-z0-9_-]+$', name): return name print(c("red", " Nom invalide. Utilisez lettres, chiffres, tirets ou underscores.")) def choose_mode() -> str: print(c("cyan", "\nMode de déploiement :")) print(" 1. Localement (sur cette machine)") print(" 2. Machine distante via SSH") while True: choice = input(" Choix (1/2) : ").strip() if choice == "1": return "local" if choice == "2": return "ssh" print(c("red", " Répondez 1 ou 2.")) def collect_ssh_info() -> dict: print(c("cyan", "\nInformations SSH :")) host = input(" Adresse IP de la machine cible : ").strip() user = input(" Nom d'utilisateur SSH : ").strip() print(" Authentification :") print(" 1. Mot de passe") print(" 2. Clé SSH") while True: auth = input(" Choix (1/2) : ").strip() if auth == "1": password = getpass.getpass(" Mot de passe SSH : ") return {"host": host, "user": user, "auth": "password", "password": password, "key_path": None} elif auth == "2": key_path = input(" Chemin de la clé SSH [~/.ssh/id_rsa] : ").strip() if not key_path: key_path = str(Path.home() / ".ssh" / "id_rsa") return {"host": host, "user": user, "auth": "key", "password": None, "key_path": key_path} print(c("red", " Répondez 1 ou 2.")) def collect_agent_config(agent_name: str, host: str) -> dict: print(c("cyan", "\nConfiguration XMPP/MQTT :")) default_jid = "{}@xmpp.ovh".format(agent_name) xmpp_jid = input(" JID XMPP [{}] : ".format(default_jid)).strip() or default_jid xmpp_pass = getpass.getpass(" Mot de passe XMPP : ") default_mqtt = host if host else "localhost" mqtt_host = input(" Broker MQTT [{}] : ".format(default_mqtt)).strip() or default_mqtt return {"xmpp_jid": xmpp_jid, "xmpp_pass": xmpp_pass, "mqtt_host": mqtt_host} # ── Déploiement local ────────────────────────────────────────────────────── def deploy_local(agent_type: str, agent_name: str, catalog: dict): info = catalog[agent_type] repo_url = info["repo_url"] main_script = info["main_script"] deps = info["dependencies"] install_path = "/opt/{}".format(agent_name) service_name = agent_name print(c("cyan", "\nDéploiement local de «{}» ({}) dans {}...".format( agent_name, agent_type, install_path))) def run(cmd): result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if result.returncode != 0: print(c("red", " Erreur : {}".format(result.stderr.strip() or result.stdout.strip()))) sys.exit(1) return result.stdout.strip() print(" [1/5] Installation des prérequis...") run("apt-get install -y -qq python3 python3-pip python3-venv git 2>&1") print(" [2/5] Clonage du dépôt...") if Path(install_path + "/.git").exists(): run("git -C {} pull".format(install_path)) else: run("git clone {} {}".format(repo_url, install_path)) print(" [3/5] Création du venv et installation des dépendances...") run("python3 -m venv {}/venv".format(install_path)) run("{}/venv/bin/pip install -q {}".format(install_path, " ".join(deps))) print(" [4/5] Configuration...") agent_cfg = collect_agent_config(agent_name, "localhost") config = { "ollama_url" : "http://192.168.7.119:11434/api/chat", "model" : "qwen3:8b", "xmpp_jid" : agent_cfg["xmpp_jid"], "xmpp_pass" : agent_cfg["xmpp_pass"], "admin_jid" : "sylvain@xmpp.ovh", "db_path" : "{}/memory.db".format(install_path), "mqtt_host" : agent_cfg["mqtt_host"], "mqtt_port" : 1883, "mqtt_client_id": agent_name, "mqtt_inbox" : "agents/{}/inbox".format(agent_name), "mqtt_outbox" : "agents/agent1/inbox", } config_path = Path(install_path) / "config" / "config.json" config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False)) print(" [5/5] Création et démarrage du service systemd...") service = """[Unit] Description=Agent {name} After=network.target mosquitto.service Wants=mosquitto.service [Service] Type=simple WorkingDirectory={path} ExecStart={path}/venv/bin/python3 {path}/{script} Restart=always RestartSec=5 StandardOutput=journal StandardError=journal [Install] WantedBy=multi-user.target """.format(name=agent_name, path=install_path, script=main_script) Path("/etc/systemd/system/{}.service".format(service_name)).write_text(service) run("systemctl daemon-reload && systemctl enable {s} && systemctl start {s}".format( s=service_name)) print(c("green", "\nDéploiement local de «{}» terminé !".format(agent_name))) print(" XMPP JID : {}".format(agent_cfg["xmpp_jid"])) print(" MQTT inbox : agents/{}/inbox".format(agent_name)) print(" Service : systemctl status {}".format(service_name)) print(" Logs : journalctl -u {} -f".format(service_name)) # ── Main ─────────────────────────────────────────────────────────────────── def main(): print(c("bold", "\n=== Agent Deploy — Déploiement interactif ===")) catalog = load_catalog() agent_type = choose_agent(catalog) agent_name = choose_name(agent_type) mode = choose_mode() if mode == "local": deploy_local(agent_type, agent_name, catalog) return # Mode SSH ssh = collect_ssh_info() agent_cfg = collect_agent_config(agent_name, ssh["host"]) print(c("cyan", "\nRécapitulatif :")) print(" Type : {}".format(agent_type)) print(" Nom : {}".format(c("bold", agent_name))) print(" Repo : {}".format(catalog[agent_type]["repo_url"])) print(" Machine : {}@{}".format(ssh["user"], ssh["host"])) print(" Auth SSH : {}".format("clé SSH" if ssh["auth"] == "key" else "mot de passe")) print(" XMPP JID : {}".format(agent_cfg["xmpp_jid"])) print(" MQTT broker: {}".format(agent_cfg["mqtt_host"])) print(" MQTT inbox : agents/{}/inbox".format(agent_name)) print(" Service : {}.service".format(agent_name)) print() confirm = input("Confirmer le déploiement ? (oui/non) : ").strip().lower() if confirm not in ("oui", "o", "yes", "y"): print(c("yellow", "Déploiement annulé.")) sys.exit(0) print(c("cyan", "\nDéploiement en cours...")) def progress(msg): print(c("yellow", " [{}]".format(msg))) success, result = deploy_agent( host = ssh["host"], ssh_user = ssh["user"], agent_type = agent_type, agent_name = agent_name, xmpp_jid = agent_cfg["xmpp_jid"], xmpp_pass = agent_cfg["xmpp_pass"], mqtt_host = agent_cfg["mqtt_host"], progress_cb = progress, ssh_password= ssh["password"], ssh_key_path= ssh["key_path"], ) if success: print(c("green", "\n" + result)) else: print(c("red", "\nÉchec : " + result)) sys.exit(1) if __name__ == "__main__": main()