Files
Payment_Reconciliation/model/Report.py

275 lines
11 KiB
Python

import config
from datetime import datetime
from flask import send_file
import openpyxl
from openpyxl.styles import Font
from model.FolderAndFile import FolderAndFile
class ReportHelper:
isSuccess = False
resultMessage = ""
data=[]
def __init__(self):
self.isSuccess = False
self.resultMessage = ""
self.data = []
@staticmethod
def execute_sp(cursor, proc_name, params=[], fetch_one=False):
cursor.callproc(proc_name, params)
return (
ReportHelper.fetch_one_result(cursor)
if fetch_one else
ReportHelper.fetch_all_results(cursor)
)
@staticmethod
def fetch_all_results(cursor):
data = []
for result in cursor.stored_results():
data = result.fetchall()
return data
@staticmethod
def fetch_one_result(cursor):
data = None
for result in cursor.stored_results():
data = result.fetchone()
return data
@staticmethod
def search_contractor(request):
subcontractor_name = request.form.get("subcontractor_name")
pmc_no = request.form.get("pmc_no")
state = request.form.get("state")
district = request.form.get("district")
block = request.form.get("block")
village = request.form.get("village")
year_from = request.form.get("year_from")
year_to = request.form.get("year_to")
connection = config.get_db_connection()
if not connection:
return []
cursor = connection.cursor(dictionary=True)
try:
data = ReportHelper.execute_sp(
cursor,
"search_contractor_info",
[
subcontractor_name or None,
pmc_no or None,
state or None,
district or None,
block or None,
village or None,
year_from or None,
year_to or None
]
)
except Exception as e:
print(f"Error in search_contractor: {e}")
data = []
finally:
cursor.close()
connection.close()
return data
@staticmethod
def get_contractor_report(contractor_id):
connection = config.get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
try:
# Contractor Info (only one fetch)
contInfo = ReportHelper.execute_sp(cursor, 'GetContractorInfo', [contractor_id], True)
# Hold Types
hold_types = ReportHelper.execute_sp(cursor, 'GetContractorHoldTypes', [contractor_id])
# Invoices
invoices = ReportHelper.execute_sp(cursor, 'GetContractorInvoices', [contractor_id])
# GST Release
gst_rel = ReportHelper.execute_sp(cursor, 'GetGSTRelease', [contractor_id])
# Hold Release
hold_release = ReportHelper.execute_sp(cursor, 'GetHoldRelease', [contractor_id])
# Credit Note
credit_note = ReportHelper.execute_sp(cursor, 'GetCreditNote', [contractor_id])
# Payments
payments = ReportHelper.execute_sp(cursor, 'GetPayments', [contractor_id])
# Totals
total = {
"sum_invo_basic_amt": float(sum(row['Basic_Amount'] or 0 for row in invoices)),
"sum_invo_debit_amt": float(sum(row['Debit_Amount'] or 0 for row in invoices)),
"sum_invo_after_debit_amt": float(sum(row['After_Debit_Amount'] or 0 for row in invoices)),
"sum_invo_amt": float(sum(row['Amount'] or 0 for row in invoices)),
"sum_invo_gst_amt": float(sum(row['GST_Amount'] or 0 for row in invoices)),
"sum_invo_tds_amt": float(sum(row['TDS_Amount'] or 0 for row in invoices)),
"sum_invo_ds_amt": float(sum(row['SD_Amount'] or 0 for row in invoices)),
"sum_invo_on_commission": float(sum(row['On_Commission'] or 0 for row in invoices)),
"sum_invo_hydro_test": float(sum(row['Hydro_Testing'] or 0 for row in invoices)),
"sum_invo_gst_sd_amt": float(sum(row['GST_SD_Amount'] or 0 for row in invoices)),
"sum_invo_final_amt": float(sum(row['Final_Amount'] or 0 for row in invoices)),
"sum_invo_hold_amt": float(sum(row['hold_amount'] or 0 for row in invoices)),
"sum_gst_basic_amt": float(sum(row['basic_amount'] or 0 for row in gst_rel)),
"sum_gst_final_amt": float(sum(row['final_amount'] or 0 for row in gst_rel)),
"sum_pay_payment_amt": float(sum(row['Payment_Amount'] or 0 for row in payments)),
"sum_pay_tds_payment_amt": float(sum(row['TDS_Payment_Amount'] or 0 for row in payments)),
"sum_pay_total_amt": float(sum(row['Total_amount'] or 0 for row in payments))
}
current_date = datetime.now().strftime('%Y-%m-%d')
finally:
cursor.close()
connection.close()
return {
"contInfo": contInfo,
"invoices": invoices,
"hold_types": hold_types,
"gst_rel": gst_rel,
"payments": payments,
"credit_note": credit_note,
"hold_release": hold_release,
"total": total,
"current_date": current_date
}
@staticmethod
def download_report(contractor_id):
try:
connection = config.get_db_connection()
cursor = connection.cursor(dictionary=True)
# -------- Contractor Info --------
contInfo = ReportHelper.execute_sp(cursor, 'GetContractorInfo', [contractor_id], True)
if not contInfo:
return "No contractor found", 404
# -------- Invoice Data --------
cursor.callproc('FetchInvoicesByContractor', [contractor_id])
invoices = []
for result in cursor.stored_results():
invoices.extend(result.fetchall())
if not invoices:
return "No invoice data found"
# -------- Create Workbook --------
workbook = openpyxl.Workbook()
sheet = workbook.active
sheet.title = "Contractor Report"
# ================= CONTRACTOR DETAILS =================
sheet.append(["SUB CONTRACTOR DETAILS"])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
sheet.append([])
sheet.append(["Name", contInfo.get("Contractor_Name") or ""])
sheet.append(["Mobile No", contInfo.get("Mobile_No") or ""])
sheet.append(["Email", contInfo.get("Email") or ""])
sheet.append(["Village", contInfo.get("Village_Name") or ""])
sheet.append(["Block", contInfo.get("Block_Name") or ""])
sheet.append(["District", contInfo.get("District_Name") or ""])
sheet.append(["State", contInfo.get("State_Name") or ""])
sheet.append(["Address", contInfo.get("Address") or ""])
sheet.append(["GST No", contInfo.get("GST_No") or ""])
sheet.append(["PAN No", contInfo.get("PAN_No") or ""])
sheet.append([])
sheet.append([])
# ================= TABLE HEADERS =================
headers = [
"PMC No", "Village", "Invoice No", "Invoice Date", "Work Type","Invoice_Details",
"Basic Amount", "Debit Amount", "After Debit Amount",
"Amount", "GST Amount", "TDS Amount", "SD Amount",
"On Commission", "Hydro Testing", "Hold Amount",
"GST SD Amount", "Final Amount",
"Payment Amount", "TDS Payment",
"Total Amount", "UTR"
]
sheet.append(headers)
for col in range(1, len(headers) + 1):
sheet.cell(row=sheet.max_row, column=col).font = Font(bold=True)
# ================= DATA =================
total_final = 0
total_payment = 0
total_amount = 0
for inv in invoices:
row = [
inv.get("PMC_No"),
inv.get("Village_Name"),
inv.get("invoice_no"),
inv.get("Invoice_Date"),
inv.get("Work_Type"),
inv.get("Invoice_Details"),
inv.get("Basic_Amount"),
inv.get("Debit_Amount"),
inv.get("After_Debit_Amount"),
inv.get("Amount"),
inv.get("GST_Amount"),
inv.get("TDS_Amount"),
inv.get("SD_Amount"),
inv.get("On_Commission"),
inv.get("Hydro_Testing"),
inv.get("Hold_Amount"),
inv.get("GST_SD_Amount"),
inv.get("Final_Amount"),
inv.get("Payment_Amount"),
inv.get("TDS_Payment_Amount"),
inv.get("Total_Amount"),
inv.get("UTR")
]
total_final += float(inv.get("Final_Amount") or 0)
total_payment += float(inv.get("Payment_Amount") or 0)
total_amount += float(inv.get("Total_Amount") or 0)
sheet.append(row)
# ================= TOTAL ROW =================
sheet.append([])
sheet.append([
"", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "",
"TOTAL",
total_final,
total_payment,
"",
total_amount,
""
])
# ================= AUTO WIDTH =================
for column in sheet.columns:
max_length = 0
column_letter = column[0].column_letter
for cell in column:
if cell.value:
max_length = max(max_length, len(str(cell.value)))
sheet.column_dimensions[column_letter].width = max_length + 2
# ================= SAVE FILE =================
filename = f"Contractor_Report_{contInfo.get('Contractor_Name')}.xlsx"
output_file = FolderAndFile.get_download_path(filename)
workbook.save(output_file)
return send_file(output_file, as_attachment=True)
except Exception as e:
return str(e)