From 20081ee3629bdd45db9d63bb6fded0f7ee4e2c3f Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 19:49:39 +0900 Subject: [PATCH] =?UTF-8?q?feat(D04):=20=EC=97=85=EC=B2=B4=20=EC=A7=80?= =?UTF-8?q?=EC=A0=95=20=C2=B7=20=ED=99=95=EC=A0=95=20API=20=C2=B7=20?= =?UTF-8?q?=EC=99=84=EB=A3=8C=20=EB=A9=94=EC=9D=BC=20=C2=B7=20=EC=8B=9C?= =?UTF-8?q?=ED=97=98=20(531-1=20=C2=B7=20531-2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- D04_PlanConfirm/D04_PlanConfirm_Email.py | 50 +++ D04_PlanConfirm/D04_PlanConfirm_Export.py | 29 +- D04_PlanConfirm/D04_PlanConfirm_Repository.py | 119 +++++ D04_PlanConfirm/D04_PlanConfirm_Router.py | 214 +++++++++ main.py | 2 + .../tester/test_531_d04_export_finalize.py | 411 ++++++++++++++++++ 6 files changed, 817 insertions(+), 8 deletions(-) create mode 100644 D04_PlanConfirm/D04_PlanConfirm_Email.py create mode 100644 D04_PlanConfirm/D04_PlanConfirm_Repository.py create mode 100644 D04_PlanConfirm/D04_PlanConfirm_Router.py create mode 100644 resources/tester/test_531_d04_export_finalize.py diff --git a/D04_PlanConfirm/D04_PlanConfirm_Email.py b/D04_PlanConfirm/D04_PlanConfirm_Email.py new file mode 100644 index 000000000..09d68aeb9 --- /dev/null +++ b/D04_PlanConfirm/D04_PlanConfirm_Email.py @@ -0,0 +1,50 @@ +"""D04 예정노선 확정 메일 — 업체에 그 연도 노선 묶음(zip)을 첨부해 보낸다 (요구 349-9).""" + +from __future__ import annotations + +import html + +from common_util.common_util_email import Attachment, send_email + + +def build_completion_mail( + *, + project_name: str, + region: str | None, + year: int, + length_m: float, + recipient: str, + attachment_names: list[str], +) -> tuple[str, str]: + """(제목, HTML 본문) — 프로젝트명 · 연도 · 연장 · 첨부 목록.""" + esc = html.escape + rows = [ + ("프로젝트", project_name), + ("사업 지역", region or "-"), + ("연도", f"{year}년"), + ("연장", f"{length_m / 1000:,.3f} km ({length_m:,.1f} m)"), + ] + table = "".join( + f"{esc(k)}" + f"{esc(v)}" + for k, v in rows + ) + files = "".join(f"
  • {esc(name)}
    • " for name in attachment_names) + body = f""" + + +

    안녕하세요, {esc(recipient)} 담당자님.

    +

    예정노선이 확정되어 {year}년 노선 파일을 보내 드립니다.

    +{table}
    +

    첨부 파일

    + +

    압축을 풀면 shapefile(.shp · .shx · .dbf · .prj · .cpg · 좌표계 UTM-K)이 들어 있으며, +임도 설계 프로그램의 노선 입력에 그대로 올릴 수 있습니다.

    +""" + return f"[예정노선 확정] {project_name} {year}년", body + + +async def send_completion_mail( + *, to_email: str, subject: str, body: str, attachments: list[Attachment] +) -> bool: + return await send_email(to_email=to_email, subject=subject, html=body, attachments=attachments) diff --git a/D04_PlanConfirm/D04_PlanConfirm_Export.py b/D04_PlanConfirm/D04_PlanConfirm_Export.py index be6b6deba..8bbb9cf78 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Export.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Export.py @@ -29,10 +29,14 @@ SHP_SUFFIXES = (".shp", ".shx", ".dbf", ".prj", ".cpg") MIME_ZIP = "application/zip" MIME_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" -# 용화 `.prj` 원문(ESRI WKT) — 양식 일치용으로 그대로 씀. +# 용화 `.prj` 원문(ESRI WKT) 그대로 — 양식 일치(시험이 용화 파일과 바이트 비교). _YONGHWA_PRJ = ( - Path(__file__).resolve().parents[1] - / "resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.prj" + 'PROJCS["PCS_ITRF2000_UTM_K",GEOGCS["GCS_ITRF_2000",DATUM["D_ITRF_2000",' + 'SPHEROID["GRS_1980",6378137.0,298.257222101]],PRIMEM["Greenwich",0.0],' + 'UNIT["Degree",0.0174532925199433]],PROJECTION["Transverse_Mercator"],' + 'PARAMETER["False_Easting",1000000.0],PARAMETER["False_Northing",2000000.0],' + 'PARAMETER["Central_Meridian",127.5],PARAMETER["Scale_Factor",0.9996],' + 'PARAMETER["Latitude_Of_Origin",38.0],UNIT["METER",1.0]]' ) _SCHEMA = { "geometry": "LineString", @@ -49,11 +53,16 @@ _BAD_FILENAME = re.compile(r'[\\/:*?"<>|\r\n\t]') @dataclass class YearRoute: - """한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점).""" + """한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점). + + 누가거리 두 값이 있으면 그대로 칸에 쓰고(D03 이은 값), 없으면 연도 순서로 이어 셈. + """ year: int points: list[tuple[float, float]] crs_epsg: int + start_station_m: float | None = None + end_station_m: float | None = None @dataclass @@ -115,7 +124,7 @@ def _write_shapefile( 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(".prj").write_text(_YONGHWA_PRJ, encoding="ascii") out.with_suffix(".cpg").write_text("949", encoding="ascii") return {suffix: out.with_suffix(suffix).read_bytes() for suffix in SHP_SUFFIXES} @@ -191,16 +200,20 @@ def build_export( if len(points) < 2: raise ValueError(f"{item.year}년 노선 점이 2개 미만입니다.") length_m = polyline_length_m(points) + if item.start_station_m is not None and item.end_station_m is not None: + start, end = float(item.start_station_m), float(item.end_station_m) + else: + start, end = station, station + length_m rows.append( { "year": item.year, "points": points, "length_m": length_m, - "start_station": station, - "end_station": station + length_m, + "start_station": start, + "end_station": end, } ) - station += length_m + station = end bundle = ExportBundle() region_name = safe_name(region) diff --git a/D04_PlanConfirm/D04_PlanConfirm_Repository.py b/D04_PlanConfirm/D04_PlanConfirm_Repository.py new file mode 100644 index 000000000..83912fb8b --- /dev/null +++ b/D04_PlanConfirm/D04_PlanConfirm_Repository.py @@ -0,0 +1,119 @@ +"""D04 확정 · 업체 지정 DB (Raw SQL · 표는 db_management/020_project_groups.sql).""" + +from __future__ import annotations + +import json +from typing import Any + +import aiomysql + +from config.config_db import get_db_pool + + +async def get_plan_project(project_id: str) -> dict[str, Any] | None: + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT id, name, region, project_kind, route_start_m FROM projects" + " WHERE id = %s AND deleted_at IS NULL", + (project_id,), + ) + return await cursor.fetchone() + + +async def list_year_routes(project_id: str) -> list[dict[str, Any]]: + """연도 구간(오름차순) — `route_json` 은 파싱해서 돌려줌(없으면 None).""" + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT year, epsg, route_json, fingerprint, confirmed_at FROM plan_route_years" + " WHERE project_id = %s ORDER BY year", + (project_id,), + ) + rows = await cursor.fetchall() + for row in rows: + raw = row["route_json"] + row["route_json"] = json.loads(raw) if isinstance(raw, str | bytes) else raw + return rows + + +async def list_assignees(project_id: str) -> list[dict[str, Any]]: + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT id, year, company_id, company_name, email, notified_at, linked_project_id" + " FROM plan_route_assignees WHERE project_id = %s ORDER BY year, id", + (project_id,), + ) + return await cursor.fetchall() + + +async def get_company_contact(company_id: int) -> dict[str, Any] | None: + """등록 회사 이름 + 대표 메일(회사 마스터 사용자).""" + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT c.id, c.name, u.email FROM companies c" + " LEFT JOIN users u ON u.id = c.master_user_id AND u.deleted_at IS NULL" + " WHERE c.id = %s AND c.deleted_at IS NULL", + (company_id,), + ) + return await cursor.fetchone() + + +async def apply_assignee_changes( + project_id: str, delete_ids: list[int], updates: list[dict], inserts: list[dict] +) -> None: + async with get_db_pool().acquire() as connection, connection.cursor() as cursor: + await connection.begin() + try: + for assignee_id in delete_ids: + await cursor.execute( + "DELETE FROM plan_route_assignees WHERE id = %s AND project_id = %s", + (assignee_id, project_id), + ) + for row in updates: + await cursor.execute( + "UPDATE plan_route_assignees SET company_name = %s, email = %s" + " WHERE id = %s AND project_id = %s", + (row["company_name"], row["email"], row["id"], project_id), + ) + for row in inserts: + await cursor.execute( + "INSERT INTO plan_route_assignees" + " (project_id, year, company_id, company_name, email) VALUES (%s,%s,%s,%s,%s)", + (project_id, row["year"], row["company_id"], row["company_name"], row["email"]), + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + + +async def mark_confirmed(project_id: str, years: list[int]) -> None: + if not years: + return + marks = ",".join(["%s"] * len(years)) + async with get_db_pool().acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + f"UPDATE plan_route_years SET confirmed_at = NOW() WHERE project_id = %s" + f" AND year IN ({marks})", + (project_id, *years), + ) + await connection.commit() + + +async def mark_notified(assignee_id: int) -> None: + async with get_db_pool().acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + "UPDATE plan_route_assignees SET notified_at = NOW() WHERE id = %s", (assignee_id,) + ) + await connection.commit() diff --git a/D04_PlanConfirm/D04_PlanConfirm_Router.py b/D04_PlanConfirm/D04_PlanConfirm_Router.py new file mode 100644 index 000000000..037101d64 --- /dev/null +++ b/D04_PlanConfirm/D04_PlanConfirm_Router.py @@ -0,0 +1,214 @@ +"""D 예정노선 확정 · 업체 지정 API (요구 345 · PLAN 531-1 · 531-2). + +- `GET /api/projects/{id}/plan/assignees` — 연도별 지정 업체 +- `PUT /api/projects/{id}/plan/assignees` — 연도별 지정 업체 저장(등록 회사 또는 업체명 + 메일) +- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 zip → 메일 첨부 → 발송 시각 + +지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이 +GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from common_util.common_util_auth import is_forest_service, verify_session +from D04_PlanConfirm import D04_PlanConfirm_Repository as repo +from D04_PlanConfirm.D04_PlanConfirm_Email import build_completion_mail, send_completion_mail +from D04_PlanConfirm.D04_PlanConfirm_Export import ( + MIME_ZIP, + YearRoute, + build_export, + company_zip, +) + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"]) + + +class AssigneeIn(BaseModel): + year: int + company_id: int | None = None # 등록 회사 — 없으면 수기 업체 + company_name: str | None = None + email: str | None = None + + +class AssigneesRequest(BaseModel): + assignees: list[AssigneeIn] + + +class FinalizeRequest(BaseModel): + years: list[int] | None = None # 비우면 분석이 저장된 모든 연도 + resend: bool = False # 이미 메일을 보낸 업체에도 다시 보냄 + + +def _require_forest(session: dict[str, Any]) -> None: + if not is_forest_service(session): + raise HTTPException( + status_code=403, detail="산림청만 예정노선을 확정 · 지정할 수 있습니다." + ) + + +def _key(row: dict[str, Any]) -> tuple: + """같은 업체 판정 — 등록 회사는 id · 수기 업체는 메일(대소문자 무시).""" + if row.get("company_id") is not None: + return (row["year"], "c", row["company_id"]) + return (row["year"], "e", (row.get("email") or "").strip().lower()) + + +def plan_assignee_changes( + existing: list[dict[str, Any]], wanted: list[dict[str, Any]], years: set[int] +) -> tuple[list[int], list[dict], list[dict]]: + """보낸 연도(`years`)만 맞춰 바꿈 — (지울 id, 고칠 행, 넣을 행). + + 같은 업체는 행을 살려 두어 발송 시각 · 연결한 프로젝트가 남는다. + """ + old = {_key(row): row for row in existing if row["year"] in years} + new = {_key(row): row for row in wanted} + delete_ids = [row["id"] for key, row in old.items() if key not in new] + updates = [{**row, "id": old[key]["id"]} for key, row in new.items() if key in old] + inserts = [row for key, row in new.items() if key not in old] + return delete_ids, updates, inserts + + +async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]: + """등록 회사면 이름 · 대표 메일을 DB 에서, 수기면 이름 + 메일을 검사.""" + email = (item.email or "").strip() + name = (item.company_name or "").strip() + if item.company_id is not None: + company = await repo.get_company_contact(item.company_id) + if not company: + raise HTTPException( + status_code=400, detail=f"등록 회사를 찾을 수 없습니다: {item.company_id}" + ) + return { + "year": item.year, + "company_id": item.company_id, + "company_name": company["name"], + "email": email or company.get("email") or None, + } + if not name or "@" not in email: + raise HTTPException(status_code=400, detail="업체명과 메일 주소가 필요합니다.") + return {"year": item.year, "company_id": None, "company_name": name, "email": email} + + +async def _plan_project(project_id: UUID) -> dict[str, Any]: + project = await repo.get_plan_project(str(project_id)) + if not project: + raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.") + if project["project_kind"] != "PLAN": + raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.") + return project + + +@router.get("/{project_id}/plan/assignees", response_model=None) +async def get_plan_assignees(project_id: UUID) -> dict[str, Any]: + await _plan_project(project_id) + return {"status": "success", "assignees": await repo.list_assignees(str(project_id))} + + +@router.put("/{project_id}/plan/assignees", response_model=None) +async def put_plan_assignees( + project_id: UUID, request: AssigneesRequest, session: dict = Depends(verify_session) +) -> dict[str, Any]: + """보낸 연도의 지정 업체를 그대로 맞춤 — 연도를 안 보낸 곳은 건드리지 않음.""" + _require_forest(session) + await _plan_project(project_id) + known = {row["year"] for row in await repo.list_year_routes(str(project_id))} + unknown = sorted({a.year for a in request.assignees} - known) + if unknown: + raise HTTPException(status_code=400, detail=f"예정노선 연도가 아닙니다: {unknown}") + wanted = [await _resolve_assignee(item) for item in request.assignees] + years = {a.year for a in request.assignees} + existing = await repo.list_assignees(str(project_id)) + delete_ids, updates, inserts = plan_assignee_changes(existing, wanted, years) + await repo.apply_assignee_changes(str(project_id), delete_ids, updates, inserts) + return {"status": "success", "assignees": await repo.list_assignees(str(project_id))} + + +def year_routes_from_rows(rows: list[dict[str, Any]]) -> list[YearRoute]: + """분석이 저장된 연도 → 내보내기 입력 (D03 `route_json["planned"]` = 프로젝트 좌표 꺾은선).""" + routes = [] + for row in rows: + body = row.get("route_json") + if not body or not body.get("planned"): + continue + routes.append( + YearRoute( + year=int(row["year"]), + points=[(p[0], p[1]) for p in body["planned"]], + crs_epsg=int(row["epsg"]), + start_station_m=body.get("start_chainage_m"), + end_station_m=body.get("end_chainage_m"), + ) + ) + return routes + + +def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]: + return { + "year": assignee["year"], + "company_name": assignee["company_name"], + "email": assignee["email"], + "state": state, + } + + +@router.post("/{project_id}/plan/finalize", response_model=None) +async def finalize_plan( + project_id: UUID, request: FinalizeRequest, session: dict = Depends(verify_session) +) -> dict[str, Any]: + """연도 노선 확정 → 지정 업체마다 그 연도 zip 을 첨부해 메일 → 보낸 곳에 발송 시각.""" + _require_forest(session) + project = await _plan_project(project_id) + routes = year_routes_from_rows(await repo.list_year_routes(str(project_id))) + saved = {r.year for r in routes} + years = sorted(set(request.years) if request.years else saved) + missing = [y for y in years if y not in saved] + if not years or missing: + raise HTTPException( + status_code=409, detail=f"분석 [저장]이 안 된 연도입니다: {missing or '없음'}" + ) + await repo.mark_confirmed(str(project_id), years) + + # 누가거리가 연도마다 이어지므로 저장된 전 연도로 한 번에 만들고, 보낼 때만 연도를 고름. + first = float(project.get("route_start_m") or 0.0) + bundle = await asyncio.to_thread( + build_export, project["name"], project["region"] or "", routes, first + ) + results = [] + for assignee in await repo.list_assignees(str(project_id)): + year = assignee["year"] + if year not in years: + continue + if assignee["notified_at"] and not request.resend: + results.append(_result(assignee, "already_sent")) + continue + if not assignee["email"]: + results.append(_result(assignee, "no_email")) + continue + zip_name, zip_bytes = company_zip(project["name"], year, bundle.by_year[year]) + subject, body = build_completion_mail( + project_name=project["name"], + region=project["region"], + year=year, + length_m=bundle.lengths_m[year], + recipient=assignee["company_name"] or assignee["email"], + attachment_names=[zip_name], + ) + sent = await send_completion_mail( + to_email=assignee["email"], + subject=subject, + body=body, + attachments=[(zip_name, zip_bytes, MIME_ZIP)], + ) + if sent: + await repo.mark_notified(assignee["id"]) + results.append(_result(assignee, "sent" if sent else "failed")) + return {"status": "success", "confirmed_years": years, "notifications": results} diff --git a/main.py b/main.py index b7b8c5dce..b9ea3974a 100644 --- a/main.py +++ b/main.py @@ -75,6 +75,7 @@ from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router +from D04_PlanConfirm.D04_PlanConfirm_Router import router as d04_plan_confirm_router from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router @@ -478,6 +479,7 @@ app.include_router(tiles_router, dependencies=protected_with_company) app.include_router(b05_route_router, dependencies=protected_with_company) app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company) app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) +app.include_router(d04_plan_confirm_router, dependencies=protected_with_company) app.include_router(b05_corridor_router, dependencies=protected_with_company) app.include_router(b05_route_replan_router, dependencies=protected_with_company) app.include_router(b05_route_terrain_router, dependencies=protected_with_company) diff --git a/resources/tester/test_531_d04_export_finalize.py b/resources/tester/test_531_d04_export_finalize.py new file mode 100644 index 000000000..ed191ff82 --- /dev/null +++ b/resources/tester/test_531_d04_export_finalize.py @@ -0,0 +1,411 @@ +# -*- coding: utf-8 -*- +"""531-1 · 531-2 D04 예정노선 내보내기 · 메일 첨부 · 업체 지정 · 확정. + +- 만든 shp 를 B03 입력 분석(`analyze_input_metadata`)이 읽음 · 용화 shp 와 칸 · .prj · 인코딩 같음 +- 메일 조립(SMTP 가짜) — 첨부 이름 · 크기 · 첨부 없는 기존 호출 그대로 +- finalize / assignees 권한(산림청 아님 403) · 보낸 업체 건너뜀 · 메일 없음 · 분석 안 저장 409 +""" + +from __future__ import annotations + +import asyncio +import io +import struct +import sys +import zipfile +from email import message_from_bytes +from pathlib import Path +from uuid import uuid4 + +import pytest +from fastapi import HTTPException +from openpyxl import load_workbook +from pyproj import Transformer + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import common_util.common_util_email as email_mod # noqa: E402 +from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Repository as repo # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Router as router # noqa: E402 +from D04_PlanConfirm.D04_PlanConfirm_Email import build_completion_mail # noqa: E402 +from D04_PlanConfirm.D04_PlanConfirm_Export import ( # noqa: E402 + YearRoute, + build_export, + company_zip, +) + +YONGHWA = ROOT / "resources/knowledge/original/실무문서/2026년 임도 용화/예정노선" +YONGHWA_BASE = YONGHWA / "대상지_일월.용화.2.2" # 이름에 점이 있어 with_suffix 못 씀 + +# 5186(중부원점) 좌표의 두 해 — 2026 끝 = 2027 시작 +Y2026 = [(210000.0, 380000.0), (210100.0, 380050.0), (210250.0, 380040.0)] +Y2027 = [(210250.0, 380040.0), (210400.0, 380120.0)] + + +def _yh(ext: str) -> Path: + return YONGHWA / f"대상지_일월.용화.2.2{ext}" + + +def _bundle(with_xlsx: bool = True): + return build_export( + "시험 프로젝트", + "일월.용화", + [YearRoute(2027, Y2027, 5186), YearRoute(2026, Y2026, 5186)], + first_station_m=100.0, + with_xlsx=with_xlsx, + ) + + +def _dbf_fields(blob: bytes) -> list[tuple]: + count = (struct.unpack(" Path: + for name, data in files.items(): + (tmp_path / name).write_bytes(data) + return next(tmp_path / n for n in files if n.endswith(".shp")) + + +# ── 내보내기 ───────────────────────────────────────────── + + +def test_year_shapefile_matches_yonghwa_format(tmp_path): + bundle = _bundle() + files = bundle.by_year[2026] + shp_name = next(n for n in files if n.endswith(".shp")) + stem = shp_name[: -len(".shp")] + assert stem.startswith("대상지_일월.용화.") and stem.endswith("_2026") + + # 칸: 용화 두 칸(이름 · 형 · 폭 · 소수) 그대로 + 덧칸 + fields = _dbf_fields(files[stem + ".dbf"]) + assert fields[:2] == _dbf_fields((_yh(".dbf")).read_bytes()) + assert fields[2:] == [("연도", "N", 4, 0), ("시점누가", "N", 11, 2), ("종점누가", "N", 11, 2)] + # .prj 원문 · .cpg · 도형 종류(폴리라인 3) + assert files[stem + ".prj"] == _yh(".prj").read_bytes() + assert files[stem + ".cpg"] == _yh(".cpg").read_bytes() == b"949" + assert struct.unpack("본문

    ", [("시험_예정노선_2026.zip", payload, "application/zip")] + ) + email_mod._send_sync("a@b.kr", "제목", "

    본문

    ") # 기존 호출 그대로 + with_file, plain = (message_from_bytes(m.as_bytes()) for m in _FakeSmtp.sent) + + parts = [p for p in with_file.walk() if p.get_filename()] + assert [p.get_filename() for p in parts] == ["시험_예정노선_2026.zip"] + assert parts[0].get_content_type() == "application/zip" + assert parts[0].get_payload(decode=True) == payload + assert not [p for p in plain.walk() if p.get_filename()] + assert plain.get_content_type() == "multipart/alternative" + + +def test_send_email_disabled_ok_with_attachments(monkeypatch): + monkeypatch.setattr(email_mod, "EMAIL_NOTIFICATION_ENABLED", False) + assert asyncio.run(email_mod.send_email("a@b.kr", "s", "

    h

    ")) is True + assert asyncio.run( + email_mod.send_email("a@b.kr", "s", "

    h

    ", [("a.zip", b"1", "application/zip")]) + ) + + +def test_completion_mail_body(): + subject, body = build_completion_mail( + project_name="용화 <예정>", + region="일월.용화", + year=2026, + length_m=2135.9, + recipient="가나다건설", + attachment_names=["용화_예정노선_2026.zip"], + ) + assert "용화" in subject and "2026" in subject + assert "2026년" in body and "2.136 km" in body and "용화_예정노선_2026.zip" in body + assert "<예정>" in body # 이스케이프 + + +# ── API ────────────────────────────────────────────────── + +FOREST = {"user_id": 1, "email": "a@korea.go.kr", "role": "MEMBER", "company_id": 7} +OUTSIDER = {"user_id": 2, "email": "b@company.co.kr", "role": "MASTER", "company_id": 8} + + +def _route_row(year: int, points, start: float, end: float): + return { + "year": year, + "epsg": 5186, + "route_json": { + "planned": [list(p) + [0.0] for p in points], + "start_chainage_m": start, + "end_chainage_m": end, + }, + "fingerprint": "f" * 40, + "confirmed_at": None, + } + + +@pytest.fixture +def fake_db(monkeypatch): + state = { + "years": [_route_row(2026, Y2026, 0.0, 330.0), _route_row(2027, Y2027, 330.0, 520.0)], + "assignees": [ + { + "id": 1, + "year": 2026, + "company_id": None, + "company_name": "가나다건설", + "email": "ga@x.kr", + "notified_at": None, + "linked_project_id": None, + }, + { + "id": 2, + "year": 2026, + "company_id": 9, + "company_name": "라마", + "email": "ra@x.kr", + "notified_at": "2026-10-01", + "linked_project_id": None, + }, + { + "id": 3, + "year": 2027, + "company_id": None, + "company_name": "메일없음", + "email": None, + "notified_at": None, + "linked_project_id": None, + }, + ], + "confirmed": [], + "notified": [], + "mails": [], + "applied": None, + } + + async def project(_pid): + return { + "id": "p", + "name": "시험", + "region": "일월.용화", + "project_kind": "PLAN", + "route_start_m": 0.0, + } + + async def years(_pid): + return state["years"] + + async def assignees(_pid): + return state["assignees"] + + async def confirmed(_pid, ys): + state["confirmed"] = ys + + async def notified(aid): + state["notified"].append(aid) + + async def apply(_pid, d, u, i): + state["applied"] = (d, u, i) + + async def contact(cid): + return {"id": cid, "name": "등록사", "email": "master@reg.kr"} if cid == 9 else None + + async def send(*, to_email, subject, body, attachments): + state["mails"].append((to_email, subject, attachments)) + return True + + monkeypatch.setattr(repo, "get_plan_project", project) + monkeypatch.setattr(repo, "list_year_routes", years) + monkeypatch.setattr(repo, "list_assignees", assignees) + monkeypatch.setattr(repo, "mark_confirmed", confirmed) + monkeypatch.setattr(repo, "mark_notified", notified) + monkeypatch.setattr(repo, "apply_assignee_changes", apply) + monkeypatch.setattr(repo, "get_company_contact", contact) + monkeypatch.setattr(router, "send_completion_mail", send) + return state + + +def test_finalize_requires_forest_service(fake_db): + with pytest.raises(HTTPException) as err: + asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(), OUTSIDER)) + assert err.value.status_code == 403 + with pytest.raises(HTTPException) as err: + asyncio.run( + router.put_plan_assignees(uuid4(), router.AssigneesRequest(assignees=[]), OUTSIDER) + ) + assert err.value.status_code == 403 + assert fake_db["confirmed"] == [] and fake_db["mails"] == [] + + +def test_finalize_sends_zip_per_assignee(fake_db): + out = asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(), FOREST)) + assert out["confirmed_years"] == [2026, 2027] + states = {(n["year"], n["company_name"]): n["state"] for n in out["notifications"]} + assert states == { + (2026, "가나다건설"): "sent", + (2026, "라마"): "already_sent", # 이미 보냄 → 건너뜀 + (2027, "메일없음"): "no_email", + } + assert fake_db["notified"] == [1] + ((to, subject, attachments),) = fake_db["mails"] + assert to == "ga@x.kr" and "2026" in subject + name, blob, mime = attachments[0] + assert name == "시험_예정노선_2026.zip" and mime == "application/zip" and len(blob) > 1000 + with zipfile.ZipFile(io.BytesIO(blob)) as archive: + assert any(n.endswith("_2026.shp") for n in archive.namelist()) + assert not any("_2027" in n for n in archive.namelist()) # 다른 연도는 안 샘 + + # 다시 보내기 + fake_db["mails"].clear() + again = asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(resend=True), FOREST)) + assert sum(n["state"] == "sent" for n in again["notifications"]) == 2 + + +def test_finalize_unsaved_year_409(fake_db): + fake_db["years"][1]["route_json"] = None + with pytest.raises(HTTPException) as err: + asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(years=[2027]), FOREST)) + assert err.value.status_code == 409 + assert fake_db["confirmed"] == [] + + +def test_put_assignees_validation_and_diff(fake_db): + def put(items): + return asyncio.run( + router.put_plan_assignees(uuid4(), router.AssigneesRequest(assignees=items), FOREST) + ) + + for bad in ( + router.AssigneeIn(year=2026, company_name="수기", email="메일아님"), + router.AssigneeIn(year=2026, email="x@y.kr"), + router.AssigneeIn(year=2030, company_name="없는해", email="x@y.kr"), + router.AssigneeIn(year=2026, company_id=99), + ): + with pytest.raises(HTTPException) as err: + put([bad]) + assert err.value.status_code == 400 + assert fake_db["applied"] is None + + # 2026: 가나다 유지(메일 바꿈) · 라마(등록사 9) 유지 · 새 수기 추가 / 2027 은 안 보냈으니 그대로 + put( + [ + router.AssigneeIn(year=2026, company_name="가나다건설", email="GA@x.kr"), + router.AssigneeIn(year=2026, company_id=9), + router.AssigneeIn(year=2026, company_name="새업체", email="new@x.kr"), + ] + ) + deleted, updated, inserted = fake_db["applied"] + assert deleted == [] # 2027 의 id 3 은 건드리지 않음 + assert {u["id"] for u in updated} == {1, 2} + assert [i["company_name"] for i in inserted] == ["새업체"] + assert next(u for u in updated if u["id"] == 2)["email"] == "master@reg.kr" + + put([router.AssigneeIn(year=2026, company_name="새업체", email="new@x.kr")]) + assert sorted(fake_db["applied"][0]) == [1, 2] # 빠진 업체 지움