script/todo/migration_form.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
+"""Écran de reprise de migration Odoo, en TUI.
+
+Pendant de `qemu_deploy_form` pour l'outil de migration. Deux interfaces
+posent la MÊME première question — « où en est-on, et par où reprend-on ? » —
+et renvoient les MÊMES chaînes de réponse (« c », « n », « r », « q »,
+« 0 »..« 4 », « 4.<version> ») que `TodoUpgrade.apply_resume_answer` traduit
+en progression. La décision est donc écrite une seule fois.
+
+- run_resume_tui(ctx, run_app=True) : renvoie la réponse, ou None pour
+  retomber sur les invites en ligne.
+
+`ctx` vient de `TodoUpgrade.resume_context()` : pure donnée, aucun accès à la
+base ni au disque depuis l'affichage.
+"""
+from __future__ import annotations
+
+try:
+    from script.todo.todo_i18n import t
+except Exception:  # pragma: no cover - repli si i18n indisponible
+
+    def t(key: str) -> str:
+        return key
+
+
+def version_line(versions):
+    """« 13✓ 14✓ 15 16 17 18 » — l'avancement des montées de version."""
+    return "  ".join(
+        f"{v['version']}{'✓' if v['done'] else ''}" for v in versions
+    )
+
+
+def next_version(versions):
+    """Première version pas encore migrée : celle que « continuer » reprend."""
+    for item in versions:
+        if not item["done"]:
+            return item["version"]
+    return None
+
+
+def run_resume_tui(ctx, run_app: bool = True):
+    """Écran de reprise. Renvoie la réponse choisie, ou None si annulé.
+    `run_app=False` renvoie l'instance sans la lancer (tests headless)."""
+    from textual.app import App, ComposeResult
+    from textual.containers import Horizontal, Vertical
+    from textual.widgets import (
+        Button,
+        DataTable,
+        Footer,
+        Header,
+        OptionList,
+        Static,
+    )
+    from textual.widgets.option_list import Option
+
+    result = {"answer": None}
+
+    class Resume(App):
+        CSS = """
+        #head { height: auto; padding: 0 1; color: $text-muted; }
+        #steps { height: auto; max-height: 12; border: solid $accent; }
+        #bumps { height: auto; max-height: 10; border: solid $panel; }
+        .grouptitle { color: $accent; text-style: bold; padding: 1 1 0 1; }
+        #actions { height: auto; padding: 1 1 0 1; }
+        #hint { height: auto; color: $text-muted; padding: 0 1; }
+        """
+        BINDINGS = [
+            ("c", "cont", t("Continue where it stopped")),
+            ("n", "new", t("New migration, erase everything")),
+            ("r", "keep_zip", t("Keep the zip only")),
+            ("q", "quit_nothing", t("Quit without doing anything")),
+            ("escape", "quit_nothing", t("Quit without doing anything")),
+        ]
+
+        def compose(self) -> ComposeResult:
+            yield Header()
+            yield Static(
+                f"  {t('File'):<9}: {ctx['file']}\n"
+                f"  {t('Database'):<9}: {ctx['database']}"
+                f"   ·   {t('Target')} : {ctx['target']}\n"
+                f"  {t('Started'):<9}: {ctx['started']}",
+                id="head",
+            )
+            yield Static(f"{t('Steps')}", classes="grouptitle")
+            yield DataTable(id="steps")
+            if ctx["versions"]:
+                yield Static(
+                    f"{t('Version bumps')}  ({version_line(ctx['versions'])})",
+                    classes="grouptitle",
+                )
+                yield OptionList(id="bumps")
+            with Vertical():
+                with Horizontal(id="actions"):
+                    yield Button(
+                        t("Continue where it stopped"),
+                        variant="primary",
+                        id="a_cont",
+                    )
+                    yield Button(t("New migration"), id="a_new")
+                    yield Button(t("Keep the zip only"), id="a_keep")
+                    yield Button(t("Quit"), id="a_quit")
+                yield Static(
+                    f"  {t('Enter on a step or a version = replay from there')}",
+                    id="hint",
+                )
+            yield Footer()
+
+        def on_mount(self) -> None:
+            self.title = t("Migration in progress")
+            table = self.query_one("#steps", DataTable)
+            table.cursor_type = "row"
+            table.add_columns("", "", t("Step"), t("Detail"))
+            for item in ctx["steps"]:
+                table.add_row(
+                    f"[{item['step']}]",
+                    item["icon"],
+                    item["label"],
+                    item["detail"],
+                )
+            # Curseur sur la première étape inachevée : c'est là que ça a
+            # calé, donc là qu'on veut probablement rejouer.
+            for index, item in enumerate(ctx["steps"]):
+                if item["icon"] != "✅":
+                    table.move_cursor(row=index)
+                    break
+            if ctx["versions"]:
+                bumps = self.query_one("#bumps", OptionList)
+                for item in ctx["versions"]:
+                    mark = "✓" if item["done"] else " "
+                    bumps.add_option(
+                        Option(
+                            f" {mark} Odoo {item['version']}.0 — "
+                            f"{t('rebuilds the intermediate database')}",
+                            id=str(item["version"]),
+                        )
+                    )
+                upcoming = next_version(ctx["versions"])
+                if upcoming is not None:
+                    bumps.highlighted = [
+                        v["version"] for v in ctx["versions"]
+                    ].index(upcoming)
+
+        # -- choix ------------------------------------------------------ #
+        def _answer(self, value):
+            result["answer"] = value
+            self.exit()
+
+        def on_data_table_row_selected(self, event) -> None:
+            index = event.cursor_row
+            if 0 <= index < len(ctx["steps"]):
+                self._answer(str(ctx["steps"][index]["step"]))
+
+        def on_option_list_option_selected(self, event) -> None:
+            self._answer(f"4.{event.option.id}")
+
+        def on_button_pressed(self, event) -> None:
+            mapping = {
+                "a_cont": "c",
+                "a_new": "n",
+                "a_keep": "r",
+                "a_quit": "q",
+            }
+            value = mapping.get(event.button.id)
+            if value:
+                self._answer(value)
+
+        def action_cont(self) -> None:
+            self._answer("c")
+
+        def action_new(self) -> None:
+            self._answer("n")
+
+        def action_keep_zip(self) -> None:
+            self._answer("r")
+
+        def action_quit_nothing(self) -> None:
+            self._answer("q")
+
+    app = Resume()
+    app._result = result  # lecture par les tests headless
+    if not run_app:
+        return app
+    app.run()
+    return result["answer"]

