Parser Ton Transactions without Memo

Скрипт парсит все транзакции на указанный адрес, отправляет в канал все транзакции, которые НЕ проходят на биржу.

Таким образом, вы получаете сотни адресов людей, которые потеряли деньги, отправляя без MEMO или с неправильным MEMO.

import requests
import time
import base64
import threading
import json
import re

BINANCE_ADDR = "UQD5mxRgCuRNLxKxeOjG6r14iSroLF5FtomPnet-sgP5xI-e"
API_URL = "https://ton-mainnet.core.chainstack.com/" #Вам нужен полный апи токен для работы
TELEGRAM_BOT_TOKEN = "7513896300:s"
TELEGRAM_CHAT_ID = "-100"
MAX_SENDER_BALANCE = 5000.0

processed_senders = set()
last_known_lt = 0
processing_lock = threading.Lock()

def decode_possible_base64(memo_text):
if not memo_text:
return memo_text

base64_pattern = re.compile(r'^[A-Za-z0-9+/]+={0,2}$')
if len(memo_text) % 4 == 0 and base64_pattern.match(memo_text):
try:
decoded = base64.b64decode(memo_text).decode('utf-8')
if re.match(r'^[\d]+$', decoded):
return decoded
if len(decoded) > 0 and not re.search(r'[\x00-\x1F\x7F]', decoded):
return decoded
except:
pass
return memo_text

def fetch_transactions(address, url, limit=50):
if address.startswith('UQ'):
try:
decoded = base64.urlsafe_b64decode(address + '==')
eq_address = base64.b64encode(decoded).decode('utf-8')
except:
eq_address = address
else:
eq_address = address

params = {
'address': eq_address,
'limit': limit,
'archival': False
}
try:
resp = requests.get(url, params=params, timeout=10)
if resp.status_code == 200:
return resp.json().get('result', [])
print(f"Ошибка API: {resp.status_code}, Ответ: {resp.text}")
return []
except Exception as e:
print(f"Ошибка соединения: {str(e)}")
return []

def has_nine_digits(memo_text):
if not memo_text:
return False

decoded_memo = decode_possible_base64(memo_text)

digit_sequences = re.findall(r'\d+', decoded_memo)

return any(len(seq) == 9 for seq in digit_sequences)

def extract_memo_text(in_msg):
if not in_msg:
return None

if in_msg.get('body'):
body = in_msg['body']
if isinstance(body, str) and body.strip() != "":
return body
if isinstance(body, str) and body.startswith("00000000"):
try:
return bytes.fromhex(body[8:]).decode('utf-8', errors='ignore')
except:
return None

decoded = in_msg.get('decoded_body', {})
if decoded.get('text'):
return decoded['text']
if decoded.get('comment'):
return decoded['comment']

msg_data = in_msg.get('msg_data', {})
if msg_data.get('@type') == 'msg.dataText':
return msg_data.get('text', '')

return None

def is_memo_missing(in_msg):
return extract_memo_text(in_msg) is None

def is_valid_memo(in_msg):
memo_text = extract_memo_text(in_msg)
if not memo_text:
return False

decoded_memo = decode_possible_base64(memo_text)

return not has_nine_digits(decoded_memo)

def get_wallet_balance(wallet_address):
try:
url = f"https://tonapi.io/v2/accounts/{wallet_address}"
response = requests.get(url, timeout=5)
if response.status_code == 200:
data = response.json()
return int(data['balance']) / 1e9
return 0.0
except:
return 0.0

def send_telegram_alert(sender_address, amount, balance, memo_status, memo_text=None):
try:
if amount >= 1:
amount_str = f"{amount:.2f} TON"
else:
amount_str = f"{amount:.9f} TON".rstrip('0').rstrip('.')
if amount_str.endswith('TON'):
amount_str = amount_str.replace('TON', '0 TON')

if memo_text and len(memo_text) > 100:
memo_text = memo_text[:100] + "..."

if memo_status == "NO_MEMO":
message = (
" *ТРАНЗАКЦИЯ БЕЗ MEMO!*\n\n"
f"• *Отправитель:* `{sender_address}`\n"
f"• *Сумма:* `{amount_str}`\n"
f"• *Баланс отправителя:* `{balance:.2f} TON`\n\n"
)
else:
decoded_memo = decode_possible_base64(memo_text)
display_memo = decoded_memo if decoded_memo != memo_text else memo_text

message = (
" *ТРАНЗАКЦИЯ С MEMO*\n\n"
f"• *Отправитель:* `{sender_address}`\n"
f"• *Сумма:* `{amount_str}`\n"
f"• *Баланс отправителя:* `{balance:.2f} TON`\n"
f"• *MEMO (raw):* `{memo_text}`\n"
f"• *MEMO (decoded):* `{display_memo}`\n\n"
)

url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown",
"disable_web_page_preview": True
}

threading.Thread(target=requests.post, args=(url,), kwargs={'json': payload}).start()
return True
except Exception as e:
print(f"Ошибка отправки в Telegram: {str(e)}")
return False

def main():
global last_known_lt, processed_senders

