야 솔직히 말해서 블록체인 결제 상태 확인하는 거 진짜 귀찮지 않음?? ㅋㅋㅋ
이더리움이나 폴리곤으로 결제했는데 "아 이거 됐나 안 됐나" 하면서 Etherscan 새로고침 누르고 있는 자신 발견한 적 있잖아요 😅
그래서 오늘은 파이썬(Python)으로 Web3 결제 상태를 자동으로 추적하는 프로그램을 만들어볼 거임!
트랜잭션 해시 넣으면 알아서 상태 체크하고, 완료되면 알림까지 보내주는 그런 거 ㄹㅇ 실용적인 거 만들어봅시다 💪

🧠 Web3 결제 추적, 왜 필요한 거임?

블록체인 트랜잭션은 은행 이체랑 달리 "즉시 완료"가 아님.
이더리움 기준으로 트랜잭션이 블록에 포함되려면 평균 12~15초가 걸리고,
"안전하다"고 볼 수 있는 컨펌 수(보통 12 confirmations)까지는 몇 분이 걸릴 수 있어요.
그 사이에 상태가 Pending → Mining → Success/Failed 이렇게 바뀌는데,
이걸 수동으로 체크하면 진짜 노가다임 ㅋㅋㅋ

🛠️ 개발 환경 세팅부터 하자

일단 필요한 라이브러리부터 설치해야지~ 파이썬 3.8 이상 환경 기준으로 설명할게요 🐍

📦 필수 패키지 설치

pip install web3
pip install python-dotenv
pip install requests
pip install aiohttp
pip install asyncio

🔌 RPC 노드 연결 설정

Web3와 블록체인을 연결하려면 RPC 엔드포인트가 필요해요.
직접 노드 돌리는 건 너무 무거우니까 아래 서비스 중 하나 쓰면 됨:

INFURA_URL=https://mainnet.infura.io/v3/YOUR_PROJECT_ID
ALCHEMY_URL=https://eth-mainnet.g.alchemy.com/v2/YOUR_API_KEY
SLACK_WEBHOOK=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
TELEGRAM_CHAT_ID=YOUR_CHAT_ID

💻 기본 트랜잭션 상태 조회 코드

자 이제 진짜 코드 짜볼 시간! 먼저 단일 트랜잭션 상태 조회하는 기본 코드부터 ㄱㄱ

🔍 트랜잭션 상태 조회 기본 구조

from web3 import Web3
from dotenv import load_dotenv
import os
import time

load_dotenv()

# Web3 연결 초기화
def init_web3(rpc_url: str) -> Web3:
    w3 = Web3(Web3.HTTPProvider(rpc_url))
    if w3.is_connected():
        print(f"✅ 블록체인 연결 성공! 현재 블록: {w3.eth.block_number}")
    else:
        raise ConnectionError("❌ 블록체인 연결 실패!")
    return w3

# 트랜잭션 상태 조회 함수
def get_tx_status(w3: Web3, tx_hash: str) -> dict:
    try:
        # 트랜잭션 기본 정보 조회
        tx = w3.eth.get_transaction(tx_hash)
        
        if tx is None:
            return {
                "status": "NOT_FOUND",
                "message": "트랜잭션을 찾을 수 없음 (잘못된 해시 or 아직 전파 안 됨)"
            }
        
        # 트랜잭션 영수증 조회 (블록에 포함됐는지 확인)
        try:
            receipt = w3.eth.get_transaction_receipt(tx_hash)
        except Exception:
            receipt = None
        
        if receipt is None:
            # 영수증 없음 = 아직 Pending 상태
            return {
                "status": "PENDING",
                "message": "⏳ 트랜잭션 대기 중 (mempool에 있음)",
                "from": tx["from"],
                "to": tx["to"],
                "value": w3.from_wei(tx["value"], "ether"),
                "gas_price": w3.from_wei(tx["gasPrice"], "gwei"),
                "nonce": tx["nonce"]
            }
        
        # 영수증 있음 = 블록에 포함됨
        current_block = w3.eth.block_number
        confirmations = current_block - receipt["blockNumber"]
        
        status_code = receipt["status"]  # 1 = 성공, 0 = 실패
        
        return {
            "status": "SUCCESS" if status_code == 1 else "FAILED",
            "message": "✅ 트랜잭션 성공!" if status_code == 1 else "❌ 트랜잭션 실패!",
            "block_number": receipt["blockNumber"],
            "confirmations": confirmations,
            "gas_used": receipt["gasUsed"],
            "from": receipt["from"],
            "to": receipt["to"],
            "tx_hash": tx_hash
        }
        
    except Exception as e:
        return {
            "status": "ERROR",
            "message": f"조회 중 오류 발생: {str(e)}"
        }

