__init__.py
3.41 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
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)
CLEANER = dict(pbb=clean_pbb, bphtb=clean_bphtb, pad=clean_pad)
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 send(
self, pajak: str, tgl_awal=date.today(),
tgl_akhir=date.today()) -> dict:
log = getLogger('send')
url = self.base_url + '/laporan/' + pajak
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, page=1, per_page=25)
r_all = []
while True:
log.info(f'Request GET {url} {p}')
r = requests.get(url, headers=self.headers, params=p, timeout=30)
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 = CLEANER[pajak]
for d in r_all:
func_cleaner(d)
result.append(d)
return result