🎯 왜 블록체인 영수증 자동화가 필요할까?

블록체인 기술이 발전하면서 암호화폐 결제, NFT 거래, DeFi 서비스 이용이 급증하고 있어. 하지만 이런 거래들은 대부분 트랜잭션 해시와 블록 번호로만 기록되기 때문에, 일반 사용자들이 이해하기 어렵고 회계 처리나 세금 신고에 활용하기도 까다로워. 😰

특히 재능넷 같은 플랫폼에서 블록체인 기반 결제 시스템을 도입한다면, 사용자들에게 친숙한 형태의 영수증을 제공하는 것이 필수적이야. 그래서 우리가 만들 시스템은 다음과 같은 문제들을 해결해줄 거야!

🛠️ 시스템 아키텍처 설계

본격적으로 코딩에 들어가기 전에, 우리가 만들 시스템의 전체 구조를 먼저 이해해보자. 크게 4개의 주요 컴포넌트로 구성될 거야! 🏗️

📦 필요한 라이브러리 설치

자, 이제 실전으로 들어가보자! 먼저 필요한 파이썬 라이브러리들을 설치해야 해. 각 라이브러리가 어떤 역할을 하는지 함께 알아보면서 설치해볼게. 🎒

web3==6.11.3
reportlab==4.0.7
Pillow==10.1.0
python-dotenv==1.0.0
requests==2.31.0
qrcode==7.4.2
APScheduler==3.10.4
pandas==2.1.3

터미널에서 다음 명령어로 한 번에 설치할 수 있어:

pip install web3 reportlab Pillow python-dotenv requests qrcode APScheduler pandas

🔗 블록체인 연동 모듈 구현

이제 본격적으로 코드를 작성해볼 시간이야! 먼저 블록체인에서 트랜잭션 정보를 가져오는 모듈부터 만들어보자. 이더리움 메인넷과 테스트넷(Sepolia) 모두 지원하도록 구현할 거야. 🚀

먼저 프로젝트 루트에 .env 파일을 만들어서 API 키와 설정값들을 저장하자:

# Infura 또는 Alchemy API 키
INFURA_PROJECT_ID=your_infura_project_id_here
ALCHEMY_API_KEY=your_alchemy_api_key_here

# Etherscan API 키 (트랜잭션 상세 정보용)
ETHERSCAN_API_KEY=your_etherscan_api_key_here

# 모니터링할 지갑 주소
WALLET_ADDRESS=0xYourWalletAddressHere

# 네트워크 설정 (mainnet, sepolia, goerli)
NETWORK=sepolia

# PDF 저장 경로
PDF_OUTPUT_PATH=./receipts/

이제 Web3.py를 사용해서 블록체인과 통신하는 클래스를 만들어보자:

from web3 import Web3
from web3.middleware import geth_poa_middleware
import os
from dotenv import load_dotenv
import requests
from datetime import datetime
import json

# 환경 변수 로드
load_dotenv()

