tryton-db-migration/extract_updates.py

#!/usr/bin/env python3
"""
Analysiert __register__-Methoden in Tryton-Modulen und extrahiert alle
Datenbank-Operationen (Spalten-/Tabellen-Änderungen, execute/insert/update/...),
die dort ausgeführt werden. Ergebnis wird pro Modul als JSON abgelegt.
"""

import re
import ast
import argparse
import json
from pathlib import Path


# --------------------------------------------------------------------------
# Konfiguration / Defaults
# --------------------------------------------------------------------------

DEFAULT_REPO = "/home/uha4/tmp/tryton"

DEFAULT_MODULES = [
    "ir",
    "res",
    "company",
    "country",
    "currency",
    "party",
    "product",
    "production_split",
    "stock_lot",
    "stock_lot_sled",
    "production",
    "stock_product_location",
    "stock_split",
    "product_measurements",
]

DEFAULT_OUTPUT = "data"

# Operationen, die find_tables() als "hier wird eine Tabelle referenziert"
# erkennt (inkl. table_exist, das selbst keine Änderung ist, aber die
# betroffene Tabelle offenlegt).
TABLE_REFERENCE_OPS = {
    "column_rename",
    "drop_column",
    "add_column",
    "drop_constraint",
    "add_constraint",
    "not_null_action",
    "index_action",
    "alter_size",
    "update",
    "insert",
    "delete",
    "select",
    "table_exist",
}

# Operationen, die extract_call() tatsächlich als DB-Operation aufnimmt.
# execute/executemany kommen hier zusätzlich dazu, table_exist bewusst nicht
# (das ist nur eine Existenzprüfung, keine Änderung).
INTERESTING_OPS = {
    "column_rename",
    "drop_column",
    "add_column",
    "drop_constraint",
    "add_constraint",
    "not_null_action",
    "index_action",
    "alter_size",
    "execute",
    "executemany",
    "update",
    "insert",
    "delete",
    "select",
}

# Aufrufe, die im "unresolved"-Log NICHT auftauchen sollen, weil sie entweder
# an anderer Stelle bereits sinnvoll aufgelöst werden (__table__,
# __table_handler__, pool.get) oder nachweislich nie eine DB-Operation sind
# (Listen-.append, Logging). Erste Fassung nach Sichtung von echten
# Tryton-Modulen (product.json) - bei Bedarf einfach erweitern.
IGNORED_UNRESOLVED_OPERATIONS = {"__table__", "__table_handler__", "append"}
IGNORED_UNRESOLVED_OBJECTS = {"logger"}
IGNORED_UNRESOLVED_CALLS = {("pool", "get")}


# --------------------------------------------------------------------------
# AST-Visitor
# --------------------------------------------------------------------------

