__init__.py
4.06 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
from datetime import date
from logging import getLogger
import requests
MONTH_LIST = [
(
'', 'JAN', 'FEB', 'MAR', 'APR', 'MAY', 'JUN', 'JUL', 'AUG', 'SEP',
'OCT', 'NOV', 'DEC'),
(
'', 'JAN', 'FEB', 'MAR', 'APR', 'MEI', 'JUN', 'JUL', 'AGT', 'SEP',
'OKT', 'NOV', 'DES'),
(
'', '01', '02', '03', '04', '05', '06', '07', '08', '09', '10', '11',
'12')]
def to_date(s: str):
t = s.split('-')
day_str = t[0]
month_str = t[1]
year_str = t[2]
if len(day_str) > 2:
year_str = day_str
day_str = t[2]
else:
year_str = '20' + year_str
day = int(day_str)
year = int(year_str)
month = None
for month_list in MONTH_LIST:
try:
month = month_list.index(month_str)
except ValueError:
pass
if not month:
raise Exception(f'Bulan {month_str} tidak dipahami, perbaiki script.')
print(s)
print(year, month, day)
return date(year, month, day)
def clean_values(d: dict, keys: list, func):
for key in keys:
d[key] = func(d[key])
def clean_pbb(d: dict):
clean_values(
d, ['DENDA_SPPT', 'DISCOUNT', 'JML_SPPT_YG_DIBAYAR',
'PBB_YG_HARUS_DIBAYAR_SPPT'], int)
clean_values(d, ['TGL_PEMBAYARAN_SPPT'], to_date)
def clean_bphtb(d: dict):
clean_values(
d, ['ID', 'BNG_LUAS', 'BUMI_LUAS', 'NILAI_BPHTB', 'NPOP'], int)
clean_values(d, ['TGL'], to_date)
def clean_pad(d: dict):
clean_values(
d, ['id', 'denda', 'jml_bayar', 'opsen_denda', 'opsen_pokok', 'pokok'],
int)
clean_values(d, ['tgl'], to_date)
def clean_retribusi(d: dict):
clean_values(d, ['id', 'denda', 'pokok', 'total_tagihan'], int)
clean_values(d, ['tanggal_jatuh_tempo'], to_date)
CLEANER = dict(
pbb=clean_pbb, bphtb=clean_bphtb, pad=clean_pad, retribusi=clean_retribusi)
class BaseError(Exception):
pass
class WebClient:
def __init__(self, base_url='https://ws.cirebonkota.go.id'):
self.base_url = base_url
self.auth_url = base_url + '/auth'
def login(self, username: str, password: str) -> dict:
log = getLogger('login')
auth = (username, password)
auth_log = (username, '-')
log.debug(f'Request GET {self.auth_url} HTTPBasicAuth {auth_log}')
r = requests.get(self.auth_url, auth=auth)
r = r.json()
token = r['token']
r_log = dict(r)
r_log['token'] = r_log['token'][:10] + ' ...'
log.debug(f'Response {r_log}')
self.headers = {"Authorization": f"Bearer {token}"}
return r
def get_transaction_url(self, pajak: str) -> str:
return self.base_url + '/laporan/' + pajak
def send(
self, pajak: str, tgl_awal=date.today(),
tgl_akhir=date.today()) -> dict:
t_awal = tgl_awal.strftime('%Y-%m-%d')
t_akhir = tgl_akhir.strftime('%Y-%m-%d')
p = dict(tanggal_dari=t_awal, tanggal_sampai=t_akhir)
return self.loop(pajak, p)
def get_cleaner_func(self, pajak: str):
return CLEANER[pajak]
def loop(self, pajak: str, params: dict) -> list:
log = getLogger('send')
url = self.get_transaction_url(pajak)
p = dict(params)
p.update(dict(page=1, per_page=25))
r_all = []
while True:
log.info(f'Request GET {url} {params}')
r = requests.get(
url, headers=self.headers, params=p, timeout=30)
if r.status_code != 200:
raise BaseError(f'HTTP Error {r.status_code}')
r = r.json()
if not r['data']:
log.info(f'Response {r}')
break
r_log = dict(r)
r_log['data'] = f'{r_log["data"][:1]} ...'
log.info(f'Response {r_log}')
r_all.extend(r['data'])
if len(r['data']) < p['per_page']:
break
p['page'] += 1
result = []
func_cleaner = self.get_cleaner_func(pajak)
for d in r_all:
func_cleaner(d)
result.append(d)
return result