"""메일 한 통을 보낸다 — Azure Communication Services 우선, SMTP 폴백. ★ 설정이 없으면 보내지 않고 False 를 돌려준다(teams_webhook 과 같은 규약). 값이 없어도 서버는 뜬다. ★ 이 파일은 "메일 한 통 보내기" 만 안다 — 무엇을 언제 보낼지는 부르는 쪽이 정한다. ★ ACS 를 1순위로 두는 이유: 회사가 이미 공용 리소스를 쓰고 있고(negodata), 발신 도메인의 SPF·DKIM 을 그쪽이 관리한다. SMTP 는 그 리소스를 못 쓰는 환경의 폴백이다. """ import os import re import smtplib import ssl from email.message import EmailMessage from email.utils import formataddr, parseaddr from common.logger import LOG ACS_ENDPOINT_ENV = "ACS_EMAIL_ENDPOINT" ACS_KEY_ENV = "ACS_EMAIL_ACCESSKEY" ACS_SENDER_ENV = "ACS_EMAIL_SENDER" HOST_ENV = "SMTP_HOST" PORT_ENV = "SMTP_PORT" USER_ENV = "SMTP_USER" PASSWORD_ENV = "SMTP_PASSWORD" FROM_ENV = "SMTP_FROM" FROM_NAME_ENV = "SMTP_FROM_NAME" TLS_ENV = "SMTP_TLS" TIMEOUT_SEC = 15.0 DEFAULT_FROM_NAME = "Web4AI" def is_configured() -> bool: return _acs_configured() or _smtp_configured() def _acs_configured() -> bool: return bool(_env(ACS_ENDPOINT_ENV) and _env(ACS_KEY_ENV) and _env(ACS_SENDER_ENV)) def _smtp_configured() -> bool: return bool(_env(HOST_ENV) and _sender()) def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() def _sender() -> str: return _env(FROM_ENV) or _env(USER_ENV) def _port() -> int: try: return int(_env(PORT_ENV) or "587") except ValueError: return 587 def _mode() -> str: """starttls | ssl | plain. 기본은 starttls(587)이고 465 는 ssl 로 떨어진다.""" value = _env(TLS_ENV).lower() if value in {"ssl", "starttls", "plain"}: return value return "ssl" if _port() == 465 else "starttls" def is_valid_address(value: str) -> bool: address = parseaddr(value or "")[1] return bool(re.fullmatch(r"[^@\s]+@[^@\s.]+(\.[^@\s.]+)+", address)) def _send_acs(*, to: str, subject: str, text: str, reply_to: str | None) -> bool: try: from azure.communication.email import EmailClient client = EmailClient.from_connection_string( f"endpoint={_env(ACS_ENDPOINT_ENV)};accesskey={_env(ACS_KEY_ENV)}" ) message = { "senderAddress": _env(ACS_SENDER_ENV), "recipients": {"to": [{"address": to}]}, "content": {"subject": subject, "plainText": text}, } if reply_to: message["replyTo"] = [{"address": reply_to}] client.begin_send(message).result() return True except Exception as error: # noqa: BLE001 — 폴백(SMTP)이 남아 있다 LOG.e(f"[mail] ACS 발송 실패: {type(error).__name__}") return False def send(*, to: str, subject: str, text: str, reply_to: str | None = None) -> bool: """보냈으면 True. 설정이 없거나 실패하면 False — 예외를 밖으로 던지지 않는다. ★ 손님이 누른 버튼 하나가 메일 서버 장애로 500 이 되면 안 된다. 부르는 쪽이 False 를 보고 "전달하지 못했다"를 손님 화면의 말로 바꾼다.""" if not is_configured(): LOG.w("[mail] SMTP 미설정 — 보내지 않는다") return False if not is_valid_address(to): LOG.w("[mail] 받는 주소가 형식에 맞지 않는다") return False if _acs_configured(): if _send_acs(to=to, subject=subject, text=text, reply_to=reply_to): return True if not _smtp_configured(): return False if not _smtp_configured(): return False message = EmailMessage() message["Subject"] = subject message["From"] = formataddr((_env(FROM_NAME_ENV) or DEFAULT_FROM_NAME, _sender())) message["To"] = to if reply_to and is_valid_address(reply_to): message["Reply-To"] = reply_to message.set_content(text) host, port, mode = _env(HOST_ENV), _port(), _mode() user, password = _env(USER_ENV), _env(PASSWORD_ENV) try: if mode == "ssl": client = smtplib.SMTP_SSL(host, port, timeout=TIMEOUT_SEC, context=ssl.create_default_context()) else: client = smtplib.SMTP(host, port, timeout=TIMEOUT_SEC) with client: if mode == "starttls": client.starttls(context=ssl.create_default_context()) if user and password: client.login(user, password) client.send_message(message) return True except Exception as error: # noqa: BLE001 — 발송 실패를 호출측 실패로 만들지 않는다 LOG.e(f"[mail] 발송 실패: {type(error).__name__}") return False