feat(D04): 업체 지정 · 확정 API · 완료 메일 · 시험 (531-1 · 531-2)

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"<tr><td style='padding:4px 12px 4px 0;color:#6b7280'>{esc(k)}</td>"
+        f"<td style='padding:4px 0'><strong>{esc(v)}</strong></td></tr>"
+        for k, v in rows
+    )
+    files = "".join(f"<li>{esc(name)}</li>" for name in attachment_names)
+    body = f"""<!doctype html>
+<html lang="ko"><head><meta charset="utf-8"></head>
+<body style="font-family:'Segoe UI',Arial,sans-serif;color:#111827;line-height:1.6">
+<p>안녕하세요, <strong>{esc(recipient)}</strong> 담당자님.</p>
+<p>예정노선이 확정되어 {year}년 노선 파일을 보내 드립니다.</p>
+<table>{table}</table>
+<p>첨부 파일</p>
+<ul>{files}</ul>
+<p>압축을 풀면 shapefile(.shp · .shx · .dbf · .prj · .cpg · 좌표계 UTM-K)이 들어 있으며,
+임도 설계 프로그램의 노선 입력에 그대로 올릴 수 있습니다.</p>
+</body></html>"""
+    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)

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)

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()

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}

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)

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("<H", blob[8:10])[0] - 33) // 32
+    out = []
+    for i in range(count):
+        d = blob[32 + i * 32 : 64 + i * 32]
+        out.append((d[:11].rstrip(b"\0").decode("cp949"), chr(d[11]), d[16], d[17]))
+    return out
+
+
+def _write(tmp_path: Path, files: dict[str, bytes]) -> 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("<i", files[stem + ".shp"][32:36])[0] == 3
+
+
+def test_b03_analysis_accepts_exported_shapefile(tmp_path):
+    bundle = _bundle()
+    shp = _write(tmp_path, bundle.by_year[2026])
+    meta = analyze_input_metadata(shp)
+    ref = analyze_input_metadata(_yh(".shp"))
+
+    assert meta["shape_type"] == ref["shape_type"]
+    assert meta["part_count"] == 1 and meta["point_count"] == len(Y2026)
+    assert meta["encoding"] == ref["encoding"] == "cp949"
+    assert meta["crs_input"] == ref["crs_input"]  # .prj 원문 같음 → 같은 좌표계
+    assert meta["missing_members"] == []
+    assert meta["route_name"] == meta["attributes"]["대상지"]
+    assert meta["route_name"].startswith("일월.용화.")
+    # 5186 → 5179 로 바뀐 시점 좌표
+    sx, sy = Transformer.from_crs(5186, 5179, always_xy=True).transform(*Y2026[0])
+    assert meta["start_point"][:2] == pytest.approx([sx, sy], abs=1e-6)
+    assert float(meta["attributes"]["연도"]) == 2026
+    assert float(meta["attributes"]["시점누가"]) == pytest.approx(100.0)
+
+
+def test_stations_chain_and_total(tmp_path):
+    bundle = _bundle()
+    l26, l27 = bundle.lengths_m[2026], bundle.lengths_m[2027]
+    shp27 = _write(tmp_path, bundle.by_year[2027])
+    attrs = analyze_input_metadata(shp27)["attributes"]
+    assert float(attrs["시점누가"]) == pytest.approx(100.0 + l26, abs=0.01)  # 직전 종점 = 시점
+    assert float(attrs["종점누가"]) == pytest.approx(100.0 + l26 + l27, abs=0.01)
+
+    total_tmp = tmp_path / "all"
+    total_tmp.mkdir()
+    meta = analyze_input_metadata(_write(total_tmp, bundle.all))
+    assert meta["point_count"] == len(Y2026) + len(Y2027) - 1  # 겹친 경계점 하나 뺌
+    assert bundle.lengths_m[0] == pytest.approx(l26 + l27, abs=0.01)
+    assert meta["attributes"].get("연도") in (
+        None,
+        "",
+        "****",
+    )  # 전체는 연도 칸 비움(dBase 널 표기)
+
+
+def test_explicit_stations_win_and_short_route_rejected():
+    routes = [YearRoute(2026, Y2026, 5186, start_station_m=500.0, end_station_m=900.0)]
+    rows = build_export("p", "r", routes, with_xlsx=False)
+    meta = {}
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        meta = analyze_input_metadata(_write(Path(tmp), rows.by_year[2026]))
+    assert float(meta["attributes"]["시점누가"]) == 500.0
+    assert float(meta["attributes"]["종점누가"]) == 900.0
+    with pytest.raises(ValueError):
+        build_export("p", "r", [YearRoute(2026, [(0.0, 0.0)], 5179)])
+
+
+def test_zip_and_coordinate_table():
+    bundle = _bundle()
+    name, blob = company_zip("시험 프로젝트", 2026, bundle.by_year[2026])
+    assert name == "시험 프로젝트_예정노선_2026.zip"
+    with zipfile.ZipFile(io.BytesIO(blob)) as archive:
+        names = archive.namelist()
+    assert sum(n.endswith(s) for n in names for s in (".shp", ".shx", ".dbf", ".prj", ".cpg")) == 5
+    assert "좌표표_시험 프로젝트.xlsx" in names
+
+    sheet = load_workbook(io.BytesIO(bundle.all["좌표표_시험 프로젝트.xlsx"])).active
+    rows = list(sheet.iter_rows(min_row=3, values_only=True))
+    assert [r[0] for r in rows] == [2026, 2027]
+    # 업체용 좌표표는 그 연도만
+    one = load_workbook(io.BytesIO(bundle.by_year[2026]["좌표표_시험 프로젝트.xlsx"])).active
+    assert [r[0] for r in one.iter_rows(min_row=3, values_only=True)] == [2026]
+
+
+# ── 메일 ─────────────────────────────────────────────────
+
+
+class _FakeSmtp:
+    sent: list = []
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def starttls(self):
+        pass
+
+    def login(self, *args):
+        pass
+
+    def send_message(self, message):
+        _FakeSmtp.sent.append(message)
+
+
+def test_mail_attachment_assembly_and_old_calls(monkeypatch):
+    monkeypatch.setattr(email_mod.smtplib, "SMTP", _FakeSmtp)
+    _FakeSmtp.sent = []
+    payload = b"PK\x03\x04" + b"x" * 5000
+
+    email_mod._send_sync(
+        "a@b.kr", "제목", "<p>본문</p>", [("시험_예정노선_2026.zip", payload, "application/zip")]
+    )
+    email_mod._send_sync("a@b.kr", "제목", "<p>본문</p>")  # 기존 호출 그대로
+    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", "<p>h</p>")) is True
+    assert asyncio.run(
+        email_mod.send_email("a@b.kr", "s", "<p>h</p>", [("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 "&lt;예정&gt;" 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]  # 빠진 업체 지움