#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Creation en masse de paquets Medulla depuis un CSV, a lancer SUR le serveur Medulla. Chaque ligne du CSV devient un paquet, cree ou mis a jour par la meme API que la console web (mmc-agent, plugin pkgs) : partage, base, liens, droits, hash et synchronisation vers les relais sont geres comme dans la console. Usage (root, sur le serveur Medulla) : python3 medulla_import_csv.py paquets.csv --verifier # TOUS les controles, rien n'est modifie python3 medulla_import_csv.py paquets.csv # memes controles, confirmation, puis import python3 medulla_import_csv.py --modele exemple.csv # ecrit un CSV d'exemple python3 medulla_import_csv.py --lister # organisations / partages accessibles Options : -v (detail de chaque etape), --adopter-existants (mettre a jour les paquets deja presents au lieu de les refuser), --oui (sans confirmation), --verifier --hors-ligne (hors serveur). Deroulement : 1. format du CSV 2. contenu (fichiers, organisations, partages, droits, quotas, uuid, doublons) 3. serveur (sauvegarde possible, ecriture, espace disque) 4. confirmation 5. sauvegarde de la base pkgs et des paquets mis a jour 6. creation / mise a jour. Une erreur aux etapes 1 a 5 arrete tout avant la moindre modification. Mot de passe du compte MMC (root par defaut) : variable MMC_PASSWORD, sinon demande. Colonnes (separateur , ou ; ; obligatoires : nom, partage ou organisation, et script_installation ou script_desinstallation) : nom nom du paquet version defaut 1.0 description defaut : nom os win (defaut), linux ou mac partage nom du partage Medulla (ex. global) organisation ou : nom de l'organisation ; le paquet va dans le partage de cette organisation. Nom complet possible (ex. "Entite racine > Client A"). Au moins partage ou organisation ; si les deux, ils doivent concorder. script_installation script lance par la section Installation arguments_installation arguments ajoutes apres le script timeout_installation secondes, defaut 3600 script_desinstallation script lance par la section Desinstallation (facultatif) arguments_desinstallation timeout_desinstallation secondes, defaut 3600 fichiers autres fichiers a embarquer (installeur...), separes par | redemarrage 0 (defaut) ou 1 : redemarrage apres succes inventaire_apres noforced (defaut, si changement), forced ou non inventaire_logiciel logiciel a associer dans l'inventaire (facultatif) inventaire_version version associee inventaire_editeur editeur associe licences nombre de licences (entier) editable oui (defaut, mode expert dans la console) ou non uuid vide = nouveau paquet ; rempli automatiquement apres creation. Relancer avec l'uuid met le paquet a jour (fichiers remplaces). Chemins de fichiers : absolus, ou relatifs au dossier du CSV. Lanceur deduit de l'extension du script : win .ps1 powershell -File | .bat/.cmd cmd /c | .vbs cscript | .msi msiexec | .exe direct linux .sh bash | .py python3 mac .sh bash | .py python3 | .pkg installer .msi sans arguments : /i ... /qn /norestart en installation, /x ... /qn /norestart en desinstallation. """ import argparse import configparser import csv import datetime import getpass import glob import io import json import os import re import shutil import subprocess import tarfile import tempfile import gzip import ssl import sys import uuid as uuidlib import xmlrpc.client import base64 PACKAGES = "/var/lib/pulse2/packages" TMPDIR = "/var/lib/pulse2/package-server-tmpdir" AGENT_CONF = ("/etc/mmc/agent/config.ini", "/etc/mmc/agent/config.ini.local") PKGS_CONF = ("/etc/mmc/plugins/pkgs.ini", "/etc/mmc/plugins/pkgs.ini.local") BACKUP_ROOT = "/var/lib/pulse2/sauvegardes_import_csv" COLUMNS = [ "nom", "version", "description", "os", "partage", "organisation", "script_installation", "arguments_installation", "timeout_installation", "script_desinstallation", "arguments_desinstallation", "timeout_desinstallation", "fichiers", "redemarrage", "inventaire_apres", "inventaire_logiciel", "inventaire_version", "inventaire_editeur", "licences", "editable", "uuid", ] OS_KEY = {"win": "win", "linux": "linux", "mac": "darwin"} # cle du workflow xmppdeploy INVENTORY_AFTER = {"noforced": "noforced", "forced": "forced", "non": "False", "no": "False", "false": "False"} YES = {"oui", "o", "yes", "y", "1", "true", "vrai"} NO = {"non", "n", "no", "0", "false", "faux"} RESERVED = {"conf.json", "xmppdeploy.json"} # Tailles des colonnes de pkgs.packages (schema Medulla) : au-dela, la base tronque ou refuse. LIMITS = {"description": 255, "version": 50, "inventaire_logiciel": 50, "inventaire_version": 50, "inventaire_editeur": 50, "licences": 20, "nom": 1024} MARGIN = 200 * 1048576 # marge disque (octets) en plus des besoins calcules EXAMPLE = [ {"nom": "7-Zip", "version": "24.08", "description": "7-Zip x64 silencieux", "os": "win", "partage": "global", "script_installation": "installeurs/7z2408-x64.msi", "script_desinstallation": "installeurs/7z2408-x64.msi", "inventaire_logiciel": "7-Zip", "inventaire_version": "24.08", "inventaire_editeur": "Igor Pavlov"}, {"nom": "Notepad++", "version": "8.9.8.1", "os": "win", "partage": "global", "script_installation": "scripts/install_npp.ps1", "arguments_installation": "-Silent", "script_desinstallation": "scripts/uninstall_npp.ps1", "fichiers": "installeurs/npp.8.9.8.1.Installer.x64.exe", "timeout_installation": "1800"}, {"nom": "Agent supervision", "version": "2.1", "os": "linux", "organisation": "Client A", "script_installation": "scripts/install_agent.sh", "fichiers": "installeurs/agent_2.1_amd64.deb", "redemarrage": "0", "inventaire_apres": "forced"}, ] class RowError(Exception): pass VERBOSE = False def debug(message): if VERBOSE: print(f" [detail] {message}") def short(value, limit=160): text = repr(value) return text if len(text) <= limit else text[:limit] + "..." def call(proxy, method, *args): """Appel XML-RPC trace en mode verbeux (nom, arguments abreges, duree, resultat).""" target = proxy for part in method.split("."): target = getattr(target, part) debug(f"appel {method}{short(args)}") start = datetime.datetime.now() result = target(*args) debug(f" -> {short(result)} ({(datetime.datetime.now() - start).total_seconds():.2f}s)") return result # ----------------------------------------------------------------------------- CSV def check_format(path): """Etape 1, toujours executee : controle du format du fichier, sans regarder le contenu metier. Retourne (lignes, separateur, en-tetes, remarques) ou arrete le script avec la liste des problemes.""" problems, notes = [], [] try: raw = open(path, "rb").read() except OSError as exc: raise SystemExit(f"ERREUR : CSV illisible : {exc}") if not raw.strip(): raise SystemExit("ERREUR : CSV vide") if raw.startswith(b"\xef\xbb\xbf"): raw = raw[3:] notes.append("BOM UTF-8 detecte (export Excel) : accepte") try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: line = raw[:exc.start].count(b"\n") + 1 raise SystemExit(f"ERREUR format : le fichier n'est pas en UTF-8 (ligne {line}, octet {exc.start}).\n" " Dans Excel : Enregistrer sous > 'CSV UTF-8 (delimite par des virgules)'.") if "\r\n" in text: notes.append("fins de ligne Windows (CRLF) : acceptees") header_line = text.splitlines()[0] nb_comma, nb_semi = header_line.count(","), header_line.count(";") if nb_comma and nb_semi: raise SystemExit(f"ERREUR format : l'en-tete melange les separateurs , ({nb_comma}) et ; ({nb_semi}).") if not nb_comma and not nb_semi: raise SystemExit("ERREUR format : aucun separateur , ou ; dans l'en-tete (une seule colonne ?)") delimiter = ";" if nb_semi else "," reader = csv.reader(io.StringIO(text, newline=""), delimiter=delimiter, strict=True) records, line_of = [], [] while True: start = reader.line_num + 1 try: record = next(reader) except StopIteration: break except csv.Error as exc: problems.append(f"ligne {start} : {exc} (guillemet non ferme ? un guillemet dans une cellule " "doit etre double : \"\") ; lecture arretee a cette ligne") break records.append(record) line_of.append(start) headers, lowered = [], {} for index, name in enumerate(records[0], start=1): clean = name.strip().lower() if not clean: problems.append(f"en-tete : colonne {index} sans nom") continue if clean != name: notes.append(f"en-tete : '{name}' lu comme '{clean}' (espaces / majuscules ignores)") if clean in lowered: problems.append(f"en-tete : colonne '{clean}' en double (colonnes {lowered[clean]} et {index})") lowered.setdefault(clean, index) headers.append(clean) unknown = [h for h in headers if h not in COLUMNS] if unknown: problems.append(f"en-tete : colonne(s) inconnue(s) {', '.join(unknown)} " f"(colonnes possibles : {', '.join(COLUMNS)})") if "nom" not in headers: problems.append("en-tete : colonne obligatoire absente : nom") if "partage" not in headers and "organisation" not in headers: problems.append("en-tete : colonne partage ou organisation obligatoire") if "script_installation" not in headers and "script_desinstallation" not in headers: problems.append("en-tete : colonne script_installation ou script_desinstallation obligatoire") rows, empty = [], 0 width = len(records[0]) for index, cells in enumerate(records[1:], start=1): number = line_of[index] if not any(c.strip() for c in cells): empty += 1 continue if len(cells) != width: hint = ("une virgule ou un ; dans une cellule doit etre entre guillemets" if len(cells) > width else "des separateurs manquent en fin de ligne") problems.append(f"ligne {number} : {len(cells)} cellule(s) au lieu de {width} ({hint})") continue for name, value in zip(records[0], cells): if "\n" in value or "\r" in value: problems.append(f"ligne {number}, colonne {name.strip()} : retour a la ligne dans la cellule") if "\t" in value: problems.append(f"ligne {number}, colonne {name.strip()} : tabulation dans la cellule") row = {c: "" for c in COLUMNS} for name, value in zip(headers, cells): if name in row: row[name] = value.strip() rows.append((number, row)) if empty: notes.append(f"{empty} ligne(s) vide(s) ignoree(s)") if not rows and not problems: problems.append("aucune ligne de paquet sous l'en-tete") print(f"Test du format CSV : {os.path.basename(path)}") print(f" encodage UTF-8, separateur '{delimiter}', {width} colonne(s), {len(rows)} ligne(s) de paquet") for note in notes: print(f" note : {note}") if problems: print(f" FORMAT REFUSE : {len(problems)} probleme(s), rien n'a ete modifie.") for problem in problems: print(f" - {problem}") raise SystemExit(1) print(" format OK") return rows, delimiter, headers def write_csv(path, delimiter, headers, rows): if "uuid" not in headers: headers = headers + ["uuid"] tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=headers, delimiter=delimiter, extrasaction="ignore") writer.writeheader() for _, row in rows: writer.writerow(row) shutil.copymode(path, tmp) if os.path.exists(path) else None os.replace(tmp, path) # ----------------------------------------------------------------------------- validation def launcher(script, args, targetos, section): name = os.path.basename(script) ext = os.path.splitext(name)[1].lower() args = args.strip() q = f'"{name}"' if targetos == "win": if ext == ".ps1": cmd = f"powershell.exe -NoProfile -ExecutionPolicy Bypass -File {q}" elif ext in (".bat", ".cmd"): cmd = f"cmd.exe /c {q}" elif ext == ".vbs": cmd = f"cscript.exe //NoLogo {q}" elif ext == ".msi": if not args: return (f"msiexec.exe /i {q} /qn /norestart" if section == "installation" else f"msiexec.exe /x {q} /qn /norestart") cmd = f"msiexec.exe {'/i' if section == 'installation' else '/x'} {q}" elif ext == ".exe": cmd = q else: raise RowError(f"script_{section} : extension {ext or '(aucune)'} non geree sous Windows " "(.ps1 .bat .cmd .vbs .msi .exe)") else: if ext == ".sh": cmd = f"/bin/bash {q}" elif ext == ".py": cmd = f"python3 {q}" elif ext == ".pkg" and targetos == "mac": if section != "installation": raise RowError("script_desinstallation : un .pkg ne sait pas desinstaller, fournir un .sh") cmd = f"installer -pkg {q} -target /" else: raise RowError(f"script_{section} : extension {ext or '(aucune)'} non geree sous {targetos} (.sh .py" + (" .pkg" if targetos == "mac" else "") + ")") return f"{cmd} {args}".strip() def resolve(path, base, column): if "|" in path: raise RowError(f"{column} : un seul fichier attendu") full = path if os.path.isabs(path) else os.path.join(base, path) full = os.path.abspath(full) if not os.path.isfile(full): near = _close_name(full) raise RowError(f"{column} : fichier introuvable {full}" + (f" (vouliez-vous dire {near} ?)" if near else "")) if not os.access(full, os.R_OK): raise RowError(f"{column} : fichier illisible (droits) {full}") if os.path.basename(full).lower() in RESERVED: raise RowError(f"{column} : nom de fichier reserve {os.path.basename(full)}") return full def _close_name(path): """Propose un fichier au nom proche (casse, espaces) pour eviter un aller-retour.""" folder, name = os.path.split(path) if not os.path.isdir(folder): return None key = re.sub(r"[\s_-]", "", name.lower()) for other in os.listdir(folder): if re.sub(r"[\s_-]", "", other.lower()) == key: return other return None def positive_int(value, default, column, allow_zero=False): value = value or default if not value.isdigit() or (int(value) == 0 and not allow_zero): raise RowError(f"{column} invalide '{value}' (entier{' >= 0' if allow_zero else ' > 0'})") return int(value) def clean_text(text): """save_xmpp_json double les antislashs avant json.loads : on evite \\ et \" dans les textes.""" return " ".join(text.replace("\\", "/").replace('"', "'").split()) def validate(row, base, shares, warnings): for column, limit in LIMITS.items(): if len(clean_text(row[column])) > limit: raise RowError(f"{column} trop long ({len(clean_text(row[column]))} caracteres, {limit} maximum)") name = clean_text(row["nom"]) if not name: raise RowError("nom obligatoire") version = clean_text(row["version"]) or "1.0" targetos = (row["os"] or "win").lower() if targetos not in OS_KEY: raise RowError(f"os invalide '{row['os']}' (win, linux ou mac)") if not row["partage"] and not row["organisation"]: raise RowError("partage ou organisation obligatoire (ex. partage global)") share = resolve_share(row["partage"], row["organisation"], shares) files, sections = [], [] for section in ("installation", "desinstallation"): script = row[f"script_{section}"] if not script: if row[f"arguments_{section}"]: raise RowError(f"arguments_{section} renseigne sans script_{section}") continue path = resolve(script, base, f"script_{section}") if os.path.getsize(path) == 0: raise RowError(f"script_{section} : fichier vide {path}") if path.lower().endswith(".exe") and not row[f"arguments_{section}"].strip(): warnings.append(f"script_{section} {os.path.basename(path)} sans arguments : sans option " "silencieuse, l'installeur peut attendre un clic et bloquer jusqu'au timeout") command = launcher(path, row[f"arguments_{section}"], targetos, section) if targetos == "win": try: command.encode("cp1252") except UnicodeEncodeError: raise RowError(f"arguments_{section} : caractere non pris en charge par Windows (cp1252)") timeout = positive_int(row[f"timeout_{section}"], "3600", f"timeout_{section}") sections.append((section, command, timeout)) files.append(path) if not sections: raise RowError("script_installation ou script_desinstallation obligatoire") for item in [f.strip() for f in row["fichiers"].split("|")] if row["fichiers"] else []: if not item: raise RowError("fichiers : entree vide (| en trop)") extra = resolve(item, base, "fichiers") if os.path.getsize(extra) == 0: warnings.append(f"fichiers : {os.path.basename(extra)} est vide (0 octet)") files.append(extra) by_name = {} for path in files: other = by_name.setdefault(os.path.basename(path), path) if other != path: raise RowError(f"deux fichiers differents nommes {os.path.basename(path)}") reboot = row["redemarrage"] or "0" if reboot not in ("0", "1"): raise RowError(f"redemarrage invalide '{reboot}' (0 ou 1)") inventory_after = INVENTORY_AFTER.get((row["inventaire_apres"] or "noforced").lower()) if not inventory_after: raise RowError(f"inventaire_apres invalide '{row['inventaire_apres']}' (noforced, forced ou non)") software = clean_text(row["inventaire_logiciel"]) if not software and (row["inventaire_version"] or row["inventaire_editeur"] or row["licences"]): raise RowError("inventaire_version/editeur ou licences renseigne sans inventaire_logiciel") licenses = row["licences"] if licenses: positive_int(licenses, "", "licences", allow_zero=True) editable = (row["editable"] or "oui").lower() if editable not in YES | NO: raise RowError(f"editable invalide '{row['editable']}' (oui ou non)") pkg_uuid = row["uuid"] if pkg_uuid and not re.fullmatch(r"[A-Za-z0-9_-]{36}", pkg_uuid): raise RowError(f"uuid invalide '{pkg_uuid}' (36 caracteres A-Z a-z 0-9 _ -)") return { "name": name, "version": version, "description": clean_text(row["description"]) or name, "targetos": targetos, "share": share, "sections": sections, "files": list(by_name.values()), "reboot": int(reboot), "inventory_after": inventory_after, "software": software, "inv_version": clean_text(row["inventaire_version"]), "inv_vendor": clean_text(row["inventaire_editeur"]), "licenses": licenses, "metagenerator": "expert" if editable in YES else "manual", "uuid": pkg_uuid, } # ----------------------------------------------------------------------------- descripteurs def xmppdeploy(pkg, pkg_uuid, now, creator, creation_date): seq = [] for section, command, timeout in pkg["sections"]: upper = "INSTALL" if section == "installation" else "UNINSTALL" seq.append({"step": len(seq), "action": f"action_section_{upper.lower()}", "actionlabel": f"label_section_{upper.lower()}"}) seq.append({ "step": len(seq), "action": "actionprocessscript", "actionlabel": f"{upper}_COMMAND", "command": base64.b64encode(command.encode("utf-8")).decode("ascii"), "timeout": str(timeout), "codereturn": "", "10@lastlines": "10@lastlines", "gotoreturncode@!=0": "END_ERROR", }) success = len(seq) + (1 if pkg["reboot"] else 0) if pkg["reboot"]: seq.append({"step": len(seq), "action": "actionrestart", "actionlabel": "REBOOT"}) seq.append({"step": success, "action": "actionsuccescompletedend", "actionlabel": "END_SUCCESS", "clear": "True", "inventory": pkg["inventory_after"]}) seq.append({"step": success + 1, "action": "actionerrorcompletedend", "actionlabel": "END_ERROR"}) oskey = OS_KEY[pkg["targetos"]] return { "info": { "name": pkg["name"], "packageUuid": pkg_uuid, "software": pkg["software"] or pkg["name"], "version": pkg["version"], "description": pkg["description"], "metagenerator": pkg["metagenerator"], "Dependency": [], "localisation_server": pkg["share"], "previous_localisation_server": pkg["share"], "creator": creator, "creation_date": creation_date, "editor": creator, "edition_date": now, "transferfile": True, "methodetransfert": "pushrsync", "limit_rate_ko": "", "spooling": "ordinary", "gotoreturncode": "!=0", "gotolabel": "END_ERROR", "targetrestart": "MA", "url": "", "gotono": "", "inventory": pkg["inventory_after"], "associateinventory": "on" if pkg["software"] else "", "licenses": pkg["licenses"], }, oskey: {"sequence": seq}, "metaparameter": {oskey: {"label": {s["actionlabel"]: s["step"] for s in seq}}, "os": [oskey], "uuid": pkg_uuid}, } def package_detail(pkg, pkg_uuid, mode, login, now, creation_date): first_command = pkg["sections"][0][1] return { "id": pkg_uuid, "label": pkg["name"], "version": pkg["version"], "description": pkg["description"], "mode": mode, "targetos": pkg["targetos"], "metagenerator": pkg["metagenerator"], "localisation_server": pkg["share"], "previous_localisation_server": pkg["share"], "creator": login, "creation_date": creation_date, "editor": login, "edition_date": now, "reboot": pkg["reboot"], "associateinventory": 1 if pkg["software"] else 0, "licenses": pkg["licenses"], "Qsoftware": pkg["software"], "Qversion": pkg["inv_version"], "Qvendor": pkg["inv_vendor"], "boolcnd": "", "command": {"name": "", "command": first_command}, "sub_packages": [], } # ----------------------------------------------------------------------------- mmc-agent class CookieTransport(xmlrpc.client.SafeTransport): def __init__(self, use_ssl, context): super().__init__(context=context) self.use_ssl, self.cookie = use_ssl, None def make_connection(self, host): if self.use_ssl: return super().make_connection(host) return xmlrpc.client.Transport.make_connection(self, host) def send_headers(self, connection, headers): if self.cookie: connection.putheader("Cookie", self.cookie) super().send_headers(connection, headers) def parse_response(self, response): cookie = response.getheader("Set-Cookie") if cookie: self.cookie = cookie.split(";", 1)[0] return super().parse_response(response) def connect(user, password): conf = configparser.RawConfigParser() if not conf.read(AGENT_CONF): raise SystemExit(f"ERREUR : configuration mmc-agent introuvable ({AGENT_CONF[0]}) : " "lancer ce script sur le serveur Medulla") host = conf.get("main", "host", fallback="127.0.0.1") port = conf.get("main", "port", fallback="7080") use_ssl = conf.getboolean("main", "enablessl", fallback=True) login = conf.get("main", "login", fallback="mmc") secret = conf.get("main", "password", fallback="s3cr3t") context = ssl._create_unverified_context() if use_ssl else None # certificat local auto-signe url = f"{'https' if use_ssl else 'http'}://{login}:{secret}@{host}:{port}/" debug(f"mmc-agent {'https' if use_ssl else 'http'}://{host}:{port}/ (identifiant HTTP {login}, " f"configuration {', '.join(f for f in AGENT_CONF if os.path.exists(f))})") proxy = xmlrpc.client.ServerProxy(url, transport=CookieTransport(use_ssl, context), allow_none=True) try: if not _auth(proxy, user, password): raise SystemExit(f"ERREUR : authentification MMC refusee pour {user}") except (OSError, xmlrpc.client.ProtocolError) as exc: raise SystemExit(f"ERREUR : mmc-agent injoignable sur {host}:{port} ({exc})") return proxy def _auth(proxy, user, password): debug(f"appel base.ldapAuth('{user}', '***')") ok = proxy.base.ldapAuth(user, password) debug(f" -> {ok}") return ok def load_shares(proxy, user): """Partages visibles par le compte, avec l'organisation associee (colonne comments de pkgs_shares).""" result = call(proxy, "pkgs.pkgs_search_share", {"login": user}) shares = [] for item in result.get("datas", []): shares.append({ "name": str(item.get("name", "")), "organisation": str(item.get("comments") or "").strip().strip('"'), "permission": str(item.get("permission", "")), "quotas": int(item.get("quotas") or 0), # Go, 0 = illimite "used": int(item.get("usedquotas") or 0), # octets "path": str(item.get("share_path", "")), }) debug(f"{len(shares)} partage(s) visibles pour {user} ; centralizedmultiplesharing=" f"{result.get('config', {}).get('centralizedmultiplesharing')}") for share in shares: debug(f" partage {share['name']!r} organisation {share['organisation']!r} " f"droits {share['permission']!r} quota {share['quotas']} Go") return shares def _norm(text): return " ".join(str(text).replace(">", " > ").split()).lower() def resolve_share(partage, organisation, shares): """Retourne le nom du partage ; hors connexion (--verifier) le nom saisi est garde tel quel.""" if shares is None: return partage or f"(partage de l'organisation {organisation})" by_share = None if partage: found = [s for s in shares if _norm(s["name"]) == _norm(partage)] or \ [s for s in shares if s["organisation"] and _norm(s["organisation"]) == _norm(partage)] if not found: raise RowError(f"partage '{partage}' inconnu pour ce compte " f"(disponibles : {', '.join(sorted(s['name'] for s in shares))} ; voir --lister)") by_share = found[0] by_org = None if organisation: found = [s for s in shares if s["organisation"] and _norm(s["organisation"]) == _norm(organisation)] if not found: # nom court : derniere partie du nom complet, ou nom du partage found = [s for s in shares if _norm(s["organisation"].split(">")[-1]) == _norm(organisation) or _norm(s["name"]) == _norm(organisation)] if not found: raise RowError(f"organisation '{organisation}' inconnue ou sans partage accessible (voir --lister)") if len({s["organisation"] for s in found}) > 1: raise RowError(f"organisation '{organisation}' ambigue : " + " ; ".join(s["organisation"] for s in found) + " -> utiliser le nom complet") by_org = found[0] share = by_share or by_org twins = [s for s in shares if s["name"] == share["name"]] if len(twins) > 1: raise RowError(f"{len(twins)} partages portent le nom '{share['name']}' (" + " ; ".join(s["organisation"] for s in twins) + ") : Medulla range les paquets par nom de partage et ne peut pas les distinguer ; " "renommer l'une des organisations avant l'import") if by_share and by_org and by_share["name"] != by_org["name"]: raise RowError(f"partage '{partage}' et organisation '{organisation}' ne concordent pas " f"(l'organisation utilise le partage '{by_org['name']}')") if "w" not in share["permission"]: raise RowError(f"partage '{share['name']}' en lecture seule pour ce compte (droits '{share['permission']}')") if share["quotas"] and share["used"] >= share["quotas"] * 1073741824: raise RowError(f"partage '{share['name']}' plein (quota {share['quotas']} Go atteint)") return share["name"] def list_shares(shares): print(f"{'PARTAGE':<28} {'DROITS':<7} {'QUOTA':>12} ORGANISATION") for share in sorted(shares, key=lambda s: (s["organisation"].lower(), s["name"].lower())): quota = (f"{share['used'] / 1073741824:.1f}/{share['quotas']} Go" if share["quotas"] else "illimite") print(f"{share['name'][:28]:<28} {share['permission']:<7} {quota:>12} {share['organisation'] or '-'}") print(f"{len(shares)} partage(s). Colonne partage = 1re colonne ; colonne organisation = derniere.") print("Seuls les partages avec le droit w (ecriture) acceptent des paquets.") def check_quotas(packages, shares): """Refuse le lot si les fichiers depassent le quota restant d'un partage.""" errors, need = [], {} for _, _, pkg in packages: need[pkg["share"]] = need.get(pkg["share"], 0) + sum(os.path.getsize(f) for f in pkg["files"]) for share in shares: if share["name"] in need: debug(f"partage {share['name']} : {need[share['name']] // 1024} Ko a ajouter") if share["quotas"]: free = share["quotas"] * 1073741824 - share["used"] if need[share["name"]] > free: errors.append(f"partage '{share['name']}' : {human(need[share['name']])} a deposer, " f"{human(max(free, 0))} libres sur le quota de {share['quotas']} Go") return errors # ----------------------------------------------------------------------------- serveur def existing_packages(): found = {} for conf in glob.glob(os.path.join(PACKAGES, "*", "conf.json")): try: with open(conf, encoding="utf-8") as fh: data = json.load(fh) found[os.path.basename(os.path.dirname(conf))] = data except (OSError, ValueError): continue return found def check_existing(pkg, existing, adopt, warnings): """Comparaison avec les paquets deja presents dans Medulla.""" key = (pkg["name"].lower(), pkg["version"], pkg["targetos"]) if pkg["uuid"]: conf = existing.get(pkg["uuid"]) if conf is None: raise RowError(f"uuid {pkg['uuid']} absent de Medulla (paquet supprime ?) : vider la cellule " "pour le recreer") current_share = conf.get("localisation_server") or "global" if current_share != pkg["share"]: raise RowError(f"le paquet {pkg['uuid']} est dans le partage '{current_share}', le CSV demande " f"'{pkg['share']}' : changement de partage a faire dans la console") if str(conf.get("targetos", "")) and str(conf.get("targetos")) != pkg["targetos"]: raise RowError(f"le paquet {pkg['uuid']} est pour '{conf.get('targetos')}', le CSV indique " f"'{pkg['targetos']}' : creer un nouveau paquet (cellule uuid vide)") if str(conf.get("name", "")).lower() != pkg["name"].lower(): warnings.append(f"le paquet {pkg['uuid']} s'appelle '{conf.get('name')}' dans Medulla : " f"il sera renomme '{pkg['name']}' (verifier que l'uuid est le bon)") return twins = [(u, c) for u, c in existing.items() if (str(c.get("name", "")).lower(), str(c.get("version", "")), str(c.get("targetos", ""))) == key] same_share = [u for u, c in twins if (c.get("localisation_server") or "global") == pkg["share"]] if same_share: if adopt: pkg["uuid"] = same_share[0] warnings.append(f"deja present ({same_share[0]}) : sera mis a jour (--adopter-existants)") return True raise RowError(f"existe deja dans le partage {pkg['share']} (uuid {same_share[0]}) : copier cet uuid " "dans le CSV, ou relancer avec --adopter-existants") for u, c in twins: warnings.append(f"un paquet identique existe dans le partage '{c.get('localisation_server') or 'global'}' " f"({u}) : un second sera cree dans '{pkg['share']}'") def stage_files(files, targetos="win"): os.makedirs(TMPDIR, exist_ok=True) random_dir = "import_csv_" + uuidlib.uuid4().hex target = os.path.join(TMPDIR, random_dir) os.mkdir(target, 0o755) for path in files: dest = os.path.join(target, os.path.basename(path)) shutil.copy2(path, dest) fixed = script_fix(path, targetos) if fixed is not None: with open(dest, "wb") as fh: fh.write(fixed) debug(f"copie {path} ({os.path.getsize(path)} octets){' corrigee' if fixed else ''} -> {target}/") return random_dir def script_fix(path, targetos): """Corrections sures appliquees a la COPIE du script (l'original n'est pas modifie) : - .sh / .py pour Linux / Mac avec fins de ligne Windows : converties (sinon bash echoue) ; - .ps1 avec accents en UTF-8 sans BOM : BOM ajoute (sinon PowerShell 5.1 lit mal les accents).""" ext = os.path.splitext(path)[1].lower() if ext not in (".sh", ".py", ".ps1") or os.path.getsize(path) > 10 * 1048576: return None data = open(path, "rb").read() if ext in (".sh", ".py") and targetos != "win" and b"\r\n" in data: return data.replace(b"\r\n", b"\n") if ext == ".ps1" and not data.startswith(b"\xef\xbb\xbf"): try: text = data.decode("utf-8") except UnicodeDecodeError: return None if any(ord(c) > 127 for c in text): return b"\xef\xbb\xbf" + data return None def clear_payload(pkg_uuid): folder = os.path.realpath(os.path.join(PACKAGES, pkg_uuid)) for entry in os.listdir(folder): if entry in RESERVED: continue full = os.path.join(folder, entry) debug(f"retrait de l'ancien fichier {full}") shutil.rmtree(full) if os.path.isdir(full) and not os.path.islink(full) else os.remove(full) def publish(proxy, pkg, existing, user, now, on_uuid): pkg_uuid = pkg["uuid"] update = bool(pkg_uuid) and pkg_uuid in existing debug(f"{'mise a jour de ' + pkg_uuid if update else 'creation'} : {pkg['name']} {pkg['version']} " f"-> partage {pkg['share']}") creation_date = (existing.get(pkg_uuid, {}).get("creation_date") or now) if update else now creator = (existing.get(pkg_uuid, {}).get("creator") or user) if update else user detail = package_detail(pkg, pkg_uuid or "", "edition" if update else "creation", user, now, creation_date) detail["creator"] = creator ret = call(proxy, "pkgs.putPackageDetail", detail, True) if not ret or not ret[0]: raise RowError(f"putPackageDetail refuse : {ret}") pkg_uuid, folder = ret[1], ret[2] if not update: on_uuid(pkg_uuid) # uuid ecrit tout de suite : une relance terminera ce paquet au lieu d'en creer un autre if update: clear_payload(pkg_uuid) random_dir = stage_files(pkg["files"], pkg["targetos"]) ok = call(proxy, "pkgs.associatePackages", pkg_uuid, [random_dir], 1) if not ok or not ok[0]: raise RowError(f"copie des fichiers en echec : {ok}") # Workflow ecrit apres les fichiers : le paquet de deploiement rapide inclut alors les fichiers. call(proxy, "pkgs.save_xmpp_json", folder, json.dumps( xmppdeploy(pkg, pkg_uuid, now, creator, creation_date), ensure_ascii=False)) # Second passage une fois les fichiers en place : hash et synchronisation relais a jour. detail.update(id=pkg_uuid, mode="edition") call(proxy, "pkgs.putPackageDetail", detail, True) call(proxy, "pkgs.update_package_size", pkg_uuid) call(proxy, "pkgs.chown", pkg_uuid) return pkg_uuid, update # ----------------------------------------------------------------------------- sauvegarde def db_settings(): conf = configparser.RawConfigParser() if not conf.read(PKGS_CONF): raise SystemExit(f"ERREUR sauvegarde : {PKGS_CONF[0]} introuvable, impossible de lire l'acces a la base") get = lambda key, default: conf.get("database", key, fallback=default) password = get("dbpasswd", "") match = re.fullmatch(r"\{(\w+)\}(.+)", password) if match: if match[1].lower() != "base64": raise SystemExit(f"ERREUR sauvegarde : mot de passe chiffre '{match[1]}' non gere dans pkgs.ini") password = base64.b64decode(match[2]).decode("utf-8") return {"host": get("dbhost", "localhost"), "port": get("dbport", "3306"), "user": get("dbuser", "mmc"), "password": password, "name": get("dbname", "pkgs")} def backup(csv_path, updated_uuids, folder_root): """Etape 3 : sauvegarde complete avant la premiere modification. Tout echec arrete le script.""" stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") os.makedirs(folder_root, mode=0o700, exist_ok=True) folder, n = os.path.join(folder_root, stamp), 1 while os.path.exists(folder): n += 1 folder = os.path.join(folder_root, f"{stamp}_{n}") os.mkdir(folder, 0o700) db = db_settings() dump = shutil.which("mysqldump") or shutil.which("mariadb-dump") if not dump: raise SystemExit("ERREUR sauvegarde : mysqldump introuvable, aucune modification faite") sql = os.path.join(folder, f"{db['name']}.sql.gz") with tempfile.NamedTemporaryFile("w", prefix="medulla_dump_", suffix=".cnf", delete=False) as cnf: os.chmod(cnf.name, 0o600) # mot de passe hors ligne de commande (invisible dans ps) cnf.write(f"[client]\nuser={db['user']}\npassword={db['password']}\n" f"host={db['host']}\nport={db['port']}\n") try: with gzip.open(sql, "wb") as out: debug(f"{dump} --single-transaction --routines --triggers --databases {db['name']} " f"(compte {db['user']}@{db['host']}:{db['port']}, mot de passe via fichier 0600) -> {sql}") proc = subprocess.Popen([dump, f"--defaults-extra-file={cnf.name}", "--single-transaction", "--routines", "--triggers", "--databases", db["name"]], stdout=subprocess.PIPE, stderr=subprocess.PIPE) for chunk in iter(lambda: proc.stdout.read(1 << 20), b""): out.write(chunk) err = proc.stderr.read().decode("utf-8", "replace").strip() code = proc.wait() finally: os.remove(cnf.name) if code != 0: raise SystemExit(f"ERREUR sauvegarde de la base {db['name']} (code {code}) : {err}\n" "Aucune modification faite.") with gzip.open(sql, "rb") as fh: fh.seek(max(0, _gzip_size(sql) - 4096)) if b"Dump completed" not in fh.read(): raise SystemExit(f"ERREUR sauvegarde : {sql} incomplet. Aucune modification faite.") archive = None if updated_uuids: archive = os.path.join(folder, "paquets_avant_mise_a_jour.tar.gz") with tarfile.open(archive, "w:gz") as tar: for pkg_uuid in updated_uuids: real = os.path.realpath(os.path.join(PACKAGES, pkg_uuid)) tar.add(real, arcname=os.path.relpath(real, "/")) shutil.copy2(csv_path, os.path.join(folder, os.path.basename(csv_path))) restore = os.path.join(folder, "RESTAURATION.txt") with open(restore, "w", encoding="utf-8") as fh: fh.write(f"Sauvegarde du {stamp} avant import de {csv_path}\n\n" f"1. Arreter mmc-agent : systemctl stop mmc-agent\n" f"2. Restaurer la base : zcat {sql} | mysql -u {db['user']} -p -h {db['host']} -P {db['port']}\n") if archive: fh.write(f"3. Restaurer les fichiers des paquets mis a jour :\n" f" tar -xzf {archive} -C /\n") fh.write("4. Redemarrer mmc-agent : systemctl start mmc-agent\n\n" "Les paquets CREES par l'import restent sur le disque : les supprimer depuis la console\n" "si besoin (leur uuid est dans le rapport et dans le CSV).\n") size = os.path.getsize(sql) // 1024 print(f"Sauvegarde OK : {folder}") print(f" base {db['name']} : {os.path.basename(sql)} ({size} Ko)" + (f" ; fichiers de {len(updated_uuids)} paquet(s) mis a jour" if archive else "") + " ; copie du CSV ; RESTAURATION.txt") return folder def _gzip_size(path): with open(path, "rb") as fh: fh.seek(-4, os.SEEK_END) return int.from_bytes(fh.read(4), "little") def db_test(db): """Verifie que la sauvegarde sera possible : outil present et identifiants acceptes (dump d'une petite table).""" dump = shutil.which("mysqldump") or shutil.which("mariadb-dump") if not dump: return "mysqldump introuvable (paquet mariadb-client / default-mysql-client)" with tempfile.NamedTemporaryFile("w", prefix="medulla_dump_", suffix=".cnf", delete=False) as cnf: os.chmod(cnf.name, 0o600) cnf.write(f"[client]\nuser={db['user']}\npassword={db['password']}\nhost={db['host']}\nport={db['port']}\n") try: proc = subprocess.run([dump, f"--defaults-extra-file={cnf.name}", "--no-data", db["name"], "packages"], capture_output=True, timeout=60) except (OSError, subprocess.TimeoutExpired) as exc: return f"test de sauvegarde impossible : {exc}" finally: os.remove(cnf.name) debug(f"test sauvegarde : {dump} --no-data {db['name']} packages -> code {proc.returncode}") if proc.returncode != 0: return f"acces a la base {db['name']} refuse pour la sauvegarde : " + \ proc.stderr.decode("utf-8", "replace").strip() return None def human(size): for unit in ("octets", "Ko", "Mo", "Go"): if size < 1024 or unit == "Go": return f"{size:.0f} {unit}" if unit == "octets" else f"{size:.1f} {unit}" size /= 1024 def free_space(path): while not os.path.exists(path): path = os.path.dirname(path) st = os.statvfs(path) return st.f_bavail * st.f_frsize def folder_size(path): total = 0 for root, _, files in os.walk(os.path.realpath(path)): for name in files: try: total += os.path.getsize(os.path.join(root, name)) except OSError: pass return total def preflight(csv_path, packages, backup_root): """Controles du serveur, une fois le contenu valide : tout ce qui ferait echouer l'import en route.""" errors = [] try: db = db_settings() problem = db_test(db) if problem: errors.append(f"sauvegarde : {problem}") except SystemExit as exc: errors.append(str(exc).replace("ERREUR sauvegarde : ", "sauvegarde : ")) if not os.access(os.path.dirname(csv_path), os.W_OK) or \ (os.path.exists(csv_path) and not os.access(csv_path, os.W_OK)): errors.append(f"CSV non modifiable ({csv_path}) : le script doit pouvoir y ecrire les uuid crees") for folder in (TMPDIR, backup_root): parent = folder while not os.path.exists(parent): parent = os.path.dirname(parent) if not os.access(parent, os.W_OK): errors.append(f"dossier non accessible en ecriture : {parent}") payload = sum(os.path.getsize(f) for _, _, pkg in packages for f in pkg["files"]) updates = [pkg["uuid"] for _, _, pkg in packages if pkg["uuid"]] backup_need = sum(folder_size(os.path.join(PACKAGES, u)) for u in updates) + MARGIN need_pkgs = 2 * payload + MARGIN # copie temporaire + copie dans le partage for label, path, need in (("partages", PACKAGES, need_pkgs), ("sauvegarde", backup_root, backup_need)): free = free_space(path) debug(f"disque {label} ({path}) : {free // 1048576} Mo libres, {need // 1048576} Mo necessaires") if free < need: errors.append(f"espace disque insuffisant pour {label} ({path}) : {human(free)} libres, " f"{human(need)} necessaires") return errors, payload def summary(packages, warnings, payload, csv_path, launching): creations = [p for _, _, p in packages if not p["uuid"]] updates = [p for _, _, p in packages if p["uuid"]] by_share = {} for _, _, pkg in packages: by_share[pkg["share"]] = by_share.get(pkg["share"], 0) + 1 print(f"Controle complet OK : {len(creations)} creation(s), {len(updates)} mise(s) a jour, " f"{human(payload)} de fichiers.") print(" par partage : " + ", ".join(f"{k} ({v})" for k, v in sorted(by_share.items()))) for number, _, pkg in packages: action = f"MAJ {pkg['uuid']}" if pkg["uuid"] else "creation" print(f" ligne {number} : {pkg['name']} {pkg['version']} [{pkg['targetos']}, {pkg['share']}] {action}") if VERBOSE or not launching: for section, command, timeout in pkg["sections"]: print(f" {section:<16}: {command} (timeout {timeout}s)") print(f" fichiers : {', '.join(os.path.basename(f) for f in pkg['files'])}") if warnings: print(f" {len(warnings)} avertissement(s) (non bloquants) :") for warning in warnings: print(f" ! {warning}") if not launching: print(f"Pret. Pour lancer l'import : python3 {os.path.basename(sys.argv[0])} {csv_path}") def ask(question): try: return input(f"{question} [o/N] ").strip().lower() in YES except EOFError: return False # ----------------------------------------------------------------------------- principal def write_example(path): if os.path.exists(path): raise SystemExit(f"ERREUR : {path} existe deja") with open(path, "w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=COLUMNS) writer.writeheader() for row in EXAMPLE: writer.writerow({c: row.get(c, "") for c in COLUMNS}) print(f"Modele ecrit : {path}") return 0 def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("csv", nargs="?", help="CSV des paquets") parser.add_argument("--verifier", action="store_true", help="faire TOUS les controles (format, fichiers, Medulla, sauvegarde, disque) " "sans rien modifier") parser.add_argument("--hors-ligne", action="store_true", help="avec --verifier : sans connexion a Medulla (format et fichiers seulement), " "utilisable hors du serveur") parser.add_argument("--adopter-existants", action="store_true", help="un paquet deja present (meme nom, version, os et partage) sans uuid dans le CSV " "est mis a jour au lieu d'etre refuse ; son uuid est ecrit dans le CSV") parser.add_argument("--oui", action="store_true", help="ne pas demander de confirmation avant l'import (lancement automatise)") parser.add_argument("--utilisateur", default="root", help="compte MMC (defaut root)") parser.add_argument("--modele", metavar="FICHIER", help="ecrire un CSV d'exemple et quitter") parser.add_argument("-v", "--verbeux", action="store_true", help="afficher le detail de chaque etape et de chaque appel a Medulla") parser.add_argument("--lister", action="store_true", help="lister les organisations et partages accessibles puis quitter") parser.add_argument("--dossier-sauvegarde", default=BACKUP_ROOT, help=f"ou ranger la sauvegarde faite avant import (defaut {BACKUP_ROOT})") args = parser.parse_args(argv) global VERBOSE VERBOSE = args.verbeux if args.lister: password = os.environ.get("MMC_PASSWORD") or getpass.getpass(f"Mot de passe MMC de {args.utilisateur} : ") list_shares(load_shares(connect(args.utilisateur, password), args.utilisateur)) return 0 if args.modele: return write_example(args.modele) if not args.csv: parser.error("CSV attendu") csv_path = os.path.abspath(args.csv) base = os.path.dirname(csv_path) rows, delimiter, headers = check_format(csv_path) proxy = shares = None existing = {} user = args.utilisateur offline = args.verifier and args.hors_ligne if args.hors_ligne and not args.verifier: parser.error("--hors-ligne s'utilise avec --verifier") if not offline: problems = [] if os.geteuid() != 0: problems.append("lancer en root sur le serveur Medulla (ou --verifier --hors-ligne ailleurs)") if not os.path.isdir(PACKAGES): problems.append(f"{PACKAGES} absent : ce n'est pas un serveur Medulla") if problems: raise SystemExit("ERREUR : " + " ; ".join(problems)) password = os.environ.get("MMC_PASSWORD") or getpass.getpass(f"Mot de passe MMC de {user} : ") proxy = connect(user, password) shares = load_shares(proxy, user) existing = existing_packages() debug(f"{len(existing)} paquet(s) deja presents dans {PACKAGES}") # 2. Controle du contenu : fichiers, organisations, partages, droits, quotas, uuid, doublons. errors, warnings, packages, uuids, keys = [], [], [], {}, {} adopted = [] for number, row in rows: row_warnings = [] try: pkg = validate(row, base, shares, row_warnings) if pkg["uuid"]: if pkg["uuid"] in uuids: raise RowError(f"uuid deja utilise ligne {uuids[pkg['uuid']]}") uuids[pkg["uuid"]] = number key = (pkg["name"].lower(), pkg["version"], pkg["targetos"]) if key in keys: raise RowError(f"meme nom/version/os que la ligne {keys[key]}") keys[key] = number if not offline and check_existing(pkg, existing, args.adopter_existants, row_warnings): adopted.append((number, row, pkg)) debug(f"ligne {number} OK : {pkg['name']} {pkg['version']} [{pkg['targetos']}] partage " f"{pkg['share']} ; {len(pkg['files'])} fichier(s) ; " + " ; ".join(f"{s}={c}" for s, c, _ in pkg["sections"])) packages.append((number, row, pkg)) except RowError as exc: errors.append(f"ligne {number} ({row['nom'] or '?'}) : {exc}") warnings += [f"ligne {number} ({row['nom'] or '?'}) : {w}" for w in row_warnings] if shares is not None: errors += check_quotas(packages, shares) # 3. Controles du serveur : sauvegarde possible, ecriture, espace disque. payload = sum(os.path.getsize(f) for _, _, pkg in packages for f in pkg["files"]) if not offline and not errors: server_errors, payload = preflight(csv_path, packages, args.dossier_sauvegarde) errors += server_errors if errors: print(f"CONTROLE REFUSE : {len(errors)} probleme(s), rien n'a ete modifie.") print("\n".join(f" - {e}" for e in errors)) if any("--adopter-existants" in e for e in errors): print(" Astuce : relancer avec --adopter-existants pour mettre a jour ces paquets " "et ecrire leur uuid dans le CSV.") for warning in warnings: print(f" ! {warning}") return 1 if args.verifier: summary(packages, warnings, payload, args.csv, launching=False) if offline: print("ATTENTION hors ligne : organisations, partages, droits, quotas, uuid, sauvegarde et disque " "n'ont pas ete controles. Relancer --verifier sur le serveur.") return 0 summary(packages, warnings, payload, args.csv, launching=True) if not args.oui and not ask("Lancer l'import (sauvegarde puis creation / mise a jour) ?"): print("Abandon, rien n'a ete modifie.") return 1 # 3. Sauvegarde de la base pkgs et des paquets mis a jour, avant toute modification. save_dir = backup(csv_path, [p["uuid"] for _, _, p in packages if p["uuid"]], args.dossier_sauvegarde) for number, row, pkg in adopted: row["uuid"] = pkg["uuid"] if adopted: write_csv(csv_path, delimiter, headers, rows) print(f"{len(adopted)} uuid de paquets existants ecrits dans le CSV.") # 4. Publication paquet par paquet ; l'uuid cree est reecrit dans le CSV a chaque succes. now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") report = [f"Import CSV Medulla {now} - {csv_path} - compte {user}", f"Sauvegarde : {save_dir}"] created = updated = failed = 0 for number, row, pkg in packages: label = f"ligne {number} : {pkg['name']} {pkg['version']}" def remember(new_uuid, row=row): row["uuid"] = new_uuid write_csv(csv_path, delimiter, headers, rows) existing[new_uuid] = {"creation_date": now, "creator": user} try: pkg_uuid, update = publish(proxy, pkg, existing, user, now, remember) except (RowError, xmlrpc.client.Fault, OSError) as exc: failed += 1 report.append(f"ECHEC {label} : {exc}" + (f" (uuid {row['uuid']} note dans le CSV : relancer pour terminer)" if row["uuid"] else "")) print(report[-1]) continue if update: updated += 1 else: created += 1 report.append(f"{'MAJ ' if update else 'CREE '} {label} -> {pkg_uuid} [{pkg['share']}]") print(report[-1]) report.append(f"Total : {created} cree(s), {updated} mis a jour, {failed} en echec") print(report[-1]) log = os.path.splitext(csv_path)[0] + "_rapport.txt" with open(log, "a", encoding="utf-8") as fh: fh.write("\n".join(report) + "\n\n") print(f"Rapport : {log}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())