from celery import Celery from src.models.user import db from src.models.subscription import Subscription, Payment, SubscriptionStatus, PaymentStatus, SubscriptionRenewalLog from datetime import datetime, timedelta import os import requests import base64 # Celery 설정 celery_app = Celery('mlink_subscription', broker=os.getenv('REDIS_URL', 'redis://localhost:6379/0'), backend=os.getenv('REDIS_URL', 'redis://localhost:6379/0')) celery_app.conf.update( task_serializer='json', accept_content=['json'], result_serializer='json', timezone='Asia/Seoul', enable_utc=True, task_track_started=True, task_time_limit=30 * 60, task_soft_time_limit=60, worker_prefetch_multiplier=1, worker_max_tasks_per_child=1000, ) # 헥토파이낸셜 설정 HECTOFINANCIAL_CONFIG = { 'client_key': os.getenv('HECTO_CLIENT_KEY'), 'secret_key': os.getenv('HECTO_SECRET_KEY'), 'base_url': os.getenv('HECTO_BASE_URL', 'https://api.hectofinancial.com') } @celery_app.task def process_subscription_renewals(): """구독 갱신 처리 작업""" try: # 갱신이 필요한 구독 조회 renewals_needed = Subscription.query.filter( Subscription.status == SubscriptionStatus.ACTIVE, Subscription.auto_renew == True, Subscription.next_billing_date <= datetime.now() ).all() for subscription in renewals_needed: process_single_renewal.delay(subscription.id) return f"Processed {len(renewals_needed)} subscription renewals" except Exception as e: return f"Error processing renewals: {str(e)}" @celery_app.task def process_single_renewal(subscription_id): """단일 구독 갱신 처리""" try: subscription = Subscription.query.get(subscription_id) if not subscription: return f"Subscription {subscription_id} not found" # 갱신 로그 생성 renewal_log = SubscriptionRenewalLog( subscription_id=subscription_id, renewal_date=datetime.now(), status='pending', amount=subscription.product.price ) db.session.add(renewal_log) db.session.commit() # 자동 결제 처리 payment_result = process_automatic_payment(subscription) if payment_result['success']: # 구독 기간 연장 if subscription.product.billing_cycle == 'monthly': new_end_date = subscription.end_date + timedelta(days=30) else: # yearly new_end_date = subscription.end_date + timedelta(days=365) subscription.end_date = new_end_date subscription.next_billing_date = new_end_date # 결제 정보 저장 payment = Payment( user_id=subscription.user_id, subscription_id=subscription.id, amount=subscription.product.price, currency=subscription.product.currency, payment_method=PaymentMethod.CARD, status=PaymentStatus.COMPLETED, pg_transaction_id=payment_result.get('transaction_id'), billing_key=payment_result.get('billing_key') ) db.session.add(payment) # 갱신 로그 업데이트 renewal_log.status = 'success' renewal_log.payment_id = payment.id else: # 결제 실패 처리 subscription.status = SubscriptionStatus.EXPIRED subscription.auto_renew = False renewal_log.status = 'failed' renewal_log.error_message = payment_result.get('error') db.session.commit() return f"Renewal processed for subscription {subscription_id}: {renewal_log.status}" except Exception as e: db.session.rollback() return f"Error processing renewal for subscription {subscription_id}: {str(e)}" def process_automatic_payment(subscription): """헥토파이낸셜 자동 결제 처리""" try: # 마지막 결제에서 빌링키 가져오기 last_payment = Payment.query.filter_by( subscription_id=subscription.id, status=PaymentStatus.COMPLETED ).order_by(Payment.created_at.desc()).first() if not last_payment or not last_payment.billing_key: return { 'success': False, 'error': 'Billing key not found' } # 헥토파이낸셜 자동결제 API 호출 payment_data = { 'billing_key': last_payment.billing_key, 'amount': int(subscription.product.price), 'currency': subscription.product.currency, 'order_id': f'renewal_{subscription.id}_{datetime.now().strftime("%Y%m%d%H%M%S")}', 'order_name': f'{subscription.product.name} 자동 갱신', 'customer_name': subscription.user.username, 'customer_email': subscription.user.email } # API 키 헤더 생성 headers = { 'Authorization': f'Bearer {HECTOFINANCIAL_CONFIG["secret_key"]}', 'Content-Type': 'application/json' } # 헥토파이낸셜 자동결제 API 호출 response = requests.post( f"{HECTOFINANCIAL_CONFIG['base_url']}/v1/billing/{last_payment.billing_key}", headers=headers, json=payment_data ) if response.status_code == 200: result = response.json() return { 'success': True, 'transaction_id': result.get('payment_id'), 'billing_key': last_payment.billing_key } else: error_data = response.json() return { 'success': False, 'error': error_data.get('message', '자동결제에 실패했습니다.') } except Exception as e: return { 'success': False, 'error': str(e) } @celery_app.task def cleanup_expired_subscriptions(): """만료된 구독 정리""" try: expired_subscriptions = Subscription.query.filter( Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date < datetime.now() ).all() for subscription in expired_subscriptions: subscription.status = SubscriptionStatus.EXPIRED subscription.auto_renew = False db.session.commit() return f"Cleaned up {len(expired_subscriptions)} expired subscriptions" except Exception as e: db.session.rollback() return f"Error cleaning up expired subscriptions: {str(e)}" @celery_app.task def send_payment_reminders(): """결제 예정 알림 발송""" try: # 3일 후 결제 예정인 구독 조회 reminder_date = datetime.now() + timedelta(days=3) subscriptions = Subscription.query.filter( Subscription.status == SubscriptionStatus.ACTIVE, Subscription.auto_renew == True, Subscription.next_billing_date <= reminder_date, Subscription.next_billing_date > datetime.now() ).all() for subscription in subscriptions: # 이메일 알림 발송 (실제 구현 시 이메일 서비스 연동) send_payment_reminder_email.delay(subscription.id) return f"Sent {len(subscriptions)} payment reminders" except Exception as e: return f"Error sending payment reminders: {str(e)}" @celery_app.task def send_payment_reminder_email(subscription_id): """결제 예정 이메일 발송""" try: subscription = Subscription.query.get(subscription_id) if not subscription: return f"Subscription {subscription_id} not found" # 이메일 발송 로직 (실제 구현 시 이메일 서비스 연동) # send_email( # to=subscription.user.email, # subject="결제 예정 안내", # template="payment_reminder", # context={ # 'user': subscription.user, # 'subscription': subscription, # 'next_billing_date': subscription.next_billing_date # } # ) return f"Payment reminder sent for subscription {subscription_id}" except Exception as e: return f"Error sending payment reminder for subscription {subscription_id}: {str(e)}" @celery_app.task def process_failed_payments(): """실패한 결제 재시도""" try: # 실패한 결제 조회 (최근 24시간 내) failed_payments = Payment.query.filter( Payment.status == PaymentStatus.FAILED, Payment.created_at >= datetime.now() - timedelta(hours=24) ).all() for payment in failed_payments: retry_payment.delay(payment.id) return f"Retrying {len(failed_payments)} failed payments" except Exception as e: return f"Error processing failed payments: {str(e)}" @celery_app.task def retry_payment(payment_id): """결제 재시도""" try: payment = Payment.query.get(payment_id) if not payment: return f"Payment {payment_id} not found" # 재시도 로직 (토스페이먼츠 API 호출) # 실제 구현에서는 토스페이먼츠 결제 재시도 API 사용 return f"Payment retry processed for payment {payment_id}" except Exception as e: return f"Error retrying payment {payment_id}: {str(e)}" # 스케줄 설정 @celery_app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): # 매일 자정에 구독 갱신 처리 sender.add_periodic_task( 86400.0, # 24시간 process_subscription_renewals.s(), name='daily-subscription-renewals' ) # 매일 자정에 만료된 구독 정리 sender.add_periodic_task( 86400.0, # 24시간 cleanup_expired_subscriptions.s(), name='daily-cleanup-expired' ) # 매일 오전 9시에 결제 예정 알림 sender.add_periodic_task( 86400.0, # 24시간 send_payment_reminders.s(), name='daily-payment-reminders' ) # 매시간 실패한 결제 재시도 sender.add_periodic_task( 3600.0, # 1시간 process_failed_payments.s(), name='hourly-failed-payment-retry' )