feat(D04): 예정노선 내보내기 엔진 · 메일 첨부 인자 (531-1 · 531-2 일부)

D04_PlanConfirm/D04_PlanConfirm_Export.py

@@ -0,0 +1,243 @@
+"""D04 예정노선 내보내기 — 연도별 노선(꺾은선)을 용화 예정노선 양식 shapefile 한 벌로 만든다.
+
+양식 = `resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.*`
+(분석 345-C): 2D 폴리라인 1레코드 · UTM-K(EPSG 5179) · cp949 · 칸 `대상지` C(20) · `길이` N(11,3) km
++ 덧칸 `연도` N(4) · `시점누가` · `종점누가` N(11,2) m. `.prj` 는 용화 원문 그대로 · `.cpg` 는 `949`.
+
+계산 자리 ③ 서버 단독 — 화면이 만지지 않는다. 순수 함수(값 → 바이트)만 두고 파일은 임시 폴더에서만 쓴다.
+"""
+
+from __future__ import annotations
+
+import io
+import math
+import re
+import tempfile
+import zipfile
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import geopandas as gpd
+from openpyxl import Workbook
+from pyproj import Transformer
+from shapely.geometry import LineString
+
+EXPORT_EPSG = 5179
+SHP_SUFFIXES = (".shp", ".shx", ".dbf", ".prj", ".cpg")
+MIME_ZIP = "application/zip"
+MIME_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+
+# 용화 `.prj` 원문(ESRI WKT) — 양식 일치용으로 그대로 씀.
+_YONGHWA_PRJ = (
+    Path(__file__).resolve().parents[1]
+    / "resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.prj"
+)
+_SCHEMA = {
+    "geometry": "LineString",
+    "properties": {
+        "대상지": "str:20",
+        "길이": "float:11.3",
+        "연도": "int:4",
+        "시점누가": "float:11.2",
+        "종점누가": "float:11.2",
+    },
+}
+_BAD_FILENAME = re.compile(r'[\\/:*?"<>|\r\n\t]')
+
+
+@dataclass
+class YearRoute:
+    """한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점)."""
+
+    year: int
+    points: list[tuple[float, float]]
+    crs_epsg: int
+
+
+@dataclass
+class ExportBundle:
+    """`by_year[연도]` = 그 연도 shp 한 벌 + 좌표표 · `all` = 전 연도 이은 한 벌 + 좌표표."""
+
+    by_year: dict[int, dict[str, bytes]] = field(default_factory=dict)
+    all: dict[str, bytes] = field(default_factory=dict)
+    lengths_m: dict[int, float] = field(default_factory=dict)
+
+
+def safe_name(text: str) -> str:
+    """파일명에 못 쓰는 글자를 `_` 로."""
+    return _BAD_FILENAME.sub("_", text).strip() or "_"
+
+
+def polyline_length_m(points: list[tuple[float, float]]) -> float:
+    return sum(math.dist(a, b) for a, b in zip(points, points[1:], strict=False))
+
+
+def to_utmk(points: list[tuple[float, float]], crs_epsg: int) -> list[tuple[float, float]]:
+    """프로젝트 좌표 → UTM-K(5179). 같은 좌표계면 그대로."""
+    pts = [(float(p[0]), float(p[1])) for p in points]
+    if crs_epsg == EXPORT_EPSG:
+        return pts
+    tf = Transformer.from_crs(crs_epsg, EXPORT_EPSG, always_xy=True)
+    return [tf.transform(x, y) for x, y in pts]
+
+
+def _clip_bytes(text: str, limit: int = 20) -> str:
+    """dbf 문자 칸(cp949 바이트) 한도에 글자 단위로 맞춤."""
+    while len(text.encode("cp949", errors="replace")) > limit:
+        text = text[:-1]
+    return text
+
+
+def _target_text(region: str, length_m: float) -> str:
+    return f"{region}.{length_m / 1000:.1f}"
+
+
+def _write_shapefile(
+    points: list[tuple[float, float]], year: int | None, stations: tuple[float, float], region: str
+) -> dict[str, bytes]:
+    """5179 꺾은선 → `.shp/.shx/.dbf/.prj/.cpg` 바이트 (파일명은 호출부가 붙임: 키 = 확장자)."""
+    length_m = polyline_length_m(points)
+    frame = gpd.GeoDataFrame(
+        {
+            "대상지": [_clip_bytes(_target_text(region, length_m))],
+            "길이": [round(length_m / 1000, 3)],
+            "연도": [year],
+            "시점누가": [round(stations[0], 2)],
+            "종점누가": [round(stations[1], 2)],
+        },
+        geometry=[LineString(points)],
+        crs=EXPORT_EPSG,
+    )
+    with tempfile.TemporaryDirectory() as tmp:
+        out = Path(tmp) / "r.shp"
+        frame.to_file(
+            out, driver="ESRI Shapefile", encoding="cp949", engine="fiona", schema=_SCHEMA
+        )
+        out.with_suffix(".prj").write_bytes(_YONGHWA_PRJ.read_bytes())
+        out.with_suffix(".cpg").write_text("949", encoding="ascii")
+        return {suffix: out.with_suffix(suffix).read_bytes() for suffix in SHP_SUFFIXES}
+
+
+def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes:
+    """사람이 읽는 좌표표 — 연도별 시점 · 종점(UTM-K · 위경도) · 누가 · 연장."""
+    to_ll = Transformer.from_crs(EXPORT_EPSG, 4326, always_xy=True)
+    book = Workbook()
+    sheet = book.active
+    sheet.title = "좌표표"
+    sheet.append([f"{project_name} 예정노선 좌표표 (UTM-K EPSG:5179)"])
+    sheet.append(
+        [
+            "연도",
+            "시점 X",
+            "시점 Y",
+            "종점 X",
+            "종점 Y",
+            "시점 경도",
+            "시점 위도",
+            "종점 경도",
+            "종점 위도",
+            "시점누가(m)",
+            "종점누가(m)",
+            "연장(m)",
+        ]
+    )
+    for row in rows:
+        (sx, sy), (ex, ey) = row["points"][0], row["points"][-1]
+        (slon, slat), (elon, elat) = to_ll.transform(sx, sy), to_ll.transform(ex, ey)
+        sheet.append(
+            [
+                row["year"],
+                round(sx, 3),
+                round(sy, 3),
+                round(ex, 3),
+                round(ey, 3),
+                round(slon, 7),
+                round(slat, 7),
+                round(elon, 7),
+                round(elat, 7),
+                round(row["start_station"], 2),
+                round(row["end_station"], 2),
+                round(row["length_m"], 2),
+            ]
+        )
+    buffer = io.BytesIO()
+    book.save(buffer)
+    return buffer.getvalue()
+
+
+def _named(base: str, parts: dict[str, bytes]) -> dict[str, bytes]:
+    return {f"{base}{suffix}": data for suffix, data in parts.items()}
+
+
+def build_export(
+    project_name: str,
+    region: str,
+    years: list[YearRoute],
+    first_station_m: float = 0.0,
+    with_xlsx: bool = True,
+) -> ExportBundle:
+    """연도 노선들 → 연도별 한 벌 + 전체 한 벌(+ 좌표표).
+
+    누가거리 = `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 = 다음 연도 시점).
+    shp 안 측점은 파일마다 0 부터가 아니라 이 누가값을 칸으로만 보관(B03 은 칸을 무시).
+    """
+    ordered = sorted(years, key=lambda y: y.year)
+    rows: list[dict] = []
+    station = float(first_station_m)
+    for item in ordered:
+        points = to_utmk(item.points, item.crs_epsg)
+        if len(points) < 2:
+            raise ValueError(f"{item.year}년 노선 점이 2개 미만입니다.")
+        length_m = polyline_length_m(points)
+        rows.append(
+            {
+                "year": item.year,
+                "points": points,
+                "length_m": length_m,
+                "start_station": station,
+                "end_station": station + length_m,
+            }
+        )
+        station += length_m
+
+    bundle = ExportBundle()
+    region_name = safe_name(region)
+    xlsx_name = f"좌표표_{safe_name(project_name)}.xlsx"
+    for row in rows:
+        base = f"대상지_{region_name}.{row['length_m'] / 1000:.1f}_{row['year']}"
+        files = _named(
+            base,
+            _write_shapefile(
+                row["points"], row["year"], (row["start_station"], row["end_station"]), region
+            ),
+        )
+        if with_xlsx:
+            files[xlsx_name] = _coordinate_xlsx(project_name, [row])
+        bundle.by_year[row["year"]] = files
+        bundle.lengths_m[row["year"]] = row["length_m"]
+
+    # 전체: 앞 연도 종점 = 다음 연도 시점이라 겹치는 점 하나는 뺌.
+    merged: list[tuple[float, float]] = []
+    for row in rows:
+        merged.extend(
+            row["points"][1:] if merged and merged[-1] == row["points"][0] else row["points"]
+        )
+    total_m = polyline_length_m(merged)
+    bundle.lengths_m[0] = total_m
+    base = f"대상지_{region_name}.{total_m / 1000:.1f}_전체"
+    bundle.all = _named(
+        base,
+        _write_shapefile(merged, None, (rows[0]["start_station"], rows[-1]["end_station"]), region),
+    )
+    if with_xlsx:
+        bundle.all[xlsx_name] = _coordinate_xlsx(project_name, rows)
+    return bundle
+
+
+def company_zip(project_name: str, year: int, files: dict[str, bytes]) -> tuple[str, bytes]:
+    """업체에 보낼 `<프로젝트명>_예정노선_<연도>.zip` (이름, 바이트)."""
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
+        for name, data in files.items():
+            archive.writestr(name, data)
+    return f"{safe_name(project_name)}_예정노선_{year}.zip", buffer.getvalue()

