Schedule Duplicate Invoice Checking

In [ ]:

# When scheduled from a knowledge model these two parameters are given from the object with type:'ml-job'
# E.g.: 
#layerId = "simon.test-model"
#objectId = "duplicate-invoice-checker"

# When scheduled from a knowledge model these two parameters are given from the object with type:'ml-job' # E.g.: #layerId = "simon.test-model" #objectId = "duplicate-invoice-checker"

In [ ]:

# Module Imports
from pycelonis import get_celonis
from pycelonis.data_deduplication.duplicate_invoice_checker import DuplicateInvoiceChecker
from pycelonis.data_deduplication.utils import get_config_from_trigger
from pycelonis.celonis_api.studio.schedule_utils import get_km_dict_by_key
import yaml

# Module Imports from pycelonis import get_celonis from pycelonis.data_deduplication.duplicate_invoice_checker import DuplicateInvoiceChecker from pycelonis.data_deduplication.utils import get_config_from_trigger from pycelonis.celonis_api.studio.schedule_utils import get_km_dict_by_key import yaml

In [ ]:

# Connect to Celonis
celonis = get_celonis(threaded=True)

# Connect to Celonis celonis = get_celonis(threaded=True)

In [ ]:

if "layerId" in locals():
    km = get_km_dict_by_key(**locals())
    config, datamodel = get_config_from_trigger(**locals())
else:
    with open(r"custom_object_config.yaml", "r") as yfile:
        custom_object = yaml.load(yfile, Loader=yaml.FullLoader)
    config = custom_object["customAttributes"]
    datamodel = celonis.datamodels.find(config["data_extraction_config"]["datamodel_id"])

if "layerId" in locals(): km = get_km_dict_by_key(**locals()) config, datamodel = get_config_from_trigger(**locals()) else: with open(r"custom_object_config.yaml", "r") as yfile: custom_object = yaml.load(yfile, Loader=yaml.FullLoader) config = custom_object["customAttributes"] datamodel = celonis.datamodels.find(config["data_extraction_config"]["datamodel_id"])

In [ ]:

datamodel

datamodel

In [ ]:

config

config

In [ ]:

# Create Duplicate Invoice Checker from config
dc_invoice = DuplicateInvoiceChecker(
    datamodel=datamodel,
    **config
)

# Create Duplicate Invoice Checker from config dc_invoice = DuplicateInvoiceChecker( datamodel=datamodel, **config )

In [ ]:

# Run duplicate checking and push results to pool.
# set reload to True if you want a partial reload upon every push of duplicates
response = dc_invoice.apply_pipeline()

# Run duplicate checking and push results to pool. # set reload to True if you want a partial reload upon every push of duplicates response = dc_invoice.apply_pipeline()