236 lines
8.3 KiB
Python
236 lines
8.3 KiB
Python
from flask import Flask, render_template, request, redirect, url_for, send_from_directory, flash, jsonify, json
|
|
from flask import current_app
|
|
|
|
from datetime import datetime
|
|
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
|
|
|
from AppCode.Utilities import RegEx, ResponseHandler, HtmlHelper
|
|
from AppCode.Log import LogData, LogHelper
|
|
|
|
import os
|
|
import config
|
|
import re
|
|
|
|
import mysql.connector
|
|
from mysql.connector import Error
|
|
|
|
class District:
|
|
|
|
isSuccess = False
|
|
resultMessage = ""
|
|
|
|
def __init__(self):
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
|
|
def AddDistrict(self, request):
|
|
"""Handles the logic for adding a new district."""
|
|
connection = config.get_db_connection()
|
|
if not connection:
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.db_connection_failure(), 500)
|
|
return
|
|
|
|
cursor = connection.cursor()
|
|
|
|
district_name = request.form['district_Name'].strip()
|
|
state_id = request.form['state_Id']
|
|
LogHelper.log_action("Add District", f"User {current_user.id} Added District '{district_name}'")
|
|
|
|
if not re.match(RegEx.patternAlphabetOnly, district_name):
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.invalid_name("district"), 400)
|
|
return
|
|
|
|
try:
|
|
cursor.callproc("GetDistrictByNameAndState", (district_name, state_id))
|
|
for data in cursor.stored_results():
|
|
rs = data.fetchone()
|
|
|
|
if rs:
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.already_exists("district"), 409)
|
|
return
|
|
|
|
cursor.callproc('SaveDistrict', (district_name, state_id))
|
|
connection.commit()
|
|
self.isSuccess = True
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.add_success("district"), 200)
|
|
return
|
|
|
|
except mysql.connector.Error as e:
|
|
print(f"Error inserting district: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.add_failure("district"), 500)
|
|
return
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
|
|
def GetAllDistricts(self, request):
|
|
"""Fetches all districts with their state names."""
|
|
districtdata = []
|
|
connection = config.get_db_connection()
|
|
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
|
|
if not connection:
|
|
return []
|
|
|
|
cursor = connection.cursor()
|
|
|
|
try:
|
|
cursor.callproc("GetAllDistricts")
|
|
for dis in cursor.stored_results():
|
|
districtdata = dis.fetchall()
|
|
self.isSuccess = True
|
|
|
|
except mysql.connector.Error as e:
|
|
print(f"Error fetching districts: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.fetch_failure("districts"), 500)
|
|
return []
|
|
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
return districtdata
|
|
|
|
|
|
def CheckDistrict(self, request):
|
|
"""Checks if a district name already exists within a specific state."""
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
connection = config.get_db_connection()
|
|
|
|
if not connection:
|
|
return HtmlHelper.json_response(ResponseHandler.db_connection_failure(), 500)
|
|
|
|
cursor = connection.cursor()
|
|
|
|
district_name = request.json.get('district_Name', '').strip()
|
|
state_id = request.json.get('state_Id', '')
|
|
LogHelper.log_action("Check District", f"User {current_user.id} Checked District '{district_name}'")
|
|
|
|
if not re.match(RegEx.patternAlphabetOnly, district_name):
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.invalid_name("district"), 400)
|
|
return self.resultMessage
|
|
|
|
try:
|
|
cursor.callproc("GetDistrictByNameAndState", (district_name, state_id,))
|
|
for result in cursor.stored_results():
|
|
existing_district = result.fetchone()
|
|
|
|
if existing_district:
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.already_exists("district"), 409)
|
|
else:
|
|
self.isSuccess = True
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.is_available("district"), 200)
|
|
|
|
return self.resultMessage
|
|
|
|
except mysql.connector.Error as e:
|
|
print(f"Error checking district: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.add_failure("district"), 500)
|
|
return self.resultMessage
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
|
|
def DeleteDistrict(self, request, id):
|
|
"""Deletes a district by its ID."""
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
connection = config.get_db_connection()
|
|
cursor = connection.cursor()
|
|
LogHelper.log_action("Delete District", f"User {current_user.id} Deleted District '{id}'")
|
|
|
|
try:
|
|
cursor.callproc("DeleteDistrict", (id,))
|
|
connection.commit()
|
|
self.resultMessage = "Successfully Deleted"
|
|
self.isSuccess = True
|
|
except mysql.connector.Error as e:
|
|
print(f"Error deleting district: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.delete_failure("district"), 500)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
return self.resultMessage
|
|
|
|
|
|
def EditDistrict(self, request, id):
|
|
"""Handles the POST logic for updating a district."""
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
connection = config.get_db_connection()
|
|
cursor = connection.cursor()
|
|
|
|
district_name = request.form['district_Name'].strip()
|
|
state_id = request.form['state_Id']
|
|
LogHelper.log_action("Edit District", f"User {current_user.id} Edited District '{id}'")
|
|
|
|
# Added validation consistent with your other Edit methods
|
|
if not re.match(RegEx.patternAlphabetOnly, district_name):
|
|
self.isSuccess = False
|
|
self.resultMessage = ResponseHandler.invalid_name("district"), 400
|
|
return self.resultMessage
|
|
|
|
try:
|
|
cursor.callproc("UpdateDistrict", (id, state_id, district_name,))
|
|
connection.commit()
|
|
self.isSuccess = True
|
|
self.resultMessage = "Successfully Edited"
|
|
except mysql.connector.Error as e:
|
|
print(f"Error updating district: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = ResponseHandler.update_failure("district"), 500
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
return self.resultMessage
|
|
|
|
|
|
def GetDistrictByID(self, request, id):
|
|
"""Fetches a single district's details by its ID."""
|
|
districtdata = None
|
|
self.isSuccess = False
|
|
self.resultMessage = ""
|
|
connection = config.get_db_connection()
|
|
|
|
if not connection:
|
|
return None
|
|
|
|
cursor = connection.cursor()
|
|
|
|
try:
|
|
cursor.callproc("GetDistrictDataByID", (id,))
|
|
for rs in cursor.stored_results():
|
|
districtdata = rs.fetchone()
|
|
|
|
if districtdata:
|
|
self.isSuccess = True
|
|
self.resultMessage = "Success in Fetching"
|
|
else:
|
|
self.isSuccess = False
|
|
self.resultMessage = "District Not Found"
|
|
|
|
except mysql.connector.Error as e:
|
|
print(f"Error fetching district data: {e}")
|
|
self.isSuccess = False
|
|
self.resultMessage = HtmlHelper.json_response(ResponseHandler.fetch_failure("district"), 500)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
return districtdata |