sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(loyalty): implement Phase 2 - company-wide points system

Browse Source 
Complete implementation of loyalty module Phase 2 features:

Database & Models:
- Add company_id to LoyaltyProgram for chain-wide loyalty
- Add company_id to LoyaltyCard for multi-location support
- Add CompanyLoyaltySettings model for admin-controlled settings
- Add points expiration, welcome bonus, and minimum redemption fields
- Add POINTS_EXPIRED, WELCOME_BONUS transaction types

Services:
- Update program_service for company-based queries
- Update card_service with enrollment and welcome bonus
- Update points_service with void_points for returns
- Update stamp_service for company context
- Update pin_service for company-wide operations

API Endpoints:
- Admin: Program listing with stats, company detail views
- Vendor: Terminal operations, card management, settings
- Storefront: Customer card/transactions, self-enrollment

UI Templates:
- Admin: Programs dashboard, company detail, settings
- Vendor: Terminal, cards list, card detail, settings, stats, enrollment
- Storefront: Dashboard, history, enrollment, success pages

Background Tasks:
- Point expiration task (daily, based on inactivity)
- Wallet sync task (hourly)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-05 22:10:27 +01:00
parent 3bdf1695fd
commit d8f3338bc8
54 changed files with 7252 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
app/modules/loyalty/services/card_service.py
View File

