""" Advanced Blockchain Ticket Verification System Secure ticket validation with distributed ledger technology """ import json import hashlib import datetime import base64 import math from typing import Dict, List, Optional, Any, Union, Tuple from dataclasses import dataclass, asdict from collections import defaultdict @dataclass class Block: index: int timestamp: str ticket_data: Dict[str, Any] previous_hash: str nonce: int hash: str class BlockchainTicketSystem: """Advanced blockchain system for ticket verification""" def __init__(self, difficulty: int = 4): self.chain: List[Block] = [] self.difficulty = difficulty self.pending_tickets = [] self.ticket_registry = {} self.ownership_transfers = [] # Create genesis block self._create_genesis_block() def _create_genesis_block(self): """Create the first block in the chain""" genesis_data = { 'type': 'genesis', 'message': 'Ticketing Platform Genesis Block', 'created_at': datetime.datetime.now().isoformat() } genesis_block = self._create_block(genesis_data, '0') self.chain.append(genesis_block) def _create_block(self, ticket_data: Dict[str, Any], previous_hash: str) -> Block: """Create a new block with proof of work""" index = len(self.chain) timestamp = datetime.datetime.now().isoformat() # Proof of Work mining nonce = 0 while True: block_string = f"{index}{timestamp}{json.dumps(ticket_data, sort_keys=True)}{previous_hash}{nonce}" block_hash = hashlib.sha256(block_string.encode()).hexdigest() if block_hash[:self.difficulty] == '0' * self.difficulty: break nonce += 1 return Block( index=index, timestamp=timestamp, ticket_data=ticket_data, previous_hash=previous_hash, nonce=nonce, hash=block_hash ) def issue_ticket(self, ticket_data: Dict[str, Any]) -> Dict[str, Any]: """Issue new ticket on blockchain""" # Add metadata ticket_data.update({ 'issued_at': datetime.datetime.now().isoformat(), 'status': 'issued', 'transfer_count': 0 }) # Create block previous_hash = self.chain[-1].hash new_block = self._create_block(ticket_data, previous_hash) # Add to chain self.chain.append(new_block) # Register ticket ticket_id = ticket_data['ticket_id'] self.ticket_registry[ticket_id] = { 'block_index': new_block.index, 'current_owner': ticket_data['owner_id'], 'status': 'active', 'issued_at': ticket_data['issued_at'], 'transfers': [] } return { 'ticket_id': ticket_id, 'block_hash': new_block.hash, 'block_index': new_block.index, 'transaction_hash': self._generate_transaction_hash(ticket_data), 'confirmed': True } def transfer_ticket(self, transfer_data: Dict[str, Any]) -> Dict[str, Any]: """Transfer ticket ownership""" ticket_id = transfer_data['ticket_id'] from_owner = transfer_data['from_owner'] to_owner = transfer_data['to_owner'] # Verify current ownership if ticket_id not in self.ticket_registry: return {'success': False, 'error': 'Ticket not found'} ticket_info = self.ticket_registry[ticket_id] if ticket_info['current_owner'] != from_owner: return {'success': False, 'error': 'Not the current owner'} # Create transfer transaction transfer_transaction = { 'type': 'transfer', 'ticket_id': ticket_id, 'from_owner': from_owner, 'to_owner': to_owner, 'timestamp': datetime.datetime.now().isoformat(), 'transfer_count': ticket_info['transfer_count'] + 1 } # Add to blockchain previous_hash = self.chain[-1].hash transfer_block = self._create_block(transfer_transaction, previous_hash) self.chain.append(transfer_block) # Update registry ticket_info['current_owner'] = to_owner ticket_info['transfer_count'] += 1 ticket_info['transfers'].append({ 'from': from_owner, 'to': to_owner, 'timestamp': transfer_transaction['timestamp'], 'block_index': transfer_block.index }) self.ownership_transfers.append(transfer_transaction) return { 'success': True, 'transfer_hash': transfer_block.hash, 'block_index': transfer_block.index, 'new_owner': to_owner } def verify_ticket(self, verification_data: Dict[str, Any]) -> Dict[str, Any]: """Comprehensive ticket verification""" ticket_id = verification_data['ticket_id'] if ticket_id not in self.ticket_registry: return {'valid': False, 'reason': 'Ticket not registered'} # Get ticket info ticket_info = self.ticket_registry[ticket_id] # Verify blockchain integrity chain_valid = self._verify_chain_integrity() # Get original ticket block ticket_block = self.chain[ticket_info['block_index']] # Check ownership current_owner = verification_data.get('current_owner') ownership_valid = current_owner == ticket_info['current_owner'] if current_owner else True # Check if ticket is voided ticket_voided = ticket_info['status'] == 'voided' verification_result = { 'valid': chain_valid and ownership_valid and not ticket_voided, 'ticket_details': { 'ticket_id': ticket_id, 'current_owner': ticket_info['current_owner'], 'issued_at': ticket_info['issued_at'], 'transfer_count': ticket_info['transfer_count'], 'status': ticket_info['status'] }, 'blockchain_valid': chain_valid, 'ownership_valid': ownership_valid, 'verification_timestamp': datetime.datetime.now().isoformat(), 'block_hash': ticket_block.hash, 'block_index': ticket_block.index } return verification_result def void_ticket(self, void_data: Dict[str, Any]) -> Dict[str, Any]: """Void a ticket (e.g., after event or due to fraud)""" ticket_id = void_data['ticket_id'] reason = void_data.get('reason', 'Event completed') if ticket_id not in self.ticket_registry: return {'success': False, 'error': 'Ticket not found'} # Create void transaction void_transaction = { 'type': 'void', 'ticket_id': ticket_id, 'reason': reason, 'timestamp': datetime.datetime.now().isoformat() } # Add to blockchain previous_hash = self.chain[-1].hash void_block = self._create_block(void_transaction, previous_hash) self.chain.append(void_block) # Update registry self.ticket_registry[ticket_id]['status'] = 'voided' return { 'success': True, 'void_hash': void_block.hash, 'block_index': void_block.index, 'voided_at': void_transaction['timestamp'] } def get_ticket_history(self, ticket_id: str) -> Dict[str, Any]: """Get complete ticket history""" if ticket_id not in self.ticket_registry: return {'error': 'Ticket not found'} ticket_info = self.ticket_registry[ticket_id] # Get all related blocks related_blocks = [] for block in self.chain: if (block.ticket_data.get('ticket_id') == ticket_id or (block.ticket_data.get('type') in ['transfer', 'void'] and block.ticket_data.get('ticket_id') == ticket_id)): related_blocks.append({ 'block_index': block.index, 'timestamp': block.timestamp, 'hash': block.hash, 'data': block.ticket_data }) return { 'ticket_id': ticket_id, 'current_info': ticket_info, 'history': related_blocks, 'transfer_count': ticket_info['transfer_count'] } def _verify_chain_integrity(self) -> bool: """Verify entire blockchain integrity""" for i in range(1, len(self.chain)): current_block = self.chain[i] previous_block = self.chain[i-1] # Verify hash linkage if current_block.previous_hash != previous_block.hash: return False # Verify block hash block_string = f"{current_block.index}{current_block.timestamp}{json.dumps(current_block.ticket_data, sort_keys=True)}{current_block.previous_hash}{current_block.nonce}" calculated_hash = hashlib.sha256(block_string.encode()).hexdigest() if calculated_hash != current_block.hash: return False # Verify proof of work if not current_block.hash[:self.difficulty] == '0' * self.difficulty: return False return True def _generate_transaction_hash(self, data: Dict[str, Any]) -> str: """Generate unique transaction hash""" data_string = json.dumps(data, sort_keys=True) return hashlib.sha256(data_string.encode()).hexdigest() def get_blockchain_stats(self) -> Dict[str, Any]: """Get blockchain statistics""" return { 'total_blocks': len(self.chain), 'total_tickets': len(self.ticket_registry), 'total_transfers': len(self.ownership_transfers), 'chain_valid': self._verify_chain_integrity(), 'difficulty': self.difficulty, 'last_block': { 'index': self.chain[-1].index, 'hash': self.chain[-1].hash, 'timestamp': self.chain[-1].timestamp } } # Global blockchain instance blockchain_system = BlockchainTicketSystem(difficulty=3) def test_blockchain_system(): """Test blockchain ticket system""" print("Testing Blockchain Ticket Verification System...") # Issue a ticket ticket_data = { 'ticket_id': 'TCKT_001', 'event_id': 'EVENT_CONCERT_2024', 'owner_id': 'USER_123', 'price': 150.00, 'seat': 'A-12', 'event_date': '2024-12-31T20:00:00Z' } issue_result = blockchain_system.issue_ticket(ticket_data) print(f"Ticket Issued: {issue_result['confirmed']}") # Verify ticket verify_result = blockchain_system.verify_ticket({'ticket_id': 'TCKT_001'}) print(f"Ticket Verified: {verify_result['valid']}") # Transfer ticket transfer_data = { 'ticket_id': 'TCKT_001', 'from_owner': 'USER_123', 'to_owner': 'USER_456' } transfer_result = blockchain_system.transfer_ticket(transfer_data) print(f"Ticket Transferred: {transfer_result['success']}") # Verify new ownership verify_result2 = blockchain_system.verify_ticket({'ticket_id': 'TCKT_001'}) print(f"New Owner Verified: {verify_result2['valid'] and verify_result2['ticket_details']['current_owner'] == 'USER_456'}") # Get ticket history history = blockchain_system.get_ticket_history('TCKT_001') print(f"History Length: {len(history['history'])}") # Get blockchain stats stats = blockchain_system.get_blockchain_stats() print(f"Blockchain Stats: {stats}") return True if __name__ == "__main__": test_blockchain_system()