import os from typing import List from openpyxl import load_workbook from sqlalchemy.orm import Session from fastapi import HTTPException from app.api.centralised_and_direct import schema from app.models.main.centralised_and_direct import CentralisedAndDirectBase, TblCentralisedAndDirect from app.dependency.authantication import JWTPayloadSchema from app.utils.schemas_utils import CustomResponse from app.locale.messages import Messages class CentralisedAndDirectService: def __init__(self, db: Session, token: JWTPayloadSchema): self.db = db self.token = token async def create_centralised_and_direct(self, request: schema.DistributionCreate): if request.distribution_type == "Centralised Warehouse Distribution": if not request.centralised: raise HTTPException(status_code=422, detail="Centralised data required") # data = CentralisedAndDirectBase.model_validate(request.centralised) new_data = TblCentralisedAndDirect.create_centralised_and_direct( data=request.centralised, db=self.db, distribution_type=request.distribution_type ) elif request.distribution_type == "Direct Store Delivery Distribution": if not request.direct: raise HTTPException(status_code=422, detail="Direct data required") # data = CentralisedAndDirectBase.model_validate(request.direct) new_data = TblCentralisedAndDirect.create_centralised_and_direct( data=request.direct, db=self.db, distribution_type=request.distribution_type ) else: raise HTTPException(status_code=400, detail="Invalid distribution type") self.db.commit() self.db.refresh(new_data) EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" os.makedirs("uploaded_files", exist_ok=True) workbook = load_workbook(EXCEL_PATH) if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) # print("📂 Loaded existing output file to keep previous data.") else: workbook = load_workbook(EXCEL_PATH) # print("🆕 Loaded clean template (first run).") sheet_a = workbook["Assu Sum Mod A"] sheet_a["K13"] = new_data.employee_cost_per_head sheet_a["K29"] = new_data.number_of_employees sheet_b = workbook["Assu Sum Mod B"] sheet_b["K13"] = new_data.employee_cost_per_head sheet_b["K29"] = new_data.number_of_employees sheet_c = workbook["Warehouse (Direct Store)"] sheet_c["B7"] = new_data.employees_in_charge_at_store_level sheet_d = workbook["Warehouse (Direct Store)"] sheet_d["C7"] = new_data.salary_of_each_employee sheet_e = workbook["Warehouse (Central)"] sheet_e["B6"] = new_data.expected_number_of_warehouse_increase_every_year sheet_f = workbook["Warehouse (Central)"] sheet_f["B8"] = new_data.expected_number_of_warehouse sheet_g = workbook["Warehouse (Central)"] sheet_g["C8"] = new_data.rental_expenses sheet_h = workbook["Warehouse (Central)"] sheet_h["D10"] = new_data.utilities_cost sheet_i = workbook["Warehouse (Central)"] sheet_i["D11"] = new_data.transportation_cost sheet_j = workbook["Warehouse (Central)"] sheet_j["B13"] = new_data.number_of_employees sheet_k = workbook["Warehouse (Central)"] sheet_k["C13"] = new_data.employee_cost_per_head workbook.save(OUTPUT_PATH) workbook.close() print("✅ Excel updated successfully at:", OUTPUT_PATH) return CustomResponse(status="1", message=Messages.CENTRALISED_AND_DIRECT) # return {"message": f"{request.distribution_type} Created Successfully"} async def get_centralised_and_direct(self, group_id: int): new_get_centralised_and_direct = TblCentralisedAndDirect.get_centralised_and_direct(group_id, self.db) if not new_get_centralised_and_direct: raise HTTPException(status_code=404, detail="Centralised is not found") return [schema.CentralisedAndDirectResponse.model_validate(get_group) for get_group in new_get_centralised_and_direct] async def update_centralised_and_direct(self, request:List[schema.CentralisedAndDirectResponse]): for req in request: updated_centralised_and_direct = CentralisedAndDirectBase.model_validate(req.model_dump()) if updated_centralised_and_direct.centralised_direct_id is None: return CustomResponse(status="-1", message=Messages.CENTRALISED_AND_DIRECT_NOT) TblCentralisedAndDirect.update_centralised_and_direct(updated_centralised_and_direct.centralised_direct_id, updated_centralised_and_direct, 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" os.makedirs("uploaded_files", exist_ok=True) workbook = load_workbook(EXCEL_PATH) if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) # print("📂 Loaded existing output file to keep previous data.") else: workbook = load_workbook(EXCEL_PATH) # print("🆕 Loaded clean template (first run).") sheet_a = workbook["Assu Sum Mod A"] sheet_a["K13"] = updated_centralised_and_direct.employee_cost_per_head sheet_a["K29"] = updated_centralised_and_direct.number_of_employees sheet_b = workbook["Assu Sum Mod B"] sheet_b["K13"] = updated_centralised_and_direct.employee_cost_per_head sheet_b["K29"] = updated_centralised_and_direct.number_of_employees sheet_c = workbook["Warehouse (Direct Store)"] sheet_c["B7"] = updated_centralised_and_direct.employees_in_charge_at_store_level sheet_d = workbook["Warehouse (Direct Store)"] sheet_d["C7"] = updated_centralised_and_direct.salary_of_each_employee sheet_e = workbook["Warehouse (Central)"] sheet_e["B6"] = updated_centralised_and_direct.expected_number_of_warehouse_increase_every_year sheet_f = workbook["Warehouse (Central)"] sheet_f["B8"] = updated_centralised_and_direct.expected_number_of_warehouse sheet_g = workbook["Warehouse (Central)"] sheet_g["C8"] = updated_centralised_and_direct.rental_expenses sheet_h = workbook["Warehouse (Central)"] sheet_h["D10"] = updated_centralised_and_direct.utilities_cost sheet_i = workbook["Warehouse (Central)"] sheet_i["D11"] = updated_centralised_and_direct.transportation_cost sheet_j = workbook["Warehouse (Central)"] sheet_j["B13"] = updated_centralised_and_direct.number_of_employees sheet_k = workbook["Warehouse (Central)"] sheet_k["C13"] = updated_centralised_and_direct.employee_cost_per_head workbook.save(OUTPUT_PATH) workbook.close() print("✅ Excel updated successfully at:", OUTPUT_PATH) return CustomResponse(status="1", message=Messages.CENTRALISED_AND_DIRECT_UPDATE) async def delete_centralised_and_direct(self, centralised_direct_id:int): deleted_centralised_and_direct = TblCentralisedAndDirect.delete_centralised_and_direct(centralised_direct_id, self.db) if not deleted_centralised_and_direct: return CustomResponse(status="-1", message=Messages.CENTRALISED_AND_DIRECT_NOT) return CustomResponse(status="1", message=Messages.CENTRALISED_AND_DIRECT_DELETE)