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]) self.gst_release_raw = GeneralUse.execute_sp(cursor, 'GetGSTRelease', [self.contractor_id]) # ---------- 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]) self.gst_release_raw = GeneralUse.execute_sp(cursor, 'GetGSTReleaseDetailsByPMC', [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.pmc_no, 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