pjpatil12/Payment_Reconciliation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimize and add new service of get report and download report.

Browse Source
This commit is contained in:
Pankaj J Patil
2026-04-03 12:10:47 +05:30
parent 0b72adef7d
commit 73cd97f3c8
27 changed files with 1416 additions and 1292 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
services/ReportService.py Normal file
View File

@@ -0,0 +1,177 @@
import config
from model.FolderAndFile import FolderAndFile
from services.Generalservice import GeneralUse
from model.Report import ReportHelper
from decimal import Decimal
def safe_decimal(value):
if value is None:
return Decimal(0)
return Decimal(value)
class ReportService:
def __init__(self, contractor_id=None, pmc_no=None):
self.contractor_id = contractor_id
self.pmc_no = pmc_no
self.contInfo = None
self.hold_types = []
self.invoices = []
self.hold_amounts = []
self.hold_data = {}
self.credit_note_raw = []
self.credit_note_map = {}
self.gst_release_raw = []
self.gst_release_map = {}
self.output_file = None
# ---------------- LOAD DATA ----------------
def load_data(self):
connection = config.get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
try:
# ---------- CONTRACTOR ----------
if self.contractor_id:
self.contInfo = GeneralUse.execute_sp(cursor, 'GetContractorInfo', [self.contractor_id], True)
self.hold_types = GeneralUse.execute_sp(cursor, 'HoldTypesByContractorId', [self.contractor_id])
self.invoices = GeneralUse.execute_sp(cursor, 'GetInvoicesByContractorOrPMCNo', [self.contractor_id, None])
# ---------- PMC ----------
elif self.pmc_no:
self.contInfo = GeneralUse.execute_sp(cursor, 'GetContractorInfoByPmcNo', [self.pmc_no], True)
self.contractor_id = self.contInfo["Contractor_Id"]
self.hold_types = GeneralUse.execute_sp(cursor, 'GetHoldTypesByContractor', [self.contractor_id])
self.invoices = GeneralUse.execute_sp(cursor, 'GetInvoicesByContractorOrPMCNo', [None, self.pmc_no])
# ---------- COMMON ----------
self.hold_amounts = GeneralUse.execute_sp(cursor, 'GetHoldAmountsByContractor', [self.contractor_id])
self.credit_note_raw = GeneralUse.execute_sp(cursor, 'GetCreditNotesByContractor', [self.contractor_id])
self.gst_release_raw = GeneralUse.execute_sp(cursor, 'GetGSTRelease', [self.contractor_id])
self.prepare_maps()
self.total = self.calculate_totals()
finally:
cursor.close()
connection.close()
return self
# ---------------- MAPS ----------------
def prepare_maps(self):
# HOLD MAP
for h in self.hold_amounts:
self.hold_data.setdefault(h['Invoice_Id'], {})[h['hold_type_id']] = h['hold_amount']
# CREDIT MAP
for cn in self.credit_note_raw:
key = str(cn.get('PMC_No') or cn.get('pmc_no')).strip()
self.credit_note_map.setdefault(key, []).append(cn)
# GST MAP (FIXED)
for gr in self.gst_release_raw:
pmc_value = gr.get('PMC_No') or gr.get('pmc_no') or gr.get('PMC_NO')
if not pmc_value:
continue # skip if missing
key = str(pmc_value).strip()
self.gst_release_map.setdefault(key, []).append(gr)
# ---------------- calculate total ----------------
def calculate_totals(self):
total = {
"sum_invo_basic_amt": Decimal(0),
"sum_invo_debit_amt": Decimal(0),
"sum_invo_after_debit_amt": Decimal(0),
"sum_invo_gst_amt": Decimal(0),
"sum_invo_amt": Decimal(0),
"sum_invo_tds_amt": Decimal(0),
"sum_invo_ds_amt": Decimal(0),
"sum_invo_on_commission": Decimal(0),
"sum_invo_hydro_test": Decimal(0),
"sum_invo_hold_amt": Decimal(0),
"sum_invo_gst_sd_amt": Decimal(0),
"sum_invo_final_amt": Decimal(0),
"sum_gst_basic_amt": Decimal(0),
"sum_gst_final_amt": Decimal(0),
"sum_pay_payment_amt": Decimal(0),
"sum_pay_tds_payment_amt": Decimal(0),
"sum_pay_total_amt": Decimal(0)
}
# ---------- INVOICE ----------
for inv in self.invoices:
total["sum_invo_basic_amt"] += safe_decimal(inv.get("Basic_Amount"))
total["sum_invo_debit_amt"] += safe_decimal(inv.get("Debit_Amount"))
total["sum_invo_after_debit_amt"] += safe_decimal(inv.get("After_Debit_Amount"))
total["sum_invo_gst_amt"] += safe_decimal(inv.get("GST_Amount"))
total["sum_invo_amt"] += safe_decimal(inv.get("Amount"))
total["sum_invo_tds_amt"] += safe_decimal(inv.get("TDS_Amount"))
total["sum_invo_ds_amt"] += safe_decimal(inv.get("SD_Amount"))
total["sum_invo_on_commission"] += safe_decimal(inv.get("On_Commission"))
total["sum_invo_hydro_test"] += safe_decimal(inv.get("Hydro_Testing"))
total["sum_invo_gst_sd_amt"] += safe_decimal(inv.get("GST_SD_Amount"))
total["sum_invo_final_amt"] += safe_decimal(inv.get("Final_Amount"))
total["sum_invo_hold_amt"] += safe_decimal(inv.get("hold_amount"))
# ---------- GST ----------
for gst in self.gst_release_raw:
total["sum_gst_basic_amt"] += safe_decimal(gst.get("basic_amount"))
total["sum_gst_final_amt"] += safe_decimal(gst.get("final_amount"))
# ---------- PAYMENTS ----------
if hasattr(self, "payments"):
for pay in self.payments:
total["sum_pay_payment_amt"] += safe_decimal(pay.get("Payment_Amount"))
total["sum_pay_tds_payment_amt"] += safe_decimal(pay.get("TDS_Payment_Amount"))
total["sum_pay_total_amt"] += safe_decimal(pay.get("Total_amount"))
return total
# ---------------- WEB DATA ----------------
def get_web_data(self):
return {
"contInfo": self.contInfo,
"invoices": self.invoices,
"hold_types": self.hold_types,
"credit_note": self.credit_note_raw,
"gst_rel": self.gst_release_raw,
"total": self.total
}
# ---------------- DOWNLOAD ----------------
def download_excel(self):
if not self.contInfo:
return None, "No data found"
filename = f"Report_{self.contractor_id or self.pmc_no}.xlsx"
self.output_file = FolderAndFile.get_download_path(filename=filename)
ReportHelper.generate_excel(
self.contractor_id or 0,
self.contInfo,
self.invoices,
self.hold_types,
self.hold_data,
self.credit_note_map,
self.gst_release_map,
self.output_file
)
return self.output_file, None