class RegisterExtractor(ast.NodeVisitor):
    """
    Läuft über eine Python-Datei und sammelt für jede __register__-Methode
    die darin ausgeführten DB-Operationen (vor/nach dem super()-Aufruf,
    inkl. eventueller if-Bedingung).
    """

    def __init__(self, source, module, filename):
        self.module = module
        self.filename = filename
        self.source = source

        self.results = []

        # Name der Model-Klasse -> __name__ (z. B. "MyClass" -> "ir.model.data")
        self.model_names = {}

        # Innerhalb der aktuell analysierten __register__-Methode:
        # lokale Variable -> Tabellenname
        self.table_vars = {}

        # lokale Variable -> Quelltext-Ausdruck, aus dem sie stammt
        # (für die spätere Ersetzung im "code"-Feld der Operation)
        self.alias_vars = {}

        # Klasse, in der wir uns gerade befinden (für cls._table etc.)
        self.current_class = None

        # Name -> FunctionDef aller Methoden der aktuellen Klasse, um
        # Helper-Aufrufe aus __register__ heraus aufzulösen.
        self.class_methods = {}

        # Aufruf-Kette der Helper-Methoden, in die wir gerade rekursiv
        # abgestiegen sind (für Zyklenschutz und die "via"-Provenienz
        # der gefundenen Operationen).
        self.call_chain = []

    # -- Modul-/Klassenebene ------------------------------------------------

    def visit_Module(self, node):
        # Erst alle __name__-Zuweisungen einsammeln, damit sie beim
        # eigentlichen Durchlauf schon bekannt sind (Reihenfolge der
        # Klassen im File soll keine Rolle spielen).
        for item in node.body:
            if isinstance(item, ast.ClassDef):
                model_name = self.get_model_name(item)
                if model_name:
                    self.model_names[item.name] = model_name

        self.generic_visit(node)

    def visit_ClassDef(self, node):
        # Alle Methoden der Klasse vormerken, damit __register__-Aufrufe an
        # eigene Helper-Methoden (z. B. `cls._migrate_x(module_name)`)
        # aufgelöst werden können.
        methods = {
            item.name: item
            for item in node.body
            if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
        }

        for item in node.body:
            if isinstance(item, ast.FunctionDef) and item.name == "__register__":
                self.extract_register(node.name, item, methods)

        self.generic_visit(node)

    def get_model_name(self, node):
        """Liest `__name__ = "some.model"` aus einem Klassenkörper."""
        for item in node.body:
            if not isinstance(item, ast.Assign):
                continue

            for target in item.targets:
                if isinstance(target, ast.Name) and target.id == "__name__":
                    try:
                        return ast.literal_eval(item.value)
                    except Exception:
                        return None

        return None

    # -- __register__ selbst -------------------------------------------------

    def extract_register(self, cls, func, methods):
        """Analysiert eine einzelne __register__-Methode."""
        operations = []
        self.table_vars = {}
        self.alias_vars = {}
        self.current_class = cls

        # Für die Auflösung von Helper-Aufrufen (cls._migrate_x(...)) und
        # als Zyklenschutz während der Rekursion in solche Helper.
        self.class_methods = methods
        self.call_chain = []

        model_name = self.model_names.get(cls)
        if model_name:
            self.table_vars["cls"] = model_name

        # Tabellen-Variablen (z. B. `table = cls.__table__()`) vorab
        # einsammeln, damit sie beim eigentlichen Walk schon aufgelöst
        # werden können.
        for stmt in func.body:
            self.collect_table_assignment(stmt)

        self.walk_statements(func.body, operations, phase="before_super", condition=None)

        self.results.append({
            "module": self.module,
            "file": self.filename,
            "class": cls,
            "method": "__register__",
            "line": func.lineno,
            "operations": operations,
        })

    # -- Tabellen-Variablen auflösen -----------------------------------------

    def collect_table_assignment(self, stmt):
        """
        Sucht rekursiv nach Zuweisungen der Form `name = <table-ausdruck>`
        und merkt sich in self.table_vars, welche Variable welche Tabelle
        referenziert.
        """
        for node in ast.walk(stmt):
            if not isinstance(node, ast.Assign):
                continue

            if len(node.targets) != 1:
                continue

            target = node.targets[0]
            if not isinstance(target, ast.Name):
                continue

            table = self.resolve_table_call(node.value)

            if table:
                self.table_vars[target.id] = table
            elif (
                isinstance(node.value, ast.Call)
                and isinstance(node.value.func, ast.Attribute)
            ):
                table = self.resolve_table_object(node.value.func.value)
                if table:
                    self.table_vars[target.id] = table

                # Ausdruck für spätere Auflösung merken
                self.alias_vars[target.id] = ast.get_source_segment(
                    self.source, node.value
                )

            elif isinstance(node.value, ast.Constant):
                self.table_vars[target.id] = node.value.value

    def resolve_table_call(self, node):
        """
        Erkennt Aufrufe, die direkt eine Tabelle liefern:
          - pool.get('some.model')
          - X.__table__() / X.__table_handler__()
        und gibt den zugehörigen (normalisierten) Tabellennamen zurück.
        """
        if not isinstance(node, ast.Call):
            return None

        # ModelClass = pool.get('ir.model.data')
        if self._is_pool_get(node):
            return node.args[0].value

        if not (
            isinstance(node.func, ast.Attribute)
            and node.func.attr in ("__table__", "__table_handler__")
        ):
            return None

        receiver = node.func.value

        # pool.__table__() o.ä. ergibt keinen sinnvollen Tabellennamen
        if isinstance(receiver, ast.Name) and receiver.id == "pool":
            return None

        if isinstance(receiver, ast.Name):
            if receiver.id == "cls":
                model_name = self.model_names.get(self.current_class)
            else:
                model_name = (
                    self.model_names.get(receiver.id)
                    or self.table_vars.get(receiver.id)
                )

            if model_name:
                return model_name.replace(".", "_")

        return None

    @staticmethod
    def _is_pool_get(node):
        """True für Aufrufe der Form `pool.get('some.model')`."""
        return (
            isinstance(node.func, ast.Attribute)
            and node.func.attr == "get"
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id == "pool"
            and len(node.args) == 1
            and isinstance(node.args[0], ast.Constant)
            and isinstance(node.args[0].value, str)
        )

    def resolve_table_object(self, node):
        """
        Löst einen beliebigen Ausdruck (Variable, cls._table, Aufruf) zu
        einem Tabellennamen auf, falls möglich.
        """
        if isinstance(node, ast.Name):
            return self.table_vars.get(node.id)

        if (
            isinstance(node, ast.Attribute)
            and node.attr == "_table"
            and isinstance(node.value, ast.Name)
            and node.value.id == "cls"
        ):
            model_name = self.model_names.get(self.current_class)
            if model_name:
                return model_name.replace(".", "_")

        return self.resolve_table_call(node)

    def find_tables(self, node):
        """
        Sammelt alle Tabellen, die innerhalb eines Ausdrucks (typischerweise
        eines Funktionsaufrufs) referenziert werden.
        """
        tables = []

        for child in ast.walk(node):
            if not isinstance(child, ast.Call):
                continue

            if not (
                isinstance(child.func, ast.Attribute)
                and child.func.attr in TABLE_REFERENCE_OPS
            ):
                continue

            # backend.TableHandler.table_exist(cls._table)
            if self._is_table_handler_call(child, attr="table_exist") and child.args:
                table = self.resolve_table_object(child.args[0])
                if table and table not in tables:
                    tables.append(table)
                continue

            table = self.resolve_table_object(child.func.value)
            if table and table not in tables:
                tables.append(table)

        return tables

    @staticmethod
    def _is_table_handler_call(call, attr=None):
        """True für Aufrufe der Form `backend.TableHandler.<attr>(...)`."""
        is_handler = (
            isinstance(call.func.value, ast.Attribute)
            and call.func.value.attr == "TableHandler"
            and isinstance(call.func.value.value, ast.Name)
            and call.func.value.value.id == "backend"
        )
        if not is_handler:
            return False
        return attr is None or call.func.attr == attr

    # -- Statements/Ausdrücke durchlaufen ------------------------------------

    def walk_statements(self, statements, operations, phase, condition):
        """
        Läuft über eine Liste von Statements und sammelt DB-Operationen ein.
        `phase` wechselt von "before_super" auf "after_super", sobald der
        super().__register__(...)-Aufruf gefunden wird. `condition` trägt
        den Text einer umschließenden if-Bedingung mit, falls vorhanden.

        Steigt dabei in ALLE geschachtelten Blöcke ab (if/for/while/with/
        try/except/finally), nicht nur in if. Aufrufe innerhalb einer
        for-Schleife, eines with-Blocks oder eines try/except wurden vorher
        stillschweigend übersprungen, weil nur ast.If speziell behandelt
        wurde und alle anderen Compound-Statements kein `.value`-Attribut
        besitzen.
        """
        for stmt in statements:

            if self.is_super_register(stmt):
                phase = "after_super"
                continue

            if isinstance(stmt, ast.If):
                cond = ast.unparse(stmt.test)
                self.walk_statements(stmt.body, operations, phase, cond)
                self.walk_statements(stmt.orelse, operations, phase, condition)
                continue

            for body in self._nested_bodies(stmt):
                self.walk_statements(body, operations, phase, condition)

            value = getattr(stmt, "value", None)
            if value is not None:
                self.walk_expression(value, operations, phase, condition)

    @staticmethod
    def _nested_bodies(stmt):
        """
        Liefert alle Statement-Listen, die direkt in `stmt` geschachtelt
        sind (außer bei ast.If, das walk_statements separat behandelt,
        weil dort zusätzlich die Bedingung mitgeführt wird).
        """
        bodies = []

        if isinstance(stmt, (ast.For, ast.AsyncFor, ast.While)):
            bodies.append(stmt.body)
            bodies.append(stmt.orelse)
        elif isinstance(stmt, (ast.With, ast.AsyncWith)):
            bodies.append(stmt.body)
        elif isinstance(stmt, ast.Try) or (
            hasattr(ast, "TryStar") and isinstance(stmt, ast.TryStar)
        ):
            bodies.append(stmt.body)
            for handler in stmt.handlers:
                bodies.append(handler.body)
            bodies.append(stmt.orelse)
            bodies.append(stmt.finalbody)

        return bodies

    def walk_expression(self, node, operations, phase, condition):
        if node is None:
            return

        if isinstance(node, ast.Call):
            op = self.extract_call(node)
            if op:
                op["phase"] = phase
                if condition:
                    op["condition"] = condition
                if self.call_chain:
                    op["via"] = [dict(entry) for entry in self.call_chain]
                operations.append(op)
            else:
                helper = self.resolve_helper_call(node)
                if helper:
                    method_name, func_node = helper
                    self.walk_into_helper(
                        method_name, func_node, node.lineno, operations, phase, condition
                    )
                else:
                    unresolved = self.extract_unresolved_call(node)
                    if unresolved:
                        unresolved["phase"] = phase
                        if condition:
                            unresolved["condition"] = condition
                        if self.call_chain:
                            unresolved["via"] = [dict(entry) for entry in self.call_chain]
                        operations.append(unresolved)

        for child in ast.iter_child_nodes(node):
            self.walk_expression(child, operations, phase, condition)

    def extract_unresolved_call(self, call):
        """
        Rohmaterial-Eintrag für Aufrufe in __register__ (bzw. einem
        verfolgten Helper), die weder eine erkannte DB-Operation noch ein
        auflösbarer cls./self.-Helper sind. Über das Flag "unresolved": true
        von echten Operationen unterscheidbar. Bewusst noch ungefiltert
        gehalten - bis auf eine kleine, aus echten Modulen (siehe
        IGNORED_UNRESOLVED_*) abgeleitete Liste von Aufrufen, die
        nachweislich nie eine DB-Operation sind bzw. bereits an anderer
        Stelle sinnvoll aufgelöst werden. Gibt None zurück, wenn der Aufruf
        gefiltert werden soll.
        """
        if isinstance(call.func, ast.Attribute):
            operation = call.func.attr
            if isinstance(call.func.value, ast.Name):
                obj = call.func.value.id
            else:
                obj = ast.unparse(call.func.value)
        elif isinstance(call.func, ast.Name):
            operation = call.func.id
            obj = ""
        else:
            operation = ast.unparse(call.func)
            obj = ""

        if (
            operation in IGNORED_UNRESOLVED_OPERATIONS
            or obj in IGNORED_UNRESOLVED_OBJECTS
            or (obj, operation) in IGNORED_UNRESOLVED_CALLS
        ):
            return None

        args = []
        for arg in call.args:
            try:
                args.append(ast.unparse(arg))
            except Exception:
                args.append("<unknown>")

        kwargs = {}
        for kw in call.keywords:
            try:
                kwargs[kw.arg] = ast.unparse(kw.value)
            except Exception:
                kwargs[kw.arg] = "<unknown>"

        return {
            "object": obj,
            "operation": operation,
            "tables": self.find_tables(call),
            "line": call.lineno,
            "args": args,
            "kwargs": kwargs,
            "code": self._render_code(call),
            "unresolved": True,
        }

    def resolve_helper_call(self, node):
        """
        Erkennt Aufrufe der Form `cls.<method>(...)` / `self.<method>(...)`
        an eine Methode, die in derselben Klasse definiert ist (z. B. eine
        ausgelagerte Migrationsmethode). Cross-Klassen-/Cross-Datei-Aufrufe
        werden bewusst nicht aufgelöst.
        """
        if not isinstance(node.func, ast.Attribute):
            return None

        receiver = node.func.value
        if not (isinstance(receiver, ast.Name) and receiver.id in ("cls", "self")):
            return None

        method_name = node.func.attr
        if method_name == "__register__":
            return None

        func_node = self.class_methods.get(method_name)
        if func_node is None:
            return None

        return method_name, func_node

    def walk_into_helper(self, method_name, func_node, call_line, operations, phase, condition):
        """
        Steigt in eine Helper-Methode ab, als wäre ihr Körper an der
        Aufrufstelle inline eingefügt. Mit Zyklenschutz gegen (indirekt)
        rekursive Helper-Aufrufe.
        """
        if any(entry["method"] == method_name for entry in self.call_chain):
            return

        self.call_chain.append({"method": method_name, "line": call_line})

        # Tabellen-Variablen, die lokal in der Helper-Methode definiert
        # werden (z. B. `table = cls.__table__()`), vorab einsammeln.
        for stmt in func_node.body:
            self.collect_table_assignment(stmt)

        self.walk_statements(func_node.body, operations, phase, condition)

        self.call_chain.pop()

    def is_super_register(self, stmt):
        """True für `super(...).__register__(...)` als eigenständiges Statement."""
        if not isinstance(stmt, ast.Expr):
            return False

        call = stmt.value
        if not isinstance(call, ast.Call):
            return False

        func = call.func
        if not isinstance(func, ast.Attribute) or func.attr != "__register__":
            return False

        if not isinstance(func.value, ast.Call):
            return False

        if not isinstance(func.value.func, ast.Name):
            return False

        return func.value.func.id == "super"

    # -- Einzelnen Aufruf in ein Operation-Dict übersetzen -------------------

    def extract_call(self, call):
        if not isinstance(call.func, ast.Attribute):
            return None

        operation = call.func.attr

        is_table_handler = self._is_table_handler_call(call)

        if operation not in INTERESTING_OPS and not is_table_handler:
            return None

        tables = self.find_tables(call)
        tables += self._extra_tables_from_table_rename(call, tables)

        # Wer wird aufgerufen?
        if isinstance(call.func.value, ast.Name):
            obj = call.func.value.id
        else:
            obj = ast.unparse(call.func.value)

        args = []
        for arg in call.args:
            try:
                args.append(ast.unparse(arg))
            except Exception:
                args.append("<unknown>")

        kwargs = {}
        for kw in call.keywords:
            try:
                kwargs[kw.arg] = ast.unparse(kw.value)
            except Exception:
                kwargs[kw.arg] = "<unknown>"

        code = self._render_code(call)

        return {
            "object": obj,
            "operation": operation,
            "tables": tables,
            "line": call.lineno,
            "args": args,
            "kwargs": kwargs,
            "code": code,
        }

    def _extra_tables_from_table_rename(self, call, existing_tables):
        """
        Sonderfall `backend.TableHandler.table_rename(old, new)`: die ersten
        beiden Argumente können selbst Tabellen(-namen) sein und werden hier
        zusätzlich zu den bereits gefundenen Tabellen ergänzt.
        """
        is_table_rename = (
            call.func.attr == "table_rename"
            and isinstance(call.func.value, ast.Attribute)
            and isinstance(call.func.value.value, ast.Name)
            and call.func.value.value.id == "backend"
            and isinstance(call.args, list)
        )
        if not is_table_rename:
            return []

        extra = []
        for arg in call.args[:2]:
            if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                table = arg.value
            elif isinstance(arg, ast.Name):
                table = self.table_vars.get(arg.id)
            elif (
                isinstance(arg, ast.Attribute)
                and isinstance(arg.value, ast.Name)
                and arg.value.id == "cls"
                and arg.attr == "_table"
            ):
                table = self.model_names.get(self.current_class)
                if table:
                    table = table.replace(".", "_")
            else:
                table = None

            if table and table not in existing_tables and table not in extra:
                extra.append(table)

        return extra

    def _render_code(self, call):
        """
        Quelltext des Aufrufs, wobei zuvor gemerkte Alias-Variablen
        (siehe alias_vars) durch ihren ursprünglichen Ausdruck ersetzt
        werden, damit der Code auch ohne Kontext lesbar ist.
        """
        code = ast.get_source_segment(self.source, call)

        for name, value in self.alias_vars.items():
            code = re.sub(
                rf"\b{re.escape(name)}\b",
                f"({value})",
                code,
            )

        return code