common_util/common_util_email.py

@@ -22,13 +22,22 @@ from config.config_system import (
 logger = logging.getLogger(__name__)
 
 
-def _send_sync(to_email: str, subject: str, html: str) -> None:
+# 첨부 = (파일명, 바이트, MIME 예 "application/zip")
+Attachment = tuple[str, bytes, str]
+
+
+def _send_sync(
+    to_email: str, subject: str, html: str, attachments: list[Attachment] | None = None
+) -> None:
     message = EmailMessage()
     message["From"] = formataddr((EMAIL_FROM_NAME, EMAIL_FROM_ADDRESS))
     message["To"] = to_email
     message["Subject"] = subject
     message.set_content("HTML 이메일을 지원하는 메일 앱에서 확인해 주세요.")
     message.add_alternative(html, subtype="html")
+    for name, data, mime in attachments or []:
+        maintype, _, subtype = mime.partition("/")
+        message.add_attachment(data, maintype=maintype, subtype=subtype, filename=name)
 
     with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as smtp:
         if SMTP_USE_TLS:
@@ -37,8 +46,10 @@ def _send_sync(to_email: str, subject: str, html: str) -> None:
         smtp.send_message(message)
 
 
-async def send_email(to_email: str, subject: str, html: str) -> bool:
-    """SMTP 발송을 워커 스레드에서 실행하고 설정된 횟수만큼 재시도한다."""
+async def send_email(
+    to_email: str, subject: str, html: str, attachments: list[Attachment] | None = None
+) -> bool:
+    """SMTP 발송을 워커 스레드에서 실행하고 설정된 횟수만큼 재시도한다. 첨부는 선택."""
     if not EMAIL_NOTIFICATION_ENABLED:
         logger.info("이메일 발송 비활성화: %s", to_email)
         return True
@@ -48,7 +59,7 @@ async def send_email(to_email: str, subject: str, html: str) -> bool:
 
     for attempt in range(1, EMAIL_RETRY_COUNT + 1):
         try:
-            await asyncio.to_thread(_send_sync, to_email, subject, html)
+            await asyncio.to_thread(_send_sync, to_email, subject, html, attachments)
             logger.info("이메일 발송 성공: %s", to_email)
             return True
         except (OSError, smtplib.SMTPException):