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, UploadFile from app.locale.messages import Messages 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.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) 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 | None): # Fetch existing record 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 from request if provided data_dict = request.model_dump(exclude_unset=True) for key, value in data_dict.items(): setattr(upload, key, value) # Handle file upload if upload_store_layout: image_filename = upload_store_layout.filename image_path = f"uploads/{image_filename}" # Ensure uploads folder exists os.makedirs("uploads", exist_ok=True) # Save new file with open(image_path, "wb") as buffer: shutil.copyfileobj(upload_store_layout.file, buffer) # Delete old file if exists if upload.upload_store_layout and os.path.exists(upload.upload_store_layout): os.remove(upload.upload_store_layout) # Update path upload.upload_store_layout = image_path # Commit DB changes 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" # Ensure output folder exists os.makedirs("uploaded_files", exist_ok=True) # Load workbook once 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 the user selected "Store Format A" if request.store_format_type == "Store Format A": sheet_a = workbook["Assu Sum Mod A"] sheet_a["I6"] = upload.store_size # print("✅ Data written to Assu Sum Mod A") # If the user selected "Store Format B" if request.store_format_type == "Store Format B": sheet_b = workbook["Assu Sum Mod B"] sheet_b["I6"] = upload.store_size # print("✅ Data written to Assu Sum Mod B") # Save both updates in one go workbook.save(OUTPUT_PATH) workbook.close() print("✅ Excel updated successfully at:", OUTPUT_PATH) 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")