from app.api.addtional_installation_fix.service import AdditionalInstallationService from app.api.carpentry.service import CarpentryService from app.api.civil.service import CivilService from app.api.commercial_equipment.service import CommercialEquipmentService from app.api.display_board.service import DisplayBoardService from app.api.display_racking_create.service import DisplayRackingUnitService from app.api.electrical_cabling.service import ElectricalCablingService from app.api.info_tech.service import InfoTechService from app.api.plumbing.service import PlumlingService from app.api.pre_operating_expenses.service import PreOperatingExpenseService from app.api.rent_expenses.service import RentExpensesService from app.api.visual_merchandising_elements.service import VisualMerchElementsService from app.dependency.authantication import JWTPayloadSchema from sqlalchemy.orm import Session from app.utils.common_utils import remove_external_links, safe_load_workbook class StoreInvestmentSummaryService: def __init__(self, db: Session, token: JWTPayloadSchema): self.db = db self.token = token async def get_store_investment_summary(self, group_id: int): # ---------------- NORMALIZE STORE FORMAT ---------------- def normalize(fmt): if fmt is None: return None s = str(fmt).upper() if "A" in s: return "Store Format A" if "B" in s: return "Store Format B" return fmt # ---------------- ACCUMULATORS ---------------- result_capital = {"Store Format A": 0, "Store Format B": 0} result_it = {"Store Format A": 0, "Store Format B": 0} rental_advance = {"Store Format A": 0, "Store Format B": 0} marketing = {"Store Format A": 0, "Store Format B": 0} pre_op = {"Store Format A": 0, "Store Format B": 0} # ---------------- SAFE HELPERS ---------------- def add_value(dic, fmt, value): fmt = normalize(fmt) if fmt in dic: dic[fmt] += float(value or 0) async def safe_async(func, default=None): try: return await func except: return default def safe(func, default=None): try: return func() except: return default # ========================================================= # CAPITAL EXPENSES (EXCLUDING IT) # ========================================================= # CIVIL civil_service = CivilService(self.db, self.token) civil = safe(lambda: civil_service.get_subtotal_by_store_format(group_id), []) for item in civil: add_value(result_capital, item["store_format_type"], item["subtotal"]) # DISPLAY RACKING display_service = DisplayRackingUnitService(self.db, self.token) display_racking = safe(lambda: display_service.get_subtotal_by_store_format(group_id), []) for item in display_racking: add_value(result_capital, item["store_format_type"], item["subtotal"]) # CARPENTRY carpentry_service = CarpentryService(self.db, self.token) carpentry = await safe_async(carpentry_service.get_and_sum_carpentry(group_id), []) for item in carpentry: add_value(result_capital, item.store_format_type, item.total_sum) # ELECTRICAL electrical_service = ElectricalCablingService(self.db, self.token) electrical = await safe_async(electrical_service.get_sum_electrical(group_id), []) for item in electrical: add_value(result_capital, item.store_format_type, item.total_sum) # DISPLAY BOARD display_board_service = DisplayBoardService(self.db, self.token) display_board = await safe_async(display_board_service.get_and_sum_display_board(group_id), []) for item in display_board: add_value(result_capital, item.store_format_type, item.total_sum) # COMMERCIAL EQUIPMENT ce_service = CommercialEquipmentService(self.db, self.token) ce_data = await safe_async(ce_service.get_commercial_equipment(group_id), []) for item in ce_data: add_value(result_capital, item.store_format_type, item.total) # VISUAL MERCH + PLUMBING + ADDITIONAL INSTALLATION (flat totals) vm_service = VisualMerchElementsService(self.db, self.token) pl_service = PlumlingService(self.db, self.token) add_service = AdditionalInstallationService(self.db, self.token) vm = await safe_async(vm_service.get_group_visual_merch_elements(group_id), []) pl = await safe_async(pl_service.get_plumbing(group_id), []) ad = await safe_async(add_service.get_group_additional_installation(group_id), []) vm_total = sum(float(getattr(x, "total", 0)) for x in vm) pl_total = sum(float(getattr(x, "total", 0)) for x in pl if not hasattr(x, "store_format_type")) ad_total = sum(float(getattr(x, "cost_per_unit", 0)) for x in ad) flat_total = vm_total + pl_total + ad_total if flat_total > 0: result_capital["Store Format A"] += flat_total result_capital["Store Format B"] += flat_total # ========================================================= # IT CAPITAL EXPENSES # ========================================================= it_service = InfoTechService(self.db, self.token) it_data = await safe_async(it_service.get_and_sum_info_tech(group_id), []) for item in it_data: add_value(result_it, item.store_format_type, item.total_sum) # ========================================================= # RENTAL ADVANCE # ========================================================= rent_service = RentExpensesService(self.db, self.token) rent = await safe_async(rent_service.get_group_rent_expenses(group_id), []) for item in rent: add_value(rental_advance, item.store_format_type, item.rental_advance_amount) # ========================================================= # MARKETING + PRE OPERATING EXPENSES # ========================================================= pre_service = PreOperatingExpenseService(self.db, self.token) pre = await safe_async(pre_service.get_pre_operating_expense(group_id), []) if pre: p = pre[0] marketing["Store Format A"] = float(p.marketing_expenses_store_a or 0) marketing["Store Format B"] = float(p.marketing_expenses_store_b or 0) pre_op["Store Format A"] = float(p.pre_operating_months_store_a or 0) pre_op["Store Format B"] = float(p.pre_operating_months_store_b or 0) # ========================================================= # FINAL TOTAL STORE INVESTMENTS # (YOUR REQUIRED CHANGE: SUM OF ALL FIELDS) # ========================================================= total_store_investments = { "Store Format A": ( result_capital["Store Format A"] + result_it["Store Format A"] + rental_advance["Store Format A"] + marketing["Store Format A"] + pre_op["Store Format A"] ), "Store Format B": ( result_capital["Store Format B"] + result_it["Store Format B"] + rental_advance["Store Format B"] + marketing["Store Format B"] + pre_op["Store Format B"] ) } # ========================================================= # RETURN FINAL STRUCTURE # ========================================================= 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_a = workbook["Assu Sum Mod A"] sheet_a["I18"] = result_capital.get("Store Format A", 0) sheet_b = workbook["Assu Sum Mod B"] sheet_b["I18"] = result_capital.get("Store Format B", 0) sheet_c = workbook["Assu Sum Mod A"] sheet_c["I20"] = result_it.get("Store Format A", 0) sheet_d = workbook["Assu Sum Mod B"] sheet_d["I20"] = result_it.get("Store Format B", 0) sheet_e = workbook["Assu Sum Mod A"] sheet_e["I21"] = result_capital["Store Format A"] + result_it["Store Format A"] sheet_f = workbook["Assu Sum Mod B"] sheet_f["I21"] = result_capital["Store Format B"] + result_it["Store Format B"] workbook.save(OUTPUT_PATH) workbook.close() print(f"✅ Excel updated successfully at: {OUTPUT_PATH}") return { "capitalExpensesExcludingIt": result_capital, "itCapitalExpenses": result_it, "totalStoreCapitalExpenses": { "Store Format A": result_capital["Store Format A"] + result_it["Store Format A"], "Store Format B": result_capital["Store Format B"] + result_it["Store Format B"], }, "rentalAdvance": rental_advance, "marketingExpenses": marketing, "preOperatingExpenses": pre_op, "totalStoreInvestments": total_store_investments, }