# 메인 실행
if __name__ == "__main__":
    rpc_url = os.getenv("INFURA_URL")
    w3 = init_web3(rpc_url)
    
    tx_hash = "0x여기에_트랜잭션_해시_입력"
    result = get_tx_status(w3, tx_hash)
    print(result)

🔄 자동 폴링 시스템 구축

한 번 조회하는 건 쉬운데, 진짜 핵심은 자동으로 계속 체크하는 거잖아요!
트랜잭션이 완료될 때까지 주기적으로 상태를 확인하는 폴링 시스템 만들어봅시다 🔄

⏱️ 폴링 간격 설정 전략

import time
import asyncio
from web3 import Web3
from typing import Callable, Optional

class TransactionTracker:
    def __init__(self, w3: Web3, poll_interval: int = 12):
        self.w3 = w3
        self.poll_interval = poll_interval  # 초 단위
        self.tracking_list = {}  # 추적 중인 트랜잭션 목록
    
    def add_transaction(
        self, 
        tx_hash: str, 
        required_confirmations: int = 1,
        callback: Optional[Callable] = None,
        label: str = ""
    ):
        """추적할 트랜잭션 등록"""
        self.tracking_list[tx_hash] = {
            "hash": tx_hash,
            "required_confirmations": required_confirmations,
            "callback": callback,
            "label": label or tx_hash[:10] + "...",
            "start_time": time.time(),
            "status": "PENDING"
        }
        print(f"📝 트랜잭션 추적 등록: {label or tx_hash[:10]}...")
    
    def check_single_tx(self, tx_hash: str) -> dict:
        """단일 트랜잭션 상태 체크"""
        try:
            receipt = self.w3.eth.get_transaction_receipt(tx_hash)
            
            if receipt is None:
                return {"status": "PENDING", "confirmations": 0}
            
            current_block = self.w3.eth.block_number
            confirmations = current_block - receipt["blockNumber"]
            tx_status = "SUCCESS" if receipt["status"] == 1 else "FAILED"
            
            return {
                "status": tx_status,
                "confirmations": confirmations,
                "block_number": receipt["blockNumber"],
                "gas_used": receipt["gasUsed"],
                "receipt": receipt
            }
        except Exception as e:
            return {"status": "ERROR", "error": str(e)}
    
    def start_tracking(self, timeout: int = 600):
        """
        모든 등록된 트랜잭션 추적 시작
        timeout: 최대 대기 시간 (초, 기본 10분)
        """
        print(f"\n🚀 추적 시작! 총 {len(self.tracking_list)}개 트랜잭션")
        print(f"⏱️  폴링 간격: {self.poll_interval}초 / 타임아웃: {timeout}초\n")
        
        start_time = time.time()
        
        while self.tracking_list:
            # 타임아웃 체크
            if time.time() - start_time > timeout:
                print("⏰ 타임아웃! 추적 종료")
                for tx_hash in list(self.tracking_list.keys()):
                    print(f"  ⚠️  미완료: {self.tracking_list[tx_hash]['label']}")
                break
            
            completed = []
            
            for tx_hash, tx_info in self.tracking_list.items():
                result = self.check_single_tx(tx_hash)
                elapsed = int(time.time() - tx_info["start_time"])
                
                if result["status"] == "PENDING":
                    print(f"⏳ [{tx_info['label']}] 대기 중... ({elapsed}초 경과)")
                
                elif result["status"] in ["SUCCESS", "FAILED"]:
                    confirmations = result["confirmations"]
                    required = tx_info["required_confirmations"]
                    
                    if confirmations >= required:
                        # 필요한 컨펌 수 달성!
                        emoji = "✅" if result["status"] == "SUCCESS" else "❌"
                        print(f"{emoji} [{tx_info['label']}] {result['status']}!")
                        print(f"   블록: {result['block_number']} | 컨펌: {confirmations} | 가스: {result['gas_used']}")
                        
                        # 콜백 함수 실행
                        if tx_info["callback"]:
                            tx_info["callback"](tx_hash, result)
                        
                        completed.append(tx_hash)
                    else:
                        print(f"⛏️  [{tx_info['label']}] 컨펌 대기: {confirmations}/{required}")
                
                elif result["status"] == "ERROR":
                    print(f"⚠️  [{tx_info['label']}] 오류: {result.get('error', '알 수 없음')}")
            
            # 완료된 트랜잭션 제거
            for tx_hash in completed:
                del self.tracking_list[tx_hash]
            
            if self.tracking_list:
                print(f"  💤 {self.poll_interval}초 후 재확인...\n")
                time.sleep(self.poll_interval)
        
        print("\n🏁 모든 트랜잭션 추적 완료!")


