pbb.py
5.67 KB
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import requests
from opensipkd.string import exception_message
from pyramid_linkaja.responses import get_template_response
from pyramid_linkaja.exceptions import (
InvoiceIdError,
AlreadyPaidError,
BaseError,
LinkError,
)
from linkaja.models import Bank
from bpd_sultra.structure import (
RC_OK,
RC_NOT_FOUND,
RC_ALREADY_PAID,
RC_LINK_DOWN,
RC_INVALID_TOKEN,
RC_INVALID_REQUEST,
)
from bpd_sultra.pbb import WebClient as BaseWebClient
NUM_FIELDS = [
('JML_POKOK', 'Pok'),
('JML_DENDA', 'Den')]
STR_FIELDS = ['ALAMAT_WP', 'KELURAHAN_WP']
def profile2name(resp):
invoice_id = 'NOP {} Tahun {}'.format(resp['NOP'], resp['THN_PAJAK'])
msg = [invoice_id, resp['NAMA_WP']]
for field, label in NUM_FIELDS:
s = resp[field]
if s:
msg.append(f'{label} {s}')
for field in STR_FIELDS:
s = resp[field]
if s:
if field == 'ALAMAT_OP':
s = s.replace(':', ' ')
msg.append(s)
return ', '.join(msg)
class WebClient(BaseWebClient):
def __init__(self, parent):
self.parent = parent
base_url = self.parent.qconf.get_value('bpd sultra base url')
username = self.parent.qconf.get_value('bpd sultra username')
password = self.parent.qconf.get_value('bpd sultra password')
super().__init__(base_url, username, password)
def before_send(self, url, data):
self.bank_log = None
if url.find('inquiry') > -1 or url.find('payment') > -1:
self.bank_log = Bank(
rpc_id=self.parent.web_log.id, conf_name='bpd sultra',
url=url, request=data)
self.parent.request.dbsession.add(self.bank_log)
self.parent.request.tm.commit()
self.parent.request.tm.begin()
def after_send(self, data):
if not self.bank_log:
return
self.bank_log.response = data
self.parent.request.dbsession.add(self.bank_log)
self.parent.request.tm.commit()
self.parent.request.tm.begin()
def send(self, url, headers, data): # Override
self.before_send(url, data)
r = super().send(url, headers, data)
self.after_send(r)
return r
class PrefixHandler:
def __init__(self, parent):
self.parent = parent
def auth(self):
try:
self.remote.auth()
except requests.exceptions.ConnectionError as e:
self.parent.log_error(e)
raise LinkError()
except Exception as e:
msg = exception_message()
self.parent.log_error(msg)
raise LinkError()
def __inquiry(self):
invoice_id = self.parent.q_web_log.get_invoice_id()
try:
return self.remote.inquiry(invoice_id)
except requests.exceptions.ConnectionError as e:
self.parent.log_error(e)
raise LinkError()
except Exception:
raise LinkError()
def inquiry(self):
resp = get_template_response(self.parent.q_web_log.data)
self.remote = WebClient(self.parent)
try:
self.auth()
except BaseError as err:
resp.from_err(err)
return resp
try:
r = self.__inquiry()
except BaseError as err:
resp.from_err(err)
return resp
if r['RC'] == RC_OK:
d = self.rc_ok(r)
for key in d:
resp[key] = d[key]
else:
err = self.rc_error(r['RC'])
resp.from_err(err)
return resp
def __payment(self):
q = self.parent.request.dbsession.query(Bank).filter_by(
rpc_id=self.parent.web_log.inquiry_id)
bank_log = q.first()
data = dict(bank_log.response) # dari inquiry response
del data['RC']
del data['RM']
del data['TRACENUM']
data['NO_ARSIP'] = self.parent.q_web_log.data['trx_id'] # NTB
try:
return self.remote.payment(data)
except requests.exceptions.ConnectionError as e:
self.parent.log_error(e)
raise LinkError()
except Exception:
raise LinkError()
def payment(self):
resp = get_template_response(self.parent.q_web_log.data)
self.remote = WebClient(self.parent)
try:
self.auth()
except BaseError as err:
resp.from_err(err)
return resp
try:
r = self.__payment()
except BaseError as err:
resp.from_err(err)
return resp
if r['RC'] == RC_OK:
d = self.rc_ok(r)
for key in d:
resp[key] = d[key]
else:
err = self.rc_error(r['RC'])
resp.from_err(err)
return resp
def rc_error(self, rc):
if rc == RC_NOT_FOUND:
return InvoiceIdError()
if rc == RC_ALREADY_PAID:
return AlreadyPaidError()
if rc in (RC_LINK_DOWN, RC_INVALID_TOKEN, RC_INVALID_REQUEST):
return LinkError()
return BaseError()
def rc_ok(self, resp):
web_log = self.parent.web_log
keterangan = profile2name(resp)
d = {'Response Code': '00'}
if self.parent.is_inquiry():
total = int(resp['JML_POKOK']) + int(resp['JML_DENDA'])
d['Bill Amount'] = str(total)
d['Bill Ref'] = str(web_log.id)
d['Biller Name'] = web_log.biller_name = keterangan
d['Notification Message'] = self.parent.notification_message
else: # payment
d['Transaction ID'] = web_log.ntp = resp['NO_NTTP'] # NTP
d['Notification Message'] = keterangan
return d