From 6e23dc8b2078e700f74e742452f7fc8662e85a45 Mon Sep 17 00:00:00 2001 From: thug0bin Date: Fri, 27 Feb 2026 14:55:07 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20=EC=84=9C=EB=B2=84=20=EC=8B=9C=EC=9E=91?= =?UTF-8?q?=20=EC=8B=9C=20=ED=8F=AC=ED=8A=B8=20=EC=B6=A9=EB=8F=8C=20?= =?UTF-8?q?=EC=9E=90=EB=8F=99=20=ED=95=B4=EA=B2=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 포트 7001 사용 중이면 기존 프로세스 자동 종료 - Flask reloader 자식 프로세스 구분 처리 - check_port_available(), kill_process_on_port() 함수 추가 --- backend/app.py | 50 +++++++++++++- backend/sms_client.py | 147 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 backend/sms_client.py diff --git a/backend/app.py b/backend/app.py index 1fa97c5..8afc171 100644 --- a/backend/app.py +++ b/backend/app.py @@ -3151,6 +3151,54 @@ def api_qr_preview(): return jsonify({'success': False, 'error': str(e)}), 500 +def check_port_available(port: int) -> bool: + """포트가 사용 가능한지 확인""" + import socket + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(1) + result = sock.connect_ex(('127.0.0.1', port)) + sock.close() + return result != 0 # 0이면 이미 사용 중, 0이 아니면 사용 가능 + + +def kill_process_on_port(port: int) -> bool: + """특정 포트를 사용하는 프로세스 종료 (Windows)""" + import subprocess + try: + # netstat으로 PID 찾기 + result = subprocess.run( + f'netstat -ano | findstr ":{port}"', + shell=True, capture_output=True, text=True + ) + + for line in result.stdout.strip().split('\n'): + if 'LISTENING' in line: + parts = line.split() + pid = parts[-1] + if pid.isdigit(): + subprocess.run(f'taskkill /F /PID {pid}', shell=True) + logging.info(f"포트 {port} 사용 중인 프로세스 종료: PID {pid}") + return True + return False + except Exception as e: + logging.error(f"프로세스 종료 실패: {e}") + return False + + if __name__ == '__main__': + import os + + PORT = 7001 + + # Flask reloader 자식 프로세스가 아닌 경우에만 체크 + if os.environ.get('WERKZEUG_RUN_MAIN') != 'true': + if not check_port_available(PORT): + logging.warning(f"포트 {PORT}이 이미 사용 중입니다. 기존 프로세스를 종료합니다...") + if kill_process_on_port(PORT): + import time + time.sleep(2) # 프로세스 종료 대기 + else: + logging.error(f"포트 {PORT} 해제 실패. 수동으로 확인하세요.") + # 개발 모드로 실행 - app.run(host='0.0.0.0', port=7001, debug=True) + app.run(host='0.0.0.0', port=PORT, debug=True) diff --git a/backend/sms_client.py b/backend/sms_client.py new file mode 100644 index 0000000..51805f2 --- /dev/null +++ b/backend/sms_client.py @@ -0,0 +1,147 @@ +# sms_client.py - NHN Cloud SMS API 클라이언트 + +import requests +import json +import logging +from typing import Dict, List + +# NHN Cloud SMS 설정 (SMS 전용 앱키) +SMS_CONFIG = { + "BASE_URL": "https://api-sms.cloud.toast.com", + "APP_KEY": "YWWBZkuJ0ck03cje", + "SECRET_KEY": "jxXbBPnQN2tUL8QnEp4O3YfraGd8ZuNh", + "SENDER_NO": "0334817390", # 발신번호 (033-481-7390) +} + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +class SMSClient: + """NHN Cloud SMS 발송 클라이언트""" + + def __init__(self): + self.base_url = SMS_CONFIG["BASE_URL"] + self.app_key = SMS_CONFIG["APP_KEY"] + self.secret_key = SMS_CONFIG["SECRET_KEY"] + self.sender_no = SMS_CONFIG["SENDER_NO"] + + def _get_headers(self) -> Dict[str, str]: + return { + "Content-Type": "application/json;charset=UTF-8", + "X-Secret-Key": self.secret_key + } + + def send_sms(self, recipients: List[Dict], message: str) -> Dict: + """ + SMS 발송 + + Args: + recipients: [{"phone": "01012345678", "name": "홍길동"}] + message: 메시지 내용 (90바이트 이하 SMS, 초과시 LMS) + + Returns: + 발송 결과 + """ + # 메시지 길이에 따라 SMS/LMS 결정 + msg_bytes = len(message.encode('utf-8')) + is_lms = msg_bytes > 90 + + url = f"{self.base_url}/sms/v3.0/appKeys/{self.app_key}/sender/{'mms' if is_lms else 'sms'}" + + # 수신자 리스트 생성 + recipient_list = [] + for r in recipients: + phone = (r.get('phone') or '').replace('-', '').replace(' ', '') + if phone and len(phone) >= 10: + recipient_list.append({ + "recipientNo": phone, + "countryCode": "82" + }) + + if not recipient_list: + return { + "success": False, + "error": "유효한 수신자가 없습니다" + } + + # 요청 데이터 + data = { + "body": message, + "sendNo": self.sender_no, + "recipientList": recipient_list + } + + # LMS인 경우 제목 추가 + if is_lms: + data["title"] = "청춘약국" + + try: + logger.info(f"SMS 발송 요청: {len(recipient_list)}명, {msg_bytes}bytes ({'LMS' if is_lms else 'SMS'})") + + response = requests.post( + url, + headers=self._get_headers(), + data=json.dumps(data), + timeout=30 + ) + + result = response.json() + logger.info(f"SMS 응답: {result}") + + header = result.get("header", {}) + if header.get("isSuccessful"): + body = result.get("body", {}) + return { + "success": True, + "message": f"SMS 발송 성공 ({len(recipient_list)}명)", + "type": "LMS" if is_lms else "SMS", + "request_id": body.get("data", {}).get("requestId"), + "sent_count": len(recipient_list) + } + else: + return { + "success": False, + "error": header.get("resultMessage", "발송 실패"), + "code": header.get("resultCode") + } + + except requests.exceptions.Timeout: + return {"success": False, "error": "요청 시간 초과"} + except requests.exceptions.RequestException as e: + logger.error(f"SMS 발송 오류: {e}") + return {"success": False, "error": str(e)} + except Exception as e: + logger.error(f"SMS 발송 예외: {e}") + return {"success": False, "error": str(e)} + + def check_balance(self) -> Dict: + """잔여 발송량 확인""" + url = f"{self.base_url}/sms/v3.0/appKeys/{self.app_key}/stats" + + try: + response = requests.get(url, headers=self._get_headers(), timeout=10) + return response.json() + except Exception as e: + return {"success": False, "error": str(e)} + + +# 싱글톤 인스턴스 +sms_client = SMSClient() + + +def send_test_sms(phone: str, message: str = None) -> Dict: + """테스트 SMS 발송""" + if not message: + message = "[청춘약국] 테스트 문자입니다. 정상 수신되었다면 회신 부탁드립니다." + + return sms_client.send_sms( + recipients=[{"phone": phone, "name": "테스트"}], + message=message + ) + + +if __name__ == "__main__": + # 테스트 발송 + result = send_test_sms("01027027390") + print(json.dumps(result, ensure_ascii=False, indent=2))