# app/modules/loyalty/tasks/point_expiration.py """ Point expiration task. Handles expiring points that are older than the configured expiration period based on card inactivity. Runs daily at 02:00 via the scheduled task configuration in definition.py. """ import logging from datetime import UTC, datetime, timedelta from celery import shared_task from sqlalchemy.orm import Session from app.core.database import SessionLocal from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram, LoyaltyTransaction from app.modules.loyalty.models.loyalty_transaction import TransactionType logger = logging.getLogger(__name__) @shared_task(name="loyalty.expire_points") def expire_points() -> dict: """ Expire points that are past their expiration date based on card inactivity. For each program with points_expiration_days configured: 1. Find cards that haven't had activity in the expiration period 2. Expire all points on those cards 3. Create POINTS_EXPIRED transaction records 4. Update card balances Returns: Summary of expired points """ logger.info("Starting point expiration task...") db: Session = SessionLocal() try: result = _process_point_expiration(db) db.commit() logger.info( f"Point expiration complete: {result['cards_processed']} cards, " f"{result['points_expired']} points expired" ) return result except Exception as e: db.rollback() logger.error(f"Point expiration task failed: {e}", exc_info=True) return { "status": "error", "error": str(e), "cards_processed": 0, "points_expired": 0, } finally: db.close() def _process_point_expiration(db: Session) -> dict: """ Process point expiration for all programs. Args: db: Database session Returns: Summary of expired points """ total_cards_processed = 0 total_points_expired = 0 programs_processed = 0 # Find all active programs with point expiration configured programs = ( db.query(LoyaltyProgram) .filter( LoyaltyProgram.is_active == True, LoyaltyProgram.points_expiration_days.isnot(None), LoyaltyProgram.points_expiration_days > 0, ) .all() ) logger.info(f"Found {len(programs)} programs with point expiration configured") for program in programs: cards_count, points_count = _expire_points_for_program(db, program) total_cards_processed += cards_count total_points_expired += points_count programs_processed += 1 logger.debug( f"Program {program.id} (merchant {program.merchant_id}): " f"{cards_count} cards, {points_count} points expired" ) return { "status": "success", "programs_processed": programs_processed, "cards_processed": total_cards_processed, "points_expired": total_points_expired, } def _expire_points_for_program(db: Session, program: LoyaltyProgram) -> tuple[int, int]: """ Expire points for a specific loyalty program. Also sends warning emails to cards approaching their expiration date (14 days before) and expired notifications after points are zeroed. Args: db: Database session program: Loyalty program to process Returns: Tuple of (cards_processed, points_expired) """ if not program.points_expiration_days: return 0, 0 now = datetime.now(UTC) # Calculate expiration threshold expiration_threshold = now - timedelta(days=program.points_expiration_days) logger.debug( f"Processing program {program.id}: expiration after {program.points_expiration_days} days " f"(threshold: {expiration_threshold})" ) # --- Phase 1: Send 14-day warning emails --- warning_days = 14 warning_threshold = now - timedelta( days=program.points_expiration_days - warning_days ) _send_expiration_warnings( db, program, warning_threshold, warning_days, now ) # --- Phase 2: Expire points --- # Find cards with: # - Points balance > 0 # - Last activity before expiration threshold # - Belonging to this program's merchant cards_to_expire = ( db.query(LoyaltyCard) .filter( LoyaltyCard.merchant_id == program.merchant_id, LoyaltyCard.points_balance > 0, LoyaltyCard.last_activity_at < expiration_threshold, LoyaltyCard.is_active == True, ) .all() ) if not cards_to_expire: logger.debug(f"No cards to expire for program {program.id}") return 0, 0 logger.info(f"Found {len(cards_to_expire)} cards to expire for program {program.id}") cards_processed = 0 points_expired = 0 for card in cards_to_expire: if card.points_balance <= 0: continue expired_points = card.points_balance # Create expiration transaction transaction = LoyaltyTransaction( card_id=card.id, merchant_id=program.merchant_id, store_id=None, # System action, no store transaction_type=TransactionType.POINTS_EXPIRED.value, points_delta=-expired_points, points_balance_after=0, stamps_delta=0, stamps_balance_after=card.stamp_count, notes=f"Points expired after {program.points_expiration_days} days of inactivity", transaction_at=now, ) db.add(transaction) # noqa: PERF006 # Update card balance and voided tracking card.expire_points(expired_points) # Note: We don't update last_activity_at for expiration # Send expired notification try: from app.modules.loyalty.services.notification_service import ( notification_service, ) notification_service.send_points_expired(db, card, expired_points) except Exception: logger.warning( f"Failed to queue expiration notification for card {card.id}", exc_info=True, ) cards_processed += 1 points_expired += expired_points logger.debug( f"Expired {expired_points} points from card {card.id} " f"(last activity: {card.last_activity_at})" ) return cards_processed, points_expired def _send_expiration_warnings( db: Session, program: LoyaltyProgram, warning_threshold: datetime, warning_days: int, now: datetime, ) -> int: """Send expiration warning emails to cards approaching expiry. Only sends one warning per expiration cycle (tracked via last_expiration_warning_at on the card). Returns: Number of warnings sent """ from sqlalchemy import or_ # Find cards in the warning window: # - Have points # - Last activity is past the warning threshold (i.e. will expire in ~14 days) # - But NOT yet past the full expiration threshold # - Haven't received a warning yet in this cycle expiration_threshold = now - timedelta(days=program.points_expiration_days) cards = ( db.query(LoyaltyCard) .filter( LoyaltyCard.merchant_id == program.merchant_id, LoyaltyCard.points_balance > 0, LoyaltyCard.is_active == True, LoyaltyCard.last_activity_at < warning_threshold, LoyaltyCard.last_activity_at >= expiration_threshold, or_( LoyaltyCard.last_expiration_warning_at.is_(None), LoyaltyCard.last_expiration_warning_at < warning_threshold, ), ) .all() ) if not cards: return 0 warnings_sent = 0 expiration_date = ( now + timedelta(days=warning_days) ).strftime("%Y-%m-%d") for card in cards: try: from app.modules.loyalty.services.notification_service import ( notification_service, ) notification_service.send_points_expiration_warning( db, card, expiring_points=card.points_balance, days_remaining=warning_days, expiration_date=expiration_date, ) card.last_expiration_warning_at = now warnings_sent += 1 except Exception: logger.warning( f"Failed to queue expiration warning for card {card.id}", exc_info=True, ) logger.info( f"Sent {warnings_sent} expiration warnings for program {program.id}" ) return warnings_sent # Allow running directly for testing if __name__ == "__main__": import sys logging.basicConfig(level=logging.DEBUG) result = expire_points() print(f"Result: {result}") sys.exit(0 if result["status"] == "success" else 1)