class BlockchainConnector:
    """블록체인 연동을 담당하는 클래스"""
    
    def __init__(self, network='sepolia'):
        self.network = network
        self.setup_web3()
        self.etherscan_api_key = os.getenv('ETHERSCAN_API_KEY')
        
    def setup_web3(self):
        """Web3 인스턴스 설정"""
        if self.network == 'mainnet':
            infura_url = f"https://mainnet.infura.io/v3/{os.getenv('INFURA_PROJECT_ID')}"
            self.etherscan_base = "https://api.etherscan.io/api"
        elif self.network == 'sepolia':
            infura_url = f"https://sepolia.infura.io/v3/{os.getenv('INFURA_PROJECT_ID')}"
            self.etherscan_base = "https://api-sepolia.etherscan.io/api"
        else:
            raise ValueError(f"지원하지 않는 네트워크: {self.network}")
        
        self.w3 = Web3(Web3.HTTPProvider(infura_url))
        
        # PoA 네트워크를 위한 미들웨어 추가 (테스트넷용)
        if self.network != 'mainnet':
            self.w3.middleware_onion.inject(geth_poa_middleware, layer=0)
        
        # 연결 확인
        if not self.w3.is_connected():
            raise ConnectionError(f"{self.network} 네트워크에 연결할 수 없습니다.")
        
        print(f"✅ {self.network} 네트워크에 성공적으로 연결되었습니다!")
    
    def get_transaction(self, tx_hash):
        """트랜잭션 정보 조회"""
        try:
            # 트랜잭션 기본 정보
            tx = self.w3.eth.get_transaction(tx_hash)
            
            # 트랜잭션 영수증 (가스 사용량, 상태 등)
            receipt = self.w3.eth.get_transaction_receipt(tx_hash)
            
            # 블록 정보 (타임스탬프)
            block = self.w3.eth.get_block(tx['blockNumber'])
            
            # 데이터 정리
            transaction_data = {
                'hash': tx_hash,
                'from': tx['from'],
                'to': tx['to'],
                'value': self.w3.from_wei(tx['value'], 'ether'),
                'gas_price': self.w3.from_wei(tx['gasPrice'], 'gwei'),
                'gas_used': receipt['gasUsed'],
                'total_gas_fee': self.w3.from_wei(
                    receipt['gasUsed'] * tx['gasPrice'], 'ether'
                ),
                'block_number': tx['blockNumber'],
                'timestamp': datetime.fromtimestamp(block['timestamp']),
                'status': 'Success' if receipt['status'] == 1 else 'Failed',
                'nonce': tx['nonce']
            }
            
            return transaction_data
            
        except Exception as e:
            print(f"❌ 트랜잭션 조회 실패: {e}")
            return None
    
    def get_wallet_transactions(self, wallet_address, start_block=0, end_block='latest'):
        """특정 지갑의 트랜잭션 목록 조회 (Etherscan API 사용)"""
        try:
            params = {
                'module': 'account',
                'action': 'txlist',
                'address': wallet_address,
                'startblock': start_block,
                'endblock': end_block,
                'sort': 'desc',
                'apikey': self.etherscan_api_key
            }
            
            response = requests.get(self.etherscan_base, params=params)
            data = response.json()
            
            if data['status'] == '1':
                transactions = []
                for tx in data['result']:
                    transactions.append({
                        'hash': tx['hash'],
                        'from': tx['from'],
                        'to': tx['to'],
                        'value': float(self.w3.from_wei(int(tx['value']), 'ether')),
                        'timestamp': datetime.fromtimestamp(int(tx['timeStamp'])),
                        'block_number': int(tx['blockNumber']),
                        'gas_used': int(tx['gasUsed']),
                        'gas_price': float(self.w3.from_wei(int(tx['gasPrice']), 'gwei'))
                    })
                return transactions
            else:
                print(f"⚠️ Etherscan API 오류: {data.get('message', 'Unknown error')}")
                return []
                
        except Exception as e:
            print(f"❌ 지갑 트랜잭션 조회 실패: {e}")
            return []
    
    def get_eth_price(self):
        """현재 ETH 가격 조회 (USD)"""
        try:
            params = {
                'module': 'stats',
                'action': 'ethprice',
                'apikey': self.etherscan_api_key
            }
            
            response = requests.get(self.etherscan_base, params=params)
            data = response.json()
            
            if data['status'] == '1':
                return float(data['result']['ethusd'])
            return None
            
        except Exception as e:
            print(f"❌ ETH 가격 조회 실패: {e}")
            return None
    
    def verify_transaction(self, tx_hash):
        """트랜잭션 검증 (블록체인에 실제로 존재하는지)"""
        try:
            receipt = self.w3.eth.get_transaction_receipt(tx_hash)
            return receipt['status'] == 1
        except Exception:
            return False

📄 PDF 영수증 생성 모듈

이제 가장 재미있는 부분이야! 블록체인 데이터를 예쁜 PDF 영수증으로 만들어보자. ReportLab을 사용해서 전문적인 디자인의 영수증을 생성할 거야. 🎨

from reportlab.lib.pagesizes import A4, letter
from reportlab.lib import colors
from reportlab.lib.units import inch, mm
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, Image
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.enums import TA_CENTER, TA_RIGHT, TA_LEFT
from reportlab.pdfgen import canvas
import qrcode
from io import BytesIO
from datetime import datetime
import os