# 사용 예시
if __name__ == "__main__":
    from dotenv import load_dotenv
    import os
    
    load_dotenv()
    w3 = Web3(Web3.HTTPProvider(os.getenv("INFURA_URL")))
    
    tracker = TransactionTracker(w3, poll_interval=12)
    
    # 콜백 함수 정의 (트랜잭션 완료 시 실행)
    def on_tx_complete(tx_hash, result):
        print(f"🎉 콜백 실행! {tx_hash[:10]}... 완료")
        # 여기서 DB 업데이트, 이메일 발송 등 처리
    
    # 트랜잭션 등록
    tracker.add_transaction(
        tx_hash="0x트랜잭션_해시_1",
        required_confirmations=3,
        callback=on_tx_complete,
        label="NFT 구매"
    )
    
    tracker.add_transaction(
        tx_hash="0x트랜잭션_해시_2",
        required_confirmations=1,
        callback=on_tx_complete,
        label="토큰 전송"
    )
    
    # 추적 시작!
    tracker.start_tracking(timeout=300)

🔔 알림 시스템 연동하기

트랜잭션 상태 감지했으면 이제 알림을 보내야지! 실무에서 많이 쓰는 3가지 방법 알려드림 ㄱㄱ

📱 텔레그램 봇 알림

import requests
import os
from dotenv import load_dotenv

load_dotenv()

class TelegramNotifier:
    def __init__(self):
        self.bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
        self.chat_id = os.getenv("TELEGRAM_CHAT_ID")
        self.base_url = f"https://api.telegram.org/bot{self.bot_token}"
    
    def send_message(self, message: str) -> bool:
        """텔레그램 메시지 발송"""
        try:
            url = f"{self.base_url}/sendMessage"
            payload = {
                "chat_id": self.chat_id,
                "text": message,
                "parse_mode": "HTML"  # HTML 포맷 지원
            }
            response = requests.post(url, json=payload, timeout=10)
            return response.status_code == 200
        except Exception as e:
            print(f"텔레그램 발송 실패: {e}")
            return False
    
    def notify_tx_success(self, tx_hash: str, label: str, confirmations: int):
        message = (
            f"✅ <b>트랜잭션 성공!</b>\n\n"
            f"📌 라벨: {label}\n"
            f"🔗 해시: <code>{tx_hash[:20]}...</code>\n"
            f"✔️ 컨펌 수: {confirmations}\n"
            f"🔍 <a href='https://etherscan.io/tx/{tx_hash}'>Etherscan에서 보기</a>"
        )
        return self.send_message(message)
    
    def notify_tx_failed(self, tx_hash: str, label: str):
        message = (
            f"❌ <b>트랜잭션 실패!</b>\n\n"
            f"📌 라벨: {label}\n"
            f"🔗 해시: <code>{tx_hash[:20]}...</code>\n"
            f"⚠️ 가스 부족 또는 컨트랙트 오류 확인 필요\n"
            f"🔍 <a href='https://etherscan.io/tx/{tx_hash}'>Etherscan에서 보기</a>"
        )
        return self.send_message(message)