# --------------------------------------------------------------------------
# Datei-/Repo-Ebene
# --------------------------------------------------------------------------

def analyse_file(module, modulepath, filename):
    """Parst eine einzelne Datei und liefert alle __register__-Ergebnisse mit Operationen."""
    path = modulepath / filename
    try:
        source = path.read_text(encoding="utf-8")
        tree = ast.parse(source)
    except Exception as e:
        print(f"Cannot parse {path}: {e}")
        return []

    visitor = RegisterExtractor(source, module, filename)
    visitor.visit(tree)

    return [r for r in visitor.results if r["operations"]]


def detect_version(repo: Path) -> str:
    """Liest die aktuelle Version aus trytond/CHANGELOG."""
    changelog = repo / "trytond" / "CHANGELOG"

    if not changelog.exists():
        return "unknown"

    with changelog.open("r", encoding="utf-8") as f:
        for line in f:
            if line.startswith("Version "):
                return line.split()[1].rsplit(".", 1)[0]

    return "unknown"


def find_module(repo: Path, module: str) -> Path | None:
    """Sucht ein Modul entweder unter modules/ oder trytond/trytond/."""
    candidates = [
        repo / "modules" / module,
        repo / "trytond" / "trytond" / module,
    ]

    for candidate in candidates:
        if candidate.exists():
            return candidate

    return None