class ReceiptPDFGenerator:
    """블록체인 영수증 PDF 생성 클래스"""
    
    def __init__(self, output_path='./receipts/'):
        self.output_path = output_path
        # 출력 디렉토리가 없으면 생성
        os.makedirs(output_path, exist_ok=True)
        
        # 스타일 설정
        self.styles = getSampleStyleSheet()
        self.setup_custom_styles()
    
    def setup_custom_styles(self):
        """커스텀 스타일 정의"""
        # 제목 스타일
        self.title_style = ParagraphStyle(
            'CustomTitle',
            parent=self.styles['Heading1'],
            fontSize=24,
            textColor=colors.HexColor('#667eea'),
            spaceAfter=30,
            alignment=TA_CENTER,
            fontName='Helvetica-Bold'
        )
        
        # 부제목 스타일
        self.subtitle_style = ParagraphStyle(
            'CustomSubtitle',
            parent=self.styles['Normal'],
            fontSize=12,
            textColor=colors.HexColor('#718096'),
            spaceAfter=20,
            alignment=TA_CENTER
        )
        
        # 섹션 헤더 스타일
        self.section_header_style = ParagraphStyle(
            'SectionHeader',
            parent=self.styles['Heading2'],
            fontSize=14,
            textColor=colors.HexColor('#2d3748'),
            spaceAfter=12,
            spaceBefore=20,
            fontName='Helvetica-Bold'
        )
        
        # 일반 텍스트 스타일
        self.normal_style = ParagraphStyle(
            'CustomNormal',
            parent=self.styles['Normal'],
            fontSize=10,
            textColor=colors.HexColor('#4a5568'),
            spaceAfter=6
        )
    
    def create_qr_code(self, data, size=100):
        """QR 코드 이미지 생성"""
        qr = qrcode.QRCode(
            version=1,
            error_correction=qrcode.constants.ERROR_CORRECT_L,
            box_size=10,
            border=4,
        )
        qr.add_data(data)
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        
        # BytesIO로 이미지를 메모리에 저장
        buffer = BytesIO()
        img.save(buffer, format='PNG')
        buffer.seek(0)
        
        return buffer
    
    def generate_receipt(self, transaction_data, eth_price=None):
        """영수증 PDF 생성"""
        # 파일명 생성
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = f"receipt_{transaction_data['hash'][:10]}_{timestamp}.pdf"
        filepath = os.path.join(self.output_path, filename)
        
        # PDF 문서 생성
        doc = SimpleDocTemplate(
            filepath,
            pagesize=A4,
            rightMargin=30*mm,
            leftMargin=30*mm,
            topMargin=20*mm,
            bottomMargin=20*mm
        )
        
        # 문서 요소들을 담을 리스트
        elements = []
        
        # 헤더 섹션
        elements.append(Paragraph("🔗 BLOCKCHAIN RECEIPT", self.title_style))
        elements.append(Paragraph(
            f"Transaction Verification Document",
            self.subtitle_style
        ))
        elements.append(Spacer(1, 20))
        
        # 거래 정보 테이블
        elements.append(Paragraph("📋 Transaction Details", self.section_header_style))
        
        tx_data = [
            ['Transaction Hash:', transaction_data['hash']],
            ['Block Number:', str(transaction_data['block_number'])],
            ['Date & Time:', transaction_data['timestamp'].strftime('%Y-%m-%d %H:%M:%S UTC')],
            ['Status:', transaction_data['status']],
            ['Network:', transaction_data.get('network', 'Ethereum')]
        ]
        
        tx_table = Table(tx_data, colWidths=[120, 350])
        tx_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (0, -1), colors.HexColor('#f7fafc')),
            ('TEXTCOLOR', (0, 0), (-1, -1), colors.HexColor('#2d3748')),
            ('ALIGN', (0, 0), (0, -1), 'RIGHT'),
            ('ALIGN', (1, 0), (1, -1), 'LEFT'),
            ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'),
            ('FONTNAME', (1, 0), (1, -1), 'Helvetica'),
            ('FONTSIZE', (0, 0), (-1, -1), 9),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e2e8f0'))
        ]))
        elements.append(tx_table)
        elements.append(Spacer(1, 20))
        
        # 금액 정보
        elements.append(Paragraph("💰 Payment Information", self.section_header_style))
        
        amount_data = [
            ['From Address:', transaction_data['from']],
            ['To Address:', transaction_data['to']],
            ['Amount (ETH):', f"{transaction_data['value']:.6f} ETH"],
        ]
        
        # USD 환산 금액 추가
        if eth_price:
            usd_value = float(transaction_data['value']) * eth_price
            amount_data.append(['Amount (USD):', f"${usd_value:,.2f} USD"])
        
        amount_table = Table(amount_data, colWidths=[120, 350])
        amount_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (0, -1), colors.HexColor('#f7fafc')),
            ('TEXTCOLOR', (0, 0), (-1, -1), colors.HexColor('#2d3748')),
            ('ALIGN', (0, 0), (0, -1), 'RIGHT'),
            ('ALIGN', (1, 0), (1, -1), 'LEFT'),
            ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'),
            ('FONTNAME', (1, 0), (1, -1), 'Helvetica'),
            ('FONTSIZE', (0, 0), (-1, -1), 9),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e2e8f0'))
        ]))
        elements.append(amount_table)
        elements.append(Spacer(1, 20))
        
        # 가스비 정보
        elements.append(Paragraph("⛽ Gas Fee Details", self.section_header_style))
        
        gas_data = [
            ['Gas Price:', f"{transaction_data['gas_price']:.2f} Gwei"],
            ['Gas Used:', f"{transaction_data['gas_used']:,}"],
            ['Total Gas Fee:', f"{transaction_data['total_gas_fee']:.6f} ETH"],
        ]
        
        gas_table = Table(gas_data, colWidths=[120, 350])
        gas_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (0, -1), colors.HexColor('#f7fafc')),
            ('TEXTCOLOR', (0, 0), (-1, -1), colors.HexColor('#2d3748')),
            ('ALIGN', (0, 0), (0, -1), 'RIGHT'),
            ('ALIGN', (1, 0), (1, -1), 'LEFT'),
            ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'),
            ('FONTNAME', (1, 0), (1, -1), 'Helvetica'),
            ('FONTSIZE', (0, 0), (-1, -1), 9),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e2e8f0'))
        ]))
        elements.append(gas_table)
        elements.append(Spacer(1, 30))
        
        # QR 코드 추가
        qr_buffer = self.create_qr_code(transaction_data['hash'])
        qr_image = Image(qr_buffer, width=100, height=100)
        
        qr_data = [[qr_image, Paragraph(
            f"<b>Scan to Verify</b><br/>"
            f"<font size=8>Scan this QR code to view<br/>"
            f"the transaction on blockchain explorer</font>",
            self.normal_style
        )]]
        
        qr_table = Table(qr_data, colWidths=[120, 350])
        qr_table.setStyle(TableStyle([
            ('ALIGN', (0, 0), (0, 0), 'CENTER'),
            ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
        ]))
        elements.append(qr_table)
        elements.append(Spacer(1, 30))
        
        # 푸터
        footer_text = Paragraph(
            f"<font size=8 color='#a0aec0'>"
            f"This receipt is automatically generated from blockchain data.<br/>"
            f"Generated on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}<br/>"
            f"Powered by Python & Web3 Technology"
            f"</font>",
            ParagraphStyle('Footer', parent=self.normal_style, alignment=TA_CENTER)
        )
        elements.append(footer_text)
        
        # PDF 빌드
        doc.build(elements)
        
        print(f"✅ PDF 영수증이 생성되었습니다: {filepath}")
        return filepath
    
    def generate_batch_receipts(self, transactions_list, eth_price=None):
        """여러 트랜잭션의 영수증을 일괄 생성"""
        generated_files = []
        
        for tx in transactions_list:
            try:
                filepath = self.generate_receipt(tx, eth_price)
                generated_files.append(filepath)
            except Exception as e:
                print(f"❌ 영수증 생성 실패 (TX: {tx.get('hash', 'Unknown')}): {e}")
        
        print(f"\n📊 총 {len(generated_files)}개의 영수증이 생성되었습니다.")
        return generated_files

