From 557681499c5e279b154444fc4f25a283c908a7b5 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 19:45:23 +0900 Subject: [PATCH] =?UTF-8?q?feat(D04):=20=EC=98=88=EC=A0=95=EB=85=B8?= =?UTF-8?q?=EC=84=A0=20=EB=82=B4=EB=B3=B4=EB=82=B4=EA=B8=B0=20=EC=97=94?= =?UTF-8?q?=EC=A7=84=20=C2=B7=20=EB=A9=94=EC=9D=BC=20=EC=B2=A8=EB=B6=80=20?= =?UTF-8?q?=EC=9D=B8=EC=9E=90=20(531-1=20=C2=B7=20531-2=20=EC=9D=BC?= =?UTF-8?q?=EB=B6=80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- D04_PlanConfirm/D04_PlanConfirm_Export.py | 243 ++++++++++++++++++++++ common_util/common_util_email.py | 19 +- 2 files changed, 258 insertions(+), 4 deletions(-) create mode 100644 D04_PlanConfirm/D04_PlanConfirm_Export.py diff --git a/D04_PlanConfirm/D04_PlanConfirm_Export.py b/D04_PlanConfirm/D04_PlanConfirm_Export.py new file mode 100644 index 000000000..b9b20806a --- /dev/null +++ b/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() diff --git a/common_util/common_util_email.py b/common_util/common_util_email.py index 1123a3685..865596187 100644 --- a/common_util/common_util_email.py +++ b/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):