From dffa9769bc9890507e3c5b1c62bd1bb740d84514 Mon Sep 17 00:00:00 2001 From: anishd100 Date: Mon, 23 Mar 2026 13:09:34 +0530 Subject: [PATCH] Village, invoice and Contractor Info model and controller added --- controllers/invoice_controller.py | 101 +++---- controllers/village_controller.py | 54 +--- model/ContractorInfo.py | 95 +++---- model/Invoice.py | 458 ++++++++++-------------------- model/Village.py | 200 ++++++++----- 5 files changed, 392 insertions(+), 516 deletions(-) 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