🤖 자동화 시스템 구현

이제 마지막 퍼즐 조각이야! 새로운 트랜잭션을 자동으로 감지하고 영수증을 생성하는 자동화 시스템을 만들어보자. APScheduler를 사용해서 주기적으로 블록체인을 모니터링할 거야. ⚡

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
from datetime import datetime, timedelta
import json
import os
from blockchain_connector import BlockchainConnector
from pdf_generator import ReceiptPDFGenerator

class ReceiptAutomationEngine:
    """영수증 자동 생성 엔진"""
    
    def __init__(self, wallet_address, network='sepolia', check_interval=60):
        self.wallet_address = wallet_address
        self.network = network
        self.check_interval = check_interval  # 초 단위
        
        # 컴포넌트 초기화
        self.blockchain = BlockchainConnector(network=network)
        self.pdf_generator = ReceiptPDFGenerator()
        
        # 처리된 트랜잭션 추적
        self.processed_file = 'processed_transactions.json'
        self.processed_txs = self.load_processed_transactions()
        
        # 스케줄러 초기화
        self.scheduler = BlockingScheduler()
        
        print(f"🤖 자동화 엔진이 초기화되었습니다.")
        print(f"📍 모니터링 지갑: {wallet_address}")
        print(f"🌐 네트워크: {network}")
        print(f"⏱️ 체크 주기: {check_interval}초")
    
    def load_processed_transactions(self):
        """이미 처리된 트랜잭션 목록 로드"""
        if os.path.exists(self.processed_file):
            with open(self.processed_file, 'r') as f:
                return set(json.load(f))
        return set()
    
    def save_processed_transactions(self):
        """처리된 트랜잭션 목록 저장"""
        with open(self.processed_file, 'w') as f:
            json.dump(list(self.processed_txs), f, indent=2)
    
    def check_new_transactions(self):
        """새로운 트랜잭션 확인 및 처리"""
        try:
            print(f"\n🔍 [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 새로운 트랜잭션 확인 중...")
            
            # 최근 트랜잭션 조회 (최근 100개)
            transactions = self.blockchain.get_wallet_transactions(
                self.wallet_address,
                start_block=0,
                end_block='latest'
            )
            
            if not transactions:
                print("   ℹ️ 조회된 트랜잭션이 없습니다.")
                return
            
            # 새로운 트랜잭션 필터링
            new_transactions = [
                tx for tx in transactions 
                if tx['hash'] not in self.processed_txs
            ]
            
            if not new_transactions:
                print(f"   ✅ 새로운 트랜잭션이 없습니다. (총 {len(transactions)}개 확인)")
                return
            
            print(f"   🆕 {len(new_transactions)}개의 새로운 트랜잭션 발견!")
            
            # ETH 가격 조회
            eth_price = self.blockchain.get_eth_price()
            if eth_price:
                print(f"   💵 현재 ETH 가격: ${eth_price:,.2f}")
            
            # 각 트랜잭션에 대해 영수증 생성
            for tx in new_transactions:
                try:
                    # 상세 정보 조회
                    detailed_tx = self.blockchain.get_transaction(tx['hash'])
                    
                    if detailed_tx:
                        # 네트워크 정보 추가
                        detailed_tx['network'] = self.network.capitalize()
                        
                        # PDF 생성
                        pdf_path = self.pdf_generator.generate_receipt(
                            detailed_tx, 
                            eth_price
                        )
                        
                        # 처리 완료 표시
                        self.processed_txs.add(tx['hash'])
                        
                        print(f"   ✅ 영수증 생성 완료: {tx['hash'][:16]}...")
                    
                except Exception as e:
                    print(f"   ❌ 영수증 생성 실패 ({tx['hash'][:16]}...): {e}")
            
            # 처리된 트랜잭션 목록 저장
            self.save_processed_transactions()
            
            print(f"   📊 처리 완료: {len(new_transactions)}개의 영수증 생성됨")
            
        except Exception as e:
            print(f"❌ 트랜잭션 확인 중 오류 발생: {e}")
    
    def manual_generate(self, tx_hash):
        """특정 트랜잭션의 영수증을 수동으로 생성"""
        try:
            print(f"\n📝 수동 영수증 생성 시작: {tx_hash}")
            
            # 트랜잭션 정보 조회
            tx_data = self.blockchain.get_transaction(tx_hash)
            
            if not tx_data:
                print("❌ 트랜잭션을 찾을 수 없습니다.")
                return None
            
            # 네트워크 정보 추가
            tx_data['network'] = self.network.capitalize()
            
            # ETH 가격 조회
            eth_price = self.blockchain.get_eth_price()
            
            # PDF 생성
            pdf_path = self.pdf_generator.generate_receipt(tx_data, eth_price)
            
            # 처리 완료 표시
            self.processed_txs.add(tx_hash)
            self.save_processed_transactions()
            
            return pdf_path
            
        except Exception as e:
            print(f"❌ 수동 생성 실패: {e}")
            return None
    
    def start_monitoring(self):
        """자동 모니터링 시작"""
        print("\n" + "="*60)
        print("🚀 블록체인 영수증 자동 생성 시스템 시작!")
        print("="*60)
        
        # 초기 체크
        self.check_new_transactions()
        
        # 주기적 체크 스케줄 등록
        self.scheduler.add_job(
            self.check_new_transactions,
            trigger=IntervalTrigger(seconds=self.check_interval),
            id='transaction_checker',
            name='트랜잭션 체크',
            replace_existing=True
        )
        
        print(f"\n⏰ {self.check_interval}초마다 자동으로 체크합니다.")
        print("   종료하려면 Ctrl+C를 누르세요.\n")
        
        try:
            self.scheduler.start()
        except (KeyboardInterrupt, SystemExit):
            print("\n\n👋 자동화 시스템을 종료합니다...")
            self.scheduler.shutdown()
    
    def generate_historical_receipts(self, days=7):
        """과거 트랜잭션의 영수증을 일괄 생성"""
        print(f"\n📚 최근 {days}일간의 트랜잭션 영수증 생성 중...")
        
        # 트랜잭션 조회
        transactions = self.blockchain.get_wallet_transactions(
            self.wallet_address,
            start_block=0,
            end_block='latest'
        )
        
        # 날짜 필터링
        cutoff_date = datetime.now() - timedelta(days=days)
        recent_txs = [
            tx for tx in transactions 
            if tx['timestamp'] >= cutoff_date
        ]
        
        print(f"   📊 총 {len(recent_txs)}개의 트랜잭션 발견")
        
        # ETH 가격 조회
        eth_price = self.blockchain.get_eth_price()
        
        # 일괄 생성
        generated = 0
        for tx in recent_txs:
            try:
                detailed_tx = self.blockchain.get_transaction(tx['hash'])
                if detailed_tx:
                    detailed_tx['network'] = self.network.capitalize()
                    self.pdf_generator.generate_receipt(detailed_tx, eth_price)
                    self.processed_txs.add(tx['hash'])
                    generated += 1
            except Exception as e:
                print(f"   ⚠️ 건너뜀 ({tx['hash'][:16]}...): {e}")
        
        self.save_processed_transactions()
        print(f"\n✅ 완료! {generated}개의 영수증이 생성되었습니다.")
        
        return generated

