^{Registro 1: 23/07/2025 - Capturando MEMOS desde Python
Versión de Python: 3.10
Modulo para interactuar: beem
Asistente AI (para programación): Microsoft Copilot
Objetivo: Lograr capturar Memos de transferencias de forma instantánea.}

Todos los proyectos, grandes o pequeños, famosos o no, tienen algo en común: Todos comienzan con una idea, hace aproximadamente dos años una idea vino a mi mente, para entonces, dados mis limitados conocimientos en programación de aquel tiempo, solo pude escribir un artículo deseando que algún desarrollador pudiese captar la idea y hacerla realidad, el artículo [¿Es posible un Messenger para STEEM?](https://steemit.com/colombiaoriginal/@jesusjacr/es-posible-un-messenger-para-steem) intentaba precisamente eso. Hoy, dos años después, mis conocimientos en programación han mejorado bastante, aún no son suficientes, pero gracias a las nuevas tecnologías se puede llegar más lejos, me refiero al uso de AI, esa que puede ser usada para hacer cosas perjudiciales a la plataforma también podemos usarla para impulsar STEEM, así que... en lo que a mí respecta creo que gracias a ello puedo ver materializada cualquier idea que me proponga, por eso el título "Una idea un proyecto" ya que son muchas las ideas que tengo, aún así vamos a desarrollar una a la vez. Como la idea de [un Messenger para STEEM](https://steemit.com/colombiaoriginal/@jesusjacr/es-posible-un-messenger-para-steem) lleva dos años esperándonos considero justo trabajar en desarrollar la misma. ¿Podemos crear algo útil? creo que sí, ¿Será exitoso? ya lo veremos. https://bing.com/th/id/BCO.5b8a46f5-ae48-4c14-b1ef-5d838bc419a1.png ^{Imagen generada por [Microsoft Copilot](https://bing.com/th/id/BCO.5b8a46f5-ae48-4c14-b1ef-5d838bc419a1.png) con fines ilustrativos para este artículo} Objetivo general: Crear una aplicación de mensajería instantánea para STEEM blockchain capaz de mantener a los usuarios de la plataforma comunicados entre sí de una forma cómoda, amigable y entretenida. Objetivos Específicos: Objetvio 1: Conseguir un script capaz de "escuchar" las operaciones de una cuenta y extraer en tiempo real las relacionadas con las transferencias con el fin de obtener datos tales como Bloque, emisor, receptor, monto, memo. La finalidad es el intento de extracción de datos en tiempo real, algo que será muy útil en el sistema de mensajería. Objetivo 2: Integrar STEEM y TELEGRAM mediante python para recibir notificaciones y mensajes de STEEM en TELEGRAM con posibilidad de respuesta. Objetivo 3: Integrar STEEM y WHATSAPP mediante python para recibir notificaciones y mensajes de STEEM en WHATSAPP con posibilidad de respuesta. Objetivo 4: Crear una APP para Android capaz de gestionar un servicio de mensajería entre usuarios de STEEM que además tenga capacidad para trasladar mensajes hasta whatsapp y telegram. https://cdn.pixabay.com/photo/2020/08/03/09/51/pencil-5459686_1280.png ^{[Pixabay](https://pixabay.com/es/vectors/l%C3%A1piz-nota-tiempo-escribiendo-5459686/)} Se que suena bastante ambicioso, por eso los objetivos pueden ser modificados durante el desarrollo del proyecto, aún así prometo dar lo mejor de mí en este proyecto que inicia hoy de manera oficial.

Rastreo de MEMOS

Por ahora no quiero ocupar mi VPS ya que es un proyecto en desarrollo y la VPS es para correr cosas de forma ininterrumpida, este proyecto lo voy a desarrollar en mi laptop con Windows Digo esto porque de nuevo me enfrenté al reto de instalar el módulo steem en windows y de nuevo sin éxito gracias a un módulo llamado pycrypto que al parecer es obsoleto y steem obliga a su instalación, no quise indagar mucho en la solución a este problema porque había una alternativa más eficiente, la instalación de beem ![image.png](https://cdn.steemitimages.com/DQmPGYMgQohs3gqsVSD5BZfYWbqNMZbDw4ASKm2k8eLzt3k/image.png) Luego que la instalación de beem fue un éxito, lo siguiente fue pedirle a Copilot algunos códigos, al principio solo obtener el número de seguidores, el saldo de la cuenta entre otras cosas. No fue tan simple como imaginé, de hecho, pensé que Copilot me daría códigos perfectos, pero luego de muchos intentos seguía cometiendo errores, no quise darle todo el trabajo de codificar porque vi que no es experto aún en steem, pero estoy seguro de que también se está entrenando (al igual que yo) para serlo. Cuando lo anterior fue posible entonces decidí avanzar en el objetivo 1, lo primero fue rastrear transferencias. Tuvimos un buen rato intentando cosas sin éxitos hasta que por fin un código pudo rastrear una transferencia hecha. ~~~ #!/usr/bin/env python3 import requests def get_transfers(account: str, limit: int = 100): """ Trae hasta `limit` operaciones de tipo 'transfer' para `account` usando directamente el RPC JSON-RPC de Steem (condenser_api). Respeta el tope de 100 ops por llamada y parsea con seguridad. """ # No pases nunca más de 100 al RPC limit = min(limit, 100) payload = { "jsonrpc": "2.0", "method": "condenser_api.get_account_history", "params": [account, -1, limit], # -1 = desde la última op hacia atrás "id": 1 } res = requests.post("https://api.steemit.com", json=payload).json() history = res.get("result", []) transfers = [] for entry in history: # 1) Cada entry debe ser lista o tupla con al menos 2 elementos if not isinstance(entry, (list, tuple)) or len(entry) < 2: continue # 2) entry[1] es la operación; puede venir en dos formatos: op = entry[1] # 2a) Formato clásico: [op_type, op_data] if isinstance(op, (list, tuple)) and len(op) >= 2: op_type, op_data = op[0], op[1] # 2b) Formato extendido (Graphene): {'op': [op_type, op_data], …} elif isinstance(op, dict) and "op" in op and isinstance(op["op"], (list, tuple)): op_type, op_data = op["op"][0], op["op"][1] else: continue # 3) Si es transferencia, guárdala if op_type == "transfer": transfers.append(op_data) return transfers if __name__ == "__main__": cuenta = "creating" # cambia por la cuenta que quieras inspeccionar # Pide hasta 100 operaciones. Si tu tx está más atrás, tendrías que paginar. for tx in get_transfers(cuenta, limit=100): print( f"💸 From: {tx['from']} → To: {tx['to']} | " f"{tx['amount']} | memo: {tx['memo']}" ) if not get_transfers(cuenta, limit=100): print("ℹ️ No apareció tu transferencia en las últimas 100 ops.") print(" Si sigue sin verse, revisa un bloque más atrás o prueba otro nodo.") ~~~ El problema que teníamos era con el límite de registros que podíamos extraer de la blockchain y desempaquetar los datos, en esto Copilot fue muy útil y el código vio su primer fruto. Bien, ya teníamos como buscar transferencias que se hayan realizado en el pasado (es lo que hace el código que te acabo de compartir). Pero solo era el principio así que fuimos un poco más profundo. Ahora, el siguiente paso era crear un bucle que se mantuviera activo esperando mensajes nuevos, al detectarlos debía imprimirlos. Aquello se volvió de nuevo complicado y de nuevo muchos errores que corregir, pero finalmente y para resumir un poco se consiguió el código capaz de hacerlo. ~~~ #!/usr/bin/env python3 import time import requests STEEM_NODE = "https://api.steemit.com" ACCOUNT = "jesusjacr" POLL_INTERVAL = 5 # segundos entre polling HISTORY_LIMIT = 100 # nunca más de 100 para no romper la API def get_account_history(account: str, limit: int = HISTORY_LIMIT): payload = { "jsonrpc": "2.0", "method": "condenser_api.get_account_history", "params": [account, -1, min(limit, HISTORY_LIMIT)], "id": 1 } res = requests.post(STEEM_NODE, json=payload).json() return res.get("result", []) def listen_transfers(account: str): last_idx = -1 print(f"⏱️ Escuchando transferencias de @{account} cada {POLL_INTERVAL}s…\n") while True: history = get_account_history(account) # tomamos solo las nuevas operaciones new_entries = [e for e in history if e and isinstance(e, (list,tuple)) and e[0] > last_idx] if new_entries: # orden cronológico new_entries.sort(key=lambda x: x[0]) for entry in new_entries: idx, op_entry = entry[0], entry[1] # extraer [op_type, op_data] if isinstance(op_entry, (list,tuple)) and len(op_entry) >= 2: op_type, op_data = op_entry[0], op_entry[1] elif isinstance(op_entry, dict) and "op" in op_entry \ and isinstance(op_entry["op"], (list,tuple)) \ and len(op_entry["op"]) >= 2: op_type, op_data = op_entry["op"][0], op_entry["op"][1] else: # formato inesperado, lo saltamos continue # solo transfers if op_type == "transfer": frm = op_data.get("from") to = op_data.get("to") amt = op_data.get("amount") memo = op_data.get("memo") # solo si involucra tu cuenta if frm == account or to == account: direction = "→" if frm == account else "←" print(f"[{idx}] {frm} {direction} {to} | {amt} | memo: {memo}") # actualizamos last_idx last_idx = idx time.sleep(POLL_INTERVAL) if __name__ == "__main__": listen_transfers(ACCOUNT) ~~~ Ese fue el código que finalmente dio resultados, al hacer una operación podía rastrearla e imprimirla, pero había un problema adicional, al parecer los bloques de STEEM tardan algo de tiempo antes de confirmarse, entonces una vez que se enviaba una transferencia, pasaban alrededor de 5 minutos antes que pudiese detectarla e imprimirla, eso no es bueno para un servicio de mensajería, es como si estuviésemos enviando palomas mensajeras en lugar de usar ondas electromagnéticas. Nadie querría una aplicación de mensajería con 5 minutos de retardo entre que el mensaje sale hasta que llega (también he notado que esto ocurre de forma global en Steemit, cuando haces alguna transferencia u otras cosas suelen tardar en actualizarse). ¿La solución? No esperar a que los bloques se confirmen, rastrear operaciones en bloques que aún no han sido confirmados, esto da la ventaja de recibir el mensaje de forma instantánea con la desventaja que un bloque no confirmado es un bloque que aún no tiene información confiable, pero solo estamos rastreando mensajes así que no es algo que afecte mucho el proceso. El código final que imprime las operaciones de transferencias sin espera: ~~~ #!/usr/bin/env python3 import time from beem import Steem STEEM_NODE = "https://api.steemit.com" ACCOUNT = "jesusjacr" POLL_INTERVAL = 3 # Segundos entre chequeos de nuevo bloque def get_head_block_number(steem): props = steem.rpc.get_dynamic_global_properties() return props.get("head_block_number", 0) def process_block(steem, block_num, account): """ Busca dentro del bloque todas las operaciones 'transfer' e imprime solo las que tienen a `account` como from o to. """ block = steem.rpc.get_block(block_num) for trx in block.get("transactions", []): # Cada trx es dict con clave "operations" for op in trx.get("operations", []): # op == [op_type, op_data] if not (isinstance(op, (list, tuple)) and len(op) == 2): continue op_type, data = op if op_type != "transfer": continue frm, to = data.get("from"), data.get("to") amt, memo = data.get("amount"), data.get("memo") if frm == account or to == account: direction = "→" if frm == account else "←" print(f"[block {block_num}] {frm} {direction} {to} | {amt} | memo: {memo}") def listen_transfers_by_block(account, poll_interval=POLL_INTERVAL): steem = Steem(node=[STEEM_NODE]) last_block = get_head_block_number(steem) print(f"⏱️ Escuchando @ {account} a nivel de bloques. Comenzamos en {last_block}\n") while True: head = get_head_block_number(steem) if head > last_block: # Procesar desde el siguiente bloque hasta el head actual for bn in range(last_block + 1, head + 1): process_block(steem, bn, account) last_block = head time.sleep(poll_interval) if __name__ == "__main__": listen_transfers_by_block(ACCOUNT) ~~~ En todos los códigos el usuario es jesusjacr, si quieres probar con tu cuenta debes cambiar este parámetro. Al final el objetivo uno fue logrado, aunque está faltando el manejo de mensajes cifrados pero el trabajo no ha hecho más que iniciar. ![image.png](https://cdn.steemitimages.com/DQmZYqVLU1wxRyAKqWSyNzWY3iS8MjVdRnrRxkFPhn1D25Z/image.png) Estos datos pueden ser estructurados en una aplicación de mensajería para ser interpretados y manejados por dicha app. Gracias por acompañarme una vez más.