Schedule Duplicate Invoice Checking
In [ ]:
Copied!
# 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 [ ]:
Copied!
# 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 [ ]:
Copied!
# Connect to Celonis
celonis = get_celonis(threaded=True)
# Connect to Celonis
celonis = get_celonis(threaded=True)
In [ ]:
Copied!
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 [ ]:
Copied!
datamodel
datamodel
In [ ]:
Copied!
config
config
In [ ]:
Copied!
# 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 [ ]:
Copied!
# 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()