carbonoffsettradingplatform-45zzph/app/services/blockchain.py

from web3 import Web3
from eth_account import Account
from typing import Optional, Dict, Any
import os
from datetime import datetime

class BlockchainService:
    def __init__(self):
        # Use environment variable for RPC URL (defaults to local for development)
        self.rpc_url = os.getenv("BLOCKCHAIN_RPC_URL", "http://localhost:8545")
        self.w3 = Web3(Web3.HTTPProvider(self.rpc_url))
        self.contract_abi = self._get_carbon_token_abi()
        
    def _get_carbon_token_abi(self) -> list:
        # Simplified ABI for a carbon credit token contract
        return [
            {
                "inputs": [
                    {"name": "to", "type": "address"},
                    {"name": "tokenId", "type": "uint256"},
                    {"name": "credits", "type": "uint256"}
                ],
                "name": "mintCarbonCredit",
                "outputs": [{"name": "", "type": "bool"}],
                "type": "function"
            },
            {
                "inputs": [
                    {"name": "from", "type": "address"},
                    {"name": "to", "type": "address"},
                    {"name": "tokenId", "type": "uint256"}
                ],
                "name": "transferFrom",
                "outputs": [{"name": "", "type": "bool"}],
                "type": "function"
            },
            {
                "inputs": [{"name": "tokenId", "type": "uint256"}],
                "name": "ownerOf",
                "outputs": [{"name": "", "type": "address"}],
                "type": "function"
            }
        ]
    
    def validate_wallet_address(self, address: str) -> bool:
        """Validate if the provided address is a valid Ethereum address"""
        try:
            return Web3.is_address(address) and Web3.is_checksum_address(Web3.to_checksum_address(address))
        except:
            return False
    
    def generate_wallet(self) -> Dict[str, str]:
        """Generate a new wallet for testing purposes"""
        account = Account.create()
        return {
            "address": account.address,
            "private_key": account.key.hex(),
            "public_key": account.address  # In Ethereum, address is derived from public key
        }
    
    def get_wallet_balance(self, address: str) -> Optional[float]:
        """Get ETH balance for a wallet address"""
        try:
            if not self.validate_wallet_address(address):
                return None
            
            balance_wei = self.w3.eth.get_balance(Web3.to_checksum_address(address))
            balance_eth = self.w3.from_wei(balance_wei, 'ether')
            return float(balance_eth)
        except Exception as e:
            print(f"Error getting balance for {address}: {e}")
            return None
    
    def create_carbon_token_transaction(
        self, 
        contract_address: str,
        from_address: str,
        to_address: str,
        token_id: int,
        private_key: str = None
    ) -> Optional[Dict[str, Any]]:
        """Create a transaction to transfer carbon credits"""
        try:
            if not all([
                self.validate_wallet_address(contract_address),
                self.validate_wallet_address(from_address),
                self.validate_wallet_address(to_address)
            ]):
                return None
            
            contract = self.w3.eth.contract(
                address=Web3.to_checksum_address(contract_address),
                abi=self.contract_abi
            )
            
            # Build transaction
            transaction = contract.functions.transferFrom(
                Web3.to_checksum_address(from_address),
                Web3.to_checksum_address(to_address),
                token_id
            ).build_transaction({
                'from': Web3.to_checksum_address(from_address),
                'gas': 200000,
                'gasPrice': self.w3.to_wei('20', 'gwei'),
                'nonce': self.w3.eth.get_transaction_count(Web3.to_checksum_address(from_address))
            })
            
            return {
                "transaction": transaction,
                "contract_address": contract_address,
                "from_address": from_address,
                "to_address": to_address,
                "token_id": token_id,
                "created_at": datetime.utcnow().isoformat()
            }
            
        except Exception as e:
            print(f"Error creating transaction: {e}")
            return None
    
    def sign_and_send_transaction(self, transaction_data: Dict[str, Any], private_key: str) -> Optional[str]:
        """Sign and send a transaction to the blockchain"""
        try:
            transaction = transaction_data["transaction"]
            signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key)
            tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)
            return tx_hash.hex()
        except Exception as e:
            print(f"Error signing/sending transaction: {e}")
            return None
    
    def get_transaction_receipt(self, tx_hash: str) -> Optional[Dict[str, Any]]:
        """Get transaction receipt from blockchain"""
        try:
            receipt = self.w3.eth.get_transaction_receipt(tx_hash)
            return {
                "transaction_hash": receipt["transactionHash"].hex(),
                "block_number": receipt["blockNumber"],
                "gas_used": receipt["gasUsed"],
                "status": receipt["status"]  # 1 for success, 0 for failure
            }
        except Exception as e:
            print(f"Error getting transaction receipt: {e}")
            return None
    
    def verify_token_ownership(self, contract_address: str, token_id: int, owner_address: str) -> bool:
        """Verify if an address owns a specific token"""
        try:
            if not all([
                self.validate_wallet_address(contract_address),
                self.validate_wallet_address(owner_address)
            ]):
                return False
            
            contract = self.w3.eth.contract(
                address=Web3.to_checksum_address(contract_address),
                abi=self.contract_abi
            )
            
            actual_owner = contract.functions.ownerOf(token_id).call()
            return actual_owner.lower() == owner_address.lower()
            
        except Exception as e:
            print(f"Error verifying ownership: {e}")
            return False

# Global instance
blockchain_service = BlockchainService()