# services/crypto_deposit_service.py from __future__ import annotations import asyncio import logging import re import sqlite3 from dataclasses import dataclass from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path from typing import Any, Optional logger = logging.getLogger(__name__) # ============================================================ # Currencies / statuses # ============================================================ CURRENCY_TON = "TON" CURRENCY_USDT = "USDT" CURRENCY_TRX = "TRX" TON = CURRENCY_TON USDT = CURRENCY_USDT TRX = CURRENCY_TRX SUPPORTED_CURRENCIES = { CURRENCY_TON: { "title": "TON", "network": "TON", "decimals": 9, }, CURRENCY_USDT: { "title": "USDT", "network": "TRC20", "decimals": 6, }, CURRENCY_TRX: { "title": "TRX", "network": "TRON", "decimals": 6, }, } SUPPORTED = SUPPORTED_CURRENCIES STATUS_PENDING = "pending_owner_review" STATUS_APPROVED = "approved" STATUS_REJECTED = "rejected" PENDING = STATUS_PENDING APPROVED = STATUS_APPROVED REJECTED = STATUS_REJECTED # ============================================================ # Exceptions # ============================================================ class CryptoDepositError(Exception): pass class CryptoDepositValidationError( CryptoDepositError ): pass class CryptoDepositDuplicateTx( CryptoDepositError ): pass class CryptoDepositAlreadyHandled( CryptoDepositError ): pass class CryptoDepositUserNotFound( CryptoDepositError ): pass # Compatibility aliases DepositError = ( CryptoDepositError ) DepositValidationError = ( CryptoDepositValidationError ) DuplicateTxError = ( CryptoDepositDuplicateTx ) AlreadyHandledError = ( CryptoDepositAlreadyHandled ) UserNotFoundError = ( CryptoDepositUserNotFound ) # ============================================================ # Models # ============================================================ @dataclass( slots=True, frozen=True, ) class OwnerWallet: currency: str network: str address: str updated_by: int updated_at: str @dataclass( slots=True, frozen=True, ) class CryptoDepositRequest: id: int user_id: int username: str full_name: str currency: str network: str crypto_amount: Decimal from_wallet: str destination_wallet: str tx_hash: str claimed_paid_at: str receipt_file_id: Optional[str] receipt_type: Optional[str] note: Optional[str] status: str created_at: str updated_at: str approved_by: Optional[int] approved_at: Optional[str] rate_toman: Optional[Decimal] rate_source: Optional[str] credited_toman: Optional[int] balance_before: Optional[int] balance_after: Optional[int] rejected_by: Optional[int] rejected_at: Optional[str] rejection_reason: Optional[str] @property def amount( self, ) -> Decimal: """ برای سازگاری با wallet.py که از request.amount استفاده می‌کند. """ return self.crypto_amount DepositRequest = ( CryptoDepositRequest ) @dataclass( slots=True, frozen=True, ) class ApprovalResult: request: CryptoDepositRequest credited_toman: int balance_before: int balance_after: int rate_toman: Decimal rate_source: str # ============================================================ # Validation # ============================================================ def normalize_currency( value: str, ) -> str: value = str( value or "" ).strip().upper() aliases = { "TONCOIN": TON, "TON COIN": TON, "TETHER": USDT, "USDT-TRC20": USDT, "USDT_TRC20": USDT, "TRC20": USDT, "TRON": TRX, } value = aliases.get( value, value, ) if value not in SUPPORTED: raise DepositValidationError( "ارز انتخاب‌شده پشتیبانی نمی‌شود." ) return value def parse_amount( value: Any, currency: str, ) -> Decimal: currency = normalize_currency( currency ) text = str( value or "" ).strip() text = ( text .replace(",", "") .replace("٬", "") .replace(" ", "") ) try: amount = Decimal( text ) except ( InvalidOperation, TypeError, ValueError, ) as exc: raise DepositValidationError( "مبلغ واردشده معتبر نیست." ) from exc if ( not amount.is_finite() or amount <= 0 ): raise DepositValidationError( "مبلغ واریز باید بیشتر از صفر باشد." ) if ( amount > Decimal( "1000000000" ) ): raise DepositValidationError( "مبلغ واردشده بیش از حد مجاز است." ) max_decimals = int( SUPPORTED[ currency ]["decimals"] ) decimal_places = max( 0, -amount.as_tuple().exponent, ) if ( decimal_places > max_decimals ): raise DepositValidationError( ( f"برای {currency} حداکثر " f"{max_decimals} رقم اعشار مجاز است." ) ) return amount def normalize_tx_hash( value: str, ) -> str: text = str( value or "" ).strip() if ( len(text) < 10 or len(text) > 180 ): raise DepositValidationError( "TX Hash واردشده معتبر نیست." ) if any( ch.isspace() for ch in text ): raise DepositValidationError( "TX Hash نباید فاصله داشته باشد." ) if not re.fullmatch( r"[A-Za-z0-9_:/+=.-]+", text, ): raise DepositValidationError( "TX Hash شامل کاراکتر نامعتبر است." ) # Hashهای Hex را lowercase می‌کنیم # تا با حروف بزرگ دوباره ثبت نشوند. if re.fullmatch( r"[0-9a-fA-F]+", text, ): return text.lower() # TON ممکن است Base64/Base64URL باشد. return text def validate_public_address( currency: str, address: str, ) -> str: currency = normalize_currency( currency ) value = str( address or "" ).strip() value = ( value .replace("\u200c", "") .replace("\u200d", "") .replace("\ufeff", "") .strip() ) if ( not value or len(value) > 160 ): raise DepositValidationError( "آدرس ولت معتبر نیست." ) # ======================================================== # TON # ======================================================== if currency == TON: raw_ok = bool( re.fullmatch( r"(?:0|-1):[0-9a-fA-F]{64}", value, ) ) friendly_ok = bool( re.fullmatch( r"[A-Za-z0-9_-]{48}", value, ) ) if not ( raw_ok or friendly_ok ): raise DepositValidationError( "آدرس TON معتبر نیست." ) return value # ======================================================== # TRON / USDT TRC20 # ======================================================== if currency in { USDT, TRX, }: base58 = ( "123456789" "ABCDEFGHJKLMNPQRSTUVWXYZ" "abcdefghijkmnopqrstuvwxyz" ) valid = ( len(value) == 34 and value.startswith("T") and all( ch in base58 for ch in value ) ) if not valid: raise DepositValidationError( ( "آدرس شبکه TRON/TRC20 معتبر نیست " "و باید با T شروع شود." ) ) return value raise DepositValidationError( "ارز انتخاب‌شده پشتیبانی نمی‌شود." ) # wallet.py از این اسم استفاده می‌کند. validate_address = ( validate_public_address ) def _decimal_or_none( value: Any, ) -> Optional[Decimal]: if value is None: return None try: result = Decimal( str(value) ) except ( InvalidOperation, TypeError, ValueError, ): return None if not result.is_finite(): return None return result # ============================================================ # Main Service # ============================================================ class CryptoDepositService: """ واریز ارزی دستی با بررسی مالک. روند: مالک ولت دریافت ثبت می‌کند ↓ کاربر ارز واریز می‌کند ↓ درخواست + TX Hash ثبت می‌شود ↓ pending_owner_review ↓ مالک تأیید می‌کند ↓ مقدار ارز × نرخ تومان ↓ balance_toman محافظ‌های مالی: - TX Hash یکتا - Snapshot آدرس مقصد - BEGIN IMMEDIATE - شارژ فقط برای Pending - Ledger مستقل - تأیید دوباره = بدون شارژ مجدد """ def __init__( self, db_path: str | Path = "matrix_bot.db", ) -> None: self.db_path = str( db_path ) # ======================================================== # Database Connection # ======================================================== 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" ) conn.execute( "PRAGMA foreign_keys = ON" ) return conn # ======================================================== # Schema # ======================================================== 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 owner_crypto_wallets ( currency TEXT PRIMARY KEY, network TEXT NOT NULL, address TEXT NOT NULL, updated_by INTEGER NOT NULL, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS crypto_deposit_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, username TEXT NOT NULL DEFAULT '', full_name TEXT NOT NULL DEFAULT '', currency TEXT NOT NULL, network TEXT NOT NULL, crypto_amount TEXT NOT NULL, from_wallet TEXT NOT NULL, destination_wallet TEXT NOT NULL, tx_hash TEXT NOT NULL, tx_hash_normalized TEXT NOT NULL UNIQUE, claimed_paid_at TEXT NOT NULL, receipt_file_id TEXT, receipt_type TEXT, note TEXT, status TEXT NOT NULL DEFAULT 'pending_owner_review', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, approved_by INTEGER, approved_at TIMESTAMP, rate_toman TEXT, rate_source TEXT, credited_toman INTEGER, balance_before INTEGER, balance_after INTEGER, rejected_by INTEGER, rejected_at TIMESTAMP, rejection_reason TEXT ); CREATE INDEX IF NOT EXISTS idx_crypto_deposit_status_created ON crypto_deposit_requests( status, created_at ); CREATE INDEX IF NOT EXISTS idx_crypto_deposit_user_created ON crypto_deposit_requests( user_id, created_at ); CREATE TABLE IF NOT EXISTS crypto_deposit_ledger ( id INTEGER PRIMARY KEY AUTOINCREMENT, request_id INTEGER NOT NULL UNIQUE, user_id INTEGER NOT NULL, currency TEXT NOT NULL, crypto_amount TEXT NOT NULL, rate_toman TEXT NOT NULL, rate_source TEXT NOT NULL, credited_toman INTEGER NOT NULL, balance_before INTEGER NOT NULL, balance_after INTEGER NOT NULL, approved_by INTEGER NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); """ ) conn.commit() finally: conn.close() # ======================================================== # Owner Wallets # ======================================================== async def set_owner_wallet( self, currency: str, address: str, admin_id: Optional[ int ] = None, *, updated_by: Optional[ int ] = None, ) -> OwnerWallet: await self.ensure_schema() currency = normalize_currency( currency ) address = validate_address( currency, address, ) if admin_id is None: admin_id = ( updated_by ) try: admin_id = int( admin_id ) except ( TypeError, ValueError, ) as exc: raise DepositValidationError( "شناسه مدیر معتبر نیست." ) from exc if ( admin_id <= 0 ): raise DepositValidationError( "شناسه مدیر معتبر نیست." ) network = str( SUPPORTED[ currency ]["network"] ) await asyncio.to_thread( self._set_owner_wallet_sync, currency, network, address, admin_id, ) wallet = ( await self.get_owner_wallet( currency ) ) if wallet is None: raise DepositError( "ثبت ولت مقصد انجام نشد." ) return wallet def _set_owner_wallet_sync( self, currency: str, network: str, address: str, admin_id: int, ) -> None: conn = self._connect() try: conn.execute( """ INSERT INTO owner_crypto_wallets ( currency, network, address, updated_by, updated_at ) VALUES ( ?, ?, ?, ?, CURRENT_TIMESTAMP ) ON CONFLICT(currency) DO UPDATE SET network = excluded.network, address = excluded.address, updated_by = excluded.updated_by, updated_at = CURRENT_TIMESTAMP """, ( currency, network, address, admin_id, ), ) conn.commit() finally: conn.close() async def get_owner_wallet( self, currency: str, ) -> Optional[ OwnerWallet ]: await self.ensure_schema() return await asyncio.to_thread( self._get_owner_wallet_sync, normalize_currency( currency ), ) def _get_owner_wallet_sync( self, currency: str, ) -> Optional[ OwnerWallet ]: conn = self._connect() try: row = conn.execute( """ SELECT currency, network, address, updated_by, updated_at FROM owner_crypto_wallets WHERE currency = ? LIMIT 1 """, ( currency, ), ).fetchone() if not row: return None return OwnerWallet( currency=str( row["currency"] ), network=str( row["network"] ), address=str( row["address"] ), updated_by=int( row["updated_by"] ), updated_at=str( row["updated_at"] or "" ), ) finally: conn.close() async def list_owner_wallets( self, ) -> dict[ str, OwnerWallet, ]: await self.ensure_schema() return await asyncio.to_thread( self._list_owner_wallets_sync ) def _list_owner_wallets_sync( self, ) -> dict[ str, OwnerWallet, ]: conn = self._connect() try: rows = conn.execute( """ SELECT currency, network, address, updated_by, updated_at FROM owner_crypto_wallets ORDER BY currency """ ).fetchall() result: dict[ str, OwnerWallet, ] = {} for row in rows: item = OwnerWallet( currency=str( row["currency"] ), network=str( row["network"] ), address=str( row["address"] ), updated_by=int( row["updated_by"] ), updated_at=str( row["updated_at"] or "" ), ) result[ item.currency ] = item return result finally: conn.close() # ======================================================== # Create Deposit Request # ======================================================== async def create_request( self, *, user_id: int, username: str, full_name: str, currency: str, crypto_amount: Any, from_wallet: str, tx_hash: str, claimed_paid_at: str, receipt_file_id: Optional[ str ] = None, receipt_type: Optional[ str ] = None, note: Optional[ str ] = None, ) -> CryptoDepositRequest: await self.ensure_schema() currency = normalize_currency( currency ) amount = parse_amount( crypto_amount, currency, ) from_wallet = validate_address( currency, from_wallet, ) tx_hash_normalized = ( normalize_tx_hash( tx_hash ) ) owner_wallet = ( await self.get_owner_wallet( currency ) ) if owner_wallet is None: raise DepositValidationError( ( "ولت مقصد این ارز هنوز " "توسط مالک تنظیم نشده است." ) ) try: user_id = int( user_id ) except ( TypeError, ValueError, ) as exc: raise DepositValidationError( "شناسه کاربر معتبر نیست." ) from exc if ( user_id <= 0 ): raise DepositValidationError( "شناسه کاربر معتبر نیست." ) claimed_paid_at = str( claimed_paid_at or "" ).strip()[:100] if not claimed_paid_at: raise DepositValidationError( "تاریخ/زمان واریز مشخص نشده است." ) username = str( username or "" ).strip()[:64] full_name = str( full_name or "" ).strip()[:160] raw_tx_hash = str( tx_hash or "" ).strip() receipt_file_id = ( str( receipt_file_id ).strip()[:512] if receipt_file_id else None ) receipt_type = ( str( receipt_type ) .strip() .lower()[:30] if receipt_type else None ) if receipt_type not in { None, "photo", "document", }: raise DepositValidationError( "نوع رسید معتبر نیست." ) note = ( str( note ).strip()[:1000] if note else None ) request_id = ( await asyncio.to_thread( self._create_request_sync, user_id, username, full_name, currency, owner_wallet.network, format( amount, "f", ), from_wallet, owner_wallet.address, raw_tx_hash, tx_hash_normalized, claimed_paid_at, receipt_file_id, receipt_type, note, ) ) request = ( await self.get_request( request_id ) ) if request is None: raise DepositError( ( "درخواست ساخته شد اما " "قابل بازیابی نیست." ) ) return request def _create_request_sync( self, user_id: int, username: str, full_name: str, currency: str, network: str, crypto_amount: str, from_wallet: str, destination_wallet: str, tx_hash: str, tx_hash_normalized: str, claimed_paid_at: str, receipt_file_id: Optional[ str ], receipt_type: Optional[ str ], note: Optional[ str ], ) -> int: conn = self._connect() try: try: cursor = conn.execute( """ INSERT INTO crypto_deposit_requests ( user_id, username, full_name, currency, network, crypto_amount, from_wallet, destination_wallet, tx_hash, tx_hash_normalized, claimed_paid_at, receipt_file_id, receipt_type, note, status ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) """, ( user_id, username, full_name, currency, network, crypto_amount, from_wallet, destination_wallet, tx_hash, tx_hash_normalized, claimed_paid_at, receipt_file_id, receipt_type, note, PENDING, ), ) conn.commit() except sqlite3.IntegrityError as exc: if ( "unique" in str( exc ).lower() ): raise DuplicateTxError( ( "این TX Hash قبلاً " "در سیستم ثبت شده است." ) ) from exc raise return int( cursor.lastrowid ) finally: conn.close() # ======================================================== # Row To Object # ======================================================== @staticmethod def _row_to_request( row: sqlite3.Row, ) -> CryptoDepositRequest: return CryptoDepositRequest( id=int( row["id"] ), user_id=int( row["user_id"] ), username=str( row["username"] or "" ), full_name=str( row["full_name"] or "" ), currency=str( row["currency"] ), network=str( row["network"] ), crypto_amount=Decimal( str( row["crypto_amount"] ) ), from_wallet=str( row["from_wallet"] ), destination_wallet=str( row["destination_wallet"] ), tx_hash=str( row["tx_hash"] ), claimed_paid_at=str( row["claimed_paid_at"] ), receipt_file_id=( str( row["receipt_file_id"] ) if row[ "receipt_file_id" ] else None ), receipt_type=( str( row["receipt_type"] ) if row[ "receipt_type" ] else None ), note=( str( row["note"] ) if row["note"] else None ), status=str( row["status"] ), created_at=str( row["created_at"] or "" ), updated_at=str( row["updated_at"] or "" ), approved_by=( int( row["approved_by"] ) if row[ "approved_by" ] is not None else None ), approved_at=( str( row["approved_at"] ) if row[ "approved_at" ] else None ), rate_toman=( _decimal_or_none( row["rate_toman"] ) ), rate_source=( str( row["rate_source"] ) if row[ "rate_source" ] else None ), credited_toman=( int( row["credited_toman"] ) if row[ "credited_toman" ] is not None else None ), balance_before=( int( row["balance_before"] ) if row[ "balance_before" ] is not None else None ), balance_after=( int( row["balance_after"] ) if row[ "balance_after" ] is not None else None ), rejected_by=( int( row["rejected_by"] ) if row[ "rejected_by" ] is not None else None ), rejected_at=( str( row["rejected_at"] ) if row[ "rejected_at" ] else None ), rejection_reason=( str( row[ "rejection_reason" ] ) if row[ "rejection_reason" ] else None ), ) # ======================================================== # Get Request # ======================================================== async def get_request( self, request_id: int, ) -> Optional[ CryptoDepositRequest ]: await self.ensure_schema() return await asyncio.to_thread( self._get_request_sync, int( request_id ), ) def _get_request_sync( self, request_id: int, ) -> Optional[ CryptoDepositRequest ]: conn = self._connect() try: row = conn.execute( """ SELECT * FROM crypto_deposit_requests WHERE id = ? LIMIT 1 """, ( request_id, ), ).fetchone() if not row: return None return self._row_to_request( row ) finally: conn.close() # ======================================================== # Pending Requests # ======================================================== async def list_pending( self, limit: int = 20, ) -> list[ CryptoDepositRequest ]: await self.ensure_schema() limit = max( 1, min( int( limit ), 100, ), ) return await asyncio.to_thread( self._list_pending_sync, limit, ) def _list_pending_sync( self, limit: int, ) -> list[ CryptoDepositRequest ]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM crypto_deposit_requests WHERE status = ? ORDER BY id ASC LIMIT ? """, ( PENDING, limit, ), ).fetchall() return [ self._row_to_request( row ) for row in rows ] finally: conn.close() # ======================================================== # User Requests # ======================================================== async def list_user_requests( self, user_id: int, limit: int = 10, ) -> list[ CryptoDepositRequest ]: await self.ensure_schema() limit = max( 1, min( int( limit ), 50, ), ) return await asyncio.to_thread( self._list_user_requests_sync, int( user_id ), limit, ) def _list_user_requests_sync( self, user_id: int, limit: int, ) -> list[ CryptoDepositRequest ]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM crypto_deposit_requests WHERE user_id = ? ORDER BY id DESC LIMIT ? """, ( user_id, limit, ), ).fetchall() return [ self._row_to_request( row ) for row in rows ] finally: conn.close() # سازگاری با wallet.py فعلی async def list_user( self, user_id: int, limit: int = 10, ) -> list[ CryptoDepositRequest ]: return await self.list_user_requests( user_id, limit, ) # ======================================================== # Detect User Balance Schema # ======================================================== @staticmethod def _users_schema( conn: sqlite3.Connection, ) -> tuple[ str, str, ]: rows = conn.execute( "PRAGMA table_info(users)" ).fetchall() columns = { str( row["name"] ) for row in rows } if ( "user_id" in columns ): user_key = ( "user_id" ) elif ( "id" in columns ): user_key = ( "id" ) else: raise UserNotFoundError( ( "ستون شناسه کاربر در " "جدول users پیدا نشد." ) ) if ( "balance_toman" in columns ): balance_key = ( "balance_toman" ) elif ( "balance" in columns ): balance_key = ( "balance" ) else: raise UserNotFoundError( ( "ستون موجودی تومانی " "کاربر پیدا نشد." ) ) return ( user_key, balance_key, ) # ======================================================== # Optional wallet_transactions Mirror # ======================================================== @staticmethod def _mirror_wallet_transaction( conn: sqlite3.Connection, *, request_id: int, user_id: int, currency: str, amount: int, before: int, after: int, ) -> None: try: rows = conn.execute( """ PRAGMA table_info( wallet_transactions ) """ ).fetchall() columns = { str( row["name"] ) for row in rows } required = { "user_id", "type", "amount", "balance_before", "balance_after", } if not required.issubset( columns ): return names = [ "user_id", "type", "amount", "balance_before", "balance_after", ] values: list[ Any ] = [ user_id, "crypto_deposit_approved", amount, before, after, ] if ( "description" in columns ): names.append( "description" ) values.append( ( f"تأیید واریز {currency} " f"- درخواست #{request_id}" ) ) if ( "reference_id" in columns ): names.append( "reference_id" ) values.append( ( "crypto_deposit:" f"{request_id}" ) ) placeholders = ( ", ".join( "?" for _ in names ) ) conn.execute( f""" INSERT INTO wallet_transactions ( {", ".join(names)} ) VALUES ( {placeholders} ) """, values, ) except Exception: # جدول wallet_transactions اجباری نیست. # Ledger اصلی در crypto_deposit_ledger ذخیره می‌شود. logger.warning( ( "Could not mirror crypto deposit " "into wallet_transactions." ), exc_info=True, ) # ======================================================== # Approve # ======================================================== async def approve_request( self, request_id: int, *, admin_id: int, rate_toman: Any, rate_source: str, ) -> ApprovalResult: await self.ensure_schema() rate = _decimal_or_none( rate_toman ) if ( rate is None or rate <= 0 ): raise DepositValidationError( "نرخ تومانی معتبر نیست." ) return await asyncio.to_thread( self._approve_sync, int( request_id ), int( admin_id ), rate, str( rate_source or "unknown" ).strip()[:120], ) # wallet.py فعلی از approve() استفاده می‌کند. async def approve( self, request_id: int, *, admin_id: int, rate_toman: Any, rate_source: str, ) -> ApprovalResult: return await self.approve_request( request_id, admin_id=( admin_id ), rate_toman=( rate_toman ), rate_source=( rate_source ), ) def _approve_sync( self, request_id: int, admin_id: int, rate_toman: Decimal, rate_source: str, ) -> ApprovalResult: conn = self._connect() try: # ================================================= # LOCK DATABASE WRITE # ================================================= conn.execute( "BEGIN IMMEDIATE" ) # ================================================= # Load Request # ================================================= row = conn.execute( """ SELECT * FROM crypto_deposit_requests WHERE id = ? LIMIT 1 """, ( request_id, ), ).fetchone() if not row: raise DepositValidationError( "درخواست پیدا نشد." ) request = ( self._row_to_request( row ) ) # ================================================= # Anti Double Approval # ================================================= if ( request.status != PENDING ): raise AlreadyHandledError( ( "این درخواست قبلاً " "بررسی شده است." ) ) # ================================================= # Crypto -> Toman # ================================================= credited_decimal = ( request.amount * rate_toman ).quantize( Decimal("1"), rounding=ROUND_HALF_UP, ) credited_toman = int( credited_decimal ) if ( credited_toman <= 0 ): raise DepositValidationError( ( "مبلغ تومانی محاسبه‌شده " "نامعتبر است." ) ) if ( credited_toman > 10**15 ): raise DepositValidationError( ( "مبلغ تومانی خارج " "از محدوده امن است." ) ) # ================================================= # User Schema # ================================================= ( user_key, balance_key, ) = self._users_schema( conn ) user_row = conn.execute( f""" SELECT {balance_key} AS balance FROM users WHERE {user_key} = ? LIMIT 1 """, ( request.user_id, ), ).fetchone() if not user_row: raise UserNotFoundError( ( "کاربر در جدول users " "پیدا نشد." ) ) balance_before = int( user_row[ "balance" ] or 0 ) balance_after = ( balance_before + credited_toman ) # ================================================= # Increase User Toman Balance # ================================================= cursor = conn.execute( f""" UPDATE users SET {balance_key} = ? WHERE {user_key} = ? """, ( balance_after, request.user_id, ), ) if ( cursor.rowcount != 1 ): raise UserNotFoundError( ( "افزایش موجودی کاربر " "انجام نشد." ) ) # ================================================= # Approve Request # ================================================= cursor = conn.execute( """ UPDATE crypto_deposit_requests SET status = ?, approved_by = ?, approved_at = CURRENT_TIMESTAMP, rate_toman = ?, rate_source = ?, credited_toman = ?, balance_before = ?, balance_after = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( APPROVED, admin_id, format( rate_toman, "f", ), rate_source, credited_toman, balance_before, balance_after, request_id, PENDING, ), ) if ( cursor.rowcount != 1 ): raise AlreadyHandledError( ( "درخواست هم‌زمان توسط " "مدیر دیگری بررسی شد." ) ) # ================================================= # Audit Ledger # ================================================= conn.execute( """ INSERT INTO crypto_deposit_ledger ( request_id, user_id, currency, crypto_amount, rate_toman, rate_source, credited_toman, balance_before, balance_after, approved_by ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) """, ( request_id, request.user_id, request.currency, format( request.amount, "f", ), format( rate_toman, "f", ), rate_source, credited_toman, balance_before, balance_after, admin_id, ), ) # ================================================= # Optional Existing Wallet History # ================================================= self._mirror_wallet_transaction( conn, request_id=( request_id ), user_id=( request.user_id ), currency=( request.currency ), amount=( credited_toman ), before=( balance_before ), after=( balance_after ), ) # ================================================= # One Atomic Commit # ================================================= conn.commit() except Exception: conn.rollback() raise finally: conn.close() # ==================================================== # Reload Updated Request # ==================================================== updated = ( self._get_request_sync( request_id ) ) if updated is None: raise DepositError( ( "درخواست پس از تأیید " "قابل بازیابی نیست." ) ) return ApprovalResult( request=( updated ), credited_toman=( credited_toman ), balance_before=( balance_before ), balance_after=( balance_after ), rate_toman=( rate_toman ), rate_source=( rate_source ), ) # ======================================================== # Reject # ======================================================== async def reject_request( self, request_id: int, *, admin_id: int, reason: str, ) -> CryptoDepositRequest: await self.ensure_schema() reason = str( reason or "" ).strip()[:1000] if not reason: raise DepositValidationError( "دلیل رد درخواست را وارد کنید." ) await asyncio.to_thread( self._reject_sync, int( request_id ), int( admin_id ), reason, ) result = ( await self.get_request( request_id ) ) if result is None: raise DepositError( ( "درخواست پس از رد " "قابل بازیابی نیست." ) ) return result # wallet.py فعلی از reject() استفاده می‌کند. async def reject( self, request_id: int, *, admin_id: int, reason: str, ) -> CryptoDepositRequest: return await self.reject_request( request_id, admin_id=( admin_id ), reason=( reason ), ) def _reject_sync( self, request_id: int, admin_id: int, reason: str, ) -> None: conn = self._connect() try: # جلوگیری از بررسی همزمان conn.execute( "BEGIN IMMEDIATE" ) row = conn.execute( """ SELECT status FROM crypto_deposit_requests WHERE id = ? LIMIT 1 """, ( request_id, ), ).fetchone() if not row: raise DepositValidationError( "درخواست پیدا نشد." ) if ( str( row["status"] ) != PENDING ): raise AlreadyHandledError( ( "این درخواست قبلاً " "بررسی شده است." ) ) cursor = conn.execute( """ UPDATE crypto_deposit_requests SET status = ?, rejected_by = ?, rejected_at = CURRENT_TIMESTAMP, rejection_reason = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( REJECTED, admin_id, reason, request_id, PENDING, ), ) if ( cursor.rowcount != 1 ): raise AlreadyHandledError( ( "درخواست هم‌زمان توسط " "مدیر دیگری بررسی شد." ) ) conn.commit() except Exception: conn.rollback() raise finally: conn.close() # ============================================================ # Public Exports # ============================================================ __all__ = [ "CryptoDepositService", "CryptoDepositRequest", "DepositRequest", "OwnerWallet", "ApprovalResult", "CryptoDepositError", "CryptoDepositValidationError", "CryptoDepositDuplicateTx", "CryptoDepositAlreadyHandled", "CryptoDepositUserNotFound", "DepositError", "DepositValidationError", "DuplicateTxError", "AlreadyHandledError", "UserNotFoundError", "CURRENCY_TON", "CURRENCY_USDT", "CURRENCY_TRX", "SUPPORTED_CURRENCIES", "TON", "USDT", "TRX", "SUPPORTED", "STATUS_PENDING", "STATUS_APPROVED", "STATUS_REJECTED", "PENDING", "APPROVED", "REJECTED", "normalize_currency", "parse_amount", "normalize_tx_hash", "validate_public_address", "validate_address", ]