"""B03 업로드 및 B04 WF1 분석 결과 이메일 발송."""
import html
import logging
from typing import Any
from uuid import UUID
from common_util.common_util_email import send_email
from config.config_system import APP_PUBLIC_BASE_URL
logger = logging.getLogger(__name__)
def _format_number(value: Any) -> str:
if isinstance(value, int | float):
return f"{value:,}"
return "N/A"
def _format_elevation(value: Any) -> str:
if isinstance(value, int | float):
return f"{value:.2f}"
return "N/A"
def _extract_elevation_range(bounds: dict[str, Any], statistics: dict[str, Any]) -> tuple[Any, Any]:
if isinstance(bounds.get("z"), list | tuple) and len(bounds["z"]) >= 2:
return bounds["z"][0], bounds["z"][1]
return (
bounds.get("min_z", statistics.get("min_z")),
bounds.get("max_z", statistics.get("max_z")),
)
def _app_url(path: str) -> str:
return f"{APP_PUBLIC_BASE_URL}/{path.lstrip('/')}"
def _email_shell(title: str, body: str, *, accent: str = "#2563eb") -> str:
box_style = (
f"border: 1px solid #e5e7eb; border-left: 4px solid {accent}; "
"padding: 18px; margin: 20px 0;"
)
row_style = (
"display: flex; justify-content: space-between; gap: 16px; padding: 9px 0; "
"border-bottom: 1px solid #f3f4f6;"
)
return f"""
"""
def _summary_row(label: str, value: str) -> str:
return (
''
f'{html.escape(label)}'
f'{value}'
"
"
)
async def send_file_upload_complete_email(
*,
to_email: str,
user_name: str,
project_name: str,
file_name: str,
file_size_mb: float,
metadata: dict[str, Any],
) -> bool:
"""파일 저장만 끝난 시점의 안내 메일.
프로젝트 업로드 경로에서는 더 이상 쓰지 않는다 — 그 흐름은 초기 설계까지 마친 뒤
[[send_initial_analysis_complete_email]] 한 통으로 알린다. 이 메일은 프로젝트를
만들기 전 임시 보관함(storage/tmp)에 자료를 올렸을 때 보관 안내용으로 쓴다
(2026-08-08 사용자 지시, 임시 보관 기능 구현 시 연결).
"""
bounds = metadata.get("bounds") or {}
statistics = metadata.get("statistics") or {}
min_z, max_z = _extract_elevation_range(bounds, statistics)
point_count = metadata.get("point_count", metadata.get("points", 0))
epsg = metadata.get("epsg", "N/A")
rows = "\n".join(
[
_summary_row("파일명", html.escape(file_name)),
_summary_row("파일 크기", f"{file_size_mb:.2f} MB"),
_summary_row("포인트 수", _format_number(point_count)),
_summary_row("좌표계", f"EPSG:{html.escape(str(epsg))}"),
_summary_row(
"고도 범위",
f"{_format_elevation(min_z)} ~ {_format_elevation(max_z)} m",
),
]
)
body = f"""
안녕하세요, {html.escape(user_name or "사용자")}님.
프로젝트 {html.escape(project_name)}의 입력 파일이 저장되었습니다.
{rows}
지표면 분석(WF1)이 백그라운드에서 자동으로 시작되었습니다.
완료되면 분석 요약과 결과 페이지 링크를 다시 보내드립니다.
"""
subject = f"파일 업로드 완료 - {project_name}"
success = await send_email(
to_email=to_email,
subject=subject,
html=_email_shell(subject, body),
)
logger.info("파일 업로드 완료 이메일 발송 %s: %s", "성공" if success else "실패", to_email)
return success
async def send_initial_analysis_complete_email(
*,
project_id: UUID,
project_name: str,
to_email: str,
analysis_result: dict[str, Any],
design_summary: dict[str, Any] | None = None,
) -> bool:
"""업로드 ~ 초기 설계(B04 전처리·B05 종단·B06 횡단)까지 마친 뒤 한 통으로 알린다.
예전에는 업로드 직후와 지표면 분석 직후에 각각 메일을 보내 같은 작업으로 두 통이
도착했다. 사용자가 실제로 확인할 시점은 초기 설계까지 끝나 화면에서 검토할 수 있게
된 때 하나뿐이므로 그때 한 통만 보낸다(2026-08-08 사용자 지시).
`design_summary`가 없으면(자동 설계 체인 미실행·실패) 전처리 결과까지만 싣는다.
"""
processed = analysis_result.get("processed") or {}
statistics = processed.get("statistics") or {}
bounds = processed.get("bounds") or {}
models = analysis_result.get("models") or []
min_z = statistics.get("min_z", bounds.get("min_z"))
max_z = statistics.get("max_z", bounds.get("max_z"))
point_count = processed.get("point_count", 0)
result_url = _app_url("/#/b05-profile")
surface_rows = "\n".join(
[
_summary_row("분석 포인트 수", _format_number(point_count)),
_summary_row(
"고도 범위",
f"{_format_elevation(min_z)} ~ {_format_elevation(max_z)} m",
),
_summary_row("생성 지표면 모델", f"{len(models):,}개"),
]
)
if design_summary:
length_m = design_summary.get("length_m")
design_rows = "\n".join(
[
_summary_row(
"노선 연장",
f"{length_m:,.1f} m" if isinstance(length_m, int | float) else "N/A",
),
_summary_row("측점 수", _format_number(design_summary.get("cross_section_count"))),
_summary_row("초기 설계", "종단 계획선·횡단 기본 설계 저장 완료"),
]
)
design_block = f"""
초기 노선·종횡단 설계
{design_rows}
기본값으로 계산한 결과입니다. 종단설계 화면에서 검토·수정한 뒤 횡단설계에서
[확정]을 누르면 다음 단계(수량 산출)로 넘어갑니다.
"""
else:
design_block = """
초기 노선·종횡단 자동 계산은 완료되지 않았습니다. 종단설계 화면에서 직접
경로를 계산해 주세요.
"""
body = f"""
프로젝트 {html.escape(project_name)}의 입력 파일 저장과
초기 분석이 모두 끝났습니다.
지표면 전처리
{surface_rows}
{design_block}
종단설계 화면 열기
로그인이 필요한 경우 인증 후 같은 화면으로 이동합니다.
"""
subject = f"임도 초기 분석 완료 - {project_name}"
success = await send_email(
to_email=to_email,
subject=subject,
html=_email_shell(subject, body),
)
logger.info("초기 분석 완료 이메일 발송 %s: %s", "성공" if success else "실패", to_email)
return success
async def send_analysis_error_email(
*,
project_id: UUID,
project_name: str,
to_email: str,
error_message: str,
) -> bool:
"""WF1 Surface 분석 실패 알림 이메일을 발송한다."""
rows = "\n".join(
[
_summary_row("프로젝트 ID", html.escape(str(project_id))),
_summary_row("오류 내용", html.escape(error_message)),
]
)
safe_name = html.escape(project_name)
body = f"""
프로젝트 {safe_name}의 지표면 분석 중 오류가 발생했습니다.
{rows}
기술 지원팀에서 확인할 수 있도록 서버 로그에도 같은 오류를 기록했습니다.
"""
subject = f"지표면 분석 실패 알림 - {project_name}"
success = await send_email(
to_email=to_email,
subject=subject,
html=_email_shell(subject, body, accent="#dc2626"),
)
logger.info("WF1 오류 이메일 발송 %s: %s", "성공" if success else "실패", to_email)
return success