Samir Boulahtit
64fe58c171
Some checks failed
The terminal redeem failed with "card not found: unknown" because CardLookupResponse used card_id while CardDetailResponse (from refreshCard) used id. After refresh, selectedCard.card_id was undefined. Fix: standardize on 'id' everywhere (the universal convention): - CardLookupResponse: card_id → id - _build_card_lookup_response: card_id= → id= - loyalty-terminal.js: selectedCard.card_id → selectedCard.id (5 refs) - Removed the card_id/model_validator approach as unnecessary Also fixes Chart.js recursion error on analytics page (inline CDN script instead of optional-libs.html include which caused infinite template recursion in test context). 342 tests pass. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
895 lines
29 KiB
Python
895 lines
29 KiB
Python
# app/modules/loyalty/routes/api/store.py
|
|
"""
|
|
Loyalty module store routes.
|
|
|
|
Merchant-based store endpoints for:
|
|
- Program management (merchant-wide, managed by store)
|
|
- Staff PINs (per-store)
|
|
- Card operations (stamps, points, redemptions, voids)
|
|
- Customer cards lookup
|
|
- Dashboard stats
|
|
|
|
All operations are scoped to the store's merchant.
|
|
Cards can be used at any store within the same merchant.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Path, Query, Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_store_api, require_module_access
|
|
from app.core.database import get_db
|
|
from app.exceptions.base import AuthorizationException
|
|
from app.modules.enums import FrontendType
|
|
from app.modules.loyalty.schemas import (
|
|
CardDetailResponse,
|
|
CardEnrollRequest,
|
|
CardListResponse,
|
|
CardLookupResponse,
|
|
CardResponse,
|
|
MerchantStatsResponse,
|
|
PinCreate,
|
|
PinListResponse,
|
|
PinResponse,
|
|
PinUpdate,
|
|
PointsAdjustRequest,
|
|
PointsAdjustResponse,
|
|
PointsEarnRequest,
|
|
PointsEarnResponse,
|
|
PointsRedeemRequest,
|
|
PointsRedeemResponse,
|
|
PointsVoidRequest,
|
|
PointsVoidResponse,
|
|
ProgramCreate,
|
|
ProgramResponse,
|
|
ProgramStatsResponse,
|
|
ProgramUpdate,
|
|
StampRedeemRequest,
|
|
StampRedeemResponse,
|
|
StampRequest,
|
|
StampResponse,
|
|
StampVoidRequest,
|
|
StampVoidResponse,
|
|
TransactionListResponse,
|
|
TransactionResponse,
|
|
)
|
|
from app.modules.loyalty.services import (
|
|
card_service,
|
|
pin_service,
|
|
points_service,
|
|
program_service,
|
|
stamp_service,
|
|
)
|
|
from app.modules.tenancy.models import User # API-007
|
|
from middleware.decorators import rate_limit
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Store router with module access control
|
|
router = APIRouter(
|
|
prefix="/loyalty",
|
|
tags=["Loyalty - Store"],
|
|
dependencies=[Depends(require_module_access("loyalty", FrontendType.STORE))],
|
|
)
|
|
|
|
|
|
def get_client_info(request: Request) -> tuple[str | None, str | None]:
|
|
"""Extract client IP and user agent from request."""
|
|
ip = request.client.host if request.client else None
|
|
user_agent = request.headers.get("user-agent")
|
|
return ip, user_agent
|
|
|
|
|
|
def get_store_merchant_id(db: Session, store_id: int) -> int:
|
|
"""Get the merchant ID for a store."""
|
|
return program_service.get_store_merchant_id(db, store_id)
|
|
|
|
|
|
# =============================================================================
|
|
# Program Management
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/program", response_model=ProgramResponse)
|
|
def get_program(
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get the merchant's loyalty program."""
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
|
|
response = ProgramResponse.model_validate(program)
|
|
response.is_stamps_enabled = program.is_stamps_enabled
|
|
response.is_points_enabled = program.is_points_enabled
|
|
response.display_name = program.display_name
|
|
|
|
return response
|
|
|
|
|
|
@router.post("/program", response_model=ProgramResponse, status_code=201)
|
|
def create_program(
|
|
data: ProgramCreate,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Create a loyalty program (merchant_owner only)."""
|
|
if current_user.role != "merchant_owner":
|
|
raise AuthorizationException("Only merchant owners can create programs")
|
|
|
|
store_id = current_user.token_store_id
|
|
merchant_id = get_store_merchant_id(db, store_id)
|
|
|
|
program = program_service.create_program(db, merchant_id, data)
|
|
|
|
response = ProgramResponse.model_validate(program)
|
|
response.is_stamps_enabled = program.is_stamps_enabled
|
|
response.is_points_enabled = program.is_points_enabled
|
|
response.display_name = program.display_name
|
|
|
|
return response
|
|
|
|
|
|
@router.put("/program", response_model=ProgramResponse)
|
|
def update_program(
|
|
data: ProgramUpdate,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Update the merchant's loyalty program (merchant_owner only)."""
|
|
if current_user.role != "merchant_owner":
|
|
raise AuthorizationException("Only merchant owners can update programs")
|
|
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
program = program_service.update_program(db, program.id, data)
|
|
|
|
response = ProgramResponse.model_validate(program)
|
|
response.is_stamps_enabled = program.is_stamps_enabled
|
|
response.is_points_enabled = program.is_points_enabled
|
|
response.display_name = program.display_name
|
|
|
|
return response
|
|
|
|
|
|
@router.delete("/program", status_code=204)
|
|
def delete_program(
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Delete the merchant's loyalty program (merchant_owner only)."""
|
|
if current_user.role != "merchant_owner":
|
|
raise AuthorizationException("Only merchant owners can delete programs")
|
|
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
program_service.delete_program(db, program.id)
|
|
logger.info(f"Store user deleted loyalty program {program.id}")
|
|
|
|
|
|
@router.get("/stats", response_model=ProgramStatsResponse)
|
|
def get_stats(
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get loyalty program statistics."""
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
|
|
stats = program_service.get_program_stats(db, program.id)
|
|
return ProgramStatsResponse(**stats)
|
|
|
|
|
|
@router.get("/stats/merchant", response_model=MerchantStatsResponse)
|
|
def get_merchant_stats(
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get merchant-wide loyalty statistics across all locations."""
|
|
store_id = current_user.token_store_id
|
|
merchant_id = get_store_merchant_id(db, store_id)
|
|
|
|
stats = program_service.get_merchant_stats(db, merchant_id)
|
|
|
|
return MerchantStatsResponse(**stats)
|
|
|
|
|
|
@router.get("/analytics/cohorts")
|
|
def get_cohort_retention(
|
|
months_back: int = Query(6, ge=1, le=24),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Cohort retention matrix for this merchant's loyalty program."""
|
|
from app.modules.loyalty.services.analytics_service import analytics_service
|
|
|
|
merchant_id = get_store_merchant_id(db, current_user.token_store_id)
|
|
return analytics_service.get_cohort_retention(
|
|
db, merchant_id, months_back=months_back
|
|
)
|
|
|
|
|
|
@router.get("/analytics/churn")
|
|
def get_at_risk_cards(
|
|
limit: int = Query(50, ge=1, le=200),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Cards at risk of churn for this merchant."""
|
|
from app.modules.loyalty.services.analytics_service import analytics_service
|
|
|
|
merchant_id = get_store_merchant_id(db, current_user.token_store_id)
|
|
return analytics_service.get_at_risk_cards(
|
|
db, merchant_id, limit=limit
|
|
)
|
|
|
|
|
|
@router.get("/analytics/revenue")
|
|
def get_revenue_attribution(
|
|
months_back: int = Query(6, ge=1, le=24),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Revenue attribution from loyalty transactions."""
|
|
from app.modules.loyalty.services.analytics_service import analytics_service
|
|
|
|
merchant_id = get_store_merchant_id(db, current_user.token_store_id)
|
|
return analytics_service.get_revenue_attribution(
|
|
db, merchant_id, months_back=months_back
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Staff PINs
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/pins", response_model=PinListResponse)
|
|
def list_pins(
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""List staff PINs for this store location."""
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
|
|
# List PINs for this store only
|
|
pins = pin_service.list_pins(db, program.id, store_id=store_id)
|
|
|
|
return PinListResponse(
|
|
pins=[PinResponse.model_validate(pin) for pin in pins],
|
|
total=len(pins),
|
|
)
|
|
|
|
|
|
@router.post("/pins", response_model=PinResponse, status_code=201)
|
|
def create_pin(
|
|
data: PinCreate,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Create a new staff PIN for this store location."""
|
|
store_id = current_user.token_store_id
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
|
|
pin = pin_service.create_pin(db, program.id, store_id, data)
|
|
return PinResponse.model_validate(pin)
|
|
|
|
|
|
@router.patch("/pins/{pin_id}", response_model=PinResponse)
|
|
def update_pin(
|
|
pin_id: int = Path(..., gt=0),
|
|
data: PinUpdate = None,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Update a staff PIN."""
|
|
pin = pin_service.require_pin(db, pin_id)
|
|
if pin.store_id != current_user.token_store_id:
|
|
from app.modules.loyalty.exceptions import StaffPinNotFoundException
|
|
raise StaffPinNotFoundException(str(pin_id))
|
|
pin = pin_service.update_pin(db, pin_id, data)
|
|
return PinResponse.model_validate(pin)
|
|
|
|
|
|
@router.delete("/pins/{pin_id}", status_code=204)
|
|
def delete_pin(
|
|
pin_id: int = Path(..., gt=0),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Delete a staff PIN."""
|
|
pin = pin_service.require_pin(db, pin_id)
|
|
if pin.store_id != current_user.token_store_id:
|
|
from app.modules.loyalty.exceptions import StaffPinNotFoundException
|
|
raise StaffPinNotFoundException(str(pin_id))
|
|
pin_service.delete_pin(db, pin_id)
|
|
|
|
|
|
@router.post("/pins/{pin_id}/unlock", response_model=PinResponse)
|
|
@rate_limit(max_requests=10, window_seconds=60)
|
|
def unlock_pin(
|
|
request: Request,
|
|
pin_id: int = Path(..., gt=0),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Unlock a locked staff PIN."""
|
|
pin = pin_service.require_pin(db, pin_id)
|
|
if pin.store_id != current_user.token_store_id:
|
|
from app.modules.loyalty.exceptions import StaffPinNotFoundException
|
|
raise StaffPinNotFoundException(str(pin_id))
|
|
pin = pin_service.unlock_pin(db, pin_id)
|
|
return PinResponse.model_validate(pin)
|
|
|
|
|
|
# =============================================================================
|
|
# Card Management
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/cards", response_model=CardListResponse)
|
|
def list_cards(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
is_active: bool | None = Query(None),
|
|
search: str | None = Query(None, max_length=100),
|
|
enrolled_here: bool = Query(False, description="Only show cards enrolled at this location"),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
List loyalty cards for the merchant.
|
|
|
|
By default lists all cards in the merchant's loyalty program.
|
|
Use enrolled_here=true to filter to cards enrolled at this location.
|
|
"""
|
|
store_id = current_user.token_store_id
|
|
merchant_id = get_store_merchant_id(db, store_id)
|
|
|
|
program = program_service.require_program_by_store(db, store_id)
|
|
|
|
# Filter by enrolled_at_store_id if requested
|
|
filter_store_id = store_id if enrolled_here else None
|
|
|
|
cards, total = card_service.list_cards(
|
|
db,
|
|
merchant_id,
|
|
store_id=filter_store_id,
|
|
skip=skip,
|
|
limit=limit,
|
|
is_active=is_active,
|
|
search=search,
|
|
)
|
|
|
|
card_responses = []
|
|
for card in cards:
|
|
customer = card.customer
|
|
response = CardResponse(
|
|
id=card.id,
|
|
card_number=card.card_number,
|
|
customer_id=card.customer_id,
|
|
merchant_id=card.merchant_id,
|
|
program_id=card.program_id,
|
|
enrolled_at_store_id=card.enrolled_at_store_id,
|
|
customer_name=customer.full_name if customer else None,
|
|
customer_email=customer.email if customer else None,
|
|
stamp_count=card.stamp_count,
|
|
stamps_target=program.stamps_target,
|
|
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
|
|
total_stamps_earned=card.total_stamps_earned,
|
|
stamps_redeemed=card.stamps_redeemed,
|
|
points_balance=card.points_balance,
|
|
total_points_earned=card.total_points_earned,
|
|
points_redeemed=card.points_redeemed,
|
|
is_active=card.is_active,
|
|
created_at=card.created_at,
|
|
has_google_wallet=bool(card.google_object_id),
|
|
has_apple_wallet=bool(card.apple_serial_number),
|
|
)
|
|
card_responses.append(response)
|
|
|
|
return CardListResponse(cards=card_responses, total=total)
|
|
|
|
|
|
def _build_card_lookup_response(card, db=None) -> CardLookupResponse:
|
|
"""Build a CardLookupResponse from a card object."""
|
|
from datetime import timedelta
|
|
|
|
program = card.program
|
|
|
|
can_stamp, _ = card.can_stamp(program.cooldown_minutes)
|
|
cooldown_ends = None
|
|
if not can_stamp and card.last_stamp_at:
|
|
cooldown_ends = card.last_stamp_at + timedelta(minutes=program.cooldown_minutes)
|
|
|
|
stamps_today = card_service.get_stamps_today(db, card.id) if db else 0
|
|
|
|
available_rewards = []
|
|
for reward in program.points_rewards or []:
|
|
if reward.get("is_active", True) and card.points_balance >= reward.get(
|
|
"points_required", 0
|
|
):
|
|
available_rewards.append(reward)
|
|
|
|
return CardLookupResponse(
|
|
id=card.id,
|
|
card_number=card.card_number,
|
|
customer_id=card.customer_id,
|
|
customer_name=card.customer.full_name if card.customer else None,
|
|
customer_email=card.customer.email if card.customer else "",
|
|
merchant_id=card.merchant_id,
|
|
merchant_name=card.merchant.name if card.merchant else None,
|
|
stamp_count=card.stamp_count,
|
|
stamps_target=program.stamps_target,
|
|
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
|
|
points_balance=card.points_balance,
|
|
can_redeem_stamps=card.stamp_count >= program.stamps_target,
|
|
stamp_reward_description=program.stamps_reward_description,
|
|
available_rewards=available_rewards,
|
|
can_stamp=can_stamp,
|
|
cooldown_ends_at=cooldown_ends,
|
|
stamps_today=stamps_today,
|
|
max_daily_stamps=program.max_daily_stamps,
|
|
can_earn_more_stamps=stamps_today < program.max_daily_stamps,
|
|
)
|
|
|
|
|
|
@router.get("/cards/lookup", response_model=CardLookupResponse)
|
|
def search_card(
|
|
request: Request,
|
|
q: str = Query(..., description="Search by email, card number, or name"),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Search for a card by email, card number, or customer name.
|
|
|
|
Tries matching in order: card number, then customer email.
|
|
Card must belong to the same merchant as the store.
|
|
"""
|
|
store_id = current_user.token_store_id
|
|
|
|
card = card_service.search_card_for_store(db, store_id, q)
|
|
if not card:
|
|
from app.modules.loyalty.exceptions import LoyaltyCardNotFoundException
|
|
|
|
raise LoyaltyCardNotFoundException(q)
|
|
|
|
return _build_card_lookup_response(card, db)
|
|
|
|
|
|
@router.post("/cards/lookup", response_model=CardLookupResponse)
|
|
def lookup_card(
|
|
request: Request,
|
|
card_id: int | None = Query(None),
|
|
qr_code: str | None = Query(None),
|
|
card_number: str | None = Query(None),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Look up a card by ID, QR code, or card number.
|
|
|
|
Card must belong to the same merchant as the store.
|
|
"""
|
|
store_id = current_user.token_store_id
|
|
|
|
# Uses lookup_card_for_store which validates merchant membership
|
|
# Raises LoyaltyCardNotFoundException if not found
|
|
card = card_service.lookup_card_for_store(
|
|
db,
|
|
store_id,
|
|
card_id=card_id,
|
|
qr_code=qr_code,
|
|
card_number=card_number,
|
|
)
|
|
|
|
return _build_card_lookup_response(card, db)
|
|
|
|
|
|
@router.get("/cards/{card_id}", response_model=CardDetailResponse)
|
|
def get_card_detail(
|
|
card_id: int = Path(..., gt=0),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get detailed loyalty card info by ID."""
|
|
store_id = current_user.token_store_id
|
|
merchant_id = get_store_merchant_id(db, store_id)
|
|
|
|
card = card_service.get_card(db, card_id)
|
|
if not card or card.merchant_id != merchant_id:
|
|
from app.modules.loyalty.exceptions import LoyaltyCardNotFoundException
|
|
|
|
raise LoyaltyCardNotFoundException(str(card_id))
|
|
|
|
program = card.program
|
|
customer = card.customer
|
|
|
|
return CardDetailResponse(
|
|
id=card.id,
|
|
card_number=card.card_number,
|
|
customer_id=card.customer_id,
|
|
merchant_id=card.merchant_id,
|
|
program_id=card.program_id,
|
|
enrolled_at_store_id=card.enrolled_at_store_id,
|
|
customer_name=customer.full_name if customer else None,
|
|
customer_email=customer.email if customer else None,
|
|
merchant_name=card.merchant.name if card.merchant else None,
|
|
qr_code_data=card.qr_code_data or card.card_number,
|
|
program_name=program.display_name,
|
|
program_type=program.loyalty_type,
|
|
reward_description=program.stamps_reward_description,
|
|
stamp_count=card.stamp_count,
|
|
stamps_target=program.stamps_target,
|
|
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
|
|
total_stamps_earned=card.total_stamps_earned,
|
|
stamps_redeemed=card.stamps_redeemed,
|
|
points_balance=card.points_balance,
|
|
total_points_earned=card.total_points_earned,
|
|
points_redeemed=card.points_redeemed,
|
|
is_active=card.is_active,
|
|
created_at=card.created_at,
|
|
last_stamp_at=card.last_stamp_at,
|
|
last_points_at=card.last_points_at,
|
|
last_redemption_at=card.last_redemption_at,
|
|
last_activity_at=card.last_activity_at,
|
|
has_google_wallet=bool(card.google_object_id),
|
|
has_apple_wallet=bool(card.apple_serial_number),
|
|
)
|
|
|
|
|
|
@router.get("/transactions", response_model=TransactionListResponse)
|
|
def list_store_transactions(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(10, ge=1, le=100),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""List recent transactions for this merchant's loyalty program."""
|
|
store_id = current_user.token_store_id
|
|
merchant_id = get_store_merchant_id(db, store_id)
|
|
|
|
transactions, total = card_service.get_store_transactions(
|
|
db, merchant_id, skip=skip, limit=limit
|
|
)
|
|
|
|
tx_responses = []
|
|
for t in transactions:
|
|
tx = TransactionResponse.model_validate(t)
|
|
if t.card and t.card.customer:
|
|
tx.customer_name = t.card.customer.full_name
|
|
tx_responses.append(tx)
|
|
|
|
return TransactionListResponse(transactions=tx_responses, total=total)
|
|
|
|
|
|
@router.post("/cards/enroll", response_model=CardResponse, status_code=201)
|
|
def enroll_customer(
|
|
data: CardEnrollRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Enroll a customer in the merchant's loyalty program.
|
|
|
|
The card will be associated with the merchant and track which
|
|
store enrolled them.
|
|
"""
|
|
store_id = current_user.token_store_id
|
|
|
|
# Resolve merchant_id for cross-store customer lookup
|
|
from app.modules.tenancy.services.store_service import store_service
|
|
|
|
store = store_service.get_store_by_id_optional(db, store_id)
|
|
merchant_id = store.merchant_id if store else None
|
|
|
|
customer_id = card_service.resolve_customer_id(
|
|
db,
|
|
customer_id=data.customer_id,
|
|
email=data.email,
|
|
store_id=store_id,
|
|
merchant_id=merchant_id,
|
|
)
|
|
|
|
card = card_service.enroll_customer_for_store(db, customer_id, store_id)
|
|
|
|
program = card.program
|
|
|
|
return CardResponse(
|
|
id=card.id,
|
|
card_number=card.card_number,
|
|
customer_id=card.customer_id,
|
|
merchant_id=card.merchant_id,
|
|
program_id=card.program_id,
|
|
enrolled_at_store_id=card.enrolled_at_store_id,
|
|
stamp_count=card.stamp_count,
|
|
stamps_target=program.stamps_target,
|
|
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
|
|
total_stamps_earned=card.total_stamps_earned,
|
|
stamps_redeemed=card.stamps_redeemed,
|
|
points_balance=card.points_balance,
|
|
total_points_earned=card.total_points_earned,
|
|
points_redeemed=card.points_redeemed,
|
|
is_active=card.is_active,
|
|
created_at=card.created_at,
|
|
)
|
|
|
|
|
|
@router.get("/cards/{card_id}/transactions", response_model=TransactionListResponse)
|
|
def get_card_transactions(
|
|
card_id: int = Path(..., gt=0),
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get transaction history for a card."""
|
|
store_id = current_user.token_store_id
|
|
|
|
# Verify card belongs to this merchant (raises LoyaltyCardNotFoundException if not found)
|
|
card_service.lookup_card_for_store(db, store_id, card_id=card_id)
|
|
|
|
transactions, total = card_service.get_card_transactions(
|
|
db, card_id, skip=skip, limit=limit
|
|
)
|
|
|
|
return TransactionListResponse(
|
|
transactions=[TransactionResponse.model_validate(t) for t in transactions],
|
|
total=total,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Stamp Operations
|
|
# =============================================================================
|
|
|
|
|
|
@router.post("/stamp", response_model=StampResponse)
|
|
@rate_limit(max_requests=60, window_seconds=60)
|
|
def add_stamp(
|
|
request: Request,
|
|
data: StampRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Add a stamp to a loyalty card."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = stamp_service.add_stamp(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return StampResponse(**result)
|
|
|
|
|
|
@router.post("/stamp/redeem", response_model=StampRedeemResponse)
|
|
@rate_limit(max_requests=30, window_seconds=60)
|
|
def redeem_stamps(
|
|
request: Request,
|
|
data: StampRedeemRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Redeem stamps for a reward."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = stamp_service.redeem_stamps(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return StampRedeemResponse(**result)
|
|
|
|
|
|
@router.post("/stamp/void", response_model=StampVoidResponse)
|
|
@rate_limit(max_requests=20, window_seconds=60)
|
|
def void_stamps(
|
|
request: Request,
|
|
data: StampVoidRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Void stamps for a return."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = stamp_service.void_stamps(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
stamps_to_void=data.stamps_to_void,
|
|
original_transaction_id=data.original_transaction_id,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return StampVoidResponse(**result)
|
|
|
|
|
|
# =============================================================================
|
|
# Points Operations
|
|
# =============================================================================
|
|
|
|
|
|
@router.post("/points/earn", response_model=PointsEarnResponse)
|
|
@rate_limit(max_requests=60, window_seconds=60)
|
|
def earn_points(
|
|
request: Request,
|
|
data: PointsEarnRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Earn points from a purchase."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = points_service.earn_points(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
purchase_amount_cents=data.purchase_amount_cents,
|
|
order_reference=data.order_reference,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return PointsEarnResponse(**result)
|
|
|
|
|
|
@router.post("/points/redeem", response_model=PointsRedeemResponse)
|
|
@rate_limit(max_requests=30, window_seconds=60)
|
|
def redeem_points(
|
|
request: Request,
|
|
data: PointsRedeemRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Redeem points for a reward."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = points_service.redeem_points(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
reward_id=data.reward_id,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return PointsRedeemResponse(**result)
|
|
|
|
|
|
@router.post("/points/void", response_model=PointsVoidResponse)
|
|
@rate_limit(max_requests=20, window_seconds=60)
|
|
def void_points(
|
|
request: Request,
|
|
data: PointsVoidRequest,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Void points for a return."""
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = points_service.void_points(
|
|
db,
|
|
store_id=store_id,
|
|
card_id=data.card_id,
|
|
qr_code=data.qr_code,
|
|
card_number=data.card_number,
|
|
points_to_void=data.points_to_void,
|
|
original_transaction_id=data.original_transaction_id,
|
|
order_reference=data.order_reference,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
notes=data.notes,
|
|
)
|
|
|
|
return PointsVoidResponse(**result)
|
|
|
|
|
|
@router.post("/cards/{card_id}/points/adjust", response_model=PointsAdjustResponse)
|
|
@rate_limit(max_requests=20, window_seconds=60)
|
|
def adjust_points(
|
|
request: Request,
|
|
data: PointsAdjustRequest,
|
|
card_id: int = Path(..., gt=0),
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Manually adjust points (merchant_owner only)."""
|
|
if current_user.role != "merchant_owner":
|
|
raise AuthorizationException("Only merchant owners can adjust points")
|
|
|
|
store_id = current_user.token_store_id
|
|
ip, user_agent = get_client_info(request)
|
|
|
|
result = points_service.adjust_points(
|
|
db,
|
|
card_id=card_id,
|
|
points_delta=data.points_delta,
|
|
store_id=store_id,
|
|
reason=data.reason,
|
|
staff_pin=data.staff_pin,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
)
|
|
|
|
return PointsAdjustResponse(**result)
|
|
|
|
|
|
# =============================================================================
|
|
# Bulk Operations (Merchant Owner only)
|
|
# =============================================================================
|
|
|
|
|
|
@router.post("/cards/bulk/deactivate")
|
|
@rate_limit(max_requests=10, window_seconds=60)
|
|
def bulk_deactivate_cards(
|
|
request: Request,
|
|
data: dict,
|
|
current_user: User = Depends(get_current_store_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Bulk deactivate multiple loyalty cards (merchant_owner only)."""
|
|
if current_user.role != "merchant_owner":
|
|
raise AuthorizationException("Only merchant owners can bulk deactivate cards")
|
|
|
|
from app.modules.tenancy.services.store_service import store_service
|
|
|
|
store = store_service.get_store_by_id_optional(db, current_user.token_store_id)
|
|
if not store:
|
|
raise AuthorizationException("Store not found")
|
|
|
|
card_ids = data.get("card_ids", [])
|
|
reason = data.get("reason", "Merchant bulk deactivation")
|
|
|
|
count = card_service.bulk_deactivate_cards(
|
|
db,
|
|
card_ids=card_ids,
|
|
merchant_id=store.merchant_id,
|
|
reason=reason,
|
|
)
|
|
|
|
return {"cards_deactivated": count, "message": f"Deactivated {count} card(s)"}
|