Hallo @mspooren, tatsächlich ist keine API-Funktion dokumentiert, die dies umsetzen kann, und entsprechend bzw. gleichzeitig kann es der enaio-Client auch nicht.
Wenn die Schränke wirklich 1:1 die gleichen Felder haben (interne Namen), könnte man ein Matching relativ leicht mit bestehenden API-Funktionen in einem kleinen Projekt aufsetzen. Das wäre aber immer eine Eigenentwicklung, die zu testen und für den Eigengebrauch mit Vorsicht zu nutzen wäre.
Eine minimale Implementation ohne Oberfläche (die also ggf. zu ergänzen wäre) würde im ecmind_blue_client etwa so aussehen:
# /// script
# requires-python = ">=3.12"
# dependencies = ["ecmind-blue-client==1.0.1"]
# ///
"""Konzept-Demo: kopiert einen Ordner samt Registern und Dokumenten in einen anderen Schrank.
Aufruf: uv run copy_folder_across_cabinets.py <server> <port> <login> <passwort> <quell-ordner-id> <ziel-schrank>
Objekttypen paart das Skript nach Art (Ordner, Register, Dokument) und der größten
Überschneidung interner Feldnamen, Feldwerte übernimmt die Kopie bei exakt gleichem
internen Namen. Nicht kopiert werden Tabellenfelder, Notizen, Rechte, Varianten und
Historie, und die Kopien bekommen neue IDs. Als Demo gedacht, nicht als Migration.
"""
import argparse
from pathlib import Path
from tempfile import TemporaryDirectory
from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.ecm.definition import ECMMainType
from ecmind_blue_client.pool import SyncPoolClient
from ecmind_blue_client.rpc import JobRequestFileFromPath
STRUCTURE = (ECMMainType.FOLDER, ECMMainType.REGISTER)
MAXHITS = 1000 # zählt die Kindobjekte mit, nicht nur die Treffer der Abfrage
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("server")
ap.add_argument("port", type=int)
ap.add_argument("login")
ap.add_argument("password")
ap.add_argument("source_folder_id", type=int)
ap.add_argument("target_cabinet")
args = ap.parse_args()
ecm = ECM(SyncPoolClient(servers=f"{args.server}:{args.port}:1", username=args.login, password=args.password))
definition = ecm.system.definition()
target_cabinet = definition.cabinets[args.target_cabinet]
root_type = definition[ecm.dms.get_object_type_by_id(args.source_folder_id)]
source_cabinet = next(c for c in definition.cabinets.values() if root_type.internal_name in c)
def target_type(source_type):
"""Zieltyp derselben Art mit den meisten gleichnamigen Feldern."""
candidates = {
ECMMainType.FOLDER: target_cabinet.folders,
ECMMainType.REGISTER: target_cabinet.registers,
}.get(source_type.maintype, target_cabinet.documents)
return max(candidates.values(), key=lambda t: len(set(source_type.all_fields) & set(t.all_fields)))
def copy(source, source_type, folder_id, register_id, tmp):
dst_type = target_type(source_type)
model = ecm.dms.model_by_name(dst_type.internal_name)
values = {name: source[name] for name in dst_type.all_fields if source[name] is not None}
files = None
if source_type.maintype not in STRUCTURE:
directory = Path(tmp, str(source.system.id))
directory.mkdir()
files = [JobRequestFileFromPath(f.store(str(directory / f.name))) for f in ecm.dms.files(source.system.id)]
new_id, _ = ecm.dms.insert(model(**values), files, folder_id=folder_id, register_id=register_id, check_mandatory=False)
print(f"{source_type.internal_name} {source.system.id} -> {dst_type.internal_name} {new_id} ({len(values)} Felder)")
if source_type.maintype not in STRUCTURE:
return
source_model = ecm.dms.model_by_name(source_type.internal_name)
tree = ecm.dms.select(source_model).where(source_model.system.id == source.system.id).with_children().limit(MAXHITS).execute()
is_folder = source_type.maintype is ECMMainType.FOLDER
for child_type in source_cabinet.values():
for child in tree[0].children_of(child_type.internal_name):
copy(child, child_type, new_id if is_folder else folder_id, None if is_folder else new_id, tmp)
with TemporaryDirectory() as tmp:
copy(ecm.dms.get(ecm.dms.model_by_name(root_type.internal_name), args.source_folder_id), root_type, None, None, tmp)
if __name__ == "__main__":
main()