# services/user_wallet_registry.py from __future__ import annotations import asyncio import re import sqlite3 from dataclasses import dataclass from pathlib import Path from typing import Optional WALLET_TON = "ton" WALLET_USDT_TRC20 = "usdt_trc20" WALLET_TRON = "tron" SUPPORTED_WALLETS = { WALLET_TON: { "title": "TON", "network": "TON", }, WALLET_USDT_TRC20: { "title": "USDT", "network": "TRC20", }, WALLET_TRON: { "title": "TRON / TRX", "network": "TRON", }, } @dataclass(slots=True, frozen=True) class SavedWallet: user_id: int wallet_type: str address: str created_at: str updated_at: str @property def title(self) -> str: return str( SUPPORTED_WALLETS.get( self.wallet_type, {}, ).get( "title", self.wallet_type, ) ) @property def network(self) -> str: return str( SUPPORTED_WALLETS.get( self.wallet_type, {}, ).get( "network", "", ) ) def normalize_wallet_type( wallet_type: str, ) -> str: value = str( wallet_type or "" ).strip().lower() if value not in SUPPORTED_WALLETS: raise ValueError( "Unsupported wallet type." ) return value def normalize_address( wallet_type: str, address: str, ) -> str: wallet_type = normalize_wallet_type( wallet_type ) value = str( address or "" ).strip() value = ( value .replace("\u200c", "") .replace("\u200d", "") .replace("\ufeff", "") .strip() ) if not value: raise ValueError( "Wallet address is empty." ) if len(value) > 160: raise ValueError( "Wallet address is too long." ) if wallet_type == WALLET_TON: raw_ok = bool( re.fullmatch( r"(?:0|-1):[0-9a-fA-F]{64}", value, ) ) user_friendly_ok = bool( re.fullmatch( r"[A-Za-z0-9_-]{48}", value, ) ) if not ( raw_ok or user_friendly_ok ): raise ValueError( "آدرس TON معتبر نیست." ) return value if wallet_type in { WALLET_USDT_TRC20, WALLET_TRON, }: tron_base58 = ( "123456789ABCDEFGHJKLMNPQRSTUVWXYZ" "abcdefghijkmnopqrstuvwxyz" ) valid = ( len(value) == 34 and value.startswith("T") and all( char in tron_base58 for char in value ) ) if not valid: raise ValueError( "آدرس شبکه TRON/TRC20 معتبر نیست." ) return value raise ValueError( "Unsupported wallet type." ) class UserWalletRegistry: def __init__( self, db_path: str | Path = "matrix_bot.db", ) -> None: self.db_path = str( db_path ) def _connect( self, ) -> sqlite3.Connection: conn = sqlite3.connect( self.db_path, timeout=30, check_same_thread=False, ) conn.row_factory = ( sqlite3.Row ) conn.execute( "PRAGMA busy_timeout = 30000" ) return conn async def ensure_schema( self, ) -> None: await asyncio.to_thread( self._ensure_schema_sync ) def _ensure_schema_sync( self, ) -> None: conn = self._connect() try: conn.executescript( """ CREATE TABLE IF NOT EXISTS user_external_wallets ( user_id INTEGER NOT NULL, wallet_type TEXT NOT NULL, address TEXT NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY ( user_id, wallet_type ) ); CREATE INDEX IF NOT EXISTS idx_user_external_wallets_user ON user_external_wallets(user_id); """ ) conn.commit() finally: conn.close() async def save( self, user_id: int, wallet_type: str, address: str, ) -> SavedWallet: await self.ensure_schema() wallet_type = ( normalize_wallet_type( wallet_type ) ) address = normalize_address( wallet_type, address, ) await asyncio.to_thread( self._save_sync, int(user_id), wallet_type, address, ) result = await self.get( user_id, wallet_type, ) if result is None: raise RuntimeError( "Wallet was not saved." ) return result def _save_sync( self, user_id: int, wallet_type: str, address: str, ) -> None: conn = self._connect() try: conn.execute( """ INSERT INTO user_external_wallets ( user_id, wallet_type, address ) VALUES (?, ?, ?) ON CONFLICT( user_id, wallet_type ) DO UPDATE SET address = excluded.address, updated_at = CURRENT_TIMESTAMP """, ( user_id, wallet_type, address, ), ) conn.commit() finally: conn.close() async def get( self, user_id: int, wallet_type: str, ) -> Optional[SavedWallet]: await self.ensure_schema() return await asyncio.to_thread( self._get_sync, int(user_id), normalize_wallet_type( wallet_type ), ) def _get_sync( self, user_id: int, wallet_type: str, ) -> Optional[SavedWallet]: conn = self._connect() try: row = conn.execute( """ SELECT user_id, wallet_type, address, created_at, updated_at FROM user_external_wallets WHERE user_id = ? AND wallet_type = ? LIMIT 1 """, ( user_id, wallet_type, ), ).fetchone() if not row: return None return SavedWallet( user_id=int( row["user_id"] ), wallet_type=str( row["wallet_type"] ), address=str( row["address"] ), created_at=str( row["created_at"] or "" ), updated_at=str( row["updated_at"] or "" ), ) finally: conn.close() async def list_user_wallets( self, user_id: int, ) -> dict[str, SavedWallet]: await self.ensure_schema() return await asyncio.to_thread( self._list_sync, int(user_id), ) def _list_sync( self, user_id: int, ) -> dict[str, SavedWallet]: conn = self._connect() try: rows = conn.execute( """ SELECT user_id, wallet_type, address, created_at, updated_at FROM user_external_wallets WHERE user_id = ? ORDER BY wallet_type """, ( user_id, ), ).fetchall() result: dict[ str, SavedWallet, ] = {} for row in rows: item = SavedWallet( user_id=int( row["user_id"] ), wallet_type=str( row["wallet_type"] ), address=str( row["address"] ), created_at=str( row["created_at"] or "" ), updated_at=str( row["updated_at"] or "" ), ) result[ item.wallet_type ] = item return result finally: conn.close() async def delete( self, user_id: int, wallet_type: str, ) -> bool: await self.ensure_schema() return await asyncio.to_thread( self._delete_sync, int(user_id), normalize_wallet_type( wallet_type ), ) def _delete_sync( self, user_id: int, wallet_type: str, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ DELETE FROM user_external_wallets WHERE user_id = ? AND wallet_type = ? """, ( user_id, wallet_type, ), ) conn.commit() return ( cursor.rowcount > 0 ) finally: conn.close() __all__ = [ "UserWalletRegistry", "SavedWallet", "SUPPORTED_WALLETS", "WALLET_TON", "WALLET_USDT_TRC20", "WALLET_TRON", "normalize_wallet_type", "normalize_address", ]