318 lines
11 KiB
Python
318 lines
11 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
rst_to_dataset.py
|
||
|
|
|
||
|
|
Extrahiert aus Tryton-Modul-Testdateien (tests/scenario_*.rst - das sind
|
||
|
|
ausführbare Doctests mit Proteus-Code in >>>-Zeilen) automatisch Record-
|
||
|
|
Definitionen im dataset.yaml-Format von create_dataset.py.
|
||
|
|
|
||
|
|
Erkennt:
|
||
|
|
>>> Party = Model.get('party.party') # Modell-Zuordnung
|
||
|
|
>>> party = Party(name='Testfirma GmbH') # Instanziierung + Felder
|
||
|
|
>>> party.vat_number = '123' # weitere Feld-Zuweisung
|
||
|
|
>>> party.save() # Abschluss -> Record wird übernommen
|
||
|
|
|
||
|
|
sowie Referenzen auf andere in derselben Datei bereits angelegte Records
|
||
|
|
(z. B. `address.party = party`) -> werden zu {"$ref": "..."}, und die
|
||
|
|
gängigsten nicht-JSON-tauglichen Werte:
|
||
|
|
Decimal('12.34') -> {"$decimal": "12.34"}
|
||
|
|
datetime.date(2024, 1, 1) -> {"$date": "2024-01-01"}
|
||
|
|
|
||
|
|
Alles andere (Wizards, .click(...), Reports, One2Many .new()/.append(),
|
||
|
|
Aufrufe von Hilfsfunktionen wie create_company()) wird NICHT automatisch
|
||
|
|
aufgelöst, sondern als {"$manual": "<Original-Ausdruck>"} markiert bzw.
|
||
|
|
komplett übersprungen und in der Zusammenfassung aufgelistet - das ist
|
||
|
|
bewusst so: lieber sichtbar unvollständig als still falsch.
|
||
|
|
|
||
|
|
Nutzung:
|
||
|
|
python3 rst_to_dataset.py tests/scenario_party.rst tests/scenario_address.rst \\
|
||
|
|
--output-dir extracted/
|
||
|
|
|
||
|
|
Erzeugt extracted/scenario_party.yaml, extracted/scenario_address.yaml.
|
||
|
|
Record-Keys werden mit dem Dateinamen präfixiert (party__party_1 statt
|
||
|
|
nur party_1), damit beim manuellen Zusammenführen mehrerer Module keine
|
||
|
|
Kollisionen entstehen.
|
||
|
|
|
||
|
|
WICHTIG: Ungetestet gegen echte .rst-Dateien (hier ohne Netzwerkzugriff auf
|
||
|
|
das Tryton-Repo) - nur gegen eine selbstgebaute Beispieldatei verifiziert
|
||
|
|
(siehe Testlauf in der Antwort). Bitte an echten Dateien gegenprüfen, bevor
|
||
|
|
du dich auf die Ausgabe verlässt - der "manual"-Anteil wird bei komplexeren
|
||
|
|
Szenarien vermutlich größer sein als im Testfall.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import ast
|
||
|
|
import re
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
try:
|
||
|
|
import yaml
|
||
|
|
except ImportError:
|
||
|
|
yaml = None
|
||
|
|
|
||
|
|
|
||
|
|
PROMPT_RE = re.compile(r'^\s*>>>\s?(.*)$')
|
||
|
|
CONTINUATION_RE = re.compile(r'^\s*\.\.\.\s?(.*)$')
|
||
|
|
|
||
|
|
|
||
|
|
def extract_doctest_source(text):
|
||
|
|
"""Zieht alle >>>/...-Zeilen aus einer .rst-Datei und reiht sie zu einem
|
||
|
|
zusammenhängenden Python-Quelltext aneinander. Alles andere (Prosa,
|
||
|
|
erwartete Ausgabe) wird ignoriert."""
|
||
|
|
lines = []
|
||
|
|
for raw_line in text.splitlines():
|
||
|
|
m = PROMPT_RE.match(raw_line)
|
||
|
|
if m:
|
||
|
|
lines.append(m.group(1))
|
||
|
|
continue
|
||
|
|
m = CONTINUATION_RE.match(raw_line)
|
||
|
|
if m:
|
||
|
|
lines.append(m.group(1))
|
||
|
|
continue
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
def is_call_to(node, *dotted_names):
|
||
|
|
"""True, wenn node ein Call ist, dessen func-Ausdruck (als unparse-Text)
|
||
|
|
zu einem der übergebenen Namen passt, z. B. 'Model.get' oder 'Decimal'."""
|
||
|
|
if not isinstance(node, ast.Call):
|
||
|
|
return False
|
||
|
|
try:
|
||
|
|
text = ast.unparse(node.func)
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
return text in dotted_names
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_model_get(node):
|
||
|
|
"""Liefert den Modellnamen aus einem Model.get('x.y')-Aufruf, sonst None."""
|
||
|
|
if not is_call_to(node, "Model.get"):
|
||
|
|
return None
|
||
|
|
if len(node.args) != 1 or not isinstance(node.args[0], ast.Constant):
|
||
|
|
return None
|
||
|
|
if not isinstance(node.args[0].value, str):
|
||
|
|
return None
|
||
|
|
return node.args[0].value
|
||
|
|
|
||
|
|
|
||
|
|
class ScenarioParser:
|
||
|
|
def __init__(self):
|
||
|
|
self.model_vars = {} # Variable -> Modellname (aus Model.get(...))
|
||
|
|
self.record_vars = {} # Variable -> Record-Dict (key/model/fields)
|
||
|
|
self.finalized = set() # Variablen, deren erster .save()-Aufruf schon verarbeitet wurde
|
||
|
|
self.output = [] # fertige Record-Dicts in Reihenfolge
|
||
|
|
self.manual_notes = [] # Zeilen, die nicht automatisch verstanden wurden
|
||
|
|
|
||
|
|
def convert_value(self, node):
|
||
|
|
if isinstance(node, ast.Constant):
|
||
|
|
return node.value
|
||
|
|
|
||
|
|
if isinstance(node, ast.Name):
|
||
|
|
if node.id in self.record_vars:
|
||
|
|
return {"$ref": self.record_vars[node.id]["key"]}
|
||
|
|
return self._manual(node)
|
||
|
|
|
||
|
|
if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
|
||
|
|
inner = self.convert_value(node.operand)
|
||
|
|
if isinstance(inner, (int, float)):
|
||
|
|
return -inner
|
||
|
|
return self._manual(node)
|
||
|
|
|
||
|
|
if isinstance(node, ast.Call):
|
||
|
|
if is_call_to(node, "Decimal") and node.args and isinstance(node.args[0], ast.Constant):
|
||
|
|
return {"$decimal": str(node.args[0].value)}
|
||
|
|
|
||
|
|
if is_call_to(node, "datetime.date", "date") and len(node.args) >= 3:
|
||
|
|
try:
|
||
|
|
y, m, d = (a.value for a in node.args[:3])
|
||
|
|
return {"$date": f"{int(y):04d}-{int(m):02d}-{int(d):02d}"}
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
return self._manual(node)
|
||
|
|
|
||
|
|
if isinstance(node, (ast.List, ast.Tuple)):
|
||
|
|
return [self.convert_value(el) for el in node.elts]
|
||
|
|
|
||
|
|
return self._manual(node)
|
||
|
|
|
||
|
|
def _manual(self, node):
|
||
|
|
try:
|
||
|
|
text = ast.unparse(node)
|
||
|
|
except Exception:
|
||
|
|
text = "<nicht darstellbar>"
|
||
|
|
self.manual_notes.append(f"Zeile {getattr(node, 'lineno', '?')}: {text}")
|
||
|
|
return {"$manual": text}
|
||
|
|
|
||
|
|
def make_key(self, var_name):
|
||
|
|
base = var_name
|
||
|
|
key = base
|
||
|
|
i = 2
|
||
|
|
existing_keys = {r["key"] for r in self.output} | {
|
||
|
|
r["key"] for r in self.record_vars.values()
|
||
|
|
}
|
||
|
|
while key in existing_keys:
|
||
|
|
key = f"{base}_{i}"
|
||
|
|
i += 1
|
||
|
|
return key
|
||
|
|
|
||
|
|
def handle_assign(self, node):
|
||
|
|
if len(node.targets) != 1:
|
||
|
|
self._manual(node)
|
||
|
|
return
|
||
|
|
target = node.targets[0]
|
||
|
|
|
||
|
|
# X = Model.get('some.model')
|
||
|
|
if isinstance(target, ast.Name):
|
||
|
|
model_name = resolve_model_get(node.value)
|
||
|
|
if model_name:
|
||
|
|
self.model_vars[target.id] = model_name
|
||
|
|
return
|
||
|
|
|
||
|
|
# x = SomeModelVar(...) ODER x = Model.get('x.y')(...)
|
||
|
|
model_name = None
|
||
|
|
kwargs = {}
|
||
|
|
if isinstance(node.value, ast.Call):
|
||
|
|
func = node.value.func
|
||
|
|
if isinstance(func, ast.Name) and func.id in self.model_vars:
|
||
|
|
model_name = self.model_vars[func.id]
|
||
|
|
else:
|
||
|
|
model_name = resolve_model_get(func)
|
||
|
|
|
||
|
|
if model_name:
|
||
|
|
for kw in node.value.keywords:
|
||
|
|
if kw.arg:
|
||
|
|
kwargs[kw.arg] = self.convert_value(kw.value)
|
||
|
|
|
||
|
|
if model_name:
|
||
|
|
key = self.make_key(target.id)
|
||
|
|
self.record_vars[target.id] = {
|
||
|
|
"key": key,
|
||
|
|
"model": model_name,
|
||
|
|
"fields": kwargs,
|
||
|
|
}
|
||
|
|
return
|
||
|
|
|
||
|
|
# x = <irgendwas anderes>, z. B. Ergebnis einer Helper-Funktion
|
||
|
|
# (create_company(), Party.find([...])[0], ...) - können wir
|
||
|
|
# nicht auflösen.
|
||
|
|
self.manual_notes.append(
|
||
|
|
f"Zeile {node.lineno}: '{target.id} = {ast.unparse(node.value)}' "
|
||
|
|
f"nicht automatisch aufgelöst (kein erkanntes Modell/Konstruktor)."
|
||
|
|
)
|
||
|
|
return
|
||
|
|
|
||
|
|
# x.feld = value
|
||
|
|
if isinstance(target, ast.Attribute) and isinstance(target.value, ast.Name):
|
||
|
|
var = target.value.id
|
||
|
|
if var in self.record_vars:
|
||
|
|
self.record_vars[var]["fields"][target.attr] = self.convert_value(node.value)
|
||
|
|
return
|
||
|
|
|
||
|
|
self._manual(node)
|
||
|
|
|
||
|
|
def handle_call_statement(self, node):
|
||
|
|
call = node.value
|
||
|
|
if not isinstance(call, ast.Call) or not isinstance(call.func, ast.Attribute):
|
||
|
|
return
|
||
|
|
if call.func.attr != "save":
|
||
|
|
return
|
||
|
|
if not isinstance(call.func.value, ast.Name):
|
||
|
|
return
|
||
|
|
|
||
|
|
var = call.func.value.id
|
||
|
|
if var in self.record_vars and var not in self.finalized:
|
||
|
|
self.output.append(dict(self.record_vars[var]))
|
||
|
|
self.finalized.add(var)
|
||
|
|
|
||
|
|
def parse(self, source):
|
||
|
|
try:
|
||
|
|
tree = ast.parse(source)
|
||
|
|
except SyntaxError as e:
|
||
|
|
raise ValueError(f"Konnte extrahierten Doctest-Code nicht parsen: {e}") from e
|
||
|
|
|
||
|
|
for stmt in tree.body:
|
||
|
|
if isinstance(stmt, ast.Assign):
|
||
|
|
self.handle_assign(stmt)
|
||
|
|
elif isinstance(stmt, ast.Expr):
|
||
|
|
self.handle_call_statement(stmt)
|
||
|
|
# alles andere (if/for/import/...) wird bewusst ignoriert
|
||
|
|
|
||
|
|
|
||
|
|
def namespace_keys(records, prefix):
|
||
|
|
"""Schreibt allen Record-Keys (und $ref-Verweisen darauf) ein Präfix
|
||
|
|
voran, damit beim manuellen Zusammenführen mehrerer Module-YAMLs keine
|
||
|
|
Kollisionen entstehen."""
|
||
|
|
rename = {r["key"]: f"{prefix}__{r['key']}" for r in records}
|
||
|
|
|
||
|
|
def rewrite(value):
|
||
|
|
if isinstance(value, dict):
|
||
|
|
if "$ref" in value and value["$ref"] in rename:
|
||
|
|
return {"$ref": rename[value["$ref"]]}
|
||
|
|
return {k: rewrite(v) for k, v in value.items()}
|
||
|
|
if isinstance(value, list):
|
||
|
|
return [rewrite(v) for v in value]
|
||
|
|
return value
|
||
|
|
|
||
|
|
result = []
|
||
|
|
for r in records:
|
||
|
|
result.append({
|
||
|
|
"key": rename[r["key"]],
|
||
|
|
"model": r["model"],
|
||
|
|
"fields": {k: rewrite(v) for k, v in r["fields"].items()},
|
||
|
|
})
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def process_file(path, output_dir):
|
||
|
|
text = Path(path).read_text(encoding="utf-8")
|
||
|
|
source = extract_doctest_source(text)
|
||
|
|
|
||
|
|
parser = ScenarioParser()
|
||
|
|
parser.parse(source)
|
||
|
|
|
||
|
|
prefix = Path(path).stem
|
||
|
|
records = namespace_keys(parser.output, prefix)
|
||
|
|
|
||
|
|
out_path = Path(output_dir) / f"{prefix}.yaml"
|
||
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
doc = {"records": records}
|
||
|
|
if yaml is not None:
|
||
|
|
out_path.write_text(
|
||
|
|
yaml.dump(doc, allow_unicode=True, sort_keys=False, default_flow_style=False),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
import json
|
||
|
|
out_path.write_text(json.dumps(doc, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
|
|
|
||
|
|
return out_path, len(records), parser.manual_notes
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
|
|
parser.add_argument("rst_files", nargs="+", help="Ein oder mehrere scenario_*.rst-Dateien")
|
||
|
|
parser.add_argument("--output-dir", default="extracted", help="Zielordner für die YAMLs")
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
for rst_file in args.rst_files:
|
||
|
|
try:
|
||
|
|
out_path, count, notes = process_file(rst_file, args.output_dir)
|
||
|
|
except ValueError as e:
|
||
|
|
print(f"{rst_file}: FEHLER - {e}", file=sys.stderr)
|
||
|
|
continue
|
||
|
|
|
||
|
|
print(f"{rst_file} -> {out_path} ({count} Record(s) erkannt)")
|
||
|
|
if notes:
|
||
|
|
print(f" {len(notes)} Ausdruck/Ausdrücke nicht automatisch aufgelöst (siehe $manual im YAML):")
|
||
|
|
for note in notes[:10]:
|
||
|
|
print(f" - {note}")
|
||
|
|
if len(notes) > 10:
|
||
|
|
print(f" ... und {len(notes) - 10} weitere")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|