"""협상 초청 메일 발송. 1순위: Azure Communication Services(ACS) Email (endpoint + accesskey + 검증된 sender). 2순위(폴백): SMTP (aiosmtplib). 둘 다 미설정이면 EmailUnavailable 을 던진다(조용한 실패 0). 설정 출처: config..toml 의 [MailConfig] (config/config_models.py:MailConfig). HTML 본문 템플릿: services/email_templates/*.html ($placeholder 치환). """ from __future__ import annotations from datetime import datetime from email.message import EmailMessage from html import escape from pathlib import Path from string import Template from zoneinfo import ZoneInfo from common.logger import LOG from config.server_configs import mail_config _KST = ZoneInfo("Asia/Seoul") # HTML 본문은 코드에 박지 않고 파일에서 읽는다(모듈 로드 시 1회). $placeholder 는 string.Template 로 치환. _TEMPLATE_DIR = Path(__file__).parent / "email_templates" _INVITE_HTML = Template((_TEMPLATE_DIR / "invite_email.html").read_text(encoding="utf-8")) class EmailUnavailable(RuntimeError): """ACS·SMTP 모두 미설정이라 발송 채널이 없음.""" async def _send_acs(to: str, subject: str, html: str, text: str) -> None: """Azure Communication Services Email — accesskey 인증.""" from azure.communication.email.aio import EmailClient conn = f"endpoint={mail_config.azure_acs_endpoint};accesskey={mail_config.azure_acs_accesskey}" message = { "senderAddress": mail_config.azure_acs_sender, "recipients": {"to": [{"address": to}]}, "content": {"subject": subject, "plainText": text, "html": html}, } async with EmailClient.from_connection_string(conn) as client: poller = await client.begin_send(message) await poller.result() LOG.i(f"email sent via ACS → {to} ({subject})") async def _send_smtp(to: str, subject: str, html: str, text: str) -> None: import aiosmtplib msg = EmailMessage() msg["From"] = mail_config.smtp_from msg["To"] = to msg["Subject"] = subject msg.set_content(text) msg.add_alternative(html, subtype="html") await aiosmtplib.send( msg, hostname=mail_config.smtp_host, port=mail_config.smtp_port, username=mail_config.smtp_user or None, password=mail_config.smtp_password or None, start_tls=mail_config.smtp_starttls, ) LOG.i(f"email sent via SMTP → {to} ({subject})") async def send_email(to: str, subject: str, html: str, text: str) -> None: """ACS(설정 시) → SMTP(폴백) 순으로 1통 발송. 둘 다 없으면 EmailUnavailable.""" if mail_config.acs_configured: await _send_acs(to, subject, html, text) return if mail_config.smtp_host: await _send_smtp(to, subject, html, text) return # endpoint/accesskey 만 있고 sender 누락 시 원인을 명확히 안내. if mail_config.azure_acs_endpoint and not mail_config.azure_acs_sender: raise EmailUnavailable("AZURE_ACS_SENDER(검증된 MailFrom 주소) 미설정") raise EmailUnavailable("이메일 미설정 — ACS(endpoint+accesskey+sender) 또는 SMTP 필요") def _fmt_deadline(end_time: datetime | None) -> str: """협상 마감 시각 → 한국시간 'YYYY-MM-DD HH:MM' 표기. 값 없으면 빈 문자열.""" if not end_time: return "" dt = end_time if end_time.tzinfo else end_time.replace(tzinfo=ZoneInfo("UTC")) return dt.astimezone(_KST).strftime("%Y-%m-%d %H:%M") def build_invite_email( *, supplier_name: str, quotation_name: str, qt_number: str, end_time: datetime | None, chat_url: str, email_header: str | None = None, ) -> tuple[str, str, str]: """협상 초청 메일 (제목/HTML/텍스트) 생성. 목표가·앵커링가는 협상 전략 값이라 메일에 담지 않는다(공급사에게 노출 금지). 공급사는 링크로 협상 화면에 진입해 입찰한다. """ sp = supplier_name or "협력사" deadline = _fmt_deadline(end_time) or "미정" subject = f"[협상 견적 {qt_number}] {quotation_name} — 협상 참여 요청" # HTML 본문은 invite_email.html 에서 읽어 치환. 값은 escape 해 HTML 인젝션 방지(견적명 등은 사용자 입력). html = _INVITE_HTML.substitute( supplier_name=escape(sp), quotation_name=escape(quotation_name), qt_number=escape(qt_number), deadline=escape(deadline), chat_url=escape(chat_url), email_header=escape(email_header or "NEGODATA"), ) text = ( f"{sp} 담당자님, 아래 견적 건의 협상에 참여해 주세요.\n\n" f" - 견적명: {quotation_name}\n" f" - 견적번호: {qt_number}\n" f" - 협상 마감: {deadline}\n\n" f"협상 참여 링크: {chat_url}\n" ) return subject, html, text