From 1bc2b50ca8c083fe7e7e83ec871fb0729912bac8 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Fri, 21 Aug 2026 22:39:48 -0400 Subject: [PATCH] [ADD] filestore: say if the record still exists, and offer the cleanups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A lost image on a deleted task is not a loss -- nobody will ever look for it. On a LIVING task it is one, and it is the only one worth regretting. The report now says which: project.task #15 and calendar.event #1 both still exist, so those two really are gone. Three states, not two: "could not check" must not read as "it is gone", or a real loss gets filed as a false alarm. Two repairs sit in the follow-up menu. Purging rows whose field no longer exists deletes by ID, never by a rebuilt domain -- replaying the reasoning in SQL would open the door to deleting more than was shown. Tidying the nested filestore moves up what is missing and deletes pure duplicates, never overwriting a file already in place. --- FR --- Une image perdue sur une tâche supprimée n'est pas une perte : personne ne la cherchera. Sur une tâche VIVANTE, c'en est une, et la seule à regretter. Le rapport le dit : project.task #15 et calendar.event #1 existent encore, ces deux-là sont bien perdues. Trois états, pas deux : « pas pu vérifier » ne doit pas se lire « a disparu », sans quoi une vraie perte passe pour une fausse alerte. Deux réparations dans le menu de suite. La purge efface par IDENTIFIANT, jamais par un domaine reconstruit -- rejouer le raisonnement en SQL ouvrirait la porte à effacer plus que ce qui a été montré. Le rangement remonte ce qui manque et supprime les doublons purs, sans jamais écraser un fichier déjà en place. Assisted-by: Claude Opus 5 --- script/analyse/check_filestore.py | 129 ++++++++++++++- script/todo/todo.py | 94 +++++++++++ test/test_check_filestore.py | 261 +++++++++++++++++++++++++++++- 3 files changed, 475 insertions(+), 9 deletions(-) diff --git a/script/analyse/check_filestore.py b/script/analyse/check_filestore.py index dc01984..23ba9db 100755 --- a/script/analyse/check_filestore.py +++ b/script/analyse/check_filestore.py @@ -130,6 +130,7 @@ def attachments(database): lignes = run_psql( database, "SELECT a.store_fname, coalesce(a.res_model, '')," + " a.id::text," " coalesce(a.res_field, ''), coalesce(a.res_id::text, '')," " coalesce(a.name, ''), coalesce(a.file_size::text, '0')," " coalesce(a.mimetype, ''), coalesce(a.create_date::text, '')" @@ -142,15 +143,16 @@ def attachments(database): { "store_fname": ligne[0], "model": ligne[1], - "field": ligne[2], - "res_id": ligne[3], - "name": ligne[4], - "size": int(ligne[5] or 0), - "mimetype": ligne[6], - "created": ligne[7][:10], + "id": ligne[2], + "field": ligne[3], + "res_id": ligne[4], + "name": ligne[5], + "size": int(ligne[6] or 0), + "mimetype": ligne[7], + "created": ligne[8][:10], } for ligne in lignes - if len(ligne) >= 8 + if len(ligne) >= 9 ] @@ -221,6 +223,45 @@ def scan_backups(dossier): return trouves +def resources_alive(database, pieces): + """{(modele, id): True/False} — l'enregistrement visé existe-t-il ? + + Une image perdue dont la tâche a été supprimée n'est pas une perte : + personne ne la cherchera jamais. Une image perdue sur une tâche + VIVANTE en est une, et c'est la seule qu'il faille regretter. + Confondre les deux, c'est pleurer au hasard. + + Une requête par modèle, sur les seules pièces jointes déjà classées + perdues — jamais sur les milliers d'autres. + """ + par_modele = {} + for piece in pieces: + if not piece.get("model") or not (piece.get("res_id") or "").isdigit(): + continue + par_modele.setdefault(piece["model"], set()).add(int(piece["res_id"])) + vivants = {} + for modele, ids in par_modele.items(): + table = modele.replace(".", "_").replace("'", "") + # Un modèle abstrait ou transitoire n'a pas de table : interroger + # une table absente rendrait None, qu'on lirait « n'existe pas ». + # Ce serait un mensonge, et le pire sens du mensonge ici. + existe = run_psql( + database, f"SELECT to_regclass('{table}') IS NOT NULL" + ) + if not existe or existe[0][0] != "t": + continue + liste = ", ".join(str(i) for i in sorted(ids)) + lignes = run_psql( + database, f"SELECT id FROM {table} WHERE id IN ({liste})" + ) + if lignes is None: + continue + trouves = {int(ligne[0]) for ligne in lignes if ligne and ligne[0]} + for identifiant in ids: + vivants[(modele, str(identifiant))] = identifiant in trouves + return vivants + + def files_on_disk(racine, database): """(au bon niveau, nichés) pour une base. Deux ensembles de noms.""" base = os.path.join(racine, database) @@ -375,6 +416,9 @@ def audit(database, config_path=None, backups=None): continue vus.add(piece["store_fname"]) groupes[verdict[0]].append(dict(piece, where=verdict[1])) + vivants = resources_alive(database, groupes["lost"]) + for piece in groupes["lost"]: + piece["alive"] = vivants.get((piece["model"], piece["res_id"])) return { "database": database, "attachments": len(pieces), @@ -427,7 +471,7 @@ def render(rapport, limit=20): lignes.append( f" {piece['name'][:44] or '(sans nom)':<46}" f" {piece['size'] // 1024:>6} ko {ou}{champ}" - f" {piece['created']}" + f" {piece['created']} {alive_mark(piece)}" ) if len(groupe) > limit: lignes.append(f" … {len(groupe) - limit} {t('more')}") @@ -444,6 +488,21 @@ def render_nested(rapport): ] +def alive_mark(piece): + """Dire si l'enregistrement visé vit encore, ou qu'on ne sait pas. + + Trois états, pas deux : « on n'a pas pu vérifier » ne doit pas se + lire comme « il a disparu », sans quoi une vraie perte serait classée + en fausse alerte. + """ + etat = piece.get("alive") + if etat is True: + return f"✓ {t('record still exists')}" + if etat is False: + return f"✗ {t('record is gone — nothing will miss it')}" + return "" + + def summarise(groupe): """« modele / champ × N », pour dire beaucoup en peu de lignes.""" compte = {} @@ -456,6 +515,60 @@ def summarise(groupe): ] +def purge_dead_sql(rapport): + """Le SQL qui efface les lignes dont le champ n'existe plus, ou "". + + On efface par IDENTIFIANT, pas par un domaine reconstruit : la liste + a été établie en confrontant `ir_model_fields` à ce que la base + porte, et rejouer ce raisonnement en SQL laisserait la porte ouverte + à effacer autre chose que ce qui a été montré. + """ + ids = sorted( + int(piece["id"]) + for piece in rapport["groups"]["dead_field"] + if str(piece.get("id", "")).isdigit() + ) + if not ids: + return "" + liste = ", ".join(str(i) for i in ids) + return f"DELETE FROM ir_attachment WHERE id IN ({liste})" + + +def nested_dir(rapport): + """Le dossier imbriqué de cette base, s'il en existe un.""" + chemin = os.path.join(rapport["root"], "filestore") + return chemin if os.path.isdir(chemin) else "" + + +def tidy_nested_plan(rapport): + """(à remonter, doublons) parmi les fichiers imbriqués. + + Deux tas très différents : ce qui MANQUE au bon niveau doit y + remonter, ce qui y est déjà est un doublon pur. Les traiter d'un + bloc écraserait des fichiers présents par des copies — inutile, et + inquiétant sur une base de production. + """ + dossier = nested_dir(rapport) + if not dossier: + return [], [] + bons, _n = files_on_disk( + os.path.dirname(rapport["root"]), os.path.basename(rapport["root"]) + ) + remonter, doublons = [], [] + for deux in sorted(os.listdir(dossier)): + profond = os.path.join(dossier, deux) + if not os.path.isdir(profond): + continue + for nom in sorted(os.listdir(profond)): + complet = os.path.join(profond, nom) + if not os.path.isfile(complet): + continue + (doublons if f"{deux}/{nom}" in bons else remonter).append( + (complet, os.path.join(rapport["root"], deux, nom)) + ) + return remonter, doublons + + def main(argv=None): import argparse diff --git a/script/todo/todo.py b/script/todo/todo.py index d8549b8..d9ccbb0 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -10295,6 +10295,10 @@ class TODO: def handler(rank): if rank == 1: print("\n".join(filestore.render(rapport, limit=0))) + elif rank == 2: + self._filestore_purge_dead(database, rapport) + elif rank == 3: + self._filestore_tidy_nested(rapport) else: self._analyse_export_json( rapport, os.path.basename(database), "filestore" @@ -10303,11 +10307,101 @@ class TODO: self._analyse_follow_up( [ {"prompt_description": t("Show every entry")}, + { + "prompt_description": t( + "🧹 Purge attachments whose field no longer exists" + ) + }, + { + "prompt_description": t( + "🧹 Tidy the nested filestore Odoo never reads" + ) + }, {"prompt_description": t("Export as JSON")}, ], handler, ) + def _filestore_purge_dead(self, database, rapport): + """Effacer les pièces jointes dont le champ a disparu. + + La seule ÉCRITURE en base de tout le menu Analyse. Elle porte sur + des lignes que plus rien ne lit — `res.country.image` est devenu + `image_url`, calculé, en 13 — mais elle reste une suppression : + question explicite, défaut à « non », et le compte est relu avant + de partir. + """ + from script.analyse import check_filestore as filestore + from script.todo import auto_ask + + lignes = rapport["groups"]["dead_field"] + sql = filestore.purge_dead_sql(rapport) + if not sql: + print(f"ℹ️ {t('Nothing to purge.')}") + return + print() + for texte in filestore.summarise(lignes): + print(f" {texte}") + question = ( + f"💬 {t('Delete these')} {len(lignes)}" + f" {t('attachment row(s) for good?')} (y/N) : " + ) + if auto_ask.ask(question, default="n").strip().lower() not in ( + "y", + "yes", + "o", + ): + print(f"ℹ️ {t('Nothing was deleted.')}") + return + status = self.execute.exec_command_live( + f'psql -d {database} -c "{sql}"', + source_erplibre=False, + single_source_erplibre=True, + ) + if status: + print(f"❌ {t('The purge failed.')}") + return + print(f"✅ {len(lignes)} {t('attachment row(s) deleted.')}") + + def _filestore_tidy_nested(self, rapport): + """Remonter ce qui manque, effacer les doublons purs. + + Deux tas, deux gestes. Écraser un fichier présent par une copie + identique ne gagnerait rien et brouillerait la trace ; c'est + pourquoi les doublons sont comptés à part et jamais déplacés. + """ + from script.analyse import check_filestore as filestore + from script.todo import auto_ask + + remonter, doublons = filestore.tidy_nested_plan(rapport) + if not remonter and not doublons: + print(f"ℹ️ {t('No nested filestore to tidy.')}") + return + print() + print(f" {len(remonter)} {t('file(s) to move up')}") + print(f" {len(doublons)} {t('pure duplicate(s) to delete')}") + print(f" {t('Directory:')} {filestore.nested_dir(rapport)}") + if auto_ask.ask( + f"💬 {t('Go ahead?')} (y/N) : ", default="n" + ).strip().lower() not in ("y", "yes", "o"): + print(f"ℹ️ {t('Nothing was moved.')}") + return + deplaces, effaces = 0, 0 + for source, cible in remonter: + os.makedirs(os.path.dirname(cible), exist_ok=True) + shutil.move(source, cible) + deplaces += 1 + for source, _cible in doublons: + os.remove(source) + effaces += 1 + dossier = filestore.nested_dir(rapport) + if dossier: + shutil.rmtree(dossier, ignore_errors=True) + print( + f"✅ {deplaces} {t('moved up')}, {effaces}" + f" {t('duplicate(s) removed')}." + ) + def _analyse_offer_install(self, database, rapport): """Proposer d'installer ce qui manque, quand c'est installable. diff --git a/test/test_check_filestore.py b/test/test_check_filestore.py index b486f38..3834252 100644 --- a/test/test_check_filestore.py +++ b/test/test_check_filestore.py @@ -28,10 +28,11 @@ from script.analyse import check_filestore as fs # noqa: E402 from script.todo import todo_i18n # noqa: E402 -def piece(store, model="", field="", res_id="1", name="x", size=1024): +def piece(store, model="", field="", res_id="1", name="x", size=1024, pid="7"): return { "store_fname": store, "model": model, + "id": pid, "field": field, "res_id": res_id, "name": name, @@ -289,6 +290,264 @@ class TestTheReport(unittest.TestCase): self.assertIn("res.country / image × 2", resume) +class TestTheLivingRecord(unittest.TestCase): + """Une image perdue sur un enregistrement supprimé n'est pas perdue.""" + + def setUp(self): + self.vrai = fs.run_psql + self.demandes = [] + + def tearDown(self): + fs.run_psql = self.vrai + + def branche(self, tables, ids): + def faux(base, sql): + self.demandes.append(sql) + if "to_regclass" in sql: + return [["t" if any(x in sql for x in tables) else "f"]] + return [[str(i)] for i in ids] + + fs.run_psql = faux + + def test_a_living_record_is_reported_as_such(self): + self.branche(["project_task"], [15]) + vivants = fs.resources_alive( + "db", [piece("a/1", "project.task", res_id="15")] + ) + self.assertIs(vivants[("project.task", "15")], True) + + def test_a_deleted_record_is_reported_as_gone(self): + self.branche(["project_task"], []) + vivants = fs.resources_alive( + "db", [piece("a/1", "project.task", res_id="15")] + ) + self.assertIs(vivants[("project.task", "15")], False) + + def test_a_model_without_a_table_is_left_UNKNOWN(self): + # « on n'a pas pu vérifier » ne doit pas se lire « il a disparu » : + # une vraie perte serait classée en fausse alerte. + self.branche([], []) + vivants = fs.resources_alive( + "db", [piece("a/1", "un.abstrait", res_id="1")] + ) + self.assertEqual(vivants, {}) + + def test_an_attachment_without_a_record_is_not_queried(self): + self.branche([], []) + fs.resources_alive("db", [piece("a/1")]) + self.assertEqual(self.demandes, []) + + def test_the_mark_says_all_three_states(self): + self.assertIn( + todo_i18n.t("record still exists"), fs.alive_mark({"alive": True}) + ) + self.assertIn( + todo_i18n.t("record is gone — nothing will miss it"), + fs.alive_mark({"alive": False}), + ) + self.assertEqual(fs.alive_mark({"alive": None}), "") + + +class TestTheRepairs(unittest.TestCase): + def rapport(self, morts=(), racine="/fs/db"): + base = {v: [] for v in fs.VERDICTS} + base["dead_field"] = list(morts) + return {"root": racine, "groups": base} + + def test_the_purge_deletes_by_id_only(self): + # Par IDENTIFIANT, jamais par un domaine reconstruit : rejouer le + # raisonnement en SQL ouvrirait la porte à effacer autre chose + # que ce qui a été montré. + sql = fs.purge_dead_sql( + self.rapport([piece("a/1", pid="3"), piece("b/2", pid="9")]) + ) + self.assertIn("WHERE id IN (3, 9)", sql) + self.assertNotIn("res_model", sql) + + def test_nothing_dead_gives_no_sql(self): + self.assertEqual(fs.purge_dead_sql(self.rapport()), "") + + def test_a_row_without_an_id_is_never_deleted(self): + sql = fs.purge_dead_sql(self.rapport([piece("a/1", pid="")])) + self.assertEqual(sql, "") + + +class TestTidyingTheNested(unittest.TestCase): + def setUp(self): + self.racine = tempfile.mkdtemp() + self.base = os.path.join(self.racine, "ma_base") + for chemin in ("aa/deja", "filestore/aa/deja", "filestore/bb/absent"): + complet = os.path.join(self.base, chemin) + os.makedirs(os.path.dirname(complet), exist_ok=True) + with open(complet, "w", encoding="utf-8") as handle: + handle.write("x") + + def tearDown(self): + shutil.rmtree(self.racine) + + def test_it_separates_what_to_move_from_pure_duplicates(self): + # Écraser un fichier présent par une copie identique ne gagne + # rien et brouille la trace : les deux tas restent distincts. + remonter, doublons = fs.tidy_nested_plan({"root": self.base}) + self.assertEqual( + [os.path.basename(a) for a, _b in remonter], ["absent"] + ) + self.assertEqual([os.path.basename(a) for a, _b in doublons], ["deja"]) + + def test_the_destination_is_the_right_level(self): + remonter, _d = fs.tidy_nested_plan({"root": self.base}) + self.assertEqual( + remonter[0][1], os.path.join(self.base, "bb", "absent") + ) + + def test_no_nested_directory_gives_an_empty_plan(self): + shutil.rmtree(os.path.join(self.base, "filestore")) + self.assertEqual(fs.tidy_nested_plan({"root": self.base}), ([], [])) + self.assertEqual(fs.nested_dir({"root": self.base}), "") + + +class TestTheRepairMenu(unittest.TestCase): + """Les réparations ÉCRIVENT. On éprouve ce qui part, pas les appels. + + Un test qui constate qu'une fonction a été appelée n'aurait pas vu + que `exec_command` n'existe pas — seul `exec_command_live` est + offert. C'est arrivé ici : le code aurait planté au premier usage. + """ + + def setUp(self): + from script.todo import auto_ask + from script.todo import todo as todo_module + + self.auto_ask = auto_ask + self.vrai_ask = auto_ask.ask + self.lancees = [] + self.obj = todo_module.TODO.__new__(todo_module.TODO) + self.obj.execute = type( + "E", + (), + { + "exec_command_live": lambda _s, cmd, **k: self.lancees.append( + cmd + ) + or 0 + }, + )() + + def tearDown(self): + self.auto_ask.ask = self.vrai_ask + + def repond(self, *reponses): + file = list(reponses) + + def faux(prompt, default="", seconds=None): + reponse = file.pop(0) if file else "" + return reponse or default + + self.auto_ask.ask = faux + + def rapport(self, morts=()): + groupes = {v: [] for v in fs.VERDICTS} + groupes["dead_field"] = list(morts) + return {"root": "/fs/db", "groups": groupes} + + def test_saying_no_deletes_nothing(self): + self.repond("n") + with redirect_stdout(io.StringIO()): + self.obj._filestore_purge_dead("db", self.rapport([piece("a/1")])) + self.assertEqual(self.lancees, []) + + def test_pressing_enter_deletes_nothing(self): + # Le défaut d'une SUPPRESSION doit être de ne rien faire. + self.repond("") + with redirect_stdout(io.StringIO()): + self.obj._filestore_purge_dead("db", self.rapport([piece("a/1")])) + self.assertEqual(self.lancees, []) + + def test_accepting_runs_a_command_that_actually_exists(self): + self.repond("y") + with redirect_stdout(io.StringIO()): + self.obj._filestore_purge_dead( + "db", self.rapport([piece("a/1", pid="3")]) + ) + self.assertEqual(len(self.lancees), 1) + self.assertIn("psql -d db", self.lancees[0]) + self.assertIn("WHERE id IN (3)", self.lancees[0]) + + def test_nothing_dead_asks_nothing(self): + demandes = [] + self.auto_ask.ask = lambda p, default="", seconds=None: ( + demandes.append(p) or "y" + ) + with redirect_stdout(io.StringIO()): + self.obj._filestore_purge_dead("db", self.rapport()) + self.assertEqual(demandes, []) + self.assertEqual(self.lancees, []) + + def test_the_menu_offers_both_repairs(self): + racine = os.path.normpath( + os.path.join(os.path.dirname(__file__), "..") + ) + with io.open( + os.path.join(racine, "script", "todo", "todo.py"), encoding="utf-8" + ) as handle: + src = handle.read() + self.assertIn("_filestore_purge_dead", src) + self.assertIn("_filestore_tidy_nested", src) + # Le garde du défaut doit rester collé à la question : c'est lui + # qui empêche Entrée de supprimer. + self.assertIn('auto_ask.ask(question, default="n")', src) + + +class TestTidyingForReal(unittest.TestCase): + def setUp(self): + from script.todo import auto_ask + from script.todo import todo as todo_module + + self.auto_ask = auto_ask + self.vrai_ask = auto_ask.ask + self.racine = tempfile.mkdtemp() + self.base = os.path.join(self.racine, "ma_base") + for chemin, contenu in ( + ("aa/deja", "bon"), + ("filestore/aa/deja", "copie"), + ("filestore/bb/absent", "utile"), + ): + complet = os.path.join(self.base, chemin) + os.makedirs(os.path.dirname(complet), exist_ok=True) + with open(complet, "w", encoding="utf-8") as handle: + handle.write(contenu) + self.obj = todo_module.TODO.__new__(todo_module.TODO) + + def tearDown(self): + self.auto_ask.ask = self.vrai_ask + shutil.rmtree(self.racine, ignore_errors=True) + + def test_refusing_leaves_every_file_where_it_was(self): + self.auto_ask.ask = lambda p, default="", seconds=None: "n" + with redirect_stdout(io.StringIO()): + self.obj._filestore_tidy_nested({"root": self.base}) + self.assertTrue( + os.path.isfile( + os.path.join(self.base, "filestore", "bb", "absent") + ) + ) + + def test_accepting_moves_up_and_removes_the_nest(self): + self.auto_ask.ask = lambda p, default="", seconds=None: "y" + with redirect_stdout(io.StringIO()): + self.obj._filestore_tidy_nested({"root": self.base}) + self.assertTrue( + os.path.isfile(os.path.join(self.base, "bb", "absent")) + ) + self.assertFalse(os.path.isdir(os.path.join(self.base, "filestore"))) + # Le fichier déjà présent n'a pas été ÉCRASÉ par sa copie : + # vérifier sa seule existence laisserait passer l'écrasement. + with io.open( + os.path.join(self.base, "aa", "deja"), encoding="utf-8" + ) as handle: + self.assertEqual(handle.read(), "bon") + + class TestVerifyingARestore(unittest.TestCase): """Le contrôle d'après-restauration, celui qui aurait vu le nichage."""