# app/modules/loyalty/services/program_service.py """ Loyalty program service. Merchant-based program management: - Programs belong to merchants, not individual stores - All stores under a merchant share the same loyalty program - One program per merchant Handles CRUD operations for loyalty programs including: - Program creation and configuration - Program updates - Program activation/deactivation - Statistics retrieval """ import logging from datetime import UTC, datetime from sqlalchemy.orm import Session from app.modules.loyalty.exceptions import ( LoyaltyProgramAlreadyExistsException, LoyaltyProgramNotFoundException, ) from app.modules.loyalty.models import ( LoyaltyProgram, MerchantLoyaltySettings, ) from app.modules.loyalty.schemas.program import ( ProgramCreate, ProgramUpdate, ) logger = logging.getLogger(__name__) class ProgramService: """Service for loyalty program operations.""" # ========================================================================= # Read Operations # ========================================================================= def get_program(self, db: Session, program_id: int) -> LoyaltyProgram | None: """Get a loyalty program by ID.""" return ( db.query(LoyaltyProgram) .filter(LoyaltyProgram.id == program_id) .first() ) def get_program_by_merchant(self, db: Session, merchant_id: int) -> LoyaltyProgram | None: """Get a merchant's loyalty program.""" return ( db.query(LoyaltyProgram) .filter(LoyaltyProgram.merchant_id == merchant_id) .first() ) def get_active_program_by_merchant(self, db: Session, merchant_id: int) -> LoyaltyProgram | None: """Get a merchant's active loyalty program.""" return ( db.query(LoyaltyProgram) .filter( LoyaltyProgram.merchant_id == merchant_id, LoyaltyProgram.is_active == True, ) .first() ) def get_program_by_store(self, db: Session, store_id: int) -> LoyaltyProgram | None: """ Get the loyalty program for a store. Looks up the store's merchant and returns the merchant's program. """ from app.modules.tenancy.services.store_service import store_service store = store_service.get_store_by_id_optional(db, store_id) if not store: return None return self.get_program_by_merchant(db, store.merchant_id) def get_active_program_by_store(self, db: Session, store_id: int) -> LoyaltyProgram | None: """ Get the active loyalty program for a store. Looks up the store's merchant and returns the merchant's active program. """ from app.modules.tenancy.services.store_service import store_service store = store_service.get_store_by_id_optional(db, store_id) if not store: return None return self.get_active_program_by_merchant(db, store.merchant_id) def require_program(self, db: Session, program_id: int) -> LoyaltyProgram: """Get a program or raise exception if not found.""" program = self.get_program(db, program_id) if not program: raise LoyaltyProgramNotFoundException(str(program_id)) return program def require_program_by_merchant(self, db: Session, merchant_id: int) -> LoyaltyProgram: """Get a merchant's program or raise exception if not found.""" program = self.get_program_by_merchant(db, merchant_id) if not program: raise LoyaltyProgramNotFoundException(f"merchant:{merchant_id}") return program def require_active_program_by_store(self, db: Session, store_id: int) -> LoyaltyProgram: """Get a store's active program or raise exception if not found.""" program = self.get_active_program_by_store(db, store_id) if not program: raise LoyaltyProgramNotFoundException(f"store:{store_id}") return program def require_program_by_store(self, db: Session, store_id: int) -> LoyaltyProgram: """Get a store's program or raise exception if not found.""" program = self.get_program_by_store(db, store_id) if not program: raise LoyaltyProgramNotFoundException(f"store:{store_id}") return program def get_store_by_code(self, db: Session, store_code: str): """ Find a store by its store_code or subdomain. Args: db: Database session store_code: Store code or subdomain Returns: Store object Raises: StoreNotFoundException: If store not found """ from app.modules.tenancy.exceptions import StoreNotFoundException from app.modules.tenancy.services.store_service import store_service store = store_service.get_store_by_code_or_subdomain(db, store_code) if not store: raise StoreNotFoundException(store_code) return store def get_store_merchant_id(self, db: Session, store_id: int) -> int: """ Get the merchant ID for a store. Args: db: Database session store_id: Store ID Returns: Merchant ID Raises: StoreNotFoundException: If store not found """ from app.modules.tenancy.exceptions import StoreNotFoundException from app.modules.tenancy.services.store_service import store_service store = store_service.get_store_by_id_optional(db, store_id) if not store: raise StoreNotFoundException(str(store_id), identifier_type="id") return store.merchant_id def get_merchant_locations(self, db: Session, merchant_id: int) -> list: """ Get all active store locations for a merchant. Args: db: Database session merchant_id: Merchant ID Returns: List of active Store objects """ from app.modules.tenancy.services.store_service import store_service return store_service.get_stores_by_merchant_id( db, merchant_id, active_only=True ) def get_program_list_stats(self, db: Session, program) -> dict: """ Get aggregation stats for a program used in list views. Args: db: Database session program: LoyaltyProgram instance Returns: Dict with merchant_name, total_cards, active_cards, total_points_issued, total_points_redeemed """ from sqlalchemy import func from app.modules.loyalty.models import LoyaltyCard, LoyaltyTransaction from app.modules.tenancy.services.merchant_service import merchant_service merchant = merchant_service.get_merchant_by_id_optional(db, program.merchant_id) merchant_name = merchant.name if merchant else None total_cards = ( db.query(func.count(LoyaltyCard.id)) .filter(LoyaltyCard.merchant_id == program.merchant_id) .scalar() or 0 ) active_cards = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.merchant_id == program.merchant_id, LoyaltyCard.is_active == True, ) .scalar() or 0 ) total_points_issued = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter( LoyaltyTransaction.merchant_id == program.merchant_id, LoyaltyTransaction.points_delta > 0, ) .scalar() or 0 ) total_points_redeemed = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter( LoyaltyTransaction.merchant_id == program.merchant_id, LoyaltyTransaction.points_delta < 0, ) .scalar() or 0 ) return { "merchant_name": merchant_name, "total_cards": total_cards, "active_cards": active_cards, "total_points_issued": total_points_issued, "total_points_redeemed": total_points_redeemed, } def get_platform_stats(self, db: Session) -> dict: """ Get platform-wide loyalty statistics. Returns dict with: - total_programs, active_programs - merchants_with_programs - total_cards, active_cards - transactions_30d - points_issued_30d, points_redeemed_30d """ from datetime import UTC, datetime, timedelta from sqlalchemy import func from app.modules.loyalty.models import ( LoyaltyCard, LoyaltyProgram, LoyaltyTransaction, ) # Program counts total_programs = db.query(func.count(LoyaltyProgram.id)).scalar() or 0 active_programs = ( db.query(func.count(LoyaltyProgram.id)) .filter(LoyaltyProgram.is_active == True) .scalar() or 0 ) # Card counts total_cards = db.query(func.count(LoyaltyCard.id)).scalar() or 0 active_cards = ( db.query(func.count(LoyaltyCard.id)) .filter(LoyaltyCard.is_active == True) .scalar() or 0 ) # Transaction counts (last 30 days) thirty_days_ago = datetime.now(UTC) - timedelta(days=30) transactions_30d = ( db.query(func.count(LoyaltyTransaction.id)) .filter(LoyaltyTransaction.transaction_at >= thirty_days_ago) .scalar() or 0 ) # Points issued/redeemed (last 30 days) points_issued_30d = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter( LoyaltyTransaction.transaction_at >= thirty_days_ago, LoyaltyTransaction.points_delta > 0, ) .scalar() or 0 ) points_redeemed_30d = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter( LoyaltyTransaction.transaction_at >= thirty_days_ago, LoyaltyTransaction.points_delta < 0, ) .scalar() or 0 ) # Merchant count with programs merchants_with_programs = ( db.query(func.count(func.distinct(LoyaltyProgram.merchant_id))).scalar() or 0 ) # All-time points total_points_issued = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter(LoyaltyTransaction.points_delta > 0) .scalar() or 0 ) total_points_redeemed = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter(LoyaltyTransaction.points_delta < 0) .scalar() or 0 ) # Outstanding points balance total_points_balance = ( db.query(func.sum(LoyaltyCard.points_balance)).scalar() or 0 ) # New members this month month_start = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0) new_this_month = ( db.query(func.count(LoyaltyCard.id)) .filter(LoyaltyCard.created_at >= month_start) .scalar() or 0 ) # Estimated liability (rough estimate: points / 100 as euros) points_value_cents = total_points_balance // 100 * 100 # Stamp liability across all programs stamp_liability = 0 programs = db.query(LoyaltyProgram).all() for prog in programs: stamp_value = prog.stamps_reward_value_cents or 0 stamps_target = prog.stamps_target or 1 current_stamps = ( db.query(func.sum(LoyaltyCard.stamp_count)) .filter(LoyaltyCard.program_id == prog.id) .scalar() or 0 ) stamp_liability += current_stamps * stamp_value // stamps_target estimated_liability_cents = stamp_liability + points_value_cents return { "total_programs": total_programs, "active_programs": active_programs, "merchants_with_programs": merchants_with_programs, "total_cards": total_cards, "active_cards": active_cards, "transactions_30d": transactions_30d, "points_issued_30d": points_issued_30d, "points_redeemed_30d": points_redeemed_30d, "total_points_issued": total_points_issued, "total_points_redeemed": total_points_redeemed, "total_points_balance": total_points_balance, "new_this_month": new_this_month, "estimated_liability_cents": estimated_liability_cents, } def check_self_enrollment_allowed(self, db: Session, merchant_id: int) -> None: """ Check if self-enrollment is allowed for a merchant. Raises: SelfEnrollmentDisabledException: If self-enrollment is disabled """ from app.modules.loyalty.exceptions import SelfEnrollmentDisabledException settings = self.get_merchant_settings(db, merchant_id) if settings and not settings.allow_self_enrollment: raise SelfEnrollmentDisabledException() def list_programs( self, db: Session, *, skip: int = 0, limit: int = 100, is_active: bool | None = None, search: str | None = None, ) -> tuple[list[LoyaltyProgram], int]: """List all loyalty programs (admin). Args: db: Database session skip: Number of records to skip limit: Maximum records to return is_active: Filter by active status search: Search by merchant name (case-insensitive) """ query = db.query(LoyaltyProgram) if is_active is not None: query = query.filter(LoyaltyProgram.is_active == is_active) if search: from app.modules.tenancy.services.merchant_service import merchant_service merchants, _ = merchant_service.get_merchants(db, search=search, limit=10000) merchant_ids = [m.id for m in merchants] query = query.filter(LoyaltyProgram.merchant_id.in_(merchant_ids)) total = query.count() programs = query.order_by(LoyaltyProgram.created_at.desc()).offset(skip).limit(limit).all() return programs, total # ========================================================================= # Write Operations # ========================================================================= def create_program( self, db: Session, merchant_id: int, data: ProgramCreate, ) -> LoyaltyProgram: """ Create a new loyalty program for a merchant. Args: db: Database session merchant_id: Merchant ID data: Program configuration Returns: Created program Raises: LoyaltyProgramAlreadyExistsException: If merchant already has a program """ # Check if merchant already has a program existing = self.get_program_by_merchant(db, merchant_id) if existing: raise LoyaltyProgramAlreadyExistsException(merchant_id) # Convert points_rewards to dict list for JSON storage points_rewards_data = [r.model_dump() for r in data.points_rewards] program = LoyaltyProgram( merchant_id=merchant_id, loyalty_type=data.loyalty_type, # Stamps stamps_target=data.stamps_target, stamps_reward_description=data.stamps_reward_description, stamps_reward_value_cents=data.stamps_reward_value_cents, # Points points_per_euro=data.points_per_euro, points_rewards=points_rewards_data, points_expiration_days=data.points_expiration_days, welcome_bonus_points=data.welcome_bonus_points, minimum_redemption_points=data.minimum_redemption_points, minimum_purchase_cents=data.minimum_purchase_cents, # Anti-fraud cooldown_minutes=data.cooldown_minutes, max_daily_stamps=data.max_daily_stamps, require_staff_pin=data.require_staff_pin, # Branding card_name=data.card_name, card_color=data.card_color, card_secondary_color=data.card_secondary_color, logo_url=data.logo_url, hero_image_url=data.hero_image_url, # Terms terms_text=data.terms_text, privacy_url=data.privacy_url, # Status is_active=True, activated_at=datetime.now(UTC), ) db.add(program) db.flush() # Create default merchant settings (idempotent — skips if already exists) self.get_or_create_merchant_settings(db, merchant_id) db.commit() db.refresh(program) logger.info( f"Created loyalty program {program.id} for merchant {merchant_id} " f"(type: {program.loyalty_type})" ) return program def update_program( self, db: Session, program_id: int, data: ProgramUpdate, ) -> LoyaltyProgram: """ Update a loyalty program. Args: db: Database session program_id: Program ID data: Update data Returns: Updated program """ program = self.require_program(db, program_id) update_data = data.model_dump(exclude_unset=True) # Handle points_rewards specially (convert to dict list) if "points_rewards" in update_data and update_data["points_rewards"] is not None: update_data["points_rewards"] = [ r.model_dump() if hasattr(r, "model_dump") else r for r in update_data["points_rewards"] ] for field, value in update_data.items(): setattr(program, field, value) db.commit() db.refresh(program) logger.info(f"Updated loyalty program {program_id}") return program def activate_program(self, db: Session, program_id: int) -> LoyaltyProgram: """Activate a loyalty program.""" program = self.require_program(db, program_id) program.activate() db.commit() db.refresh(program) logger.info(f"Activated loyalty program {program_id}") return program def deactivate_program(self, db: Session, program_id: int) -> LoyaltyProgram: """Deactivate a loyalty program.""" program = self.require_program(db, program_id) program.deactivate() db.commit() db.refresh(program) logger.info(f"Deactivated loyalty program {program_id}") return program def delete_program(self, db: Session, program_id: int) -> None: """Soft-delete a loyalty program and associated cards.""" from app.core.soft_delete import soft_delete_cascade program = self.require_program(db, program_id) merchant_id = program.merchant_id # Hard delete merchant settings (config data, not business records) db.query(MerchantLoyaltySettings).filter( MerchantLoyaltySettings.merchant_id == merchant_id ).delete() soft_delete_cascade(db, program, deleted_by_id=None, cascade_rels=[ ("cards", []), ]) db.commit() logger.info(f"Soft-deleted loyalty program {program_id} for merchant {merchant_id}") # ========================================================================= # Merchant Settings # ========================================================================= def get_merchant_settings(self, db: Session, merchant_id: int) -> MerchantLoyaltySettings | None: """Get merchant loyalty settings.""" return ( db.query(MerchantLoyaltySettings) .filter(MerchantLoyaltySettings.merchant_id == merchant_id) .first() ) def get_or_create_merchant_settings(self, db: Session, merchant_id: int) -> MerchantLoyaltySettings: """Get or create merchant loyalty settings.""" settings = self.get_merchant_settings(db, merchant_id) if not settings: settings = MerchantLoyaltySettings(merchant_id=merchant_id) db.add(settings) db.commit() db.refresh(settings) return settings # ========================================================================= # Statistics # ========================================================================= def get_program_stats(self, db: Session, program_id: int) -> dict: """ Get statistics for a loyalty program. Returns dict with: - total_cards, active_cards - total_stamps_issued, total_stamps_redeemed - total_points_issued, total_points_redeemed - etc. """ from datetime import timedelta from sqlalchemy import func from app.modules.loyalty.models import LoyaltyCard, LoyaltyTransaction program = self.require_program(db, program_id) # Card counts total_cards = ( db.query(func.count(LoyaltyCard.id)) .filter(LoyaltyCard.program_id == program_id) .scalar() or 0 ) active_cards = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.program_id == program_id, LoyaltyCard.is_active == True, ) .scalar() or 0 ) # Stamp totals from cards stamp_stats = ( db.query( func.sum(LoyaltyCard.total_stamps_earned), func.sum(LoyaltyCard.stamps_redeemed), ) .filter(LoyaltyCard.program_id == program_id) .first() ) total_stamps_issued = stamp_stats[0] or 0 total_stamps_redeemed = stamp_stats[1] or 0 # Points totals from cards points_stats = ( db.query( func.sum(LoyaltyCard.total_points_earned), func.sum(LoyaltyCard.points_redeemed), ) .filter(LoyaltyCard.program_id == program_id) .first() ) total_points_issued = points_stats[0] or 0 total_points_redeemed = points_stats[1] or 0 # This month's activity month_start = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0) stamps_this_month = ( db.query(func.count(LoyaltyTransaction.id)) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.transaction_type == "stamp_earned", LoyaltyTransaction.transaction_at >= month_start, ) .scalar() or 0 ) redemptions_this_month = ( db.query(func.count(LoyaltyTransaction.id)) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.transaction_type == "stamp_redeemed", LoyaltyTransaction.transaction_at >= month_start, ) .scalar() or 0 ) # 30-day active cards thirty_days_ago = datetime.now(UTC) - timedelta(days=30) cards_with_activity_30d = ( db.query(func.count(func.distinct(LoyaltyTransaction.card_id))) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) # Averages avg_stamps = total_stamps_issued / total_cards if total_cards > 0 else 0 avg_points = total_points_issued / total_cards if total_cards > 0 else 0 # New this month (cards created since month start) new_this_month = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.program_id == program_id, LoyaltyCard.created_at >= month_start, ) .scalar() or 0 ) # Points activity this month points_this_month = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.points_delta > 0, LoyaltyTransaction.transaction_at >= month_start, ) .scalar() or 0 ) points_redeemed_this_month = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.points_delta < 0, LoyaltyTransaction.transaction_at >= month_start, ) .scalar() or 0 ) # 30-day transaction metrics transactions_30d = ( db.query(func.count(LoyaltyTransaction.id)) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) points_issued_30d = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.points_delta > 0, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) points_redeemed_30d = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .join(LoyaltyCard) .filter( LoyaltyCard.program_id == program_id, LoyaltyTransaction.points_delta < 0, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) # Estimated liability (unredeemed value) current_stamps = ( db.query(func.sum(LoyaltyCard.stamp_count)) .filter(LoyaltyCard.program_id == program_id) .scalar() or 0 ) stamp_value = program.stamps_reward_value_cents or 0 current_points = ( db.query(func.sum(LoyaltyCard.points_balance)) .filter(LoyaltyCard.program_id == program_id) .scalar() or 0 ) total_points_balance = current_points # Rough estimate: assume 100 points = €1 points_value_cents = current_points // 100 * 100 estimated_liability = ( (current_stamps * stamp_value // program.stamps_target) + points_value_cents ) avg_points_per_member = round(current_points / active_cards, 2) if active_cards > 0 else 0 return { "total_cards": total_cards, "active_cards": active_cards, "new_this_month": new_this_month, "total_stamps_issued": total_stamps_issued, "total_stamps_redeemed": total_stamps_redeemed, "stamps_this_month": stamps_this_month, "redemptions_this_month": redemptions_this_month, "total_points_issued": total_points_issued, "total_points_redeemed": total_points_redeemed, "total_points_balance": total_points_balance, "points_this_month": points_this_month, "points_redeemed_this_month": points_redeemed_this_month, "cards_with_activity_30d": cards_with_activity_30d, "average_stamps_per_card": round(avg_stamps, 2), "average_points_per_card": round(avg_points, 2), "avg_points_per_member": avg_points_per_member, "transactions_30d": transactions_30d, "points_issued_30d": points_issued_30d, "points_redeemed_30d": points_redeemed_30d, "estimated_liability_cents": estimated_liability, } def get_wallet_integration_status(self, db: Session) -> dict: """Get wallet integration status for admin dashboard.""" from app.modules.loyalty.models import LoyaltyCard from app.modules.loyalty.services.apple_wallet_service import ( apple_wallet_service, ) from app.modules.loyalty.services.google_wallet_service import ( google_wallet_service, ) # Google Wallet google_config = google_wallet_service.validate_config() google_classes = [] if google_config["credentials_valid"]: programs_with_class = ( db.query(LoyaltyProgram) .filter(LoyaltyProgram.google_class_id.isnot(None)) .all() ) for prog in programs_with_class: status = google_wallet_service.get_class_status( prog.google_class_id, ) google_classes.append({ "program_id": prog.id, "program_name": prog.display_name, "class_id": prog.google_class_id, "review_status": status["review_status"] if status else "UNKNOWN", }) google_objects = ( db.query(LoyaltyCard) .filter(LoyaltyCard.google_object_id.isnot(None)) .count() ) # Apple Wallet apple_config = apple_wallet_service.validate_config() apple_passes = ( db.query(LoyaltyCard) .filter(LoyaltyCard.apple_serial_number.isnot(None)) .count() ) return { "google_wallet": { **google_config, "classes": google_classes, "total_objects": google_objects, }, "apple_wallet": { **apple_config, "total_passes": apple_passes, }, } def get_merchant_stats(self, db: Session, merchant_id: int) -> dict: """ Get statistics for a merchant's loyalty program across all locations. Returns dict with per-store breakdown. """ from datetime import UTC, datetime, timedelta from sqlalchemy import func from app.modules.loyalty.models import LoyaltyCard, LoyaltyTransaction from app.modules.tenancy.services.store_service import store_service program = self.get_program_by_merchant(db, merchant_id) # Base stats dict stats = { "merchant_id": merchant_id, "program_id": program.id if program else None, "total_cards": 0, "active_cards": 0, "total_points_issued": 0, "total_points_redeemed": 0, "points_issued_30d": 0, "points_redeemed_30d": 0, "transactions_30d": 0, "program": None, "locations": [], } if not program: return stats # Add program info stats["program"] = { "id": program.id, "display_name": program.display_name, "card_name": program.card_name, "loyalty_type": program.loyalty_type.value if hasattr(program.loyalty_type, "value") else str(program.loyalty_type), "points_per_euro": program.points_per_euro, "welcome_bonus_points": program.welcome_bonus_points, "minimum_redemption_points": program.minimum_redemption_points, "points_expiration_days": program.points_expiration_days, "is_active": program.is_active, "stamps_target": program.stamps_target, "stamps_reward_description": program.stamps_reward_description, "stamps_reward_value_cents": program.stamps_reward_value_cents, "minimum_purchase_cents": program.minimum_purchase_cents, "cooldown_minutes": program.cooldown_minutes, "max_daily_stamps": program.max_daily_stamps, "require_staff_pin": program.require_staff_pin, "card_color": program.card_color, "card_secondary_color": program.card_secondary_color, "logo_url": program.logo_url, "hero_image_url": program.hero_image_url, "terms_text": program.terms_text, "privacy_url": program.privacy_url, "points_rewards": program.points_rewards, } thirty_days_ago = datetime.now(UTC) - timedelta(days=30) month_start = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0) # Total cards stats["total_cards"] = ( db.query(func.count(LoyaltyCard.id)) .filter(LoyaltyCard.merchant_id == merchant_id) .scalar() or 0 ) # Active cards stats["active_cards"] = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.merchant_id == merchant_id, LoyaltyCard.is_active == True, ) .scalar() or 0 ) # Total points issued (all time) stats["total_points_issued"] = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.points_delta > 0, ) .scalar() or 0 ) # Total points redeemed (all time) stats["total_points_redeemed"] = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.points_delta < 0, ) .scalar() or 0 ) # Points issued (30 days) stats["points_issued_30d"] = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.points_delta > 0, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) # Points redeemed (30 days) stats["points_redeemed_30d"] = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.points_delta < 0, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) # Transactions (30 days) stats["transactions_30d"] = ( db.query(func.count(LoyaltyTransaction.id)) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) # New members this month stats["new_this_month"] = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.merchant_id == merchant_id, LoyaltyCard.created_at >= month_start, ) .scalar() or 0 ) # Estimated liability (unredeemed value) current_stamps = ( db.query(func.sum(LoyaltyCard.stamp_count)) .filter(LoyaltyCard.merchant_id == merchant_id) .scalar() or 0 ) stamp_value = program.stamps_reward_value_cents or 0 stamps_target = program.stamps_target or 1 current_points = ( db.query(func.sum(LoyaltyCard.points_balance)) .filter(LoyaltyCard.merchant_id == merchant_id) .scalar() or 0 ) points_value_cents = current_points // 100 * 100 stats["estimated_liability_cents"] = ( (current_stamps * stamp_value // stamps_target) + points_value_cents ) # Get all stores for this merchant for location breakdown stores = store_service.get_stores_by_merchant_id(db, merchant_id) location_stats = [] for store in stores: # Cards enrolled at this store enrolled_count = ( db.query(func.count(LoyaltyCard.id)) .filter( LoyaltyCard.merchant_id == merchant_id, LoyaltyCard.enrolled_at_store_id == store.id, ) .scalar() or 0 ) # Points earned at this store points_earned = ( db.query(func.sum(LoyaltyTransaction.points_delta)) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.store_id == store.id, LoyaltyTransaction.points_delta > 0, ) .scalar() or 0 ) # Points redeemed at this store points_redeemed = ( db.query(func.sum(func.abs(LoyaltyTransaction.points_delta))) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.store_id == store.id, LoyaltyTransaction.points_delta < 0, ) .scalar() or 0 ) # Transactions (30 days) at this store transactions_30d = ( db.query(func.count(LoyaltyTransaction.id)) .filter( LoyaltyTransaction.merchant_id == merchant_id, LoyaltyTransaction.store_id == store.id, LoyaltyTransaction.transaction_at >= thirty_days_ago, ) .scalar() or 0 ) location_stats.append({ "store_id": store.id, "store_name": store.name, "store_code": store.store_code, "enrolled_count": enrolled_count, "points_earned": points_earned, "points_redeemed": points_redeemed, "transactions_30d": transactions_30d, }) stats["locations"] = location_stats return stats def restore_deleted_programs(self, db: Session, merchant_id: int) -> int: """Restore all soft-deleted programs for a merchant. Returns number of programs restored. """ from sqlalchemy import update result = db.execute( update(LoyaltyProgram) .where( LoyaltyProgram.merchant_id == merchant_id, LoyaltyProgram.deleted_at.isnot(None), ) .values(deleted_at=None, deleted_by_id=None) .execution_options(include_deleted=True) ) db.commit() count = result.rowcount if count: logger.info( f"Restored {count} soft-deleted programs for merchant {merchant_id}" ) return count # Singleton instance program_service = ProgramService()