import json import requests OAUTH_CLIENT_ID = 'qHhNuALPINvJ8UHAcBGsX1uVRWu3AmRMwRL5kVbi' # OAUTH_CLIENT_SECRET = 'pbkdf2_sha256$390000$A8ru6iwhcjU1wEgPL6n6b8$GOhCcrfqw4Xkw6USpuEL6esjmNgqJe1A8q1Ec2dLxhw=' OAUTH_CLIENT_SECRET = 'pbkdf2_sha256$390000$A8ru6iwhcjU1wEgPL6n6b8$GOhCcrfqw4Xkw6USpuEL6esjmNgqJe1A8q1Ec2dLxhw=' API_URI = 'http://127.0.0.1:8000' # API_URI = 'https://accounts.gooyal.com' # NOTIFICATION_URI = 'https://notifications.gooyal.com' # NOTIFICATION_URI = 'http://127.0.0.1:8001' class ApiClient(): def __init__(self, phone_number, auth_data=None): self.auth_data = {} if auth_data: self.auth_data = auth_data self.phone_number = phone_number def otp_login(self, password): phone_number = self.phone_number data = { "grant_type": "password", "username": phone_number, "password": password, # "scope": 'introspection', "scope": 'introspection superapp.applications:list accounts.account:request_ott accounts.profile:list accounts.profile:retrieve accounts.account:retrieve accounts.account:update accounts.profile:inquiry accounts.account:change_password wallet.transaction:list wallet.invoice:create wallet.transaction:retrieve wallet.invoice:pay wallet.invoice:receipt wallet.deposit:submit wallet.deposit:verify wallet.withdraw:submit wallet.withdraw:verify ', "auth_fields": 'phone_number:otp' } auth = (OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET) response = requests.post(f'{API_URI}/oauth2/token/', data=data, auth=auth) auth_data = response.json() if 'access_token' in auth_data: self.auth_data = auth_data return auth_data def ott_login(self, token): phone_number = self.phone_number data = { "grant_type": "password", "username": phone_number, "password": token, "scope": 'introspection superapp.applications:list accounts.account:request_ott accounts.profile:list accounts.profile:retrieve accounts.account:retrieve accounts.account:update accounts.profile:inquiry accounts.account:change_password wallet.transaction:list wallet.invoice:create wallet.transaction:retrieve wallet.invoice:pay wallet.invoice:receipt wallet.deposit:submit wallet.deposit:verify wallet.withdraw:submit wallet.withdraw:verify ', "auth_fields": 'phone_number:ott' } auth = (OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET) response = requests.post(f'{API_URI}/oauth2/token/', data=data, auth=auth) auth_data = response.json() if 'access_token' in auth_data: self.auth_data = auth_data return auth_data def _request(self, path, data=None, files=None, method='get', with_auth=True, encode=True): if files: header = {} else: header = { "Content-Type": "application/json", "charset": "utf-8" } if with_auth: access_token = self.auth_data.get('access_token', '') header.setdefault("Authorization", f"Bearer {access_token}", ) if data and encode: data = json.dumps(data) response = requests.request(method, f'{API_URI}/{path}', headers=header, data=data, files=files) try: return response.json() except: return response.content.decode() def request_otp(self): phone_number = self.phone_number path = 'users/api/request_otp/' method = 'post' data = { 'phone_number': phone_number } result = self._request(path=path, data=data, method=method, with_auth=False) return result def get_account(self): path = 'users/api/account/' result = self._request(path=path) return result def update_account(self, **kwargs): path = 'users/api/account/' data = {} fields = ( 'username', "first_name", "last_name", 'name', 'iban', 'iban_card_number', 'iban_account_number', ) for key, value in kwargs.items(): if key in fields: data.setdefault(key, value) avatar = kwargs.get('avatar') if avatar: files = {'avatar': avatar} encode = False else: files = None encode = True result = self._request(path=path, data=data, method='put', files=files, encode=encode) return result def change_password(self, old_password, new_password, old_password_gateway='otp'): path = 'users/api/change_password/' data = {'old_password': old_password, 'new_password': new_password, 'old_password_gateway': old_password_gateway } result = self._request(path=path, data=data, method='put') return result def get_user_profile(self, code): path = f'users/api/users/{code}/' result = self._request(path=path) return result def get_application_list(self, page=None): path = 'api/applications/' if page: page = int(page) path = f'{path}?page={page}' return self._request(path=path) def get_transactions(self, page=None): path = 'wallet/api/transactions/' if page: page = int(page) path = f'{path}?page={page}' return self._request(path=path) def get_transaction(self, code): path = f'transactions/{code}' return self._request(path=path) # create user invoice # def create_invoice(self, amount, delay, payer=None): # data = { # 'delay': delay, # 'amount': amount # } # if payer: # data.setdefault('payer', {'code': payer}) # response = self._request('transactions/', method='post', data=data) # return response and 'code' in response, response # create application invoice def get_application_token(self): response = self._request('users/api/request_ott/', method='get') return response and 'code' in response, response def create_invoice(self, amount, delay=0): data = { 'delay': delay, 'amount': amount } response = self._request('wallet/api/transactions/', method='post', data=data) return response and 'code' in response, response def pay_invoice(self, code): path = f'wallet/api/transactions/{code}/pay' return self._request(path=path) def receipt_transaction(self, code): path = f'transactions/{code}/receipt' return self._request(path=path) def get_introspection(self): path = 'oauth2/introspect/' access_token = self.auth_data.get('access_token', '') data = { 'token': access_token, } return self._request(path=path, data=data) def notify(self): data = { 'phone_number': "+989106853582", 'body': 'this is a test' } # data = json.dumps(data) header = { # "Content-Type": "application/json", "charset": "utf-8" } access_token = self.auth_data.get('access_token', '') # access_token = 'Px7WLUKoynua6qJq5IkIG8Fn5dPLXq' header.setdefault("Authorization", f"Bearer {access_token}", ) response = requests.request('post', f'{NOTIFICATION_URI}/notifications/', headers=header, json=data) try: return response.json() except: return response.text client = ApiClient('+989106853582') client.request_otp() client.otp_login('12345') print(client.get_application_list()) # print(client.get_application_token()) # print(client.ott_login('abcdef12345')) # print(client.create_invoice(2000, 10)) # print(client.pay_invoice('7967a76a-5de1-48b2-92ac-1b0f39f7223b'))