print("Мониторинг транзакций запущен...")
print(f"Отслеживаем адрес: {BINANCE_ADDR}")
print(f"Фильтр: все транзакции, кроме MEMO с 9 цифрами")

init_txs = fetch_transactions(BINANCE_ADDR, API_URL, limit=10)
if init_txs:
last_known_lt = max(int(tx['transaction_id']['lt']) for tx in init_txs)
print(f"Инициализирован last_known_lt: {last_known_lt}")

while True:
try:
txs = fetch_transactions(BINANCE_ADDR, API_URL, limit=50)
if not txs:
time.sleep(5)
continue

current_max_lt = max(int(tx['transaction_id']['lt']) for tx in txs)
new_txs = [tx for tx in txs if int(tx['transaction_id']['lt']) > last_known_lt]

if new_txs:
for tx in sorted(new_txs, key=lambda x: int(x['transaction_id']['lt'])):
in_msg = tx.get('in_msg', {})
value = int(in_msg.get('value', 0))
amount_ton = value / 1e9

if amount_ton <= 1:
continue

sender_address = in_msg.get('source', '')
if not sender_address:
continue

with processing_lock:
if sender_address in processed_senders:
continue

balance = get_wallet_balance(sender_address)
if balance > MAX_SENDER_BALANCE:
continue

memo_text = extract_memo_text(in_msg)

if is_memo_missing(in_msg):
print(f"Обнаружена транзакция БЕЗ MEMO: {sender_address} -> {amount_ton} TON")
if send_telegram_alert(sender_address, amount_ton, balance, "NO_MEMO"):
processed_senders.add(sender_address)
elif is_valid_memo(in_msg):
print(f"Обнаружена транзакция с MEMO: {sender_address} -> {amount_ton} TON | MEMO: {memo_text}")
if send_telegram_alert(sender_address, amount_ton, balance, "WITH_MEMO", memo_text):
processed_senders.add(sender_address)
else:
decoded_memo = decode_possible_base64(memo_text)
print(f"Пропущена транзакция с 9 цифрами в MEMO: {memo_text} -> {decoded_memo}")

last_known_lt = current_max_lt
print(f"Обновлен last_known_lt: {last_known_lt}")

time.sleep(1)

except Exception as e:
print(f"Критическая ошибка: {str(e)}")
time.sleep(10)

if __name__ == "__main__":
main()

Python

import requests

import time

import base64

import threading

import json

import re



BINANCE_ADDR = "UQD5mxRgCuRNLxKxeOjG6r14iSroLF5FtomPnet-sgP5xI-e"

API_URL = "https://ton-mainnet.core.chainstack.com/" #Вам нужен полный апи токен для работы

TELEGRAM_BOT_TOKEN = "7513896300:s"

TELEGRAM_CHAT_ID = "-100"

MAX_SENDER_BALANCE = 5000.0



processed_senders = set()

last_known_lt = 0

processing_lock = threading.Lock()



def decode_possible_base64(memo_text):

    if not memo_text:

        return memo_text

        

    base64_pattern = re.compile(r'^[A-Za-z0-9+/]+={0,2}$')

    if len(memo_text) % 4 == 0 and base64_pattern.match(memo_text):

        try:

            decoded = base64.b64decode(memo_text).decode('utf-8')

            if re.match(r'^[\d]+$', decoded):

                return decoded

            if len(decoded) > 0 and not re.search(r'[\x00-\x1F\x7F]', decoded):

                return decoded

        except:

            pass

    return memo_text



def fetch_transactions(address, url, limit=50):

    if address.startswith('UQ'):

        try:

            decoded = base64.urlsafe_b64decode(address + '==')

            eq_address = base64.b64encode(decoded).decode('utf-8')

        except:

            eq_address = address

    else:

        eq_address = address

    

    params = {

        'address': eq_address,

        'limit': limit,

        'archival': False

    }

    try:

        resp = requests.get(url, params=params, timeout=10)

        if resp.status_code == 200:

            return resp.json().get('result', [])

        print(f"Ошибка API: {resp.status_code}, Ответ: {resp.text}")

        return []

    except Exception as e:

        print(f"Ошибка соединения: {str(e)}")

        return []



def has_nine_digits(memo_text):

    if not memo_text:

        return False

        

    decoded_memo = decode_possible_base64(memo_text)

    

    digit_sequences = re.findall(r'\d+', decoded_memo)

    

    return any(len(seq) == 9 for seq in digit_sequences)



def extract_memo_text(in_msg):

    if not in_msg:

        return None

        

    if in_msg.get('body'):

        body = in_msg['body']

        if isinstance(body, str) and body.strip() != "":

            return body

        if isinstance(body, str) and body.startswith("00000000"):

            try:

                return bytes.fromhex(body[8:]).decode('utf-8', errors='ignore')

            except:

                return None

                

    decoded = in_msg.get('decoded_body', {})

    if decoded.get('text'):

        return decoded['text']

    if decoded.get('comment'):

        return decoded['comment']

        

    msg_data = in_msg.get('msg_data', {})

    if msg_data.get('@type') == 'msg.dataText':

        return msg_data.get('text', '')

        

    return None



