Commit 31b46d12 by Owo Sugiana

Kali pertama

0 parents
build
*egg-info
Laporan Pajak Kota Cirebon
==========================
Ini adalah web client untuk mendapatkan data transaksi PBB, BPHTB, dan PAD.
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
from configparser import RawConfigParser
import logging
import logging.config
def setup_logging(conf_file):
conf = RawConfigParser()
conf.read(conf_file)
d = {
'version': 1,
'formatters': {},
'handlers': {},
'loggers': {},
}
for section in conf.sections():
if section.find('formatter_') == 0:
name = section.split('formatter_')[1]
data = {'format': conf.get(section, 'format')}
d['formatters'][name] = data
elif section.find('handler_') == 0:
name = section.split('handler_')[1]
data = {'formatter': conf.get(section, 'formatter')}
data['class'] = 'logging.' + conf.get(section, 'class')
if conf.has_option(section, 'stream'):
data['stream'] = 'ext://' + conf.get(section, 'stream')
else:
data['filename'] = conf.get(section, 'filename')
d['handlers'][name] = data
elif section.find('logger_') == 0:
name = section.split('logger_')[1]
if name == 'root':
name = ''
data = {'level': conf.get(section, 'level')}
value = conf.get(section, 'handlers')
data['handlers'] = [x.strip() for x in value.split(',')]
d['loggers'][name] = data
logging.config.dictConfig(d)
import sys
from datetime import (
datetime,
date,
)
from argparse import ArgumentParser
from configparser import ConfigParser
from pprint import pprint
from cirebon_kota_payment_report import WebClient
from .logger import setup_logging
def main(argv=sys.argv[1:]):
pajak = ['pbb', 'bphtb', 'pad']
tgl = date.today().strftime('%d-%m-%Y')
help_tgl = f'default {tgl}'
pars = ArgumentParser()
pars.add_argument('conf')
pars.add_argument('pajak', choices=pajak)
pars.add_argument('--tgl', default=tgl, help=help_tgl)
option = pars.parse_args(argv)
setup_logging(option.conf)
t = option.tgl.split(',')
tgl_awal = t[0]
if t[1:]:
tgl_akhir = t[1]
else:
tgl_akhir = tgl_awal
tgl_awal = datetime.strptime(tgl_awal, '%d-%m-%Y')
tgl_akhir = datetime.strptime(tgl_akhir, '%d-%m-%Y')
tgl_awal = tgl_awal.date()
tgl_akhir = tgl_akhir.date()
conf = ConfigParser()
conf.read(option.conf)
cf = dict(conf.items('main'))
client = WebClient()
r = client.login(cf['username'], cf['password'])
r = client.send(option.pajak, tgl_awal, tgl_akhir)
pprint(r)
count = len(r)
print(f'Ada {count} transaksi.')
[main]
username = your-user
password = your-pass
[formatter_generic]
format = %(asctime)s %(levelname)s %(name)s %(message)s
[handler_console]
class = StreamHandler
stream = sys.stdout
formatter = generic
[logger_root]
handlers = console
level = DEBUG
[build-system]
requires = ['setuptools >= 64']
[project]
name = 'cirebon-kota-web-service'
version = '0.1'
dependencies = [
'requests',
]
requires-python = '>= 3.9'
authors = [
{name='Owo Sugiana', email='sugiana@gmail.com'},
]
description = 'Web service client Kota Cirebon'
readme = 'README.rst'
license = {text='PostgreSQL License'}
classifiers = [
'Programming Language :: Python :: 3',
'Operating System :: OS Independent',
]
[project.scripts]
cirebon_kota_payment_report = 'cirebon_kota_payment_report.scripts.payment_report:main'
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!