按区块计算

from blockx import (
    SOURCE_ROW, BlockTableCallConfig, BlockTableWriteHandler,
    Function, TaskBuilder,
)

source_code = """
def _(chain_id, record):
    value = int(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None
    return {
        "id": record["id"],
        "chain_id": chain_id,
        "tx_id": record["tx_id"],
        "from_addr": record["from_addr"],
        "to_addr": record["to_addr"],
        "value": value,
    }
"""


def submit_block(block, source_table, target_table):
    task = TaskBuilder.build(
        BlockTableCallConfig(block=block, triggerSources=[{
            "table": source_table,
            "func": Function(source_code=source_code),
            "params": ["eth", SOURCE_ROW],
        }]),
        BlockTableWriteHandler(targetTable=target_table, block=block),
    )
    result = task.submit(timeout=120)
    if not result.task_result.success:
        raise RuntimeError(result.task_result.failure_code)
    return task.task_id, result.handler_result

from blockx import NormalTableWriteHandler


def increasing_value_handler(target_table):
    return NormalTableWriteHandler(
        targetTable=target_table,
        condition=("value", "UpdateIfLarger"),
    )