129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
from AppCode.Config import DBConfig
|
|
|
|
|
|
class MatCreditHandler:
|
|
|
|
def __init__(self):
|
|
self.conn = DBConfig.get_db_connection()
|
|
self.cursor = self.conn.cursor(dictionary=True)
|
|
|
|
# --------------------------------------------------
|
|
# FETCH ALL MAT CREDIT + UTILIZATION (For UI Display)
|
|
# --------------------------------------------------
|
|
def fetch_all(self):
|
|
try:
|
|
|
|
self.cursor.callproc("GetMatCedit")
|
|
result_sets = self.cursor.stored_results()
|
|
mat_rows = next(result_sets).fetchall()
|
|
utilization_rows = next(result_sets).fetchall()
|
|
|
|
return mat_rows, utilization_rows
|
|
finally:
|
|
self.cursor.close()
|
|
self.conn.close()
|
|
|
|
# --------------------------------------------------
|
|
# SAVE / UPDATE SINGLE MAT ROW (FROM MANUAL UI)
|
|
# --------------------------------------------------
|
|
@staticmethod
|
|
def save_single(data):
|
|
conn = DBConfig.get_db_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
try:
|
|
# Save / Update MAT Credit
|
|
cur.callproc(
|
|
"SaveOrUpdateMatCredit",
|
|
(
|
|
data["financial_year"],
|
|
data["mat_credit"],
|
|
data["balance"],
|
|
data.get("remarks", "")
|
|
)
|
|
)
|
|
|
|
mat_id = None
|
|
for result in cur.stored_results():
|
|
mat_id = result.fetchone()["mat_id"]
|
|
|
|
# Save utilization rows
|
|
for u in data.get("utilization", []):
|
|
if float(u["amount"]) > 0:
|
|
cur.callproc(
|
|
"InsertMatUtilization",
|
|
(mat_id, u["year"], u["amount"])
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
# --------------------------------------------------
|
|
# AUTO SAVE MAT FROM ITR (MAIN LOGIC)
|
|
# --------------------------------------------------
|
|
@staticmethod
|
|
def save_from_itr(year, mat_created, mat_utilized, remarks="Auto from"):
|
|
conn = DBConfig.get_db_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
try:
|
|
mat_created = float(mat_created or 0)
|
|
mat_utilized = float(mat_utilized or 0)
|
|
|
|
balance = mat_created - mat_utilized
|
|
|
|
# Save / Update MAT Credit
|
|
cur.callproc(
|
|
"SaveOrUpdateMatCredit",
|
|
(year, mat_created, balance, remarks)
|
|
)
|
|
|
|
mat_id = None
|
|
for result in cur.stored_results():
|
|
mat_id = result.fetchone()["mat_id"]
|
|
|
|
# Save utilization only if used
|
|
if mat_utilized > 0:
|
|
cur.callproc(
|
|
"InsertMatUtilization",
|
|
(mat_id, year, mat_utilized)
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
# --------------------------------------------------
|
|
# DELETE MAT CREDIT SAFELY (OPTIONAL)
|
|
# --------------------------------------------------
|
|
def delete_by_year(self, financial_year):
|
|
try:
|
|
self.cursor.execute(
|
|
"DELETE FROM mat_credit WHERE financial_year=%s",
|
|
(financial_year,)
|
|
)
|
|
self.conn.commit()
|
|
finally:
|
|
self.cursor.close()
|
|
self.conn.close()
|
|
|
|
# --------------------------------------------------
|
|
# CLOSE CONNECTION (MANUAL USE)
|
|
# --------------------------------------------------
|
|
def close(self):
|
|
if self.cursor:
|
|
self.cursor.close()
|
|
if self.conn:
|
|
self.conn.close()
|