284 lines
9.7 KiB
Python
284 lines
9.7 KiB
Python
"""
|
|
텔레그램 봇 유틸리티
|
|
회원가입 알림 등을 위한 텔레그램 메시지 전송 기능
|
|
"""
|
|
|
|
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
|
|
"""
|
|
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
|