Make Array

1import json2import os3import shutil4import sys56INPUT_DIR = "/data/input"7OUTPUT_DIR = "/data/output"8910def _looks_like_file_value(value):11    """Cheap filename heuristic — used only to decide whether to probe /data/input.1213    The authoritative signal is os.path.exists(INPUT_DIR/<value>): we will not14    copy anything we cannot actually find on disk.15    """16    if not isinstance(value, str) or not value:17        return False18    if "/" in value or "\\" in value or value.startswith("."):19        return False20    return "." in value212223def main():24    try:25        execution_input = json.loads(sys.stdin.read())26        inputs = execution_input.get("inputs", {})27        input_schema = execution_input.get("inputSchema", {})2829        schema_props = input_schema.get("properties", {})3031        def sort_key(k):32            prop = schema_props.get(k, {})33            if "order" in prop:34                return prop["order"]35            try:36                return int(k.split("_")[1])37            except (ValueError, IndexError):38                return float("inf")3940        item_keys = sorted(41            [k for k in inputs if k.startswith("item_")],42            key=sort_key,43        )4445        items = []46        files_passed_through = []47        for key in item_keys:48            value = inputs.get(key)49            if value is None or value == "":50                continue51            items.append(value)52            if _looks_like_file_value(value):53                staged_path = os.path.join(INPUT_DIR, value)54                if os.path.isfile(staged_path):55                    files_passed_through.append(value)5657        # When any item resolves to an actually-staged file in /data/input,58        # copy it to /data/output. The Docker executor uploads everything in59        # /data/output to MinIO under this node's (node_id, execution_id),60        # which is the key downstream edge resolvers reconstruct.61        if files_passed_through:62            os.makedirs(OUTPUT_DIR, exist_ok=True)63            for filename in files_passed_through:64                shutil.copy(65                    os.path.join(INPUT_DIR, filename),66                    os.path.join(OUTPUT_DIR, filename),67                )68            print(69                "make_array: passed through {} file(s)".format(len(files_passed_through)),70                file=sys.stderr,71            )7273        print(74            "make_array: collected {} item(s) from {} port(s)".format(len(items), len(item_keys)),75            file=sys.stderr,76        )7778        output = {"items": items}79        # Companion key consumed by the edge resolver to override the80        # statically-Any source_port_type. "File" is enough — is_file_type()81        # treats it as a file family and the resolver will download each82        # item from MinIO using this node's (source_node_id, src_exec_id).83        if files_passed_through:84            output["_items_port_type"] = "File"8586        print(json.dumps(output))8788    except Exception as e:89        import traceback90        error_output = {91            "error": str(e),92            "errorType": type(e).__name__,93            "traceback": traceback.format_exc(),94        }95        print(json.dumps(error_output), file=sys.stderr)96        sys.exit(1)979899if __name__ == "__main__":100    main()