import os from typing import List from fastapi import HTTPException from openpyxl import load_workbook from sqlalchemy.orm import Session from app.api.rent_expenses import schema from app.dependency.authantication import JWTPayloadSchema from app.locale.messages import Messages from app.models.main.rent_expenses import RentExpensesBase, TblRentExpenses from app.models.main.store_formate import TblStoreFormat from app.utils.schemas_utils import CustomResponse class RentExpensesService: def __init__(self, db:Session, token:JWTPayloadSchema): self.db = db self.token = token async def create_rent_expenses(self, request:schema.RentExpensesCreate): created_rent_expenses = RentExpensesBase.model_validate(request.model_dump()) TblRentExpenses.create_rent_expenses(created_rent_expenses, self.db) self.db.commit() EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" # Ensure output folder exists os.makedirs("uploaded_files", exist_ok=True) # 🔹 Load workbook (prefer latest saved version) if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) else: workbook = load_workbook(EXCEL_PATH) # 🔹 Mapping between Store Format and Excel Sheet format_to_sheet = { "Store Format A": "Assu Sum Mod A", "Store Format B": "Assu Sum Mod B", } # 🔹 Update based on selected store format for format_type, sheet_name in format_to_sheet.items(): if request.store_format_type == format_type: sheet = workbook[sheet_name] sheet["I12"] = created_rent_expenses.rent_per_month sheet["I23"] = created_rent_expenses.rental_advance_amount sheet["I24"] = created_rent_expenses.rental_advance_period # print(f"✅ Data written to {sheet_name}") # 🔹 Save changes workbook.save(OUTPUT_PATH) workbook.close() print(f"✅ Excel updated successfully at: {OUTPUT_PATH}") return CustomResponse(status="1", message=Messages.RENT_EXPENSES) async def get_rent_expenses(self, rent_get_id:int): new_get_rent_expenses = TblRentExpenses.get_rent_expenses(rent_get_id, self.db) if not new_get_rent_expenses: raise HTTPException(status_code=404, detail="Get Rent Expenses ID not found") return schema.RentExpensesResponse.model_validate(new_get_rent_expenses) async def get_group_rent_expenses(self, group_id:int) ->"TblRentExpenses": new_get_group_rent_expenses = self.db.query(TblRentExpenses).filter(TblRentExpenses.group_id == group_id).all() if not new_get_group_rent_expenses: raise HTTPException(status_code=404, detail="Get group ID not found") return [schema.RentExpensesResponse.model_validate(get_group) for get_group in new_get_group_rent_expenses] # async def post_store_size(self) -> TblRentExpenses: # # Fetch the record # new_post_store_size = ( # self.db.query(TblRentExpenses, TblStoreFormat) # .filter(TblRentExpenses.group_id == TblStoreFormat.group_id) # .first() # ) # if not new_post_store_size: # raise HTTPException(status_code=404, detail="No matching records found") # rent_expense, store_format = new_post_store_size # # Update instance value # rent_expense.store_size = store_format.store_size # self.db.commit() # self.db.refresh(rent_expense) # return rent_expense async def post_store_size(self) ->"TblRentExpenses": new_post_store_size = self.db.query(TblRentExpenses,TblStoreFormat).filter(TblRentExpenses.group_id == TblStoreFormat.group_id).all() if not new_post_store_size: raise HTTPException(status_code=404, detail="Data not found") for rent_expense, store_format in new_post_store_size: rent_expense.store_size = store_format.store_size self.db.commit() return [data for data, _ in new_post_store_size] # async def post_store_size(self) -> list[dict]: # new_post_store_size = ( # self.db.query(TblRentExpenses, TblStoreFormat) # .filter(TblRentExpenses.group_id == TblStoreFormat.group_id) # .all() # ) # if not new_post_store_size: # raise HTTPException(status_code=404, detail="Data not found") # for rent_expense, store_format in new_post_store_size: # rent_expense.store_size = store_format.store_size # self.db.commit() # return [ # {"rent_expense": rent_expense, "store_format": store_format} # for rent_expense, store_format in new_post_store_size # ] async def update_rent_expenses(self, request:List[schema.RentExpensesUpdate]): for req in request: updated_rent_expenses = RentExpensesBase.model_validate(req.model_dump()) if updated_rent_expenses.rent_id is None: return CustomResponse(status="-1", message=Messages.RENT_EXPENSES_NOT) TblRentExpenses.update_rent_expenses(updated_rent_expenses.rent_id, updated_rent_expenses, self.db) self.db.commit() EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" # Ensure output folder exists os.makedirs("uploaded_files", exist_ok=True) # 🔹 Load workbook (prefer latest saved version) if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) else: workbook = load_workbook(EXCEL_PATH) # 🔹 Mapping between Store Format and Excel Sheet format_to_sheet = { "Store Format A": "Assu Sum Mod A", "Store Format B": "Assu Sum Mod B", } # 🔹 Update based on selected store format for format_type, sheet_name in format_to_sheet.items(): if req.store_format_type == format_type: sheet = workbook[sheet_name] sheet["I12"] = updated_rent_expenses.rent_per_month sheet["I23"] = updated_rent_expenses.rental_advance_amount sheet["I24"] = updated_rent_expenses.rental_advance_period # print(f"✅ Data written to {sheet_name}") # 🔹 Save changes workbook.save(OUTPUT_PATH) workbook.close() print(f"✅ Excel updated successfully at: {OUTPUT_PATH}") return CustomResponse(status="1", message=Messages.RENT_EXPENSES_UPDATE) async def delete_rent_expenses(self, rent_id:int): deleted_rent_expenses = TblRentExpenses.delete_rent_expenses(rent_id, self.db) if not deleted_rent_expenses: return CustomResponse(status="-1", message=Messages.RENT_EXPENSES_NOT) return CustomResponse(status="1", message=Messages.RENT_EXPENSES_DELETE)