import os import shutil from typing import Optional from openpyxl import load_workbook from sqlalchemy import desc, select import app.api.store_formate.schemas as schemas from sqlalchemy.orm import Session from app.dependency.authantication import JWTPayloadSchema from fastapi import Form, HTTPException, UploadFile from app.locale.messages import Messages from app.models.main.civil import TblCivil from app.models.main.display_board import TblDisplayBoard from app.models.main.electrical_cabling import TblElectricalCabling from app.models.main.group import TblGroup from app.models.main.main_data import TblMainData from app.models.main.store_formate import StoreFormatBase, TblStoreFormat from app.utils.common_utils import remove_external_links, safe_load_workbook from app.utils.schemas_utils import CustomResponse from app.api.store_formate.schemas import StoreFormateUpdate class StoreFormateService: def __init__(self, db: Session,token:JWTPayloadSchema): self.db = db self.token = token async def create_store_formate(self, request: schemas.StoreFormateCreate): created_user = StoreFormatBase.model_validate(request.model_dump()) TblStoreFormat.create(created_user, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.STOREFORMATE_CREAT) # def create_store_format3(self, request: schemas.StoreFormateCreate11, upload_store_layout: UploadFile = Form(None)): # group = self.db.execute( # select(TblGroup).where(TblGroup.group_id == request.group_id) # ).scalar_one_or_none() # if not group: # return CustomResponse(status="0", message="Group not found") # result = self.db.execute( # select(TblMainData) # .where( # (TblMainData.format_type == group.category_assigned) | # (TblMainData.format_type == group.category) # ) # .order_by(desc(TblMainData.data_id)) # .limit(1) # ) # main_data = result.scalar_one_or_none() # if not main_data: # return CustomResponse(status="0", message="Main data not found for the group's category") # merchandise_map = { # 'Low': main_data.no_of_cats_l or 0, # 'Medium': main_data.no_of_cats_m or 0, # 'High': main_data.no_of_cats_h or 0 # } # service_parameters_map = { # 'Low': main_data.services_l or 0, # 'Medium': main_data.services_m or 0, # 'High': main_data.services_h or 0 # } # technology_adoption_map = { # 'Low': main_data.tech_inv_l or 0, # 'Medium': main_data.tech_inv_m or 0, # 'High': main_data.tech_inv_h or 0 # } # merchandise_text = request.merchandise # if merchandise_text not in merchandise_map: # return CustomResponse(status="0", message="Invalid merchandise value (Low, Medium, High required)") # merchandise_value = merchandise_map[merchandise_text] # service_text = request.service_parameters # if service_text not in service_parameters_map: # return CustomResponse(status="0", message="Invalid service parameters value (Low, Medium, High required)") # service_value = service_parameters_map[service_text] # tech_text = request.technology_adoption # if tech_text not in technology_adoption_map: # return CustomResponse(status="0", message="Invalid technology adoption value (Low, Medium, High required)") # tech_value = technology_adoption_map[tech_text] # if not upload_store_layout: # return CustomResponse(status="0", message="Upload Store Layout is required") # image_filename = upload_store_layout.filename # image_path = f"uploads/{image_filename}" # with open(image_path, "wb") as buffer: # shutil.copyfileobj(upload_store_layout.file, buffer) # new_store = StoreFormatBase.model_validate(request) # new_store.upload_store_layout = image_path # new_store.merchandise=str(merchandise_value) # new_store.service_parameters=str(service_value) # new_store.technology_adoption=str(tech_value) # new_store_db = TblStoreFormat(**new_store.model_dump()) # self.db.add(new_store_db) # self.db.commit() # self.db.refresh(new_store_db) # EXCEL_PATH = "files/Simulation_Test_file_clean.xlsm" # OUTPUT_PATH = "uploaded_files/Simulation_Test_file_output.xlsm" # # Ensure output folder exists # os.makedirs("uploaded_files", exist_ok=True) # 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).") # format_to_sheet = { # "Store Format A": "Assu Sum Mod A", # "Store Format B": "Assu Sum Mod B", # } # for format_type, sheet_name in format_to_sheet.items(): # if request.store_format_type == format_type: # sheet = workbook[sheet_name] # sheet["I6"] = new_store.store_size # # print(f"✅ Data written to {sheet_name}") # workbook.save(OUTPUT_PATH) # workbook.close() # print(f"✅ Excel updated successfully at: {OUTPUT_PATH}") # return CustomResponse(status="1", message=Messages.STOREFORMATE_CREAT) def create_store_format3( self, request: schemas.StoreFormateCreate11, upload_store_layout: UploadFile = Form(None) ): # 1️⃣ Validate Group group = self.db.execute( select(TblGroup).where(TblGroup.group_id == request.group_id) ).scalar_one_or_none() if not group: return CustomResponse(status="0", message="Group not found") # 2️⃣ Fetch Latest MainData For group's assigned category or category category_list = [ group.category_assigned, group.category ] result = self.db.execute( select(TblMainData) .where(TblMainData.format_type.in_(category_list)) .order_by(desc(TblMainData.data_id)) .limit(1) ) main_data = result.scalar_one_or_none() if not main_data: return CustomResponse( status="0", message="Main data not found for the group's category" ) # 3️⃣ Mapping values merchandise_map = { 'Low': main_data.no_of_cats_l or 0, 'Medium': main_data.no_of_cats_m or 0, 'High': main_data.no_of_cats_h or 0 } service_parameters_map = { 'Low': main_data.services_l or 0, 'Medium': main_data.services_m or 0, 'High': main_data.services_h or 0 } technology_adoption_map = { 'Low': main_data.tech_inv_l or 0, 'Medium': main_data.tech_inv_m or 0, 'High': main_data.tech_inv_h or 0 } # 4️⃣ Validate & lookup values merchandise_value = merchandise_map.get(request.merchandise) if merchandise_value is None: return CustomResponse(status="0", message="Invalid merchandise level") service_value = service_parameters_map.get(request.service_parameters) if service_value is None: return CustomResponse(status="0", message="Invalid service parameter level") tech_value = technology_adoption_map.get(request.technology_adoption) if tech_value is None: return CustomResponse(status="0", message="Invalid technology adoption level") # 5️⃣ Handle Upload File (Optional) image_path = None if upload_store_layout: image_filename = upload_store_layout.filename image_path = f"uploads/{image_filename}" os.makedirs("uploads", exist_ok=True) with open(image_path, "wb") as buffer: shutil.copyfileobj(upload_store_layout.file, buffer) # 6️⃣ Save Store Format new_store = StoreFormatBase.model_validate(request) new_store.upload_store_layout = image_path new_store.merchandise = str(merchandise_value) new_store.service_parameters = str(service_value) new_store.technology_adoption = str(tech_value) new_store_db = TblStoreFormat(**new_store.model_dump()) self.db.add(new_store_db) self.db.commit() self.db.refresh(new_store_db) # # 7️⃣ Excel Update Logic # 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) # if os.path.exists(OUTPUT_PATH): # workbook = load_workbook(OUTPUT_PATH) # else: # workbook = load_workbook(EXCEL_PATH) OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" workbook = safe_load_workbook(OUTPUT_PATH, EXCEL_PATH) remove_external_links(workbook) workbook.save(OUTPUT_PATH) format_to_sheet = { "Store Format A": "Assu Sum Mod A", "Store Format B": "Assu Sum Mod B", } sheet_name = format_to_sheet.get(request.store_format_type) if sheet_name: sheet = workbook[sheet_name] sheet["I6"] = new_store.store_size workbook.save(OUTPUT_PATH) workbook.close() return CustomResponse(status="1", message=Messages.STOREFORMATE_CREAT) # async def update_store_formate( # self, # store_id: int, # request: StoreFormateUpdate, # upload_store_layout: UploadFile = Form(None) # ): # upload = ( # self.db.query(TblStoreFormat) # .filter(TblStoreFormat.store_id == store_id) # .first() # ) # if not upload: # return CustomResponse(status="0", message="Store format not found") # update_fields = request.model_dump(exclude_unset=True) # if any(k in update_fields for k in ["merchandise", "service_parameters", "technology_adoption"]): # group = ( # self.db.query(TblGroup) # .filter(TblGroup.group_id == upload.group_id) # .first() # ) # if not group: # return CustomResponse(status="0", message="Group not found for mapping") # category_list = [group.category_assigned, group.category] # main_data = ( # self.db.execute( # select(TblMainData) # .where(TblMainData.format_type.in_(category_list)) # .order_by(desc(TblMainData.data_id)) # .limit(1) # ) # .scalar_one_or_none() # ) # if not main_data: # return CustomResponse(status="0", message="Main data not found for mapping") # merchandise_map = { # 'Low': main_data.no_of_cats_l or 0, # 'Medium': main_data.no_of_cats_m or 0, # 'High': main_data.no_of_cats_h or 0 # } # service_parameters_map = { # 'Low': main_data.services_l or 0, # 'Medium': main_data.services_m or 0, # 'High': main_data.services_h or 0 # } # technology_adoption_map = { # 'Low': main_data.tech_inv_l or 0, # 'Medium': main_data.tech_inv_m or 0, # 'High': main_data.tech_inv_h or 0 # } # if "merchandise" in update_fields: # merch_val = merchandise_map.get(request.merchandise) # if merch_val is None: # return CustomResponse(status="0", message="Invalid merchandise level") # upload.merchandise = str(merch_val) # if "service_parameters" in update_fields: # serv_val = service_parameters_map.get(request.service_parameters) # if serv_val is None: # return CustomResponse(status="0", message="Invalid service parameter level") # upload.service_parameters = str(serv_val) # if "technology_adoption" in update_fields: # tech_val = technology_adoption_map.get(request.technology_adoption) # if tech_val is None: # return CustomResponse(status="0", message="Invalid technology adoption level") # upload.technology_adoption = str(tech_val) # for key, value in update_fields.items(): # if key not in ["merchandise", "service_parameters", "technology_adoption"]: # setattr(upload, key, value) # if upload_store_layout and upload_store_layout.filename: # os.makedirs("uploads", exist_ok=True) # new_image_path = f"uploads/{upload_store_layout.filename}" # with open(new_image_path, "wb") as buffer: # shutil.copyfileobj(upload_store_layout.file, buffer) # if upload.upload_store_layout and os.path.exists(upload.upload_store_layout): # os.remove(upload.upload_store_layout) # upload.upload_store_layout = new_image_path # self.db.commit() # self.db.refresh(upload) # 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(OUTPUT_PATH) if os.path.exists(OUTPUT_PATH) else load_workbook(EXCEL_PATH) # sheet_map = { # "Store Format A": "Assu Sum Mod A", # "Store Format B": "Assu Sum Mod B", # } # sheet_name = sheet_map.get(upload.store_format_type) # if sheet_name: # try: # sheet = workbook[sheet_name] # sheet["I6"] = upload.store_size # except Exception as e: # return CustomResponse(status="0", message=f"Excel update failed: {str(e)}") # workbook.save(OUTPUT_PATH) # workbook.close() # group_civil = self.db.query(TblCivil).filter(TblCivil.group_id == upload.group_id).first() # group_electric = self.db.query(TblElectricalCabling).filter(TblElectricalCabling.group_id == upload.group_id).first() # group_display = self.db.query(TblDisplayBoard).filter(TblDisplayBoard.group_id == upload.group_id).first() # #civil # total_sqft = group_civil.rate_per_sqft * upload.store_size # group_civil.total_sqft = total_sqft # #electrical # if TblElectricalCabling.feature_type in ["Lighting", "Cabling"]: # total = group_electric.cost_unit * upload.store_size # group_electric.total = total # elif group_electric.choice == "Rate per SQFT": # total = group_electric.rate_unit * upload.store_size # group_electric.total = total # #displayBoard # elif group_display.choice == "Rate per SQFT": # total = upload.store_size * group_display.rate_per_sq # group_display.total = total # return CustomResponse( # status="1", # message="Store format updated successfully", # data={"image_url": upload.upload_store_layout}, # ) async def update_store_formate( self, store_id: int, request: StoreFormateUpdate, upload_store_layout: UploadFile = Form(None) ): upload = ( self.db.query(TblStoreFormat) .filter(TblStoreFormat.store_id == store_id) .first() ) if not upload: raise HTTPException(status_code=404, detail="Store format not found") update_fields = request.model_dump(exclude_unset=True) if any(k in update_fields for k in ["merchandise", "service_parameters", "technology_adoption"]): group = ( self.db.query(TblGroup) .filter(TblGroup.group_id == upload.group_id) .first() ) if not group: raise HTTPException(status_code=404, detail="Group not found for mapping") category_list = [group.category_assigned, group.category] main_data = ( self.db.execute( select(TblMainData) .where(TblMainData.format_type.in_(category_list)) .order_by(desc(TblMainData.data_id)) .limit(1) ) .scalar_one_or_none() ) if not main_data: raise HTTPException(status_code=404, detail="Main data not found") merchandise_map = { 'Low': main_data.no_of_cats_l or 0, 'Medium': main_data.no_of_cats_m or 0, 'High': main_data.no_of_cats_h or 0 } service_parameters_map = { 'Low': main_data.services_l or 0, 'Medium': main_data.services_m or 0, 'High': main_data.services_h or 0 } technology_adoption_map = { 'Low': main_data.tech_inv_l or 0, 'Medium': main_data.tech_inv_m or 0, 'High': main_data.tech_inv_h or 0 } # ✔ merchandise → mapped value if "merchandise" in update_fields: merch_val = merchandise_map.get(request.merchandise) if merch_val is None: raise HTTPException(status_code=400, detail="Invalid merchandise level") upload.merchandise = str(merch_val) # ✔ service parameters → mapped value if "service_parameters" in update_fields: serv_val = service_parameters_map.get(request.service_parameters) if serv_val is None: raise HTTPException(status_code=400, detail="Invalid service parameter level") upload.service_parameters = str(serv_val) # ✔ technology adoption → mapped value if "technology_adoption" in update_fields: tech_val = technology_adoption_map.get(request.technology_adoption) if tech_val is None: raise HTTPException(status_code=400, detail="Invalid technology adoption level") upload.technology_adoption = str(tech_val) for key, value in update_fields.items(): if key not in ["merchandise", "service_parameters", "technology_adoption"]: setattr(upload, key, value) if upload_store_layout and upload_store_layout.filename: os.makedirs("uploads", exist_ok=True) new_image_path = f"uploads/{upload_store_layout.filename}" with open(new_image_path, "wb") as f: shutil.copyfileobj(upload_store_layout.file, f) if upload.upload_store_layout and os.path.exists(upload.upload_store_layout): os.remove(upload.upload_store_layout) upload.upload_store_layout = new_image_path self.db.commit() self.db.refresh(upload) # 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(OUTPUT_PATH) # if os.path.exists(OUTPUT_PATH) # else load_workbook(EXCEL_PATH) # ) OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" workbook = safe_load_workbook(OUTPUT_PATH, EXCEL_PATH) remove_external_links(workbook) workbook.save(OUTPUT_PATH) sheet_map = { "Store Format A": "Assu Sum Mod A", "Store Format B": "Assu Sum Mod B", } sheet_name = sheet_map.get(upload.store_format_type) if sheet_name: try: sheet = workbook[sheet_name] sheet["I6"] = upload.store_size except Exception as e: raise HTTPException(status_code=500, detail=f"Excel update failed: {str(e)}") workbook.save(OUTPUT_PATH) workbook.close() group_civil = ( self.db.query(TblCivil) .filter(TblCivil.group_id == upload.group_id) .first() ) group_electric = ( self.db.query(TblElectricalCabling) .filter(TblElectricalCabling.group_id == upload.group_id) .first() ) group_display = ( self.db.query(TblDisplayBoard) .filter(TblDisplayBoard.group_id == upload.group_id) .first() ) # ------- CIVIL ------- if group_civil: group_civil.total_sqft = group_civil.rate_per_sqft * upload.store_size # ------- ELECTRICAL ------- if group_electric: if group_electric.feature_type in ["Lighting", "Cabling"]: group_electric.total = group_electric.cost_unit * upload.store_size elif group_electric.choice == "Rate per SQFT": group_electric.total = group_electric.rate_unit * upload.store_size # ------- DISPLAY BOARD ------- if group_display and group_display.choice == "Rate per SQFT": group_display.total = upload.store_size * group_display.rate_per_sq self.db.commit() return CustomResponse( status="1", message="Store format updated successfully", data={"image_url": upload.upload_store_layout}, ) async def get_store_formate(self, group_id: int): store_format = TblStoreFormat.get_by_group_id(group_id, self.db) return [schemas.StoreFormateRes.model_validate(get_group) for get_group in store_format] async def delete_store_formate(self, store_id: int): deleted = TblStoreFormat.delete(store_id, self.db) if not deleted: return CustomResponse(status="-1", message="Store Formate not found") return CustomResponse(status="1", message="Store Formate deleted successfully")