@@ -2,9 +2,14 @@
"""
Loyalty card service.
Company-based card operations:
- Cards belong to a company's loyalty program
- One card per customer per company
- Can be used at any vendor within the company
Handles card operations including:
- Customer enrollment
- Card lookup (by ID, QR code, card number)
- Customer enrollment (with welcome bonus)
- Card lookup (by ID, QR code, card number, email, phone)
- Card management (activation, deactivation)
"""
@@ -19,7 +24,12 @@ from app.modules.loyalty.exceptions import (
LoyaltyProgramInactiveException,
LoyaltyProgramNotFoundException,
)
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram, LoyaltyTransaction, TransactionType
from app.modules.loyalty.models import (
LoyaltyCard,
LoyaltyProgram,
LoyaltyTransaction,
TransactionType,
)
logger = logging.getLogger(__name__)
@@ -51,10 +61,31 @@ class CardService:
def get_card_by_number(self, db: Session, card_number: str) -> LoyaltyCard | None:
"""Get a loyalty card by card number."""
# Normalize card number (remove dashes)
normalized = card_number.replace("-", "").replace(" ", "")
return (
db.query(LoyaltyCard)
.options(joinedload(LoyaltyCard.program))
.filter(LoyaltyCard.card_number == card_number)
.filter(
LoyaltyCard.card_number.replace("-", "") == normalized
)
.first()
)
def get_card_by_customer_and_company(
self,
db: Session,
customer_id: int,
company_id: int,
) -> LoyaltyCard | None:
"""Get a customer's card for a company's program."""
return (
db.query(LoyaltyCard)
.options(joinedload(LoyaltyCard.program))
.filter(
LoyaltyCard.customer_id == customer_id,
LoyaltyCard.company_id == company_id,
)
.first()
)
@@ -89,6 +120,7 @@ class CardService:
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
company_id: int | None = None,
) -> LoyaltyCard:
"""
Look up a card by any identifier.
@@ -97,7 +129,8 @@ class CardService:
db: Database session
card_id: Card ID
qr_code: QR code data
card_number: Card number
card_number: Card number (with or without dashes)
company_id: Optional company filter
Returns:
Found card
@@ -118,28 +151,73 @@ class CardService:
identifier = card_id or qr_code or card_number or "unknown"
raise LoyaltyCardNotFoundException(str(identifier))
# Filter by company if specified
if company_id and card.company_id != company_id:
raise LoyaltyCardNotFoundException(str(card_id or qr_code or card_number))
return card
def list_cards(
def lookup_card_for_vendor(
self,
db: Session,
vendor_id: int,
*,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
) -> LoyaltyCard:
"""
Look up a card for a specific vendor (must be in same company).
Args:
db: Database session
vendor_id: Vendor ID (to get company context)
card_id: Card ID
qr_code: QR code data
card_number: Card number
Returns:
Found card (verified to be in vendor's company)
Raises:
LoyaltyCardNotFoundException: If no card found or wrong company
"""
from app.modules.tenancy.models import Vendor
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
raise LoyaltyCardNotFoundException("vendor not found")
return self.lookup_card(
db,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
company_id=vendor.company_id,
)
def list_cards(
self,
db: Session,
company_id: int,
*,
vendor_id: int | None = None,
skip: int = 0,
limit: int = 50,
is_active: bool | None = None,
search: str | None = None,
) -> tuple[list[LoyaltyCard], int]:
"""
List loyalty cards for a vendor.
List loyalty cards for a company.
Args:
db: Database session
vendor_id: Vendor ID
company_id: Company ID
vendor_id: Optional filter by enrolled vendor
skip: Pagination offset
limit: Pagination limit
is_active: Filter by active status
search: Search by card number or customer email
search: Search by card number, email, or name
Returns:
(cards, total_count)
@@ -149,18 +227,24 @@ class CardService:
query = (
db.query(LoyaltyCard)
.options(joinedload(LoyaltyCard.customer))
.filter(LoyaltyCard.vendor_id == vendor_id)
.filter(LoyaltyCard.company_id == company_id)
)
if vendor_id:
query = query.filter(LoyaltyCard.enrolled_at_vendor_id == vendor_id)
if is_active is not None:
query = query.filter(LoyaltyCard.is_active == is_active)
if search:
# Normalize search term for card number matching
search_normalized = search.replace("-", "").replace(" ", "")
query = query.join(Customer).filter(
(LoyaltyCard.card_number.ilike(f"%{search}%"))
(LoyaltyCard.card_number.replace("-", "").ilike(f"%{search_normalized}%"))
| (Customer.email.ilike(f"%{search}%"))
| (Customer.first_name.ilike(f"%{search}%"))
| (Customer.last_name.ilike(f"%{search}%"))
| (Customer.phone.ilike(f"%{search}%"))
)
total = query.count()
@@ -181,7 +265,7 @@ class CardService:
"""List all loyalty cards for a customer."""
return (
db.query(LoyaltyCard)
.options(joinedload(LoyaltyCard.program))
.options(joinedload(LoyaltyCard.program), joinedload(LoyaltyCard.company))
.filter(LoyaltyCard.customer_id == customer_id)
.all()
)
@@ -194,18 +278,18 @@ class CardService:
self,
db: Session,
customer_id: int,
vendor_id: int,
company_id: int,
*,
program_id: int | None = None,
enrolled_at_vendor_id: int | None = None,
) -> LoyaltyCard:
"""
Enroll a customer in a loyalty program.
Enroll a customer in a company's loyalty program.
Args:
db: Database session
customer_id: Customer ID
vendor_id: Vendor ID
program_id: Optional program ID (defaults to vendor's program)
company_id: Company ID
enrolled_at_vendor_id: Vendor where customer enrolled (for analytics)
Returns:
Created loyalty card
@@ -216,35 +300,29 @@ class CardService:
LoyaltyCardAlreadyExistsException: If customer already enrolled
"""
# Get the program
if program_id:
program = (
db.query(LoyaltyProgram)
.filter(LoyaltyProgram.id == program_id)
.first()
)
else:
program = (
db.query(LoyaltyProgram)
.filter(LoyaltyProgram.vendor_id == vendor_id)
.first()
)
program = (
db.query(LoyaltyProgram)
.filter(LoyaltyProgram.company_id == company_id)
.first()
)
if not program:
raise LoyaltyProgramNotFoundException(f"vendor:{vendor_id}")
raise LoyaltyProgramNotFoundException(f"company:{company_id}")
if not program.is_active:
raise LoyaltyProgramInactiveException(program.id)
# Check if customer already has a card
existing = self.get_card_by_customer_and_program(db, customer_id, program.id)
existing = self.get_card_by_customer_and_company(db, customer_id, company_id)
if existing:
raise LoyaltyCardAlreadyExistsException(customer_id, program.id)
# Create the card
card = LoyaltyCard(
company_id=company_id,
customer_id=customer_id,
program_id=program.id,
vendor_id=vendor_id,
enrolled_at_vendor_id=enrolled_at_vendor_id,
)
db.add(card)
@@ -252,32 +330,88 @@ class CardService:
# Create enrollment transaction
transaction = LoyaltyTransaction(
company_id=company_id,
card_id=card.id,
vendor_id=vendor_id,
vendor_id=enrolled_at_vendor_id,
transaction_type=TransactionType.CARD_CREATED.value,
transaction_at=datetime.now(UTC),
)
db.add(transaction)
# Award welcome bonus if configured
if program.welcome_bonus_points > 0:
card.add_points(program.welcome_bonus_points)
bonus_transaction = LoyaltyTransaction(
company_id=company_id,
card_id=card.id,
vendor_id=enrolled_at_vendor_id,
transaction_type=TransactionType.WELCOME_BONUS.value,
points_delta=program.welcome_bonus_points,
points_balance_after=card.points_balance,
notes="Welcome bonus on enrollment",
transaction_at=datetime.now(UTC),
)
db.add(bonus_transaction)
db.commit()
db.refresh(card)
logger.info(
f"Enrolled customer {customer_id} in loyalty program {program.id} "
f"(card: {card.card_number})"
f"Enrolled customer {customer_id} in company {company_id} loyalty program "
f"(card: {card.card_number}, bonus: {program.welcome_bonus_points} pts)"
)
return card
def deactivate_card(self, db: Session, card_id: int) -> LoyaltyCard:
def enroll_customer_for_vendor(
self,
db: Session,
customer_id: int,
vendor_id: int,
) -> LoyaltyCard:
"""
Enroll a customer through a specific vendor.
Looks up the vendor's company and enrolls in the company's program.
Args:
db: Database session
customer_id: Customer ID
vendor_id: Vendor ID
Returns:
Created loyalty card
"""
from app.modules.tenancy.models import Vendor
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
raise LoyaltyProgramNotFoundException(f"vendor:{vendor_id}")
return self.enroll_customer(
db,
customer_id,
vendor.company_id,
enrolled_at_vendor_id=vendor_id,
)
def deactivate_card(
self,
db: Session,
card_id: int,
*,
vendor_id: int | None = None,
) -> LoyaltyCard:
"""Deactivate a loyalty card."""
card = self.require_card(db, card_id)
card.is_active = False
# Create deactivation transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
transaction_type=TransactionType.CARD_DEACTIVATED.value,
transaction_at=datetime.now(UTC),
)
@@ -334,6 +468,7 @@ class CardService:
"""Get transaction history for a card."""
query = (
db.query(LoyaltyTransaction)
.options(joinedload(LoyaltyTransaction.vendor))
.filter(LoyaltyTransaction.card_id == card_id)
.order_by(LoyaltyTransaction.transaction_at.desc())
)

