from typing import List from sqlalchemy.orm import Session from app.api.summary import schema from app.models.main.summary import SummaryBase, TblSummary from app.utils.schemas_utils import CustomResponse from app.locale.messages import Messages from fastapi import HTTPException from app.dependency.authantication import JWTPayloadSchema # Import all services for capital expenses calculation from app.api.civil.service import CivilService from app.api.display_racking_create.service import DisplayRackingUnitService from app.api.carpentry.service import CarpentryService from app.api.electrical_cabling.service import ElectricalCablingService from app.api.display_board.service import DisplayBoardService from app.api.commercial_equipment.service import CommercialEquipmentService from app.api.visual_merchandising_elements.service import VisualMerchElementsService from app.api.plumbing.service import PlumlingService from app.api.addtional_installation_fix.service import AdditionalInstallationService from app.api.info_tech.service import InfoTechService # Import services for sales calculations from app.api.customer_location.service import Customer_location from app.api.catchment_potential.service import Catchment_Potential from app.api.competitor_intensity.service import CompetitorIntensityService from app.api.location_spillage_factor_analysis.service import LocationSpillageFactorAnalysisService from app.api.sales_estimate.service import SalesEstimateService class SummaryService: def __init__(self, db:Session, token:JWTPayloadSchema): self.db = db self.token = token async def created(self, request:schema.SummaryCreate): created_data = SummaryBase.model_validate(request.model_dump()) TblSummary.create(created_data, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.SUMMARY) async def geted(self, group_id:int): geted_data = TblSummary.get(group_id, self.db) if not geted_data: raise HTTPException(status_code=404, detail="Get group ID not found") return [schema.SummaryResponse.model_validate(get_group) for get_group in geted_data] async def updated(self, request:List[schema.SummaryUpdate]): for req in request: updated_data = SummaryBase.model_validate(req.model_dump()) if updated_data.summary_id is None: return CustomResponse(status="-1", message=Messages.SUMMARY_NOT) TblSummary.update(updated_data.summary_id, updated_data, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.SUMMARY_UPDATE) async def deleted(self, summary_id:int): deleted_data = TblSummary.delete(summary_id, self.db) if not deleted_data: return CustomResponse(status="-1", message=Messages.SUMMARY_NOT) return CustomResponse(status="1", message=Messages.SUMMARY_DELETE) async def get_capital_expenses_calculation(self, group_id: int): """Calculate capital expenses (excluding IT) + IT capital expenses = total store capital expenses""" # Initialize services civil_service = CivilService(self.db, self.token) display_racking_service = DisplayRackingUnitService(self.db, self.token) carpentry_service = CarpentryService(self.db, self.token) electrical_service = ElectricalCablingService(self.db, self.token) display_board_service = DisplayBoardService(self.db, self.token) commercial_equipment_service = CommercialEquipmentService(self.db, self.token) visual_merch_service = VisualMerchElementsService(self.db, self.token) plumbing_service = PlumlingService(self.db, self.token) additional_installation_service = AdditionalInstallationService(self.db, self.token) info_tech_service = InfoTechService(self.db, self.token) capital_expenses_excluding_it = {} it_capital_expenses = {} try: # Get data from each service with error handling try: civil_totals = civil_service.get_subtotal_by_store_format(group_id) except: civil_totals = [] try: display_racking_totals = display_racking_service.get_subtotal_by_store_format(group_id) except: display_racking_totals = [] try: carpentry_totals = await carpentry_service.get_and_sum_carpentry(group_id) except: carpentry_totals = [] try: electrical_totals = await electrical_service.get_sum_electrical(group_id) except: electrical_totals = [] try: display_board_totals = await display_board_service.get_and_sum_display_board(group_id) except: display_board_totals = [] try: commercial_equipment_data = await commercial_equipment_service.get_commercial_equipment(group_id) visual_merch_data = await visual_merch_service.get_group_visual_merch_elements(group_id) plumbing_data = await plumbing_service.get_plumbing(group_id) additional_installation_data = await additional_installation_service.get_group_additional_installation(group_id) except HTTPException: commercial_equipment_data = [] visual_merch_data = [] plumbing_data = [] additional_installation_data = [] try: it_totals = await info_tech_service.get_and_sum_info_tech(group_id) except HTTPException: it_totals = [] def normalize_format(fmt): fmt_str = str(fmt) if 'STORE_FORMAT_A' in fmt_str or 'Store Format A' in fmt_str: return 'Store Format A' elif 'STORE_FORMAT_B' in fmt_str or 'Store Format B' in fmt_str: return 'Store Format B' return fmt_str # Aggregate Capital Expenses (Excluding IT) for item in civil_totals: store_format = normalize_format(item["store_format_type"]) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + item["subtotal"] for item in display_racking_totals: store_format = normalize_format(item["store_format_type"]) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + item["subtotal"] for item in carpentry_totals: store_format = normalize_format(item.store_format_type) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + float(item.total_sum or 0) for item in electrical_totals: store_format = normalize_format(item.store_format_type) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + float(item.total_sum or 0) for item in display_board_totals: store_format = normalize_format(item.store_format_type) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + float(item.total_sum or 0) for item in commercial_equipment_data: if hasattr(item, 'store_format_type') and hasattr(item, 'total'): store_format = normalize_format(item.store_format_type) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + float(item.total or 0) for item in plumbing_data: if hasattr(item, 'store_format_type') and hasattr(item, 'total'): store_format = normalize_format(item.store_format_type) capital_expenses_excluding_it[store_format] = capital_expenses_excluding_it.get(store_format, 0) + float(item.total or 0) # Visual merchandising and additional installation - same total for both formats visual_merch_total = sum(float(getattr(item, 'total', 0) or 0) for item in visual_merch_data) additional_installation_total = sum(float(getattr(item, 'total', 0) or 0) for item in additional_installation_data) for format_name in ['Store Format A', 'Store Format B']: capital_expenses_excluding_it[format_name] = capital_expenses_excluding_it.get(format_name, 0) + visual_merch_total + additional_installation_total # Aggregate IT Capital Expenses for item in it_totals: store_format = normalize_format(item.store_format_type) it_capital_expenses[store_format] = it_capital_expenses.get(store_format, 0) + float(item.total_sum or 0) # Calculate total store capital expenses total_store_capital_expenses = {} for format_name in ['Store Format A', 'Store Format B']: total_store_capital_expenses[format_name] = ( capital_expenses_excluding_it.get(format_name, 0) + it_capital_expenses.get(format_name, 0) ) return { "capital_expenses_excluding_it": capital_expenses_excluding_it, "it_capital_expenses": it_capital_expenses, "total_store_capital_expenses": total_store_capital_expenses } except Exception as e: raise HTTPException(status_code=500, detail=f"Error calculating capital expenses: {str(e)}") async def get_sales_calculations(self, group_id: int): """Calculate sales potential, sales estimate, and validation for both store formats""" customer_location_service = Customer_location(self.db, self.token) catchment_service = Catchment_Potential(self.db, self.token) competitor_service = CompetitorIntensityService(self.db, self.token) spillage_service = LocationSpillageFactorAnalysisService(self.db, self.token) sales_estimate_service = SalesEstimateService(self.db, self.token) try: # Get data from services customer_data = await customer_location_service.get_customer_locations_by_group(group_id) catchment_data = await catchment_service.get_catchment_poteneial_by_group(group_id) competitor_data = await competitor_service.get_group_competitor(group_id) spillage_data = await spillage_service.get_location(group_id) print(f"DEBUG - Customer data: {len(customer_data)} records") print(f"DEBUG - Catchment data: {len(catchment_data)} records") print(f"DEBUG - Competitor data: {len(competitor_data)} records") print(f"DEBUG - Spillage data: {len(spillage_data)} records") # Check if tables have store_format_type column has_format_column = ( customer_data and hasattr(customer_data[0], 'store_format_type') and getattr(customer_data[0], 'store_format_type', None) is not None ) print(f"DEBUG - Has format column: {has_format_column}") sales_potential = {} if has_format_column: # Format-specific calculation for format_type in ['Store Format A', 'Store Format B']: format_key = 'FORMAT_A' if format_type == 'Store Format A' else 'FORMAT_B' # Get household consumption for this format household_consumption = 0 for item in customer_data: item_format = str(getattr(item, 'store_format_type', '')) if format_key in item_format or format_type in item_format: household_consumption += float(getattr(item, 'household_consumption_per_month', 0) or 0) # Get catchment potential for this format catchment_potential = 0 for item in catchment_data: item_format = str(getattr(item, 'store_format_type', '')) if format_key in item_format or format_type in item_format: percentage = float(getattr(item, 'percentage_of_segment', 0) or 0) households = float(getattr(item, 'number_of_households', 0) or 0) catchment_potential += (percentage / 100) * households # Get competition factor for this format competition_factor = 1.0 for item in competitor_data: item_format = str(getattr(item, 'store_format_type', '')) if format_key in item_format or format_type in item_format: intensity = str(getattr(item, 'total_square_footage', '')).lower() if 'high' in intensity: competition_factor = 0.7 elif 'medium' in intensity: competition_factor = 0.85 break # Get spillage factor for this format spillage_factor = 1.0 for item in spillage_data: item_format = str(getattr(item, 'store_format_type', '')) if format_key in item_format or format_type in item_format: spillage = str(getattr(item, 'estimated_spillage_factor', '')).lower() if 'high' in spillage: spillage_factor = 1.2 elif 'low' in spillage: spillage_factor = 0.8 break print(f"DEBUG - {format_type}: consumption={household_consumption}, catchment={catchment_potential}, competition={competition_factor}, spillage={spillage_factor}") # Calculate sales potential for this format potential_value = catchment_potential * household_consumption * 0.15 * competition_factor * spillage_factor sales_potential[format_type] = potential_value else: # Fallback: Use same data for both formats (original logic) household_consumption = sum(float(getattr(item, 'household_consumption_per_month', 0) or 0) for item in customer_data) catchment_potential = 0 for item in catchment_data: percentage = float(getattr(item, 'percentage_of_segment', 0) or 0) households = float(getattr(item, 'number_of_households', 0) or 0) catchment_potential += (percentage / 100) * households competition_factor = 1.0 if competitor_data: intensity = str(getattr(competitor_data[0], 'total_square_footage', '')).lower() if 'high' in intensity: competition_factor = 0.7 elif 'medium' in intensity: competition_factor = 0.85 spillage_factor = 1.0 if spillage_data: spillage = str(getattr(spillage_data[0], 'estimated_spillage_factor', '')).lower() if 'high' in spillage: spillage_factor = 1.2 elif 'low' in spillage: spillage_factor = 0.8 print(f"DEBUG - Shared: consumption={household_consumption}, catchment={catchment_potential}, competition={competition_factor}, spillage={spillage_factor}") potential_value = catchment_potential * household_consumption * 0.15 * competition_factor * spillage_factor sales_potential = { "Store Format A": potential_value, "Store Format B": potential_value } # Get sales estimates sales_estimate = {} sales_data = await sales_estimate_service.get_group_sales_estimate(group_id) # Calculate for each store format for format_type in ['Store Format A', 'Store Format B']: format_key = 'FORMAT_A' if format_type == 'Store Format A' else 'FORMAT_B' bills_per_month = 0 items_per_bill = 0 avg_price = 0 for item in sales_data: store_format = getattr(item, 'store_formate_type', '') parameter = getattr(item, 'parameter', '') value = float(getattr(item, 'value', 0) or 0) # Match store format format_match = ( store_format == format_key or format_key in store_format or store_format == format_type or format_type in store_format ) if format_match: if 'Bills per Month' in parameter or 'BILLS_PER_MONTH' in parameter: bills_per_month = value elif 'Items per Bill' in parameter or 'ITEMS_PER_BILL' in parameter: items_per_bill = value elif 'Average Price' in parameter or 'AVG_PRICE' in parameter: avg_price = value sales_estimate[format_type] = bills_per_month * items_per_bill * avg_price return { "sales_potential": sales_potential, "sales_estimate": sales_estimate } except Exception as e: raise HTTPException(status_code=500, detail=f"Error calculating sales: {str(e)}")