🎮 메인 실행 스크립트

이제 모든 컴포넌트를 통합해서 실제로 실행할 수 있는 메인 스크립트를 만들어보자! 사용자가 쉽게 사용할 수 있도록 CLI 인터페이스를 구현할 거야. 🎯

#!/usr/bin/env python3
"""
블록체인 결제 영수증 PDF 자동 생성 시스템
Blockchain Receipt PDF Auto-Generator
"""

import os
import sys
from dotenv import load_dotenv
from automation_engine import ReceiptAutomationEngine
import argparse

# 환경 변수 로드
load_dotenv()

def print_banner():
    """시작 배너 출력"""
    banner = """
    ╔═══════════════════════════════════════════════════════════╗
    ║                                                           ║
    ║   🔗 블록체인 영수증 자동 생성 시스템                    ║
    ║   Blockchain Receipt PDF Auto-Generator                  ║
    ║                                                           ║
    ║   Powered by Python, Web3, & ReportLab                   ║
    ║                                                           ║
    ╚═══════════════════════════════════════════════════════════╝
    """
    print(banner)

def main():
    """메인 실행 함수"""
    print_banner()
    
    # 명령줄 인자 파서 설정
    parser = argparse.ArgumentParser(
        description='블록체인 트랜잭션 영수증 자동 생성 시스템'
    )
    
    parser.add_argument(
        '--mode',
        choices=['auto', 'manual', 'history'],
        default='auto',
        help='실행 모드 선택 (auto: 자동 모니터링, manual: 수동 생성, history: 과거 내역 생성)'
    )
    
    parser.add_argument(
        '--wallet',
        type=str,
        default=os.getenv('WALLET_ADDRESS'),
        help='모니터링할 지갑 주소'
    )
    
    parser.add_argument(
        '--network',
        choices=['mainnet', 'sepolia', 'goerli'],
        default=os.getenv('NETWORK', 'sepolia'),
        help='블록체인 네트워크 선택'
    )
    
    parser.add_argument(
        '--interval',
        type=int,
        default=60,
        help='자동 모드에서 체크 주기 (초)'
    )
    
    parser.add_argument(
        '--tx-hash',
        type=str,
        help='수동 모드에서 생성할 트랜잭션 해시'
    )
    
    parser.add_argument(
        '--days',
        type=int,
        default=7,
        help='히스토리 모드에서 조회할 일수'
    )
    
    args = parser.parse_args()
    
    # 지갑 주소 확인
    if not args.wallet:
        print("❌ 오류: 지갑 주소가 설정되지 않았습니다.")
        print("   .env 파일에 WALLET_ADDRESS를 설정하거나 --wallet 옵션을 사용하세요.")
        sys.exit(1)
    
    # 자동화 엔진 초기화
    try:
        engine = ReceiptAutomationEngine(
            wallet_address=args.wallet,
            network=args.network,
            check_interval=args.interval
        )
    except Exception as e:
        print(f"❌ 초기화 실패: {e}")
        sys.exit(1)
    
    # 모드별 실행
    if args.mode == 'auto':
        # 자동 모니터링 모드
        print("\n🤖 자동 모니터링 모드로 시작합니다...")
        engine.start_monitoring()
        
    elif args.mode == 'manual':
        # 수동 생성 모드
        if not args.tx_hash:
            print("❌ 오류: 수동 모드에서는 --tx-hash 옵션이 필요합니다.")
            sys.exit(1)
        
        print(f"\n📝 수동 생성 모드: {args.tx_hash}")
        pdf_path = engine.manual_generate(args.tx_hash)
        
        if pdf_path:
            print(f"\n✅ 성공! 영수증이 생성되었습니다: {pdf_path}")
        else:
            print("\n❌ 영수증 생성에 실패했습니다.")
            sys.exit(1)
    
    elif args.mode == 'history':
        # 히스토리 생성 모드
        print(f"\n📚 히스토리 모드: 최근 {args.days}일")
        count = engine.generate_historical_receipts(days=args.days)
        
        if count > 0:
            print(f"\n✅ 성공! {count}개의 영수증이 생성되었습니다.")
        else:
            print("\n⚠️ 생성된 영수증이 없습니다.")

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n👋 프로그램을 종료합니다.")
        sys.exit(0)
    except Exception as e:
        print(f"\n❌ 예상치 못한 오류 발생: {e}")
        sys.exit(1)