def analyse_module(repo: Path, module: str) -> list | None:
    """Analysiert alle .py-Dateien eines Moduls. None, falls Modul nicht gefunden."""
    module_path = find_module(repo, module)
    if not module_path:
        return None

    results = []
    for py in sorted(module_path.rglob("*.py")):
        filename = str(py.relative_to(module_path))
        results.extend(analyse_file(module, module_path, filename))

    return results


def write_results(output: Path, version: str, module: str, results: list) -> Path:
    """Schreibt die Ergebnisse eines Moduls als JSON und gibt den Pfad zurück."""
    outdir = output / version
    outdir.mkdir(parents=True, exist_ok=True)

    outfile = outdir / f"{module}.json"
    with outfile.open("w", encoding="utf-8") as f:
        json.dump(
            {
                "version": version,
                "module": module,
                "files": results,
            },
            f,
            indent=2,
        )

    return outfile


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

def parse_args():
    parser = argparse.ArgumentParser()

    parser.add_argument("--repo", help="Path to Tryton repository")
    parser.add_argument("--modules", nargs="+", help="Module names")
    parser.add_argument("--output", help="Output directory")

    return parser.parse_args()


def main():
    args = parse_args()

    repo = Path(args.repo or DEFAULT_REPO)
    modules = args.modules or DEFAULT_MODULES
    output = Path(args.output or DEFAULT_OUTPUT)

    version = detect_version(repo)

    print(f"Repository : {repo}")
    print(f"Version    : {version}")
    print(f"Modules    : {', '.join(modules)}")
    print()

    for module in modules:
        results = analyse_module(repo, module)

        if results is None:
            print(f"Module not found: {module}")
            continue

        outfile = write_results(output, version, module, results)
        print(f"{module:25} -> {outfile}")


if __name__ == "__main__":
    main()