Spaces:
Sleeping
Sleeping
| """Stellar SDK Client for ZeroSense | |
| Handles all Stellar blockchain interactions: | |
| - Contract invocations (Soroban) | |
| - XLM payments | |
| - Event streaming | |
| - ZREP token management | |
| """ | |
| import os | |
| from typing import Optional | |
| from stellar_sdk import Keypair, Network, Server | |
| # stellar-sdk v10 moved SorobanServer — try both paths | |
| try: | |
| from stellar_sdk import SorobanServer | |
| except ImportError: | |
| try: | |
| from stellar_sdk.soroban_server import SorobanServer | |
| except ImportError: | |
| SorobanServer = None | |
| class StellarClient: | |
| def __init__(self, secret_key: str, network: str = "testnet"): | |
| if secret_key: | |
| try: | |
| self.keypair = Keypair.from_secret(secret_key) | |
| except Exception as e: | |
| print(f"[Stellar] ⚠️ Invalid secret key ({e}) — falling back to random keypair") | |
| self.keypair = Keypair.random() | |
| else: | |
| self.keypair = Keypair.random() | |
| print(f"[Stellar] ⚠️ No secret key — using random keypair for dev") | |
| print(f"[Stellar] Public key: {self.keypair.public_key}") | |
| self.network = network | |
| if network == "testnet": | |
| self.network_passphrase = Network.TESTNET_NETWORK_PASSPHRASE | |
| self.horizon_url = "https://horizon-testnet.stellar.org" | |
| self.soroban_url = "https://soroban-testnet.stellar.org" | |
| else: | |
| self.network_passphrase = Network.PUBLIC_NETWORK_PASSPHRASE | |
| self.horizon_url = "https://horizon.stellar.org" | |
| self.soroban_url = "https://soroban.stellar.org" | |
| try: | |
| self.server = Server(horizon_url=self.horizon_url) | |
| except Exception as e: | |
| print(f"[Stellar] ⚠️ Horizon server init failed: {e}") | |
| self.server = None | |
| try: | |
| self.soroban = SorobanServer(server_url=self.soroban_url) if SorobanServer else None | |
| except Exception as e: | |
| print(f"[Stellar] ⚠️ Soroban server init failed: {e}") | |
| self.soroban = None | |
| # Contract IDs from .env | |
| self.verifier_contract = os.getenv("VERIFIER_CONTRACT_ID", "") | |
| self.payment_contract = os.getenv("PAYMENT_CONTRACT_ID", "") | |
| self.reputation_contract = os.getenv("REPUTATION_CONTRACT_ID", "") | |
| self.insurance_contract = os.getenv("INSURANCE_CONTRACT_ID", "") | |
| print(f"[Stellar] Connected to {network}") | |
| print(f"[Stellar] Account: {self.keypair.public_key[:20]}...") | |
| async def verify_proof_on_chain( | |
| self, | |
| proof_hex: str, | |
| robot_id: str, | |
| task_id: str, | |
| model_hash: str, | |
| confidence: int, | |
| action_type: int, | |
| ) -> str: | |
| """Submit ZK proof to ZeroSenseVerifier Soroban contract.""" | |
| print(f"[Stellar] Submitting proof for task {task_id}...") | |
| # TODO: Build Soroban transaction with stellar_sdk TransactionBuilder | |
| mock_tx = f"tx_{task_id}_verified" | |
| print(f"[Stellar] ✅ Proof verified on-chain: {mock_tx}") | |
| return mock_tx | |
| async def claim_payment(self, task_id: str, confidence: int) -> str: | |
| """Trigger XLM payment via RobotPaymentRouter contract.""" | |
| print(f"[Stellar] ⚡ Claiming payment for task {task_id}... (confidence: {confidence}%)") | |
| mock_tx = f"tx_{task_id}_paid" | |
| print(f"[Stellar] ✅ XLM paid: {mock_tx}") | |
| return mock_tx | |
| async def file_insurance_claim( | |
| self, robot_id: str, proof_hash: str, claim_amount: int | |
| ) -> int: | |
| """File insurance claim via InsuranceClaim contract.""" | |
| print(f"[Stellar] 🛡️ Filing insurance claim for robot {robot_id[:8]}...") | |
| return 1 # claim_id | |
| async def get_zrep_balance(self, robot_id: str) -> int: | |
| """Query ZREP token balance for a robot.""" | |
| return 0 # TODO: Query ZRepToken contract | |
| async def mint_soulbound_identity_token( | |
| self, robot_id: str, fingerprint_hash: str | |
| ) -> str: | |
| """Mint soul-bound robot identity token on Stellar.""" | |
| print(f"[Stellar] 🧬 Minting identity token for {robot_id}...") | |
| return f"tx_identity_{robot_id[:8]}" | |
| async def get_recent_contract_events(self, event_type: str) -> list: | |
| """Stream recent contract events from Stellar ledger.""" | |
| # TODO: Use Horizon /effects or Soroban event streaming | |
| return [] | |