SillaAMP_V2/A_core/telegram_utils.py

289 lines
9.9 KiB
Python
Raw Normal View History

"""
텔레그램 유틸리티
회원가입 알림 등을 위한 텔레그램 메시지 전송 기능
"""
import requests
import threading
from django.conf import settings
from datetime import datetime
def get_client_ip(request):
"""
클라이언트의 실제 IP 주소를 추출
프록시, 로드밸런서, CDN 등을 고려하여 실제 IP를 찾음
"""
# 프록시나 로드밸런서를 통한 실제 클라이언트 IP 확인
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
# 첫 번째 IP가 실제 클라이언트 IP (쉼표로 구분된 경우)
ip = x_forwarded_for.split(',')[0].strip()
if ip and ip != '127.0.0.1':
return ip
# Cloudflare 등 CDN에서 사용하는 헤더
cf_connecting_ip = request.META.get('HTTP_CF_CONNECTING_IP')
if cf_connecting_ip and cf_connecting_ip != '127.0.0.1':
return cf_connecting_ip
# Nginx 등 리버스 프록시에서 사용
x_real_ip = request.META.get('HTTP_X_REAL_IP')
if x_real_ip and x_real_ip != '127.0.0.1':
return x_real_ip
# 기본 REMOTE_ADDR
remote_addr = request.META.get('REMOTE_ADDR', '알 수 없음')
# 로컬 환경 표시
if remote_addr == '127.0.0.1':
return f"{remote_addr} (로컬 개발환경)"
elif remote_addr.startswith('192.168.') or remote_addr.startswith('10.') or remote_addr.startswith('172.'):
return f"{remote_addr} (내부 네트워크)"
return remote_addr
def get_device_info(user_agent):
"""User-Agent에서 기기 정보 추출"""
if not user_agent:
return "알 수 없음"
# 모바일 기기 체크
if any(mobile in user_agent for mobile in ['iPhone', 'iPad', 'Android', 'Mobile']):
if 'iPhone' in user_agent:
return "iPhone"
elif 'iPad' in user_agent:
return "iPad"
elif 'Android' in user_agent:
if 'Mobile' in user_agent:
return "Android 폰"
else:
return "Android 태블릿"
else:
return "모바일"
# 데스크톱 OS 체크
if 'Windows NT' in user_agent:
if 'Windows NT 10.0' in user_agent:
return "Windows 10/11"
elif 'Windows NT 6.3' in user_agent:
return "Windows 8.1"
elif 'Windows NT 6.1' in user_agent:
return "Windows 7"
else:
return "Windows"
elif 'Macintosh' in user_agent or 'Mac OS X' in user_agent:
return "Mac"
elif 'Linux' in user_agent and 'Android' not in user_agent:
return "Linux"
return "알 수 없음"
def get_browser_info(user_agent):
"""User-Agent에서 브라우저 정보 추출"""
if not user_agent:
return "알 수 없음"
# 브라우저 체크 (순서 중요 - Chrome이 Safari 문자열도 포함하므로)
if 'Edg/' in user_agent:
return "Microsoft Edge"
elif 'Chrome/' in user_agent and 'Safari/' in user_agent:
return "Chrome"
elif 'Firefox/' in user_agent:
return "Firefox"
elif 'Safari/' in user_agent and 'Chrome' not in user_agent:
return "Safari"
elif 'Opera' in user_agent or 'OPR/' in user_agent:
return "Opera"
return "알 수 없음"
def send_telegram_message(message, chat_id=None):
"""
텔레그램 봇으로 메시지 전송
Args:
message (str): 전송할 메시지
chat_id (str, optional): 채팅 ID. 없으면 기본 설정값 사용
Returns:
dict: {'success': bool, 'error': str}
"""
try:
if not hasattr(settings, 'TELEGRAM_BOT_TOKEN') or not settings.TELEGRAM_BOT_TOKEN:
return {'success': False, 'error': 'TELEGRAM_BOT_TOKEN이 설정되지 않았습니다.'}
if not chat_id:
chat_id = getattr(settings, 'TELEGRAM_CHAT_ID', None)
if not chat_id:
return {'success': False, 'error': 'TELEGRAM_CHAT_ID가 설정되지 않았습니다.'}
url = f"https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
'chat_id': chat_id,
'text': message,
'parse_mode': 'HTML' # HTML 포맷 지원
}
response = requests.post(url, json=payload, timeout=10)
if response.status_code == 200:
result = response.json()
if result.get('ok'):
print(f"[TELEGRAM_SUCCESS] 메시지 전송 성공: {message[:50]}...")
return {'success': True, 'error': None}
else:
error_msg = result.get('description', '알 수 없는 오류')
print(f"[TELEGRAM_ERROR] API 오류: {error_msg}")
return {'success': False, 'error': f'텔레그램 API 오류: {error_msg}'}
else:
print(f"[TELEGRAM_ERROR] HTTP 오류: {response.status_code}")
return {'success': False, 'error': f'HTTP 오류: {response.status_code}'}
except requests.exceptions.Timeout:
print("[TELEGRAM_ERROR] 요청 시간 초과")
return {'success': False, 'error': '요청 시간 초과'}
except requests.exceptions.RequestException as e:
print(f"[TELEGRAM_ERROR] 네트워크 오류: {str(e)}")
return {'success': False, 'error': f'네트워크 오류: {str(e)}'}
except Exception as e:
print(f"[TELEGRAM_ERROR] 예상치 못한 오류: {str(e)}")
return {'success': False, 'error': f'예상치 못한 오류: {str(e)}'}
def send_telegram_message_async(message, chat_id=None):
"""
비동기로 텔레그램 메시지 전송 (백그라운드)
Args:
message (str): 전송할 메시지
chat_id (str, optional): 채팅 ID
"""
# 텔레그램 알림이 비활성화된 경우 전송하지 않음
if not getattr(settings, 'TELEGRAM_NOTIFICATIONS_ENABLED', True):
print("[TELEGRAM_INFO] 텔레그램 알림이 비활성화되어 있습니다.")
return
def _send():
send_telegram_message(message, chat_id)
thread = threading.Thread(target=_send)
thread.daemon = True
thread.start()
def send_signup_notification(name, phone, request):
"""
회원가입 문자인증 요청 알림
Args:
name (str): 가입자 이름
phone (str): 전화번호
request: Django request 객체 (IP, User-Agent 정보)
"""
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 클라이언트 정보 추출
ip = get_client_ip(request)
user_agent = request.META.get('HTTP_USER_AGENT', '알 수 없음')
device_info = get_device_info(user_agent)
browser_info = get_browser_info(user_agent)
# 디버그 정보 출력
print(f"[TELEGRAM_DEBUG] 회원가입 알림 - IP: {ip}, 기기: {device_info}, 브라우저: {browser_info}")
# 상세 헤더 정보 (개발 환경에서만)
if ip.endswith('(로컬 개발환경)'):
print(f"[DEBUG] REMOTE_ADDR: {request.META.get('REMOTE_ADDR')}")
print(f"[DEBUG] HTTP_X_FORWARDED_FOR: {request.META.get('HTTP_X_FORWARDED_FOR')}")
print(f"[DEBUG] HTTP_X_REAL_IP: {request.META.get('HTTP_X_REAL_IP')}")
print(f"[DEBUG] HTTP_CF_CONNECTING_IP: {request.META.get('HTTP_CF_CONNECTING_IP')}")
print(f"[DEBUG] User-Agent: {user_agent[:100]}...")
message = f"""
🔔 <b>신라AMP 회원가입 알림</b>
👤 <b>이름:</b> {name}
📱 <b>전화번호:</b> {phone}
<b>요청시간:</b> {current_time}
🌐 <b>접속 정보:</b>
IP 주소: <code>{ip}</code>
기기: {device_info}
브라우저: {browser_info}
💡 새로운 회원가입 요청이 있습니다!
""".strip()
# 비동기로 전송 (사용자 대기시간 없음)
send_telegram_message_async(message)
def send_password_reset_notification(phone, request, user_exists=True):
"""
비밀번호 찾기 문자인증 요청 알림
Args:
phone (str): 전화번호
request: Django request 객체 (IP, User-Agent 정보)
user_exists (bool): 사용자가 등록되어 있는지 여부
"""
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 클라이언트 정보 추출
ip = get_client_ip(request)
user_agent = request.META.get('HTTP_USER_AGENT', '알 수 없음')
device_info = get_device_info(user_agent)
browser_info = get_browser_info(user_agent)
# 디버그 정보 출력
print(f"[TELEGRAM_DEBUG] 비밀번호 찾기 알림 - IP: {ip}, 기기: {device_info}, 브라우저: {browser_info}")
# 사용자 존재 여부에 따른 메시지 구성
if user_exists:
status_icon = ""
status_text = "등록된 사용자"
action_text = "비밀번호 찾기 요청이 있습니다!"
else:
status_icon = ""
status_text = "미등록 사용자"
action_text = "등록되지 않은 전화번호로 비밀번호 찾기 시도!"
message = f"""
🔑 <b>신라AMP 비밀번호 찾기 알림</b>
📱 <b>전화번호:</b> {phone}
{status_icon} <b>상태:</b> {status_text}
<b>요청시간:</b> {current_time}
🌐 <b>접속 정보:</b>
IP 주소: <code>{ip}</code>
기기: {device_info}
브라우저: {browser_info}
💡 {action_text}
""".strip()
# 비동기로 전송 (사용자 대기시간 없음)
send_telegram_message_async(message)
def test_telegram_bot():
"""
텔레그램 연결 테스트
"""
test_message = f"🧪 <b>신라AMP 봇 테스트</b>\n\n⏰ 테스트 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n✅ 봇이 정상적으로 작동하고 있습니다!"
result = send_telegram_message(test_message)
if result['success']:
print("✅ 텔레그램 봇 테스트 성공!")
return True
else:
print(f"❌ 텔레그램 봇 테스트 실패: {result['error']}")
return False