python main.py --mode manual --tx-hash 0x123abc...

🔧 고급 기능 추가하기

기본 시스템은 완성했어! 이제 더 강력한 기능들을 추가해보자. 실무에서 정말 유용한 기능들이야. 💪

영수증이 생성되면 자동으로 이메일로 발송하는 기능을 추가해보자:

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import os
from dotenv import load_dotenv

load_dotenv()

class EmailSender:
    """영수증 이메일 발송 클래스"""
    
    def __init__(self):
        self.smtp_server = os.getenv('SMTP_SERVER', 'smtp.gmail.com')
        self.smtp_port = int(os.getenv('SMTP_PORT', '587'))
        self.sender_email = os.getenv('SENDER_EMAIL')
        self.sender_password = os.getenv('SENDER_PASSWORD')
    
    def send_receipt(self, recipient_email, pdf_path, transaction_data):
        """영수증 PDF를 이메일로 발송"""
        try:
            # 이메일 메시지 생성
            msg = MIMEMultipart()
            msg['From'] = self.sender_email
            msg['To'] = recipient_email
            msg['Subject'] = f"🔗 Blockchain Receipt - TX: {transaction_data['hash'][:16]}..."
            
            # 이메일 본문
            body = f"""
            안녕하세요!
            
            블록체인 트랜잭션 영수증을 보내드립니다.
            
            📋 거래 정보:
            - 트랜잭션 해시: {transaction_data['hash']}
            - 금액: {transaction_data['value']} ETH
            - 날짜: {transaction_data['timestamp']}
            - 상태: {transaction_data['status']}
            
            첨부된 PDF 파일을 확인해주세요.
            
            감사합니다!
            
            ---
            이 이메일은 자동으로 생성되었습니다.
            """
            
            msg.attach(MIMEText(body, 'plain'))
            
            # PDF 첨부
            with open(pdf_path, 'rb') as attachment:
                part = MIMEBase('application', 'octet-stream')
                part.set_payload(attachment.read())
            
            encoders.encode_base64(part)
            part.add_header(
                'Content-Disposition',
                f'attachment; filename= {os.path.basename(pdf_path)}'
            )
            msg.attach(part)
            
            # SMTP 서버 연결 및 발송
            with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
                server.starttls()
                server.login(self.sender_email, self.sender_password)
                server.send_message(msg)
            
            print(f"✅ 이메일 발송 완료: {recipient_email}")
            return True
            
        except Exception as e:
            print(f"❌ 이메일 발송 실패: {e}")
            return False

