diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..5e3f01d --- /dev/null +++ b/.flake8 @@ -0,0 +1,13 @@ +[flake8] +max-line-length = 127 +extend-ignore = E203, W503, E501 +exclude = + .git, + __pycache__, + build, + dist, + .eggs, + *.egg-info, + .venv, + venv, + example diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 24430fb..383fd9d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -32,9 +32,9 @@ jobs: run: | pip install flake8 # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics + flake8 sofizpay tests --count --select=E9,F63,F7,F82 --show-source --statistics # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics + flake8 sofizpay tests --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics - name: Run tests with pytest run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 82f5e70..d734bef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,60 +5,46 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.2.0] - 2026-08-22 - - - -## [1.0.2] - 2025-08-01 -sd - -## [1.0.2] - 2025-08-01 -sd +### Added +- **CIB & EDAHABIA Sandbox Environment Support**: + - `is_sandbox` parameter in `SofizPayClient` and `make_cib_transaction`. + - Dedicated `make_sandbox_cib_transaction` method. + - Dedicated `check_sandbox_cib_status` method. + - Support for `webhook_url`, `invoice_id`, `language`, `keep_return_url`, and `redirect` parameters. +- **CIB Status Verification**: + - `check_cib_transaction` and `check_cib_status` methods. + - Parsing and status classification (`paid`, `pending`, `Amount`, `errorMessage`, `orderStatus`). +- **Algerian Utility & Telecom Services**: + - `get_products`: Products catalog search & listing. + - `get_operation_history`: Service operation history. + - `get_operation_details`: Individual operation status tracking by UUID. + - `pay_bill`, `pay_ade_bill`: Water bill payment for Algรฉrienne Des Eaux. + - `pay_sonelgaz_bill`: Electricity & gas bill payment for Sonelgaz. + - `pay_algerie_telecom_bill`: Landline and internet bill payment for Algรฉrie Tรฉlรฉcom. + - `recharge_phone`: Flexy mobile balance top-ups (Mobilis, Djezzy, Ooredoo). + - `recharge_internet`: IDOOM 4G and ADSL subscriptions. + - `recharge_game`: Gaming vouchers (PUBG UC, Free Fire Diamonds). +- **Search by Memo**: + - `search_transactions_by_memo` in `SofizPayClient` and `TransactionManager`. +- **Top-level Convenience Exports**: + - All utility and CIB methods exposed at the package root level in `sofizpay`. +- **Documentation**: + - Comprehensive guide, Sandbox test card directory, and security best practices in `README.md`. ## [1.0.2] - 2025-08-01 -hello -## [1.0.2] - 2025-08-01 -hello +### Added +- Bug fixes and optimizations for Stellar transaction handlers. ## [1.0.1] - 2025-07-16 ### Added -- Initial release of SofizPay SDK for Python -- Payment operations using Stellar network -- DZT asset support with default issuer configuration -- Real-time transaction streaming and monitoring -- Balance checking for DZT and other assets -- Transaction history retrieval -- Transaction lookup by hash -- Comprehensive error handling and validation -- Async/await support for all operations -- Rate limiting for API calls -- Context manager support -- Extensive documentation and examples -- Full test suite with pytest - -### Features -- **Payment Management**: Send DZT payments with memo support -- **Transaction Monitoring**: Real-time transaction streaming -- **Balance Checking**: Get DZT balances and all account assets -- **Transaction History**: Retrieve transaction records with filtering -- **Error Handling**: Comprehensive exception hierarchy -- **Validation**: Input validation for all operations -- **Rate Limiting**: Built-in rate limiting for API stability -- **Examples**: Complete examples for all major operations - -### Technical Details -- Compatible with Python 3.8+ -- Uses stellar-sdk 8.0+ for Stellar operations -- Async/await support throughout -- Comprehensive test coverage -- Type hints for better IDE support -- PEP 8 compliant code style - -### Documentation -- Complete API documentation -- Usage examples in Arabic and English -- Installation and setup instructions -- Error handling guidelines -- Development setup guide +- Initial release of SofizPay SDK for Python. +- Payment operations using Stellar network (DZT asset). +- Real-time transaction streaming and monitoring. +- Balance checking for DZT and account assets. +- Transaction history and lookup by hash. +- Cryptographic RSA signature verification. +- Async/await support for core operations. diff --git a/README.md b/README.md index 25488e8..495c3c9 100644 --- a/README.md +++ b/README.md @@ -1,420 +1,422 @@
SofizPay Logo - -

SofizPay Python SDK

-

The official Python SDK for secure digital payments on the SofizPay platform.

