tryton-db-migration/extract.py

#!/usr/bin/env python3

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

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

DEFAULT_MODULES = [
    "party",
]

DEFAULT_OUTPUT = "data"

class RegisterExtractor(ast.NodeVisitor):

    def __init__(self, filename):
        self.filename = filename
        self.results = []

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

        self.generic_visit(node)

    def extract_register(self, cls, func):

        phase = "before_super"

        operations = []

        for stmt in func.body:

            # Wurde super().__register__() aufgerufen?
            if self.is_super_register(stmt):
                phase = "after_super"
                continue

            for call in ast.walk(stmt):
                if isinstance(call, ast.Call):

                    op = self.extract_call(call)

                    if op:
                        op["phase"] = phase
                        operations.append(op)

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

    def is_super_register(self, stmt):

        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):
            return False

        if 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"

    def extract_call(self, call):
    
        if not isinstance(call.func, ast.Attribute):
            return None
    
        interesting = {
            "column_rename",
            "drop_column",
            "add_column",
            "drop_constraint",
            "add_constraint",
            "not_null_action",
            "index_action",
            "alter_size",
            "execute",
            "executemany",
            "update",
            "insert",
            "delete",
            "select",
        }
    
        operation = call.func.attr
    
        if operation not in interesting:
            return None
    
        # 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>"
    
        return {
            "object": obj,
            "operation": operation,
            "line": call.lineno,
            "args": args,
            "kwargs": kwargs,
            "code": ast.unparse(call),
        }

def analyse_file(path):

    try:
        tree = ast.parse(path.read_text(encoding="utf-8"))
    except Exception as e:
        print(f"Cannot parse {path}: {e}")
        return []

    visitor = RegisterExtractor(str(path))
    visitor.visit(tree)

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

def detect_version(repo: Path) -> str:
    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]

    return "unknown"

def find_module(repo: Path, module: str) -> Path | None:

    candidates = [
        repo / "modules" / module,
        repo / "trytond" / "modules" / module,
    ]

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

    return None

def main():

    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",
    )

    args = parser.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:

        module_path = repo / "modules" / module

        if not module_path.exists():
            print(f"Module not found: {module}")
            continue

        results = []

        for py in sorted(module_path.rglob("*.py")):
            results.extend(analyse_file(py))

        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,
            )

        print(f"{module:25} -> {outfile}")


if __name__ == "__main__":
    main()