bin/pbb_inquiry bisa rekonsiliasi menggunakan CSV

CHANGES.txt

+0.2.2 2023-05-08
+----------------
+- Tambah Kota Banjar bila diarahkan ke Oracle. Kalau ke Postgres cukup pakai
+  modul default. 
+- bin/pbb_inquiry ada pilihan --payment-file untuk rekonsiliasi menggunakan CSV
+
 0.2.1 2023-04-14
 ----------------
 - Bug fixed total bayar negatif

sismiop/models/banjar.py

+from sqlalchemy import (
+    Column,
+    Float,
+    )
+from sqlalchemy.ext.declarative import declarative_base
+from .propinsi import PropinsiMixin
+from .kelurahan import KelurahanMixin
+from .kecamatan import KecamatanMixin
+from .objek_pajak import ObjekPajakMixin
+from .sppt import SpptMixin
+from .pembayaran_sppt import BasePembayaranSpptMixin
+
+
+Base = declarative_base()
+
+
+class Propinsi(PropinsiMixin, Base):
+    __table_args__ = dict(schema='pbb')
+
+
+class Kelurahan(KelurahanMixin, Base):
+    __table_args__ = dict(schema='pbb')
+
+
+class Kecamatan(KecamatanMixin, Base):
+    __table_args__ = dict(schema='pbb')
+
+
+class ObjekPajak(ObjekPajakMixin, Base):
+    __table_args__ = dict(schema='pbb')
+
+
+class Sppt(SpptMixin, Base):
+    __table_args__ = dict(schema='pbb')
+
+
+class PembayaranSppt(BasePembayaranSpptMixin, Base):
+    __table_args__ = dict(schema='pbb')
+    jml_sppt_yg_dibayar = Column(Float, nullable=False)
+    denda_sppt = Column(Float, nullable=False)

sismiop/scripts/inquiry.py

 import sys
+import csv
 from datetime import datetime
 from configparser import ConfigParser
 from argparse import ArgumentParser
@@ -40,15 +41,19 @@ def date_from_str(s):
     return datetime(int(y), int(m), int(d))
 
 
+help_tgl = 'butuh --payment'
+help_status_bayar = 'Field status_pembayaran_sppt = 1 jika lunas'
+help_payment_file = 'File CSV rekonsiliasi'
+
+
 def get_option(argv):
-    help_tgl = 'butuh --payment'
-    help_status_bayar = 'Field status_pembayaran_sppt = 1 jika lunas'
     pars = ArgumentParser()
     pars.add_argument('conf')
     pars.add_argument('--invoice-id')
     pars.add_argument('--payment', action='store_true')
     pars.add_argument('--tgl-bayar', help=help_tgl)
     pars.add_argument('--reversal', action='store_true')
+    pars.add_argument('--payment-file', help=help_payment_file)
     pars.add_argument(
         '--update-status-bayar', action='store_true', help=help_status_bayar)
     pars.add_argument('--sql-debug', action='store_true')
@@ -167,6 +172,37 @@ def show_reversal(inq, pay):
     show_fields(row, fields)
 
 
+def get_tp(d):
+    cf = dict()
+    for key in d:
+        key_lower = key.lower()
+        if key_lower.find('kd_') < 0:
+            continue
+        cf[key_lower] = d[key]
+    if registry['module_name'] == 'sukabumi_kota':
+        tp = dict(
+                KD_KANWIL_BANK=cf['kd_kanwil_bank'],
+                KD_KPPBB_BANK=cf['kd_kppbb_bank'],
+                KD_BANK_TUNGGAL=cf['kd_bank_tunggal'],
+                KD_BANK_PERSEPSI=cf['kd_bank_persepsi'],
+                KD_TP=cf['kd_tp'])
+    else:
+        tp = dict(kd_tp=cf['kd_tp'])
+        if cf.get('kd_kantor'):
+            tp.update(dict(
+                    kd_kanwil=cf['kd_kanwil'],
+                    kd_kantor=cf['kd_kantor']))
+        if cf.get('kd_kanwil_bank'):
+            tp.update(dict(
+                kd_kanwil_bank=cf['kd_kanwil_bank'],
+                kd_kppbb_bank=cf['kd_kppbb_bank']))
+        if cf.get('kd_bank_tunggal'):
+            tp.update(dict(
+                kd_bank_tunggal=cf['kd_bank_tunggal'],
+                kd_bank_persepsi=cf['kd_bank_persepsi']))
+    return tp
+
+
 ERR_PAYMENT_NOT_FOUND = 'Pembayaran tidak ditemukan, '\
     'tidak ada yang perlu dibatalkan.'
 