- - [![PyPI version](https://badge.fury.io/py/sofizpay-sdk-python.svg)](https://pypi.org/project/sofizpay-sdk-python/) - [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) - [![Python](https://img.shields.io/badge/Python-3.8%2B-blue.svg)](https://www.python.org/)
---- +# SofizPay SDK Python -## ๐Ÿ“‹ Table of Contents +**The official Python SDK for secure digital payments, EDAHABIA / CIB transactions, utility bill payments, and telecom recharges in Algeria.** -- [Overview](#overview) -- [Installation](#installation) -- [Quick Start](#quick-start) -- [Core Methods](#core-methods) -- [Digital Services (Missions)](#digital-services-missions) -- [Bank Integration (CIB)](#bank-integration-cib) -- [Real-time Transaction Streaming](#real-time-transaction-streaming) -- [Response Format](#response-format) -- [Security Best Practices](#security-best-practices) -- [Support](#support) +[![PyPI version](https://badge.fury.io/py/sofizpay-sdk-python.svg)](https://pypi.org/project/sofizpay-sdk-python/) +[![Python Versions](https://img.shields.io/pypi/pyversions/sofizpay-sdk-python.svg)](https://pypi.org/project/sofizpay-sdk-python/) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) --- -## ๐ŸŒŸ Overview - -The SofizPay Python SDK provides a complete `async` interface for integrating **DZT digital payments** into Python applications, Django backends, and FastAPI services. It offers high-level abstractions for Stellar payments, transaction history, and digital service recharges. - -**Key Benefits:** -- โšก **Fully Async:** Built on `aiohttp` for non-blocking I/O -- ๐ŸŒ **Framework Agnostic:** Works perfectly with Django, FastAPI, Flask, or CLI tools -- ๐Ÿ“Š **Exhaustive History:** Captures Path Payments, Trustlines, and account events -- ๐Ÿฆ **CIB/Dahabia:** Simple bank deposit link generation -- ๐Ÿ“ฑ **Missions:** Phone, Internet, and Game top-ups via Mission APIs - ---- - -## ๐Ÿ“ฆ Installation +## ๐Ÿš€ Quick Start -### pip +### Installation ```bash pip install sofizpay-sdk-python ``` -**Requirements:** -- Python `>= 3.8` -- Dependencies: `aiohttp`, `stellar-sdk` - ---- - -## ๐Ÿš€ Quick Start +### Basic Usage ```python -import asyncio, os -from sofizpay.client import SofizPayClient +import asyncio +from sofizpay import SofizPayClient async def main(): - # Initialize the async client + # Production mode (default) client = SofizPayClient() - - # 1. Check DZT balance - balance = await client.get_balance('YOUR_PUBLIC_KEY') - if balance['success']: - print(f"๐Ÿ’ฐ Balance: {balance['balance']} DZT") - - # 2. Send a DZT payment + + # Or Sandbox mode for testing: + # client = SofizPayClient(is_sandbox=True) + + # Send direct wallet payment (DZT) result = await client.send_payment( - source_secret='YOUR_SECRET_KEY', # 56-char Stellar seed starting with 'S' - destination_public_key='RECIPIENT_PUBLIC_KEY', # Recipient's public key - amount='100', # Amount as string - memo='Invoice #1234' # Optional memo (max 28 chars) + source_secret='YOUR_SECRET_KEY', + destination_public_key='RECIPIENT_PUBLIC_KEY', + amount='100', + memo='Payment description' ) + print('Payment sent!' if result.get('successful') else result.get('error')) - if result.get('success'): - print(f"โœ… Payment sent! Hash: {result['transactionHash']}") - else: - print(f"โŒ Error: {result.get('error')}") - -if __name__ == "__main__": - # Run the async main function - asyncio.run(main()) +asyncio.run(main()) ``` --- -## ๐Ÿ”ง Core Methods +## โœจ Features + +- ๐Ÿ’ณ **CIB & EDAHABIA Gateway** - Accept bank card and postal card payments with full 3D Secure, Webhooks, and Sandbox testing. +- ๐Ÿ” **CIB Transaction Status Check** - Real-time verification of CIB/EDAHABIA payment status. +- โšก **Send Secure Payments** - Instant DZT wallet digital transactions on Stellar network. +- ๐Ÿ’ฐ **Account Balance** - Real-time balance checking for DZT. +- ๐Ÿ“œ **Transaction History & Search** - Complete transaction history and memo/hash searching. +- ๐Ÿ“ก **Real-time Streaming** - Live transaction monitoring with customizable intervals. +- ๐Ÿ›๏ธ **Products Catalog** - Browse available products, gaming vouchers, and telecom packages. +- ๐Ÿงพ **Utility Bill Payments** - Pay Sonelgaz, ADE (water), and Algรฉrie Tรฉlรฉcom bills programmatically. +- ๐Ÿ“ฑ **Mobile & Internet Top-ups** - Flexy (Mobilis, Djezzy, Ooredoo) and IDOOM (ADSL / 4G LTE). +- ๐ŸŽฎ **Gaming Credits** - Instant top-ups for PUBG UC, Free Fire Diamonds, and more. +- ๐Ÿ” **Digital Signature Verification** - Verify cryptographic RSA SHA-256 signatures from SofizPay callbacks. +- ๐ŸŒ **Multi-platform** - Fully asynchronous with async/await support across Linux, Windows, macOS. -### `get_balance(public_key)` - -Returns the current **DZT** balance for a given Stellar account. +--- -```python -# Fetch balance for a specific public key -result = await client.get_balance('GCAZI...YOUR_PUBLIC_KEY') - -# Response -{ - 'success': True, - 'balance': '1500.0000000', - 'publicKey': 'GCAZI...', - 'asset_code': 'DZT', - 'asset_issuer': 'GCAZI7YBLIDJWIVEL7ETNAZGPP3LC24NO6KAOBWZHUERXQ7M5BC52DLV', - 'timestamp': '2025-07-28T10:30:00Z' -} -``` +## ๐Ÿ“‹ Core Methods Reference + +| Method | Description | Example | +|--------|-------------|---------| +| `make_cib_transaction(data)` | Create CIB / EDAHABIA payment link | `await client.make_cib_transaction({...})` | +| `make_sandbox_cib_transaction(data)` | Create Sandbox CIB payment link | `await client.make_sandbox_cib_transaction({...})` | +| `check_cib_transaction(data)` | Check CIB transaction status | `await client.check_cib_transaction('order_id')` | +| `check_cib_status(cib_id)` | Production CIB status check | `await client.check_cib_status('2517039448')` | +| `check_sandbox_cib_status(cib_id)` | Sandbox CIB status check | `await client.check_sandbox_cib_status('40a11881...')` | +| `send_payment(secret, dest, amount, memo)` | Send DZT Stellar payment | `await client.send_payment(...)` | +| `get_balance(public_key)` | Get DZT account balance | `await client.get_balance('GXXX...')` | +| `get_transactions(public_key, limit)` | Get transaction history | `await client.get_transactions('GXXX...', 50)` | +| `search_transactions_by_memo(pk, memo, limit)` | Search transactions by memo | `await client.search_transactions_by_memo('GXXX...', 'Order #1')` | +| `get_transaction_by_hash(hash)` | Find transaction by hash | `await client.get_transaction_by_hash('abc123...')` | +| `setup_transaction_stream(pk, cb, from_now)` | Stream live transactions | `await client.setup_transaction_stream('GXXX...', cb)` | +| `stop_transaction_stream(stream_id)` | Stop live stream | `client.stop_transaction_stream(stream_id)` | +| `get_products(encrypted_sk, search)` | Browse products catalog | `await client.get_products('SXXX', search='PUBG')` | +| `get_operation_history(encrypted_sk, limit)` | View service operations history | `await client.get_operation_history('SXXX')` | +| `get_operation_details(id, encrypted_sk)` | Operation status tracking | `await client.get_operation_details('UUID', 'SXXX')` | +| `pay_ade_bill(data)` | Pay ADE water bill | `await client.pay_ade_bill({...})` | +| `pay_sonelgaz_bill(data)` | Pay Sonelgaz electricity/gas | `await client.pay_sonelgaz_bill({...})` | +| `pay_algerie_telecom_bill(data)` | Pay Algรฉrie Tรฉlรฉcom bill | `await client.pay_algerie_telecom_bill({...})` | +| `recharge_phone(data)` | Flexy mobile credit | `await client.recharge_phone({...})` | +| `recharge_internet(data)` | IDOOM ADSL / 4G recharge | `await client.recharge_internet({...})` | +| `recharge_game(data)` | PUBG / Free Fire recharge | `await client.recharge_game({...})` | +| `verify_signature(data)` | Verify RSA SHA-256 webhook | `SofizPayClient.verify_signature({...})` | --- -### `send_payment(source_secret, destination_public_key, amount, memo?)` +## ๐Ÿ“– API Reference & Examples + +### 1. CIB & EDAHABIA Transactions (`make_cib_transaction`) -Submits a DZT payment to the Stellar network. +Generate a secure payment URL to accept CIB or EDAHABIA payments with 3D Secure support, webhook callbacks, and sandbox testing. ```python -result = await client.send_payment( - source_secret='SXXX...YOUR_SECRET', # 56-char Stellar seed starting with 'S' - destination_public_key='GXXX...RECIPIENT', # Recipient's public key - amount='250.50', # Amount in DZT - memo='Order #5567' # Optional memo (max 28 chars) -) +import asyncio +from sofizpay import SofizPayClient -# Success Response -{ - 'success': True, - 'transactionId': 'abc123...hash', - 'transactionHash': 'abc123...hash', - 'amount': '250.50', - 'memo': 'Order #5567', - 'destinationPublicKey': 'GXXX...', - 'timestamp': '2025-07-28T10:30:00Z' -} +async def create_payment(): + client = SofizPayClient() + + response = await client.make_cib_transaction({ + 'account': 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', # Your Sofizpay receiving account + 'amount': 1500, # Amount in DZD + 'full_name': 'Ahmed Ben Ali', # Customer name + 'phone': '+213555123456', # Customer phone + 'email': 'ahmed.benali@example.com', # Customer email + 'return_url': 'https://mystore.com/payment-callback', # Redirect URL after checkout + 'webhook_url': 'https://mystore.com/api/cib-webhook', # Real-time async webhook URL + 'invoice_id': 'INV-2026-001', # Optional invoice ID + 'language': 'ar', # 'ar' | 'en' | 'fr' + 'memo': 'Order #12345', # Payment note (max 28 bytes) + 'redirect': 'yes', # 'yes' | 'no' + 'keep_return_url': 'True', # Include signed callback params + 'is_sandbox': False # Set True for Sandbox testing + }) + + if response.get('success'): + print('Payment URL:', response.get('payment_url')) + print('Transaction ID:', response.get('transaction_id')) + print('CIB Transaction ID:', response.get('cib_transaction_id')) + # Redirect customer to response['payment_url'] + else: + print('Failed to initiate payment:', response.get('error')) + +asyncio.run(create_payment()) ``` -> โš ๏ธ **Memo Truncation:** Memos longer than 28 characters are automatically truncated. +#### Dedicated Sandbox Helper: +```python +# Creates a CIB transaction directly in the Sandbox environment +sandbox_res = await client.make_sandbox_cib_transaction({ + 'account': 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', + 'amount': 150.0, + 'full_name': 'Sandbox Tester', + 'phone': '0661000000', + 'email': 'sandbox@sofizpay.com', + 'memo': 'Python Sandbox Test' +}) +print('Sandbox Payment URL:', sandbox_res.get('payment_url')) +``` --- -### `get_transactions(public_key, limit?)` +### 2. Check CIB Transaction Status (`check_cib_transaction` / `check_cib_status`) -Fetches **exhaustive transaction history** via Stellar. Includes payments, trustlines, and account creation. +Verify the payment status of an order after the customer completes payment on the SATIM payment page. ```python -# Fetch up to 100 recent transactions -transactions = await client.get_transactions('YOUR_PUBLIC_KEY', 100) +# Query by order number / CIB transaction ID +check = await client.check_cib_transaction({ + 'order_number': '2517039448', + 'is_sandbox': False # Set True if checking a sandbox transaction +}) -for tx in transactions: - # Print details for each captured operation - print(f"[{tx['created_at']}] {tx['type']} โ€” {tx['amount']} DZT") +if check.get('success') and check.get('status') == 'paid': + print(f"Order {check.get('order_number')} was successfully paid! Amount: {check.get('amount')} DZD") +else: + print('Payment status:', check.get('status'), check.get('error_message')) ``` ---- - -### `search_transactions_by_memo(public_key, memo, limit?)` - -Performs a case-insensitive search over recent transactions. - +#### Dedicated Status Check Helpers: ```python -# Perform case-insensitive search over user's recent transactions -results = await client.search_transactions_by_memo('YOUR_PUBLIC_KEY', 'Order #1234', 10) -if results['success']: - print(f"Found {results['totalFound']} matches") +# Check status in Production +prod_status = await client.check_cib_status('2517039448') + +# Check status in Sandbox +sandbox_status = await client.check_sandbox_cib_status('40a11881d8764fe9a371') ``` --- -## ๐Ÿ“ฑ Digital Services (Missions) - -Mission APIs allow users to spend DZT on real-world digital services. All calls require `encrypted_sk`. +### ๐Ÿ’ก Best Practice: Secure Order Flow -### Phone Recharge +For maximum security, store the `cib_transaction_id` in your database server-side and verify status before fulfilling orders: ```python -result = await client.recharge_phone({ - 'encrypted_sk': 'USER_ENCRYPTED_SECRET_KEY', # User's encrypted secret key - 'phone': '0661000000', # Recipient's phone number - 'operator': 'mobilis', # 'mobilis' | 'djezzy' | 'ooredoo' - 'amount': '100', # Recharge amount - 'offer': 'Top' # Offer type (e.g., 'Top', 'Pix') +# 1. Server initiates transaction +result = await client.make_cib_transaction({ + 'account': 'YOUR_PUBLIC_KEY', + 'amount': 5000, + 'full_name': 'Customer Name', + 'phone': '0555000000', + 'email': 'customer@example.com', + 'memo': 'Order #9921' }) -if result['success']: - # Process successful recharge response - print('โœ… Phone recharged:', result['data']) +if result.get('success'): + cib_id = result.get('cib_transaction_id') + # Save cib_id to database linked to order #9921 + # db.orders.update_one({'id': 9921}, {'$set': {'cib_id': cib_id}}) + payment_url = result.get('payment_url') + +# 2. When customer returns or webhook triggers, verify server-side: +status = await client.check_cib_status(cib_id) +if status.get('success') and status.get('status') == 'paid': + # Mark order as PAID in database and dispatch goods + pass ``` -### Internet Recharge (Idoom 4G) +--- -```python -result = await client.recharge_internet({ - 'encrypted_sk': 'USER_ENCRYPTED_SECRET_KEY', # User's encrypted secret key - 'phone': '0661000000', # Account phone number - 'operator': 'algerie-telecom', # Network provider - 'amount': '2000', # Recharge amount - 'offer': 'prepaid' # Offer type (e.g., 'prepaid', 'postpaid') -}) -``` +### 3. Products Catalog (`get_products`) -### Game Top-up (FreeFire, PUBG) +Retrieve available products and services with their prices in DZT, with optional search filtering. ```python -result = await client.recharge_game({ - 'encrypted_sk': 'USER_ENCRYPTED_SECRET_KEY', - 'operator': 'freefire', # e.g., 'freefire', 'pubg' - 'playerId': '123456789', - 'amount': '500', # 'amount' from getProducts() - 'offer': 'diamonds' # 'name' from getProducts() -}) -``` +# Get all available products +catalog = await client.get_products('YOUR_SECRET_KEY') -### Bill Payment +print(f"Available products count: {catalog.get('count')}") +for product in catalog.get('products', []): + print(f"{product.get('name')}: {product.get('price')} DZT") -```python -result = await client.pay_bill({ - 'encrypted_sk': 'USER_ENCRYPTED_SECRET_KEY', - 'operator': 'sonelgaz', # e.g., 'sonelgaz', 'ade' - 'bill_id': 'BILL_999', - 'amount': '1500' +# Search for specific products (e.g., PUBG, Free Fire, Mobilis) +search_result = await client.get_products({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'search': 'PUBG' }) +print('Found products:', search_result.get('products')) ``` -### Get Available Products +--- -Fetches the list of available services, operators, and their associated offers (products). +### 4. Utility Bill Payments (`pay_bill`) -```python -products = await client.get_products() -if products['success']: - # Use products['data'] to populate Missions 'offer' and 'amount' - print('Available services:', products['data']) -``` +Pay Algerian utility bills directly via the Python SDK: -> [!TIP] -> Each product in the `data` array usually contains `name` and `amount`. When calling recharge methods, use the product's `name` for the `offer` field and the product's `amount` for the `amount` field. +#### ADE (Algรฉrienne Des Eaux - Water Bill) +```python +ade_payment = await client.pay_ade_bill({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'amount': 2500, + 'bill': '0123456789' # Bill reference number +}) -### Operation History & Details +if ade_payment.get('success'): + print('ADE Bill Paid! Operation ID:', ade_payment.get('operation_id')) +``` +#### Sonelgaz (Electricity & Gas) ```python -# Recent operations (paginated) -history = await client.get_operation_history('USER_ENCRYPTED_SK', limit=10, offset=0) -if history['success']: - print('Last 10 operations:', history['data']) +sonelgaz_payment = await client.pay_sonelgaz_bill({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'amount': 3500, + 'customerId': 'CUST-100234', # Customer ID + 'ebb': 'EBB-987654', # EBB Number + 'bill': 'BILL-456789' # Bill Number +}) -# Details of a specific operation -details = await client.get_operation_details('OPERATION_ID', 'USER_ENCRYPTED_SK') +if sonelgaz_payment.get('success'): + print('Sonelgaz Bill Paid! Operation ID:', sonelgaz_payment.get('operation_id')) ``` -### Get Available Products - +#### Algรฉrie Tรฉlรฉcom Bill ```python -# Fetch available charging services and products -products = await client.get_products() -if products.get('success'): - # Iterate through available products to find names and amounts - print('Available services:', products['data']) +telecom_payment = await client.pay_algerie_telecom_bill({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'amount': 2000, + 'phone': '021234567', # Landline or subscription number + 'bill': 'BILL-00129' +}) ``` -> [!TIP] -> Use the product `name` for the `offer` field and the product `amount` for the `amount` field. - --- -## ๐Ÿฆ Bank Integration (CIB) - -Generate secure Dahabia/CIB bank payment links. +### 5. Mobile, Internet & Game Top-ups +#### Phone Recharge (Flexy: Mobilis, Djezzy, Ooredoo) ```python -result = await client.make_cib_transaction({ - 'account': 'YOUR_STELLAR_PUBLIC_KEY', # Your merchant Stellar public key - 'amount': 2500, # Amount in DZD (Algerian Dinars) - 'full_name': 'Ahmed Benali', # Customer's full name - 'phone': '0661234567', # Customer's phone number - 'email': 'ahmed@example.com', # Customer's email - 'memo': 'Order #789', # Internal order reference - 'return_url': 'https://yoursite.com/callback', # Redirect after payment - 'redirect': 'no' # 'yes' for auto-redirect +flexy = await client.recharge_phone({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'phone': '0661234567', + 'operator': 'djezzy', # 'mobilis' | 'djezzy' | 'ooredoo' + 'amount': 500, + 'offer': 'prepaid' }) - -if result.get('success'): - # Get total generated hosted payment URL - payment_url = result['data']['payment_url'] - print(f"Redirect customer to: {payment_url}") ``` -### ๐Ÿงช Sandbox Environment - -For safe testing without real money, use the dedicated sandbox methods. These methods always point to the SofizPay Sandbox environment. - +#### IDOOM Internet Recharge (ADSL & 4G LTE) ```python -# 1. Generate a sandbox payment link -result = await client.make_sandbox_cib_transaction({ - 'account': 'YOUR_PUBLIC_KEY', - 'amount': 150.0, - 'full_name': 'Sandbox Tester', - 'phone': '0555000000', - 'email': 'sandbox@example.com' +internet = await client.recharge_internet({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'phone': '0458230823', # 10 digits for 4G, 9 digits for ADSL + 'operator': 'idoom', + 'amount': 1000, + 'offer': 'IDOOM 4G 1000' # e.g. 'IDOOM 4G 1000' or 'IDOOM ADSL 2000' }) - -if result.get('success'): - print(f"Sandbox URL: {result['data']['payment_url']}") - cib_id = result['data']['cib_transaction_id'] - - # 2. Check sandbox status - status = await client.check_sandbox_cib_status(cib_id) - print(f"Sandbox Status: {status['data']['status']}") ``` -### Check CIB Status (Production) - +#### Gaming Credits (PUBG & Free Fire) ```python -# Monitor progress of a real CIB transaction using its ID -status = await client.check_cib_status('CIB_TRANSACTION_ID') -if status['success']: - # Current transaction status (e.g., 'success', 'pending', 'failed') - print(f"Status: {status['data']['status']}") +game = await client.recharge_game({ + 'encrypted_sk': 'YOUR_SECRET_KEY', + 'operator': 'pubg', # 'pubg' | 'freefire' + 'playerId': '5123456789', # Player in-game ID + 'amount': 1200, + 'offer': '60' # '60' | '325' | '660' for PUBG, '110' | '210' for Free Fire +}) ``` -### ๐Ÿ’ก Best Practice: Secure Order Flow +--- -For maximum security, never expose the `cib_transaction_id` (order_number) to the end-user. Always store it in your database and verify the status server-side. +### 6. Operation Details & History ```python -# 1. Initiate transaction and store the ID invisibly -result = await client.make_cib_transaction({ - 'account': 'YOUR_PUBLIC_KEY', - 'amount': 5000, - 'memo': 'Order #9988' +# Get details of a specific operation +details = await client.get_operation_details({ + 'operation_id': '550e8400-e29b-41d4-a716-446655440000', + 'encrypted_sk': 'YOUR_SECRET_KEY' }) -if result['success']: - cib_id = result['data']['cib_transaction_id'] - # โœ… SAVE to database: UPDATE orders SET cib_id = '{cib_id}' WHERE id = 9988; - - # Send user to payment_url - print(f"Redirecting user to {result['data']['payment_url']}") - -# 2. Later, when checking status, fetch from database -order = await db.fetch_one("SELECT * FROM orders WHERE id=9988") -status = await client.check_cib_status(order['cib_id']) - -if status['success'] and status['data']['status'] == 'success': - # โœ… Update order to PAID state in your database - await db.execute("UPDATE orders SET status='paid' WHERE id=9988") +# Get operation history +history = await client.get_operation_history('YOUR_SECRET_KEY', limit=10, offset=0) +print('Recent Operations:', history.get('data')) ``` --- -## ๐Ÿ”ด Real-time Transaction Streaming +### 7. Digital Signature Verification (`verify_signature`) + +Verify webhook callbacks signed with RSA SHA-256: + +```python +from fastapi import FastAPI, Request, HTTPException +from sofizpay import SofizPayClient + +app = FastAPI() -The Python SDK uses polling to monitor accounts for new transactions. +@app.post("/api/cib-webhook") +async def cib_webhook(request: Request): + payload = await request.json() + message = payload.get("message") + signature = payload.get("signature_url_safe") + + is_valid = SofizPayClient.verify_signature({ + "message": message, + "signature_url_safe": signature + }) + + if not is_valid: + raise HTTPException(status_code=400, detail="Invalid signature") + + # Process authentic payment confirmation + return {"status": "success", "received": True} +``` -### `setup_transaction_stream(public_key, callback, from_now?, check_interval?)` +--- -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `public_key` | `str` | required | Stellar account to monitor | -| `callback` | `async function` | required | Called for each new transaction | -| `from_now` | `bool` | `True` | `True`: new txs only; `False`: load history first | -| `check_interval` | `int` | `30` | Polling interval in seconds | +### 8. Direct Stellar Wallet Payments & Balance ```python -async def on_tx(tx): - print(f"New {tx['type']}: {tx['amount']} DZT โ€” memo: {tx['memo']}") +# Get DZT balance +balance = await client.get_balance('GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ') +print('Balance:', balance) + +# Search transactions by memo +search_res = await client.search_transactions_by_memo( + 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', + 'Order #12345', + limit=50 +) +print('Search Results:', search_res.get('transactions')) + +# Real-time transaction streaming +def handle_tx(tx): + print('Live Payment Received:', tx.get('amount'), tx.get('memo'), tx.get('from')) stream_id = await client.setup_transaction_stream( - 'YOUR_PUBLIC_KEY', on_tx, from_now=True, check_interval=15 + 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', + handle_tx, + from_now=True, + check_interval=30 ) + +# Later, stop stream: +client.stop_transaction_stream(stream_id) ``` --- -## ๐Ÿ“ค Response Format +## ๐Ÿงช Testing with CIB Sandbox -All async methods return a dictionary with a `success` flag: +SofizPay provides a mock testing environment to test CIB / EDAHABIA payments without real cards: -```python -# โœ… Success -{ - 'success': True, - 'data': {...}, # method-specific - 'timestamp': '2025-07-28T10:30:00Z' -} - -# โŒ Failure -{ - 'success': False, - 'error': 'Error message', - 'timestamp': '2025-07-28T10:30:00Z' -} -``` +Set `is_sandbox=True` in `SofizPayClient(is_sandbox=True)` or pass `'is_sandbox': True` to `make_cib_transaction` / `check_cib_transaction`. ---- +### Test Card Numbers: -## ๐Ÿ›ก๏ธ Security Best Practices +| Card Number | Expiry | CVV | Expected Result | +|-------------|--------|-----|-----------------| +| `6280581001234567` | `12/28` | `123` | **Approved** | +| `6280581009876543` | `12/29` | `456` | **Approved** | +| `6280581005555555` | `06/28` | `999` | **Declined by issuer** | +| `6280581004444444` | `09/28` | `444` | **Insufficient funds** | +| `6280581003333333` | `03/21` | `333` | **Expired card** | +| `6280581002222222` | `11/28` | `222` | **Timeout** | -- โŒ Never expose secret keys in public repositories or client-side code. -- โœ… Use environment variables (`os.getenv('SOFIZPAY_SECRET')`). -- โœ… Keep `encrypted_sk` secure in your backend database. -- โœ… Use `encrypted_sk` for all Mission API calls (never the raw secret). -- โœ… Run payment logic on backend only. +> **Sandbox OTP:** Use `123456` for any test transaction. --- -## ๐Ÿ“ž Support +## ๐Ÿ”’ Security Best Practices -- ๐ŸŒ **Website**: [SofizPay.com](https://sofizpay.com) -- ๐Ÿ“š **Full Docs**: [docs.sofizpay.com](https://docs.sofizpay.com) -- ๐Ÿ› **Bug Reports**: [GitHub Issues](https://github.com/kenandarabeh/sofizpay-sdk-python/issues) +1. **Protect Secret Keys:** Never hardcode secret keys in frontend or public code. Always use environment variables (`os.environ.get('SOFIZPAY_SECRET_KEY')`). +2. **Verify Webhooks:** Always use `SofizPayClient.verify_signature()` to validate incoming webhook payloads before updating database records. +3. **Check Status Server-side:** When a customer returns to your `return_url`, verify the transaction using `check_cib_transaction()` from your server before granting access or fulfilling orders. --- -## License +## ๐Ÿ“œ License MIT ยฉ [SofizPay Team](https://github.com/kenandarabeh) -**Built with โค๏ธ for Python developers | Version `1.1.0`** - +**Built with โค๏ธ for Algerian Fintech | [docs.sofizpay.com](https://docs.sofizpay.com/)** diff --git a/example/test_sandbox.py b/example/test_sandbox.py new file mode 100644 index 0000000..4df6454 --- /dev/null +++ b/example/test_sandbox.py @@ -0,0 +1,55 @@ +import os +import sys + +# ุฅุถุงูุฉ ู…ุณุงุฑ ุงู„ุญุฒู…ุฉ ู„ู„ุชุดุบูŠู„ ุงู„ู…ุจุงุดุฑ +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +import asyncio +import json +from sofizpay import SofizPayClient + +async def run_sandbox_test(): + print("--- Starting SofizPay Python SDK Sandbox Test ---") + print("Current Mode: SANDBOX") + + # 1. Initialize SDK in Sandbox Mode + client = SofizPayClient(is_sandbox=True) + + account_key = os.environ.get('SOFIZPAY_PUBLIC_KEY', 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ') + + try: + # 2. Test make_sandbox_cib_transaction + print("\n1. Testing make_sandbox_cib_transaction (Dedicated)...") + cib_result = await client.make_sandbox_cib_transaction({ + 'account': account_key, + 'amount': 150.0, + 'full_name': 'Sandbox Tester', + 'phone': '0661000000', + 'email': 'sandbox@sofizpay.com', + 'memo': 'Python Sandbox Test' + }) + + print("Result:", json.dumps(cib_result, indent=2, ensure_ascii=False)) + + # 3. Test check_sandbox_cib_status + cib_id = cib_result.get('cib_transaction_id') or ( + cib_result.get('data', {}).get('cib_transaction_id') + if isinstance(cib_result.get('data'), dict) else None + ) + + if cib_id: + print(f"\n2. Testing check_sandbox_cib_status for ID: {cib_id}...") + status_result = await client.check_sandbox_cib_status(cib_id) + print("Status Result:", json.dumps(status_result, indent=2, ensure_ascii=False)) + else: + print("\n2. Testing check_sandbox_cib_status with sample order number...") + sample_status = await client.check_sandbox_cib_status("40a11881d8764fe9a371") + print("Sample Status Result:", json.dumps(sample_status, indent=2, ensure_ascii=False)) + + except Exception as error: + print("Test Error:", str(error)) + + print("\n--- Sandbox Test Completed ---") + +if __name__ == '__main__': + asyncio.run(run_sandbox_test()) diff --git a/example/test_sdk.py b/example/test_sdk.py new file mode 100644 index 0000000..e5f04de --- /dev/null +++ b/example/test_sdk.py @@ -0,0 +1,192 @@ +import os +import sys +import json +import asyncio + +# ุฏุนู… ุทุจุงุนุฉ ุงู„ุญุฑูˆู ุงู„ุนุฑุจูŠุฉ ุนู„ู‰ Windows +if sys.platform == 'win32' and hasattr(sys.stdout, 'reconfigure'): + try: + sys.stdout.reconfigure(encoding='utf-8', errors='replace') + except Exception: + pass + +# ุฅุถุงูุฉ ู…ุณุงุฑ ุงู„ุญุฒู…ุฉ ู„ู„ุชุดุบูŠู„ ุงู„ู…ุจุงุดุฑ +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +from sofizpay import SofizPayClient, ValidationError + +# ============================================================= +# ๐Ÿ”‘ ุถุน ู…ูุงุชูŠุญูƒ ู‡ู†ุง (ุชู…ุงู…ุงู‹ ู…ุซู„ ุงุฎุชุจุงุฑ ุงู„ู€ JS) +# ============================================================= + +MY_SECRET_KEY = os.environ.get('SOFIZPAY_SECRET_KEY', 'YOUR_SECRET_KEY_HERE') # ู…ูุชุงุญูƒ ุงู„ุณุฑูŠ (ูŠุจุฏุฃ ุจู€ S) +MY_PUBLIC_KEY = os.environ.get('SOFIZPAY_PUBLIC_KEY', 'YOUR_PUBLIC_KEY_HERE') # ู…ูุชุงุญูƒ ุงู„ุนุงู… (ูŠุจุฏุฃ ุจู€ G) +RECIPIENT_KEY = os.environ.get('SOFIZPAY_RECIPIENT_KEY', 'RECIPIENT_PUBLIC_KEY_HERE') # ุงู„ู…ูุชุงุญ ุงู„ุนุงู… ู„ู„ู…ุณุชู‚ุจู„ +MY_ENCRYPTED_SK = MY_SECRET_KEY # ู†ูุณ ุงู„ู…ูุชุงุญ ุงู„ุณุฑูŠ (ูŠูุณุชุฎุฏู… ู…ุน ุฎุฏู…ุงุช ุงู„ููˆุงุชูŠุฑ ูˆุงู„ู…ู†ุชุฌุงุช) + +# ============================================================= + +async def run_tests(): + print('--- Starting SofizPay Python SDK Tests ---') + client = SofizPayClient() + + print('SDK Version:', client.get_version()) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 1: ุงู„ุชุญู‚ู‚ ู…ู† ูˆุฌูˆุฏ ุฌู…ูŠุน ุงู„ู€ Methods + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + required_methods = [ + 'send_payment', + 'get_balance', + 'get_transactions', + 'get_all_transactions', + 'get_public_key_from_secret', + 'setup_transaction_stream', + 'stop_transaction_stream', + 'search_transactions_by_memo', + 'get_transaction_by_hash', + 'make_cib_transaction', + 'make_sandbox_cib_transaction', + 'check_cib_transaction', + 'check_cib_status', + 'check_sandbox_cib_status', + 'cib_transaction_check', + 'get_products', + 'execute_service_operation', + 'pay_bill', + 'pay_ade_bill', + 'pay_sonelgaz_bill', + 'pay_algerie_telecom_bill', + 'recharge_phone', + 'recharge_internet', + 'recharge_game', + 'get_operation_details', + 'get_operation_history', + 'verify_signature', + 'verify_sofizpay_signature' + ] + + missing = [] + for method in required_methods: + if not hasattr(client, method) or not callable(getattr(client, method)): + missing.append(method) + + if not missing: + print(f'[OK] All {len(required_methods)} expected methods exist on client instance.') + else: + print('[ERROR] Missing methods:', missing) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 2: ุงุณุชุฎุฑุงุฌ ุงู„ู…ูุชุงุญ ุงู„ุนุงู… ู…ู† ุงู„ุณุฑูŠ + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุงุณุชุฎุฑุงุฌ ุงู„ู…ูุชุงุญ ุงู„ุนุงู… ู…ู† ุงู„ู…ูุชุงุญ ุงู„ุณุฑูŠ ---') + try: + derived_pk = client.get_public_key_from_secret(MY_SECRET_KEY) + print('[OK] Public Key ุงู„ู…ุณุชุฎุฑุฌ:', derived_pk) + except Exception as err: + print('[ERROR] ูุดู„ ุงุณุชุฎุฑุงุฌ ุงู„ู…ูุชุงุญ:', err) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 3: ุฑุตูŠุฏ ุญุณุงุจูƒ + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุงู„ุชุญู‚ู‚ ู…ู† ุฑุตูŠุฏ ุงู„ุญุณุงุจ ---') + try: + balance = await client.get_balance(MY_PUBLIC_KEY) + print(f'[OK] ุงู„ุฑุตูŠุฏ: {balance} DZT') + except Exception as err: + print('[ERROR] ุฎุทุฃ ููŠ ุฌู„ุจ ุงู„ุฑุตูŠุฏ:', err) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 4: ุฅุฑุณุงู„ ุฏูุนุฉ ู…ุจุงุดุฑุฉ DZT + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุฅุฑุณุงู„ ุฏูุนุฉ DZT ู…ุจุงุดุฑุฉ ---') + try: + payment = await client.send_payment( + source_secret=MY_SECRET_KEY, + destination_public_key=RECIPIENT_KEY, + amount='1', + memo='ุงุฎุชุจุงุฑ SDK' + ) + if payment.get('successful') or payment.get('hash'): + print('[OK] ุงู„ุฏูุนุฉ ุงุฑุณู„ุช! Hash:', payment.get('hash')) + else: + print('[INFO] ู†ุชูŠุฌุฉ ุงู„ุฏูุนุฉ:', payment) + except Exception as err: + print('[ERROR] ูุดู„ ุงู„ุฅุฑุณุงู„:', err) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 5: ุฅู†ุดุงุก ู…ุนุงู…ู„ุฉ CIB / ุงู„ุฐู‡ุจูŠุฉ (Sandbox) + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุฅู†ุดุงุก ู…ุนุงู…ู„ุฉ CIB (Sandbox) ---') + try: + cib = await client.make_cib_transaction({ + 'account': MY_PUBLIC_KEY, + 'amount': 1000, + 'full_name': 'Ahmed Ben Ali', + 'phone': '+213661234567', + 'email': 'test@example.com', + 'return_url': 'https://mystore.com/callback', + 'webhook_url': 'https://mystore.com/api/webhook', + 'memo': 'ุทู„ุจ ุงุฎุชุจุงุฑูŠ #001', + 'is_sandbox': True # ุจูŠุฆุฉ ุงุฎุชุจุงุฑ (ู„ุง ูŠุฎุตู… ู…ุงู„ ุญู‚ูŠู‚ูŠ) + }) + if cib.get('success'): + print('[OK] CIB Transaction created!') + print(' Payment URL:', cib.get('payment_url')) + print(' CIB ID:', cib.get('cib_transaction_id')) + else: + print('[ERROR] ูุดู„ CIB:', cib.get('error')) + except Exception as err: + print('[ERROR] ุฎุทุฃ ููŠ CIB:', err) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 6: ุฌู„ุจ ู‚ุงุฆู…ุฉ ุงู„ู…ู†ุชุฌุงุช + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุฌู„ุจ ูƒุชุงู„ูˆุฌ ุงู„ู…ู†ุชุฌุงุช ---') + try: + products = await client.get_products({'encrypted_sk': MY_ENCRYPTED_SK}) + if products.get('success'): + print(f"[OK] {products.get('count', 0)} ู…ู†ุชุฌ ู…ุชูˆูุฑ") + print('\nู‚ุงุฆู…ุฉ ุจุงู„ู…ู†ุชุฌุงุช:') + print('=' * 60) + for i, p in enumerate(products.get('products', [])): + name = p.get('name') or p.get('title') or p.get('product_name') or 'ุจุฏูˆู† ุงุณู…' + price = p.get('price') or p.get('amount') or p.get('cost') or 'โ€”' + category = p.get('category') or p.get('type') or p.get('operator') or '' + offer = p.get('offer') or p.get('offer_id') or '' + offer_str = f" ({offer})" if offer else "" + print(f"{str(i + 1).rjust(3)}. [{category or 'โ€”'}] {name} -> {price} DZT{offer_str}") + print('=' * 60) + else: + print('[ERROR] ูุดู„ ุฌู„ุจ ุงู„ู…ู†ุชุฌุงุช:', products.get('error')) + except Exception as err: + print('[ERROR] ุฎุทุฃ ููŠ ุฌู„ุจ ุงู„ู…ู†ุชุฌุงุช:', err) + + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + # โœ… Test 7: ุงุฎุชุจุงุฑ ุงู„ุชุญู‚ู‚ ู…ู† validations + # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + print('\n--- ุงุฎุชุจุงุฑ ุงู„ุชุญู‚ู‚ ู…ู† ุงู„ู…ุฏุฎู„ุงุช ---') + try: + await client.make_cib_transaction({}) + except ValidationError as err: + print('[OK] make_cib_transaction validation:', str(err)) + + try: + await client.check_cib_transaction('') + except ValidationError as err: + print('[OK] check_cib_transaction validation:', str(err)) + + try: + await client.pay_ade_bill({'encrypted_sk': 'X', 'amount': 100}) + except ValidationError as err: + print('[OK] pay_ade_bill validation:', str(err)) + + try: + await client.recharge_game({'encrypted_sk': 'X', 'operator': 'pubg', 'amount': 100}) + except ValidationError as err: + print('[OK] recharge_game validation:', str(err)) + + print('\n[OK] ุงู†ุชู‡ู‰ ุงุฎุชุจุงุฑ ุงู„ู€ SDK ุจู†ุฌุงุญ!') + +if __name__ == '__main__': + asyncio.run(run_tests()) diff --git a/pyproject.toml b/pyproject.toml index 2d49750..b41ae09 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [build-system] -requires = ["setuptools>=45", "wheel", "setuptools_scm[toml]>=6.2"] +requires = ["setuptools>=61.0", "wheel"] build-backend = "setuptools.build_meta" [project] name = "sofizpay-sdk-python" -version = "1.0.2" -description = "Professional Python SDK for SofizPay payments using Stellar blockchain" +version = "1.2.0" +description = "Professional Python SDK for SofizPay payments, CIB & EDAHABIA gateway, utility bill payments, and telecom recharges" readme = "README.md" requires-python = ">=3.8" license = {text = "MIT"} @@ -15,7 +15,7 @@ authors = [ maintainers = [ {name = "SofizPay Team", email = "support@sofizpay.com"} ] -keywords = ["stellar", "payment", "blockchain", "cryptocurrency", "DZT", "sofizpay", "fintech"] +keywords = ["stellar", "payment", "blockchain", "cryptocurrency", "DZT", "sofizpay", "fintech", "cib", "edahabia", "satim", "algeria"] classifiers = [ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", diff --git a/setup.cfg b/setup.cfg index 03cc3ed..9addd38 100644 --- a/setup.cfg +++ b/setup.cfg @@ -14,8 +14,8 @@ markers = asyncio_mode = auto [flake8] -max-line-length = 88 -extend-ignore = E203, W503 +max-line-length = 127 +extend-ignore = E203, W503, E501 exclude = .git, __pycache__, @@ -24,7 +24,8 @@ exclude = .eggs, *.egg-info, .venv, - venv + venv, + example [mypy] python_version = 3.8 diff --git a/setup.py b/setup.py index 986f8ea..9f46c87 100644 --- a/setup.py +++ b/setup.py @@ -7,15 +7,15 @@ requirements = [line.strip() for line in fh if line.strip() and not line.startswith("#")] setup( - name="sofizpay-sdk", - version = "1.0.2", + name="sofizpay-sdk-python", + version="1.2.0", author="SofizPay Team", author_email="support@sofizpay.com", - description="Professional Python SDK for SofizPay payments using Stellar blockchain", + description="Professional Python SDK for SofizPay payments, CIB & EDAHABIA gateway, utility bill payments, and telecom recharges", long_description=long_description, long_description_content_type="text/markdown", url="https://github.com/kenandarabeh/sofizpay-sdk-python", - packages=find_packages(), + packages=find_packages(exclude=["tests*", "example*"]), classifiers=[ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", @@ -34,7 +34,7 @@ ], python_requires=">=3.8", install_requires=requirements, - keywords="stellar, payment, blockchain, cryptocurrency, DZT, sofizpay, fintech, stellar-network", + keywords="stellar, payment, blockchain, cryptocurrency, DZT, sofizpay, fintech, cib, edahabia, satim, algeria", project_urls={ "Bug Reports": "https://github.com/kenandarabeh/sofizpay-sdk-python/issues", "Source": "https://github.com/kenandarabeh/sofizpay-sdk-python", diff --git a/sofizpay/__init__.py b/sofizpay/__init__.py index 92a27d3..c015306 100644 --- a/sofizpay/__init__.py +++ b/sofizpay/__init__.py @@ -1,10 +1,11 @@ """ -SofizPay SDK - Python library for Stellar-based payments +SofizPay SDK - Python library for digital payments, CIB/EDAHABIA gateway, and Algerian services This SDK provides easy-to-use functions for integrating SofizPay -payment functionality into Python applications. +payment functionality, CIB gateway, utility bill payments, and telecom recharges into Python applications. """ +from typing import Dict, Any, Optional, Union, List from .client import SofizPayClient from .payments import PaymentManager from .transactions import TransactionManager @@ -13,10 +14,14 @@ PaymentError, TransactionError, NetworkError, - ValidationError + ValidationError, + RateLimitError, + InsufficientBalanceError, + InvalidAccountError, + InvalidAssetError ) -__version__ = "1.0.2" +__version__ = "1.2.0" __author__ = "SofizPay Team" __email__ = "support@sofizpay.com" @@ -28,16 +33,151 @@ "PaymentError", "TransactionError", "NetworkError", - "ValidationError" + "ValidationError", + "RateLimitError", + "InsufficientBalanceError", + "InvalidAccountError", + "InvalidAssetError", + "make_cib_transaction", + "make_sandbox_cib_transaction", + "check_cib_transaction", + "check_cib_status", + "check_sandbox_cib_status", + "cib_transaction_check", + "verify_sofizpay_signature", + "verify_signature", + "get_products", + "get_operation_history", + "get_operation_details", + "execute_service_operation", + "pay_bill", + "pay_ade_bill", + "pay_sonelgaz_bill", + "pay_algerie_telecom_bill", + "recharge_phone", + "recharge_internet", + "recharge_game", + "search_transactions_by_memo" ] -# Convenience functions for easy access -def make_cib_transaction(transaction_data): - """Convenience function to make CIB transaction""" + +# ========================================================================= +# Convenience functions for easy top-level access +# ========================================================================= + +async def make_cib_transaction(transaction_data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to make CIB / EDAHABIA transaction""" + client = SofizPayClient() + return await client.make_cib_transaction(transaction_data) + + +async def make_sandbox_cib_transaction(transaction_data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to make Sandbox CIB transaction""" + client = SofizPayClient(is_sandbox=True) + return await client.make_sandbox_cib_transaction(transaction_data) + + +async def check_cib_transaction(data: Union[str, int, Dict[str, Any]]) -> Dict[str, Any]: + """Convenience function to check CIB transaction status""" client = SofizPayClient() - return client.make_cib_transaction(transaction_data) + return await client.check_cib_transaction(data) -def verify_sofizpay_signature(verification_data): + +async def check_cib_status(cib_transaction_id: str) -> Dict[str, Any]: + """Convenience function to check CIB status in Production""" + client = SofizPayClient() + return await client.check_cib_status(cib_transaction_id) + + +async def check_sandbox_cib_status(cib_transaction_id: str) -> Dict[str, Any]: + """Convenience function to check CIB status in Sandbox""" + client = SofizPayClient(is_sandbox=True) + return await client.check_sandbox_cib_status(cib_transaction_id) + + +async def cib_transaction_check(data: Union[str, int, Dict[str, Any]]) -> Dict[str, Any]: + """Alias convenience function to check CIB transaction""" + client = SofizPayClient() + return await client.check_cib_transaction(data) + + +def verify_sofizpay_signature(verification_data: Dict[str, str]) -> bool: """Convenience function to verify SofizPay signature""" + return SofizPayClient.verify_signature(verification_data) + + +def verify_signature(verification_data: Union[Dict[str, str], str], signature: Optional[str] = None) -> bool: + """Convenience function to verify cryptographic signature from SofizPay""" + return SofizPayClient.verify_signature(verification_data, signature) + + +async def get_products(options: Union[str, Dict[str, Any]], search: Optional[str] = None) -> Dict[str, Any]: + """Convenience function to retrieve product catalog""" + client = SofizPayClient() + return await client.get_products(options, search=search) + + +async def get_operation_history(encrypted_sk: str, limit: int = 10, offset: int = 0) -> Dict[str, Any]: + """Convenience function to get operation history""" + client = SofizPayClient() + return await client.get_operation_history(encrypted_sk, limit=limit, offset=offset) + + +async def get_operation_details(options: Union[str, Dict[str, Any]], encrypted_sk: Optional[str] = None) -> Dict[str, Any]: + """Convenience function to get operation details""" + client = SofizPayClient() + return await client.get_operation_details(options, encrypted_sk=encrypted_sk) + + +async def execute_service_operation(operation_data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to execute service operation""" + client = SofizPayClient() + return await client.execute_service_operation(operation_data) + + +async def pay_bill(bill_data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to pay utility bills""" + client = SofizPayClient() + return await client.pay_bill(bill_data) + + +async def pay_ade_bill(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to pay ADE water bill""" + client = SofizPayClient() + return await client.pay_ade_bill(data) + + +async def pay_sonelgaz_bill(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to pay Sonelgaz electricity/gas bill""" + client = SofizPayClient() + return await client.pay_sonelgaz_bill(data) + + +async def pay_algerie_telecom_bill(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to pay Algรฉrie Tรฉlรฉcom bill""" + client = SofizPayClient() + return await client.pay_algerie_telecom_bill(data) + + +async def recharge_phone(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to recharge phone credit (Flexy)""" + client = SofizPayClient() + return await client.recharge_phone(data) + + +async def recharge_internet(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to recharge IDOOM internet""" + client = SofizPayClient() + return await client.recharge_internet(data) + + +async def recharge_game(data: Dict[str, Any]) -> Dict[str, Any]: + """Convenience function to recharge game credits (PUBG / Free Fire)""" + client = SofizPayClient() + return await client.recharge_game(data) + + +async def search_transactions_by_memo(public_key: str, memo: str, limit: int = 50) -> Dict[str, Any]: + """Convenience function to search transactions by memo""" client = SofizPayClient() - return client.verify_sofizpay_signature(verification_data) + return await client.search_transactions_by_memo(public_key, memo, limit=limit) diff --git a/sofizpay/client.py b/sofizpay/client.py index 9de2013..82ce22c 100644 --- a/sofizpay/client.py +++ b/sofizpay/client.py @@ -2,11 +2,12 @@ import urllib.parse from datetime import datetime -from typing import Optional, Dict, Any, List, Callable +from typing import Optional, Dict, Any, List, Callable, Union import requests import base64 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey from cryptography.exceptions import InvalidSignature from .payments import PaymentManager @@ -20,10 +21,11 @@ class SofizPayClient: This class provides a unified interface for all SofizPay operations including payments, transaction monitoring, balance management, - CIB transactions, and signature verification. + CIB & EDAHABIA transactions, CIB status checks, Algerian utility payments, + telecom & gaming recharges, and digital signature verification. """ - VERSION = "1.0.2" + VERSION = "1.2.0" SOFIZPAY_PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY----- MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1N+bDPxpqeB9QB0affr/ @@ -35,19 +37,32 @@ class SofizPayClient: 9wIDAQAB -----END PUBLIC KEY-----""" - def __init__(self, server_url: str = "https://horizon.stellar.org", sandbox: bool = False): + def __init__( + self, + is_sandbox: bool = False, + server_url: str = "https://horizon.stellar.org" + ): """ Initialize SofizPay client Args: + is_sandbox: Default to sandbox environment if True server_url: Stellar Horizon server URL (defaults to mainnet) - sandbox: Enable sandbox mode for CIB transactions (defaults to False) """ + self.version = self.VERSION + self.is_sandbox = is_sandbox self.server_url = server_url - self.sandbox = sandbox self.payment_manager = PaymentManager(server_url) self.transaction_manager = TransactionManager(server_url) + def get_version(self) -> str: + """Return SDK version""" + return self.version + + # ========================================================================= + # STELLAR WALLET & DIRECT PAYMENTS + # ========================================================================= + async def send_payment( self, source_secret: str, @@ -56,7 +71,7 @@ async def send_payment( memo: Optional[str] = None ) -> Dict[str, Any]: """ - Send a payment on Sofizpay (ูŠุนู…ู„ ู…ุซู„ JavaScript ุชู…ุงู…ุงู‹) + Send a payment on Sofizpay / Stellar network Args: source_secret: Secret key of the source account @@ -67,176 +82,70 @@ async def send_payment( Returns: Dictionary with transaction result """ - try: - result = await self.payment_manager.send_payment( - source_secret=source_secret, - destination_public_key=destination_public_key, - amount=amount, - memo=memo - ) - - if result.get('success'): - # Align return structure with JS SDK - return { - 'success': True, - 'transactionId': result.get('hash'), - 'transactionHash': result.get('hash'), - 'amount': float(amount), - 'memo': memo, - 'destinationPublicKey': destination_public_key, - 'duration': result.get('duration'), - 'timestamp': datetime.now().isoformat() - } - else: - return { - 'success': False, - 'error': result.get('error', 'Transaction failed'), - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return { - 'success': False, - 'error': str(e), - 'timestamp': datetime.now().isoformat() - } - - async def submit(self, data: Dict[str, Any]) -> Dict[str, Any]: - """Alias for send_payment to match JS SDK nomenclature""" - return await self.send_payment( - source_secret=data.get('secretkey'), - destination_public_key=data.get('destinationPublicKey'), - amount=str(data.get('amount')), - memo=data.get('memo') + return await self.payment_manager.send_payment( + source_secret=source_secret, + destination_public_key=destination_public_key, + amount=amount, + memo=memo ) - async def get_balance(self, public_key: str) -> Dict[str, Any]: + async def get_balance(self, public_key: str) -> float: """ - Get balance for an account + Get DZT balance for an account Args: public_key: Public key of the account Returns: - balance dictionary (aligned with JS SDK) + Balance as float """ - try: - balance = await self.payment_manager.get_balance(public_key) - return { - 'success': True, - 'balance': balance, - 'publicKey': public_key, - 'asset_code': 'DZT', - 'asset_issuer': 'GCAZI7YBLIDJWIVEL7ETNAZGPP3LC24NO6KAOBWZHUERXQ7M5BC52DLV', - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return { - 'success': False, - 'error': str(e), - 'balance': 0, - 'publicKey': public_key, - 'timestamp': datetime.now().isoformat() - } - + return await self.payment_manager.get_balance(public_key) - async def search_transactions_by_memo( - self, - public_key: str, - memo: str, - limit: int = 50 - ) -> Dict[str, Any]: + def get_public_key_from_secret(self, secret_key: str) -> str: """ - Search transactions by memo content (aligned with JS SDK) + Extract public key from secret key Args: - public_key: Public key of the account - memo: Text to search for in memo - limit: Maximum number of transactions to return + secret_key: The secret key Returns: - Standard result object with filtered transactions + The corresponding public key """ - try: - # First fetch a larger set to search within - raw_transactions = await self.transaction_manager.get_all_transactions(public_key, limit=200) - - # Filter by memo - filtered = [ - tx for tx in raw_transactions - if memo.lower() in tx.get('memo', '').lower() - ] - - # Apply limit - limited = filtered[:limit] - - return { - 'success': True, - 'transactions': limited, - 'total': len(limited), - 'totalFound': len(filtered), - 'searchMemo': memo, - 'publicKey': public_key, - 'message': f'Found {len(filtered)} transactions containing "{memo}"', - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return { - 'success': False, - 'error': str(e), - 'transactions': [], - 'timestamp': datetime.now().isoformat() - } + return self.payment_manager.get_public_key_from_secret(secret_key) - def get_public_key_from_secret(self, secret_key: str) -> str: + async def get_transactions( + self, + public_key: str, + limit: int = 200 + ) -> List[Dict[str, Any]]: """ - Extract public key from secret key + Get DZT transactions for an account Args: - secret_key: The secret key + public_key: Public key of the account + limit: Maximum number of transactions to retrieve Returns: - The corresponding public key - - Example: - ```python - client = SofizPayClient() - public_key = client.get_public_key_from_secret("SECRET_KEY_HERE") - ``` + List of transaction dictionaries """ - return self.payment_manager.get_public_key_from_secret(secret_key) + return await self.transaction_manager.get_transactions(public_key, limit) - async def get_transactions( + async def get_all_transactions( self, public_key: str, limit: int = 200 - ) -> Dict[str, Any]: + ) -> List[Dict[str, Any]]: """ - Get transactions for an account + Get all raw Stellar transactions for an account Args: public_key: Public key of the account limit: Maximum number of transactions to retrieve Returns: - Standard result object with filtered transactions + List of all transaction dictionaries """ - try: - transactions = await self.transaction_manager.get_transactions(public_key, limit) - return { - 'success': True, - 'transactions': transactions, - 'total': len(transactions), - 'publicKey': public_key, - 'message': f'Fetched DZT transactions ({len(transactions)})', - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return { - 'success': False, - 'error': str(e), - 'transactions': [], - 'timestamp': datetime.now().isoformat() - } + return await self.transaction_manager.get_all_transactions(public_key, limit) async def get_transaction_by_hash(self, transaction_hash: str) -> Dict[str, Any]: """ @@ -247,17 +156,28 @@ async def get_transaction_by_hash(self, transaction_hash: str) -> Dict[str, Any] Returns: Detailed transaction information - - Example: - ```python - client = SofizPayClient() - result = await client.get_transaction_by_hash("TRANSACTION_HASH_HERE") - if result['found']: - tx = result['transaction'] - ``` """ return await self.transaction_manager.get_transaction_by_hash(transaction_hash) + async def search_transactions_by_memo( + self, + public_key: str, + memo: str, + limit: int = 50 + ) -> Dict[str, Any]: + """ + Search transactions by memo keyword + + Args: + public_key: Public key of the account + memo: Memo text to search for + limit: Maximum number of transactions to return + + Returns: + Dictionary containing matched transactions and search metadata + """ + return await self.transaction_manager.search_transactions_by_memo(public_key, memo, limit) + async def setup_transaction_stream( self, public_key: str, @@ -266,39 +186,16 @@ async def setup_transaction_stream( check_interval: int = 30 ) -> str: """ - Set up real-time transaction streaming for an account + Set up real-time transaction streaming for an account Args: public_key: Public key to monitor transaction_callback: Callback function to handle new transactions - from_now: If True, only new transactions will be streamed; if False, both new and historical transactions will be included - check_interval: Duration in seconds for repeated network checks (default 30 seconds) + from_now: If True, only new transactions are streamed; if False, includes history + check_interval: Duration in seconds for network polling checks (default 30s) Returns: Stream ID for managing the stream - - Example: - ```python - client = SofizPayClient() - - def handle_transaction(transaction): - print("New transaction:", transaction) - - - stream_id = await client.setup_transaction_stream( - "PUBLIC_KEY_HERE", - handle_transaction, - from_now=True, - check_interval=10 - ) - - stream_id = await client.setup_transaction_stream( - "PUBLIC_KEY_HERE", - handle_transaction, - from_now=False, - check_interval=60 - ) - ``` """ return await self.transaction_manager.setup_transaction_stream( public_key, transaction_callback, from_now=from_now, check_interval=check_interval @@ -313,61 +210,42 @@ def stop_transaction_stream(self, stream_id: str) -> bool: Returns: True if stream was stopped, False if not found - - Example: - ```python - client = SofizPayClient() - success = client.stop_transaction_stream(stream_id) - if success: - ``` """ return self.transaction_manager.stop_transaction_stream(stream_id) - - @classmethod - def verify_signature(cls, message: str, signature: str) -> bool: + + # ========================================================================= + # CIB & EDAHABIA GATEWAY & STATUS CHECK + # ========================================================================= + + async def make_cib_transaction(self, transaction_data: Dict[str, Any]) -> Dict[str, Any]: """ - Verify a signature against a message using SofizPay's official public key + Make a CIB / EDAHABIA payment transaction through SofizPay Args: - message: The original message - signature: The signature to verify - + transaction_data: Dictionary containing transaction details: + - account (str): Required. SofizPay account / public key + - amount (float|int): Required. Payment amount in DZD (must be > 0) + - full_name (str): Required. Customer full name + - phone (str): Required. Customer phone number + - email (str): Required. Customer email address + - return_url (str, optional): Redirect URL after payment + - webhook_url (str, optional): Async webhook notification URL + - invoice_id (str, optional): Optional linked invoice ID + - language (str, optional): Language for payment gateway ('ar' | 'en' | 'fr') + - memo (str, optional): Optional memo for the transaction + - redirect (bool|str, optional): 'yes' | 'no' (defaults to 'no') + - keep_return_url (bool|str, optional): 'True' | 'False' + - is_sandbox (bool, optional): Use Sandbox environment + Returns: - True if signature is valid, False otherwise + Dictionary with transaction result including payment_url and identifiers - Example: - ```python - client = SofizPayClient() - is_valid = client.verify_signature( - message="Hello, world!", - signature="SIGNATURE_HERE" - ) - ``` + Raises: + ValidationError: When required fields are missing or invalid """ - try: - decoded_signature = base64.b64decode(signature) - - public_key_obj = serialization.load_pem_public_key(cls.SOFIZPAY_PUBLIC_KEY_PEM.encode()) - - public_key_obj.verify( - decoded_signature, - message.encode(), - padding.PKCS1v15(), - hashes.SHA256() - ) - return True - except (InvalidSignature, ValueError, TypeError) as e: - return False - - async def make_cib_transaction(self, transaction_data: Dict[str, Any]) -> Dict[str, Any]: - """Initiate a CIB transaction (Production only)""" - return await self._make_cib_request(transaction_data, False) - - async def make_sandbox_cib_transaction(self, transaction_data: Dict[str, Any]) -> Dict[str, Any]: - """Initiate a CIB transaction specifically in Sandbox mode""" - return await self._make_cib_request(transaction_data, True) - - async def _make_cib_request(self, transaction_data: Dict[str, Any], use_sandbox: bool) -> Dict[str, Any]: + if not transaction_data: + raise ValidationError('Transaction data is required') + if not transaction_data.get('account'): raise ValidationError('Account is required') @@ -384,31 +262,60 @@ async def _make_cib_request(self, transaction_data: Dict[str, Any], use_sandbox: raise ValidationError('Email is required') try: + is_sandbox = bool( + transaction_data.get('is_sandbox') + if 'is_sandbox' in transaction_data + else transaction_data.get('isSandbox', self.is_sandbox) + ) + base_url = ( 'https://sofizpay.com/sandbox/make-cib-transaction/' - if use_sandbox - else 'https://www.sofizpay.com/make-cib-transaction/' + if is_sandbox + else 'https://sofizpay.com/make-cib-transaction/' ) query_params = [] query_params.append(f"account={urllib.parse.quote(str(transaction_data['account']))}") query_params.append(f"amount={transaction_data['amount']}") - query_params.append(f"full_name={urllib.parse.quote(transaction_data['full_name'])}") - query_params.append(f"phone={urllib.parse.quote(transaction_data['phone'])}") - query_params.append(f"email={urllib.parse.quote(transaction_data['email'])}") + query_params.append(f"full_name={urllib.parse.quote(str(transaction_data['full_name']))}") + query_params.append(f"phone={urllib.parse.quote(str(transaction_data['phone']))}") + query_params.append(f"email={urllib.parse.quote(str(transaction_data['email']))}") - # Add optional parameters if transaction_data.get('return_url'): - query_params.append(f"return_url={urllib.parse.quote(transaction_data['return_url'])}") + query_params.append(f"return_url={urllib.parse.quote(str(transaction_data['return_url']))}") + + if transaction_data.get('webhook_url'): + query_params.append(f"webhook_url={urllib.parse.quote(str(transaction_data['webhook_url']))}") + + if transaction_data.get('invoice_id'): + query_params.append(f"invoice_id={urllib.parse.quote(str(transaction_data['invoice_id']))}") + + if transaction_data.get('language'): + query_params.append(f"language={urllib.parse.quote(str(transaction_data['language']))}") if transaction_data.get('memo'): - safe_memo = urllib.parse.quote(transaction_data['memo']) + safe_memo = urllib.parse.quote(str(transaction_data['memo'])) query_params.append(f"memo={safe_memo}") - query_params.append("redirect=no") + if 'redirect' in transaction_data: + redirect_val = transaction_data['redirect'] + if isinstance(redirect_val, bool): + redirect_str = 'yes' if redirect_val else 'no' + else: + redirect_str = str(redirect_val) + query_params.append(f"redirect={urllib.parse.quote(redirect_str)}") + else: + query_params.append("redirect=no") - full_url = f"{base_url}?{'&'.join(query_params)}" + if 'keep_return_url' in transaction_data: + kru_val = transaction_data['keep_return_url'] + if isinstance(kru_val, bool): + kru_str = 'True' if kru_val else 'False' + else: + kru_str = str(kru_val) + query_params.append(f"keep_return_url={urllib.parse.quote(kru_str)}") + full_url = f"{base_url}?{'&'.join(query_params)}" response = requests.get( full_url, @@ -420,50 +327,70 @@ async def _make_cib_request(self, transaction_data: Dict[str, Any], use_sandbox: timeout=30 ) - response.raise_for_status() + response_data = response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text + + payment_url = None + transaction_id = None + cib_transaction_id = None + order_id = None + + if isinstance(response_data, dict): + payment_url = response_data.get('payment_url') + if not payment_url and isinstance(response_data.get('cib_response'), dict): + payment_url = response_data['cib_response'].get('formUrl') + + transaction_id = response_data.get('transaction_id') + cib_transaction_id = response_data.get('cib_transaction_id') + order_id = response_data.get('order_id') + is_success = response_data.get('status') != 'error' and response_data.get('success') is not False + else: + is_success = response.status_code == 200 return { - 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, - 'status': response.status_code, - 'status_text': response.reason, - 'headers': dict(response.headers), + 'success': is_success, + 'data': response_data, + 'payment_url': payment_url, + 'transaction_id': transaction_id, + 'cib_transaction_id': cib_transaction_id, + 'order_id': order_id, + 'webhook_url': (response_data.get('webhook_url') if isinstance(response_data, dict) else None) or transaction_data.get('webhook_url'), + 'account': transaction_data['account'], + 'amount': transaction_data['amount'], + 'full_name': transaction_data['full_name'], + 'phone': transaction_data['phone'], + 'email': transaction_data['email'], + 'memo': transaction_data.get('memo'), + 'is_sandbox': is_sandbox, + 'status_code': response.status_code, 'url': full_url, - 'request_data': { - 'account': transaction_data['account'], - 'amount': transaction_data['amount'], - 'full_name': transaction_data['full_name'], - 'phone': transaction_data['phone'], - 'email': transaction_data['email'], - 'return_url': transaction_data.get('return_url'), - 'memo': transaction_data.get('memo'), - 'redirect': 'no' - }, 'timestamp': datetime.now().isoformat() } except requests.exceptions.HTTPError as e: error_message = f"HTTP Error: {e.response.status_code} - {e.response.reason}" - + error_data = None try: error_data = e.response.json() - if 'error' in error_data: - error_message += f" - {error_data['error']}" - except: + if isinstance(error_data, dict): + if 'message' in error_data: + error_message += f" - {error_data['message']}" + elif 'error' in error_data: + error_message += f" - {error_data['error']}" + except Exception: pass return { 'success': False, 'error': error_message, - 'account': transaction_data['account'], - 'amount': transaction_data['amount'], + 'error_data': error_data, + 'account': transaction_data.get('account'), + 'amount': transaction_data.get('amount'), 'timestamp': datetime.now().isoformat(), - 'status_code': e.response.status_code + 'status_code': e.response.status_code if hasattr(e, 'response') and e.response else None } except requests.exceptions.RequestException as e: error_message = f"Request error: {str(e)}" - if isinstance(e, requests.exceptions.Timeout): error_message = "Request timeout: Server took too long to respond" elif isinstance(e, requests.exceptions.ConnectionError): @@ -472,8 +399,8 @@ async def _make_cib_request(self, transaction_data: Dict[str, Any], use_sandbox: return { 'success': False, 'error': error_message, - 'account': transaction_data['account'], - 'amount': transaction_data['amount'], + 'account': transaction_data.get('account'), + 'amount': transaction_data.get('amount'), 'timestamp': datetime.now().isoformat() } @@ -481,193 +408,691 @@ async def _make_cib_request(self, transaction_data: Dict[str, Any], use_sandbox: return { 'success': False, 'error': f"Unexpected error: {str(e)}", - 'account': transaction_data['account'], - 'amount': transaction_data['amount'], + 'account': transaction_data.get('account'), + 'amount': transaction_data.get('amount'), 'timestamp': datetime.now().isoformat() } - def verify_sofizpay_signature(self, verification_data: Dict[str, str]) -> bool: + async def make_sandbox_cib_transaction(self, transaction_data: Dict[str, Any]) -> Dict[str, Any]: """ - Verify a signature from SofizPay using the official public key + Dedicated helper to initiate a CIB transaction specifically in Sandbox mode Args: - verification_data: Dictionary containing: - - message (str): The original message that was signed - - signature_url_safe (str): The URL-safe base64 encoded signature - - Returns: - True if signature is valid, False otherwise + transaction_data: CIB transaction data dictionary + Returns: + Dictionary with sandbox transaction result + """ + data = dict(transaction_data) + data['is_sandbox'] = True + return await self.make_cib_transaction(data) - ``` + async def check_cib_transaction(self, data: Union[str, int, Dict[str, Any]]) -> Dict[str, Any]: """ - if not verification_data.get('message'): - return False + Check CIB transaction status by order number / CIB transaction ID - if not verification_data.get('signature_url_safe'): - return False + Args: + data: Order number string/int OR dictionary containing: + - order_number / orderNumber / cib_transaction_id / order_id: Order ID + - is_sandbox / isSandbox (bool, optional): Use sandbox endpoint + + Returns: + Dictionary with verification status (success, status='paid'/'pending', amount, etc.) + """ + order_number = None + is_sandbox = self.is_sandbox + + if isinstance(data, (str, int)): + order_number = str(data) + elif isinstance(data, dict): + order_number = ( + data.get('order_number') or + data.get('orderNumber') or + data.get('cib_transaction_id') or + data.get('order_id') or + data.get('orderId') + ) + if 'is_sandbox' in data: + is_sandbox = bool(data['is_sandbox']) + elif 'isSandbox' in data: + is_sandbox = bool(data['isSandbox']) + + if not order_number: + raise ValidationError('Order number is required') try: - signature_url_safe = verification_data['signature_url_safe'] - base64_signature = signature_url_safe.replace('-', '+').replace('_', '/') - - while len(base64_signature) % 4: - base64_signature += '=' - - signature_bytes = base64.b64decode(base64_signature) - - public_key = serialization.load_pem_public_key( - self.SOFIZPAY_PUBLIC_KEY_PEM.encode() + base_url = ( + 'https://sofizpay.com/sandbox/cib-transaction-check/' + if is_sandbox + else 'https://sofizpay.com/cib-transaction-check/' ) - public_key.verify( - signature_bytes, - verification_data['message'].encode('utf-8'), - padding.PKCS1v15(), - hashes.SHA256() + response = requests.get( + base_url, + params={'order_number': str(order_number)}, + headers={ + 'Accept': 'application/json', + 'Content-Type': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, + timeout=30 ) - return True + response_data = response.json() if response.headers.get('content-type', '').startswith('application/json') else {} - except InvalidSignature: - return False - except Exception as e: - return False + is_success = False + if isinstance(response_data, dict): + is_success = ( + response_data.get('errorCode') == 0 or + response_data.get('orderStatus') == 2 or + response_data.get('status') == 'success' or + response_data.get('respCode') == '00' + ) + + return { + 'success': is_success, + 'data': response_data, + 'order_number': response_data.get('order_number') or str(order_number), + 'order_status': response_data.get('orderStatus'), + 'status': 'paid' if is_success else (response_data.get('status') or 'pending'), + 'amount': response_data.get('Amount') or response_data.get('amount'), + 'error_message': response_data.get('errorMessage'), + 'is_sandbox': is_sandbox, + 'timestamp': datetime.now().isoformat() + } + + except Exception as error: + return self._handle_requests_error(error, extra={'order_number': str(order_number)}) - async def recharge_phone(self, data: Dict[str, Any]) -> Dict[str, Any]: + async def check_cib_status(self, cib_transaction_id: str) -> Dict[str, Any]: """ - Phone recharge service + Check status of a CIB transaction in Production mode + Args: - data: {encrypted_sk, phone, operator, amount, offer} + cib_transaction_id: The CIB transaction ID / order number + + Returns: + Status check result dictionary """ - return await self._perform_service_operation(data) + return await self.check_cib_transaction({ + 'order_number': cib_transaction_id, + 'is_sandbox': False + }) - async def recharge_internet(self, data: Dict[str, Any]) -> Dict[str, Any]: + async def check_sandbox_cib_status(self, cib_transaction_id: str) -> Dict[str, Any]: """ - Internet recharge service + Check status of a CIB transaction specifically in Sandbox mode + Args: - data: {encrypted_sk, phone, operator, amount, offer} + cib_transaction_id: The CIB transaction ID / order number + + Returns: + Status check result dictionary """ - return await self._perform_service_operation(data) + return await self.check_cib_transaction({ + 'order_number': cib_transaction_id, + 'is_sandbox': True + }) - async def recharge_game(self, data: Dict[str, Any]) -> Dict[str, Any]: - """ - Game recharge service - Args: - data: {encrypted_sk, operator, playerId, amount, offer} - """ - return await self._perform_service_operation(data) + async def cib_transaction_check(self, data: Union[str, int, Dict[str, Any]]) -> Dict[str, Any]: + """Alias for check_cib_transaction""" + return await self.check_cib_transaction(data) - async def pay_bill(self, data: Dict[str, Any]) -> Dict[str, Any]: - """ - Bill payment service - Args: - data: {encrypted_sk, operator, ...utility_fields} - """ - return await self._perform_service_operation(data) + # ========================================================================= + # ALGERIAN SERVICES & UTILITIES (SONELGAZ, ADE, TELECOM, FLEXY, GAMES) + # ========================================================================= + + def _handle_requests_error(self, error: Exception, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """Internal helper for handling request errors uniformly""" + error_message = str(error) + error_data = None + + if hasattr(error, 'response') and error.response is not None: + try: + error_data = error.response.json() + if isinstance(error_data, dict): + if 'message' in error_data: + error_message = error_data['message'] + elif 'error' in error_data: + error_message = error_data['error'] + else: + error_message = f"HTTP Error: {error.response.status_code} - {error.response.reason}" + except Exception: + error_message = f"HTTP Error: {error.response.status_code} - {error.response.reason}" + + result = { + 'success': False, + 'error': error_message, + 'error_data': error_data, + 'timestamp': datetime.now().isoformat() + } + if extra: + result.update(extra) + return result async def _perform_service_operation(self, data: Dict[str, Any]) -> Dict[str, Any]: - """Internal helper for service operations""" + """Internal helper to execute service operations POST request""" try: + url = 'https://sofizpay.com/services/operation_post' response = requests.post( - 'https://www.sofizpay.com/services/operation_post', + url, json=data, - headers={'Content-Type': 'application/json'}, - timeout=30 + headers={ + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, + timeout=35 ) response.raise_for_status() + response_data = response.json() + + is_success = ( + response_data.get('status') == 'success' or + response_data.get('transaction_status') == 'confirmed' + ) + return { - 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return { - 'success': False, - 'error': str(e), + 'success': is_success, + 'status': response_data.get('status') or ('success' if is_success else 'failed'), + 'message': response_data.get('message'), + 'operation_id': response_data.get('operation_id'), + 'transaction_id': response_data.get('transaction_id'), + 'transaction_status': response_data.get('transaction_status'), + 'data': response_data, 'timestamp': datetime.now().isoformat() } + except Exception as error: + return self._handle_requests_error(error) - async def get_operation_details(self, operation_id: str, encrypted_sk: str) -> Dict[str, Any]: - """Get details of a specific operation""" - try: - response = requests.get( - f'https://sofizpay.com/services/operation-detail/{operation_id}/', - params={'encrypted_sk': encrypted_sk}, - timeout=30 + async def get_products( + self, + options: Union[str, Dict[str, Any]], + search: Optional[str] = None + ) -> Dict[str, Any]: + """ + Retrieve catalog of available products and services + + Args: + options: Encrypted secret key string OR options dictionary: + - encrypted_sk (str): Encrypted or plain Stellar secret key (starts with 'S') + - search (str, optional): Search keyword filter + search: Optional search filter keyword (if options is string) + + Returns: + Dictionary containing available products and count + """ + encrypted_sk = None + search_kw = None + + if isinstance(options, str): + encrypted_sk = options + search_kw = search + elif isinstance(options, dict): + encrypted_sk = ( + options.get('encrypted_sk') or + options.get('secretKey') or + options.get('secretkey') or + options.get('secret_key') ) - response.raise_for_status() + search_kw = options.get('search') or search + + if not encrypted_sk: + raise ValidationError('encrypted_sk (or secret key) is required.') + + try: + url = 'https://sofizpay.com/services/get_products/' + payload = {'encrypted_sk': encrypted_sk} + if search_kw: + payload['search'] = search_kw + + # First try POST, fallback to GET + try: + response = requests.post( + url, + json=payload, + headers={ + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, + timeout=30 + ) + response.raise_for_status() + except Exception: + response = requests.get( + url, + params=payload, + headers={ + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, + timeout=30 + ) + response.raise_for_status() + + data = response.json() + + products_list = [] + if isinstance(data, dict) and 'products' in data and isinstance(data['products'], list): + products_list = data['products'] + elif isinstance(data, list): + products_list = data + + count = len(products_list) + is_success = ( + isinstance(data, dict) and data.get('status') == 'success' + ) or isinstance(data, list) or len(products_list) > 0 + return { - 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, + 'success': is_success, + 'status': data.get('status', 'success') if isinstance(data, dict) else 'success', + 'count': count, + 'products': products_list, + 'raw': data, 'timestamp': datetime.now().isoformat() } - except Exception as e: - return {'success': False, 'error': str(e), 'timestamp': datetime.now().isoformat()} + except Exception as error: + return self._handle_requests_error(error) - async def get_operation_history(self, encrypted_sk: str, limit: int = 10, offset: int = 0) -> Dict[str, Any]: - """Get operation history""" + async def get_operation_history( + self, + encrypted_sk: str, + limit: int = 10, + offset: int = 0 + ) -> Dict[str, Any]: + """ + Get service operation history + + Args: + encrypted_sk: Encrypted secret key or plain secret key + limit: Page limit (default 10) + offset: Page offset (default 0) + + Returns: + Dictionary containing operation history + """ + if not encrypted_sk: + raise ValidationError('encrypted_sk is required.') + try: response = requests.get( 'https://sofizpay.com/services/operation-history/', - params={'encrypted_sk': encrypted_sk, 'limit': limit, 'offset': offset}, + params={ + 'encrypted_sk': encrypted_sk, + 'limit': limit, + 'offset': offset + }, + headers={ + 'Accept': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, timeout=30 ) response.raise_for_status() return { 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, + 'data': response.json(), 'timestamp': datetime.now().isoformat() } - except Exception as e: - return {'success': False, 'error': str(e), 'timestamp': datetime.now().isoformat()} + except Exception as error: + return self._handle_requests_error(error) - async def get_products(self, encrypted_sk: str = None) -> Dict[str, Any]: - """Get available products/services""" + async def get_operation_details( + self, + options: Union[str, Dict[str, Any]], + encrypted_sk: Optional[str] = None + ) -> Dict[str, Any]: + """ + Retrieve operation details by operation UUID + + Args: + options: Operation UUID string OR dictionary with operation_id & encrypted_sk + encrypted_sk: Encrypted secret key (if options is string) + + Returns: + Dictionary containing operation details + """ + operation_id = None + sk = None + + if isinstance(options, str): + operation_id = options + sk = encrypted_sk + elif isinstance(options, dict): + operation_id = ( + options.get('operation_id') or + options.get('operationId') or + options.get('id') + ) + sk = ( + options.get('encrypted_sk') or + options.get('secretKey') or + options.get('secretkey') or + options.get('secret_key') or + encrypted_sk + ) + + if not operation_id: + raise ValidationError('Operation ID is required.') + if not sk: + raise ValidationError('encrypted_sk is required.') + try: + url = f'https://sofizpay.com/services/operation-detail/{operation_id}/' response = requests.get( - 'https://sofizpay.com/services/get_products/', - json={'encrypted_sk': encrypted_sk} if encrypted_sk else {}, + url, + params={'encrypted_sk': sk}, + headers={ + 'Accept': 'application/json', + 'User-Agent': f'SofizPay-Python-SDK/{self.VERSION}' + }, timeout=30 ) response.raise_for_status() return { 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, + 'data': response.json(), + 'operation_id': operation_id, 'timestamp': datetime.now().isoformat() } - except Exception as e: - return {'success': False, 'error': str(e), 'timestamp': datetime.now().isoformat()} + except Exception as error: + return self._handle_requests_error(error) - async def check_cib_status(self, cib_transaction_id: str) -> Dict[str, Any]: - """Check status of a CIB transaction (Production only)""" - return await self._check_cib_status_request(cib_transaction_id, False) + async def execute_service_operation(self, operation_data: Dict[str, Any]) -> Dict[str, Any]: + """ + Generic execution of /services/operation_post for bills, recharges, and games + + Args: + operation_data: Operation payload dictionary + + Returns: + Operation result dictionary + """ + if not operation_data: + raise ValidationError('Operation data is required.') + + sk = ( + operation_data.get('encrypted_sk') or + operation_data.get('secretKey') or + operation_data.get('secretkey') or + operation_data.get('secret_key') + ) + if not sk: + raise ValidationError('encrypted_sk (or secret key) is required.') + + if not operation_data.get('operator'): + raise ValidationError('Operator is required.') + + amount = operation_data.get('amount') + if amount is None or float(amount) <= 0: + raise ValidationError('Valid amount is required.') + + payload = dict(operation_data) + payload['encrypted_sk'] = sk + + return await self._perform_service_operation(payload) - async def check_sandbox_cib_status(self, cib_transaction_id: str) -> Dict[str, Any]: - """Check status of a CIB transaction specifically in Sandbox mode""" - return await self._check_cib_status_request(cib_transaction_id, True) + async def pay_bill(self, bill_data: Dict[str, Any]) -> Dict[str, Any]: + """ + Pay utility bills (Sonelgaz, ADE, Algรฉrie Tรฉlรฉcom) + + Args: + bill_data: Bill payment dictionary: + - encrypted_sk: Encrypted secret key or plain secret key + - amount: Payment amount in DZD + - operator: 'ade' | 'sonelgaz' | 'algerie_telecom' + - offer (optional): Offer name (defaults to operator name) + - bill (optional): Bill number (Required for ADE and Sonelgaz) + - customerId / customer_id (optional): Customer ID (Required for Sonelgaz) + - ebb (optional): EBB number (Required for Sonelgaz) + - phone (optional): Phone number (For Algรฉrie Tรฉlรฉcom) + + Returns: + Payment result dictionary + """ + if not bill_data: + raise ValidationError('Bill payment data is required.') + + operator = str(bill_data.get('operator', '')).lower() + sk = ( + bill_data.get('encrypted_sk') or + bill_data.get('secretKey') or + bill_data.get('secretkey') or + bill_data.get('secret_key') + ) + + payload = { + 'encrypted_sk': sk, + 'amount': bill_data.get('amount'), + 'operator': operator, + 'offer': bill_data.get('offer') or operator + } + + if operator == 'ade': + if not bill_data.get('bill'): + raise ValidationError('Bill number ("bill") is required for ADE water bill payment.') + payload['bill'] = bill_data['bill'] + elif operator == 'sonelgaz': + if not bill_data.get('bill'): + raise ValidationError('Bill number ("bill") is required for Sonelgaz bill payment.') + customer_id = bill_data.get('customerId') or bill_data.get('customer_id') + if not customer_id: + raise ValidationError('Customer ID ("customerId") is required for Sonelgaz bill payment.') + if not bill_data.get('ebb'): + raise ValidationError('EBB number ("ebb") is required for Sonelgaz bill payment.') + payload['customerId'] = customer_id + payload['ebb'] = bill_data['ebb'] + payload['bill'] = bill_data['bill'] + elif operator in ('algerie_telecom', 'telecom'): + payload['operator'] = 'algerie_telecom' + payload['offer'] = bill_data.get('offer') or 'algerie_telecom' + if bill_data.get('phone'): + payload['phone'] = bill_data['phone'] + if bill_data.get('bill'): + payload['bill'] = bill_data['bill'] + else: + payload.update(bill_data) + + return await self.execute_service_operation(payload) + + async def pay_ade_bill(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Helper to pay ADE (Algรฉrienne Des Eaux) water bill + + Args: + data: Dictionary with encrypted_sk, amount, bill + """ + params = dict(data) + params['operator'] = 'ade' + params['offer'] = 'ade' + return await self.pay_bill(params) + + async def pay_sonelgaz_bill(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Helper to pay Sonelgaz electricity & gas bill + + Args: + data: Dictionary with encrypted_sk, amount, customerId/customer_id, ebb, bill + """ + params = dict(data) + params['operator'] = 'sonelgaz' + params['offer'] = 'sonelgaz' + return await self.pay_bill(params) + + async def pay_algerie_telecom_bill(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Helper to pay Algรฉrie Tรฉlรฉcom bill + + Args: + data: Dictionary with encrypted_sk, amount, phone, bill + """ + params = dict(data) + params['operator'] = 'algerie_telecom' + params['offer'] = 'algerie_telecom' + return await self.pay_bill(params) + + async def recharge_phone(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Recharge phone credit (Flexy: Mobilis, Djezzy, Ooredoo) + + Args: + data: Dictionary containing: + - encrypted_sk: Encrypted or plain secret key + - phone: 10-digit Algerian phone number + - operator: 'mobilis' | 'djezzy' | 'ooredoo' + - amount: Flexy amount in DZD + - offer (optional): 'prepaid' | 'postpaid' (defaults to 'prepaid') + """ + if not data: + raise ValidationError('Phone recharge data is required.') + if not data.get('phone'): + raise ValidationError('Phone number is required.') + + return await self.execute_service_operation({ + 'encrypted_sk': ( + data.get('encrypted_sk') or + data.get('secretKey') or + data.get('secretkey') or + data.get('secret_key') + ), + 'phone': data['phone'], + 'operator': str(data.get('operator', '')).lower(), + 'amount': data.get('amount'), + 'offer': data.get('offer') or 'prepaid' + }) + + async def recharge_internet(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Recharge IDOOM Internet (ADSL / 4G LTE) + + Args: + data: Dictionary containing: + - encrypted_sk: Encrypted or plain secret key + - phone: Phone/subscription number (10 digits for 4G, 9 digits for ADSL) + - amount: Recharge amount in DZD + - offer: Offer name (e.g. 'IDOOM 4G 1000' or 'IDOOM ADSL 2000') + - operator (optional): Defaults to 'idoom' + """ + if not data: + raise ValidationError('Internet recharge data is required.') + if not data.get('phone'): + raise ValidationError('Phone/subscription number is required.') + if not data.get('offer'): + raise ValidationError('Offer name is required (e.g., "IDOOM 4G 1000").') + + return await self.execute_service_operation({ + 'encrypted_sk': ( + data.get('encrypted_sk') or + data.get('secretKey') or + data.get('secretkey') or + data.get('secret_key') + ), + 'phone': data['phone'], + 'operator': str(data.get('operator') or 'idoom').lower(), + 'amount': data.get('amount'), + 'offer': data['offer'] + }) + + async def recharge_game(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + Purchase gaming credits (PUBG, Free Fire, etc.) + + Args: + data: Dictionary containing: + - encrypted_sk: Encrypted or plain secret key + - operator: 'pubg' | 'freefire' + - player_id / playerId: In-game Player ID + - amount: Recharge amount in DZD + - offer: Offer code (e.g., "60" for PUBG, "110" for Free Fire) + """ + if not data: + raise ValidationError('Game recharge data is required.') + player_id = data.get('playerId') or data.get('player_id') + if not player_id: + raise ValidationError('Player ID is required.') + if not data.get('offer'): + raise ValidationError('Offer is required (e.g. "60" or "110").') + + return await self.execute_service_operation({ + 'encrypted_sk': ( + data.get('encrypted_sk') or + data.get('secretKey') or + data.get('secretkey') or + data.get('secret_key') + ), + 'operator': str(data.get('operator', '')).lower(), + 'playerId': player_id, + 'amount': data.get('amount'), + 'offer': str(data['offer']) + }) + + # ========================================================================= + # SIGNATURE VERIFICATION + # ========================================================================= - async def _check_cib_status_request(self, cib_transaction_id: str, use_sandbox: bool) -> Dict[str, Any]: - """Internal helper for checking CIB status""" + @classmethod + def verify_signature(cls, verification_data: Union[Dict[str, str], str], signature: Optional[str] = None) -> bool: + """ + Verify a signature against a message using SofizPay's official public key + + Args: + verification_data: Dict with 'message' and 'signature_url_safe' OR string message + signature: String signature (if verification_data is a message string) + + Returns: + True if signature is valid, False otherwise + """ + message = "" + sig = "" + + if isinstance(verification_data, dict): + message = verification_data.get('message', '') + sig = verification_data.get('signature_url_safe') or verification_data.get('signature', '') + elif isinstance(verification_data, str): + message = verification_data + sig = signature or "" + + if not message or not sig: + return False + try: - base_url = ( - 'https://sofizpay.com/sandbox/cib-transaction-check/' - if use_sandbox - else 'https://www.sofizpay.com/cib-transaction-check/' - ) - response = requests.get( - base_url, - params={'order_number': cib_transaction_id}, - timeout=30 + # Handle url-safe base64 + base64_sig = sig.replace('-', '+').replace('_', '/') + while len(base64_sig) % 4: + base64_sig += '=' + + signature_bytes = base64.b64decode(base64_sig) + + public_key_obj = serialization.load_pem_public_key(cls.SOFIZPAY_PUBLIC_KEY_PEM.encode()) + + if not isinstance(public_key_obj, RSAPublicKey): + return False + + public_key_obj.verify( + signature_bytes, + message.encode('utf-8'), + padding.PKCS1v15(), + hashes.SHA256() ) - response.raise_for_status() - return { - 'success': True, - 'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text, - 'timestamp': datetime.now().isoformat() - } - except Exception as e: - return {'success': False, 'error': str(e), 'timestamp': datetime.now().isoformat()} + return True + except (InvalidSignature, ValueError, TypeError, Exception): + return False + + def verify_sofizpay_signature(self, verification_data: Dict[str, str]) -> bool: + """ + Verify a signature from SofizPay using the official public key + + Args: + verification_data: Dictionary containing: + - message (str): The original message that was signed + - signature_url_safe (str): The URL-safe base64 encoded signature + + Returns: + True if signature is valid, False otherwise + """ + return self.verify_signature(verification_data) async def __aenter__(self): """Async context manager entry""" diff --git a/sofizpay/payments.py b/sofizpay/payments.py index 15608f7..050004d 100644 --- a/sofizpay/payments.py +++ b/sofizpay/payments.py @@ -1,11 +1,9 @@ -"""Payment management for SofizPay SDK""" - import time from datetime import datetime from typing import Optional, Dict, Any from stellar_sdk import ( Server, Keypair, Asset, TransactionBuilder, - Network, Memo + Network ) from stellar_sdk.operation import Payment from stellar_sdk.exceptions import SdkError diff --git a/sofizpay/transactions.py b/sofizpay/transactions.py index 07743d0..0cbdf29 100644 --- a/sofizpay/transactions.py +++ b/sofizpay/transactions.py @@ -377,6 +377,70 @@ async def get_transaction_by_hash(self, transaction_hash: str) -> Dict[str, Any] except Exception as e: return {} + async def search_transactions_by_memo( + self, + public_key: str, + memo: str, + limit: int = 50 + ) -> Dict[str, Any]: + """ + Search transactions by memo substring + + Args: + public_key: Public key of the account + memo: Memo text to search for + limit: Maximum number of matching transactions to return + + Returns: + Dictionary containing search results and metadata + """ + if not validate_public_key(public_key): + raise ValidationError("Invalid public key") + + if not memo: + raise ValidationError("Memo is required for search") + + try: + transactions = await self.get_transactions(public_key, 200) + + if not transactions: + return { + 'success': True, + 'transactions': [], + 'total': 0, + 'totalFound': 0, + 'searchMemo': memo, + 'publicKey': public_key, + 'message': 'There are no transactions in this account', + 'timestamp': datetime.now(timezone.utc).isoformat() + } + + filtered = [ + tx for tx in transactions + if tx.get('memo') and memo.lower() in str(tx.get('memo', '')).lower() + ] + + limited = filtered[:limit] + + return { + 'success': True, + 'transactions': limited, + 'total': len(limited), + 'totalFound': len(filtered), + 'searchMemo': memo, + 'publicKey': public_key, + 'message': f'Found {len(filtered)} transactions containing "{memo}"', + 'timestamp': datetime.now(timezone.utc).isoformat() + } + except Exception as e: + return { + 'success': False, + 'error': str(e), + 'transactions': [], + 'searchMemo': memo, + 'timestamp': datetime.now(timezone.utc).isoformat() + } + def __del__(self): """Cleanup streaming tasks when object is destroyed""" for stream_id in list(self._streaming_tasks.keys()): diff --git a/sofizpay/utils.py b/sofizpay/utils.py index 4928d22..ecc07bc 100644 --- a/sofizpay/utils.py +++ b/sofizpay/utils.py @@ -2,7 +2,7 @@ import asyncio import time -from typing import Callable, Any, Optional +from typing import Callable, Any, Optional, Tuple from stellar_sdk import Keypair from .exceptions import ValidationError, NetworkError import requests @@ -127,7 +127,7 @@ def validate_amount(amount: str) -> bool: return False -def validate_memo(memo: str) -> tuple[bool, str]: +def validate_memo(memo: str) -> Tuple[bool, str]: """ Validate and optionally truncate memo diff --git a/test_sandbox.py b/test_sandbox.py deleted file mode 100644 index e24ce09..0000000 --- a/test_sandbox.py +++ /dev/null @@ -1,38 +0,0 @@ -import asyncio -import json -from sofizpay.client import SofizPayClient - -async def main(): - # 1. Initialize client in Sandbox Mode - client = SofizPayClient(sandbox=True) - - print("--- Starting SofizPay Python SDK Sandbox Test ---") - print("Current Mode: SANDBOX\n") - - # 2. Test make_sandbox_cib_transaction - print("1. Testing make_sandbox_cib_transaction (Dedicated)...") - transaction_data = { - "account": "GB3R3DRQXBPSC2XSFLPDRVCAVRCVJXAPJGBPMJ45JBRJC5QJPM7QTUSO", - "amount": 120.25, - "full_name": "Python Sandbox Tester", - "phone": "+213000000000", - "email": "python_test@sofizpay.com", - "memo": "Python Sandbox Test" - } - - result = await client.make_sandbox_cib_transaction(transaction_data) - print(f"Result: {json.dumps(result, indent=2)}\n") - - # 3. Test check_sandbox_cib_status - if result.get('success') and 'data' in result and isinstance(result['data'], dict) and result['data'].get('cib_transaction_id'): - cib_transaction_id = result['data']['cib_transaction_id'] - print(f"2. Testing check_sandbox_cib_status for ID: {cib_transaction_id}...") - status = await client.check_sandbox_cib_status(cib_transaction_id) - print(f"Status Result: {json.dumps(status, indent=2)}\n") - else: - print("2. Skipping check_cib_status (no order number received).\n") - - print("--- Sandbox Test Completed ---") - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/test_sdk.py b/test_sdk.py deleted file mode 100644 index 9112a00..0000000 --- a/test_sdk.py +++ /dev/null @@ -1,135 +0,0 @@ -import asyncio -import json -from sofizpay.client import SofizPayClient - -async def test_all(): - # Test Credentials - pub = "GB3R3DRQXBPSC2XSFLPDRVCAVRCVJXAPJGBPMJ45JBRJC5QJPM7QTUSO" - enc_sk = "SCILSE4IMSKSZ7PPDP26CXOYFXWLUER47X5ROMYE6XLWSCZX2UPFKBCO" - tx_hash = "0000000000000000000000000000000000000000000000000000000000000000" - - client = SofizPayClient() - print(f"--- Starting SofizPay Python SDK Test (v{client.VERSION}) ---") - - try: - # 1. Core: Fetch Balance - balance = await client.get_balance(pub) - print(f"1. Get Balance: {balance}") - - # 2. Core: Transaction History - transactions_result = await client.get_transactions(pub) - print(f"2. Get Transactions: {len(transactions_result.get('transactions', []))} found") - - # 3. Core: Public Key Discovery - # get_public_key_from_secret is sync - try: - pk = client.get_public_key_from_secret(pub) - print(f"3. Get Public Key (Validation): Success") - except: - print(f"3. Get Public Key (Validation): Expected fail (mock key)") - - # 4. Core: Search Transactions by Memo - search_result = await client.search_transactions_by_memo(pub, memo="test") - print(f"4. Search Transactions by Memo: {len(search_result.get('transactions', []))} found") - - # 5. Core: Transaction by Hash - tx_details = await client.get_transaction_by_hash(tx_hash) - print(f"5. Get Transaction by Hash: Found={tx_details.get('found')}") - - # 6. CIB: Create Transaction - cib_create = await client.make_cib_transaction({ - "account": pub, - "amount": 100.0, - "full_name": "Python SDK Tester", - "phone": "0661000000", - "email": "test@sofizpay.com", - "memo": "Test CIB Pay" - }) - print(f"6. CIB Create: Success={cib_create.get('success')}") - - # 7. CIB: Check Status - if cib_create.get('success') and 'data' in cib_create and isinstance(cib_create['data'], dict): - order_no = cib_create['data'].get('order_number') - if order_no: - status = await client.check_cib_status(order_no) - print(f"7. CIB Status: Success={status.get('success')}") - else: - print("7. CIB Status: Skipped (no order number)") - - # 8. Services: Get Products - products = await client.get_products(enc_sk) - print(f"8. Get Products: Success={products.get('success')}") - - # 9. Services: Operation History - history = await client.get_operation_history(enc_sk, limit=10) - print(f"9. Operation History: Success={history.get('success')}") - - # 10. Services: Operation Details - details = await client.get_operation_details("OP_12345", enc_sk) - print(f"10. Operation Details: Success={details.get('success')}") - - # 11. Mission: Phone Recharge - recharge = await client.recharge_phone({ - "encrypted_sk": enc_sk, - "phone": "0661000000", - "operator": "mobilis", - "amount": 100, - "offer": "pix" - }) - print(f"11. Recharge Phone: Success={recharge.get('success')}") - - # 12. Mission: Internet Recharge - internet = await client.recharge_internet({ - "encrypted_sk": enc_sk, - "phone": "0661000000", - "operator": "idoom", - "amount": 2000, - "offer": "adsl" - }) - print(f"12. Recharge Internet: Success={internet.get('success')}") - - # 13. Mission: Game Recharge - game = await client.recharge_game({ - "encrypted_sk": enc_sk, - "operator": "freefire", - "playerId": "123456789", - "amount": 100, - "offer": "diamonds" - }) - print(f"13. Recharge Game: Success={game.get('success')}") - - # 14. Mission: Pay Bill - bill = await client.pay_bill({ - "encrypted_sk": enc_sk, - "operator": "sonelgaz", - "bill_id": "BILL_999", - "amount": 5500 - }) - print(f"14. Pay Bill: Success={bill.get('success')}") - - # 15. Utility: Signature Verification - is_valid = client.verify_sofizpay_signature({ - "message": "test_message", - "signature_url_safe": "jHrONYl2NuBhjAYTgRq3xwRuW2ZYZIQlx1VWgiObu5FrSnY78pQ" - }) - print(f"15. Signature Verification: {is_valid}") - - # 16. Stream: Monitoring - print("16. Stream - Setting up...") - def handle_tx(tx): - print(f"STREAM EVENT: {tx.get('hash')}") - - stream_id = await client.setup_transaction_stream(pub, handle_tx, from_now=True, check_interval=15) - print(f"Stream started with ID: {stream_id}") - - # Stop stream - stopped = client.stop_transaction_stream(stream_id) - print(f"Stream stopped: {stopped}") - - except Exception as e: - print(f"Critical Test Failure: {str(e)}") - - print("--- SDK Test Completed ---") - -if __name__ == "__main__": - asyncio.run(test_all()) diff --git a/test_transactions.py b/test_transactions.py deleted file mode 100644 index 33e5969..0000000 --- a/test_transactions.py +++ /dev/null @@ -1,34 +0,0 @@ -import asyncio -from sofizpay.transactions import TransactionManager -import time - -async def main(): - # Use the same public key used in JS tests - public_key = "GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ" - - print(f"--- SofizPay Python SDK: Testing get_transactions ---") - print(f"Public Key: {public_key}") - print("-" * 50) - - manager = TransactionManager() - - start_time = time.time() - try: - # Fetch all transactions - transactions = await manager.get_transactions(public_key) - duration = time.time() - start_time - - print(f"โœ… Success! Fetched {len(transactions)} DZT transactions.") - print(f"โฑ๏ธ Duration: {duration:.2f} seconds") - print("-" * 50) - - if transactions: - # Show all transactions with full response - for i, tx in enumerate(transactions): - print(f"[{i}] {tx}") - - except Exception as e: - print(f"โŒ Failed: {e}") - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..03c3400 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for SofizPay SDK""" diff --git a/tests/test_sdk.py b/tests/test_sdk.py new file mode 100644 index 0000000..9af9c41 --- /dev/null +++ b/tests/test_sdk.py @@ -0,0 +1,167 @@ +import os +import sys +from unittest.mock import patch, MagicMock + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +import pytest +from sofizpay import SofizPayClient, ValidationError + + +def test_client_init(): + client_prod = SofizPayClient() + assert client_prod.is_sandbox is False + assert client_prod.version == "1.2.0" + + client_sandbox = SofizPayClient(is_sandbox=True) + assert client_sandbox.is_sandbox is True + + +@pytest.mark.asyncio +async def test_make_cib_validation(): + client = SofizPayClient() + + # Missing required account + with pytest.raises(ValidationError): + await client.make_cib_transaction({ + 'amount': 100, + 'full_name': 'Tester', + 'phone': '0555000000', + 'email': 'test@example.com' + }) + + # Missing / invalid amount + with pytest.raises(ValidationError): + await client.make_cib_transaction({ + 'account': 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', + 'amount': -10, + 'full_name': 'Tester', + 'phone': '0555000000', + 'email': 'test@example.com' + }) + + +@patch('requests.get') +@pytest.mark.asyncio +async def test_make_cib_sandbox_url_generation(mock_get): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {'content-type': 'application/json'} + mock_response.json.return_value = { + 'status': 'success', + 'payment_url': 'https://sofizpay.com/sandbox/payment/?mdOrder=123', + 'cib_transaction_id': '999888777' + } + mock_get.return_value = mock_response + + client = SofizPayClient(is_sandbox=True) + + res = await client.make_cib_transaction({ + 'account': 'GDNS27ISCGOIJFXC6CM4O5SVHVJPSWR42QEBWUFF24N5VVHGW73ZSJNQ', + 'amount': 200, + 'full_name': 'Ali Tester', + 'phone': '0661000000', + 'email': 'ali@example.com', + 'webhook_url': 'https://mysite.com/webhook', + 'invoice_id': 'INV-123', + 'language': 'fr', + 'redirect': 'yes', + 'keep_return_url': 'True' + }) + + assert res['is_sandbox'] is True + assert 'url' in res + assert 'https://sofizpay.com/sandbox/make-cib-transaction/' in res['url'] + assert 'webhook_url' in res['url'] + assert 'redirect=yes' in res['url'] + assert res['payment_url'] == 'https://sofizpay.com/sandbox/payment/?mdOrder=123' + + +@patch('requests.get') +@pytest.mark.asyncio +async def test_check_cib_status_sandbox(mock_get): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {'content-type': 'application/json'} + mock_response.json.return_value = { + 'errorCode': 0, + 'orderStatus': 2, + 'status': 'success', + 'order_number': 'dummy_order_id', + 'Amount': '200' + } + mock_get.return_value = mock_response + + client = SofizPayClient() + check = await client.check_sandbox_cib_status('dummy_order_id') + assert check['is_sandbox'] is True + assert check['order_number'] == 'dummy_order_id' + assert check['status'] == 'paid' + assert check['success'] is True + + +def test_signature_verification(): + # Negative test with invalid signature + is_valid = SofizPayClient.verify_signature({ + 'message': 'Test Order Payload', + 'signature_url_safe': 'bad_sig_base64' + }) + assert is_valid is False + + # Missing parameters + assert SofizPayClient.verify_signature({'message': ''}) is False + assert SofizPayClient.verify_signature({'signature_url_safe': ''}) is False + + +@pytest.mark.asyncio +async def test_bill_payment_validations(): + client = SofizPayClient() + + # ADE requires bill + with pytest.raises(ValidationError): + await client.pay_ade_bill({'encrypted_sk': 'SXXX', 'amount': 500}) + + # Sonelgaz requires bill, customerId, ebb + with pytest.raises(ValidationError): + await client.pay_sonelgaz_bill({ + 'encrypted_sk': 'SXXX', + 'amount': 500, + 'bill': '12345' + }) + + +@pytest.mark.asyncio +async def test_recharge_validations(): + client = SofizPayClient() + + # Phone recharge requires phone + with pytest.raises(ValidationError): + await client.recharge_phone({ + 'encrypted_sk': 'SXXX', + 'operator': 'mobilis', + 'amount': 100 + }) + + # Game recharge requires playerId and offer + with pytest.raises(ValidationError): + await client.recharge_game({ + 'encrypted_sk': 'SXXX', + 'operator': 'pubg', + 'amount': 1200 + }) + + +if __name__ == '__main__': + import asyncio + + async def run_all(): + test_client_init() + await test_make_cib_validation() + await test_make_cib_sandbox_url_generation() + await test_check_cib_status_sandbox() + test_signature_verification() + await test_bill_payment_validations() + await test_recharge_validations() + print("All 7 tests passed successfully!") + + asyncio.run(run_all())