106
app/modules/loyalty/services/pin_service.py
View File

@@ -2,9 +2,14 @@
"""
Staff PIN service.
Company-based PIN operations:
- PINs belong to a company's loyalty program
- Each vendor (location) has its own set of staff PINs
- Staff can only use PINs at their assigned location
Handles PIN operations including:
- PIN creation and management
- PIN verification with lockout
- PIN verification with lockout (per vendor)
- PIN security (failed attempts, lockout)
"""
@@ -41,16 +46,17 @@ class PinService:
db: Session,
program_id: int,
staff_id: str,
*,
vendor_id: int | None = None,
) -> StaffPin | None:
"""Get a staff PIN by employee ID."""
return (
db.query(StaffPin)
.filter(
StaffPin.program_id == program_id,
StaffPin.staff_id == staff_id,
)
.first()
query = db.query(StaffPin).filter(
StaffPin.program_id == program_id,
StaffPin.staff_id == staff_id,
)
if vendor_id:
query = query.filter(StaffPin.vendor_id == vendor_id)
return query.first()
def require_pin(self, db: Session, pin_id: int) -> StaffPin:
"""Get a PIN or raise exception if not found."""
@@ -64,16 +70,61 @@ class PinService:
db: Session,
program_id: int,
*,
vendor_id: int | None = None,
is_active: bool | None = None,
) -> list[StaffPin]:
"""List all staff PINs for a program."""
"""
List staff PINs for a program.
Args:
db: Database session
program_id: Program ID
vendor_id: Optional filter by vendor (location)
is_active: Filter by active status
Returns:
List of StaffPin objects
"""
query = db.query(StaffPin).filter(StaffPin.program_id == program_id)
if vendor_id is not None:
query = query.filter(StaffPin.vendor_id == vendor_id)
if is_active is not None:
query = query.filter(StaffPin.is_active == is_active)
return query.order_by(StaffPin.name).all()
def list_pins_for_company(
self,
db: Session,
company_id: int,
*,
vendor_id: int | None = None,
is_active: bool | None = None,
) -> list[StaffPin]:
"""
List staff PINs for a company.
Args:
db: Database session
company_id: Company ID
vendor_id: Optional filter by vendor (location)
is_active: Filter by active status
Returns:
List of StaffPin objects
"""
query = db.query(StaffPin).filter(StaffPin.company_id == company_id)
if vendor_id is not None:
query = query.filter(StaffPin.vendor_id == vendor_id)
if is_active is not None:
query = query.filter(StaffPin.is_active == is_active)
return query.order_by(StaffPin.vendor_id, StaffPin.name).all()
# =========================================================================
# Write Operations
# =========================================================================
@@ -91,13 +142,21 @@ class PinService:
Args:
db: Database session
program_id: Program ID
vendor_id: Vendor ID
vendor_id: Vendor ID (location where staff works)
data: PIN creation data
Returns:
Created PIN
"""
from app.modules.loyalty.models import LoyaltyProgram
# Get company_id from program
program = db.query(LoyaltyProgram).filter(LoyaltyProgram.id == program_id).first()
if not program:
raise StaffPinNotFoundException(f"program:{program_id}")
pin = StaffPin(
company_id=program.company_id,
program_id=program_id,
vendor_id=vendor_id,
name=data.name,
@@ -109,7 +168,9 @@ class PinService:
db.commit()
db.refresh(pin)
logger.info(f"Created staff PIN {pin.id} for '{pin.name}' in program {program_id}")
logger.info(
f"Created staff PIN {pin.id} for '{pin.name}' at vendor {vendor_id}"
)
return pin
@@ -158,11 +219,12 @@ class PinService:
"""Delete a staff PIN."""
pin = self.require_pin(db, pin_id)
program_id = pin.program_id
vendor_id = pin.vendor_id
db.delete(pin)
db.commit()
logger.info(f"Deleted staff PIN {pin_id} from program {program_id}")
logger.info(f"Deleted staff PIN {pin_id} from vendor {vendor_id}")
def unlock_pin(self, db: Session, pin_id: int) -> StaffPin:
"""Unlock a locked staff PIN."""
@@ -184,16 +246,21 @@ class PinService:
db: Session,
program_id: int,
plain_pin: str,
*,
vendor_id: int | None = None,
) -> StaffPin:
"""
Verify a staff PIN.
Checks all active PINs for the program and returns the matching one.
For company-wide programs, if vendor_id is provided, only checks
PINs assigned to that vendor. This ensures staff can only use
their PIN at their assigned location.
Args:
db: Database session
program_id: Program ID
plain_pin: Plain text PIN to verify
vendor_id: Optional vendor ID to restrict PIN lookup
Returns:
Verified StaffPin object
@@ -202,8 +269,8 @@ class PinService:
InvalidStaffPinException: PIN is invalid
StaffPinLockedException: PIN is locked
"""
# Get all active PINs for the program
pins = self.list_pins(db, program_id, is_active=True)
# Get active PINs (optionally filtered by vendor)
pins = self.list_pins(db, program_id, vendor_id=vendor_id, is_active=True)
if not pins:
raise InvalidStaffPinException()
@@ -220,7 +287,9 @@ class PinService:
pin.record_success()
db.commit()
logger.debug(f"PIN verified for '{pin.name}' in program {program_id}")
logger.debug(
f"PIN verified for '{pin.name}' at vendor {pin.vendor_id}"
)
return pin
@@ -254,6 +323,8 @@ class PinService:
db: Session,
program_id: int,
plain_pin: str,
*,
vendor_id: int | None = None,
) -> StaffPin | None:
"""
Find a matching PIN without recording attempts.
@@ -264,11 +335,12 @@ class PinService:
db: Database session
program_id: Program ID
plain_pin: Plain text PIN to check
vendor_id: Optional vendor ID to restrict lookup
Returns:
Matching StaffPin or None
"""
pins = self.list_pins(db, program_id, is_active=True)
pins = self.list_pins(db, program_id, vendor_id=vendor_id, is_active=True)
for pin in pins:
if not pin.is_locked and pin.verify_pin(plain_pin):

203
app/modules/loyalty/services/points_service.py
View File

@@ -2,9 +2,15 @@
"""
Points service.
Company-based points operations:
- Points earned at any vendor count toward company total
- Points can be redeemed at any vendor within the company
- Supports voiding points for returns
Handles points operations including:
- Earning points from purchases
- Redeeming points for rewards
- Voiding points (for returns)
- Points balance management
"""
@@ -34,6 +40,7 @@ class PointsService:
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
@@ -51,6 +58,7 @@ class PointsService:
Args:
db: Database session
vendor_id: Vendor ID (where purchase is being made)
card_id: Card ID
qr_code: QR code data
card_number: Card number
@@ -64,9 +72,10 @@ class PointsService:
Returns:
Dict with operation result
"""
# Look up the card
card = card_service.lookup_card(
# Look up the card (validates it belongs to vendor's company)
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
@@ -85,12 +94,26 @@ class PointsService:
logger.warning(f"Points attempted on stamps-only program {program.id}")
raise LoyaltyCardInactiveException(card.id)
# Check minimum purchase amount
if program.minimum_purchase_cents > 0 and purchase_amount_cents < program.minimum_purchase_cents:
return {
"success": True,
"message": f"Purchase below minimum of €{program.minimum_purchase_cents/100:.2f}",
"points_earned": 0,
"points_per_euro": program.points_per_euro,
"purchase_amount_cents": purchase_amount_cents,
"card_id": card.id,
"card_number": card.card_number,
"points_balance": card.points_balance,
"total_points_earned": card.total_points_earned,
}
# Verify staff PIN if required
verified_pin = None
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin)
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Calculate points
# points_per_euro is per full euro, so divide cents by 100
@@ -115,11 +138,13 @@ class PointsService:
card.points_balance += points_earned
card.total_points_earned += points_earned
card.last_points_at = now
card.last_activity_at = now
# Create transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.POINTS_EARNED.value,
points_delta=points_earned,
@@ -138,7 +163,7 @@ class PointsService:
db.refresh(card)
logger.info(
f"Added {points_earned} points to card {card.id} "
f"Added {points_earned} points to card {card.id} at vendor {vendor_id} "
f"(purchase: €{purchase_euros:.2f}, balance: {card.points_balance})"
)
@@ -152,12 +177,14 @@ class PointsService:
"card_number": card.card_number,
"points_balance": card.points_balance,
"total_points_earned": card.total_points_earned,
"vendor_id": vendor_id,
}
def redeem_points(
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
@@ -172,6 +199,7 @@ class PointsService:
Args:
db: Database session
vendor_id: Vendor ID (where redemption is happening)
card_id: Card ID
qr_code: QR code data
card_number: Card number
@@ -188,9 +216,10 @@ class PointsService:
InvalidRewardException: Reward not found or inactive
InsufficientPointsException: Not enough points
"""
# Look up the card
card = card_service.lookup_card(
# Look up the card (validates it belongs to vendor's company)
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
@@ -215,6 +244,10 @@ class PointsService:
points_required = reward["points_required"]
reward_name = reward["name"]
# Check minimum redemption
if points_required < program.minimum_redemption_points:
raise InvalidRewardException(reward_id)
# Check if enough points
if card.points_balance < points_required:
raise InsufficientPointsException(card.points_balance, points_required)
@@ -224,18 +257,20 @@ class PointsService:
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin)
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Redeem points
now = datetime.now(UTC)
card.points_balance -= points_required
card.points_redeemed += points_required
card.last_redemption_at = now
card.last_activity_at = now
# Create transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.POINTS_REDEEMED.value,
points_delta=-points_required,
@@ -254,7 +289,7 @@ class PointsService:
db.refresh(card)
logger.info(
f"Redeemed {points_required} points from card {card.id} "
f"Redeemed {points_required} points from card {card.id} at vendor {vendor_id} "
f"(reward: {reward_name}, balance: {card.points_balance})"
)
@@ -268,6 +303,140 @@ class PointsService:
"card_number": card.card_number,
"points_balance": card.points_balance,
"total_points_redeemed": card.points_redeemed,
"vendor_id": vendor_id,
}
def void_points(
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
points_to_void: int | None = None,
original_transaction_id: int | None = None,
order_reference: str | None = None,
staff_pin: str | None = None,
ip_address: str | None = None,
user_agent: str | None = None,
notes: str | None = None,
) -> dict:
"""
Void points for a return.
Args:
db: Database session
vendor_id: Vendor ID
card_id: Card ID
qr_code: QR code data
card_number: Card number
points_to_void: Number of points to void (if not using original_transaction_id)
original_transaction_id: ID of original earn transaction to void
order_reference: Order reference (to find original transaction)
staff_pin: Staff PIN for verification
ip_address: Request IP for audit
user_agent: Request user agent for audit
notes: Reason for voiding
Returns:
Dict with operation result
"""
# Look up the card
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
)
program = card.program
# Verify staff PIN if required
verified_pin = None
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Determine points to void
original_transaction = None
if original_transaction_id:
original_transaction = (
db.query(LoyaltyTransaction)
.filter(
LoyaltyTransaction.id == original_transaction_id,
LoyaltyTransaction.card_id == card.id,
LoyaltyTransaction.transaction_type == TransactionType.POINTS_EARNED.value,
)
.first()
)
if original_transaction:
points_to_void = original_transaction.points_delta
elif order_reference:
original_transaction = (
db.query(LoyaltyTransaction)
.filter(
LoyaltyTransaction.order_reference == order_reference,
LoyaltyTransaction.card_id == card.id,
LoyaltyTransaction.transaction_type == TransactionType.POINTS_EARNED.value,
)
.first()
)
if original_transaction:
points_to_void = original_transaction.points_delta
if not points_to_void or points_to_void <= 0:
return {
"success": False,
"message": "No points to void",
"card_id": card.id,
"card_number": card.card_number,
"points_balance": card.points_balance,
}
# Void the points (can reduce balance below what was earned)
now = datetime.now(UTC)
actual_voided = min(points_to_void, card.points_balance)
card.points_balance = max(0, card.points_balance - points_to_void)
card.last_activity_at = now
# Create void transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.POINTS_VOIDED.value,
points_delta=-actual_voided,
stamps_balance_after=card.stamp_count,
points_balance_after=card.points_balance,
related_transaction_id=original_transaction.id if original_transaction else None,
order_reference=order_reference,
ip_address=ip_address,
user_agent=user_agent,
notes=notes or "Points voided for return",
transaction_at=now,
)
db.add(transaction)
db.commit()
db.refresh(card)
logger.info(
f"Voided {actual_voided} points from card {card.id} at vendor {vendor_id} "
f"(balance: {card.points_balance})"
)
return {
"success": True,
"message": "Points voided successfully",
"points_voided": actual_voided,
"card_id": card.id,
"card_number": card.card_number,
"points_balance": card.points_balance,
"vendor_id": vendor_id,
}
def adjust_points(
@@ -276,18 +445,20 @@ class PointsService:
card_id: int,
points_delta: int,
*,
vendor_id: int | None = None,
reason: str,
staff_pin: str | None = None,
ip_address: str | None = None,
user_agent: str | None = None,
) -> dict:
"""
Manually adjust points (admin operation).
Manually adjust points (admin/vendor operation).
Args:
db: Database session
card_id: Card ID
points_delta: Points to add (positive) or remove (negative)
vendor_id: Vendor ID
reason: Reason for adjustment
staff_pin: Staff PIN for verification
ip_address: Request IP for audit
@@ -299,14 +470,15 @@ class PointsService:
card = card_service.require_card(db, card_id)
program = card.program
# Verify staff PIN if required
# Verify staff PIN if required and vendor provided
verified_pin = None
if program.require_staff_pin and staff_pin:
verified_pin = pin_service.verify_pin(db, program.id, staff_pin)
if program.require_staff_pin and staff_pin and vendor_id:
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Apply adjustment
now = datetime.now(UTC)
card.points_balance += points_delta
card.last_activity_at = now
if points_delta > 0:
card.total_points_earned += points_delta
@@ -320,8 +492,9 @@ class PointsService:
# Create transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.POINTS_ADJUSTMENT.value,
points_delta=points_delta,

332
app/modules/loyalty/services/program_service.py
View File

@@ -2,6 +2,11 @@
"""
Loyalty program service.
Company-based program management:
- Programs belong to companies, not individual vendors
- All vendors under a company share the same loyalty program
- One program per company
Handles CRUD operations for loyalty programs including:
- Program creation and configuration
- Program updates
@@ -18,7 +23,11 @@ from app.modules.loyalty.exceptions import (
LoyaltyProgramAlreadyExistsException,
LoyaltyProgramNotFoundException,
)
from app.modules.loyalty.models import LoyaltyProgram, LoyaltyType
from app.modules.loyalty.models import (
LoyaltyProgram,
LoyaltyType,
CompanyLoyaltySettings,
)
from app.modules.loyalty.schemas.program import (
ProgramCreate,
ProgramUpdate,
@@ -42,25 +51,53 @@ class ProgramService:
.first()
)
def get_program_by_vendor(self, db: Session, vendor_id: int) -> LoyaltyProgram | None:
"""Get a vendor's loyalty program."""
def get_program_by_company(self, db: Session, company_id: int) -> LoyaltyProgram | None:
"""Get a company's loyalty program."""
return (
db.query(LoyaltyProgram)
.filter(LoyaltyProgram.vendor_id == vendor_id)
.filter(LoyaltyProgram.company_id == company_id)
.first()
)
def get_active_program_by_vendor(self, db: Session, vendor_id: int) -> LoyaltyProgram | None:
"""Get a vendor's active loyalty program."""
def get_active_program_by_company(self, db: Session, company_id: int) -> LoyaltyProgram | None:
"""Get a company's active loyalty program."""
return (
db.query(LoyaltyProgram)
.filter(
LoyaltyProgram.vendor_id == vendor_id,
LoyaltyProgram.company_id == company_id,
LoyaltyProgram.is_active == True,
)
.first()
)
def get_program_by_vendor(self, db: Session, vendor_id: int) -> LoyaltyProgram | None:
"""
Get the loyalty program for a vendor.
Looks up the vendor's company and returns the company's program.
"""
from app.modules.tenancy.models import Vendor
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
return None
return self.get_program_by_company(db, vendor.company_id)
def get_active_program_by_vendor(self, db: Session, vendor_id: int) -> LoyaltyProgram | None:
"""
Get the active loyalty program for a vendor.
Looks up the vendor's company and returns the company's active program.
"""
from app.modules.tenancy.models import Vendor
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
return None
return self.get_active_program_by_company(db, vendor.company_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)
@@ -68,6 +105,13 @@ class ProgramService:
raise LoyaltyProgramNotFoundException(str(program_id))
return program
def require_program_by_company(self, db: Session, company_id: int) -> LoyaltyProgram:
"""Get a company's program or raise exception if not found."""
program = self.get_program_by_company(db, company_id)
if not program:
raise LoyaltyProgramNotFoundException(f"company:{company_id}")
return program
def require_program_by_vendor(self, db: Session, vendor_id: int) -> LoyaltyProgram:
"""Get a vendor's program or raise exception if not found."""
program = self.get_program_by_vendor(db, vendor_id)
@@ -82,15 +126,32 @@ class ProgramService:
skip: int = 0,
limit: int = 100,
is_active: bool | None = None,
search: str | None = None,
) -> tuple[list[LoyaltyProgram], int]:
"""List all loyalty programs (admin)."""
query = db.query(LoyaltyProgram)
"""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 company name (case-insensitive)
"""
from app.modules.tenancy.models import Company
query = db.query(LoyaltyProgram).join(
Company, LoyaltyProgram.company_id == Company.id
)
if is_active is not None:
query = query.filter(LoyaltyProgram.is_active == is_active)
if search:
search_pattern = f"%{search}%"
query = query.filter(Company.name.ilike(search_pattern))
total = query.count()
programs = query.offset(skip).limit(limit).all()
programs = query.order_by(LoyaltyProgram.created_at.desc()).offset(skip).limit(limit).all()
return programs, total
@@ -101,33 +162,33 @@ class ProgramService:
def create_program(
self,
db: Session,
vendor_id: int,
company_id: int,
data: ProgramCreate,
) -> LoyaltyProgram:
"""
Create a new loyalty program for a vendor.
Create a new loyalty program for a company.
Args:
db: Database session
vendor_id: Vendor ID
company_id: Company ID
data: Program configuration
Returns:
Created program
Raises:
LoyaltyProgramAlreadyExistsException: If vendor already has a program
LoyaltyProgramAlreadyExistsException: If company already has a program
"""
# Check if vendor already has a program
existing = self.get_program_by_vendor(db, vendor_id)
# Check if company already has a program
existing = self.get_program_by_company(db, company_id)
if existing:
raise LoyaltyProgramAlreadyExistsException(vendor_id)
raise LoyaltyProgramAlreadyExistsException(company_id)
# Convert points_rewards to dict list for JSON storage
points_rewards_data = [r.model_dump() for r in data.points_rewards]
program = LoyaltyProgram(
vendor_id=vendor_id,
company_id=company_id,
loyalty_type=data.loyalty_type,
# Stamps
stamps_target=data.stamps_target,
@@ -136,6 +197,10 @@ class ProgramService:
# 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,
@@ -155,11 +220,19 @@ class ProgramService:
)
db.add(program)
db.flush()
# Create default company settings
settings = CompanyLoyaltySettings(
company_id=company_id,
)
db.add(settings)
db.commit()
db.refresh(program)
logger.info(
f"Created loyalty program {program.id} for vendor {vendor_id} "
f"Created loyalty program {program.id} for company {company_id} "
f"(type: {program.loyalty_type})"
)
@@ -224,12 +297,39 @@ class ProgramService:
def delete_program(self, db: Session, program_id: int) -> None:
"""Delete a loyalty program and all associated data."""
program = self.require_program(db, program_id)
vendor_id = program.vendor_id
company_id = program.company_id
# Also delete company settings
db.query(CompanyLoyaltySettings).filter(
CompanyLoyaltySettings.company_id == company_id
).delete()
db.delete(program)
db.commit()
logger.info(f"Deleted loyalty program {program_id} for vendor {vendor_id}")
logger.info(f"Deleted loyalty program {program_id} for company {company_id}")
# =========================================================================
# Company Settings
# =========================================================================
def get_company_settings(self, db: Session, company_id: int) -> CompanyLoyaltySettings | None:
"""Get company loyalty settings."""
return (
db.query(CompanyLoyaltySettings)
.filter(CompanyLoyaltySettings.company_id == company_id)
.first()
)
def get_or_create_company_settings(self, db: Session, company_id: int) -> CompanyLoyaltySettings:
"""Get or create company loyalty settings."""
settings = self.get_company_settings(db, company_id)
if not settings:
settings = CompanyLoyaltySettings(company_id=company_id)
db.add(settings)
db.commit()
db.refresh(settings)
return settings
# =========================================================================
# Statistics
@@ -374,6 +474,196 @@ class ProgramService:
"estimated_liability_cents": estimated_liability,
}
def get_company_stats(self, db: Session, company_id: int) -> dict:
"""
Get statistics for a company's loyalty program across all locations.
Returns dict with per-vendor breakdown.
"""
from datetime import UTC, datetime, timedelta
from sqlalchemy import func
from app.modules.loyalty.models import LoyaltyCard, LoyaltyTransaction
from app.modules.tenancy.models import Vendor
program = self.get_program_by_company(db, company_id)
# Base stats dict
stats = {
"company_id": company_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,
}
thirty_days_ago = datetime.now(UTC) - timedelta(days=30)
# Total cards
stats["total_cards"] = (
db.query(func.count(LoyaltyCard.id))
.filter(LoyaltyCard.company_id == company_id)
.scalar()
or 0
)
# Active cards
stats["active_cards"] = (
db.query(func.count(LoyaltyCard.id))
.filter(
LoyaltyCard.company_id == company_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.company_id == company_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.company_id == company_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.company_id == company_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.company_id == company_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.company_id == company_id,
LoyaltyTransaction.transaction_at >= thirty_days_ago,
)
.scalar()
or 0
)
# Get all vendors for this company for location breakdown
vendors = db.query(Vendor).filter(Vendor.company_id == company_id).all()
location_stats = []
for vendor in vendors:
# Cards enrolled at this vendor
enrolled_count = (
db.query(func.count(LoyaltyCard.id))
.filter(
LoyaltyCard.company_id == company_id,
LoyaltyCard.enrolled_at_vendor_id == vendor.id,
)
.scalar()
or 0
)
# Points earned at this vendor
points_earned = (
db.query(func.sum(LoyaltyTransaction.points_delta))
.filter(
LoyaltyTransaction.company_id == company_id,
LoyaltyTransaction.vendor_id == vendor.id,
LoyaltyTransaction.points_delta > 0,
)
.scalar()
or 0
)
# Points redeemed at this vendor
points_redeemed = (
db.query(func.sum(func.abs(LoyaltyTransaction.points_delta)))
.filter(
LoyaltyTransaction.company_id == company_id,
LoyaltyTransaction.vendor_id == vendor.id,
LoyaltyTransaction.points_delta < 0,
)
.scalar()
or 0
)
# Transactions (30 days) at this vendor
transactions_30d = (
db.query(func.count(LoyaltyTransaction.id))
.filter(
LoyaltyTransaction.company_id == company_id,
LoyaltyTransaction.vendor_id == vendor.id,
LoyaltyTransaction.transaction_at >= thirty_days_ago,
)
.scalar()
or 0
)
location_stats.append({
"vendor_id": vendor.id,
"vendor_name": vendor.name,
"vendor_code": vendor.vendor_code,
"enrolled_count": enrolled_count,
"points_earned": points_earned,
"points_redeemed": points_redeemed,
"transactions_30d": transactions_30d,
})
stats["locations"] = location_stats
return stats
# Singleton instance
program_service = ProgramService()