트랜잭션 정보를 데이터베이스에 저장해서 나중에 검색하고 분석할 수 있게 해보자:

import sqlite3
from datetime import datetime
import json

class DatabaseManager:
    """트랜잭션 데이터베이스 관리 클래스"""
    
    def __init__(self, db_path='receipts.db'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """데이터베이스 초기화"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS transactions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                tx_hash TEXT UNIQUE NOT NULL,
                from_address TEXT NOT NULL,
                to_address TEXT NOT NULL,
                value REAL NOT NULL,
                gas_price REAL,
                gas_used INTEGER,
                total_gas_fee REAL,
                block_number INTEGER,
                timestamp DATETIME,
                status TEXT,
                network TEXT,
                pdf_path TEXT,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_tx_hash ON transactions(tx_hash)
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_timestamp ON transactions(timestamp)
        ''')
        
        conn.commit()
        conn.close()
        
        print("✅ 데이터베이스 초기화 완료")
    
    def save_transaction(self, transaction_data, pdf_path):
        """트랜잭션 정보 저장"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                INSERT OR REPLACE INTO transactions 
                (tx_hash, from_address, to_address, value, gas_price, 
                 gas_used, total_gas_fee, block_number, timestamp, 
                 status, network, pdf_path)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                transaction_data['hash'],
                transaction_data['from'],
                transaction_data['to'],
                float(transaction_data['value']),
                float(transaction_data['gas_price']),
                transaction_data['gas_used'],
                float(transaction_data['total_gas_fee']),
                transaction_data['block_number'],
                transaction_data['timestamp'],
                transaction_data['status'],
                transaction_data.get('network', 'Unknown'),
                pdf_path
            ))
            
            conn.commit()
            conn.close()
            
            return True
            
        except Exception as e:
            print(f"❌ 데이터베이스 저장 실패: {e}")
            return False
    
    def get_transaction(self, tx_hash):
        """트랜잭션 조회"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute(
            'SELECT * FROM transactions WHERE tx_hash = ?',
            (tx_hash,)
        )
        
        result = cursor.fetchone()
        conn.close()
        
        return result
    
    def get_transactions_by_date(self, start_date, end_date):
        """날짜 범위로 트랜잭션 조회"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT * FROM transactions 
            WHERE timestamp BETWEEN ? AND ?
            ORDER BY timestamp DESC
        ''', (start_date, end_date))
        
        results = cursor.fetchall()
        conn.close()
        
        return results
    
    def get_statistics(self):
        """통계 정보 조회"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 총 트랜잭션 수
        cursor.execute('SELECT COUNT(*) FROM transactions')
        total_count = cursor.fetchone()[0]
        
        # 총 거래 금액
        cursor.execute('SELECT SUM(value) FROM transactions')
        total_value = cursor.fetchone()[0] or 0
        
        # 총 가스비
        cursor.execute('SELECT SUM(total_gas_fee) FROM transactions')
        total_gas = cursor.fetchone()[0] or 0
        
        # 네트워크별 통계
        cursor.execute('''
            SELECT network, COUNT(*), SUM(value) 
            FROM transactions 
            GROUP BY network
        ''')
        network_stats = cursor.fetchall()
        
        conn.close()
        
        return {
            'total_transactions': total_count,
            'total_value': total_value,
            'total_gas_fee': total_gas,
            'network_stats': network_stats
        }

Flask를 사용해서 간단한 웹 대시보드를 만들어보자. 브라우저에서 영수증을 확인하고 다운로드할 수 있어!

from flask import Flask, render_template, send_file, jsonify, request
from database_manager import DatabaseManager
from automation_engine import ReceiptAutomationEngine
import os

app = Flask(__name__)
db = DatabaseManager()

@app.route('/')
def index():
    """메인 대시보드"""
    stats = db.get_statistics()
    return render_template('dashboard.html', stats=stats)

@app.route('/api/transactions')
def get_transactions():
    """트랜잭션 목록 API"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    # 페이징 처리된 트랜잭션 조회
    # (실제 구현에서는 LIMIT, OFFSET 사용)
    
    return jsonify({
        'transactions': [],  # 실제 데이터
        'total': 0,
        'page': page,
        'per_page': per_page
    })

@app.route('/download/<tx_hash>')
def download_receipt(tx_hash):
    """영수증 다운로드"""
    tx = db.get_transaction(tx_hash)
    if tx and tx[12]:  # pdf_path 컬럼
        return send_file(tx[12], as_attachment=True)
    return "영수증을 찾을 수 없습니다.", 404

@app.route('/api/generate', methods=['POST'])
def generate_receipt():
    """영수증 수동 생성 API"""
    data = request.json
    tx_hash = data.get('tx_hash')
    
    if not tx_hash:
        return jsonify({'error': '트랜잭션 해시가 필요합니다'}), 400
    
    # 영수증 생성 로직
    # (실제 구현)
    
    return jsonify({'success': True, 'message': '영수증이 생성되었습니다'})

if __name__ == '__main__':
    app.run(debug=True, port=5000)

🛡️ 보안 및 최적화

실제 프로덕션 환경에서 사용하려면 보안과 성능 최적화가 필수야. 중요한 포인트들을 짚어볼게! 🔒

import redis
import json
from functools import wraps

class CacheManager:
    """Redis 캐시 매니저"""
    
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis_client = redis.Redis(
            host=host, 
            port=port, 
            db=db,
            decode_responses=True
        )
    
    def cache_transaction(self, tx_hash, data, expire=3600):
        """트랜잭션 데이터 캐싱 (1시간)"""
        key = f"tx:{tx_hash}"
        self.redis_client.setex(
            key,
            expire,
            json.dumps(data, default=str)
        )
    
    def get_cached_transaction(self, tx_hash):
        """캐시된 트랜잭션 조회"""
        key = f"tx:{tx_hash}"
        data = self.redis_client.get(key)
        return json.loads(data) if data else None

# 데코레이터로 캐싱 적용
def cached(expire=3600):
    def decorator(func):
        @wraps(func)
        def wrapper(self, tx_hash, *args, **kwargs):
            cache = CacheManager()
            
            # 캐시 확인
            cached_data = cache.get_cached_transaction(tx_hash)
            if cached_data:
                print(f"✅ 캐시 히트: {tx_hash[:16]}...")
                return cached_data
            
            # 캐시 미스 - 실제 조회
            result = func(self, tx_hash, *args, **kwargs)
            
            # 캐시 저장
            if result:
                cache.cache_transaction(tx_hash, result, expire)
            
            return result
        return wrapper
    return decorator

🚀 실전 배포 가이드

개발은 끝났어! 이제 실제 서버에 배포해서 운영해보자. Docker를 사용한 배포 방법을 알려줄게. 🐳

FROM python:3.11-slim

# 작업 디렉토리 설정
WORKDIR /app

# 시스템 패키지 업데이트 및 필요한 라이브러리 설치
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Python 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 영수증 저장 디렉토리 생성
RUN mkdir -p /app/receipts

# 환경 변수 설정
ENV PYTHONUNBUFFERED=1

# 실행 권한 부여
RUN chmod +x main.py

# 기본 명령어
CMD ["python", "main.py", "--mode", "auto"]

version: '3.8'

services:
  receipt-generator:
    build: .
    container_name: blockchain-receipt-generator
    env_file:
      - .env
    volumes:
      - ./receipts:/app/receipts
      - ./data:/app/data
    restart: unless-stopped
    networks:
      - receipt-network
  
  redis:
    image: redis:7-alpine
    container_name: receipt-redis
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    networks:
      - receipt-network
  
  postgres:
    image: postgres:15-alpine
    container_name: receipt-db
    environment:
      POSTGRES_DB: receipts
      POSTGRES_USER: receipt_user
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - postgres-data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    networks:
      - receipt-network

volumes:
  redis-data:
  postgres-data:

networks:
  receipt-network:
    driver: bridge

# Docker 이미지 빌드
docker-compose build

# 컨테이너 시작
docker-compose up -d

# 로그 확인
docker-compose logs -f receipt-generator

# 컨테이너 중지
docker-compose down

# 전체 재시작 (데이터 유지)
docker-compose restart

💡 실전 활용 사례

이 시스템을 실제로 어떻게 활용할 수 있을까? 다양한 비즈니스 시나리오를 살펴보자! 🎯

🎓 학습 리소스 및 다음 단계

여기까지 따라왔다면 정말 대단해! 🎉 하지만 블록체인 개발은 끝이 없어. 더 깊이 공부하고 싶다면 이런 것들을 추천해:

🔍 트러블슈팅 가이드

개발하다 보면 꼭 마주치는 문제들이 있어. 자주 발생하는 이슈와 해결 방법을 정리해봤어! 🛠️

🎯 마무리하며

와! 여기까지 정말 긴 여정이었어. 🎉 우리는 블록체인 기술과 파이썬을 결합해서 실용적인 자동화 시스템을 만들어봤어. 이 프로젝트를 통해 배운 것들을 정리해보자:

이 프로젝트가 블록체인 개발 여정에 도움이 되었으면 좋겠어. 궁금한 점이 있거나 더 발전시킨 버전을 만들었다면 커뮤니티에 공유해줘! 함께 성장하는 게 개발자 문화의 아름다움이니까. 😊

그리고 재능넷에서 블록체인 관련 재능을 공유하거나, 이런 시스템 개발을 의뢰받을 수도 있으니 한번 둘러보는 것도 추천해!

자, 이제 당신 차례야! 코드를 실행하고, 실험하고, 실패하고, 배우고, 성공하는 과정을 즐겨봐. 해피 코딩! 🚀✨