import os from typing import List from fastapi import HTTPException from openpyxl import load_workbook from sqlalchemy import func from sqlalchemy.orm import Session from app.api.sales_estimate import schema from app.dependency.authantication import JWTPayloadSchema from app.locale.messages import Messages from app.models.main.sales_estimate import SalesEstimateBase, TblSalesEstimate from app.utils.common_utils import remove_external_links, safe_load_workbook from app.utils.schemas_utils import CustomResponse class SalesEstimateService: def __init__(self, db:Session, token:JWTPayloadSchema): self.db = db self.token = token async def create_sales_estimate(self, request: List[schema.SalesEstimateCreate]): 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) # 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 only once # if os.path.exists(OUTPUT_PATH): # workbook = load_workbook(OUTPUT_PATH) # else: # workbook = load_workbook(EXCEL_PATH) param_to_cell = { "Number of Bills per Month": "I7", "Number of Items per Bill": "I8", "Average Price per item (in Rupees)": "I9", "Gross Margin % Estimate": "I10", } format_to_sheet = { "Store Format A": "Assu Sum Mod A", "Store Format B": "Assu Sum Mod B", } # Loop through each request item for req in request: created_sales_estimate = SalesEstimateBase.model_validate(req.model_dump()) TblSalesEstimate.create_sales_estimate(created_sales_estimate, self.db) self.db.commit() # Write to Excel for format_type, sheet_name in format_to_sheet.items(): if req.store_formate_type == format_type: sheet = workbook[sheet_name] cell = param_to_cell.get(req.parameter) if cell: sheet[cell] = created_sales_estimate.value workbook.save(OUTPUT_PATH) workbook.close() return CustomResponse(status="1", message=Messages.SALES_ESTIMATE) # async def create_sales_estimate(self, request:List[schema.SalesEstimateCreate]): # EXCEL_PATH = "files/Simulation_Test_file_clean.xlsx" # OUTPUT_PATH = "uploaded_files/Simulation_Test_file_output.xlsx" # for req in request: # created_sales_estimate = SalesEstimateBase.model_validate(req.model_dump()) # TblSalesEstimate.create_sales_estimate(created_sales_estimate, self.db) # self.db.commit() # 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).") # param_to_cell = { # "Number of Bills per Month": "I7", # "Number of Items per Bill": "I8", # "Average Price per item (in Rupees)": "I9", # "Gross Margin % Estimate": "I10", # } # 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(): # sheet = workbook[sheet_name] # if request.store_formate_type == format_type: # cell = param_to_cell.get(request.parameter) # if cell: # sheet[cell] = created_sales_estimate.value # # print(f"✅ Updated {format_type} → {request.parameter} = {created_sales_estimate.value}") # # else: # # print(f"⚠️ Unknown parameter: {request.parameter}") # workbook.save(OUTPUT_PATH) # workbook.close() # print("✅ Excel updated successfully at:", OUTPUT_PATH) # return CustomResponse(status="1", message=Messages.SALES_ESTIMATE) async def get_sales_estimate(self, sales_get_id:int): new_get_sales_estimate = TblSalesEstimate.get_sales_estimate(sales_get_id, self.db) if not new_get_sales_estimate: raise HTTPException(status_code=404, detail="Get Sales Estimate ID not found") return schema.SalesEstimateResponse.model_validate(new_get_sales_estimate) async def get_group_sales_estimate(self, group_id) ->"TblSalesEstimate": new_get_group_sales_estimate = self.db.query(TblSalesEstimate).filter(TblSalesEstimate.group_id == group_id).all() if not new_get_group_sales_estimate: raise HTTPException(status_code=404, detail="Get group ID not found") return [schema.SalesEstimateResponse.model_validate(get_group) for get_group in new_get_group_sales_estimate] async def get_sales_value_sales_estimate(self, group_id:int) ->"TblSalesEstimate": new_get_sales_value_sales_estimate = ( self.db.query (func.exp(func.sum(func.ln(TblSalesEstimate.value))).label("sales_value")) # self.db.query( # TblSalesEstimate.parameter, # func.mult(TblSalesEstimate.value ).label("sales_value") # ) .filter( TblSalesEstimate.parameter.in_([ "Number of Bills per Month", "Number of Items per Bill", "Average Price per item (in Rupees)" ]), TblSalesEstimate.group_id == group_id ) .scalar() # .group_by(TblSalesEstimate.parameter) # .all() ) if not new_get_sales_value_sales_estimate: raise HTTPException(status_code=404, detail="Get Multiple ID not found") # return [schema.SalesEstimateSum.model_validate(get_multiple) for get_multiple in new_get_sales_value_sales_estimate] return new_get_sales_value_sales_estimate # new_get_sales_value_sales_estimate = ( # self.db.query( # TblSalesEstimate.parameter.in_([ # "Number of Bills per Month", # "Number of Items per Bill", # "Average Price per item (in Rupees)" # ]) # ) # ) async def update_sales_estimate(self, request:List[schema.SalesEstimateUpdate]): # EXCEL_PATH = "files/Simulation_Test_file_clean_final.xlsx" # OUTPUT_PATH = "uploaded_files/Simulation_Test_file_clean_final_output.xlsx" for req in request: updated_sales_estimate = SalesEstimateBase.model_validate(req.model_dump()) if updated_sales_estimate.sales_id is None: return CustomResponse(status="-1", message=Messages.SALES_ESTIMATE_NOT) TblSalesEstimate.update_sales_estimate(updated_sales_estimate.sales_id, updated_sales_estimate, self.db) self.db.commit() 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) # 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).") param_to_cell = { "Number of Bills per Month": "I7", "Number of Items per Bill": "I8", "Average Price per item (in Rupees)": "I9", "Gross Margin % Estimate": "I10", } 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(): sheet = workbook[sheet_name] if req.store_formate_type == format_type: cell = param_to_cell.get(req.parameter) if cell: sheet[cell] = updated_sales_estimate.value # print(f"✅ Updated {format_type} → {request.parameter} = {created_sales_estimate.value}") # else: # print(f"⚠️ Unknown parameter: {request.parameter}") workbook.save(OUTPUT_PATH) workbook.close() # print("✅ Excel updated successfully at:", OUTPUT_PATH) return CustomResponse(status="1" , message=Messages.SALES_ESTIMATE_UPDATE) async def delete_sales_estimate(self, sales_id:int): deleted_sales_estimate = TblSalesEstimate.delete_sales_estimate(sales_id, self.db) if not deleted_sales_estimate: return CustomResponse(status="-1", message=Messages.SALES_ESTIMATE_NOT) return CustomResponse(status="1", message=Messages.SALES_ESTIMATE_DELETE)