156
app/modules/loyalty/services/stamp_service.py
View File

@@ -2,9 +2,15 @@
"""
Stamp service.
Company-based stamp operations:
- Stamps earned at any vendor count toward company total
- Stamps can be redeemed at any vendor within the company
- Supports voiding stamps for returns
Handles stamp operations including:
- Adding stamps with anti-fraud checks
- Redeeming stamps for rewards
- Voiding stamps (for returns)
- Daily limit tracking
"""
@@ -36,6 +42,7 @@ class StampService:
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
@@ -54,6 +61,7 @@ class StampService:
Args:
db: Database session
vendor_id: Vendor ID (where stamp is being added)
card_id: Card ID
qr_code: QR code data
card_number: Card number
@@ -74,9 +82,10 @@ class StampService:
StampCooldownException: Cooldown period not elapsed
DailyStampLimitException: Daily limit reached
"""
# Look up the card
card = card_service.lookup_card(
# Look up the card (validates it belongs to vendor's company)
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
@@ -100,7 +109,7 @@ class StampService:
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin)
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Check cooldown
now = datetime.now(UTC)
@@ -121,14 +130,16 @@ class StampService:
card.stamp_count += 1
card.total_stamps_earned += 1
card.last_stamp_at = now
card.last_activity_at = now
# Check if reward earned
reward_earned = card.stamp_count >= program.stamps_target
# Create transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.STAMP_EARNED.value,
stamps_delta=1,
@@ -147,7 +158,7 @@ class StampService:
stamps_today += 1
logger.info(
f"Added stamp to card {card.id} "
f"Added stamp to card {card.id} at vendor {vendor_id} "
f"(stamps: {card.stamp_count}/{program.stamps_target}, "
f"today: {stamps_today}/{program.max_daily_stamps})"
)
@@ -168,12 +179,14 @@ class StampService:
"next_stamp_available_at": next_stamp_at,
"stamps_today": stamps_today,
"stamps_remaining_today": max(0, program.max_daily_stamps - stamps_today),
"vendor_id": vendor_id,
}
def redeem_stamps(
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
@@ -187,6 +200,7 @@ class StampService:
Args:
db: Database session
vendor_id: Vendor ID (where redemption is happening)
card_id: Card ID
qr_code: QR code data
card_number: Card number
@@ -203,9 +217,10 @@ class StampService:
InsufficientStampsException: Not enough stamps
StaffPinRequiredException: PIN required but not provided
"""
# Look up the card
card = card_service.lookup_card(
# Look up the card (validates it belongs to vendor's company)
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
@@ -228,7 +243,7 @@ class StampService:
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin)
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Redeem stamps
now = datetime.now(UTC)
@@ -236,11 +251,13 @@ class StampService:
card.stamp_count -= stamps_redeemed
card.stamps_redeemed += 1
card.last_redemption_at = now
card.last_activity_at = now
# Create transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=card.vendor_id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.STAMP_REDEEMED.value,
stamps_delta=-stamps_redeemed,
@@ -258,7 +275,7 @@ class StampService:
db.refresh(card)
logger.info(
f"Redeemed stamps from card {card.id} "
f"Redeemed stamps from card {card.id} at vendor {vendor_id} "
f"(reward: {program.stamps_reward_description}, "
f"total redemptions: {card.stamps_redeemed})"
)
@@ -272,6 +289,125 @@ class StampService:
"stamps_target": program.stamps_target,
"reward_description": program.stamps_reward_description,
"total_redemptions": card.stamps_redeemed,
"vendor_id": vendor_id,
}
def void_stamps(
self,
db: Session,
*,
vendor_id: int,
card_id: int | None = None,
qr_code: str | None = None,
card_number: str | None = None,
stamps_to_void: int | None = None,
original_transaction_id: int | None = None,
staff_pin: str | None = None,
ip_address: str | None = None,
user_agent: str | None = None,
notes: str | None = None,
) -> dict:
"""
Void stamps for a return.
Args:
db: Database session
vendor_id: Vendor ID
card_id: Card ID
qr_code: QR code data
card_number: Card number
stamps_to_void: Number of stamps to void (if not using original_transaction_id)
original_transaction_id: ID of original stamp transaction to void
staff_pin: Staff PIN for verification
ip_address: Request IP for audit
user_agent: Request user agent for audit
notes: Reason for voiding
Returns:
Dict with operation result
"""
# Look up the card
card = card_service.lookup_card_for_vendor(
db,
vendor_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
)
program = card.program
# Verify staff PIN if required
verified_pin = None
if program.require_staff_pin:
if not staff_pin:
raise StaffPinRequiredException()
verified_pin = pin_service.verify_pin(db, program.id, staff_pin, vendor_id=vendor_id)
# Determine stamps to void
original_transaction = None
if original_transaction_id:
original_transaction = (
db.query(LoyaltyTransaction)
.filter(
LoyaltyTransaction.id == original_transaction_id,
LoyaltyTransaction.card_id == card.id,
LoyaltyTransaction.transaction_type == TransactionType.STAMP_EARNED.value,
)
.first()
)
if original_transaction:
stamps_to_void = original_transaction.stamps_delta
if not stamps_to_void or stamps_to_void <= 0:
return {
"success": False,
"message": "No stamps to void",
"card_id": card.id,
"card_number": card.card_number,
"stamp_count": card.stamp_count,
}
# Void the stamps (can reduce balance below what was earned)
now = datetime.now(UTC)
actual_voided = min(stamps_to_void, card.stamp_count)
card.stamp_count = max(0, card.stamp_count - stamps_to_void)
card.last_activity_at = now
# Create void transaction
transaction = LoyaltyTransaction(
company_id=card.company_id,
card_id=card.id,
vendor_id=vendor_id,
staff_pin_id=verified_pin.id if verified_pin else None,
transaction_type=TransactionType.STAMP_VOIDED.value,
stamps_delta=-actual_voided,
stamps_balance_after=card.stamp_count,
points_balance_after=card.points_balance,
related_transaction_id=original_transaction.id if original_transaction else None,
ip_address=ip_address,
user_agent=user_agent,
notes=notes or "Stamps voided for return",
transaction_at=now,
)
db.add(transaction)
db.commit()
db.refresh(card)
logger.info(
f"Voided {actual_voided} stamps from card {card.id} at vendor {vendor_id} "
f"(balance: {card.stamp_count})"
)
return {
"success": True,
"message": "Stamps voided successfully",
"stamps_voided": actual_voided,
"card_id": card.id,
"card_number": card.card_number,
"stamp_count": card.stamp_count,
"vendor_id": vendor_id,
}