1414 lines
58 KiB
Python
1414 lines
58 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
|||
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|||
|
|
|
|||
|
|
"""Le menu « Transform data » : décrire un fichier externe, puis le copier.
|
|||
|
|
|
|||
|
|
Le moteur vit dans `script/data/external_file.py` et tourne en
|
|||
|
|
SOUS-PROCESSUS, sous l'interpréteur que `transform_setup.engine_python()`
|
|||
|
|
désigne — le venv dédié pour Excel et Access, celui du CLI pour les
|
|||
|
|
formats en pur stdlib. Même séparation que `lib_analyse` appelant `psql`
|
|||
|
|
plutôt que d'importer psycopg2 : le CLI n'a pas à porter les
|
|||
|
|
bibliothèques de lecture.
|
|||
|
|
|
|||
|
|
L'ordre du dialogue est la règle, pas une préférence : rapport, questions,
|
|||
|
|
puis MARCHE À BLANC, et l'écriture seulement après. C'est la convention du
|
|||
|
|
dépôt pour l'action de menu qui écrit — « une question posée avant de
|
|||
|
|
savoir ce qui sera touché n'est pas un consentement : c'est un pari ».
|
|||
|
|
|
|||
|
|
Ce que ce mixin EMPRUNTE à `TODO`
|
|||
|
|
---------------------------------
|
|||
|
|
`_is_yes`, `_is_yes_default_yes`, `fill_help_info`, `execute`,
|
|||
|
|
`db_manager`, `_dir_path`, et — pour l'anonymisation d'une base —
|
|||
|
|
`_monitoring_restore` et `_monitoring_write_flow`. La liste est écrite
|
|||
|
|
ici parce qu'un emprunt tacite se casse en silence : une méthode
|
|||
|
|
renommée dans `todo.py` ne lève qu'à l'exécution de l'entrée, et un
|
|||
|
|
bouchon de test qui ne la fournit pas passe au vert sans rien éprouver.
|
|||
|
|
|
|||
|
|
Les deux emprunts de `monitoring` sont VOULUS : `anonymize.py` sait déjà
|
|||
|
|
remplacer les données d'une base, et `_monitoring_write_flow` sait déjà
|
|||
|
|
montrer puis demander avec le nom retapé. Les réécrire ici ferait deux
|
|||
|
|
dialogues à tenir d'accord pour la même chose.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import datetime
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import shlex
|
|||
|
|
import shutil
|
|||
|
|
import subprocess
|
|||
|
|
|
|||
|
|
import click
|
|||
|
|
|
|||
|
|
from script.todo import transform_setup
|
|||
|
|
from script.todo.todo_i18n import t
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
from script.todo import todo_file_browser
|
|||
|
|
except Exception: # pragma: no cover - urwid peut manquer
|
|||
|
|
todo_file_browser = None
|
|||
|
|
|
|||
|
|
MOTEUR = os.path.join("script", "data", "external_file.py")
|
|||
|
|
SORTIE_PAR_DEFAUT = os.path.join("private", "transform")
|
|||
|
|
|
|||
|
|
# Ce que chaque format peut porter en sortie. Les formats en lecture seule
|
|||
|
|
# n'ont pas de graveur : leur copie repart en .xlsx.
|
|||
|
|
EXTENSION_PAR_FORMAT = {
|
|||
|
|
"xlsx": ".xlsx",
|
|||
|
|
"xls": ".xlsx",
|
|||
|
|
"xlsb": ".xlsx",
|
|||
|
|
"access": ".xlsx",
|
|||
|
|
"csv": ".csv",
|
|||
|
|
"xml": ".xml",
|
|||
|
|
"json": ".json",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
CIBLES = ("xlsx", "csv", "json", "xml")
|
|||
|
|
|
|||
|
|
# Ce que le chemin COURT prend, quand l'opérateur refuse les options
|
|||
|
|
# avancées. Sept questions dont chacune a un défaut évident : les répondre
|
|||
|
|
# une par une pour arriver au même endroit fait passer les invites par
|
|||
|
|
# réflexe, et une invite qu'on passe par réflexe ne consent à rien.
|
|||
|
|
#
|
|||
|
|
# `calibre_chiffres` est le seul qui ne suit pas le défaut de son invite :
|
|||
|
|
# le chemin court GARDE les largeurs. L'aperçu dit alors que l'étendue
|
|||
|
|
# mesurée ne borne plus les nombres, ce qui est ce qui rend le raccourci
|
|||
|
|
# honnête — un taux ou une année peut sortir de sa plage.
|
|||
|
|
DEFAUTS_RAPIDES = {
|
|||
|
|
"feuilles": [],
|
|||
|
|
"colonnes_intactes": [],
|
|||
|
|
"nombres": True,
|
|||
|
|
"calibre_chiffres": True,
|
|||
|
|
"texte": True,
|
|||
|
|
"entetes": False,
|
|||
|
|
"graine": "",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# La famille d'une valeur laissée intacte, telle qu'on la nomme à l'écran.
|
|||
|
|
# Une date et un booléen sont de VRAIES valeurs du client : les taire dans
|
|||
|
|
# le bilan laissait croire à une copie entièrement remplacée.
|
|||
|
|
_LIBELLE_INTACTES = {
|
|||
|
|
"formule": "formula(s)",
|
|||
|
|
"date": "date(s)",
|
|||
|
|
"booleen": "boolean(s)",
|
|||
|
|
"erreur": "error value(s)",
|
|||
|
|
"binaire": "binary value(s)",
|
|||
|
|
"texte": "text",
|
|||
|
|
"nombre": "numeric",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TransformMenuMixin:
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Menu
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def prompt_execute_transform(self):
|
|||
|
|
print(f"🪄 {t('Transform your data: read, describe, then copy.')}")
|
|||
|
|
choices = [
|
|||
|
|
# « Source » et non « Source file » : l'entrée couvre
|
|||
|
|
# désormais un fichier ET une base Odoo, et un libellé qui
|
|||
|
|
# dit « fichier » ferait chercher ailleurs l'anonymiseur de
|
|||
|
|
# base.
|
|||
|
|
{"section": t("Source")},
|
|||
|
|
{"prompt_description": t("Open a file and read its report")},
|
|||
|
|
{
|
|||
|
|
"prompt_description": t(
|
|||
|
|
"Anonymise an Odoo database or a backup"
|
|||
|
|
)
|
|||
|
|
},
|
|||
|
|
{"section": t("Environment")},
|
|||
|
|
{"prompt_description": t("Install the reading environment")},
|
|||
|
|
{"prompt_description": t("What can this machine read?")},
|
|||
|
|
{"prompt_description": t("Copies produced")},
|
|||
|
|
{"prompt_description": t("Databases produced")},
|
|||
|
|
]
|
|||
|
|
help_info = self.fill_help_info(choices)
|
|||
|
|
|
|||
|
|
while True:
|
|||
|
|
status = click.prompt(help_info)
|
|||
|
|
print()
|
|||
|
|
if status == "0":
|
|||
|
|
return False
|
|||
|
|
elif status == "1":
|
|||
|
|
self._transform_open_and_report()
|
|||
|
|
elif status == "2":
|
|||
|
|
self._transform_anonymise_base()
|
|||
|
|
elif status == "3":
|
|||
|
|
self._transform_install_env()
|
|||
|
|
elif status == "4":
|
|||
|
|
self._transform_capabilities()
|
|||
|
|
elif status == "5":
|
|||
|
|
self._transform_copies()
|
|||
|
|
elif status == "6":
|
|||
|
|
self._transform_bases_produites()
|
|||
|
|
else:
|
|||
|
|
print(t("Command not found !"))
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Le moteur, en sous-processus
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_run(self, arguments, fmt=None):
|
|||
|
|
"""Lancer le moteur. Rend l'objet JSON, ou None sur erreur.
|
|||
|
|
|
|||
|
|
stdout ne porte qu'un objet JSON ; stderr porte la progression et
|
|||
|
|
les traces. Un stdout vide ou inanalysable est traité comme une
|
|||
|
|
erreur — et non relayé en exception — pour que le menu affiche les
|
|||
|
|
dernières lignes de stderr plutôt que de tomber à son tour.
|
|||
|
|
"""
|
|||
|
|
interpreteur = transform_setup.engine_python(fmt)
|
|||
|
|
racine = transform_setup.racine()
|
|||
|
|
environnement = dict(os.environ)
|
|||
|
|
environnement["PYTHONPATH"] = racine
|
|||
|
|
commande = [interpreteur, os.path.join(racine, MOTEUR)] + list(
|
|||
|
|
arguments
|
|||
|
|
)
|
|||
|
|
try:
|
|||
|
|
acheve = subprocess.run(
|
|||
|
|
commande,
|
|||
|
|
capture_output=True,
|
|||
|
|
text=True,
|
|||
|
|
cwd=racine,
|
|||
|
|
env=environnement,
|
|||
|
|
)
|
|||
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|||
|
|
print(f"❌ {exc}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
for ligne in (acheve.stderr or "").splitlines():
|
|||
|
|
if ligne.startswith("# "):
|
|||
|
|
print(f" {ligne[2:]}")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
resultat = json.loads(acheve.stdout or "")
|
|||
|
|
except (ValueError, TypeError):
|
|||
|
|
print(f"❌ {t('Command not found !')}")
|
|||
|
|
print(f" {shlex.join(commande)}")
|
|||
|
|
for ligne in (acheve.stderr or "").splitlines()[-20:]:
|
|||
|
|
print(f" {ligne}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
if "erreur" in resultat:
|
|||
|
|
print(f"❌ {t(resultat['erreur'])}{resultat.get('detail', '')}")
|
|||
|
|
if resultat.get("conseil"):
|
|||
|
|
print(f" → {t(resultat['conseil'])}")
|
|||
|
|
return None
|
|||
|
|
return resultat
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Le fichier d'entrée
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _on_dir_selected(self, chemin):
|
|||
|
|
"""Le chemin choisi dans le navigateur, mis de côté.
|
|||
|
|
|
|||
|
|
Ce mixin fournit SON rappel au lieu d'emprunter celui de sa classe
|
|||
|
|
d'accueil : `TODO` nomme le sien `on_dir_selected` et pose
|
|||
|
|
`dir_path`, tandis que le préfixé vit sur le gestionnaire de bases,
|
|||
|
|
qui n'est pas un mixin mais un objet à part. Emprunter faisait
|
|||
|
|
lever `AttributeError` à l'ouverture du navigateur — donc au
|
|||
|
|
premier usage de l'entrée, et sur la première ligne du menu.
|
|||
|
|
|
|||
|
|
Le navigateur sort de sa boucle lui-même après avoir appelé le
|
|||
|
|
rappel : celui-ci n'a que le chemin à retenir.
|
|||
|
|
"""
|
|||
|
|
self._dir_path = chemin
|
|||
|
|
|
|||
|
|
def _transform_select_file(self):
|
|||
|
|
"""Le parcours d'abord, la saisie en repli.
|
|||
|
|
|
|||
|
|
La garde d'import est celle de `database_manager` : urwid peut
|
|||
|
|
manquer, et sans elle « p » lèverait sur None.
|
|||
|
|
"""
|
|||
|
|
depart = os.path.join(os.getcwd(), SORTIE_PAR_DEFAUT)
|
|||
|
|
if not os.path.isdir(depart):
|
|||
|
|
depart = os.getcwd()
|
|||
|
|
if todo_file_browser is not None:
|
|||
|
|
self._dir_path = ""
|
|||
|
|
navigateur = todo_file_browser.FileBrowser(
|
|||
|
|
depart, self._on_dir_selected
|
|||
|
|
)
|
|||
|
|
navigateur.run_main_frame()
|
|||
|
|
if self._dir_path and os.path.isfile(self._dir_path):
|
|||
|
|
print(self._dir_path)
|
|||
|
|
return self._dir_path
|
|||
|
|
reponse = input(
|
|||
|
|
t("Enter to accept, p to browse, or type a path: ")
|
|||
|
|
).strip()
|
|||
|
|
if not reponse or reponse == "0":
|
|||
|
|
return None
|
|||
|
|
chemin = os.path.expanduser(reponse)
|
|||
|
|
if not os.path.isfile(chemin):
|
|||
|
|
print(f"❌ {t('Not an ordinary file.')} {chemin}")
|
|||
|
|
return None
|
|||
|
|
return chemin
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Le rapport
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_render_report(self, rapport):
|
|||
|
|
lignes = []
|
|||
|
|
nom = os.path.basename(rapport.get("chemin", ""))
|
|||
|
|
taille = rapport.get("taille", 0)
|
|||
|
|
lignes.append(f"🪄 {nom} — {taille} o — {rapport.get('format', '')}")
|
|||
|
|
if rapport.get("divergence"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" ⚠ {t('Contents do not match the extension: read as ')}"
|
|||
|
|
f"{rapport.get('format', '')}"
|
|||
|
|
)
|
|||
|
|
if rapport.get("encodage"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('Encoding detected by ')}"
|
|||
|
|
f"{t(rapport['encodage_source'])}: {rapport['encodage']}"
|
|||
|
|
)
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('Delimiter detected by ')}"
|
|||
|
|
f"{t(rapport['delimiteur_source'])}:"
|
|||
|
|
f" {rapport['delimiteur']!r}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
feuilles = rapport.get("feuilles") or []
|
|||
|
|
if feuilles:
|
|||
|
|
lignes.append("")
|
|||
|
|
lignes.append(f" {len(feuilles)} {t('sheet(s)')}")
|
|||
|
|
for feuille in feuilles:
|
|||
|
|
marque = f" ({t('hidden')})" if feuille.get("masquee") else ""
|
|||
|
|
lignes.append(
|
|||
|
|
f" {feuille['nom']}{marque}"
|
|||
|
|
f" {feuille['lignes']} × {feuille['colonnes_n']}"
|
|||
|
|
f" {feuille['formules']} {t('formula(s)')}"
|
|||
|
|
)
|
|||
|
|
if feuille.get("formules_litteral"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" ⚠ {feuille['formules_litteral']}"
|
|||
|
|
f" {t('formula(s) holding a text literal')}"
|
|||
|
|
)
|
|||
|
|
for colonne in feuille.get("colonnes", [])[:12]:
|
|||
|
|
borne = ""
|
|||
|
|
if colonne.get("min") is not None:
|
|||
|
|
borne = f" {colonne['min']}..{colonne['max']}"
|
|||
|
|
plancher = " ⛔" if colonne.get("plancher") else ""
|
|||
|
|
lignes.append(
|
|||
|
|
f" {colonne['index']:>3} "
|
|||
|
|
f"{str(colonne['etiquette'] or ''):<22}"
|
|||
|
|
f"{colonne['type']:<9}"
|
|||
|
|
f"{colonne['remplies']:>6} {t('row(s)')}"
|
|||
|
|
f"{colonne['distinctes']:>6}"
|
|||
|
|
f" {t('distinct value(s)')}{borne}{plancher}"
|
|||
|
|
)
|
|||
|
|
reste = len(feuille.get("colonnes", [])) - 12
|
|||
|
|
if reste > 0:
|
|||
|
|
lignes.append(f" … {reste}")
|
|||
|
|
# Les colonnes au-delà du plafond des exemples n'en portent
|
|||
|
|
# pas : une colonne sans exemple se lit comme une colonne
|
|||
|
|
# vide, et le taire faisait juger une colonne sur rien.
|
|||
|
|
manquants = feuille.get("exemples_manquants") or 0
|
|||
|
|
if manquants:
|
|||
|
|
lignes.append(
|
|||
|
|
f" ⚠ {manquants} {t('column(s) without examples')}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
requetes = rapport.get("requetes") or []
|
|||
|
|
if requetes:
|
|||
|
|
# Nommées, non comptées : une base dont le travail vit dans
|
|||
|
|
# ses requêtes se transmet amputée, et un chiffre ne dit pas
|
|||
|
|
# ce qui manque.
|
|||
|
|
lignes.append("")
|
|||
|
|
lignes.append(f" {len(requetes)} {t('saved query(ies)')}")
|
|||
|
|
for nom in requetes[:12]:
|
|||
|
|
lignes.append(f" {nom}")
|
|||
|
|
if len(requetes) > 12:
|
|||
|
|
lignes.append(
|
|||
|
|
f" … {len(requetes) - 12} {t('more, not listed')}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
hors = rapport.get("hors_cellules") or {}
|
|||
|
|
if hors:
|
|||
|
|
lignes.append("")
|
|||
|
|
lignes.append(f" {t('Text outside cells')}")
|
|||
|
|
if hors.get("macros"):
|
|||
|
|
lignes.append(f" ⚠ {t('VBA macros present')}")
|
|||
|
|
elif rapport.get("format") == "xlsx":
|
|||
|
|
lignes.append(f" {t('No VBA macro')}")
|
|||
|
|
for cle, libelle in (
|
|||
|
|
("croises", "pivot table(s)"),
|
|||
|
|
("graphiques", "chart(s) with cached series"),
|
|||
|
|
("feuilles_graphiques", "chart sheet(s)"),
|
|||
|
|
("images", "image(s) and drawing(s)"),
|
|||
|
|
("liens_externes", "external link(s)"),
|
|||
|
|
("hyperliens", "cell hyperlink(s)"),
|
|||
|
|
("commentaires", "comment(s)"),
|
|||
|
|
("auteurs_commentaires", "comment author(s)"),
|
|||
|
|
("entetes_pieds", "header(s)/footer(s)"),
|
|||
|
|
("validations", "validation(s)"),
|
|||
|
|
("conditionnelles", "conditional format(s)"),
|
|||
|
|
):
|
|||
|
|
if hors.get(cle):
|
|||
|
|
lignes.append(f" {hors[cle]} {t(libelle)}")
|
|||
|
|
if hors.get("createur") or hors.get("modifie_par"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('Author')}: {hors.get('createur') or ''}"
|
|||
|
|
f" / {hors.get('modifie_par') or ''}"
|
|||
|
|
)
|
|||
|
|
if hors.get("cibles_liens"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('Link targets')}:"
|
|||
|
|
f" {', '.join(hors['cibles_liens'][:5])}"
|
|||
|
|
)
|
|||
|
|
if hors.get("plages_nommees"):
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('named range(s)')}:"
|
|||
|
|
f" {', '.join(hors['plages_nommees'][:8])}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
comptes = rapport.get("comptes") or {}
|
|||
|
|
lignes.append("")
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('Anonymisable')}:"
|
|||
|
|
f" {comptes.get('texte', 0)} {t('text')},"
|
|||
|
|
f" {comptes.get('nombre', 0)} {t('numeric')}"
|
|||
|
|
)
|
|||
|
|
lignes.append(
|
|||
|
|
f" {t('word(s) in the pool')}: {rapport.get('vivier', 0)}"
|
|||
|
|
)
|
|||
|
|
intacts = []
|
|||
|
|
for cle, libelle in (
|
|||
|
|
("formule", "formula(s)"),
|
|||
|
|
("date", "date(s)"),
|
|||
|
|
("booleen", "boolean(s)"),
|
|||
|
|
("erreur", "error value(s)"),
|
|||
|
|
("binaire", "binary value(s)"),
|
|||
|
|
):
|
|||
|
|
if comptes.get(cle):
|
|||
|
|
intacts.append(f"{comptes[cle]} {t(libelle)}")
|
|||
|
|
if intacts:
|
|||
|
|
lignes.append(f" {t('Left intact')}: {', '.join(intacts)}")
|
|||
|
|
for avertissement in rapport.get("avertissements", []):
|
|||
|
|
lignes.append(f" ⚠ {t(avertissement)}")
|
|||
|
|
return "\n".join(lignes)
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Les questions
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_ask(libelle):
|
|||
|
|
"""La réponse, ou None quand l'opérateur tape « 0 ».
|
|||
|
|
|
|||
|
|
Une sortie doit exister à CHAQUE question : s'apercevoir à la
|
|||
|
|
onzième qu'on a ouvert le mauvais fichier ne doit pas obliger à
|
|||
|
|
répondre à tout puis à refuser un nom de fichier.
|
|||
|
|
"""
|
|||
|
|
reponse = input(libelle).strip()
|
|||
|
|
return None if reponse == "0" else reponse
|
|||
|
|
|
|||
|
|
def _transform_ask_options(self, rapport):
|
|||
|
|
"""Les réponses, ou None si l'opérateur renonce."""
|
|||
|
|
print(f" {t('0 to cancel')}")
|
|||
|
|
options = {}
|
|||
|
|
reponse = self._transform_ask(t("Anonymise this file? (y/N): "))
|
|||
|
|
if reponse is None or not self._is_yes(reponse):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
cible = self._transform_ask(
|
|||
|
|
t("Convert at the same time? (xlsx/csv/json/xml, empty = keep): ")
|
|||
|
|
)
|
|||
|
|
if cible is None:
|
|||
|
|
return None
|
|||
|
|
cible = cible.strip().lower()
|
|||
|
|
if cible and cible not in CIBLES:
|
|||
|
|
print(f"❌ {t('Format not recognised: ')}{cible}")
|
|||
|
|
return None
|
|||
|
|
options["conversion"] = cible
|
|||
|
|
|
|||
|
|
# La table AVANT l'écran : elle dit si ce fichier appartient à
|
|||
|
|
# un lot déjà entamé, et l'écran montre alors les lignes
|
|||
|
|
# d'en-tête qu'un fichier précédent a fait corriger. La poser
|
|||
|
|
# après l'écran laissait cette mémoire hors de vue.
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Mapping table to reuse (empty = a new one): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["table_chemin"] = (
|
|||
|
|
os.path.expanduser(reponse) if reponse else ""
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# L'écran AVANT les questions de portée : c'est lui qui répond
|
|||
|
|
# aux deux qu'une invite ne sait pas poser — quelles lignes
|
|||
|
|
# nomment les colonnes, et lesquelles rester intactes. Il rend
|
|||
|
|
# `{}` pour dire « pose-moi les questions », ce qui laisse la
|
|||
|
|
# suite se dérouler comme avant.
|
|||
|
|
ecran = self._transform_ecran(rapport, options.get("table_chemin"))
|
|||
|
|
if ecran is None:
|
|||
|
|
return None
|
|||
|
|
options.update(ecran)
|
|||
|
|
|
|||
|
|
reponse = self._transform_ask(t("Advanced options? (y/N): "))
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
if not self._is_yes(reponse):
|
|||
|
|
# L'écran a pu répondre par feuille : ses réponses passent
|
|||
|
|
# AVANT les défauts, sans quoi refuser les options avancées
|
|||
|
|
# effacerait ce qu'on vient de cocher.
|
|||
|
|
for cle, valeur in DEFAUTS_RAPIDES.items():
|
|||
|
|
options.setdefault(cle, valeur)
|
|||
|
|
if "colonnes_intactes_index_par_feuille" in options:
|
|||
|
|
options["colonnes_intactes"] = []
|
|||
|
|
self._transform_dire_les_defauts(options)
|
|||
|
|
return self._transform_ask_hors_cellules(rapport, options)
|
|||
|
|
|
|||
|
|
noms = [f["nom"] for f in rapport.get("feuilles") or []]
|
|||
|
|
if len(noms) > 1:
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Sheets to process (empty = all): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
demandees = [n.strip() for n in reponse.split(",") if n.strip()]
|
|||
|
|
resolues = []
|
|||
|
|
for demandee in demandees:
|
|||
|
|
trouve = next(
|
|||
|
|
(n for n in noms if n.strip().lower() == demandee.lower()),
|
|||
|
|
None,
|
|||
|
|
)
|
|||
|
|
if trouve is None:
|
|||
|
|
manque = t(
|
|||
|
|
"The selection matches no sheet;"
|
|||
|
|
" nothing was written."
|
|||
|
|
)
|
|||
|
|
print(f"❌ {manque} {demandee}")
|
|||
|
|
return None
|
|||
|
|
resolues.append(trouve)
|
|||
|
|
options["feuilles"] = resolues
|
|||
|
|
if resolues:
|
|||
|
|
print(f" {', '.join(resolues)}")
|
|||
|
|
|
|||
|
|
if "colonnes_intactes_index_par_feuille" in options:
|
|||
|
|
# L'écran a répondu par feuille, ce que cette invite ne sait
|
|||
|
|
# pas faire : la reposer inviterait à une réponse GLOBALE qui
|
|||
|
|
# gèlerait la même colonne sur toutes les feuilles.
|
|||
|
|
options["colonnes_intactes"] = []
|
|||
|
|
else:
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Columns to leave untouched (empty = none): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["colonnes_intactes"] = [
|
|||
|
|
c.strip() for c in reponse.split(",") if c.strip()
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
reponse = self._transform_ask(t("Replace numbers? (Y/n): "))
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["nombres"] = self._is_yes_default_yes(reponse)
|
|||
|
|
if options["nombres"]:
|
|||
|
|
# Posée SOUS la question des nombres : sans eux elle n'a pas
|
|||
|
|
# d'objet, et une invite qui ne change rien apprend à passer
|
|||
|
|
# les invites.
|
|||
|
|
reponse = self._transform_ask(t("Keep the digit count? (y/N): "))
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["calibre_chiffres"] = self._is_yes(reponse)
|
|||
|
|
|
|||
|
|
reponse = self._transform_ask(t("Replace text? (Y/n): "))
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["texte"] = self._is_yes_default_yes(reponse)
|
|||
|
|
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Anonymise the header row too? (y/N): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["entetes"] = self._is_yes(reponse)
|
|||
|
|
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Random seed (empty = not reproducible): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["graine"] = reponse
|
|||
|
|
|
|||
|
|
return self._transform_ask_hors_cellules(rapport, options)
|
|||
|
|
|
|||
|
|
def _transform_ask_hors_cellules(self, rapport, options):
|
|||
|
|
"""Ce qu'un classeur porte HORS des cellules : macros, graphiques.
|
|||
|
|
|
|||
|
|
Ces deux-là ne sont pas des options avancées et restent posées
|
|||
|
|
dans les deux chemins : elles ne se voient pas dans les cellules,
|
|||
|
|
elles ne se devinent pas, et la réponse change ce que la copie
|
|||
|
|
porte. Elles ne se posent que si le fichier en a et que la copie
|
|||
|
|
peut les porter — une question dont la réponse ne change rien
|
|||
|
|
apprend à passer les questions.
|
|||
|
|
"""
|
|||
|
|
hors = rapport.get("hors_cellules") or {}
|
|||
|
|
garde_xlsm = (options["conversion"] or rapport["format"]) == "xlsx"
|
|||
|
|
if hors.get("macros") and garde_xlsm:
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Keep the VBA macros in the copy? (y/N): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["garder_macros"] = self._is_yes(reponse)
|
|||
|
|
if hors.get("graphiques") and garde_xlsm:
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Keep the charts in the copy? (y/N): ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
options["garder_graphiques"] = self._is_yes(reponse)
|
|||
|
|
return options
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_dire_les_defauts(options):
|
|||
|
|
"""Montrer ce que le chemin court a pris, sans le demander.
|
|||
|
|
|
|||
|
|
Un raccourci qui ne dit pas ce qu'il choisit n'est pas un
|
|||
|
|
raccourci mais une surprise : l'opérateur consent à ce qu'il LIT.
|
|||
|
|
|
|||
|
|
Des PHRASES et non un tableau de oui/non : ce sont des décisions
|
|||
|
|
prises, pas des questions posées, et « Remplacer les nombres →
|
|||
|
|
oui » se lit comme une invite dont on aurait perdu la réponse.
|
|||
|
|
"""
|
|||
|
|
# L'écran a pu geler des colonnes, et le chemin court les GARDE :
|
|||
|
|
# annoncer « aucune colonne laissée intacte » dirait alors le
|
|||
|
|
# contraire de ce qui s'applique, sur le seul écran où
|
|||
|
|
# l'opérateur consent.
|
|||
|
|
gelees = options.get("colonnes_intactes_index_par_feuille") or {}
|
|||
|
|
combien = sum(len(rangs) for rangs in gelees.values())
|
|||
|
|
perimetre = (
|
|||
|
|
f"{combien} {t('column(s) left untouched, chosen on screen')}"
|
|||
|
|
if combien
|
|||
|
|
else t("no column left untouched")
|
|||
|
|
)
|
|||
|
|
# Un empan VIDE veut dire « cette feuille n'a pas d'en-tête », et
|
|||
|
|
# l'écran y mène d'une frappe. Sa première ligne est alors de la
|
|||
|
|
# donnée et se fait remplacer : annoncer l'en-tête protégé dirait
|
|||
|
|
# le contraire de ce qui s'applique. Une feuille ABSENTE de la
|
|||
|
|
# table veut dire « mesure-la », ce qui laisse l'en-tête en place.
|
|||
|
|
empans = options.get("entetes_par_feuille") or {}
|
|||
|
|
sans_entete = [nom for nom, rangs in empans.items() if not rangs]
|
|||
|
|
entete = (
|
|||
|
|
f"{len(sans_entete)} {t('sheet(s) with no header row on screen')}"
|
|||
|
|
if sans_entete
|
|||
|
|
else t("the header row is left readable")
|
|||
|
|
)
|
|||
|
|
print(f" {t('Taken as given:')}")
|
|||
|
|
for ligne in (
|
|||
|
|
t("every sheet"),
|
|||
|
|
perimetre,
|
|||
|
|
t(
|
|||
|
|
"numbers replaced, widths kept"
|
|||
|
|
if options.get("calibre_chiffres")
|
|||
|
|
else "numbers replaced within the measured extent"
|
|||
|
|
),
|
|||
|
|
t("text replaced"),
|
|||
|
|
entete,
|
|||
|
|
t("no seed: not reproducible"),
|
|||
|
|
):
|
|||
|
|
print(f" {ligne}")
|
|||
|
|
|
|||
|
|
def _transform_ecran(self, rapport, table_chemin=""):
|
|||
|
|
"""Le périmètre choisi à l'écran, ou {} pour les invites.
|
|||
|
|
|
|||
|
|
Trois issues, et non deux : la spec porte le périmètre, `{}`
|
|||
|
|
demande les invites, et None annule. Confondre les deux dernières
|
|||
|
|
supprimerait le repli en silence.
|
|||
|
|
"""
|
|||
|
|
if not (rapport.get("feuilles") or []):
|
|||
|
|
return {}
|
|||
|
|
reponse = self._transform_ask(t("Open the TUI for setup? (Y/n): "))
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
if not self._is_yes_default_yes(reponse):
|
|||
|
|
return {}
|
|||
|
|
try:
|
|||
|
|
from script.todo import transform_form
|
|||
|
|
except ImportError: # pragma: no cover - textual peut manquer
|
|||
|
|
print(
|
|||
|
|
" ⚠ %s"
|
|||
|
|
% t("The screen needs textual; falling back to prompts.")
|
|||
|
|
)
|
|||
|
|
return {}
|
|||
|
|
try:
|
|||
|
|
spec = transform_form.run_transform_form(
|
|||
|
|
transform_form.contexte_depuis_rapport(
|
|||
|
|
rapport, self._transform_memoire(table_chemin)
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
except Exception as exc: # pragma: no cover - pas de terminal
|
|||
|
|
# Un écran qui ne peut pas s'ouvrir ne doit pas emporter le
|
|||
|
|
# travail : les invites savent tout demander.
|
|||
|
|
print(f" ⚠ {exc}")
|
|||
|
|
return {}
|
|||
|
|
if spec is None:
|
|||
|
|
return None
|
|||
|
|
return spec
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_memoire(table_chemin):
|
|||
|
|
"""Les lignes d'en-tête qu'une table de lot se rappelle.
|
|||
|
|
|
|||
|
|
Lue ICI et non par le moteur : l'écran vit dans le processus du
|
|||
|
|
menu, et la table est un simple JSON. Une table absente, illisible
|
|||
|
|
ou d'une version antérieure rend un dictionnaire vide — la mémoire
|
|||
|
|
est un confort, jamais une condition pour ouvrir l'écran.
|
|||
|
|
"""
|
|||
|
|
if not table_chemin or not os.path.isfile(table_chemin):
|
|||
|
|
return {}
|
|||
|
|
try:
|
|||
|
|
with open(table_chemin, "r", encoding="utf-8") as flux:
|
|||
|
|
brut = json.load(flux)
|
|||
|
|
# Une racine qui n'est pas un dictionnaire — un autre fichier
|
|||
|
|
# de `private/` désigné à l'invite — lèverait `AttributeError`
|
|||
|
|
# sur `.get`, hors de ce que la garde attrape, et emporterait
|
|||
|
|
# l'ouverture de l'écran. Le moteur refusera ce chemin en
|
|||
|
|
# NOMMANT la table ; ici la mémoire est un confort, et son
|
|||
|
|
# absence n'a rien à empêcher.
|
|||
|
|
if not isinstance(brut, dict):
|
|||
|
|
return {}
|
|||
|
|
return brut.get("entetes") or {}
|
|||
|
|
except (OSError, ValueError):
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# L'aperçu
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_preview(self, apercu):
|
|||
|
|
"""Montrer, PUIS demander. False si l'opérateur refuse."""
|
|||
|
|
print(f"\n {t('Preview — nothing written yet')}")
|
|||
|
|
print(
|
|||
|
|
f" {apercu.get('remplacees', 0)} {t('cell(s) replaced')}"
|
|||
|
|
f" — {apercu.get('texte', 0)} {t('text')},"
|
|||
|
|
f" {apercu.get('nombre', 0)} {t('numeric')}"
|
|||
|
|
)
|
|||
|
|
if apercu.get("hors_portee"):
|
|||
|
|
print(
|
|||
|
|
f" {apercu['hors_portee']}"
|
|||
|
|
f" {t('cell(s) left out of scope')}"
|
|||
|
|
)
|
|||
|
|
ecartees = apercu.get("colonnes_ecartees") or []
|
|||
|
|
if ecartees:
|
|||
|
|
noms = ", ".join(
|
|||
|
|
sorted(
|
|||
|
|
{str(c["etiquette"]) for c in ecartees if c["etiquette"]}
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
print(
|
|||
|
|
f" {t('column(s) left alone: they hold identifiers')}:"
|
|||
|
|
f" {noms}"
|
|||
|
|
)
|
|||
|
|
intactes = apercu.get("intactes") or {}
|
|||
|
|
if intactes:
|
|||
|
|
# Une date, un booléen, une valeur d'erreur traversent par
|
|||
|
|
# RÈGLE, non par oubli — et ce sont de vraies valeurs du
|
|||
|
|
# client. Ne les compter nulle part faisait signer un
|
|||
|
|
# consentement sur un fichier dont une colonne entière part
|
|||
|
|
# en clair sans que rien ne le dise.
|
|||
|
|
detail = ", ".join(
|
|||
|
|
f"{compte} {t(_LIBELLE_INTACTES.get(famille, famille))}"
|
|||
|
|
for famille, compte in sorted(intactes.items())
|
|||
|
|
if compte
|
|||
|
|
)
|
|||
|
|
print(f" ⚠ {t('Left intact')} : {detail}")
|
|||
|
|
# Le compte par famille dit POURQUOI, non OÙ : sur quarante
|
|||
|
|
# colonnes, « 7 date(s) laissées » ne dit pas laquelle sort. Le
|
|||
|
|
# critère est le même quelle que soit la raison — une colonne qui
|
|||
|
|
# porte quelque chose et dont rien n'a été remplacé part entière.
|
|||
|
|
# Celles qu'une réponse ou le plancher expliquent sont déjà
|
|||
|
|
# nommées au-dessus, et n'y reviennent pas.
|
|||
|
|
en_clair = apercu.get("colonnes_en_clair") or []
|
|||
|
|
if en_clair:
|
|||
|
|
noms = ", ".join(
|
|||
|
|
"%s/%s"
|
|||
|
|
% (
|
|||
|
|
colonne["feuille"],
|
|||
|
|
colonne["etiquette"] or "#%s" % colonne["index"],
|
|||
|
|
)
|
|||
|
|
for colonne in en_clair
|
|||
|
|
)
|
|||
|
|
print(f" ⚠ {t('column(s) entirely in clear')} : {noms}")
|
|||
|
|
gardee = apercu.get("entete_gardee") or []
|
|||
|
|
if gardee:
|
|||
|
|
# L'empan est MESURÉ, et il porte plusieurs lignes sur un
|
|||
|
|
# export mis en page : un compteur ne dirait pas qu'un nom est
|
|||
|
|
# dedans. Chaque valeur se montre donc, avec sa coordonnée.
|
|||
|
|
print(f" ⚠ {t('These cells are copied verbatim:')}")
|
|||
|
|
for cellule in gardee:
|
|||
|
|
print(f" {cellule['cellule']} {cellule['valeur']!r}")
|
|||
|
|
# Le plafond borne la LISTE, pas le compte : le taire faisait
|
|||
|
|
# consentir sur un extrait pris pour le tout.
|
|||
|
|
for nom, omises in sorted(
|
|||
|
|
(apercu.get("entete_gardee_omises") or {}).items()
|
|||
|
|
):
|
|||
|
|
print(f" {nom} : {omises} {t('more, not listed')}")
|
|||
|
|
print(" " + t("Correct the header rows to anonymise them."))
|
|||
|
|
for exemple in apercu.get("apercu") or []:
|
|||
|
|
print(
|
|||
|
|
f" {exemple['cellule']}"
|
|||
|
|
f" {exemple['avant']!r} → {exemple['apres']!r}"
|
|||
|
|
)
|
|||
|
|
for fichier in apercu.get("fichiers") or []:
|
|||
|
|
if fichier:
|
|||
|
|
print(f" → {fichier}")
|
|||
|
|
for avertissement in apercu.get("avertissements") or []:
|
|||
|
|
print(f" ⚠ {t(avertissement)}")
|
|||
|
|
return self._is_yes_default_yes(input(t("Write? (Y/n): ")))
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# La destination
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_select_destination(self, source, fmt, macros=False):
|
|||
|
|
"""Le chemin de sortie, ou None si l'opérateur renonce.
|
|||
|
|
|
|||
|
|
Le défaut NE REPREND PAS le nom source : sur un export réel, ce nom
|
|||
|
|
porte le client, la base ou l'année, et toute l'entrée existe pour
|
|||
|
|
produire un fichier transmissible. Le nom source reste offert, et
|
|||
|
|
l'invite dit alors qu'il n'a pas été anonymisé.
|
|||
|
|
|
|||
|
|
Un classeur qui GARDE ses macros se nomme `.xlsm` : Excel lie
|
|||
|
|
l'extension au contenu, et refuse d'ouvrir un `.xlsx` qui porte un
|
|||
|
|
projet VBA. La copie était écrite correctement et n'ouvrait pas.
|
|||
|
|
"""
|
|||
|
|
horodatage = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|||
|
|
extension = EXTENSION_PAR_FORMAT.get(fmt, ".out")
|
|||
|
|
if macros and extension == ".xlsx":
|
|||
|
|
extension = ".xlsm"
|
|||
|
|
defaut = os.path.join(
|
|||
|
|
SORTIE_PAR_DEFAUT, f"{horodatage}.anon{extension}"
|
|||
|
|
)
|
|||
|
|
print(f"\n {t('Destination')}: {defaut}")
|
|||
|
|
if extension == ".xlsm":
|
|||
|
|
print(
|
|||
|
|
" ⚠ %s"
|
|||
|
|
% t(
|
|||
|
|
"Macros are kept, so the copy is named .xlsm:"
|
|||
|
|
" Excel refuses a .xlsx holding a VBA project."
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
avis = t(
|
|||
|
|
"The file name is not anonymised;"
|
|||
|
|
" the default does not reuse it."
|
|||
|
|
)
|
|||
|
|
print(f" ⚠ {avis}")
|
|||
|
|
reponse = self._transform_ask(
|
|||
|
|
t("Enter to accept, p to browse, or type a path: ")
|
|||
|
|
)
|
|||
|
|
if reponse is None:
|
|||
|
|
return None
|
|||
|
|
if not reponse:
|
|||
|
|
return defaut
|
|||
|
|
if reponse.lower() == "p":
|
|||
|
|
if todo_file_browser is None:
|
|||
|
|
print(f" {defaut}")
|
|||
|
|
return defaut
|
|||
|
|
depart = os.path.dirname(os.path.abspath(defaut))
|
|||
|
|
if not os.path.isdir(depart):
|
|||
|
|
depart = os.getcwd()
|
|||
|
|
self._dir_path = ""
|
|||
|
|
navigateur = todo_file_browser.FileBrowser(
|
|||
|
|
depart, self._on_dir_selected, open_dir=True
|
|||
|
|
)
|
|||
|
|
navigateur.run_main_frame()
|
|||
|
|
if not self._dir_path:
|
|||
|
|
return None
|
|||
|
|
choisi = os.path.join(self._dir_path, os.path.basename(defaut))
|
|||
|
|
print(f" {choisi}")
|
|||
|
|
return choisi
|
|||
|
|
return os.path.expanduser(reponse)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_table_par_defaut(destination):
|
|||
|
|
"""Le chemin de la table, TOUJOURS sous private/transform/.
|
|||
|
|
|
|||
|
|
La table porte chaque valeur d'origine en clair : elle
|
|||
|
|
ré-identifie la copie à elle seule. La poser à côté du fichier à
|
|||
|
|
transmettre — ce que faisait le défaut, puisque la destination est
|
|||
|
|
justement choisie hors de private/ — fait partir la clé avec le
|
|||
|
|
chiffré au premier `zip -r` ou `scp -r` du dossier de livraison.
|
|||
|
|
|
|||
|
|
Le nom DÉRIVE de la destination, ce qui reste le plus utile pour
|
|||
|
|
retrouver la table d'une copie une semaine plus tard. Mais il ne
|
|||
|
|
tient qu'au nom de BASE : deux livraisons rangées par année,
|
|||
|
|
« 2026/export.xlsx » et « 2027/export.xlsx », donnaient le même
|
|||
|
|
chemin. L'invite promet « vide = nouvelle » et le moteur chargeait
|
|||
|
|
la table de la livraison précédente — ses mots, ses nombres, ses
|
|||
|
|
empans — sans que rien ne signale sa présence, l'écran ayant déjà
|
|||
|
|
été bâti sur une mémoire vide.
|
|||
|
|
|
|||
|
|
Un chemin déjà pris prend donc un suffixe frais, et l'appelant le
|
|||
|
|
DIT. Le partage voulu à l'intérieur d'un lot passe par l'invite,
|
|||
|
|
où l'opérateur tape le chemin de la table précédente.
|
|||
|
|
"""
|
|||
|
|
chemin = TransformMenuMixin._transform_table_derivee(destination)
|
|||
|
|
if not os.path.exists(chemin):
|
|||
|
|
return chemin
|
|||
|
|
horodatage = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|||
|
|
racine = chemin[: -len(".table.json")]
|
|||
|
|
return f"{racine}-{horodatage}.table.json"
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_table_derivee(destination):
|
|||
|
|
"""Le nom que la destination donne à sa table, SANS voir le disque.
|
|||
|
|
|
|||
|
|
Séparé pour que l'appelant sache si le chemin retenu est bien
|
|||
|
|
celui-là — comparer deux chemins vaut mieux que renifler un tiret
|
|||
|
|
dans un nom de base, qui en porte un dès que la destination garde
|
|||
|
|
l'horodatage du défaut.
|
|||
|
|
"""
|
|||
|
|
base = os.path.basename(os.path.splitext(destination)[0])
|
|||
|
|
return os.path.join(SORTIE_PAR_DEFAUT, f"{base}.table.json")
|
|||
|
|
|
|||
|
|
def _transform_confirm_overwrite(self, cibles):
|
|||
|
|
"""Le nom tapé en entier, comme le reste du dépôt l'exige."""
|
|||
|
|
existants = [c for c in cibles if c and os.path.exists(c)]
|
|||
|
|
if not existants:
|
|||
|
|
return True
|
|||
|
|
print(f" {t('These files would be overwritten: ')}")
|
|||
|
|
for cible in existants:
|
|||
|
|
print(f" {cible}")
|
|||
|
|
attendu = os.path.basename(existants[0])
|
|||
|
|
reponse = input(
|
|||
|
|
t(
|
|||
|
|
"This file already exists. Type its name in full to"
|
|||
|
|
" overwrite: "
|
|||
|
|
)
|
|||
|
|
).strip()
|
|||
|
|
if reponse != attendu:
|
|||
|
|
print(f"❌ {t('Name does not match, nothing was written.')}")
|
|||
|
|
return False
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Le déroulé complet
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_open_and_report(self):
|
|||
|
|
chemin = self._transform_select_file()
|
|||
|
|
if not chemin:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
format_devine = self._transform_format(chemin)
|
|||
|
|
if not transform_setup.ensure(format_devine):
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
rapport = self._transform_run(["--report", chemin], format_devine)
|
|||
|
|
if rapport is None:
|
|||
|
|
return
|
|||
|
|
print(self._transform_render_report(rapport))
|
|||
|
|
|
|||
|
|
if rapport.get("arret"):
|
|||
|
|
print(f"\n❌ {t(rapport['arret'])}")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
options = self._transform_ask_options(rapport)
|
|||
|
|
if options is None:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
sortie_fmt = options["conversion"] or rapport["format"]
|
|||
|
|
fmt_moteur = self._transform_fmt_moteur(rapport["format"], sortie_fmt)
|
|||
|
|
destination = self._transform_select_destination(
|
|||
|
|
chemin, sortie_fmt, options.get("garder_macros")
|
|||
|
|
)
|
|||
|
|
if not destination:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
options["destination"] = destination
|
|||
|
|
if not options.get("table_chemin"):
|
|||
|
|
options["table_chemin"] = self._transform_table_par_defaut(
|
|||
|
|
destination
|
|||
|
|
)
|
|||
|
|
# « vide = nouvelle » est une promesse : le dire quand le nom
|
|||
|
|
# dérivé était déjà pris, plutôt que d'adopter en silence la
|
|||
|
|
# table d'une autre livraison.
|
|||
|
|
if options["table_chemin"] != self._transform_table_derivee(
|
|||
|
|
destination
|
|||
|
|
):
|
|||
|
|
print(
|
|||
|
|
f" {t('A table of that name exists; using a new one.')}"
|
|||
|
|
)
|
|||
|
|
print(f" {t('Mapping table: ')}{options['table_chemin']}")
|
|||
|
|
|
|||
|
|
arguments = [
|
|||
|
|
"--plan",
|
|||
|
|
chemin,
|
|||
|
|
"--options",
|
|||
|
|
json.dumps(options, ensure_ascii=False),
|
|||
|
|
]
|
|||
|
|
if options.get("table_chemin"):
|
|||
|
|
arguments += ["--table", options["table_chemin"]]
|
|||
|
|
apercu = self._transform_run(arguments, fmt_moteur)
|
|||
|
|
if apercu is None:
|
|||
|
|
return
|
|||
|
|
if not self._transform_preview(apercu):
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
if not self._transform_confirm_overwrite(
|
|||
|
|
apercu.get("fichiers") or [destination]
|
|||
|
|
):
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
arguments = [
|
|||
|
|
"--apply",
|
|||
|
|
chemin,
|
|||
|
|
"--out",
|
|||
|
|
destination,
|
|||
|
|
"--options",
|
|||
|
|
json.dumps(options, ensure_ascii=False),
|
|||
|
|
]
|
|||
|
|
if options.get("table_chemin"):
|
|||
|
|
arguments += ["--table", options["table_chemin"]]
|
|||
|
|
bilan = self._transform_run(arguments, fmt_moteur)
|
|||
|
|
if bilan is None:
|
|||
|
|
return
|
|||
|
|
self._transform_render_bilan(bilan, destination)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_fmt_moteur(source_fmt, sortie_fmt):
|
|||
|
|
"""Le format qui DÉCIDE de l'interpréteur du moteur.
|
|||
|
|
|
|||
|
|
Le processus doit lire la source ET écrire la cible : il lui faut
|
|||
|
|
l'union des bibliothèques. La source seule laissait csv→xlsx
|
|||
|
|
importer openpyxl sous l'interpréteur du CLI, qui ne l'a pas ; la
|
|||
|
|
cible seule enverrait un classeur au même interpréteur, qui ne sait
|
|||
|
|
pas le lire.
|
|||
|
|
"""
|
|||
|
|
if source_fmt not in transform_setup.FORMATS_STDLIB:
|
|||
|
|
return source_fmt
|
|||
|
|
return sortie_fmt
|
|||
|
|
|
|||
|
|
def _transform_render_bilan(self, bilan, destination):
|
|||
|
|
for fichier in bilan.get("fichiers") or [destination]:
|
|||
|
|
print(f"✅ {t('Written: ')}{fichier}")
|
|||
|
|
print(
|
|||
|
|
f" {bilan.get('remplacees', 0)} {t('cell(s) replaced')}"
|
|||
|
|
f" — {bilan.get('texte', 0)} {t('text')},"
|
|||
|
|
f" {bilan.get('nombre', 0)} {t('numeric')}"
|
|||
|
|
)
|
|||
|
|
intactes = bilan.get("intactes") or {}
|
|||
|
|
if intactes:
|
|||
|
|
detail = ", ".join(
|
|||
|
|
f"{compte} {t(_LIBELLE_INTACTES.get(famille, famille))}"
|
|||
|
|
for famille, compte in sorted(intactes.items())
|
|||
|
|
if compte
|
|||
|
|
)
|
|||
|
|
print(f" {t('Left intact')} : {detail}")
|
|||
|
|
if bilan.get("hors_portee"):
|
|||
|
|
print(
|
|||
|
|
f" {bilan['hors_portee']}"
|
|||
|
|
f" {t('cell(s) left out of scope')}"
|
|||
|
|
)
|
|||
|
|
hors = bilan.get("hors_cellules") or {}
|
|||
|
|
# `isinstance(True, int)` vaut True : sans exclure les booléens, le
|
|||
|
|
# drapeau des macros comptait pour un élément effacé.
|
|||
|
|
efface = sum(
|
|||
|
|
v
|
|||
|
|
for v in hors.values()
|
|||
|
|
if isinstance(v, int) and not isinstance(v, bool)
|
|||
|
|
)
|
|||
|
|
if efface:
|
|||
|
|
print(f" {efface} {t('element(s) outside cells wiped')}")
|
|||
|
|
non_verifiees = bilan.get("valeurs_non_verifiees") or 0
|
|||
|
|
if non_verifiees:
|
|||
|
|
print(
|
|||
|
|
f" ⚠ {non_verifiees}"
|
|||
|
|
f" {t('value(s) not verified, above the cap')}"
|
|||
|
|
)
|
|||
|
|
print(f" {t('The original was not modified.')}")
|
|||
|
|
if bilan.get("table"):
|
|||
|
|
print(f" {t('Mapping table written: ')}{bilan['table']}")
|
|||
|
|
avis = t("This table re-identifies the copy; keep it in private/.")
|
|||
|
|
print(f" ⚠ {avis}")
|
|||
|
|
for avertissement in bilan.get("avertissements") or []:
|
|||
|
|
print(f" ⚠ {t(avertissement)}")
|
|||
|
|
if self._transform_sous_private(destination):
|
|||
|
|
avis = t(
|
|||
|
|
"private/ is tracked by git — this file will show in"
|
|||
|
|
" « git status »."
|
|||
|
|
)
|
|||
|
|
print(f" ⚠ {avis}")
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_sous_private(destination):
|
|||
|
|
"""La destination est-elle sous `private/` du dépôt ?
|
|||
|
|
|
|||
|
|
Comparaison de `realpath` ancrés : un test de préfixe sur le chemin
|
|||
|
|
tel qu'il est tapé taisait l'avertissement pour le chemin ABSOLU
|
|||
|
|
que rend le navigateur — c'est-à-dire le cas le plus courant — et
|
|||
|
|
le levait à tort pour « privateer/ ».
|
|||
|
|
"""
|
|||
|
|
prive = os.path.realpath(
|
|||
|
|
os.path.join(transform_setup.racine(), "private")
|
|||
|
|
)
|
|||
|
|
cible = os.path.realpath(os.path.abspath(str(destination)))
|
|||
|
|
return cible == prive or cible.startswith(prive + os.sep)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_format(chemin):
|
|||
|
|
"""Le format présumé, pour choisir l'interpréteur AVANT de lire.
|
|||
|
|
|
|||
|
|
Le moteur retranchera par les octets ; ici l'extension suffit,
|
|||
|
|
puisqu'il ne s'agit que de savoir s'il faut le venv dédié.
|
|||
|
|
"""
|
|||
|
|
extension = os.path.splitext(chemin)[1].lower()
|
|||
|
|
return {
|
|||
|
|
".xlsx": "xlsx",
|
|||
|
|
".xlsm": "xlsx",
|
|||
|
|
".xlsb": "xlsx",
|
|||
|
|
".xls": "xls",
|
|||
|
|
".mdb": "access",
|
|||
|
|
".accdb": "access",
|
|||
|
|
".csv": "csv",
|
|||
|
|
".xml": "xml",
|
|||
|
|
".json": "json",
|
|||
|
|
}.get(extension, "xlsx")
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# Une base Odoo, ou une sauvegarde
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_anonymise_base(self):
|
|||
|
|
"""Anonymiser une base Odoo, ou une sauvegarde qui en redevient une.
|
|||
|
|
|
|||
|
|
Rien n'est réécrit ici : `anonymize.py` sait déjà remplacer les
|
|||
|
|
données d'une base, et `_monitoring_write_flow` sait déjà montrer
|
|||
|
|
puis demander, avec le nom retapé pour confirmer. Cette méthode ne
|
|||
|
|
fait que choisir la SOURCE et, quand c'en était un, refaire un zip.
|
|||
|
|
|
|||
|
|
Un zip n'est jamais modifié : il est restauré dans une base, qui
|
|||
|
|
est anonymisée, puis vidée dans un zip NEUF. Une base locale, elle,
|
|||
|
|
est modifiée EN PLACE — l'invite le dit, et le nom retapé le
|
|||
|
|
confirme.
|
|||
|
|
"""
|
|||
|
|
from script.analyse import monitoring
|
|||
|
|
|
|||
|
|
analyse = next(
|
|||
|
|
(a for a in monitoring.ANALYSES if a["key"] == "anonymize"), None
|
|||
|
|
)
|
|||
|
|
if analyse is None: # pragma: no cover - la table est en dur
|
|||
|
|
print(f"❌ {t('Command not found !')}")
|
|||
|
|
return
|
|||
|
|
choix = self._transform_anonymise_source()
|
|||
|
|
if choix is None:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
base, depuis_zip = choix
|
|||
|
|
print()
|
|||
|
|
print(monitoring.describe_source(monitoring.KIND_DATABASE, base))
|
|||
|
|
if not depuis_zip:
|
|||
|
|
# La base MÊME est modifiée : le dire avant, non après.
|
|||
|
|
print(f"⚠ {t('The database ITSELF is modified; no copy.')}")
|
|||
|
|
ecrit = self._monitoring_write_flow(analyse, base)
|
|||
|
|
# APRÈS le travail, et quel que soit son sort : une base restaurée
|
|||
|
|
# existe sur le serveur même si l'anonymisation a été refusée, et
|
|||
|
|
# c'est précisément celle-là qu'il faut pouvoir retrouver.
|
|||
|
|
self._transform_noter_base(base, depuis_zip)
|
|||
|
|
if not ecrit:
|
|||
|
|
# Sans cette garde, un renoncement produisait un zip de la
|
|||
|
|
# base NON anonymisée, annoncé comme le résultat d'une
|
|||
|
|
# anonymisation — le pire des deux mondes : l'opérateur croit
|
|||
|
|
# tenir une copie transmissible et tient l'original.
|
|||
|
|
print(f" {t('Nothing was written; no backup was drawn.')}")
|
|||
|
|
print(f" {t('Database kept: ')}{base}")
|
|||
|
|
return
|
|||
|
|
if depuis_zip:
|
|||
|
|
self._transform_export_zip(base)
|
|||
|
|
|
|||
|
|
def _transform_anonymise_source(self):
|
|||
|
|
"""(nom de base, venait d'un zip), ou None si l'on renonce.
|
|||
|
|
|
|||
|
|
Les trois mêmes provenances que « Monitoring », dans le même
|
|||
|
|
ordre : deux ordres différents dans un même logiciel se paient en
|
|||
|
|
hésitation à chaque usage.
|
|||
|
|
"""
|
|||
|
|
print(f"[1] {t('A local database')}")
|
|||
|
|
print(f"[2] {t('A local backup .zip')}")
|
|||
|
|
print(f"[3] {t('A remote backup (https + master password)')}")
|
|||
|
|
print(f"[0] {t('Back')}")
|
|||
|
|
reponse = click.prompt(t("Command:"), prompt_suffix=" ")
|
|||
|
|
print()
|
|||
|
|
if reponse == "1":
|
|||
|
|
base = self.db_manager.select_database()
|
|||
|
|
return (base, False) if base else None
|
|||
|
|
if reponse == "2":
|
|||
|
|
chemin = self.db_manager.select_backup_path()
|
|||
|
|
base = self._monitoring_restore(chemin) if chemin else None
|
|||
|
|
return (base, True) if base else None
|
|||
|
|
if reponse == "3":
|
|||
|
|
statut, chemin, _nom = (
|
|||
|
|
self.db_manager.download_database_backup_cli()
|
|||
|
|
)
|
|||
|
|
if statut or not chemin or not os.path.isfile(chemin):
|
|||
|
|
print(f"❌ {t('The download did not produce a usable file.')}")
|
|||
|
|
return None
|
|||
|
|
base = self._monitoring_restore(chemin)
|
|||
|
|
return (base, True) if base else None
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _transform_export_zip(self, base):
|
|||
|
|
"""Vider la base anonymisée dans un zip NEUF.
|
|||
|
|
|
|||
|
|
Par la sauvegarde d'Odoo lui-même — `odoo_bin.sh db --backup` —
|
|||
|
|
et non par un `pg_dump` à nous : c'est Odoo qui écrit le
|
|||
|
|
`dump.sql`, le `filestore/` et le `manifest.json`, donc le format
|
|||
|
|
est juste par construction. Le zip atterrit dans `image_db/`,
|
|||
|
|
d'où les sauvegardes viennent et où le navigateur les cherche.
|
|||
|
|
|
|||
|
|
Un nom déjà pris n'est jamais écrasé en silence : l'invite le dit
|
|||
|
|
et propose un nom frais, que l'opérateur peut changer.
|
|||
|
|
"""
|
|||
|
|
defaut = "%s_anon_%s" % (
|
|||
|
|
base,
|
|||
|
|
datetime.datetime.now().strftime("%Y%m%d-%H%M%S"),
|
|||
|
|
)
|
|||
|
|
nom = self._transform_ask(
|
|||
|
|
f"{t('Backup name (Enter to accept): ')}{defaut} "
|
|||
|
|
)
|
|||
|
|
if nom is None:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
nom = (nom or defaut).removesuffix(".zip")
|
|||
|
|
cible = os.path.join(
|
|||
|
|
transform_setup.racine(), "image_db", nom + ".zip"
|
|||
|
|
)
|
|||
|
|
if os.path.isfile(cible):
|
|||
|
|
print(f" {t('These files would be overwritten: ')}")
|
|||
|
|
print(f" {cible}")
|
|||
|
|
if not self._transform_confirm_overwrite([cible]):
|
|||
|
|
return
|
|||
|
|
statut, _sortie = self.execute.exec_command_live(
|
|||
|
|
f"./odoo_bin.sh db --backup --database {base}"
|
|||
|
|
f" --restore_image {nom}",
|
|||
|
|
return_status_and_output=True,
|
|||
|
|
single_source_erplibre=True,
|
|||
|
|
source_erplibre=False,
|
|||
|
|
)
|
|||
|
|
if statut:
|
|||
|
|
print(f"❌ {t('The download did not produce a usable file.')}")
|
|||
|
|
return
|
|||
|
|
print(f"✅ {t('Backup written: ')}{cible}")
|
|||
|
|
print(f" {t('The original was not modified.')}")
|
|||
|
|
print(f" {t('Database kept: ')}{base}")
|
|||
|
|
|
|||
|
|
def _transform_noter_base(self, base, depuis_zip):
|
|||
|
|
"""Inscrire la base produite au registre, sans jamais lever.
|
|||
|
|
|
|||
|
|
Le serveur ne dit pas QUI a créé une base : sans registre, cette
|
|||
|
|
entrée ne pourrait que lister toutes les bases de la machine, ce
|
|||
|
|
que le menu Database fait déjà. Le registre est ce qui lui permet
|
|||
|
|
de ne montrer que ce que Transform data a produit.
|
|||
|
|
|
|||
|
|
Un registre illisible ou impossible à écrire ne doit pas emporter
|
|||
|
|
le travail : il est un confort, comme la mémoire de lot.
|
|||
|
|
"""
|
|||
|
|
registre = os.path.join(SORTIE_PAR_DEFAUT, "bases.json")
|
|||
|
|
connues = self._transform_bases_lues()
|
|||
|
|
connues = [b for b in connues if b.get("base") != base]
|
|||
|
|
connues.append(
|
|||
|
|
{
|
|||
|
|
"base": base,
|
|||
|
|
"depuis_zip": bool(depuis_zip),
|
|||
|
|
"date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
try:
|
|||
|
|
os.makedirs(SORTIE_PAR_DEFAUT, mode=0o700, exist_ok=True)
|
|||
|
|
with open(registre, "w", encoding="utf-8") as flux:
|
|||
|
|
json.dump(connues, flux, ensure_ascii=False, indent=1)
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_bases_lues():
|
|||
|
|
"""Le registre, ou une liste vide s'il ne se lit pas."""
|
|||
|
|
registre = os.path.join(SORTIE_PAR_DEFAUT, "bases.json")
|
|||
|
|
try:
|
|||
|
|
with open(registre, "r", encoding="utf-8") as flux:
|
|||
|
|
lues = json.load(flux)
|
|||
|
|
except (OSError, ValueError):
|
|||
|
|
return []
|
|||
|
|
return [b for b in lues if isinstance(b, dict) and b.get("base")]
|
|||
|
|
|
|||
|
|
def _transform_bases_produites(self):
|
|||
|
|
"""Lister les bases que cette entrée a produites, et les effacer.
|
|||
|
|
|
|||
|
|
Le même raisonnement que « Copies produites » : sans elle, le
|
|||
|
|
serveur devient un tas que rien n'annonce, et le bruit finit par
|
|||
|
|
se lire comme du fond. Une base anonymisée ne porte plus de
|
|||
|
|
donnée client, mais elle occupe le serveur et elle se confond avec
|
|||
|
|
une base de travail.
|
|||
|
|
|
|||
|
|
Une base inscrite au registre peut avoir été détruite ailleurs —
|
|||
|
|
par le menu Database, par un `db_drop_all`. L'entrée le DIT plutôt
|
|||
|
|
que de la proposer à l'effacement.
|
|||
|
|
"""
|
|||
|
|
inscrites = self._transform_bases_lues()
|
|||
|
|
if not inscrites:
|
|||
|
|
print(f" {t('Empty file.')} {SORTIE_PAR_DEFAUT}")
|
|||
|
|
return
|
|||
|
|
vivantes = self._transform_bases_vivantes()
|
|||
|
|
presentes, disparues = [], []
|
|||
|
|
for entree in inscrites:
|
|||
|
|
cible = presentes if entree["base"] in vivantes else disparues
|
|||
|
|
cible.append(entree)
|
|||
|
|
for entree in presentes:
|
|||
|
|
marque = "📦" if entree.get("depuis_zip") else "🗄"
|
|||
|
|
print(f" {marque} {entree['base']:<40} {entree['date']}")
|
|||
|
|
print(f" {len(presentes)} {t('Database:')}")
|
|||
|
|
if disparues:
|
|||
|
|
print()
|
|||
|
|
print(f" {t('Gone already, only in the register:')}")
|
|||
|
|
for entree in disparues:
|
|||
|
|
print(f" {entree['base']}")
|
|||
|
|
if not presentes:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
print(f" {t('Use Database to drop one.')}")
|
|||
|
|
|
|||
|
|
def _transform_bases_vivantes(self):
|
|||
|
|
"""Les bases que le serveur porte VRAIMENT, ou () si on ne sait pas.
|
|||
|
|
|
|||
|
|
La MÊME commande que `select_database` — `odoo_bin.sh db --list` —
|
|||
|
|
et non un appel à psql de plus : deux façons de demander la même
|
|||
|
|
chose se contredisent le jour où l'une change.
|
|||
|
|
|
|||
|
|
Une liste vide et une ignorance ne se confondent pas : sans
|
|||
|
|
serveur joignable, tout le registre paraîtrait disparu. Les deux
|
|||
|
|
rendent `()`, et l'appelant ne propose alors aucun effacement.
|
|||
|
|
"""
|
|||
|
|
statut, sortie = self.execute.exec_command_live(
|
|||
|
|
"./odoo_bin.sh db --list",
|
|||
|
|
return_status_and_output=True,
|
|||
|
|
quiet=True,
|
|||
|
|
source_erplibre=False,
|
|||
|
|
single_source_erplibre=True,
|
|||
|
|
)
|
|||
|
|
if statut:
|
|||
|
|
return ()
|
|||
|
|
return {ligne.strip() for ligne in sortie if ligne.strip()}
|
|||
|
|
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
# L'environnement
|
|||
|
|
# ------------------------------------------------------------------
|
|||
|
|
def _transform_install_env(self):
|
|||
|
|
if transform_setup.available("xlsx"):
|
|||
|
|
print(f"✅ {t('The environment is ready.')}")
|
|||
|
|
else:
|
|||
|
|
transform_setup.create()
|
|||
|
|
# La PRÉSENCE d'abord, le paquet ensuite. Demander au gestionnaire
|
|||
|
|
# de paquets si l'outil est là rendait « aucun paquet connu »
|
|||
|
|
# devant un `mdb-queries` installé — vrai sur une distribution
|
|||
|
|
# dont ce module ne connaît pas le paquet, et faux sur ce que
|
|||
|
|
# l'opérateur voit dans son PATH.
|
|||
|
|
if transform_setup.requetes_access_lisibles():
|
|||
|
|
print(f"✅ {t('Access saved queries are readable here.')}")
|
|||
|
|
return
|
|||
|
|
commande = transform_setup.system_packages_cmd()
|
|||
|
|
if commande is None:
|
|||
|
|
print(f" {t('No package known here for:')} mdb-queries")
|
|||
|
|
print(
|
|||
|
|
f" {t('Install it by hand to read Access saved queries.')}"
|
|||
|
|
)
|
|||
|
|
return
|
|||
|
|
print(f"\n{t('Access files need a system package.')}")
|
|||
|
|
print(f"{t('The installation requires sudo.')}")
|
|||
|
|
from script.todo import todo_install
|
|||
|
|
|
|||
|
|
todo_install.ask_and_install(
|
|||
|
|
self.execute,
|
|||
|
|
commande,
|
|||
|
|
t("Create it now? (Y/n): "),
|
|||
|
|
self._is_yes_default_yes,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def _transform_capabilities(self):
|
|||
|
|
for fmt, disponible in transform_setup.capabilities().items():
|
|||
|
|
etat = t("readable") if disponible else t("not readable")
|
|||
|
|
print(f" {fmt:<8} {etat}")
|
|||
|
|
|
|||
|
|
def _transform_copies(self):
|
|||
|
|
"""Lister les copies produites, et proposer de les effacer.
|
|||
|
|
|
|||
|
|
Rien n'est ajouté à `.gitignore` : les copies restent visibles dans
|
|||
|
|
« git status », ce qui est le rappel qu'elles existent et qu'elles
|
|||
|
|
portent de la donnée client. Cette entrée est ce qui rend ce choix
|
|||
|
|
vivable — sans elle, le répertoire deviendrait un tas que rien
|
|||
|
|
n'annonce, et le bruit finirait par se lire comme du fond.
|
|||
|
|
"""
|
|||
|
|
if not os.path.isdir(SORTIE_PAR_DEFAUT):
|
|||
|
|
print(f" {t('Empty file.')} {SORTIE_PAR_DEFAUT}")
|
|||
|
|
return
|
|||
|
|
produites, etrangeres = self._transform_inventaire()
|
|||
|
|
if not produites and not etrangeres:
|
|||
|
|
print(f" {t('Empty file.')} {SORTIE_PAR_DEFAUT}")
|
|||
|
|
return
|
|||
|
|
total = 0
|
|||
|
|
for nom, taille in produites:
|
|||
|
|
total += taille
|
|||
|
|
print(f" {nom:<44} {taille:>10} o")
|
|||
|
|
print(f" {len(produites)} {t('File')} — {total} o")
|
|||
|
|
if etrangeres:
|
|||
|
|
# Le navigateur de fichiers ouvre son parcours dans ce même
|
|||
|
|
# répertoire : ce qui s'y trouve n'est pas toujours une copie
|
|||
|
|
# produite ici, et l'effacement en bloc l'emportait aussi.
|
|||
|
|
print(f"\n {t('Not produced here, left alone:')}")
|
|||
|
|
for nom, taille in etrangeres:
|
|||
|
|
print(f" {nom:<44} {taille:>10} o")
|
|||
|
|
if not produites:
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
reponse = input(
|
|||
|
|
t("Delete all copies in private/transform/? (y/N): ")
|
|||
|
|
).strip()
|
|||
|
|
if not self._is_yes(reponse):
|
|||
|
|
print(t("Nothing to do."))
|
|||
|
|
return
|
|||
|
|
for nom, _taille in produites:
|
|||
|
|
chemin = os.path.join(SORTIE_PAR_DEFAUT, nom)
|
|||
|
|
try:
|
|||
|
|
if os.path.isdir(chemin):
|
|||
|
|
# Une source à plusieurs feuilles convertie en csv
|
|||
|
|
# écrit un RÉPERTOIRE. Il était listé comme un fichier
|
|||
|
|
# avec sa taille d'inode, puis laissé sur le disque,
|
|||
|
|
# plein de lignes dérivées du client.
|
|||
|
|
shutil.rmtree(chemin)
|
|||
|
|
else:
|
|||
|
|
os.unlink(chemin)
|
|||
|
|
except OSError as exc:
|
|||
|
|
print(f" ⚠ {exc}")
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _transform_inventaire():
|
|||
|
|
"""(produites ici, étrangères) — chacune en (nom, octets).
|
|||
|
|
|
|||
|
|
Ce que l'outil produit se reconnaît à son nom : « .anon. » quelque
|
|||
|
|
part, ou le suffixe de la table. Tout le reste est à quelqu'un
|
|||
|
|
d'autre et n'est pas à effacer.
|
|||
|
|
"""
|
|||
|
|
produites, etrangeres = [], []
|
|||
|
|
for nom in sorted(os.listdir(SORTIE_PAR_DEFAUT)):
|
|||
|
|
chemin = os.path.join(SORTIE_PAR_DEFAUT, nom)
|
|||
|
|
taille = 0
|
|||
|
|
if os.path.isdir(chemin):
|
|||
|
|
for racine, _dossiers, fichiers in os.walk(chemin):
|
|||
|
|
for fichier in fichiers:
|
|||
|
|
try:
|
|||
|
|
taille += os.path.getsize(
|
|||
|
|
os.path.join(racine, fichier)
|
|||
|
|
)
|
|||
|
|
except OSError:
|
|||
|
|
continue
|
|||
|
|
else:
|
|||
|
|
try:
|
|||
|
|
taille = os.path.getsize(chemin)
|
|||
|
|
except OSError:
|
|||
|
|
continue
|
|||
|
|
cible = (
|
|||
|
|
produites
|
|||
|
|
if ".anon." in nom or nom.endswith(".table.json")
|
|||
|
|
else etrangeres
|
|||
|
|
)
|
|||
|
|
cible.append((nom, taille))
|
|||
|
|
return produites, etrangeres
|