diff --git a/D01_PlanRegister/D01_PlanRegister_Router.py b/D01_PlanRegister/D01_PlanRegister_Router.py index 9919509ec..031072192 100644 --- a/D01_PlanRegister/D01_PlanRegister_Router.py +++ b/D01_PlanRegister/D01_PlanRegister_Router.py @@ -4,6 +4,7 @@ = 설계속도 작업임도 20 · 측점 없음(DB 칸 그대로). """ +import json from typing import Any import aiomysql @@ -72,8 +73,11 @@ async def post_plan_project( async def assigned_plan_routes( session: dict[str, Any] = Depends(require_company), ) -> dict[str, Any]: - """우리 회사에 지정된 예정노선 연도 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트.""" + """우리 회사에 지정된 예정노선 조각(옛 = 연도) — 확정된 것(또는 이미 연결한 것) · 연결 상태 · + 업데이트 · 조각 이름 · 길이(조각 행 · 옛 = 연도 행).""" + from D04_PlanConfirm import D04_PlanConfirm_Repository as plan_repo from D04_PlanConfirm.D04_PlanConfirm_Link import update_available + from D04_PlanConfirm.D04_PlanConfirm_Router import piece_label async with ( get_db_pool().acquire() as connection, @@ -81,23 +85,43 @@ async def assigned_plan_routes( ): await cursor.execute( """SELECT a.id AS assignee_id, a.project_id, p.name AS project_name, a.year, + a.piece_id, c.length_m AS piece_length_m, l.id AS linked_project_id, l.name AS linked_project_name, - l.source_fingerprint, y.route_json, y.confirmed_at + l.source_fingerprint, COALESCE(c.route_json, y.route_json) AS route_json, + y.confirmed_at FROM plan_route_assignees a JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year + LEFT JOIN plan_route_pieces c + ON c.project_id = a.project_id AND c.piece_id = a.piece_id LEFT JOIN projects l ON l.id = a.linked_project_id AND l.deleted_at IS NULL WHERE a.company_id = %s AND (y.confirmed_at IS NOT NULL OR l.id IS NOT NULL) - ORDER BY p.name, a.year""", + ORDER BY p.name, a.year, a.piece_id""", (session.get("company_id"),), ) rows = await cursor.fetchall() + pieces: dict[tuple[str, str], dict[str, Any]] = {} + for project_id in {r["project_id"] for r in rows if r.get("piece_id")}: + for piece in await plan_repo.list_pieces(str(project_id)): + pieces[(str(project_id), piece["piece_id"])] = piece routes = [] for row in rows: route_json, fingerprint = row.pop("route_json"), row.pop("source_fingerprint") + piece_length = row.pop("piece_length_m", None) confirmed = row["confirmed_at"] is not None + body = json.loads(route_json) if isinstance(route_json, str | bytes) else route_json + piece = pieces.get((str(row["project_id"]), row.get("piece_id"))) routes.append( - {**row, "update_available": update_available(route_json, fingerprint, confirmed)} + { + **row, + "piece_id": row.get("piece_id"), + "piece_label": piece + and piece_label(piece["line_name"], piece["from_m"], piece["to_m"]), + "length_m": piece_length + if piece_length is not None + else (body or {}).get("length_m"), + "update_available": update_available(route_json, fingerprint, confirmed), + } ) return {"status": "success", "routes": routes} diff --git a/D04_PlanConfirm/D04_PlanConfirm_Export.py b/D04_PlanConfirm/D04_PlanConfirm_Export.py index 633d98630..df82e4742 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Export.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Export.py @@ -1,4 +1,5 @@ -"""D04 예정노선 내보내기 — 연도별 노선(꺾은선)을 용화 예정노선 양식 shapefile 한 벌로 만든다. +"""D04 예정노선 내보내기 — 조각(구역 × 갈래)마다 · 옛 프로젝트는 연도마다 노선(꺾은선)을 +용화 예정노선 양식 shapefile 한 벌로 만든다. 양식 = `resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.*` (분석 345-C): 2D 폴리라인 1레코드 · UTM-K(EPSG 5179) · cp949 · 칸 `대상지` C(20) · @@ -70,9 +71,10 @@ def station_label(chainage_m: float, interval_m: float = STATION_INTERVAL_M) -> @dataclass class YearRoute: - """한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점). + """한 조각(또는 옛 연도)의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점). - 누가거리 두 값이 있으면 그대로 칸에 쓰고(D03 이은 값), 없으면 연도 순서로 이어 셈. + 누가거리 두 값이 있으면 그대로 칸에 쓰고(조각 = 선마다 No.0), 없으면 연도 순서로 이어 셈. + `piece` = 조각 id(옛 연도 노선은 None) · `line` = 선 이름(파일 이름 · 전체 묶음 열쇠). """ year: int @@ -80,11 +82,18 @@ class YearRoute: crs_epsg: int start_station_m: float | None = None end_station_m: float | None = None + piece: str | None = None + line: str | None = None + + @property + def key(self) -> str | int: + return self.piece or self.year @dataclass class ExportBundle: - """`by_year[연도]` = 그 연도 shp 한 벌 + 좌표표 · `all` = 전 연도 이은 한 벌 + 좌표표.""" + """`by_year[열쇠]` = 그 조각(열쇠 = 조각 id · 옛은 연도) shp 한 벌 + 좌표표 · + `all` = 선마다 이은 레코드 한 벌 + 좌표표.""" by_year: dict[int, dict[str, bytes]] = field(default_factory=dict) all: dict[str, bytes] = field(default_factory=dict) @@ -120,20 +129,22 @@ 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) +Record = tuple[list[tuple[float, float]], int | None, tuple[float, float]] + + +def _write_shapefile(records: list[Record], region: str) -> dict[str, bytes]: + """5179 꺾은선 레코드(점 · 연도 · 누가 두 값) → `.shp/.shx/.dbf/.prj/.cpg` 바이트 + (파일명은 호출부가 붙임: 키 = 확장자).""" + lengths = [polyline_length_m(points) for points, _, _ in records] frame = gpd.GeoDataFrame( { - "대상지": [_clip_bytes(_target_text(region, length_m))], - "길이": [round(length_m / 1000, 3)], - "연도": [year], - "시점누가": [round(stations[0], 2)], - "종점누가": [round(stations[1], 2)], + "대상지": [_clip_bytes(_target_text(region, m)) for m in lengths], + "길이": [round(m / 1000, 3) for m in lengths], + "연도": [year for _, year, _ in records], + "시점누가": [round(stations[0], 2) for _, _, stations in records], + "종점누가": [round(stations[1], 2) for _, _, stations in records], }, - geometry=[LineString(points)], + geometry=[LineString(points) for points, _, _ in records], crs=EXPORT_EPSG, ) with tempfile.TemporaryDirectory() as tmp: @@ -169,6 +180,8 @@ def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes: "시점누가(m)", "종점누가(m)", "연장(m)", + "선", + "조각", ] ) for row in rows: @@ -190,6 +203,8 @@ def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes: round(row["start_station"], 2), round(row["end_station"], 2), round(row["length_m"], 2), + row["line"], + row["piece"], ] ) buffer = io.BytesIO() @@ -208,10 +223,10 @@ def build_export( first_station_m: float = 0.0, with_xlsx: bool = True, ) -> ExportBundle: - """연도 노선들 → 연도별 한 벌 + 전체 한 벌(+ 좌표표). + """조각(옛 = 연도) 노선들 → 조각마다 한 벌 + 전체 한 벌(+ 좌표표). - 누가거리 = `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 = 다음 연도 시점). - shp 안 측점은 파일마다 0 부터가 아니라 이 누가값을 칸으로만 보관(B03 은 칸을 무시). + 누가거리 = 값이 없으면 `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 = + 다음 연도 시점). shp 안 측점은 칸으로만 보관(B03 은 칸을 무시). """ ordered = sorted(years, key=lambda y: y.year) rows: list[dict] = [] @@ -232,6 +247,9 @@ def build_export( "length_m": length_m, "start_station": start, "end_station": end, + "key": item.key, + "piece": item.piece, + "line": item.line, } ) station = end @@ -241,39 +259,40 @@ def build_export( 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 row["piece"]: + base += f"_{safe_name(row['line'] or '')}_{safe_name(row['piece'])}" + stations = (row["start_station"], row["end_station"]) + files = _named(base, _write_shapefile([(row["points"], row["year"], stations)], 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"] + bundle.by_year[row["key"]] = files + bundle.lengths_m[row["key"]] = row["length_m"] - # 전체: 앞 연도 종점 = 다음 연도 시점이라 겹치는 점 하나는 뺌. - merged: list[tuple[float, float]] = [] + # 전체: 선마다 한 레코드 · 앞 조각 종점 = 다음 조각 시점이라 겹치는 점 하나는 뺌. + by_line: dict[str | None, list[dict]] = {} 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) + by_line.setdefault(row["line"], []).append(row) + records: list[Record] = [] + for parts in by_line.values(): + parts.sort(key=lambda r: r["start_station"]) + merged: list[tuple[float, float]] = [] + for row in parts: + same = merged and math.dist(merged[-1], row["points"][0]) < 1e-6 + merged.extend(row["points"][1:] if same else row["points"]) + records.append((merged, None, (parts[0]["start_station"], parts[-1]["end_station"]))) + total_m = sum(polyline_length_m(points) for points, _, _ in records) 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), - ) + bundle.all = _named(base, _write_shapefile(records, 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` (이름, 바이트).""" +def company_zip(project_name: str, label: str | 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() + return f"{safe_name(project_name)}_예정노선_{safe_name(str(label))}.zip", buffer.getvalue() diff --git a/D04_PlanConfirm/D04_PlanConfirm_Link.py b/D04_PlanConfirm/D04_PlanConfirm_Link.py index e6ae64914..21084eca6 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Link.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Link.py @@ -4,9 +4,9 @@ new = 임도 프로젝트 새로(B02 저장소 함수) · existing = 고른 우리 회사 임도 프로젝트 - `GET /api/projects/{id}/plan-source` — 임도 프로젝트가 이어 받은 예정노선 · 업데이트 있음 -연결 = 그 연도 shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에 원래 이름으로 + -`input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길. 입력이 다 차면 -B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림. +연결 = 지정 행의 조각(옛 = 연도) shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에 +원래 이름으로 + `input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길. +입력이 다 차면 B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림. 연결 시점 복사 · 원본이 바뀌면 알림만(내용 지문 비교 · 자동 반영 없음). """ @@ -50,8 +50,7 @@ from common_util.common_util_auth import require_company from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_workflow_state import is_analysis_running, merge_stage_params from config.config_db import get_db_pool -from D04_PlanConfirm.D04_PlanConfirm_Export import build_export -from D04_PlanConfirm.D04_PlanConfirm_Router import year_routes_from_rows +from D04_PlanConfirm.D04_PlanConfirm_Export import YearRoute, build_export router = APIRouter(prefix="/api/projects", tags=["D04 Plan Link"]) dashboard_router = APIRouter(prefix="/api/dashboard", tags=["D04 Plan Link"]) @@ -102,18 +101,53 @@ async def _fetch_one(sql: str, args: tuple) -> dict[str, Any] | None: return await cursor.fetchone() +# 지정 행의 노선 = 조각 행(조각 지정) · 연도 행(옛 연도 지정) — 확정은 연도 행 +PIECE_JOIN = ( + " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year" + " LEFT JOIN plan_route_pieces c ON c.project_id = a.project_id AND c.piece_id = a.piece_id" +) +PIECE_COLUMNS = ( + " a.piece_id, c.epsg AS piece_epsg, c.route_json AS piece_json," + " c.from_m, c.to_m, c.length_m AS piece_length_m," +) + + +def unit_row(row: dict[str, Any]) -> dict[str, Any]: + """조각 지정이면 `epsg` · `route_json` 을 조각 것으로 갈아 끼움(누가 = 선 위 거리 · + 길이 = 조각) — 아래 지문 · shp · 새 임도 길이가 그대로 돎. 조각 칸은 뺌.""" + out = {k: v for k, v in row.items() if k not in ("piece_epsg", "piece_json", "from_m", "to_m")} + out.pop("piece_length_m", None) + if not row.get("piece_id"): + return out + body = _parsed(row.get("piece_json")) or {} + out["epsg"] = row.get("piece_epsg") + out["route_json"] = ( + { + "planned": body["planned"], + "start_chainage_m": row["from_m"], + "end_chainage_m": row["to_m"], + "length_m": row["piece_length_m"], + } + if body.get("planned") + else None + ) + return out + + async def load_assignee(assignee_id: int) -> dict[str, Any] | None: - """지정 행 + 예정노선 프로젝트 칸 + 그 연도 노선.""" - return await _fetch_one( + """지정 행 + 예정노선 프로젝트 칸 + 그 조각(옛 = 연도) 노선.""" + row = await _fetch_one( "SELECT a.id, a.project_id, a.year, a.company_id, a.linked_project_id," + f"{PIECE_COLUMNS}" " p.name, p.region, p.road_type, p.design_speed_kph, p.route_start_m," " y.epsg, y.route_json, y.confirmed_at" " FROM plan_route_assignees a" " JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL" - " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year" + f"{PIECE_JOIN}" " WHERE a.id = %s", (assignee_id,), ) + return unit_row(row) if row else None async def load_road_project(project_id: str) -> dict[str, Any] | None: @@ -152,11 +186,17 @@ async def record_link( def year_files(assignee: dict[str, Any]) -> dict[str, bytes]: - """그 연도 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트.""" - row = {**assignee, "route_json": _parsed(assignee["route_json"])} - routes = year_routes_from_rows([row]) - bundle = build_export(assignee["name"], assignee["region"] or "", routes, with_xlsx=False) - return bundle.by_year[int(assignee["year"])] + """그 조각(옛 = 연도) shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트.""" + body = _parsed(assignee["route_json"]) or {} + route = YearRoute( + year=int(assignee["year"]), + points=[(p[0], p[1]) for p in body["planned"]], + crs_epsg=int(assignee["epsg"]), + start_station_m=body.get("start_chainage_m"), + end_station_m=body.get("end_chainage_m"), + ) + bundle = build_export(assignee["name"], assignee["region"] or "", [route], with_xlsx=False) + return bundle.by_year[route.key] async def _supersede_route_set(connection: aiomysql.Connection, project_id: str) -> None: @@ -285,7 +325,7 @@ async def seed_design_speed(project_id: str, speed: Any) -> None: async def link_plan_route( assignee_id: int, payload: LinkRequest, session: dict[str, Any] = Depends(require_company) ) -> dict[str, Any] | JSONResponse: - """지정받은 연도 노선을 임도 프로젝트 B03 입력으로 — 지정 회사만.""" + """지정받은 조각(옛 = 연도) 노선을 임도 프로젝트 B03 입력으로 — 지정 회사만.""" assignee = await load_assignee(assignee_id) if not assignee: raise HTTPException(status_code=404, detail="지정 업체 행을 찾을 수 없습니다.") @@ -319,7 +359,13 @@ async def link_plan_route( raise HTTPException(status_code=400, detail=str(exc)) from exc fingerprint = route_fingerprint(assignee["route_json"]) await record_link(project_id, assignee, fingerprint, assignee["design_speed_kph"]) - return {"status": "success", "project_id": project_id, "year": assignee["year"], **result} + return { + "status": "success", + "project_id": project_id, + "year": assignee["year"], + "piece_id": assignee.get("piece_id"), + **result, + } @router.get("/{project_id}/plan-source", response_model=None) @@ -327,22 +373,25 @@ async def plan_source(project_id: UUID) -> dict[str, Any]: """임도 프로젝트가 이어 받은 예정노선(B03 머리 띠) — 없으면 linked False.""" row = await _fetch_one( "SELECT p.source_fingerprint, s.name AS plan_name, a.id AS assignee_id, a.year," - " y.route_json, y.confirmed_at" + f"{PIECE_COLUMNS}" + " y.epsg, y.route_json, y.confirmed_at" " FROM projects p" " JOIN projects s ON s.id = p.source_project_id AND s.project_kind = 'PLAN'" " JOIN plan_route_assignees a ON a.linked_project_id = p.id" - " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year" + f"{PIECE_JOIN}" " WHERE p.id = %s LIMIT 1", (str(project_id),), ) if not row: return {"status": "success", "linked": False} + row = unit_row(row) return { "status": "success", "linked": True, "assignee_id": row["assignee_id"], "plan_name": row["plan_name"], "year": row["year"], + "piece_id": row.get("piece_id"), "update_available": update_available( row["route_json"], row["source_fingerprint"], row["confirmed_at"] is not None ), diff --git a/D04_PlanConfirm/D04_PlanConfirm_Repository.py b/D04_PlanConfirm/D04_PlanConfirm_Repository.py index 83912fb8b..91b34d073 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Repository.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Repository.py @@ -41,13 +41,45 @@ async def list_year_routes(project_id: str) -> list[dict[str, Any]]: return rows +async def list_pieces(project_id: str) -> list[dict[str, Any]]: + """조각 행(연도 · 선 차례 · 선 위 거리 차례) + `line_name`(본선 · 갈래 N = 정본 선 차례).""" + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT piece_id, line_id, year, from_m, to_m, length_m, epsg, route_json" + " FROM plan_route_pieces WHERE project_id = %s", + (project_id,), + ) + rows = list(await cursor.fetchall()) + await cursor.execute( + "SELECT network_json FROM plan_networks WHERE project_id = %s", (project_id,) + ) + network = await cursor.fetchone() + lines = [] + if network: + raw = network["network_json"] + body = json.loads(raw) if isinstance(raw, str | bytes) else raw + lines = [line["id"] for line in (body or {}).get("lines", [])] + order = {line: i for i, line in enumerate(lines)} + for row in rows: + raw = row["route_json"] + row["route_json"] = json.loads(raw) if isinstance(raw, str | bytes) else raw + index = order.get(row["line_id"], len(order)) + row["line_name"] = "본선" if index == 0 else f"갈래 {index}" + rows.sort(key=lambda r: (r["year"], order.get(r["line_id"], len(order)), r["from_m"])) + 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" + "SELECT id, year, piece_id, company_id, company_name, email, notified_at," + " linked_project_id" " FROM plan_route_assignees WHERE project_id = %s ORDER BY year, id", (project_id,), ) @@ -89,8 +121,16 @@ async def apply_assignee_changes( 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"]), + " (project_id, year, piece_id, company_id, company_name, email)" + " VALUES (%s,%s,%s,%s,%s,%s)", + ( + project_id, + row["year"], + row.get("piece_id"), + row["company_id"], + row["company_name"], + row["email"], + ), ) await connection.commit() except Exception: diff --git a/D04_PlanConfirm/D04_PlanConfirm_Router.py b/D04_PlanConfirm/D04_PlanConfirm_Router.py index 41897e5af..8686e7261 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Router.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Router.py @@ -1,10 +1,14 @@ -"""D 예정노선 확정 · 업체 지정 API (요구 345 · PLAN 531-1 · 531-2). +"""D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · PLAN 531-1 · 531-2 · 563-2). -- `GET /api/projects/{id}/plan/assignees` — 연도별 지정 업체(지정 회사 = 자기 행만) -- `PUT /api/projects/{id}/plan/assignees` — 연도별 지정 업체 저장(등록 회사 또는 업체명 + 메일) -- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 zip → 메일 첨부 → 발송 시각 -- `GET /api/projects/{id}/plan/summary` — D04 화면 연도 요약(지정 회사 = 자기 연도 · 자기 행만) -- `GET /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 연도 + 전체 · 공유 = 자기 연도) +- `GET /api/projects/{id}/plan/assignees` — 조각별 지정 업체(지정 회사 = 자기 행만) +- `PUT /api/projects/{id}/plan/assignees` — 조각별 지정 업체 저장(등록 회사 또는 업체명 + 메일) +- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 조각 zip → 메일 첨부 → 발송 시각 +- `GET /api/projects/{id}/plan/summary` — D04 화면 연도 · 조각 요약(지정 회사 = 자기 조각 · 행만) +- `GET /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 조각 + 전체 · 공유 = 자기 조각) + +단위 = 조각(`plan_route_pieces` · 구역 × 갈래 · 요구 391) · 조각이 없는 연도 행(옛 프로젝트)은 +연도 하나가 단위. 확정은 연도 단위(`plan_route_years.confirmed_at`). 업체 행 `piece_id` 가 비면 +옛 연도 지정 — 그 연도 첫 단위에 붙음. 지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이 GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦. @@ -34,6 +38,7 @@ from D04_PlanConfirm.D04_PlanConfirm_Export import ( build_export, company_zip, safe_name, + station_label, ) logger = logging.getLogger(__name__) @@ -42,6 +47,7 @@ router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"]) class AssigneeIn(BaseModel): year: int + piece_id: str | None = None # 조각 — 비면 옛 연도 지정(조각 없는 프로젝트) company_id: int | None = None # 등록 회사 — 없으면 수기 업체 company_name: str | None = None email: str | None = None @@ -65,10 +71,11 @@ def _require_forest(session: dict[str, Any]) -> None: def _key(row: dict[str, Any]) -> tuple: - """같은 업체 판정 — 등록 회사는 id · 수기 업체는 메일(대소문자 무시).""" + """같은 업체 판정 — 같은 조각(옛 = 연도)에서 등록 회사 = id · 수기 = 메일(대소문자 무시).""" + unit = (row["year"], row.get("piece_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()) + return (*unit, "c", row["company_id"]) + return (*unit, "e", (row.get("email") or "").strip().lower()) def plan_assignee_changes( @@ -90,6 +97,7 @@ async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]: """등록 회사면 이름 · 대표 메일을 DB 에서, 수기면 이름 + 메일을 검사.""" email = (item.email or "").strip() name = (item.company_name or "").strip() + unit = {"year": item.year, "piece_id": item.piece_id} if item.company_id is not None: company = await repo.get_company_contact(item.company_id) if not company: @@ -97,14 +105,14 @@ async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]: status_code=400, detail=f"등록 회사를 찾을 수 없습니다: {item.company_id}" ) return { - "year": item.year, + **unit, "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} + return {**unit, "company_id": None, "company_name": name, "email": email} async def _plan_project(project_id: UUID) -> dict[str, Any]: @@ -116,16 +124,89 @@ async def _plan_project(project_id: UUID) -> dict[str, Any]: return project +def route_units(rows: list[dict[str, Any]], pieces: list[dict[str, Any]]) -> list[dict[str, Any]]: + """표 한 줄 단위(연도 차례) — 조각마다 · 조각이 없는 연도 행(옛 · 분석 전)은 연도 하나.""" + confirmed = {int(r["year"]): r.get("confirmed_at") for r in rows} + units = [ + { + "piece_id": p["piece_id"], + "line": p["line_id"], + "line_name": p.get("line_name"), + "year": int(p["year"]), + "epsg": int(p["epsg"]), + "planned": (p.get("route_json") or {}).get("planned") or [], + "start_chainage_m": float(p["from_m"]), + "end_chainage_m": float(p["to_m"]), + "length_m": float(p["length_m"]), + "confirmed_at": confirmed.get(int(p["year"])), + } + for p in pieces + ] + taken = {u["year"] for u in units} + for row in rows: + if int(row["year"]) in taken: + continue + body = row.get("route_json") or {} + units.append( + { + "piece_id": None, + "line": None, + "line_name": None, + "year": int(row["year"]), + "epsg": row.get("epsg"), + "planned": body.get("planned") or [], + "start_chainage_m": body.get("start_chainage_m"), + "end_chainage_m": body.get("end_chainage_m"), + "length_m": body.get("length_m"), + "confirmed_at": row.get("confirmed_at"), + } + ) + units.sort(key=lambda u: u["year"]) # 같은 연도 안 = 조각 차례 그대로 + return units + + +def unit_key(unit: dict[str, Any]) -> str | int: + """내보내기 묶음 열쇠 — 조각 id · 옛 연도.""" + return unit["piece_id"] or unit["year"] + + +def unit_label(unit: dict[str, Any]) -> str: + """zip 폴더 · 첨부 이름 — 「2026」(옛) · 「2026_갈래1_p3」(조각).""" + if not unit["piece_id"]: + return str(unit["year"]) + line = (unit["line_name"] or unit["line"]).replace(" ", "") + return f"{unit['year']}_{line}_{unit['piece_id']}" + + +def unit_of(assignee: dict[str, Any], units: list[dict[str, Any]]) -> dict[str, Any] | None: + """업체 행의 단위 — 조각 지정 = 그 조각 · 옛 연도 지정 = 그 연도 첫 단위.""" + piece = assignee.get("piece_id") + for unit in units: + if (unit["piece_id"] == piece) if piece else (unit["year"] == assignee["year"]): + return unit + return None + + +def _saved(unit: dict[str, Any]) -> bool: + return len(unit["planned"]) >= 2 + + def visible_to( - session: dict[str, Any], years: list[dict[str, Any]], assignees: list[dict[str, Any]] + session: dict[str, Any], units: list[dict[str, Any]], assignees: list[dict[str, Any]] ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: - """산림청 = 전부 · 지정 회사 = 자기 회사가 지정된 연도 줄 · 자기 업체 행만(남의 이름 · 메일 뺌).""" + """산림청 = 전부 · 지정 회사 = 자기 회사가 지정된 단위 · 자기 업체 행만(남의 이름 · 메일 뺌).""" if is_forest_service(session): - return years, assignees + return units, assignees company = session.get("company_id") mine = [a for a in assignees if a["company_id"] is not None and a["company_id"] == company] - own_years = {a["year"] for a in mine} - return [y for y in years if y["year"] in own_years], mine + own = [unit_of(a, units) for a in mine] + return [u for u in units if any(u is o for o in own)], mine + + +async def _units(project_id: UUID) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """(연도 행, 단위).""" + rows = await repo.list_year_routes(str(project_id)) + return rows, route_units(rows, await repo.list_pieces(str(project_id))) @router.get("/{project_id}/plan/assignees", response_model=None) @@ -141,10 +222,19 @@ async def get_plan_assignees( 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))} + piece_years = {p["piece_id"]: int(p["year"]) for p in await repo.list_pieces(str(project_id))} + for item in request.assignees: + if item.piece_id is None: + continue + if item.piece_id not in piece_years: + raise HTTPException(status_code=400, detail=f"조각이 없습니다: {item.piece_id}") + item.year = piece_years[item.piece_id] years = {a.year for a in request.assignees} | set(request.years or []) unknown = sorted(years - known) if unknown: @@ -156,28 +246,27 @@ async def put_plan_assignees( 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"), - ) +def unit_routes(units: list[dict[str, Any]]) -> list[YearRoute]: + """저장된 단위 → 내보내기 입력(`planned` = 프로젝트 좌표 꺾은선).""" + return [ + YearRoute( + year=unit["year"], + points=[(p[0], p[1]) for p in unit["planned"]], + crs_epsg=int(unit["epsg"]), + start_station_m=unit["start_chainage_m"], + end_station_m=unit["end_chainage_m"], + piece=unit["piece_id"], + line=(unit["line_name"] or unit["line"] or "").replace(" ", "") or None, ) - return routes + for unit in units + if _saved(unit) + ] def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]: return { "year": assignee["year"], + "piece_id": assignee.get("piece_id"), "company_name": assignee["company_name"], "email": assignee["email"], "state": state, @@ -188,10 +277,11 @@ def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]: async def finalize_plan( project_id: UUID, request: FinalizeRequest, session: dict = Depends(verify_session) ) -> dict[str, Any]: - """연도 노선 확정 → 지정 업체마다 그 연도 zip 을 첨부해 메일 → 보낸 곳에 발송 시각.""" + """연도 노선 확정 → 지정 업체마다 그 조각 zip 을 첨부해 메일 → 보낸 곳에 발송 시각.""" _require_forest(session) project = await _plan_project(project_id) - routes = year_routes_from_rows(await repo.list_year_routes(str(project_id))) + _, units = await _units(project_id) + routes = unit_routes(units) 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] @@ -201,7 +291,7 @@ async def finalize_plan( ) 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 @@ -217,12 +307,17 @@ async def finalize_plan( if not assignee["email"]: results.append(_result(assignee, "no_email")) continue - zip_name, zip_bytes = company_zip(project["name"], year, bundle.by_year[year]) + unit = unit_of(assignee, units) + if unit is None or unit_key(unit) not in bundle.by_year: + results.append(_result(assignee, "no_route")) + continue + key = unit_key(unit) + zip_name, zip_bytes = company_zip(project["name"], unit_label(unit), bundle.by_year[key]) subject, body = build_completion_mail( project_name=project["name"], region=project["region"], year=year, - length_m=bundle.lengths_m[year], + length_m=bundle.lengths_m[key], recipient=assignee["company_name"] or assignee["email"], attachment_names=[zip_name], ) @@ -263,25 +358,58 @@ def year_summary(row: dict[str, Any]) -> dict[str, Any]: return out +def unit_summary(unit: dict[str, Any]) -> dict[str, Any]: + """표 한 줄(조각 · 옛 연도) — 저장 전이면 saved False · label 은 조각만(「본선 0+0~12+10」).""" + out = {key: unit[key] for key in ("piece_id", "line", "line_name", "year", "length_m")} + out.update(saved=_saved(unit), confirmed_at=None, label=None) + if unit["confirmed_at"]: + out["confirmed_at"] = str(unit["confirmed_at"]) + if out["saved"]: + epsg = int(unit["epsg"]) + out.update( + start_chainage_m=unit["start_chainage_m"], + end_chainage_m=unit["end_chainage_m"], + start=_end_point(unit["planned"][0], epsg), + end=_end_point(unit["planned"][-1], epsg), + ) + if unit["piece_id"]: + out["label"] = piece_label( + unit["line_name"] or unit["line"], unit["start_chainage_m"], unit["end_chainage_m"] + ) + return out + + +def piece_label(line_name: str, from_m: float, to_m: float) -> str: + """조각 이름 「본선 0+0~12+10」(측점 = 선마다 No.0).""" + return f"{line_name} {station_label(from_m)}~{station_label(to_m)}" + + @router.get("/{project_id}/plan/summary", response_model=None) async def get_plan_summary( project_id: UUID, session: dict = Depends(verify_session) ) -> dict[str, Any]: + """`years` = 연도 행(확정 고르기) · `pieces` = 표 한 줄 단위(조각 · 옛 연도).""" project = await _plan_project(project_id) - rows = await repo.list_year_routes(str(project_id)) - years, assignees = visible_to( - session, [year_summary(row) for row in rows], await repo.list_assignees(str(project_id)) - ) - return {"status": "success", "name": project["name"], "years": years, "assignees": assignees} + rows, units = await _units(project_id) + units, assignees = visible_to(session, units, await repo.list_assignees(str(project_id))) + own_years = {u["year"] for u in units} + years = [year_summary(row) for row in rows if int(row["year"]) in own_years] + return { + "status": "success", + "name": project["name"], + "years": years, + "pieces": [unit_summary(u) for u in units], + "assignees": assignees, + } -def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes: - """연도마다 `<연도>/` · 전체는 `전체/` — 좌표표 이름이 연도마다 같아 폴더로 가름.""" +def export_zip(bundle: Any, units: list[dict[str, Any]], with_all: bool) -> bytes: + """단위마다 `<연도>/`(옛) · `<연도>_<선>_<조각>/` · 전체 `전체/` — 좌표표 이름이 같아 폴더로.""" buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: - for year in years: - for name, data in bundle.by_year[year].items(): - archive.writestr(f"{year}/{name}", data) + for unit in units: + for name, data in bundle.by_year[unit_key(unit)].items(): + archive.writestr(f"{unit_label(unit)}/{name}", data) if with_all: for name, data in bundle.all.items(): archive.writestr(f"전체/{name}", data) @@ -290,30 +418,24 @@ def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes: @router.get("/{project_id}/plan/export", response_model=None) async def export_plan(project_id: UUID, session: dict = Depends(verify_session)) -> Response: - """산림청 = 저장된 전 연도 + 전체 · 지정 회사 = 확정된 자기 연도만(전체 없음).""" + """산림청 = 저장된 전 조각 + 전체 · 지정 회사 = 확정된 자기 조각만(전체 없음).""" project = await _plan_project(project_id) - rows = await repo.list_year_routes(str(project_id)) - routes = year_routes_from_rows(rows) + _, units = await _units(project_id) + routes = unit_routes(units) if not routes: raise HTTPException(status_code=409, detail="분석 [저장]이 된 연도가 없습니다.") forest = is_forest_service(session) - if forest: - years = [r.year for r in routes] - else: - confirmed = {int(r["year"]) for r in rows if r.get("confirmed_at")} - mine = { - a["year"] - for a in await repo.list_assignees(str(project_id)) - if a["company_id"] is not None and a["company_id"] == session.get("company_id") - } - years = sorted({r.year for r in routes} & confirmed & mine) - if not years: - raise HTTPException(status_code=403, detail="받을 수 있는 연도가 없습니다.") + picked = [u for u in units if _saved(u)] + if not forest: + own, _ = visible_to(session, picked, await repo.list_assignees(str(project_id))) + picked = [u for u in own if u["confirmed_at"]] + if not picked: + raise HTTPException(status_code=403, detail="받을 수 있는 조각이 없습니다.") first = float(project.get("route_start_m") or 0.0) bundle = await asyncio.to_thread( build_export, project["name"], project["region"] or "", routes, first ) - data = await asyncio.to_thread(export_zip, bundle, years, forest) + data = await asyncio.to_thread(export_zip, bundle, picked, forest) name = f"{safe_name(project['name'])}_예정노선.zip" return Response( content=data, diff --git a/resources/tester/test_531_3_plan_link.py b/resources/tester/test_531_3_plan_link.py index 746c07746..93a6ffbea 100644 --- a/resources/tester/test_531_3_plan_link.py +++ b/resources/tester/test_531_3_plan_link.py @@ -438,9 +438,13 @@ def plan_repo(monkeypatch): {"id": 10, "year": 2028, "company_id": 6, "company_name": "다", "email": "c@da.kr"}, ] + async def no_pieces(_pid): # 옛 프로젝트 — 조각 없음 + return [] + monkeypatch.setattr(repo, "get_plan_project", project) monkeypatch.setattr(repo, "list_year_routes", years) monkeypatch.setattr(repo, "list_assignees", assignees) + monkeypatch.setattr(repo, "list_pieces", no_pieces) def _zip_names(response): diff --git a/resources/tester/test_531_d04_export_finalize.py b/resources/tester/test_531_d04_export_finalize.py index cd4142df8..621fca05b 100644 --- a/resources/tester/test_531_d04_export_finalize.py +++ b/resources/tester/test_531_d04_export_finalize.py @@ -328,9 +328,13 @@ def fake_db(monkeypatch): state["mails"].append((to_email, subject, attachments)) return True + async def no_pieces(_pid): # 옛 프로젝트 — 조각 없음 + return [] + monkeypatch.setattr(repo, "get_plan_project", project) monkeypatch.setattr(repo, "list_year_routes", years) monkeypatch.setattr(repo, "list_assignees", assignees) + monkeypatch.setattr(repo, "list_pieces", no_pieces) monkeypatch.setattr(repo, "mark_confirmed", confirmed) monkeypatch.setattr(repo, "mark_notified", notified) monkeypatch.setattr(repo, "apply_assignee_changes", apply) diff --git a/resources/tester/test_563_2_plan_pieces.py b/resources/tester/test_563_2_plan_pieces.py new file mode 100644 index 000000000..987420ae0 --- /dev/null +++ b/resources/tester/test_563_2_plan_pieces.py @@ -0,0 +1,331 @@ +# -*- coding: utf-8 -*- +"""563-2 예정노선 아래 단계 = 조각 단위(요구 391). + +견본 = 같은 연도 조각 둘(본선 p1 · 갈래 p3, 2026) + 본선 p2(2027) · 업체 둘(p1 = 회사 5 · p3 = 회사 6). +- D04 요약 = 조각 줄 · 이름 「본선 0+0~12+10」 · 지정 회사 = 자기 조각 · 자기 행만 +- 내보내기 = 조각마다 폴더 · 전체 = 선마다 레코드 · 지정 회사 = 확정된 자기 조각만 +- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘) +- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행 +- 연결 · 가져오기 = 조각 꺾은선 shp · 받은 목록 = 조각 이름 · 길이 +""" + +from __future__ import annotations + +import asyncio +import io +import json +import sys +import tempfile +import zipfile +from pathlib import Path +from uuid import uuid4 + +import geopandas as gpd +import pytest +from fastapi import HTTPException + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata # noqa: E402 +from D01_PlanRegister import D01_PlanRegister_Router as d01 # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Link as link # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Repository as repo # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Router as d04 # noqa: E402 + +PLAN_ID = str(uuid4()) +P1 = [[210000.0, 380000.0], [210200.0, 380150.0]] # 본선 0 ~ 250 +P2 = [[210200.0, 380150.0], [210300.0, 380400.0]] # 본선 250 ~ 519.258 +P3 = [[210200.0, 380150.0], [210330.0, 380150.0]] # 갈래 0 ~ 130 +L2 = (100.0**2 + 250.0**2) ** 0.5 + +FOREST = {"role": "SYSTEM_ADMIN", "email": "root@x.kr", "company_id": 1, "user_id": 1} +FIRM5 = {"role": "USER", "email": "a@ga.kr", "company_id": 5, "user_id": 3} +FIRM6 = {"role": "USER", "email": "b@na.kr", "company_id": 6, "user_id": 4} + + +def _piece(pid, line, name, year, start, end, planned): + return { + "piece_id": pid, + "line_id": line, + "line_name": name, + "year": year, + "from_m": start, + "to_m": end, + "length_m": end - start, + "epsg": 5186, + "route_json": {"planned": planned, "stations": [start, end]}, + } + + +PIECES = [ + _piece("p1", "main", "본선", 2026, 0.0, 250.0, P1), + _piece("p3", "b1", "갈래 1", 2026, 0.0, 130.0, P3), + _piece("p2", "main", "본선", 2027, 250.0, 250.0 + L2, P2), +] + + +def _assignee(aid, piece, year, company, name, email, notified=None): + return { + "id": aid, + "year": year, + "piece_id": piece, + "company_id": company, + "company_name": name, + "email": email, + "notified_at": notified, + "linked_project_id": None, + } + + +@pytest.fixture +def plan(monkeypatch): + state = { + "years": [ + {"year": 2026, "epsg": 5186, "route_json": {"planned": P1}, "confirmed_at": "x"}, + {"year": 2027, "epsg": 5186, "route_json": {"planned": P2}, "confirmed_at": None}, + ], + "assignees": [ + _assignee(7, "p1", 2026, 5, "가", "a@ga.kr"), + _assignee(8, "p3", 2026, 6, "나", "b@na.kr"), + ], + "mails": [], + "applied": None, + } + + async def project(_pid): + return {"id": PLAN_ID, "name": "시험", "region": "일월.용화", "project_kind": "PLAN"} + + async def years(_pid): + return state["years"] + + async def pieces(_pid): + return PIECES + + async def assignees(_pid): + return state["assignees"] + + async def noop(*a): + return None + + async def apply(_pid, d, u, i): + state["applied"] = (d, u, i) + + async def contact(cid): + return {"id": cid, "name": f"등록{cid}", "email": f"m{cid}@reg.kr"} + + async def send(*, to_email, subject, body, attachments): + state["mails"].append((to_email, attachments)) + return True + + monkeypatch.setattr(repo, "get_plan_project", project) + monkeypatch.setattr(repo, "list_year_routes", years) + monkeypatch.setattr(repo, "list_pieces", pieces) + monkeypatch.setattr(repo, "list_assignees", assignees) + monkeypatch.setattr(repo, "mark_confirmed", noop) + monkeypatch.setattr(repo, "mark_notified", noop) + monkeypatch.setattr(repo, "apply_assignee_changes", apply) + monkeypatch.setattr(repo, "get_company_contact", contact) + monkeypatch.setattr(d04, "send_completion_mail", send) + return state + + +def _names(response) -> list[str]: + return sorted(zipfile.ZipFile(io.BytesIO(response.body)).namelist()) + + +def test_summary_piece_rows_and_firm_view(plan): + out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST)) + rows = out["pieces"] + assert [(r["piece_id"], r["year"], r["label"]) for r in rows] == [ + ("p1", 2026, "본선 0+0~12+10"), + ("p3", 2026, "갈래 1 0+0~6+10"), + ("p2", 2027, "본선 12+10~25+19.3"), + ] + assert rows[1]["length_m"] == 130.0 and rows[1]["start"]["x"] == P3[0][0] + assert rows[0]["confirmed_at"] == "x" and rows[2]["confirmed_at"] is None + assert [y["year"] for y in out["years"]] == [2026, 2027] + # 같은 연도 다른 업체 — 자기 조각 · 자기 행만 + for session, piece, aid in ((FIRM5, "p1", 7), (FIRM6, "p3", 8)): + mine = asyncio.run(d04.get_plan_summary(PLAN_ID, session)) + assert [r["piece_id"] for r in mine["pieces"]] == [piece] + assert [a["id"] for a in mine["assignees"]] == [aid] + assert [y["year"] for y in mine["years"]] == [2026] + assert "b@na.kr" not in str(asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5))) + + +def test_export_piece_folders_and_firm_only_own(plan): + names = _names(asyncio.run(d04.export_plan(PLAN_ID, FOREST))) + assert {n.split("/")[0] for n in names} == { + "2026_본선_p1", + "2026_갈래1_p3", + "2027_본선_p2", + "전체", + } + assert sum(n.endswith(".shp") for n in names) == 4 + with zipfile.ZipFile(io.BytesIO(asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body)) as z: + with tempfile.TemporaryDirectory() as tmp: + for n in z.namelist(): + if n.startswith("전체/") and not n.endswith(".xlsx"): + (Path(tmp) / n.split("/")[1]).write_bytes(z.read(n)) + whole = gpd.read_file(next(Path(tmp).glob("*.shp"))) + # 전체 = 선마다 한 레코드 — 본선(p1 + p2 이음) · 갈래 + assert len(whole) == 2 + assert sorted(round(g.length, 1) for g in whole.to_crs(5186).geometry) == [ + 130.0, + round(250.0 + L2, 1), + ] + firm = _names(asyncio.run(d04.export_plan(PLAN_ID, FIRM6))) + assert {n.split("/")[0] for n in firm} == {"2026_갈래1_p3"} + plan["years"][0]["confirmed_at"] = None # 확정 전 = 받을 것 없음 + with pytest.raises(HTTPException) as err: + asyncio.run(d04.export_plan(PLAN_ID, FIRM6)) + assert err.value.status_code == 403 + + +def test_finalize_mails_own_piece_zip(plan): + out = asyncio.run(d04.finalize_plan(PLAN_ID, d04.FinalizeRequest(years=[2026]), FOREST)) + assert [(n["piece_id"], n["state"]) for n in out["notifications"]] == [ + ("p1", "sent"), + ("p3", "sent"), + ] + for (to, attachments), piece in zip(plan["mails"], ("본선_p1", "갈래1_p3"), strict=True): + name, blob, _ = attachments[0] + assert name == f"시험_예정노선_2026_{piece}.zip" + shp = [n for n in zipfile.ZipFile(io.BytesIO(blob)).namelist() if n.endswith(".shp")] + assert len(shp) == 1 and shp[0].endswith(f"_2026_{piece}.shp") + + +def test_put_assignees_by_piece(plan): + def put(items, years=None): + request = d04.AssigneesRequest(assignees=items, years=years) + return asyncio.run(d04.put_plan_assignees(PLAN_ID, request, FOREST)) + + with pytest.raises(HTTPException) as err: + put([d04.AssigneeIn(year=2026, piece_id="없음", company_id=5)]) + assert err.value.status_code == 400 + # 같은 업체(회사 5)를 같은 연도 두 조각에 — 두 행 · 연도는 조각에서(보낸 2099 무시) + put( + [ + d04.AssigneeIn(year=2026, piece_id="p1", company_id=5), + d04.AssigneeIn(year=2099, piece_id="p3", company_id=5), + ], + years=[2026], + ) + deleted, updates, inserts = plan["applied"] + assert deleted == [8] and [u["id"] for u in updates] == [7] + assert [(i["piece_id"], i["year"], i["company_id"]) for i in inserts] == [("p3", 2026, 5)] + + +def test_legacy_year_assignee_sits_on_first_piece(plan): + """조각이 생기기 전 연도 지정(piece_id 없음) — 그 연도 첫 조각에 붙어 보임.""" + plan["assignees"] = [_assignee(9, None, 2026, 5, "가", "a@ga.kr")] + mine = asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5)) + assert [r["piece_id"] for r in mine["pieces"]] == ["p1"] + + +def test_link_uses_piece_route(): + row = { + "id": 7, + "year": 2026, + "piece_id": "p3", + "piece_epsg": 5186, + "piece_json": json.dumps({"planned": P3, "stations": [0.0, 130.0]}), + "from_m": 0.0, + "to_m": 130.0, + "piece_length_m": 130.0, + "epsg": 5186, + "route_json": json.dumps({"planned": P1, "length_m": 380.0}), + "name": "2026 홍천 노선 계획", + "region": "일월.용화", + } + unit = link.unit_row(row) + assert unit["route_json"]["planned"] == P3 and unit["route_json"]["length_m"] == 130.0 + assert "piece_json" not in unit and "from_m" not in unit + assert link.route_fingerprint(unit["route_json"]) != link.route_fingerprint(row["route_json"]) + files = link.year_files(unit) + with tempfile.TemporaryDirectory() as tmp: + for name, data in files.items(): + (Path(tmp) / name).write_bytes(data) + meta = analyze_input_metadata(next(Path(tmp).glob("*.shp"))) + assert meta["point_count"] == 2 and float(meta["attributes"]["종점누가"]) == 130.0 + # 옛 연도 지정 = 연도 노선 그대로 + legacy = link.unit_row({**row, "piece_id": None}) + assert legacy["route_json"] == row["route_json"] and legacy["epsg"] == 5186 + + +class _Cursor: + def __init__(self, rows): + self.rows = rows + + async def execute(self, sql, args=None): + self.sql = sql + + async def fetchall(self): + return self.rows + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +class _Conn: + def __init__(self, rows): + self.rows = rows + + def acquire(self): + return self + + def cursor(self, *a): + return _Cursor(self.rows) + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +def test_assigned_list_piece_label_length(monkeypatch): + rows = [ + { + "assignee_id": 8, + "project_id": PLAN_ID, + "project_name": "시험", + "year": 2026, + "piece_id": "p3", + "piece_length_m": 130.0, + "linked_project_id": None, + "linked_project_name": None, + "source_fingerprint": None, + "route_json": json.dumps({"planned": P3}), + "confirmed_at": "x", + }, + { + "assignee_id": 9, + "project_id": PLAN_ID, + "project_name": "옛", + "year": 2025, + "piece_id": None, + "piece_length_m": None, + "linked_project_id": None, + "linked_project_name": None, + "source_fingerprint": None, + "route_json": json.dumps({"planned": P1, "length_m": 250.0}), + "confirmed_at": "x", + }, + ] + + async def pieces(_pid): + return PIECES + + monkeypatch.setattr(d01, "get_db_pool", lambda: _Conn(rows)) + monkeypatch.setattr(repo, "list_pieces", pieces) + out = asyncio.run(d01.assigned_plan_routes(FIRM6))["routes"] + assert [(r["piece_id"], r["piece_label"], r["length_m"]) for r in out] == [ + ("p3", "갈래 1 0+0~6+10", 130.0), + (None, None, 250.0), + ]