1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# Winpay Create Virtual Account
# Ikuti petunjuk di https://git.opensipkd.com/snippets/157
# pip install requests
import sys
import json
from base64 import b64encode
from datetime import (
datetime,
timedelta,
)
from argparse import ArgumentParser
from pprint import pprint
from logging import getLogger
import logging.config
import requests
from opensipkd.waktu import create_now
from signature import (
generator,
time_to_str,
)
class HttpErr(Exception):
pass
def create_va(
url: str, private_key: bytes, partner_id: str, customer_no: str,
va_name: str, amount: int, channel: str, trx_id: str, expired_days=1):
log = getLogger('create_va()')
timestamp = create_now()
timestamp_str = time_to_str(timestamp)
expired_timestamp = timestamp + timedelta(expired_days)
expired_str = time_to_str(expired_timestamp)
data = dict(
customerNo=customer_no,
virtualAccountName=va_name,
trxId=trx_id,
totalAmount=dict(value=amount, currency='IDR'),
virtualAccountTrxType='c',
expiredDate=expired_str,
additionalInfo=dict(channel=channel))
signature = generator(private_key, data, timestamp)
signature_b64 = b64encode(signature).decode('utf-8')
headers = {
'X-TIMESTAMP': timestamp_str,
'X-SIGNATURE': signature_b64,
'X-PARTNER-ID': partner_id,
'X-EXTERNAL-ID': trx_id,
'CHANNEL-ID': channel}
log.info(f'Request: Headers {headers}, Data {data}')
r = requests.post(url, headers=headers, json=data)
if r.status_code != 200:
log.error(f'HTTP {r.status_code}, {[r.text]}')
raise HttpErr(r.status_code, r.text)
d = r.json()
log.info(f'Response: {d}')
return d
def main(argv=sys.argv[1:]):
url = 'https://sandbox-api.bmstaging.id/snap/v1.0/transfer-va/create-va'
help_url = f'default {url}'
expired_days = 1
help_expired = f'default {expired_days}'
channels = [
'BRI', 'BNI', 'MANDIRI', 'PERMATA', 'BSI', 'MUAMALAT', 'BCA', 'CIMB',
'SINARMAS', 'BNC', 'INDOMARET', 'ALFAMART']
pars = ArgumentParser()
pars.add_argument('--private-file', required=True)
pars.add_argument('--partner-id', required=True)
pars.add_argument('--customer-no', required=True)
pars.add_argument('--va-name', required=True)
pars.add_argument('--amount', type=int, required=True)
pars.add_argument('--channel', choices=channels)
pars.add_argument('--url', default=url, help=help_url)
pars.add_argument('--expired-days', type=int, default=expired_days, help=help_expired)
pars.add_argument('--trx-id')
option = pars.parse_args(sys.argv[1:])
logging.config.dictConfig({
'version': 1,
'formatters': {
'generic': {
'format': '%(asctime)s %(levelname)s %(name)s %(message)s',
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'stream': 'ext://sys.stdout',
'formatter': 'generic',
},
},
'loggers': {
'': {
'handlers': ['console'],
'level': 'DEBUG',
},
},
})
with open(option.private_file, 'rb') as f:
private_key = f.read()
if option.trx_id:
trx_id = option.trx_id
else:
trx_id = datetime.now().strftime('%H%M%S')
try:
r = create_va(
option.url, private_key, option.partner_id, option.customer_no,
option.va_name, option.amount, option.channel, trx_id,
option.expired_days)
pprint(r)
except HttpErr as e:
print('ERROR', e)
if __name__ == '__main__':
main()