Merge branch 'main' of https://gitea.lcepl.org/pjpatil12/Payment_Reconciliation into pankaj-dev
This commit is contained in:
@@ -1,72 +1,65 @@
|
||||
import mysql.connector
|
||||
|
||||
|
||||
|
||||
from mysql.connector import Error
|
||||
import config
|
||||
import openpyxl
|
||||
import os
|
||||
import re
|
||||
import ast
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class ContractorInfo:
|
||||
ID = ""
|
||||
contInfo = None
|
||||
def __init__(self, id):
|
||||
self.ID = id
|
||||
print(id)
|
||||
def __init__(self, contractor_id):
|
||||
self.ID = contractor_id
|
||||
self.contInfo = None
|
||||
self.fetchData()
|
||||
|
||||
def fetchData(self):
|
||||
"""Fetch basic contractor info by ID."""
|
||||
try:
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True, buffered=True)
|
||||
cursor.callproc('GetContractorInfoById', [self.ID])
|
||||
for result in cursor.stored_results():
|
||||
self.contInfo = result.fetchone()
|
||||
|
||||
print(self.contInfo,flush=True)
|
||||
with connection.cursor(dictionary=True, buffered=True) as cursor:
|
||||
cursor.callproc('GetContractorInfoById', [self.ID])
|
||||
# Get the first result set
|
||||
for result in cursor.stored_results():
|
||||
self.contInfo = result.fetchone()
|
||||
except Error as e:
|
||||
print(f"Error fetching contractor info: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
if connection.is_connected():
|
||||
connection.close()
|
||||
|
||||
def fetchalldata(self):
|
||||
|
||||
"""Fetch hold types and invoices for contractor."""
|
||||
data = {}
|
||||
try:
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True, buffered=True)
|
||||
print("here", flush=True)
|
||||
|
||||
with connection.cursor(dictionary=True, buffered=True) as cursor:
|
||||
# Fetch Hold Types
|
||||
cursor.callproc('GetHoldTypesByContractor', [self.ID])
|
||||
hold_types = []
|
||||
for result in cursor.stored_results():
|
||||
hold_types = result.fetchall()
|
||||
hold_type_map = {ht['hold_type_id']: ht['hold_type'] for ht in hold_types}
|
||||
data['hold_types'] = hold_type_map
|
||||
|
||||
# ---------------- Hold Types ----------------
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
# Fetch Invoices
|
||||
cursor.callproc('GetInvoicesByContractor', [self.ID])
|
||||
invoices = []
|
||||
for result in cursor.stored_results():
|
||||
invoices = result.fetchall()
|
||||
|
||||
cursor.callproc('GetHoldTypesByContractor', [self.ID])
|
||||
|
||||
hold_types = []
|
||||
for result in cursor.stored_results():
|
||||
hold_types = result.fetchall()
|
||||
hold_type_map = {ht['hold_type_id']: ht['hold_type'] for ht in hold_types}
|
||||
|
||||
# ---------------- Invoices ----------------
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
cursor.callproc('GetInvoicesByContractor', [self.ID])
|
||||
|
||||
invoices = []
|
||||
for result in cursor.stored_results():
|
||||
invoices = result.fetchall()
|
||||
|
||||
# Remove duplicate invoices
|
||||
invoice_ids_seen = set()
|
||||
unique_invoices = []
|
||||
for inv in invoices:
|
||||
if inv["Invoice_Id"] not in invoice_ids_seen:
|
||||
invoice_ids_seen.add(inv["Invoice_Id"])
|
||||
unique_invoices.append(inv)
|
||||
invoices = unique_invoices
|
||||
# Remove duplicate invoices
|
||||
seen_ids = set()
|
||||
unique_invoices = []
|
||||
for inv in invoices:
|
||||
if inv['Invoice_Id'] not in seen_ids:
|
||||
seen_ids.add(inv['Invoice_Id'])
|
||||
unique_invoices.append(inv)
|
||||
data['invoices'] = unique_invoices
|
||||
|
||||
except Error as e:
|
||||
print(f"Error fetching contractor data: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
if connection.is_connected():
|
||||
connection.close()
|
||||
|
||||
|
||||
return data
|
||||
|
||||
458
model/Invoice.py
458
model/Invoice.py
@@ -1,35 +1,76 @@
|
||||
|
||||
import config
|
||||
import mysql.connector
|
||||
|
||||
# ------------------- Helper -------------------
|
||||
# ------------------- Helper Functions -------------------
|
||||
def clear_results(cursor):
|
||||
"""Consume all stored results to prevent cursor issues."""
|
||||
for r in cursor.stored_results():
|
||||
r.fetchall()
|
||||
|
||||
def fetch_one(cursor):
|
||||
"""Fetch first row from stored results."""
|
||||
for result in cursor.stored_results():
|
||||
return result.fetchone()
|
||||
return None
|
||||
|
||||
# ------------------- Get Village Id -------------------
|
||||
def get_village_id(village_name):
|
||||
def fetch_all(cursor):
|
||||
"""Fetch all rows from stored results."""
|
||||
data = []
|
||||
for result in cursor.stored_results():
|
||||
data.extend(result.fetchall())
|
||||
return data
|
||||
|
||||
def get_numeric_values(data):
|
||||
"""Return numeric fields for invoices safely."""
|
||||
return [
|
||||
float(data.get('basic_amount') or 0),
|
||||
float(data.get('debit_amount') or 0),
|
||||
float(data.get('after_debit_amount') or 0),
|
||||
float(data.get('amount') or 0),
|
||||
float(data.get('gst_amount') or 0),
|
||||
float(data.get('tds_amount') or 0),
|
||||
float(data.get('sd_amount') or 0),
|
||||
float(data.get('on_commission') or 0),
|
||||
float(data.get('hydro_testing') or 0),
|
||||
float(data.get('gst_sd_amount') or 0),
|
||||
float(data.get('final_amount') or 0),
|
||||
]
|
||||
|
||||
def execute_db_operation(operation_func):
|
||||
"""General DB operation wrapper with commit/rollback."""
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
cursor.callproc("GetVillageIdByName", (village_name,))
|
||||
village_result = None
|
||||
|
||||
for rs in cursor.stored_results():
|
||||
village_result = rs.fetchone()
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return village_result
|
||||
|
||||
|
||||
# ------------------- Insert Invoice -------------------
|
||||
def insert_invoice(data, village_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
# 1. Insert Invoice
|
||||
result = operation_func(cursor)
|
||||
connection.commit()
|
||||
return result
|
||||
except Exception:
|
||||
connection.rollback()
|
||||
raise
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Village Functions -------------------
|
||||
def get_village_id(village_name):
|
||||
def operation(cursor):
|
||||
cursor.callproc("GetVillageIdByName", (village_name,))
|
||||
return fetch_one(cursor)
|
||||
return execute_db_operation(operation)
|
||||
|
||||
def get_all_villages():
|
||||
def operation(cursor):
|
||||
cursor.callproc("GetAllVillages")
|
||||
return fetch_all(cursor)
|
||||
return execute_db_operation(operation)
|
||||
|
||||
|
||||
# ------------------- Invoice Functions -------------------
|
||||
def insert_invoice(data, village_id):
|
||||
def operation(cursor):
|
||||
# Insert invoice
|
||||
cursor.callproc('InsertInvoice', [
|
||||
data.get('pmc_no'),
|
||||
village_id,
|
||||
@@ -37,29 +78,14 @@ def insert_invoice(data, village_id):
|
||||
data.get('invoice_details'),
|
||||
data.get('invoice_date'),
|
||||
data.get('invoice_no'),
|
||||
float(data.get('basic_amount') or 0),
|
||||
float(data.get('debit_amount') or 0),
|
||||
float(data.get('after_debit_amount') or 0),
|
||||
float(data.get('amount') or 0),
|
||||
float(data.get('gst_amount') or 0),
|
||||
float(data.get('tds_amount') or 0),
|
||||
float(data.get('sd_amount') or 0),
|
||||
float(data.get('on_commission') or 0),
|
||||
float(data.get('hydro_testing') or 0),
|
||||
float(data.get('gst_sd_amount') or 0),
|
||||
float(data.get('final_amount') or 0)
|
||||
*get_numeric_values(data)
|
||||
])
|
||||
|
||||
invoice_id = None
|
||||
for result in cursor.stored_results():
|
||||
row = result.fetchone()
|
||||
if row:
|
||||
invoice_id = row.get('invoice_id')
|
||||
|
||||
if not invoice_id:
|
||||
invoice_row = fetch_one(cursor)
|
||||
if not invoice_row:
|
||||
raise Exception("Invoice ID not returned")
|
||||
invoice_id = invoice_row.get('invoice_id')
|
||||
|
||||
# 2. Insert Inpayment
|
||||
# Insert inpayment
|
||||
cursor.callproc('InsertInpayment', [
|
||||
data.get('pmc_no'),
|
||||
village_id,
|
||||
@@ -67,221 +93,41 @@ def insert_invoice(data, village_id):
|
||||
data.get('invoice_details'),
|
||||
data.get('invoice_date'),
|
||||
data.get('invoice_no'),
|
||||
float(data.get('basic_amount') or 0),
|
||||
float(data.get('debit_amount') or 0),
|
||||
float(data.get('after_debit_amount') or 0),
|
||||
float(data.get('amount') or 0),
|
||||
float(data.get('gst_amount') or 0),
|
||||
float(data.get('tds_amount') or 0),
|
||||
float(data.get('sd_amount') or 0),
|
||||
float(data.get('on_commission') or 0),
|
||||
float(data.get('hydro_testing') or 0),
|
||||
float(data.get('gst_sd_amount') or 0),
|
||||
float(data.get('final_amount') or 0),
|
||||
*get_numeric_values(data),
|
||||
data.get('subcontractor_id')
|
||||
])
|
||||
clear_results(cursor)
|
||||
|
||||
connection.commit()
|
||||
return invoice_id
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
return execute_db_operation(operation)
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Assign Subcontractor -------------------
|
||||
def assign_subcontractor(data, village_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
cursor.callproc('AssignSubcontractor', [
|
||||
data.get('pmc_no'),
|
||||
data.get('subcontractor_id'),
|
||||
village_id
|
||||
])
|
||||
clear_results(cursor)
|
||||
|
||||
connection.commit()
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Insert Hold Types -------------------
|
||||
def insert_hold_types(data, invoice_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
hold_types = data.getlist('hold_type[]')
|
||||
hold_amounts = data.getlist('hold_amount[]')
|
||||
|
||||
for hold_type, hold_amount in zip(hold_types, hold_amounts):
|
||||
if not hold_type:
|
||||
continue
|
||||
|
||||
cursor.callproc('GetHoldTypeIdByName', [hold_type])
|
||||
hold_type_result = None
|
||||
|
||||
for result in cursor.stored_results():
|
||||
hold_type_result = result.fetchone()
|
||||
|
||||
if not hold_type_result:
|
||||
cursor.callproc('InsertHoldType', [hold_type, 0])
|
||||
cursor.execute("SELECT @_InsertHoldType_1")
|
||||
hold_type_id = cursor.fetchone()[0]
|
||||
else:
|
||||
hold_type_id = hold_type_result['hold_type_id']
|
||||
|
||||
hold_amount = float(hold_amount or 0)
|
||||
|
||||
cursor.callproc('InsertInvoiceSubcontractorHold', [
|
||||
data.get('subcontractor_id'),
|
||||
invoice_id,
|
||||
hold_type_id,
|
||||
hold_amount
|
||||
])
|
||||
clear_results(cursor)
|
||||
|
||||
connection.commit()
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Get All Invoices -------------------
|
||||
def get_all_invoice_details():
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
def operation(cursor):
|
||||
cursor.callproc('GetAllInvoiceDetails')
|
||||
return fetch_all(cursor)
|
||||
return execute_db_operation(operation)
|
||||
|
||||
cursor.callproc('GetAllInvoiceDetails')
|
||||
invoices = []
|
||||
|
||||
for result in cursor.stored_results():
|
||||
invoices = result.fetchall()
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return invoices
|
||||
|
||||
|
||||
# ------------------- Get All Villages -------------------
|
||||
def get_all_villages():
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
cursor.callproc("GetAllVillages")
|
||||
villages = []
|
||||
|
||||
for result in cursor.stored_results():
|
||||
villages = result.fetchall()
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return villages
|
||||
|
||||
|
||||
# ------------------- Search Contractors -------------------
|
||||
def search_contractors(sub_query):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
cursor.callproc('SearchContractorsByName', [sub_query])
|
||||
results = []
|
||||
|
||||
for result in cursor.stored_results():
|
||||
results = result.fetchall()
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return results
|
||||
|
||||
|
||||
# ------------------- Get All Hold Types -------------------
|
||||
def get_all_hold_types():
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
cursor.callproc("GetAllHoldTypes")
|
||||
hold_types = []
|
||||
|
||||
for result in cursor.stored_results():
|
||||
hold_types = result.fetchall()
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return hold_types
|
||||
|
||||
|
||||
# ------------------- Get Invoice By Id -------------------
|
||||
def get_invoice_by_id(invoice_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
def operation(cursor):
|
||||
cursor.callproc('GetInvoiceDetailsById', [invoice_id])
|
||||
invoice = fetch_one(cursor)
|
||||
|
||||
cursor.callproc('GetInvoiceDetailsById', [invoice_id])
|
||||
invoice = None
|
||||
cursor.callproc('GetHoldAmountsByInvoiceId', [invoice_id])
|
||||
hold_amounts = fetch_all(cursor)
|
||||
|
||||
for result in cursor.stored_results():
|
||||
invoice = result.fetchone()
|
||||
if invoice:
|
||||
invoice["hold_amounts"] = hold_amounts
|
||||
return invoice
|
||||
return execute_db_operation(operation)
|
||||
|
||||
cursor.callproc('GetHoldAmountsByInvoiceId', [invoice_id])
|
||||
hold_amounts = []
|
||||
|
||||
for result in cursor.stored_results():
|
||||
hold_amounts = result.fetchall()
|
||||
|
||||
if invoice:
|
||||
invoice["hold_amounts"] = hold_amounts
|
||||
|
||||
cursor.close()
|
||||
connection.close()
|
||||
return invoice
|
||||
|
||||
|
||||
# ------------------- Update Invoice -------------------
|
||||
def update_invoice(data, invoice_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
def operation(cursor):
|
||||
cursor.callproc("GetVillageIdByName", (data.get('village'),))
|
||||
village = None
|
||||
|
||||
for rs in cursor.stored_results():
|
||||
village = rs.fetchone()
|
||||
|
||||
village = fetch_one(cursor)
|
||||
if not village:
|
||||
raise Exception("Village not found")
|
||||
village_id = village['Village_Id']
|
||||
|
||||
numeric = [
|
||||
float(data.get('basic_amount') or 0),
|
||||
float(data.get('debit_amount') or 0),
|
||||
float(data.get('after_debit_amount') or 0),
|
||||
float(data.get('amount') or 0),
|
||||
float(data.get('gst_amount') or 0),
|
||||
float(data.get('tds_amount') or 0),
|
||||
float(data.get('sd_amount') or 0),
|
||||
float(data.get('on_commission') or 0),
|
||||
float(data.get('hydro_testing') or 0),
|
||||
float(data.get('gst_sd_amount') or 0),
|
||||
float(data.get('final_amount') or 0),
|
||||
]
|
||||
|
||||
cursor.callproc('UpdateInvoice', [
|
||||
data.get('pmc_no'),
|
||||
village_id,
|
||||
@@ -289,91 +135,101 @@ def update_invoice(data, invoice_id):
|
||||
data.get('invoice_details'),
|
||||
data.get('invoice_date'),
|
||||
data.get('invoice_no'),
|
||||
*numeric,
|
||||
*get_numeric_values(data),
|
||||
invoice_id
|
||||
])
|
||||
clear_results(cursor)
|
||||
execute_db_operation(operation)
|
||||
|
||||
connection.commit()
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Update Inpayment -------------------
|
||||
def update_inpayment(data):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
numeric = [
|
||||
float(data.get('basic_amount') or 0),
|
||||
float(data.get('debit_amount') or 0),
|
||||
float(data.get('after_debit_amount') or 0),
|
||||
float(data.get('amount') or 0),
|
||||
float(data.get('gst_amount') or 0),
|
||||
float(data.get('tds_amount') or 0),
|
||||
float(data.get('sd_amount') or 0),
|
||||
float(data.get('on_commission') or 0),
|
||||
float(data.get('hydro_testing') or 0),
|
||||
float(data.get('gst_sd_amount') or 0),
|
||||
float(data.get('final_amount') or 0),
|
||||
]
|
||||
|
||||
def operation(cursor):
|
||||
cursor.callproc('UpdateInpayment', [
|
||||
data.get('work_type'),
|
||||
data.get('invoice_details'),
|
||||
data.get('invoice_date'),
|
||||
*numeric,
|
||||
*get_numeric_values(data),
|
||||
data.get('pmc_no'),
|
||||
data.get('invoice_no')
|
||||
])
|
||||
clear_results(cursor)
|
||||
execute_db_operation(operation)
|
||||
|
||||
connection.commit()
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
|
||||
# ------------------- Delete Invoice -------------------
|
||||
def delete_invoice_data(invoice_id, user_id):
|
||||
connection = config.get_db_connection()
|
||||
cursor = connection.cursor(dictionary=True)
|
||||
|
||||
try:
|
||||
cursor.callproc('GetInvoicePMCById', [invoice_id])
|
||||
|
||||
def operation(cursor):
|
||||
# Fetch PMC and Invoice_No from DB
|
||||
cursor.callproc('GetInvoicePMCById', (invoice_id,))
|
||||
record = {}
|
||||
for result in cursor.stored_results():
|
||||
record = result.fetchone() or {}
|
||||
if not record:
|
||||
raise Exception("Invoice not found")
|
||||
|
||||
# Use exact DB keys
|
||||
pmc_no = record['PMC_No']
|
||||
invoice_no = record['Invoice_No']
|
||||
|
||||
# Delete invoice
|
||||
cursor.callproc("DeleteInvoice", (invoice_id,))
|
||||
clear_results(cursor)
|
||||
|
||||
cursor.callproc(
|
||||
'DeleteInpaymentByPMCInvoice',
|
||||
[record['PMC_No'], record['invoice_no']]
|
||||
)
|
||||
# Delete inpayment
|
||||
cursor.callproc('DeleteInpaymentByPMCInvoice', (pmc_no, invoice_no))
|
||||
clear_results(cursor)
|
||||
|
||||
connection.commit()
|
||||
execute_db_operation(operation)
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
raise e
|
||||
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
# ------------------- Subcontractor Functions -------------------
|
||||
def assign_subcontractor(data, village_id):
|
||||
def operation(cursor):
|
||||
cursor.callproc('AssignSubcontractor', [
|
||||
data.get('pmc_no'),
|
||||
data.get('subcontractor_id'),
|
||||
village_id
|
||||
])
|
||||
clear_results(cursor)
|
||||
execute_db_operation(operation)
|
||||
|
||||
|
||||
# ------------------- Hold Types Functions -------------------
|
||||
def insert_hold_types(data, invoice_id):
|
||||
def operation(cursor):
|
||||
hold_types = data.getlist('hold_type[]')
|
||||
hold_amounts = data.getlist('hold_amount[]')
|
||||
|
||||
for hold_type, hold_amount in zip(hold_types, hold_amounts):
|
||||
if not hold_type:
|
||||
continue
|
||||
|
||||
cursor.callproc('GetHoldTypeIdByName', [hold_type])
|
||||
hold_type_result = fetch_one(cursor)
|
||||
|
||||
if not hold_type_result:
|
||||
cursor.callproc('InsertHoldType', [hold_type, 0])
|
||||
cursor.execute("SELECT @_InsertHoldType_1")
|
||||
hold_type_id = cursor.fetchone()[0]
|
||||
else:
|
||||
hold_type_id = hold_type_result['hold_type_id']
|
||||
|
||||
cursor.callproc('InsertInvoiceSubcontractorHold', [
|
||||
data.get('subcontractor_id'),
|
||||
invoice_id,
|
||||
hold_type_id,
|
||||
float(hold_amount or 0)
|
||||
])
|
||||
clear_results(cursor)
|
||||
execute_db_operation(operation)
|
||||
|
||||
def get_all_hold_types():
|
||||
def operation(cursor):
|
||||
cursor.callproc("GetAllHoldTypes")
|
||||
return fetch_all(cursor)
|
||||
return execute_db_operation(operation)
|
||||
|
||||
|
||||
# ------------------- Contractor Functions -------------------
|
||||
def search_contractors(sub_query):
|
||||
def operation(cursor):
|
||||
cursor.callproc('SearchContractorsByName', [sub_query])
|
||||
return fetch_all(cursor)
|
||||
return execute_db_operation(operation)
|
||||
200
model/Village.py
200
model/Village.py
@@ -1,14 +1,7 @@
|
||||
|
||||
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
||||
|
||||
from model.Utilities import RegEx, ResponseHandler, HtmlHelper, ItemCRUDType
|
||||
from model.Log import LogData, LogHelper
|
||||
|
||||
# return blocks
|
||||
from model.Utilities import ResponseHandler, HtmlHelper, ItemCRUDType
|
||||
import config
|
||||
|
||||
import mysql.connector
|
||||
from mysql.connector import Error
|
||||
|
||||
from model.ItemCRUD import ItemCRUD
|
||||
|
||||
|
||||
@@ -19,103 +12,162 @@ class Village:
|
||||
def __init__(self):
|
||||
self.isSuccess = False
|
||||
self.resultMessage = ""
|
||||
self.village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
|
||||
# 🔹 Helper: sync status
|
||||
def _set_status(self, village):
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
|
||||
# 🔹 Helper: get request data
|
||||
def _get_form_data(self, request):
|
||||
block_id = request.form.get('block_Id')
|
||||
village_name = request.form.get('Village_Name', '').strip()
|
||||
return block_id, village_name
|
||||
|
||||
def AddVillage(self, request):
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
block_id, village_name = self._get_form_data(request)
|
||||
|
||||
block_id = request.form.get('block_Id')
|
||||
village_name = request.form.get('Village_Name', '').strip()
|
||||
if not village_name:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = "Village name cannot be empty"
|
||||
return
|
||||
|
||||
village.AddItem(request=request, parentid=block_id, childname=village_name, storedprocfetch="GetVillageByNameAndBlock", storedprocadd="SaveVillage" )
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
return
|
||||
#self.isSuccess = False
|
||||
try:
|
||||
self.village.AddItem(
|
||||
request=request,
|
||||
parentid=block_id,
|
||||
childname=village_name,
|
||||
storedprocfetch="GetVillageByNameAndBlock",
|
||||
storedprocadd="SaveVillage"
|
||||
)
|
||||
self._set_status(self.village)
|
||||
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
|
||||
def GetAllVillages(self, request):
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
villagesdata = village.GetAllData(request=request, storedproc="GetAllVillages")
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
return villagesdata
|
||||
|
||||
try:
|
||||
villagesdata = self.village.GetAllData(
|
||||
request=request,
|
||||
storedproc="GetAllVillages"
|
||||
)
|
||||
self._set_status(self.village)
|
||||
return villagesdata
|
||||
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
return []
|
||||
|
||||
def CheckVillage(self, request):
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
block_id = request.form.get('block_Id')
|
||||
village_name = request.form.get('Village_Name', '').strip()
|
||||
result = village.CheckItem(request=request, parentid=block_id, childname=village_name, storedprocfetch="GetVillageByNameAndBlocks")
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
return result
|
||||
block_id, village_name = self._get_form_data(request)
|
||||
|
||||
if not village_name:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = "Village name cannot be empty"
|
||||
return None
|
||||
|
||||
try:
|
||||
result = self.village.CheckItem(
|
||||
request=request,
|
||||
parentid=block_id,
|
||||
childname=village_name,
|
||||
storedprocfetch="GetVillageByNameAndBlocks"
|
||||
)
|
||||
self._set_status(self.village)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
return None
|
||||
|
||||
|
||||
def DeleteVillage(self, request, village_id):
|
||||
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
try:
|
||||
self.village.DeleteItem(
|
||||
request=request,
|
||||
itemID=village_id,
|
||||
storedprocDelete="DeleteVillage"
|
||||
)
|
||||
self._set_status(self.village)
|
||||
|
||||
village.DeleteItem(request=request, itemID=village_id, storedprocDelete="DeleteVillage" )
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
return
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
|
||||
def EditVillage(self, request, village_id):
|
||||
corsor=None
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
block_id, village_name = self._get_form_data(request)
|
||||
|
||||
block_id = request.form.get('block_Id')
|
||||
village_name = request.form.get('Village_Name', '').strip()
|
||||
|
||||
village.EditItem(request=request,childid=village_id,parentid=block_id,childname=village_name,storedprocupdate="UpdateVillage" )
|
||||
|
||||
self.isSuccess = village.isSuccess
|
||||
self.resultMessage = village.resultMessage
|
||||
return
|
||||
|
||||
# def GetVillageByID(self, request, id):
|
||||
|
||||
# village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
# villagedetailsdata = village.GetAllData(request=request, storedproc="GetVillageDetailsById")
|
||||
# self.isSuccess = village.isSuccess
|
||||
# self.resultMessage = village.resultMessage
|
||||
# return villagedetailsdata
|
||||
|
||||
def GetVillageByID(self, request, id):
|
||||
village = ItemCRUD(itemType=ItemCRUDType.Village)
|
||||
villagedetailsdata = village.GetDataByID(id=id,storedproc="GetVillageDetailsById")
|
||||
if villagedetailsdata:
|
||||
self.isSuccess = True
|
||||
else:
|
||||
if not village_name:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = "Village not found"
|
||||
self.resultMessage = "Village name cannot be empty"
|
||||
return
|
||||
|
||||
return villagedetailsdata
|
||||
try:
|
||||
self.village.EditItem(
|
||||
request=request,
|
||||
childid=village_id,
|
||||
parentid=block_id,
|
||||
childname=village_name,
|
||||
storedprocupdate="UpdateVillage"
|
||||
)
|
||||
self._set_status(self.village)
|
||||
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
|
||||
def GetAllBlocks(self, request):
|
||||
def GetVillageByID(self, id):
|
||||
|
||||
try:
|
||||
villagedetailsdata = self.village.GetDataByID(
|
||||
id=id,
|
||||
storedproc="GetVillageDetailsById"
|
||||
)
|
||||
|
||||
if villagedetailsdata:
|
||||
self.isSuccess = True
|
||||
else:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = "Village not found"
|
||||
|
||||
return villagedetailsdata
|
||||
|
||||
except Exception as e:
|
||||
self.isSuccess = False
|
||||
self.resultMessage = str(e)
|
||||
return None
|
||||
|
||||
def GetAllBlocks(self):
|
||||
blocks = []
|
||||
self.isSuccess = False
|
||||
self.resultMessage = ""
|
||||
connection = config.get_db_connection()
|
||||
|
||||
connection = config.get_db_connection()
|
||||
if not connection:
|
||||
return []
|
||||
|
||||
cursor = connection.cursor()
|
||||
|
||||
try:
|
||||
cursor.callproc('GetAllBlocks')
|
||||
for result in cursor.stored_results():
|
||||
blocks = result.fetchall()
|
||||
with connection.cursor() as cursor:
|
||||
cursor.callproc('GetAllBlocks')
|
||||
|
||||
for result in cursor.stored_results():
|
||||
blocks.extend(result.fetchall())
|
||||
|
||||
self.isSuccess = True
|
||||
return blocks
|
||||
|
||||
except mysql.connector.Error as e:
|
||||
print(f"Error fetching blocks: {e}")
|
||||
self.isSuccess = False
|
||||
self.resultMessage = HtmlHelper.json_response(ResponseHandler.fetch_failure("block"), 500)
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
self.resultMessage = HtmlHelper.json_response(
|
||||
ResponseHandler.fetch_failure("block"), 500
|
||||
)
|
||||
return []
|
||||
|
||||
return blocks
|
||||
finally:
|
||||
connection.close()
|
||||
Reference in New Issue
Block a user