class SlackNotifier:
    def __init__(self):
        self.webhook_url = os.getenv("SLACK_WEBHOOK")
    
    def send_message(self, text: str, color: str = "#36a64f") -> bool:
        """슬랙 웹훅으로 메시지 발송"""
        try:
            payload = {
                "attachments": [{
                    "color": color,
                    "text": text,
                    "mrkdwn_in": ["text"]
                }]
            }
            response = requests.post(self.webhook_url, json=payload, timeout=10)
            return response.status_code == 200
        except Exception as e:
            print(f"슬랙 발송 실패: {e}")
            return False
    
    def notify_tx_success(self, tx_hash: str, label: str):
        text = f"✅ *트랜잭션 성공!* | {label} | `{tx_hash[:16]}...`"
        return self.send_message(text, color="#2eb886")
    
    def notify_tx_failed(self, tx_hash: str, label: str):
        text = f"❌ *트랜잭션 실패!* | {label} | `{tx_hash[:16]}...`"
        return self.send_message(text, color="#e01e5a")

🔗 트래커에 알림 연동

from web3 import Web3
from auto_tracker import TransactionTracker
from notifier import TelegramNotifier, SlackNotifier
from dotenv import load_dotenv
import os

load_dotenv()

# 초기화
w3 = Web3(Web3.HTTPProvider(os.getenv("INFURA_URL")))
tracker = TransactionTracker(w3, poll_interval=12)
telegram = TelegramNotifier()
slack = SlackNotifier()

# 콜백 함수 (알림 포함)
def on_complete(tx_hash: str, result: dict):
    label = result.get("label", "Unknown")
    
    if result["status"] == "SUCCESS":
        telegram.notify_tx_success(tx_hash, label, result["confirmations"])
        slack.notify_tx_success(tx_hash, label)
        print(f"📨 알림 발송 완료!")
    else:
        telegram.notify_tx_failed(tx_hash, label)
        slack.notify_tx_failed(tx_hash, label)
        print(f"📨 실패 알림 발송 완료!")

# 트랜잭션 등록 및 추적 시작
tracker.add_transaction(
    tx_hash="0x실제_트랜잭션_해시",
    required_confirmations=3,
    callback=on_complete,
    label="결제 확인"
)

tracker.start_tracking(timeout=600)

⚡ 비동기(Async) 버전으로 업그레이드

여러 트랜잭션을 동시에 추적할 때는 비동기 처리가 훨씬 효율적임!
동기 방식은 하나씩 순서대로 체크하지만, 비동기는 동시에 여러 개 체크 가능 ㄷㄷ

import asyncio
import aiohttp
import time
from web3 import Web3
from typing import List, Dict, Callable, Optional

