import os from typing import List from openpyxl import load_workbook from sqlalchemy.orm import Session from app.api.physical_store_formats import schema from app.dependency.authantication import JWTPayloadSchema from app.models.main.physical_store_formats import PhysicalStoreFormatsBase, TblPhysicalStoreFormats from app.utils.schemas_utils import CustomResponse from app.locale.messages import Messages from fastapi import HTTPException class PhysicalStoreFormatsService: def __init__(self, db:Session, token:JWTPayloadSchema): self.db = db self.token = token async def created(self, request: List[schema.PhysicalStoreFormatsCreate]): 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).") for req in request: created_data = PhysicalStoreFormatsBase.model_validate(req.model_dump()) TblPhysicalStoreFormats.create(created_data, self.db) self.db.commit() if req.metric == "Investment & Capital Expenses": sheet_a = workbook["Plan Working A"] sheet_a["F15"] = created_data.y1_y2 sheet_a["H15"] = created_data.y3_y4 sheet_a["J15"] = created_data.y5_y6 sheet_b = workbook["Plan Working B"] sheet_b["F15"] = created_data.y1_y2 sheet_b["H15"] = created_data.y3_y4 sheet_b["J15"] = created_data.y5_y6 elif req.metric == "Pre-Operating Expenses": sheet_c = workbook["Plan Working A"] sheet_c["F39"] = created_data.y1_y2 sheet_c["H39"] = created_data.y3_y4 sheet_c["J39"] = created_data.y5_y6 sheet_d = workbook["Plan Working B"] sheet_d["F39"] = created_data.y1_y2 sheet_d["H39"] = created_data.y3_y4 sheet_d["J39"] = created_data.y5_y6 elif req.metric == "Rental Advance (Deposit)": sheet_e = workbook["Plan Working A"] sheet_e["F48"] = created_data.y1_y2 sheet_e["H48"] = created_data.y3_y4 sheet_e["J48"] = created_data.y5_y6 sheet_f = workbook["Plan Working B"] sheet_f["F48"] = created_data.y1_y2 sheet_f["H48"] = created_data.y3_y4 sheet_f["J48"] = created_data.y5_y6 elif req.metric == "Rent": sheet_g = workbook["Plan Working A"] sheet_g["F142"] = created_data.y1_y2 sheet_g["H142"] = created_data.y3_y4 sheet_g["J142"] = created_data.y5_y6 sheet_h = workbook["Plan Working B"] sheet_h["F142"] = created_data.y1_y2 sheet_h["H142"] = created_data.y3_y4 sheet_h["J142"] = created_data.y5_y6 elif req.metric == "Bills per month": sheet_i = workbook["Plan Working A"] sheet_i["F66"] = created_data.y1_y2 sheet_i["H66"] = created_data.y3_y4 sheet_i["J66"] = created_data.y5_y6 sheet_j = workbook["Plan Working B"] sheet_j["F66"] = created_data.y1_y2 sheet_j["H66"] = created_data.y3_y4 sheet_j["J66"] = created_data.y5_y6 elif req.metric == "Average Price per item": sheet_k = workbook["Plan Working A"] sheet_k["F70"] = created_data.y1_y2 sheet_k["H70"] = created_data.y3_y4 sheet_k["J70"] = created_data.y5_y6 sheet_l = workbook["Plan Working B"] sheet_l["F70"] = created_data.y1_y2 sheet_l["H70"] = created_data.y3_y4 sheet_l["J70"] = created_data.y5_y6 elif req.metric == "Other Income": sheet_m = workbook["Plan Working A"] sheet_m["F111"] = created_data.y1_y2 sheet_m["H111"] = created_data.y3_y4 sheet_m["J111"] = created_data.y5_y6 sheet_n = workbook["Plan Working B"] sheet_n["F111"] = created_data.y1_y2 sheet_n["H111"] = created_data.y3_y4 sheet_n["J111"] = created_data.y5_y6 elif req.metric == "Utilities": sheet_o = workbook["Plan Working A"] sheet_o["F184"] = created_data.y1_y2 sheet_o["H184"] = created_data.y3_y4 sheet_o["J184"] = created_data.y5_y6 sheet_p = workbook["Plan Working B"] sheet_p["F184"] = created_data.y1_y2 sheet_p["H184"] = created_data.y3_y4 sheet_p["J184"] = created_data.y5_y6 elif req.metric == "Logistics and Transportation": sheet_q = workbook["Plan Working Online"] sheet_q["F86"] = created_data.y1_y2 sheet_q["H86"] = created_data.y3_y4 sheet_q["J86"] = created_data.y5_y6 elif req.metric == "Corporate Employee Salary": sheet_r = workbook["Org Structure"] sheet_r["F86"] = created_data.y1_y2 sheet_r["H16"] = created_data.y3_y4 sheet_r["J86"] = created_data.y5_y6 elif req.metric == "Concept Employee Salary": sheet_s = workbook["Org Structure"] sheet_s["F86"] = created_data.y1_y2 sheet_s["H16"] = created_data.y3_y4 sheet_s["J86"] = created_data.y5_y6 elif req.metric == "Store Level Employee Salary": sheet_t = workbook["Org Structure"] sheet_t["F86"] = created_data.y1_y2 sheet_t["H16"] = created_data.y3_y4 sheet_t["J86"] = created_data.y5_y6 elif req.metric == "Store Level Employee Salary": sheet_t = workbook["Org Structure"] sheet_t["F86"] = created_data.y1_y2 sheet_t["H16"] = created_data.y3_y4 sheet_t["J86"] = created_data.y5_y6 elif req.metric == "Store Level Employee Salary": sheet_t = workbook["Org Structure"] sheet_t["F86"] = created_data.y1_y2 sheet_t["H16"] = created_data.y3_y4 sheet_t["J86"] = created_data.y5_y6 workbook.save(OUTPUT_PATH) workbook.close() print("✅ Excel updated successfully at:", OUTPUT_PATH) return CustomResponse(status="1", message=Messages.PHYSICAL_STORE_FORMATS) async def geted(self, group_id:int): geted_data= TblPhysicalStoreFormats.get(group_id, self.db) if not geted_data: raise HTTPException(status_code=404, detail="Get group ID not found") return [schema.PhysicalStoreFormatsResponse.model_validate(get_group) for get_group in geted_data] # async def updated(self, request:List[schema.PhysicalStoreFormatsUpdate]): # EXCEL_PATH = "files/Simulation_Test_file_clean.xlsm" # OUTPUT_PATH = "uploaded_files/Simulation_Test_file_output.xlsm" # 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).") # for req in request: # updated = PhysicalStoreFormatsBase.model_validate(req.model_dump()) # if updated.physical_id is None: # return CustomResponse(status="-1", message=Messages.PHYSICAL_STORE_FORMATS_NOT) # TblPhysicalStoreFormats.update(updated.physical_id, updated, self.db) # self.db.commit() # self.db.refresh(updated) # metric = req.metric # if req.metric == "Investment & Capital Expenses": # sheet_a = workbook["Plan Working A"] # sheet_a["F15"] = updated.y1_y2 # sheet_a["H15"] = updated.y3_y4 # sheet_a["J15"] = updated.y5_y6 # sheet_b = workbook["Plan Working B"] # sheet_b["F15"] = updated.y1_y2 # sheet_b["H15"] = updated.y3_y4 # sheet_b["J15"] = updated.y5_y6 # elif req.metric == "Pre-Operating Expenses": # sheet_c = workbook["Plan Working A"] # sheet_c["F39"] = updated.y1_y2 # sheet_c["H39"] = updated.y3_y4 # sheet_c["J39"] = updated.y5_y6 # sheet_d = workbook["Plan Working B"] # sheet_d["F39"] = updated.y1_y2 # sheet_d["H39"] = updated.y3_y4 # sheet_d["J39"] = updated.y5_y6 # elif req.metric == "Rental Advance (Deposit)": # sheet_e = workbook["Plan Working A"] # sheet_e["F48"] = updated.y1_y2 # sheet_e["H48"] = updated.y3_y4 # sheet_e["J48"] = updated.y5_y6 # sheet_f = workbook["Plan Working B"] # sheet_f["F48"] = updated.y1_y2 # sheet_f["H48"] = updated.y3_y4 # sheet_f["J48"] = updated.y5_y6 # elif req.metric == "Rent": # sheet_g = workbook["Plan Working A"] # sheet_g["F142"] = updated.y1_y2 # sheet_g["H142"] = updated.y3_y4 # sheet_g["J142"] = updated.y5_y6 # sheet_h = workbook["Plan Working B"] # sheet_h["F142"] = updated.y1_y2 # sheet_h["H142"] = updated.y3_y4 # sheet_h["J142"] = updated.y5_y6 # elif req.metric == "Bills per month": # sheet_i = workbook["Plan Working A"] # sheet_i["F66"] = updated.y1_y2 # sheet_i["H66"] = updated.y3_y4 # sheet_i["J66"] = updated.y5_y6 # sheet_j = workbook["Plan Working B"] # sheet_j["F66"] = updated.y1_y2 # sheet_j["H66"] = updated.y3_y4 # sheet_j["J66"] = updated.y5_y6 # elif req.metric == "Average Price per item": # sheet_k = workbook["Plan Working A"] # sheet_k["F70"] = updated.y1_y2 # sheet_k["H70"] = updated.y3_y4 # sheet_k["J70"] = updated.y5_y6 # sheet_l = workbook["Plan Working B"] # sheet_l["F70"] = updated.y1_y2 # sheet_l["H70"] = updated.y3_y4 # sheet_l["J70"] = updated.y5_y6 # elif req.metric == "Other Income": # sheet_m = workbook["Plan Working A"] # sheet_m["F111"] = updated.y1_y2 # sheet_m["H111"] = updated.y3_y4 # sheet_m["J111"] = updated.y5_y6 # sheet_n = workbook["Plan Working B"] # sheet_n["F111"] = updated.y1_y2 # sheet_n["H111"] = updated.y3_y4 # sheet_n["J111"] = updated.y5_y6 # elif req.metric == "Utilities": # sheet_o = workbook["Plan Working A"] # sheet_o["F184"] = updated.y1_y2 # sheet_o["H184"] = updated.y3_y4 # sheet_o["J184"] = updated.y5_y6 # sheet_p = workbook["Plan Working B"] # sheet_p["F184"] = updated.y1_y2 # sheet_p["H184"] = updated.y3_y4 # sheet_p["J184"] = updated.y5_y6 # elif req.metric == "Logistics and Transportation": # sheet_q = workbook["Plan Working Online"] # sheet_q["F86"] = updated.y1_y2 # sheet_q["H86"] = updated.y3_y4 # sheet_q["J86"] = updated.y5_y6 # elif req.metric == "Corporate Employee Salary": # sheet_r = workbook["Org Structure"] # sheet_r["F86"] = updated.y1_y2 # sheet_r["H16"] = updated.y3_y4 # sheet_r["J86"] = updated.y5_y6 # elif req.metric == "Concept Employee Salary": # sheet_s = workbook["Org Structure"] # sheet_s["F86"] = updated.y1_y2 # sheet_s["H16"] = updated.y3_y4 # sheet_s["J86"] = updated.y5_y6 # elif req.metric == "Store Level Employee Salary": # sheet_t = workbook["Org Structure"] # sheet_t["F86"] = updated.y1_y2 # sheet_t["H16"] = updated.y3_y4 # sheet_t["J86"] = updated.y5_y6 # workbook.save(OUTPUT_PATH) # workbook.close() # print("✅ Excel updated successfully at:", OUTPUT_PATH) # return CustomResponse(status="1", message=Messages.PHYSICAL_STORE_FORMATS_UPDATE) async def updated(self, request: List[schema.PhysicalStoreFormatsUpdate]): 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) # Load Excel if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) else: workbook = load_workbook(EXCEL_PATH) for req in request: validated = PhysicalStoreFormatsBase.model_validate(req.model_dump()) if validated.physical_id is None: return CustomResponse(status="-1", message=Messages.PHYSICAL_STORE_FORMATS_NOT) orm_obj = TblPhysicalStoreFormats.update( validated.physical_id, validated, self.db ) self.db.commit() self.db.refresh(orm_obj) metric = req.metric if metric == "Investment & Capital Expenses": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F15"] = validated.y1_y2 ws["H15"] = validated.y3_y4 ws["J15"] = validated.y5_y6 elif metric == "Pre-Operating Expenses": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F39"] = validated.y1_y2 ws["H39"] = validated.y3_y4 ws["J39"] = validated.y5_y6 elif metric == "Rental Advance (Deposit)": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F48"] = validated.y1_y2 ws["H48"] = validated.y3_y4 ws["J48"] = validated.y5_y6 elif metric == "Rent": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F142"] = validated.y1_y2 ws["H142"] = validated.y3_y4 ws["J142"] = validated.y5_y6 elif metric == "Bills per month": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F66"] = validated.y1_y2 ws["H66"] = validated.y3_y4 ws["J66"] = validated.y5_y6 elif metric == "Average Price per item": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F70"] = validated.y1_y2 ws["H70"] = validated.y3_y4 ws["J70"] = validated.y5_y6 elif metric == "Other Income": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F111"] = validated.y1_y2 ws["H111"] = validated.y3_y4 ws["J111"] = validated.y5_y6 elif metric == "Utilities": for sheet_name in ["Plan Working A", "Plan Working B"]: ws = workbook[sheet_name] ws["F184"] = validated.y1_y2 ws["H184"] = validated.y3_y4 ws["J184"] = validated.y5_y6 elif metric == "Logistics and Transportation": ws = workbook["Plan Working Online"] ws["F86"] = validated.y1_y2 ws["H86"] = validated.y3_y4 ws["J86"] = validated.y5_y6 elif metric in ( "Corporate Employee Salary", "Concept Employee Salary", "Store Level Employee Salary" ): ws = workbook["Org Structure"] ws["F86"] = validated.y1_y2 ws["H16"] = validated.y3_y4 ws["J86"] = validated.y5_y6 # Save Excel workbook.save(OUTPUT_PATH) workbook.close() return CustomResponse(status="1", message=Messages.PHYSICAL_STORE_FORMATS_UPDATE) async def deleted(self, physical_id:int): deleted_data = TblPhysicalStoreFormats.delete(physical_id, self.db) if not deleted_data: return CustomResponse(status="-1", message=Messages.PHYSICAL_STORE_FORMATS_NOT) return CustomResponse(status="1", message=Messages.PHYSICAL_STORE_FORMATS_DELETE)