script/todo/todo.py

@@ -642,6 +642,14 @@ class TODO:
                 ("tui", "TUI, collapsible blocks per VM"),
             ),
         ),
+        "migration_ui": (
+            "Odoo migration interface",
+            (
+                ("ask", "Ask every time"),
+                ("tui", "TUI form"),
+                ("cli", "Classic questions (line by line)"),
+            ),
+        ),
     }
 
     def _pref_label(self, key):
@@ -690,6 +698,12 @@ class TODO:
                         f"({self._pref_label('qemu_deploy_progress')})"
                     )
                 },
+                {
+                    "prompt_description": (
+                        f"{t('Odoo migration interface')}  "
+                        f"({self._pref_label('migration_ui')})"
+                    )
+                },
                 {"section": t("Maintenance")},
                 {"prompt_description": t("Reset all preferences")},
             ]
@@ -704,6 +718,8 @@ class TODO:
             elif status == "3":
                 self._pref_edit("qemu_deploy_progress")
             elif status == "4":
+                self._pref_edit("migration_ui")
+            elif status == "5":
                 n = todo_prefs.reset()
                 print(f"✅ {t('Preferences reset')} ({n})")
             else:

script/todo/todo_i18n.py

@@ -1399,6 +1399,42 @@ TRANSLATIONS = {
         "fr": "Choix (numéro, vide = garder) :",
         "en": "Choice (number, blank = keep):",
     },