class AsyncTransactionTracker:
    def __init__(self, rpc_url: str, poll_interval: int = 12):
        self.rpc_url = rpc_url
        self.poll_interval = poll_interval
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        self.tracking_list: Dict[str, dict] = {}
    
    def add_transaction(
        self,
        tx_hash: str,
        required_confirmations: int = 1,
        callback: Optional[Callable] = None,
        label: str = ""
    ):
        self.tracking_list[tx_hash] = {
            "hash": tx_hash,
            "required_confirmations": required_confirmations,
            "callback": callback,
            "label": label or f"{tx_hash[:8]}...",
            "start_time": time.time()
        }
    
    async def check_tx_async(self, session: aiohttp.ClientSession, tx_hash: str) -> dict:
        """비동기 방식으로 트랜잭션 상태 조회"""
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_getTransactionReceipt",
            "params": [tx_hash],
            "id": 1
        }
        
        try:
            async with session.post(
                self.rpc_url,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:
                data = await response.json()
                receipt = data.get("result")
                
                if receipt is None:
                    return {"status": "PENDING", "confirmations": 0}
                
                # 현재 블록 번호 조회
                block_payload = {
                    "jsonrpc": "2.0",
                    "method": "eth_blockNumber",
                    "params": [],
                    "id": 2
                }
                async with session.post(self.rpc_url, json=block_payload) as block_resp:
                    block_data = await block_resp.json()
                    current_block = int(block_data["result"], 16)
                
                tx_block = int(receipt["blockNumber"], 16)
                confirmations = current_block - tx_block
                status_int = int(receipt["status"], 16)
                
                return {
                    "status": "SUCCESS" if status_int == 1 else "FAILED",
                    "confirmations": confirmations,
                    "block_number": tx_block,
                    "gas_used": int(receipt["gasUsed"], 16)
                }
        
        except asyncio.TimeoutError:
            return {"status": "TIMEOUT", "confirmations": 0}
        except Exception as e:
            return {"status": "ERROR", "error": str(e)}
    
    async def run(self, timeout: int = 600):
        """비동기 추적 메인 루프"""
        print(f"🚀 비동기 추적 시작! {len(self.tracking_list)}개 트랜잭션")
        
        start_time = time.time()
        
        async with aiohttp.ClientSession() as session:
            while self.tracking_list:
                if time.time() - start_time > timeout:
                    print("⏰ 타임아웃!")
                    break
                
                # 모든 트랜잭션 동시 체크!
                tasks = {
                    tx_hash: self.check_tx_async(session, tx_hash)
                    for tx_hash in self.tracking_list
                }
                
                results = await asyncio.gather(*tasks.values(), return_exceptions=True)
                
                completed = []
                
                for tx_hash, result in zip(tasks.keys(), results):
                    if isinstance(result, Exception):
                        print(f"⚠️ 오류: {result}")
                        continue
                    
                    tx_info = self.tracking_list[tx_hash]
                    
                    if result["status"] == "PENDING":
                        elapsed = int(time.time() - tx_info["start_time"])
                        print(f"⏳ [{tx_info['label']}] 대기 중 ({elapsed}초)")
                    
                    elif result["status"] in ["SUCCESS", "FAILED"]:
                        if result["confirmations"] >= tx_info["required_confirmations"]:
                            emoji = "✅" if result["status"] == "SUCCESS" else "❌"
                            print(f"{emoji} [{tx_info['label']}] 완료! 컨펌: {result['confirmations']}")
                            
                            if tx_info["callback"]:
                                # 비동기 콜백 지원
                                if asyncio.iscoroutinefunction(tx_info["callback"]):
                                    await tx_info["callback"](tx_hash, result)
                                else:
                                    tx_info["callback"](tx_hash, result)
                            
                            completed.append(tx_hash)
                        else:
                            print(f"⛏️  [{tx_info['label']}] 컨펌: {result['confirmations']}/{tx_info['required_confirmations']}")
                
                for tx_hash in completed:
                    del self.tracking_list[tx_hash]
                
                if self.tracking_list:
                    await asyncio.sleep(self.poll_interval)
        
        print("🏁 추적 완료!")
    
    def start(self, timeout: int = 600):
        """동기 방식으로 비동기 루프 실행"""
        asyncio.run(self.run(timeout))


# 사용 예시
if __name__ == "__main__":
    from dotenv import load_dotenv
    import os
    
    load_dotenv()
    
    tracker = AsyncTransactionTracker(
        rpc_url=os.getenv("INFURA_URL"),
        poll_interval=12
    )
    
    def my_callback(tx_hash, result):
        print(f"🎉 완료 콜백: {tx_hash[:10]}... | {result['status']}")
    
    # 여러 트랜잭션 동시 추적
    hashes = [
        ("0x해시1", "결제A"),
        ("0x해시2", "결제B"),
        ("0x해시3", "결제C"),
    ]
    
    for tx_hash, label in hashes:
        tracker.add_transaction(
            tx_hash=tx_hash,
            required_confirmations=3,
            callback=my_callback,
            label=label
        )
    
    tracker.start(timeout=300)

🌐 멀티체인 지원 확장

요즘은 이더리움만 쓰는 게 아니잖아요~ 폴리곤, BSC, 아비트럼 등 다양한 체인 지원하는 버전도 만들어봅시다 💪

from web3 import Web3
from typing import Dict

# 체인 설정 딕셔너리
CHAIN_CONFIG = {
    "ethereum": {
        "name": "이더리움 메인넷",
        "rpc": "https://eth.llamarpc.com",
        "explorer": "https://etherscan.io/tx/",
        "poll_interval": 12,
        "chain_id": 1
    },
    "polygon": {
        "name": "폴리곤",
        "rpc": "https://polygon-rpc.com",
        "explorer": "https://polygonscan.com/tx/",
        "poll_interval": 2,
        "chain_id": 137
    },
    "bsc": {
        "name": "바이낸스 스마트체인",
        "rpc": "https://bsc-dataseed.binance.org/",
        "explorer": "https://bscscan.com/tx/",
        "poll_interval": 3,
        "chain_id": 56
    },
    "arbitrum": {
        "name": "아비트럼 원",
        "rpc": "https://arb1.arbitrum.io/rpc",
        "explorer": "https://arbiscan.io/tx/",
        "poll_interval": 1,
        "chain_id": 42161
    },
    "optimism": {
        "name": "옵티미즘",
        "rpc": "https://mainnet.optimism.io",
        "explorer": "https://optimistic.etherscan.io/tx/",
        "poll_interval": 2,
        "chain_id": 10
    }
}

class MultiChainTracker:
    def __init__(self):
        self.connections: Dict[str, Web3] = {}
        self._init_connections()
    
    def _init_connections(self):
        """모든 체인 연결 초기화"""
        for chain_key, config in CHAIN_CONFIG.items():
            try:
                w3 = Web3(Web3.HTTPProvider(config["rpc"]))
                if w3.is_connected():
                    self.connections[chain_key] = w3
                    print(f"✅ {config['name']} 연결 성공")
                else:
                    print(f"⚠️  {config['name']} 연결 실패")
            except Exception as e:
                print(f"❌ {config['name']} 오류: {e}")
    
    def get_tx_status(self, chain: str, tx_hash: str) -> dict:
        """특정 체인의 트랜잭션 상태 조회"""
        if chain not in self.connections:
            return {"status": "ERROR", "message": f"지원하지 않는 체인: {chain}"}
        
        w3 = self.connections[chain]
        config = CHAIN_CONFIG[chain]
        
        try:
            receipt = w3.eth.get_transaction_receipt(tx_hash)
            
            if receipt is None:
                return {
                    "status": "PENDING",
                    "chain": config["name"],
                    "explorer_url": config["explorer"] + tx_hash
                }
            
            current_block = w3.eth.block_number
            confirmations = current_block - receipt["blockNumber"]
            
            return {
                "status": "SUCCESS" if receipt["status"] == 1 else "FAILED",
                "chain": config["name"],
                "confirmations": confirmations,
                "block_number": receipt["blockNumber"],
                "gas_used": receipt["gasUsed"],
                "explorer_url": config["explorer"] + tx_hash
            }
        
        except Exception as e:
            return {"status": "ERROR", "message": str(e)}
    
    def auto_detect_chain(self, tx_hash: str) -> str:
        """트랜잭션 해시로 어느 체인인지 자동 감지"""
        print(f"🔍 체인 자동 감지 중: {tx_hash[:10]}...")
        
        for chain_key, w3 in self.connections.items():
            try:
                tx = w3.eth.get_transaction(tx_hash)
                if tx is not None:
                    print(f"✅ {CHAIN_CONFIG[chain_key]['name']}에서 발견!")
                    return chain_key
            except Exception:
                continue
        
        return "NOT_FOUND"


# 사용 예시
if __name__ == "__main__":
    tracker = MultiChainTracker()
    
    # 폴리곤 트랜잭션 조회
    result = tracker.get_tx_status("polygon", "0x폴리곤_트랜잭션_해시")
    print(result)
    
    # 체인 자동 감지
    chain = tracker.auto_detect_chain("0x알_수_없는_트랜잭션_해시")
    print(f"감지된 체인: {chain}")

📊 데이터베이스 연동 & 이력 관리

실제 서비스에서는 추적 이력을 DB에 저장해야 하잖아요.
SQLite로 간단하게 트랜잭션 이력 관리하는 방법 알려드림!

import sqlite3
import json
from datetime import datetime
from typing import Optional, List

class TxDatabase:
    def __init__(self, db_path: str = "transactions.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """데이터베이스 초기화 및 테이블 생성"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS transactions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    tx_hash TEXT UNIQUE NOT NULL,
                    chain TEXT NOT NULL,
                    label TEXT,
                    status TEXT DEFAULT 'PENDING',
                    block_number INTEGER,
                    confirmations INTEGER DEFAULT 0,
                    gas_used INTEGER,
                    from_address TEXT,
                    to_address TEXT,
                    value_eth REAL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    extra_data TEXT
                )
            """)
            conn.commit()
        print("✅ 데이터베이스 초기화 완료")
    
    def insert_tx(self, tx_hash: str, chain: str, label: str = "", extra: dict = None):
        """새 트랜잭션 등록"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR IGNORE INTO transactions 
                (tx_hash, chain, label, extra_data)
                VALUES (?, ?, ?, ?)
            """, (tx_hash, chain, label, json.dumps(extra or {})))
            conn.commit()
    
    def update_tx_status(self, tx_hash: str, status: str, **kwargs):
        """트랜잭션 상태 업데이트"""
        fields = ["status = ?", "updated_at = CURRENT_TIMESTAMP"]
        values = [status]
        
        for key, value in kwargs.items():
            fields.append(f"{key} = ?")
            values.append(value)
        
        values.append(tx_hash)
        
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                f"UPDATE transactions SET {', '.join(fields)} WHERE tx_hash = ?",
                values
            )
            conn.commit()
    
    def get_pending_txs(self) -> List[dict]:
        """Pending 상태 트랜잭션 목록 조회"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(
                "SELECT * FROM transactions WHERE status = 'PENDING' ORDER BY created_at"
            )
            return [dict(row) for row in cursor.fetchall()]
    
    def get_tx_history(self, limit: int = 50) -> List[dict]:
        """최근 트랜잭션 이력 조회"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(
                "SELECT * FROM transactions ORDER BY created_at DESC LIMIT ?",
                (limit,)
            )
            return [dict(row) for row in cursor.fetchall()]
    
    def get_stats(self) -> dict:
        """통계 조회"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT 
                    COUNT(*) as total,
                    SUM(CASE WHEN status='SUCCESS' THEN 1 ELSE 0 END) as success,
                    SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as failed,
                    SUM(CASE WHEN status='PENDING' THEN 1 ELSE 0 END) as pending
                FROM transactions
            """)
            row = cursor.fetchone()
            return {
                "total": row[0],
                "success": row[1],
                "failed": row[2],
                "pending": row[3]
            }

🎯 실전 팁 & 주의사항

⚠️ 흔히 하는 실수들

🔒 보안 고려사항

import re

def validate_tx_hash(tx_hash: str) -> bool:
    """트랜잭션 해시 유효성 검사"""
    pattern = r'^0x[0-9a-fA-F]{64}$'
    return bool(re.match(pattern, tx_hash))

def validate_address(address: str) -> bool:
    """이더리움 주소 유효성 검사"""
    pattern = r'^0x[0-9a-fA-F]{40}$'
    return bool(re.match(pattern, address))

# 사용 예시
tx_hash = "0x1234...abcd"
if not validate_tx_hash(tx_hash):
    raise ValueError(f"유효하지 않은 트랜잭션 해시: {tx_hash}")

🚀 Alchemy 웹훅으로 더 스마트하게

폴링 방식 말고 웹훅(Webhook) 방식도 있음!
Alchemy에서 제공하는 웹훅을 쓰면 트랜잭션 완료 시 자동으로 내 서버에 알림이 와요.
폴링보다 훨씬 효율적이고 실시간성도 좋음 ㄷㄷ

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
from notifier import TelegramNotifier

app = Flask(__name__)
telegram = TelegramNotifier()

# Alchemy 웹훅 서명 검증
ALCHEMY_SIGNING_KEY = "your_alchemy_signing_key"

def verify_alchemy_signature(payload: bytes, signature: str) -> bool:
    """Alchemy 웹훅 서명 검증"""
    expected = hmac.new(
        ALCHEMY_SIGNING_KEY.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhook/alchemy", methods=["POST"])
def alchemy_webhook():
    # 서명 검증
    signature = request.headers.get("X-Alchemy-Signature", "")
    if not verify_alchemy_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    data = request.json
    
    # 이벤트 타입 확인
    event_type = data.get("type")
    
    if event_type == "MINED_TRANSACTION":
        # 트랜잭션 채굴 완료
        tx_hash = data["event"]["transaction"]["hash"]
        from_addr = data["event"]["transaction"]["from"]
        to_addr = data["event"]["transaction"]["to"]
        
        print(f"⛏️  트랜잭션 채굴됨: {tx_hash[:16]}...")
        
        telegram.send_message(
            f"⛏️ <b>트랜잭션 채굴 완료!</b>\n"
            f"해시: <code>{tx_hash[:20]}...</code>\n"
            f"From: <code>{from_addr[:16]}...</code>"
        )
    
    elif event_type == "DROPPED_TRANSACTION":
        # 트랜잭션 드랍됨
        tx_hash = data["event"]["transaction"]["hash"]
        print(f"🗑️  트랜잭션 드랍됨: {tx_hash[:16]}...")
        
        telegram.send_message(
            f"🗑️ <b>트랜잭션 드랍!</b>\n"
            f"해시: <code>{tx_hash[:20]}...</code>\n"
            f"⚠️ 가스비 부족 또는 nonce 충돌 확인 필요"
        )
    
    return jsonify({"status": "ok"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)

📋 전체 프로젝트 구조 정리

지금까지 만든 코드들을 하나의 프로젝트로 정리하면 이런 구조가 됨!

web3-tx-tracker/
├── .env                    # API 키, 웹훅 URL (깃 제외!)
├── .gitignore
├── requirements.txt
├── main.py                 # 메인 실행 파일
├── config.py               # 체인 설정, 상수 정의
├── tracker/
│   ├── __init__.py
│   ├── sync_tracker.py     # 동기 방식 트래커
│   └── async_tracker.py    # 비동기 방식 트래커
├── notifier/
│   ├── __init__.py
│   ├── telegram.py         # 텔레그램 알림
│   ├── slack.py            # 슬랙 알림
│   └── email_notifier.py   # 이메일 알림
├── database/
│   ├── __init__.py
│   └── db_manager.py       # SQLite DB 관리
├── webhook/
│   ├── __init__.py
│   └── server.py           # 웹훅 수신 서버
└── utils/
    ├── __init__.py
    └── validators.py       # 입력값 검증

web3>=6.0.0
python-dotenv>=1.0.0
requests>=2.31.0
aiohttp>=3.9.0
flask>=3.0.0
sqlite3  # 파이썬 내장

🌟 마무리 & 다음 단계

오늘 만든 Web3 결제 상태 자동 추적 프로그램 어때요? ㅋㅋㅋ
기본 구조부터 비동기 처리, 멀티체인 지원, 알림 연동, DB 저장까지 꽤 완성도 있는 시스템이 됐죠! 🎉

이런 Web3 개발 스킬은 재능넷 같은 플랫폼에서도 꽤 수요가 있는 분야예요!
블록체인 결제 시스템 구축, 트랜잭션 모니터링 툴 개발 등 실용적인 프로젝트들이 많이 의뢰되고 있거든요 😊

파이썬 + Web3 조합은 진입장벽도 낮고 실용성도 높아서 블록체인 개발 입문으로 딱 좋음!
오늘 코드 직접 돌려보면서 익혀보세요 ~ 화이팅!! 🔥