@@ -194,11 +230,30 @@ def main(argv=sys.argv):
     sismiop.services.base.DBSession = DBSession = session_factory()
     register(sismiop.services.base.DBSession)
     persen_denda = float(cf['persen_denda'])
+    if option.payment_file:
+        with open(option.payment_file) as f:
+            c = csv.DictReader(f)
+            for r in c:
+                invoice_id = r['NOP'] + r['TAHUN']
+                tgl_bayar = datetime.strptime(r['TANGGAL'], '%Y%m%d')
+                inq = Inquiry(invoice_id, persen_denda, tgl_bayar=tgl_bayar)
+                ket = None
+                if inq.invoice:
+                    if inq.is_paid():
+                        ket = 'memang sudah dibayar'
+                else:
+                    ket = 'tidak ditemukan'
+                if not ket:
+                    tp = get_tp(r)
+                    pay = inq.do_payment(tp, cf['nip_pencatat'])
+                    ket = 'berhasil dibayar'
+                print(f'{invoice_id} {ket}')
+        return
     with transaction.manager:
         inq = Inquiry(
                 invoice_id, persen_denda, tgl_bayar=tgl_bayar, debug=True)
         if not inq.invoice:
-            print('Invoice ID {} tidak ada.'.format(invoice_id))
+            print(f'Invoice ID {invoice_id} tidak ada.')
             if inq.debug_invoice and tgl_bayar:
                 if tgl_bayar.year < int(inq.debug_invoice.thn_pajak_sppt):
                     print('Ada tapi tahun pajak lebih besar dari tahun '
@@ -215,27 +270,7 @@ def main(argv=sys.argv):
             if not inq.is_available():
                 print('Tagihan tidak diperkenankan dibayar melalui jalur ini.')
                 return
-            if cf['module'] == 'sukabumi_kota':
-                tp = dict(
-                        KD_KANWIL_BANK=cf['kd_kanwil_bank'],
-                        KD_KPPBB_BANK=cf['kd_kppbb_bank'],
-                        KD_BANK_TUNGGAL=cf['kd_bank_tunggal'],
-                        KD_BANK_PERSEPSI=cf['kd_bank_persepsi'],
-                        KD_TP=cf['kd_tp'])
-            else:
-                tp = dict(kd_tp=cf['kd_tp'])
-                if cf.get('kd_kantor'):
-                    tp.update(dict(
-                            kd_kanwil=cf['kd_kanwil'],
-                            kd_kantor=cf['kd_kantor']))
-                if cf.get('kd_kanwil_bank'):
-                    tp.update(dict(
-                        kd_kanwil_bank=cf['kd_kanwil_bank'],
-                        kd_kppbb_bank=cf['kd_kppbb_bank']))
-                if cf.get('kd_bank_tunggal'):
-                    tp.update(dict(
-                        kd_bank_tunggal=cf['kd_bank_tunggal'],
-                        kd_bank_persepsi=cf['kd_bank_persepsi']))
+            tp = get_tp(cf)
             pay = inq.do_payment(tp, cf['nip_pencatat'])
             show_payment(inq, pay)
         if option.reversal:

sismiop/services/banjar_oracle.py

+from datetime import date
+from sqlalchemy import func
+from ..models.banjar import (
+    Propinsi,
+    Kelurahan,
+    Kecamatan,
+    ObjekPajak,
+    Sppt,
+    PembayaranSppt,
+    )
+from .default import (
+    Inquiry as BaseInquiry,
+    Reversal,
+    AvailableInvoice as BaseAvailableInvoice,
+    )
+from .base import get_db_session
+
+
+class Inquiry(BaseInquiry):
+    def get_propinsi_model(self):  # Override
+        return Propinsi
+
+    def get_kelurahan_model(self):  # Override
+        return Kelurahan
+
+    def get_kecamatan_model(self):  # Override
+        return Kecamatan
+
+    def get_op_model(self):  # Override
+        return ObjekPajak
+
+    def get_invoice_model(self):  # Override
+        return Sppt
+
+    def get_payment_model(self):  # Override
+        return PembayaranSppt
+
+    def get_payment_sum(self):  # Override
+        Payment = self.get_payment_model()
+        DBSession = get_db_session()
+        return DBSession.query(
+                    func.sum(Payment.jml_sppt_yg_dibayar).
+                    label('jml_sppt_yg_dibayar'),
+                    func.sum(Payment.denda_sppt).
+                    label('denda_sppt'),
+                    func.sum(0).
+                    label('discount'))
+
+    def before_save(self, payment):  # Override
+        payment.denda_sppt = 0  # self.denda - self.discount
+
+
+class AvailableInvoice(BaseAvailableInvoice):
+    def get_invoice_model(self):  # Override
+        return Sppt
+
+    def get_inquiry_class(self):  # Override
+        return Inquiry