+    "Odoo migration interface": {
+        "fr": "🚚 Interface de la migration Odoo",
+        "en": "🚚 Odoo migration interface",
+    },
+    "No migration in progress to resume.": {
+        "fr": "Aucune migration en cours à reprendre.",
+        "en": "No migration in progress to resume.",
+    },
+    "Quit without doing anything": {
+        "fr": "Quitter sans rien faire",
+        "en": "Quit without doing anything",
+    },
+    "Version bumps": {
+        "fr": "Montées de version",
+        "en": "Version bumps",
+    },
+    "Step": {
+        "fr": "Étape",
+        "en": "Step",
+    },
+    "Detail": {
+        "fr": "Détail",
+        "en": "Detail",
+    },
+    "Enter on a step or a version = replay from there": {
+        "fr": "Entrée sur une étape ou une version = rejouer depuis là",
+        "en": "Enter on a step or a version = replay from there",
+    },
+    "New migration": {
+        "fr": "Nouvelle migration",
+        "en": "New migration",
+    },
+    "Keep the zip only": {
+        "fr": "Garder seulement le zip",
+        "en": "Keep the zip only",
+    },
     "Reset all preferences": {
         "fr": "🧹 Réinitialiser toutes les préférences",
         "en": "🧹 Reset all preferences",

script/todo/todo_prefs.py

@@ -28,6 +28,8 @@ DEFAULTS = {
     # Affichage pendant le déploiement : "cli" (sortie texte, facile à copier
     # depuis le terminal) ou "tui" (blocs repliables + copie OSC 52).
     "qemu_deploy_progress": "cli",
+    # Interface de la migration Odoo : "ask" / "tui" / "cli".
+    "migration_ui": "ask",
 }
 
 

script/todo/todo_upgrade.py

@@ -156,54 +156,91 @@ class TodoUpgrade:
             return "✅", t("done")
         return "⏳", t("partially done")
 
-    def prompt_resume(self, old_dct_progression):
-        """Show where the migration stands and ask what to do next.
+    def resume_context(self, old_dct_progression):
+        """Everything the resume screen shows, as plain data.
 
-        Returns (progression, changed). The previous menu exposed internal key
-        names (« Reuse database without state_4 »), which said nothing about
-        what would actually happen. This shows the real state of every step and
-        lets the user replay from any of them.
+        No I/O: the line-by-line prompt and the TUI both render THIS, so the
+        two can never describe the migration differently.
         """
         migration_file = old_dct_progression.get("migration_file") or "?"
+        steps = []
+        for step, label in MIGRATION_STEP:
+            icon, detail = self.step_status(old_dct_progression, step)
+            steps.append(
+                {
+                    "step": step,
+                    "icon": icon,
+                    "label": t(label),
+                    "detail": detail,
+                }
+            )
+        lst_version = self.version_bumps(old_dct_progression)
+        done = old_dct_progression.get("state_4_upgrade_odoo_lst") or []
+        return {
+            "file": os.path.basename(migration_file),
+            "database": old_dct_progression.get("config_database_name") or "?",
+            "target": old_dct_progression.get("target_odoo_version") or "?",
+            "started": old_dct_progression.get("date_create") or "?",
+            "steps": steps,
+            "versions": [
+                {
+                    "version": version,
+                    "done": bool(i < len(done) and done[i]),
+                }
+                for i, version in enumerate(lst_version)
+            ],
+        }
+
+    @staticmethod
+    def print_resume(ctx):
+        """Render the resume screen on the terminal."""
         print()
         print(f"📍 {t('Migration in progress')}")
         # Pad in code, not in the translations: the labels differ in length
         # between languages and a hardcoded padding misaligns the colons.
-        print(f"   {t('File'):<9}: {os.path.basename(migration_file)}")
+        print(f"   {t('File'):<9}: {ctx['file']}")
         print(
-            f"   {t('Database'):<9}: "
-            f"{old_dct_progression.get('config_database_name') or '?'}"
-            f"   ·   {t('Target')} :"
-            f" {old_dct_progression.get('target_odoo_version') or '?'}"
-        )
-        print(
-            f"   {t('Started'):<9}: "
-            f"{old_dct_progression.get('date_create')}"
+            f"   {t('Database'):<9}: {ctx['database']}"
+            f"   ·   {t('Target')} : {ctx['target']}"
         )
+        print(f"   {t('Started'):<9}: {ctx['started']}")
         print()
         print(f"   {t('Steps')} :")
-        for step, label in MIGRATION_STEP:
-            icon, detail = self.step_status(old_dct_progression, step)
-            print(f"     [{step}] {icon} {t(label):<44} {detail}")
+        for item in ctx["steps"]:
+            print(
+                f"     [{item['step']}] {item['icon']} "
+                f"{item['label']:<44} {item['detail']}"
+            )
         print()
-        lst_version = self.version_bumps(old_dct_progression)
         print(f"   [c] {t('Continue where it stopped')}")
         print(
             f"   [0-4] {t('Replay from that step')}"
             f" ({t('erases the progression of that step and the next ones')})"
         )
-        if lst_version:
+        if ctx["versions"]:
+            versions = "/".join(str(v["version"]) for v in ctx["versions"])
             print(
                 f"   [4.N] {t('Replay the upgrade from version N')}"
-                f" ({'/'.join(str(v) for v in lst_version)}) —"
+                f" ({versions}) —"
                 f" {t('rebuilds the intermediate database')}"
             )
         print(f"   [n] {t('New migration, erase everything')}")
         print(f"   [r] {t('Keep the zip only, ask every question again')}")
-        answer = input(f"💬 {t('Your choice')} : ").strip().lower()
+        print(f"   [q] {t('Quit without doing anything')}")
+
+    def apply_resume_answer(self, old_dct_progression, answer, ctx):
+        """Turn the answer into (progression, changed), or None to quit.
+
+        THE decision point, shared by both interfaces: the TUI returns the
+        same answer strings as the prompt, so this logic is written once.
+        """
+        answer = (answer or "").strip().lower()
+        lst_version = [v["version"] for v in ctx["versions"]]
 
         if answer in ("", "c"):
             return old_dct_progression, False
+        if answer == "q":
+            return None
         if answer == "n":
             return {}, True
         if answer == "r":
@@ -231,6 +268,57 @@ class TodoUpgrade:
         print(f"⚠️ {t('Unknown choice, continuing where it stopped')}.")
         return old_dct_progression, False
 
+    def prompt_resume(self, old_dct_progression, use_tui=False):
+        """Show where the migration stands and ask what to do next.
+
+        Returns (progression, changed), or None if the user quits. The old
+        menu exposed internal key names (« Reuse database without state_4 »),
+        which said nothing about what would happen; this shows the real state
+        of every step and lets the user replay from any of them.
+        """
+        ctx = self.resume_context(old_dct_progression)
+        answer = None
+        if use_tui:
+            answer = self.resume_tui(ctx)
+        if answer is None:
+            self.print_resume(ctx)
+            answer = input(f"💬 {t('Your choice')} : ")
+        return self.apply_resume_answer(old_dct_progression, answer, ctx)
+
+    @staticmethod
+    def ask_ui():
+        """Interface of the migration: TUI or line-by-line prompts.
+
+        The preference can settle it in advance (TODO > Configuration);
+        « ask » asks. Same contract as the QEMU deployment.
+        """
+        try:
+            from script.todo import todo_prefs
+
+            pref = todo_prefs.get("migration_ui")
+        except Exception:
+            pref = "ask"
+        if pref in ("tui", "cli"):
+            return pref
+        print(f"\n{t('Interface:')}")
+        print(f"  [1] {t('TUI form')} *")
+        print(f"  [2] {t('Classic questions (line by line)')}")
+        print(f"  {t('(change the default in TODO > Configuration)')}")
+        answer = input(t("Choice (1-2, default 1): ")).strip()
+        return "cli" if answer == "2" else "tui"
+
+    @staticmethod
+    def resume_tui(ctx):
+        """Resume screen as a TUI. Returns the SAME answer strings as the
+        prompt, or None when textual is missing (fall back to the prompt)."""
+        try:
+            from script.todo.migration_form import run_resume_tui
+
+            return run_resume_tui(ctx)
+        except ImportError:
+            print(t("Install textual for the dashboard (pip)."))
+            return None
+
     @staticmethod
     def version_bumps(dct_progression):
         """Odoo versions the step 4 loop walks through, e.g. [13, 14, ..., 18].
@@ -634,14 +722,21 @@ class TodoUpgrade:
         self.dct_module_per_dct_version_path = {}
         default_database_name = "test"
 
+        use_tui = self.ask_ui() == "tui"
+
         if os.path.exists(UPGRADE_DATABASE_CONFIG_LOG):
             old_dct_progression = self.read_progression()
             if old_dct_progression:
-                self.dct_progression, changed = self.prompt_resume(
-                    old_dct_progression
-                )
+                resumed = self.prompt_resume(old_dct_progression, use_tui)
+                if resumed is None:
+                    return
+                self.dct_progression, changed = resumed
                 if changed:
                     self.write_config()
+            elif use_tui:
+                print(f"ℹ️  {t('No migration in progress to resume.')}")
+        elif use_tui:
+            print(f"ℹ️  {t('No migration in progress to resume.')}")
 
         if "migration_file" in self.dct_progression:
             self.file_path = self.dct_progression["migration_file"]