def is_memo_missing(in_msg):

    return extract_memo_text(in_msg) is None



def is_valid_memo(in_msg):

    memo_text = extract_memo_text(in_msg)

    if not memo_text:

        return False

        

    decoded_memo = decode_possible_base64(memo_text)

    

    return not has_nine_digits(decoded_memo)



def get_wallet_balance(wallet_address):

    try:

        url = f"https://tonapi.io/v2/accounts/{wallet_address}"

        response = requests.get(url, timeout=5)

        if response.status_code == 200:

            data = response.json()

            return int(data['balance']) / 1e9

        return 0.0

    except:

        return 0.0



def send_telegram_alert(sender_address, amount, balance, memo_status, memo_text=None):

    try:

        if amount >= 1:

            amount_str = f"{amount:.2f} TON"

        else:

            amount_str = f"{amount:.9f} TON".rstrip('0').rstrip('.')

            if amount_str.endswith('TON'):

                amount_str = amount_str.replace('TON', '0 TON')

        

        if memo_text and len(memo_text) > 100:

            memo_text = memo_text[:100] + "..."

        

        if memo_status == "NO_MEMO":

            message = (

                " *ТРАНЗАКЦИЯ БЕЗ MEMO!*\n\n"

                f"• *Отправитель:* `{sender_address}`\n"

                f"• *Сумма:* `{amount_str}`\n"

                f"• *Баланс отправителя:* `{balance:.2f} TON`\n\n"

            )

        else:

            decoded_memo = decode_possible_base64(memo_text)

            display_memo = decoded_memo if decoded_memo != memo_text else memo_text

            

            message = (

                " *ТРАНЗАКЦИЯ С MEMO*\n\n"

                f"• *Отправитель:* `{sender_address}`\n"

                f"• *Сумма:* `{amount_str}`\n"

                f"• *Баланс отправителя:* `{balance:.2f} TON`\n"

                f"• *MEMO (raw):* `{memo_text}`\n"

                f"• *MEMO (decoded):* `{display_memo}`\n\n"

            )

        

        url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"

        payload = {

            "chat_id": TELEGRAM_CHAT_ID,

            "text": message,

            "parse_mode": "Markdown",

            "disable_web_page_preview": True

        }

        

        threading.Thread(target=requests.post, args=(url,), kwargs={'json': payload}).start()

        return True

    except Exception as e:

        print(f"Ошибка отправки в Telegram: {str(e)}")

        return False



def main():

    global last_known_lt, processed_senders

    

    print("Мониторинг транзакций запущен...")

    print(f"Отслеживаем адрес: {BINANCE_ADDR}")

    print(f"Фильтр: все транзакции, кроме MEMO с 9 цифрами")

    

    init_txs = fetch_transactions(BINANCE_ADDR, API_URL, limit=10)

    if init_txs:

        last_known_lt = max(int(tx['transaction_id']['lt']) for tx in init_txs)

        print(f"Инициализирован last_known_lt: {last_known_lt}")

    

    while True:

        try:

            txs = fetch_transactions(BINANCE_ADDR, API_URL, limit=50)

            if not txs:

                time.sleep(5)

                continue

            

            current_max_lt = max(int(tx['transaction_id']['lt']) for tx in txs)

            new_txs = [tx for tx in txs if int(tx['transaction_id']['lt']) > last_known_lt]

            

            if new_txs:

                for tx in sorted(new_txs, key=lambda x: int(x['transaction_id']['lt'])):

                    in_msg = tx.get('in_msg', {})

                    value = int(in_msg.get('value', 0))

                    amount_ton = value / 1e9

                    

                    if amount_ton <= 1:

                        continue

                        

                    sender_address = in_msg.get('source', '')

                    if not sender_address:

                        continue

                        

                    with processing_lock:

                        if sender_address in processed_senders:

                            continue

                            

                        balance = get_wallet_balance(sender_address)

                        if balance > MAX_SENDER_BALANCE:

                            continue

                            

                        memo_text = extract_memo_text(in_msg)

                        

                        if is_memo_missing(in_msg):

                            print(f"Обнаружена транзакция БЕЗ MEMO: {sender_address} -> {amount_ton} TON")

                            if send_telegram_alert(sender_address, amount_ton, balance, "NO_MEMO"):

                                processed_senders.add(sender_address)

                        elif is_valid_memo(in_msg):

                            print(f"Обнаружена транзакция с MEMO: {sender_address} -> {amount_ton} TON | MEMO: {memo_text}")

                            if send_telegram_alert(sender_address, amount_ton, balance, "WITH_MEMO", memo_text):

                                processed_senders.add(sender_address)

                        else:

                            decoded_memo = decode_possible_base64(memo_text)

                            print(f"Пропущена транзакция с 9 цифрами в MEMO: {memo_text} -> {decoded_memo}")

                

                last_known_lt = current_max_lt

                print(f"Обновлен last_known_lt: {last_known_lt}")

            

            time.sleep(1)

            

        except Exception as e:

            print(f"Критическая ошибка: {str(e)}")

            time.sleep(10)



if __name__ == "__main__":

    main()