Rossum.ai: STORE and RESTORE datapoint values from annotation metadata so that they survive annotation re-extraction.

RESTORE.py

import requests
from txscript import TxScript, default_to, substitute


# Configuration: Map { "Metadata_Source_ID": "Target_Field_ID" }
FIELD_MAPPING = {
    "document_type_calculated": "document_type_calculated_memorized",
    "approval_number": "approval_number",
}


# Restore values from the metadata (run on "initialize" event):
def rossum_hook_request_handler(payload):
    x = TxScript.from_payload(payload)

    memorized_datapoints = x.annotation.metadata.get("memorized_datapoints", {})

    for source_id, target_id in FIELD_MAPPING.items():
        memorized_data = memorized_datapoints.get(source_id)
        if memorized_data:
            target_field = getattr(x.field, target_id)
            target_field.attr.page = memorized_data.get("attr", {}).get("page")
            target_field.attr.position = memorized_data.get("attr", {}).get("position")
            target_field.attr.value = memorized_data.get("value")

    return x.hook_response()

STORE.py

import requests
from txscript import TxScript, default_to, substitute


FIELD_IDS = ["document_type_calculated", "approval_number"]


# Store values to the metadata (run on "initialize", "started", and "updated" event):
def rossum_hook_request_handler(payload):
    x = TxScript.from_payload(payload)

    memorize_payload = {}

    for field_id in FIELD_IDS:
        field = getattr(x.field, field_id)
        memorize_payload[field.attr.schema_id] = {
            "value": field.value,
            "attr": {
                "page": field.attr.page,
                "position": field.attr.position,
            }
        }

    print(f"Payload to memorize: {memorize_payload}")

    current_metadata = x.annotation.metadata
    current_metadata['memorized_datapoints'] = memorize_payload
    response = requests.patch(
        f"{payload.get('base_url')}/api/v1/annotations/{x.annotation.id}",
        json={
            "metadata": current_metadata
        },
        headers={
            "Authorization": f"Bearer {payload.get('rossum_authorization_token')}",
            "Content-Type": "application/json"
        }
    )

    response.raise_for_status()

    return x.hook_response()