import os from typing import List from fastapi import HTTPException from openpyxl import load_workbook from app.api.network.schemas import NetworkPlanningCreate, NetworkPlanningResponse, NetworkPlanningUpdate from app.dependency.authantication import JWTPayloadSchema from app.locale.messages import Messages from app.models.main.network_plan import NetworkPlanningBase, TblNetworkPlanning from app.utils.common_utils import calculate_total_store_counts, calculate_total_store_counts_update from sqlalchemy.orm import Session from app.utils.schemas_utils import CustomResponse class NetworkPlanningService: def __init__(self, db: Session, token: JWTPayloadSchema): self.db = db self.token = token # async def create_network_plan(self, request: NetworkPlanningCreate): # store_counts = calculate_total_store_counts(request) # new_data = NetworkPlanningBase(**request.model_dump(),**store_counts) # TblNetworkPlanning.create(new_data, self.db) # self.db.commit() # return CustomResponse(status="1", message=Messages.NETWORK_CREAT) async def create_network_plan(self, request: List[NetworkPlanningCreate]): 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 workbook once if os.path.exists(OUTPUT_PATH): workbook = load_workbook(OUTPUT_PATH) else: workbook = load_workbook(EXCEL_PATH) sheet = workbook["Rollout Plan"] for req in request: store_counts = calculate_total_store_counts(req) new_data = NetworkPlanningBase(**req.model_dump(), **store_counts) test = TblNetworkPlanning.create(new_data, self.db) self.db.commit() if test.network_id <= 9: # Dynamic row calculation (C6 → C14) row = 5 + test.network_id # Write values sheet[f"C{row}"] = new_data.location sheet[f"D{row}"] = new_data.format_a_y1 sheet[f"E{row}"] = new_data.format_a_y2 sheet[f"F{row}"] = new_data.format_a_y3 sheet[f"G{row}"] = new_data.format_a_y4 sheet[f"H{row}"] = new_data.format_a_y5 sheet[f"K{row}"] = new_data.format_b_y1 sheet[f"L{row}"] = new_data.format_b_y2 sheet[f"M{row}"] = new_data.format_b_y3 sheet[f"N{row}"] = new_data.format_b_y4 sheet[f"O{row}"] = new_data.format_b_y5 # ✅ Save and close once at the end workbook.save(OUTPUT_PATH) workbook.close() print("✅ Excel updated successfully at:", OUTPUT_PATH) return CustomResponse(status="1", message=Messages.NETWORK_CREAT) # async def create_network_plan(self, request: List[NetworkPlanningCreate]): # for req in request: # store_counts = calculate_total_store_counts(req) # new_data = NetworkPlanningBase(**req.model_dump(),**store_counts) # test = TblNetworkPlanning.create(new_data, self.db) # self.db.commit() # 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).") # if test.network_id <= 9: # if test.network_id == 1: # sheet_a = workbook["Rollout Plan"] # sheet_a["C6"] = new_data.location # sheet_a["D6"] = new_data.format_a_y1 # sheet_a["E6"] = new_data.format_a_y2 # sheet_a["F6"] = new_data.format_a_y3 # sheet_a["G6"] = new_data.format_a_y4 # sheet_a["H6"] = new_data.format_a_y5 # sheet_a["K6"] = new_data.format_b_y1 # sheet_a["L6"] = new_data.format_b_y2 # sheet_a["M6"] = new_data.format_b_y3 # sheet_a["N6"] = new_data.format_b_y4 # sheet_a["O6"] = new_data.format_b_y5 # elif test.network_id == 2: # sheet_b = workbook["Rollout Plan"] # sheet_b["C7"] = new_data.location # sheet_b["D7"] = new_data.format_a_y1 # sheet_b["E7"] = new_data.format_a_y2 # sheet_b["F7"] = new_data.format_a_y3 # sheet_b["G7"] = new_data.format_a_y4 # sheet_b["H7"] = new_data.format_a_y5 # sheet_b["K7"] = new_data.format_b_y1 # sheet_b["L7"] = new_data.format_b_y2 # sheet_b["M7"] = new_data.format_b_y3 # sheet_b["N7"] = new_data.format_b_y4 # sheet_b["O7"] = new_data.format_b_y5 # elif test.network_id == 3: # sheet_c = workbook["Rollout Plan"] # sheet_c["C8"] = new_data.location # sheet_c["D8"] = new_data.format_a_y1 # sheet_c["E8"] = new_data.format_a_y2 # sheet_c["F8"] = new_data.format_a_y3 # sheet_c["G8"] = new_data.format_a_y4 # sheet_c["H8"] = new_data.format_a_y5 # sheet_c["K8"] = new_data.format_b_y1 # sheet_c["L8"] = new_data.format_b_y2 # sheet_c["M8"] = new_data.format_b_y3 # sheet_c["N8"] = new_data.format_b_y4 # sheet_c["O8"] = new_data.format_b_y5 # elif test.network_id == 4: # sheet_d = workbook["Rollout Plan"] # sheet_d["C9"] = new_data.location # sheet_d["D9"] = new_data.format_a_y1 # sheet_d["E9"] = new_data.format_a_y2 # sheet_d["F9"] = new_data.format_a_y3 # sheet_d["G9"] = new_data.format_a_y4 # sheet_d["H9"] = new_data.format_a_y5 # sheet_d["K9"] = new_data.format_b_y1 # sheet_d["L9"] = new_data.format_b_y2 # sheet_d["M9"] = new_data.format_b_y3 # sheet_d["N9"] = new_data.format_b_y4 # sheet_d["O9"] = new_data.format_b_y5 # elif test.network_id == 5: # sheet_e = workbook["Rollout Plan"] # sheet_e["C10"] = new_data.location # sheet_e["D10"] = new_data.format_a_y1 # sheet_e["E10"] = new_data.format_a_y2 # sheet_e["F10"] = new_data.format_a_y3 # sheet_e["G10"] = new_data.format_a_y4 # sheet_e["H10"] = new_data.format_a_y5 # sheet_e["K10"] = new_data.format_b_y1 # sheet_e["L10"] = new_data.format_b_y2 # sheet_e["M10"] = new_data.format_b_y3 # sheet_e["N10"] = new_data.format_b_y4 # sheet_e["O10"] = new_data.format_b_y5 # elif test.network_id == 6: # sheet_f = workbook["Rollout Plan"] # sheet_f["C11"] = new_data.location # sheet_f["D11"] = new_data.format_a_y1 # sheet_f["E11"] = new_data.format_a_y2 # sheet_f["F11"] = new_data.format_a_y3 # sheet_f["G11"] = new_data.format_a_y4 # sheet_f["H11"] = new_data.format_a_y5 # sheet_f["K11"] = new_data.format_b_y1 # sheet_f["L11"] = new_data.format_b_y2 # sheet_f["M11"] = new_data.format_b_y3 # sheet_f["N11"] = new_data.format_b_y4 # sheet_f["O11"] = new_data.format_b_y5 # elif test.network_id == 7: # sheet_g = workbook["Rollout Plan"] # sheet_g["C12"] = new_data.location # sheet_g["D12"] = new_data.format_a_y1 # sheet_g["E12"] = new_data.format_a_y2 # sheet_g["F12"] = new_data.format_a_y3 # sheet_g["G12"] = new_data.format_a_y4 # sheet_g["H12"] = new_data.format_a_y5 # sheet_g["K12"] = new_data.format_b_y1 # sheet_g["L12"] = new_data.format_b_y2 # sheet_g["M12"] = new_data.format_b_y3 # sheet_g["N12"] = new_data.format_b_y4 # sheet_g["O12"] = new_data.format_b_y5 # elif test.network_id == 8: # sheet_h = workbook["Rollout Plan"] # sheet_h["C13"] = new_data.location # sheet_h["D13"] = new_data.format_a_y1 # sheet_h["E13"] = new_data.format_a_y2 # sheet_h["F13"] = new_data.format_a_y3 # sheet_h["G13"] = new_data.format_a_y4 # sheet_h["H13"] = new_data.format_a_y5 # sheet_h["K13"] = new_data.format_b_y1 # sheet_h["L13"] = new_data.format_b_y2 # sheet_h["M13"] = new_data.format_b_y3 # sheet_h["N13"] = new_data.format_b_y4 # sheet_h["O13"] = new_data.format_b_y5 # elif test.network_id == 9: # sheet_i = workbook["Rollout Plan"] # sheet_i["C14"] = new_data.location # sheet_i["D14"] = new_data.format_a_y1 # sheet_i["E14"] = new_data.format_a_y2 # sheet_i["F14"] = new_data.format_a_y3 # sheet_i["G14"] = new_data.format_a_y4 # sheet_i["H14"] = new_data.format_a_y5 # sheet_i["K14"] = new_data.format_b_y1 # sheet_i["L14"] = new_data.format_b_y2 # sheet_i["M14"] = new_data.format_b_y3 # sheet_i["N14"] = new_data.format_b_y4 # sheet_i["O14"] = new_data.format_b_y5 # workbook.save(OUTPUT_PATH) # workbook.close() # print("✅ Excel updated successfully at:", OUTPUT_PATH) # return CustomResponse(status="1", message=Messages.NETWORK_CREAT) async def get_network(self, group_id: int): network = TblNetworkPlanning.get_by_group_id(group_id, self.db) if not network: return CustomResponse(status="-1", message=Messages.NETWORK_NOT_FOUND, data=None) network_data = NetworkPlanningResponse.model_validate(network) return CustomResponse(status="0", message="Success", data=network_data) async def get_network_by_group(self, group_id: int): results = (self.db.query(TblNetworkPlanning).filter(TblNetworkPlanning.group_id == group_id).all()) return [NetworkPlanningResponse.model_validate(row, from_attributes=True) for row in results] async def update_network_plan(self, request: NetworkPlanningUpdate): network_id = request.network_id existing_plan = TblNetworkPlanning.get_by_id(network_id, self.db) if not existing_plan: raise HTTPException(status_code=404, detail=Messages.NETWORK_NOT_FOUND) update_data = request.model_dump(exclude_unset=True) store_counts = calculate_total_store_counts_update(update_data, existing_plan) update_data.update(store_counts) updated_data = NetworkPlanningBase(**update_data) TblNetworkPlanning.update(network_id, updated_data, self.db) return CustomResponse(status="1", message=Messages.NETWORK_UPDATE) async def delete_network_plan(self, network_id:int): deleted_network_plan = TblNetworkPlanning.delete(network_id, self.db) if not deleted_network_plan: return CustomResponse(status="-1", message=Messages.NETWORK_NOT_FOUND) return CustomResponse(status="1", message=Messages.NETWORK_DELETE)