diff --git a/controllers/invoice_controller.py b/controllers/invoice_controller.py
index d8af37a..1170741 100644
--- a/controllers/invoice_controller.py
+++ b/controllers/invoice_controller.py
@@ -1,98 +1,101 @@
+
+
+
# controllers/invoice_controller.py
from flask import Blueprint, request, jsonify, render_template
from flask_login import login_required, current_user
from model.Invoice import *
-from model.Log import LogHelper
+from model.Log import LogHelper
invoice_bp = Blueprint('invoice', __name__)
-# -------------------------------- Add Invoice ---------------------------------
-@invoice_bp.route('/add_invoice', methods=['GET', 'POST'])
-@login_required
-def add_invoice():
- if request.method == 'POST':
+# ------------------------------- Helpers -------------------------------
+def handle_exception(func):
+ """Decorator to handle exceptions and return JSON error responses."""
+ def wrapper(*args, **kwargs):
try:
- village_name = request.form.get('village')
- village_result = get_village_id(village_name)
-
- if not village_result:
- return jsonify({"status": "error", "message": f"Village '{village_name}' not found"}), 400
-
- village_id = village_result['Village_Id']
- data = request.form
-
- invoice_id = insert_invoice(data, village_id)
- assign_subcontractor(data, village_id)
- insert_hold_types(data, invoice_id)
-
- LogHelper.log_action("Add invoice", f"User {current_user.id} Added invoice '{data.get('pmc_no')}'")
-
- return jsonify({"status": "success", "message": "Invoice added successfully"}), 201
-
+ return func(*args, **kwargs)
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
+ wrapper.__name__ = func.__name__
+ return wrapper
+
+def log_action(action: str, detail: str):
+ LogHelper.log_action(action, f"User {current_user.id} {detail}")
+
+
+# ------------------------------- Add Invoice -------------------------------
+@invoice_bp.route('/add_invoice', methods=['GET', 'POST'])
+@login_required
+@handle_exception
+def add_invoice():
+ if request.method == 'POST':
+ data = request.form
+ village_name = data.get('village')
+ village_result = get_village_id(village_name)
+
+ if not village_result:
+ return jsonify({"status": "error", "message": f"Village '{village_name}' not found"}), 400
+
+ village_id = village_result['Village_Id']
+ invoice_id = insert_invoice(data, village_id)
+ assign_subcontractor(data, village_id)
+ insert_hold_types(data, invoice_id)
+
+ log_action("Add invoice", f"added invoice '{data.get('pmc_no')}'")
+ return jsonify({"status": "success", "message": "Invoice added successfully"}), 201
invoices = get_all_invoice_details()
villages = get_all_villages()
return render_template('add_invoice.html', invoices=invoices, villages=villages)
-# ------------------- Search Subcontractor -------------------
+# ------------------------------- Search Subcontractor -------------------------------
@invoice_bp.route('/search_subcontractor', methods=['POST'])
@login_required
+@handle_exception
def search_subcontractor():
- sub_query = request.form.get("query")
- results = search_contractors(sub_query)
+ query = request.form.get("query", "").strip()
+ results = search_contractors(query)
if not results:
return "No subcontractor found"
- output = "".join(
- f"{row['Contractor_Name']}"
- for row in results
- )
- return output
+ return "".join(f"{r['Contractor_Name']}" for r in results)
-# ------------------- Get Hold Types -------------------
+# ------------------------------- Get Hold Types -------------------------------
@invoice_bp.route('/get_hold_types', methods=['GET'])
@login_required
+@handle_exception
def get_hold_types():
hold_types = get_all_hold_types()
- LogHelper.log_action("Get hold type", f"User {current_user.id} Get hold type '{hold_types}'")
+ log_action("Get hold type", f"retrieved hold types '{hold_types}'")
return jsonify(hold_types)
-# ------------------- Edit Invoice -------------------
+# ------------------------------- Edit Invoice -------------------------------
@invoice_bp.route('/edit_invoice/', methods=['GET', 'POST'])
@login_required
+@handle_exception
def edit_invoice(invoice_id):
if request.method == 'POST':
data = request.form
update_invoice(data, invoice_id)
update_inpayment(data)
-
- LogHelper.log_action("Edit invoice", f"User {current_user.id} Edit invoice '{invoice_id}'")
+ log_action("Edit invoice", f"edited invoice '{invoice_id}'")
return jsonify({"status": "success", "message": "Invoice updated successfully"}), 200
invoice = get_invoice_by_id(invoice_id)
return render_template('edit_invoice.html', invoice=invoice)
-# ------------------- Delete Invoice -------------------
+# ------------------------------- Delete Invoice -------------------------------
@invoice_bp.route('/delete_invoice/', methods=['GET'])
@login_required
+@handle_exception
def delete_invoice_route(invoice_id):
- try:
- delete_invoice_data(invoice_id, current_user.id)
- LogHelper.log_action("Delete Invoice", f"User {current_user.id} deleted Invoice '{invoice_id}'")
- return jsonify({
- "message": f"Invoice {invoice_id} deleted successfully.",
- "status": "success"
- })
- except Exception as e:
- return jsonify({
- "message": str(e),
- "status": "error"
- }), 500
\ No newline at end of file
+ delete_invoice_data(invoice_id, current_user.id)
+ log_action("Delete invoice", f"deleted invoice '{invoice_id}'")
+ return jsonify({"status": "success", "message": f"Invoice {invoice_id} deleted successfully."})
\ No newline at end of file
diff --git a/controllers/village_controller.py b/controllers/village_controller.py
index 02df862..64f8a65 100644
--- a/controllers/village_controller.py
+++ b/controllers/village_controller.py
@@ -1,8 +1,11 @@
+
+
+
+
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify
from flask_login import login_required
import config
-
from model.Village import Village
from model.State import State
@@ -23,7 +26,6 @@ def add_village():
state = State()
states = state.GetAllStates(request=request)
-
villages = village.GetAllVillages(request=request)
return render_template(
@@ -42,7 +44,6 @@ def get_districts(state_id):
cursor = connection.cursor()
cursor.callproc("GetDistrictByStateID", [state_id])
-
districts = []
for rs in cursor.stored_results():
@@ -51,15 +52,7 @@ def get_districts(state_id):
cursor.close()
connection.close()
- district_list = []
-
- for d in districts:
- district_list.append({
- "id": d[0],
- "name": d[1]
- })
-
- return jsonify(district_list)
+ return jsonify([{"id": d[0], "name": d[1]} for d in districts])
# ------------------------- Fetch Blocks -------------------------
@@ -71,7 +64,6 @@ def get_blocks(district_id):
cursor = connection.cursor()
cursor.callproc("GetBlocksByDistrictID", [district_id])
-
blocks = []
for rs in cursor.stored_results():
@@ -80,22 +72,13 @@ def get_blocks(district_id):
cursor.close()
connection.close()
- block_list = []
-
- for b in blocks:
- block_list.append({
- "id": b[0],
- "name": b[1]
- })
-
- return jsonify(block_list)
+ return jsonify([{"id": b[0], "name": b[1]} for b in blocks])
# ------------------------- Check Village -------------------------
@village_bp.route('/check_village', methods=['POST'])
@login_required
def check_village():
-
village = Village()
return village.CheckVillage(request=request)
@@ -106,14 +89,9 @@ def check_village():
def delete_village(village_id):
village = Village()
-
village.DeleteVillage(request=request, village_id=village_id)
- if not village.isSuccess:
- flash(village.resultMessage, "error")
- else:
- flash(village.resultMessage, "success")
-
+ flash(village.resultMessage, "success" if village.isSuccess else "error")
return redirect(url_for('village.add_village'))
@@ -135,8 +113,8 @@ def edit_village(village_id):
else:
flash(village.resultMessage, "error")
- village_data = village.GetVillageByID(request=request, id=village_id)
- blocks = village.GetAllBlocks(request=request)
+ village_data = village.GetVillageByID(id=village_id) or []
+ blocks = village.GetAllBlocks() or []
return render_template(
'edit_village.html',
@@ -145,23 +123,17 @@ def edit_village(village_id):
)
else:
-
- village_data = village.GetVillageByID(request=request, id=village_id)
+ # ✅ FIXED HERE (removed request)
+ village_data = village.GetVillageByID(id=village_id)
if not village.isSuccess:
flash(village.resultMessage, "error")
return redirect(url_for('village.add_village'))
- blocks = village.GetAllBlocks(request=request)
-
- if village_data is None:
- village_data = []
-
- if blocks is None:
- blocks = []
+ blocks = village.GetAllBlocks() or []
return render_template(
'edit_village.html',
- village_data=village_data,
+ village_data=village_data or [],
blocks=blocks
)
\ No newline at end of file
diff --git a/model/ContractorInfo.py b/model/ContractorInfo.py
index a4eb36b..edfa3cd 100644
--- a/model/ContractorInfo.py
+++ b/model/ContractorInfo.py
@@ -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
diff --git a/model/Invoice.py b/model/Invoice.py
index ac148ac..b002fb3 100644
--- a/model/Invoice.py
+++ b/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()
\ No newline at end of file
+# ------------------- 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)
\ No newline at end of file
diff --git a/model/Village.py b/model/Village.py
index 3b7b0cb..fa5b9bd 100644
--- a/model/Village.py
+++ b/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
\ No newline at end of file
+ finally:
+ connection.close()
\ No newline at end of file