167 lines
6.5 KiB
Python
167 lines
6.5 KiB
Python
import requests
|
|
import json
|
|
import time
|
|
import hashlib
|
|
import hmac
|
|
import base64
|
|
from django.conf import settings
|
|
from typing import Dict, Any, Optional
|
|
|
|
class NaverCloudSMS:
|
|
"""네이버 클라우드 플랫폼 SMS 서비스 클래스"""
|
|
|
|
def __init__(self):
|
|
self.access_key = getattr(settings, 'NAVER_CLOUD_ACCESS_KEY', '')
|
|
self.secret_key = getattr(settings, 'NAVER_CLOUD_SECRET_KEY', '')
|
|
self.service_id = getattr(settings, 'NAVER_CLOUD_SMS_SERVICE_ID', '')
|
|
self.sender_phone = getattr(settings, 'NAVER_CLOUD_SMS_SENDER_PHONE', '')
|
|
|
|
# API 엔드포인트
|
|
self.base_url = "https://sens.apigw.ntruss.com"
|
|
self.sms_url = f"{self.base_url}/sms/v2/services/{self.service_id}/messages"
|
|
|
|
def _make_signature(self, timestamp: str) -> str:
|
|
"""네이버 클라우드 API 서명 생성"""
|
|
space = " "
|
|
new_line = "\n"
|
|
method = "POST"
|
|
url = f"/sms/v2/services/{self.service_id}/messages"
|
|
|
|
message = method + space + url + new_line + timestamp + new_line + self.access_key
|
|
message = message.encode('utf-8')
|
|
|
|
signing_key = base64.b64encode(
|
|
hmac.new(
|
|
self.secret_key.encode('utf-8'),
|
|
message,
|
|
digestmod=hashlib.sha256
|
|
).digest()
|
|
).decode('utf-8')
|
|
|
|
return signing_key
|
|
|
|
def send_sms(self, phone_number: str, message: str) -> Dict[str, Any]:
|
|
"""SMS 발송"""
|
|
try:
|
|
timestamp = str(int(time.time() * 1000))
|
|
signature = self._make_signature(timestamp)
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json; charset=utf-8',
|
|
'x-ncp-apigw-timestamp': timestamp,
|
|
'x-ncp-iam-access-key': self.access_key,
|
|
'x-ncp-apigw-signature-v2': signature
|
|
}
|
|
|
|
data = {
|
|
'type': 'SMS',
|
|
'contentType': 'COMM',
|
|
'countryCode': '82',
|
|
'from': self.sender_phone,
|
|
'content': message,
|
|
'messages': [
|
|
{
|
|
'to': phone_number
|
|
}
|
|
]
|
|
}
|
|
|
|
response = requests.post(
|
|
self.sms_url,
|
|
headers=headers,
|
|
data=json.dumps(data)
|
|
)
|
|
|
|
if response.status_code == 202:
|
|
result = response.json()
|
|
return {
|
|
'success': True,
|
|
'request_id': result.get('requestId'),
|
|
'status_code': result.get('statusCode'),
|
|
'status_name': result.get('statusName')
|
|
}
|
|
else:
|
|
return {
|
|
'success': False,
|
|
'error': f'HTTP {response.status_code}: {response.text}'
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'error': str(e)
|
|
}
|
|
|
|
def send_verification_code(self, phone_number: str, verification_code: str) -> Dict[str, Any]:
|
|
"""인증번호 SMS 발송"""
|
|
message = f"[신라AMP] 인증번호는 [{verification_code}] 입니다. 3분 이내에 입력해주세요."
|
|
|
|
# 발신번호가 설정되지 않은 경우 에러 반환
|
|
if not self.sender_phone:
|
|
print(f"[ERROR] 발신번호가 설정되지 않았습니다.")
|
|
return {
|
|
'success': False,
|
|
'error': '발신번호가 설정되지 않았습니다. .env 파일을 확인해주세요.'
|
|
}
|
|
|
|
print(f"[INFO] 실제 SMS 발송 시도: {phone_number} - {verification_code}")
|
|
return self.send_sms(phone_number, message)
|
|
|
|
def send_withdrawal_approval_sms(self, phone_number: str, name: str) -> Dict[str, Any]:
|
|
"""탈퇴 승인 SMS 발송"""
|
|
message = f"[신라AMP] {name}님의 회원탈퇴 요청이 처리되었습니다."
|
|
|
|
if not self.sender_phone:
|
|
print(f"[ERROR] 발신번호가 설정되지 않았습니다.")
|
|
return {
|
|
'success': False,
|
|
'error': '발신번호가 설정되지 않았습니다.'
|
|
}
|
|
|
|
print(f"[INFO] 탈퇴 승인 SMS 발송: {phone_number} - {name}")
|
|
return self.send_sms(phone_number, message)
|
|
|
|
def send_withdrawal_rejection_sms(self, phone_number: str, name: str, reason: str = None) -> Dict[str, Any]:
|
|
"""탈퇴 거부 SMS 발송"""
|
|
if reason:
|
|
# message = f"[신라AMP] {name}님의 회원탈퇴 요청이 거부되었습니다. 사유: {reason}"
|
|
message = f"[신라AMP] {name}님의 회원탈퇴 요청이 거부되었습니다. 자세한 내용은 관리자에게 문의해주세요."
|
|
else:
|
|
message = f"[신라AMP] {name}님의 회원탈퇴 요청이 거부되었습니다. 자세한 내용은 관리자에게 문의해주세요."
|
|
|
|
if not self.sender_phone:
|
|
print(f"[ERROR] 발신번호가 설정되지 않았습니다.")
|
|
return {
|
|
'success': False,
|
|
'error': '발신번호가 설정되지 않았습니다.'
|
|
}
|
|
|
|
print(f"[INFO] 탈퇴 거부 SMS 발송: {phone_number} - {name}")
|
|
return self.send_sms(phone_number, message)
|
|
|
|
# 전역 인스턴스
|
|
sms_service = NaverCloudSMS()
|
|
|
|
def send_verification_sms(phone_number: str, verification_code: str) -> Dict[str, Any]:
|
|
"""인증번호 SMS 발송 함수 (편의 함수)"""
|
|
# 실제 SMS 발송 시도
|
|
print(f"[DEBUG] SMS 발송 시도: {phone_number} - {verification_code}")
|
|
print(f"[DEBUG] Access Key: {sms_service.access_key}")
|
|
print(f"[DEBUG] Service ID: {sms_service.service_id}")
|
|
print(f"[DEBUG] Sender Phone: {sms_service.sender_phone}")
|
|
|
|
result = sms_service.send_verification_code(phone_number, verification_code)
|
|
print(f"[DEBUG] SMS 발송 결과: {result}")
|
|
|
|
return result
|
|
|
|
|
|
def send_withdrawal_approval_sms(phone_number: str, name: str) -> Dict[str, Any]:
|
|
"""탈퇴 승인 SMS 발송 함수 (편의 함수)"""
|
|
return sms_service.send_withdrawal_approval_sms(phone_number, name)
|
|
|
|
|
|
def send_withdrawal_rejection_sms(phone_number: str, name: str, reason: str = None) -> Dict[str, Any]:
|
|
"""탈퇴 거부 SMS 발송 함수 (편의 함수)"""
|
|
return sms_service.send_withdrawal_rejection_sms(phone_number, name, reason)
|