eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(D 예정노선): 설계 내용 = 프로젝트 폴더 한 파일 D03_PlanAnalysis/plan_network.json — 저장 모듈 · 서버 길 옮김 · 업체 연도 = 파일 조각 (PLAN 590-1 · 요구 452 · 453)

Browse Source 
- 저장 모듈 D03_PlanAnalysis_Store — 읽기 · 원자 쓰기 · 판 · 옛 DB 칸 읽을 때 옮김(지점 셋 꼴 · 연도 행 접기 · 옛 연도 노선) · plan_state(조각 · 연도 지문 · 확정 = 지문 같을 때만)
- D03 [저장] · [초기화] · D02 지점 = 파일만 + 연도 상태 줄(지문 · 확정) · 프로젝트 잠금 · 파일 쓰기 실패 = DB 되돌림 · 빈 연도 줄 지움
- D04 요약 · 업체 지정 · 확정 메일 · 내보내기 · 대시보드 [연결] · 받은 예정노선 · 산림청 홈 · 개요 연결 = 파일에서 · 업체 줄 연도 칸 안 읽음(⑤)
- 산림청 홈 「분석 전」 = 노선 안 그은 예정노선 프로젝트 수 · plan-source 지운 원본 거름
- 옛 연도 단위 지정 · 옛 연도 행 노선 길 걷음(DB 실측 0) · 연결 때 설계속도 씨앗 걷음(591-1 이음)
- SQL 023 = 업체 줄 year NULL 허용(옛 칸은 590-3 에서 지움)
- 시험: 새 지킴 590_1(옮김 · 원자 쓰기 · 연도 읽기 · 잠금) · 낡은 시험 고침 — 529_3 · 530_2 · 531_3 · 531_d04 · 547_2 · 563_2(DB 조각 · 연도 행 흉내 → 파일 흉내 · 옛 연도 지정 시험 · 설계속도 씨앗 시험 지움 = 그 길 걷음)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
eomsangdon
2026-10-30 09:52:51 +09:00
co-authored by Claude Opus 5.5
parent 1c0c44f9e2
commit 54e08d0d65
18 changed files with 1032 additions and 798 deletions
Download Patch File Download Diff File Expand all files Collapse all files
B01_Dashboard/B01_Dashboard_Router_Home.py
+23 -12
View File
@@ -4,7 +4,7 @@ from datetime import datetime, timedelta
from typing import Any
import aiomysql
from fastapi import APIRouter, Depends, Query
from fastapi import APIRouter, Depends, HTTPException, Query
from common_util.common_util_auth import is_forest_email, verify_session
from config.config_db import get_db_pool
@@ -138,22 +138,33 @@ async def _company(cursor: Any, company_id: int) -> dict[str, Any]:
async def _forest(cursor: Any, company_id: int) -> dict[str, Any]:
"""예정노선 연도 상태별 수(분석 전 · 분석됨 · 확정) · 업체 지정 대기(확정 · 업체 없음)."""
"""예정노선 수 — 분석 전 = 노선을 아직 안 그은 프로젝트 · 분석됨 · 확정 = 연도(설계 내용 파일) ·
업체 지정 대기 = 확정 연도 중 그 연도 조각에 업체가 하나도 없는 것."""
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
await cursor.execute(
"""SELECT y.route_json IS NOT NULL AS analyzed, y.confirmed_at IS NOT NULL AS confirmed,
EXISTS (SELECT 1 FROM plan_route_assignees a
WHERE a.project_id = y.project_id AND a.year = y.year) AS assigned
FROM plan_route_years y
JOIN projects p ON p.id = y.project_id AND p.deleted_at IS NULL
WHERE p.company_id = %s""",
"""SELECT p.id, a.piece_id FROM projects p
LEFT JOIN plan_route_assignees a ON a.project_id = p.id
WHERE p.company_id = %s AND p.project_kind = 'PLAN' AND p.deleted_at IS NULL""",
(company_id,),
)
assigned: dict[str, set] = {}
for r in await cursor.fetchall():
assigned.setdefault(str(r["id"]), set()).add(r["piece_id"])
status = {"draft": 0, "analyzed": 0, "confirmed": 0}
waiting = 0
for r in await cursor.fetchall():
key = "confirmed" if r["confirmed"] else "analyzed" if r["analyzed"] else "draft"
status[key] += 1
waiting += bool(r["confirmed"] and not r["assigned"])
for project_id, pieces in assigned.items():
try:
state = await store.plan_state(project_id)
except HTTPException: # 노선 파일이 틀린 프로젝트 = 건너뜀
continue
if not state["plan"]["lines"]:
status["draft"] += 1
for year, row in state["years"].items():
key = "confirmed" if row["confirmed_at"] else "analyzed"
status[key] += 1
ids = {p["id"] for p in row["route"]["pieces"]}
waiting += bool(row["confirmed_at"] and not ids & pieces)
return {"plan_status": status, "assign_waiting": waiting}
B01_Dashboard/B01_Dashboard_Router_Overview.py
+16 -4
View File
@@ -232,6 +232,17 @@ def _link(kind: str, row: dict[str, Any], note: str | None = None) -> dict[str,
}
async def _plan_year_note(project_id: str, piece_id: str | None) -> str | None:
"""「N 년」 — 그 조각의 연도(설계 내용 파일) · 조각 · 파일이 없으면 None."""
from D04_PlanConfirm.D04_PlanConfirm_Link import piece_route
try:
year = (await piece_route(project_id, piece_id))["year"]
except HTTPException:
return None
return f"{year} 년" if year else None
async def _links(cursor: Any, row: dict[str, Any]) -> list[dict[str, Any]]:
"""연결된 것 — 원본 · 이 프로젝트에서 나온 것 · 연결한 예정노선(원본과 같으면 한 번만)."""
links: list[dict[str, Any]] = []
@@ -249,16 +260,17 @@ async def _links(cursor: Any, row: dict[str, Any]) -> list[dict[str, Any]]:
)
links += [_link("derived", r) for r in await cursor.fetchall()]
await cursor.execute(
"""SELECT p.id, p.name, a.year FROM plan_route_assignees a
"""SELECT p.id, p.name, a.piece_id FROM plan_route_assignees a
JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL
WHERE a.linked_project_id = %s ORDER BY a.year""",
WHERE a.linked_project_id = %s ORDER BY a.id""",
(row["id"],),
)
for r in await cursor.fetchall():
note = await _plan_year_note(str(r["id"]), r["piece_id"])
if r["id"] == source:
links[0]["note"] = f"{r['year']} 년"
links[0]["note"] = note
else:
links.append(_link("plan", r, f"{r['year']} 년"))
links.append(_link("plan", r, note))
return links
D01_PlanRegister/D01_PlanRegister_Router.py
+25 -27
View File
@@ -1,10 +1,9 @@
"""D01 예정노선 등록 API · 대시보드 「받은 예정노선」 (요구 345 · 요구서 3-1 · 2-2).
"""D01 예정노선 등록 API · 대시보드 「받은 예정노선」 (요구 345 · 452 · 요구서 3-1 · 2-2).
화면은 이름 · 시작 연도 · 사업 지역만 보냄(요구서 389 3장 1 · 3) — 끝 연도 = 시작 · 임도 종류 없음
= 설계속도 작업임도 20 · 측점 없음(DB 칸 그대로).
"""
import json
from typing import Any
import aiomysql
@@ -73,9 +72,9 @@ 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
"""우리 회사에 지정된 예정노선 조각 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트 ·
조각 이름 · 길이 · 연도(설계 내용 파일 `plan_state` · 업체 줄은 조각 번호만)."""
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
from D04_PlanConfirm.D04_PlanConfirm_Link import update_available
from D04_PlanConfirm.D04_PlanConfirm_Router import piece_label
@@ -84,44 +83,43 @@ async def assigned_plan_routes(
connection.cursor(aiomysql.DictCursor) as cursor,
):
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,
"""SELECT a.id AS assignee_id, a.project_id, p.name AS project_name, a.piece_id,
l.id AS linked_project_id, l.name AS linked_project_name,
l.source_fingerprint, COALESCE(c.route_json, y.route_json) AS route_json,
y.confirmed_at
l.source_fingerprint
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, a.piece_id""",
ORDER BY p.name, 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
for project_id in {str(r["project_id"]) for r in rows}:
try:
units = (await store.plan_state(project_id))["units"]
except HTTPException: # 노선 파일이 틀린 프로젝트 = 건너뜀(목록 전체를 막지 않음)
continue
for unit in units:
pieces[(project_id, unit["id"])] = unit
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")))
fingerprint = row.pop("source_fingerprint")
piece = pieces.get((str(row["project_id"]), row["piece_id"]))
confirmed_at = piece and piece["confirmed_at"]
if confirmed_at is None and row["linked_project_id"] is None:
continue
route = piece and {"planned": piece["planned"]}
routes.append(
{
**row,
"piece_id": row.get("piece_id"),
"year": piece and piece["year"],
"confirmed_at": confirmed_at,
"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),
"length_m": piece and piece["length_m"],
"update_available": update_available(route, fingerprint, bool(confirmed_at)),
}
)
routes.sort(key=lambda r: (r["project_name"], r["year"] or 0, r["piece_id"]))
return {"status": "success", "routes": routes}
D02_PlanBase/D02_PlanBase_Router.py
+22 -77
View File
@@ -1,28 +1,25 @@
"""D 예정노선 1단계 기준 지정 — 지점 목록 하나 정본 API
(요구서 389 3장 1 · 440 1 · 445 A2 · PLAN 560-2 · 579-1 · 585-1).
"""D 예정노선 1단계 기준 지정 — 지점 목록 API
(요구서 389 3장 1 · 440 1 · 445 A2 · 452 · PLAN 560-2 · 579-1 · 585-1 · 590-1).
- `GET /api/projects/{id}/plan/points` — {epsg, points: [점]} · 점 = {lon, lat, x, y}
- `PUT …/plan/points` — {points: [{lon, lat}]} → TM 셈 → `projects.plan_base_json`
(`{"epsg", "points": [{"x", "y"}]}` · 좌표계 = 저장된 값 · 없으면 점 평균 경도의 TM 띠)
- 시점 · 종점 구분 없음 — 옛 꼴(`starts` + `ends` 목록 · 더 옛 `start` · `end` 한 점)은 읽을 때
이어 붙임(옮김 없음 · 쓰기는 새 꼴).
- 칸이 비면(옛 프로젝트) [첫 연도 시점, 마지막 연도 종점]으로 접어 보임 · 연도 행은 D03 [저장] 몫.
- `PUT …/plan/points` — {points: [{lon, lat}]} → TM 셈 → 설계 내용 파일 `points`
(`D03_PlanAnalysis_Store` · 좌표계 = 파일 머리 · 없으면 점 평균 경도의 TM 띠)
- 시점 · 종점 구분 없음 — 옛 꼴(DB `plan_base_json` · 연도 행 접기)은 저장 모듈의 옮김 몫.
- 고침 = 산림청만(지정 회사 GET 은 공용 판정 `require_project_access` 몫).
"""
from __future__ import annotations
import json
from typing import Any
from uuid import UUID
import aiomysql
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from common_util.common_util_auth import is_forest_service, require_company
from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat
from config.config_db import get_db_pool
from common_util.common_util_save_guard import project_lock
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
router = APIRouter(prefix="/api/projects", tags=["D02 Plan Base"])
@@ -38,31 +35,6 @@ class BaseRequest(BaseModel):
points: list[LonLat] = Field(default_factory=list, max_length=MAX_POINTS)
async def _load(project_id: UUID) -> tuple[dict[str, Any], dict[int, dict[str, Any]]]:
"""(예정노선 프로젝트 칸, 연도 → 지금 행)."""
async with (
get_db_pool().acquire() as connection,
connection.cursor(aiomysql.DictCursor) as cursor,
):
await cursor.execute(
"SELECT project_kind, project_year, project_year_end, plan_base_json FROM projects"
" WHERE id = %s AND deleted_at IS NULL",
(str(project_id),),
)
project = await cursor.fetchone()
await cursor.execute(
"SELECT year, start_x, start_y, end_x, end_y, epsg FROM plan_route_years"
" WHERE project_id = %s",
(str(project_id),),
)
rows = await cursor.fetchall()
if not project:
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
if project["project_kind"] != "PLAN" or project["project_year"] is None:
raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
return project, {int(row["year"]): row for row in rows}
def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None:
if x is None or y is None or epsg is None:
return None
@@ -70,43 +42,21 @@ def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None:
return {"lon": lon, "lat": lat, "x": float(x), "y": float(y)}
def _stored(project: dict[str, Any]) -> dict[str, Any] | None:
stored = project.get("plan_base_json")
return json.loads(stored) if isinstance(stored, str) else stored
def base_view(project: dict[str, Any], rows: dict[int, dict[str, Any]]) -> dict[str, Any]:
"""{epsg, points} — 정본 칸(옛 시점 · 종점 꼴은 이어 붙임) ·
비면 [첫 연도 시점, 마지막 연도 종점]으로 접음."""
stored = _stored(project)
if stored:
epsg = stored.get("epsg")
items = stored.get("points")
if items is None:
starts = stored.get("starts", [stored["start"]] if stored.get("start") else [])
ends = stored.get("ends", [stored["end"]] if stored.get("end") else [])
items = [*starts, *ends]
elif rows:
first, last = rows[min(rows)], rows[max(rows)]
epsg = first.get("epsg")
items = [
{"x": first.get("start_x"), "y": first.get("start_y")},
{"x": last.get("end_x"), "y": last.get("end_y")},
]
else:
epsg, items = None, []
points = [p for p in (_point(i.get("x"), i.get("y"), epsg) for i in items) if p]
return {"epsg": epsg, "points": points}
def base_view(plan: dict[str, Any]) -> dict[str, Any]:
"""{epsg, points} — 설계 내용의 지점(TM) + 경위도 · 지점이 없으면 epsg None."""
epsg = plan.get("epsg")
points = [p for p in (_point(i.get("x"), i.get("y"), epsg) for i in plan["points"]) if p]
return {"epsg": epsg if points else None, "points": points}
async def load_base(project_id: UUID) -> dict[str, Any]:
"""지점 목록 — D03 이 읽는 길."""
project, rows = await _load(project_id)
return base_view(project, rows)
_, plan = await store.load_plan(project_id)
return base_view(plan)
def stored_base(payload: BaseRequest, epsg: int | None = None) -> dict[str, Any]:
"""PUT 몸 → 정본 칸 꼴(TM m) — 좌표계 = `epsg`(저장된 값) · 없으면 점 평균 경도의 TM 띠."""
"""PUT 몸 → 파일 꼴(TM m) — 좌표계 = `epsg`(저장된 값) · 없으면 점 평균 경도의 TM 띠."""
given = payload.points
if not given:
epsg = None
@@ -116,14 +66,6 @@ def stored_base(payload: BaseRequest, epsg: int | None = None) -> dict[str, Any]
return {"epsg": epsg, "points": points}
async def _store_base(project_id: UUID, base: dict[str, Any]) -> None:
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"UPDATE projects SET plan_base_json = %s WHERE id = %s",
(json.dumps(base), str(project_id)),
)
@router.get("/{project_id}/plan/points", response_model=None)
async def get_plan_points(project_id: UUID) -> dict[str, Any]:
return {"status": "success", **await load_base(project_id)}
@@ -137,7 +79,10 @@ async def put_plan_points(
) -> dict[str, Any]:
if not is_forest_service(session):
raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
project, _ = await _load(project_id) # 없음 · 예정노선 아님 거절
epsg = (_stored(project) or {}).get("epsg")
await _store_base(project_id, stored_base(payload, int(epsg) if epsg else None))
return {"status": "success", **await load_base(project_id)}
async with project_lock(str(project_id)):
root, plan = await store.load_plan(project_id) # 없음 · 예정노선 아님 거절
base = stored_base(payload, plan["epsg"])
epsg = plan["epsg"] if plan["lines"] else base["epsg"] # 선이 있으면 좌표계 그대로
plan = {**plan, "epsg": epsg, "points": base["points"]}
store.save_plan(root, plan)
return {"status": "success", **base_view(plan)}
D03_PlanAnalysis/D03_PlanAnalysis_Network.py
+63 -145
View File
@@ -1,17 +1,14 @@
"""D 예정노선 2단계 노선 정본 — 손으로 그은 직선 · 갈래 · 구역
(요구 389 6 · 11 · 13 · 14 · 391 · PLAN 562-2).
(요구 389 6 · 11 · 13 · 14 · 391 · 452 · PLAN 562-2 · 590-1).
- `GET /api/projects/{id}/plan/network` — 정본(`plan_networks`) + 서버가 다시 센 길이 ·
측점 · 조각 · 저장 없으면 옛 연도 노선(`plan_route_years.route_json.planned`)을 한 선으로 바꿔
보임(legacy)
- `GET /api/projects/{id}/plan/network` — 설계 내용 파일(`D03_PlanAnalysis_Store`) + 서버가 다시 센
길이 · 측점 · 조각 · 파일이 없으면 옛 DB 칸을 옮겨 보임(쓰지 않음)
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
다시 셈 → `plan_networks` · 조각마다 한 줄(`plan_route_pieces`) · 연도 행(`plan_route_years`) ·
사업연도 끝
다시 셈 → 파일(원자 쓰기) · 연도 상태 줄(`plan_route_years` 지문 · 확정) · 사업연도 끝
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
지점 목록(`base` · 585-1) = D02 정본(`plan_base.load_base`) — 비어 옛 연도 행을 접어 보이던 것은
[저장] 전에 칸에 굳힘(연도 행이 바뀌어도 그대로).
지점 목록(`base`) = 같은 파일의 `points`(D02 `base_view`).
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
"""
@@ -23,14 +20,15 @@ import math
from typing import Any
from uuid import UUID
import aiomysql
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from common_util.common_util_auth import is_forest_service, require_company
from common_util.common_util_save_guard import project_lock
from config.config_db import get_db_pool
from D02_PlanBase import D02_PlanBase_Router as plan_base
from D03_PlanAnalysis import D03_PlanAnalysis_Draw as draw
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Network"])
@@ -166,8 +164,8 @@ def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, A
def year_rows(pieces: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
"""조각 → 연도 행(`plan_route_years`) — `planned` = 첫 조각(옛 읽는 곳 그대로) ·
`pieces` = 그 연도 전부."""
"""조각 → 연도 묶음 — `planned` = 첫 조각 · `pieces` = 그 연도 전부 · `fingerprint` = 연도 상태
줄(`plan_route_years`) 지문."""
by_year: dict[int, list[dict[str, Any]]] = {}
for piece in pieces:
by_year.setdefault(piece["year"], []).append(piece)
@@ -251,150 +249,69 @@ def legacy_network(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
}
async def _load(connection: Any, project_id: UUID) -> tuple[dict[str, Any] | None, bool]:
"""(정본 · 옛 연도 노선을 바꾼 것인가) — 둘 다 없으면 (None, False)."""
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT network_json FROM plan_networks WHERE project_id = %s", (str(project_id),)
)
row = await cursor.fetchone()
if row:
return _json(row["network_json"]), False
await cursor.execute(
"SELECT year, epsg, route_json FROM plan_route_years"
" WHERE project_id = %s ORDER BY year",
(str(project_id),),
)
legacy = legacy_network(list(await cursor.fetchall()))
return legacy, legacy is not None
def _view(
network: dict[str, Any] | None, legacy: bool, base: dict[str, Any], year_start: int | None
) -> dict[str, Any]:
def _view(plan: dict[str, Any], year_start: int | None) -> dict[str, Any]:
network = {key: plan[key] for key in ("epsg", "lines", "marks", "pieces")}
return {
"status": "success",
"base": base,
"base": plan_base.base_view(plan),
"year_start": year_start,
"network": network,
"derived": derive(network) if network else None,
"legacy": legacy,
"network": network if plan["lines"] else None,
"derived": derive(network) if plan["lines"] else None,
"legacy": False,
}
async def _year_start(project_id: UUID) -> int | None:
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
await cursor.execute("SELECT project_year FROM projects WHERE id = %s", (str(project_id),))
row = await cursor.fetchone()
return row[0] if row else None
@router.get("/{project_id}/plan/network", response_model=None)
async def get_plan_network(project_id: UUID) -> dict[str, Any]:
base = await plan_base.load_base(project_id)
async with get_db_pool().acquire() as connection:
network, legacy = await _load(connection, project_id)
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT project_year FROM projects WHERE id = %s", (str(project_id),)
)
row = await cursor.fetchone()
_, plan = await store.load_plan(project_id)
try:
return _view(network, legacy, base, row["project_year"] if row else None)
return _view(plan, await _year_start(project_id))
except NetworkError as error:
raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
async def _store(
connection: Any, project_id: UUID, network: dict[str, Any], derived: dict[str, Any]
async def _store_status(
connection: Any, project_id: UUID, pieces: list[dict[str, Any]], years: dict[int, Any]
) -> None:
"""연도 상태 줄 — 있는 연도만(지문이 바뀐 연도만 확정 풂) · 없는 연도 줄 지움 · 사업연도 끝."""
pid = str(project_id)
epsg = network["epsg"]
years = year_rows(derived["pieces"])
async with connection.cursor() as cursor:
await cursor.execute(
"SELECT DISTINCT piece_id FROM plan_route_assignees"
" WHERE project_id = %s AND piece_id IS NOT NULL",
(pid,),
)
assigned = {r[0] for r in await cursor.fetchall()}
await cursor.execute(
"SELECT DISTINCT year FROM plan_route_assignees"
" WHERE project_id = %s AND piece_id IS NULL",
(pid,),
)
assigned_years = {int(r[0]) for r in await cursor.fetchall()}
lost = sorted(assigned - {p["id"] for p in derived["pieces"]})
lost_years = sorted(assigned_years - set(years))
if lost or lost_years:
lost = sorted({r[0] for r in await cursor.fetchall()} - {p["id"] for p in pieces})
if lost:
raise HTTPException(
status_code=409,
detail=f"업체가 지정된 조각 · 연도를 지울 수 없습니다: {lost + lost_years}",
)
await cursor.execute(
"INSERT INTO plan_networks (project_id, network_json) VALUES (%s, %s)"
" ON DUPLICATE KEY UPDATE network_json = VALUES(network_json)",
(pid, json.dumps(network, ensure_ascii=False)),
)
await cursor.execute("DELETE FROM plan_route_pieces WHERE project_id = %s", (pid,))
for p in derived["pieces"]:
route = {"planned": p["planned"], "stations": p["stations"]}
await cursor.execute(
"INSERT INTO plan_route_pieces (project_id, piece_id, line_id, year, from_m, to_m,"
" length_m, epsg, route_json, fingerprint)"
" VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
pid,
p["id"],
p["line"],
p["year"],
p["from_m"],
p["to_m"],
p["length_m"],
epsg,
json.dumps(route),
_sha1(route),
),
status_code=409, detail=f"업체가 지정된 조각을 지울 수 없습니다: {lost}"
)
for year, row in years.items():
# 내용이 바뀐 연도만 확정 풂(confirmed_at 을 fingerprint 보다 먼저 셈)
# confirmed_at 을 fingerprint 보다 먼저 셈
await cursor.execute(
"INSERT INTO plan_route_years (project_id, year, start_x, start_y, end_x, end_y,"
" epsg, route_json, fingerprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
" ON DUPLICATE KEY UPDATE start_x = VALUES(start_x), start_y = VALUES(start_y),"
" end_x = VALUES(end_x), end_y = VALUES(end_y), epsg = VALUES(epsg),"
" route_json = VALUES(route_json),"
"INSERT INTO plan_route_years (project_id, year, fingerprint) VALUES (%s, %s, %s)"
" ON DUPLICATE KEY UPDATE"
" confirmed_at = IF(fingerprint <=> VALUES(fingerprint), confirmed_at, NULL),"
" fingerprint = VALUES(fingerprint)",
(
pid,
year,
*row["start"],
*row["end"],
epsg,
json.dumps(row["route"], ensure_ascii=False),
row["fingerprint"],
),
(pid, year, row["fingerprint"]),
)
marks = ", ".join(["%s"] * len(years))
await cursor.execute(
"DELETE FROM plan_route_years WHERE project_id = %s"
+ (f" AND year NOT IN ({marks})" if years else ""),
(pid, *years),
)
if years:
marks = ", ".join(["%s"] * len(years))
await cursor.execute(
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
f" confirmed_at = NULL WHERE project_id = %s AND year NOT IN ({marks})",
(pid, *years),
)
await cursor.execute(
"UPDATE projects SET project_year_end = %s WHERE id = %s", (max(years), pid)
)
else:
await cursor.execute(
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
" confirmed_at = NULL WHERE project_id = %s",
(pid,),
)
async def _freeze_base(connection: Any, project_id: UUID, base: dict[str, Any]) -> None:
"""옛 연도 행을 접은 지점 → D02 정본 칸(비었을 때만)."""
stored = {"epsg": base["epsg"], "points": [{"x": p["x"], "y": p["y"]} for p in base["points"]]}
async with connection.cursor() as cursor:
await cursor.execute(
"UPDATE projects SET plan_base_json = %s WHERE id = %s AND plan_base_json IS NULL",
(json.dumps(stored), str(project_id)),
)
@router.put("/{project_id}/plan/network", response_model=None)
@@ -403,31 +320,32 @@ async def put_plan_network(
payload: Network,
session: dict[str, Any] = Depends(require_company),
) -> dict[str, Any]:
"""[저장] · [초기화](lines 빈 것) — 서버가 정본으로 다시 셈 → 정본 · 조각 행 · 연도 행."""
"""[저장] · [초기화](lines 빈 것) — 서버가 정본으로 다시 셈 → 파일 · 연도 상태 줄.
파일은 DB 커밋 직전에 씀 — 파일 쓰기가 깨지면 DB 되돌림 · 커밋이 깨지면 다음 [저장]이 바로잡음
(확정은 지문이 같을 때만 보이므로 노선이 틀리게 확정으로 보이지 않음)."""
if not is_forest_service(session):
raise HTTPException(status_code=403, detail="예정노선은 산림청만 고칠 수 있습니다.")
network = payload.model_dump()
base = await plan_base.load_base(project_id) # 예정노선 프로젝트가 아니면 404 · 400
if base["epsg"] is not None and int(base["epsg"]) != network["epsg"]:
raise HTTPException(status_code=400, detail="좌표계가 기준 지정과 다릅니다.")
async with get_db_pool().acquire() as connection:
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT project_year, plan_base_json FROM projects WHERE id = %s",
(str(project_id),),
)
project = await cursor.fetchone()
async with project_lock(str(project_id)):
root, plan = await store.load_plan(project_id) # 없음 404 · 예정노선 아님 400
if plan["epsg"] is not None and plan["epsg"] != network["epsg"]:
raise HTTPException(status_code=400, detail="좌표계가 기준 지정과 다릅니다.")
year_start = await _year_start(project_id)
try:
derived = derive(network, project["project_year"])
derived = derive(network, year_start)
except NetworkError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
await connection.begin()
try:
if project["plan_base_json"] is None and base["points"]:
await _freeze_base(connection, project_id, base)
await _store(connection, project_id, network, derived)
await connection.commit()
except Exception:
await connection.rollback()
raise
return {**_view(network, False, base, project["project_year"]), "derived": derived}
plan = {**plan, **network}
async with get_db_pool().acquire() as connection:
await connection.begin()
try:
await _store_status(
connection, project_id, derived["pieces"], year_rows(derived["pieces"])
)
store.save_plan(root, plan)
await connection.commit()
except Exception:
await connection.rollback()
raise
return {**_view(plan, year_start), "derived": derived}
D03_PlanAnalysis/D03_PlanAnalysis_Store.py
+214
View File
@@ -0,0 +1,214 @@
"""D 예정노선 설계 내용 한 파일 — `<프로젝트>/D03_PlanAnalysis/plan_network.json`
(요구 452 · 453 · PLAN 590-1).
- 꼴 = `{version, epsg, points: [{x, y}], lines, marks, pieces: [{id, line, from_m, to_m, year}]}` —
좌표계 = 머리 한 곳 · 연도 = 조각의 `year` 뿐(빈 연도 줄 없음) · 길이 · 측점 · 조각 좌표 ·
연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
- DB 에 남는 것 = 등록 칸 · 연도 상태(`plan_route_years` 의 year · fingerprint · confirmed_at) ·
업체 줄(`plan_route_assignees` 조각 번호 · 업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
- 옮김 = 파일이 없으면 옛 DB 칸(`projects.plan_base_json` · `plan_networks` · `plan_route_years`)을
메모리에서 새 꼴로(`migrate`) · 파일은 첫 쓰기 때 생김 · 옛 칸은 손대지 않음(590-3 에서 지움).
- 쓰기 = `project_lock(project_id)` 안에서 `save_plan`(같은 폴더 임시 파일 → 바꿔치기 · 반쯤 쓴
파일이 안 보임) · 읽기는 잠금 없음.
- 확정 = 연도 상태 줄 지문이 파일 지문과 같을 때만(파일만 바뀌고 DB 가 못 따라간 틈 =
풀린 것으로 봄).
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from uuid import UUID
import aiomysql
from fastapi import HTTPException
from common_util.common_util_crs import lonlat_to_tm, tm_to_lonlat
from common_util.common_util_json import atomic_write_json
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
from D04_PlanConfirm.D04_PlanConfirm_Repository import line_names
FILE = Path("D03_PlanAnalysis") / "plan_network.json"
VERSION = 1
KEYS = ("epsg", "points", "lines", "marks", "pieces")
def empty_plan(epsg: int | None = None) -> dict[str, Any]:
return {"version": VERSION, "epsg": epsg, "points": [], "lines": [], "marks": [], "pieces": []}
def plan_path(root: str | Path) -> Path:
return Path(root) / FILE
def _clean(plan: dict[str, Any]) -> dict[str, Any]:
out = empty_plan()
out.update({key: plan[key] for key in KEYS if plan.get(key) is not None})
out["epsg"] = int(out["epsg"]) if out["epsg"] is not None else None
return out
def read_plan(root: str | Path) -> dict[str, Any] | None:
"""파일 — 없으면 None · 이 코드보다 새 판이면 거절(덮어써 잃지 않게)."""
path = plan_path(root)
if not path.is_file():
return None
plan = json.loads(path.read_text("utf-8"))
if int(plan.get("version") or 0) > VERSION:
raise HTTPException(status_code=409, detail="새 판 노선 파일입니다 — 서버를 새로 받으세요.")
return _clean(plan)
def save_plan(root: str | Path, plan: dict[str, Any]) -> None:
atomic_write_json(plan_path(root), _clean(plan))
def _parsed(value: Any) -> Any:
if isinstance(value, (str, bytes)):
try:
return json.loads(value)
except ValueError:
return None
return value
def _old_points(base: dict[str, Any] | None, rows: list[dict[str, Any]]) -> tuple[Any, list]:
"""옛 지점 칸 → (epsg, [{x, y}]) — points · 옛 starts/ends 목록 · 더 옛 start/end 한 점 ·
칸이 비면 [첫 연도 시점, 마지막 연도 종점]으로 접음."""
if base:
items = base.get("points")
if items is None:
starts = base.get("starts", [base["start"]] if base.get("start") else [])
ends = base.get("ends", [base["end"]] if base.get("end") else [])
items = [*starts, *ends]
epsg = base.get("epsg")
else:
dated = [r for r in rows if r.get("start_x") is not None or r.get("end_x") is not None]
if not dated:
return None, []
first, last = dated[0], dated[-1]
epsg = first.get("epsg")
items = [
{"x": first.get("start_x"), "y": first.get("start_y")},
{"x": last.get("end_x"), "y": last.get("end_y")},
]
points = [
{"x": float(p["x"]), "y": float(p["y"])}
for p in items
if p and p.get("x") is not None and p.get("y") is not None
]
return (epsg if points else None), points
def migrate(base_json: Any, network_json: Any, year_rows: list[dict[str, Any]]) -> dict[str, Any]:
"""옛 DB 칸 → 새 꼴(메모리) — 노선 = `plan_networks` · 없으면 옛 연도 노선을 한 선으로."""
rows = sorted(year_rows, key=lambda r: int(r["year"]))
epsg, points = _old_points(_parsed(base_json), rows)
net = _parsed(network_json) or network.legacy_network(rows)
plan = empty_plan(int(epsg) if epsg is not None else None)
plan["points"] = points
if net and net.get("lines"):
net_epsg = int(net["epsg"]) if net.get("epsg") is not None else plan["epsg"]
if points and plan["epsg"] is not None and net_epsg != plan["epsg"]:
plan["points"] = [
dict(
zip(
("x", "y"),
lonlat_to_tm(*tm_to_lonlat(p["x"], p["y"], plan["epsg"]), net_epsg),
)
)
for p in points
]
plan.update(
epsg=net_epsg,
lines=net.get("lines", []),
marks=net.get("marks", []),
pieces=net.get("pieces", []),
)
return plan
async def _project(cursor: Any, project_id: UUID | str) -> dict[str, Any]:
await cursor.execute(
"SELECT project_kind, project_year, storage_path, plan_base_json FROM projects"
" WHERE id = %s AND deleted_at IS NULL",
(str(project_id),),
)
project = await cursor.fetchone()
if not project:
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
if project["project_kind"] != "PLAN" or project["project_year"] is None:
raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
if not project["storage_path"]:
raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
return project
async def load_plan(project_id: UUID | str) -> tuple[Path, dict[str, Any]]:
"""(프로젝트 실경로, 설계 내용) — 파일 · 없으면 옛 DB 칸을 옮긴 것(쓰지 않음)."""
pid = str(project_id)
async with (
get_db_pool().acquire() as connection,
connection.cursor(aiomysql.DictCursor) as cursor,
):
project = await _project(cursor, pid)
root = Path(resolve_stored_project_path(project["storage_path"]))
plan = read_plan(root)
if plan is not None:
return root, plan
await cursor.execute("SELECT network_json FROM plan_networks WHERE project_id = %s", (pid,))
row = await cursor.fetchone()
await cursor.execute(
"SELECT year, start_x, start_y, end_x, end_y, epsg, route_json FROM plan_route_years"
" WHERE project_id = %s ORDER BY year",
(pid,),
)
years = list(await cursor.fetchall())
return root, migrate(project["plan_base_json"], row and row["network_json"], years)
def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
"""조각마다 {id · line · line_name · year · epsg · from_m · to_m · length_m · planned ·
stations}
— 연도 · 선 차례 · 선 위 거리 차례 · 틀리면 `NetworkError`."""
pieces = network.derive(plan)["pieces"] if plan["lines"] else []
names = dict(zip((line["id"] for line in plan["lines"]), line_names(plan["lines"])))
for piece in pieces:
piece.update(line_name=names.get(piece["line"], "갈래"), epsg=plan["epsg"])
pieces.sort(key=lambda p: p["year"]) # 같은 연도 안 = derive 차례(선 · 거리) 그대로
return pieces
async def year_status(project_id: UUID | str) -> dict[int, dict[str, Any]]:
"""연도 상태 줄 — {연도: {fingerprint, confirmed_at}}."""
async with (
get_db_pool().acquire() as connection,
connection.cursor(aiomysql.DictCursor) as cursor,
):
await cursor.execute(
"SELECT year, fingerprint, confirmed_at FROM plan_route_years WHERE project_id = %s",
(str(project_id),),
)
return {int(r["year"]): r for r in await cursor.fetchall()}
async def plan_state(project_id: UUID | str) -> dict[str, Any]:
"""{root, plan, units, years} — `years` = `year_rows` + `confirmed_at`(지문이 같을 때만) ·
단위마다 `confirmed_at` · 노선이 틀리면 409."""
root, plan = await load_plan(project_id)
try:
units = plan_units(plan)
except network.NetworkError as error:
raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
years = network.year_rows(units)
status = await year_status(project_id)
for year, row in years.items():
saved = status.get(year)
same = saved is not None and saved["fingerprint"] == row["fingerprint"]
row["confirmed_at"] = saved["confirmed_at"] if same else None
for unit in units:
unit["confirmed_at"] = years[unit["year"]]["confirmed_at"]
return {"root": root, "plan": plan, "units": units, "years": years}
D04_PlanConfirm/D04_PlanConfirm_Link.py
+43 -72
View File
@@ -1,13 +1,15 @@
"""D 예정노선 → 임도 연결 (요구 345 · 요구서 2-3 · 3-5 · PLAN 531-3).
"""D 예정노선 → 임도 연결 (요구 345 · 452 · 요구서 2-3 · 3-5 · PLAN 531-3 · 590-1).
- `POST /api/dashboard/plan-routes/{assignee_id}/link` — {mode: new | existing, project_id?}
new = 임도 프로젝트 새로(B02 저장소 함수) · existing = 고른 우리 회사 임도 프로젝트
- `GET /api/projects/{id}/plan-source` — 임도 프로젝트가 이어 받은 예정노선 · 업데이트 있음
연결 = 지정 행의 조각(옛 = 연도) shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에
연결 = 지정 행의 조각 shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에
원래 이름으로 + `input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길.
입력이 다 차면 B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림.
연결 시점 복사 · 원본이 바뀌면 알림만(내용 지문 비교 · 자동 반영 없음).
조각 노선 · 연도 · 확정 = 설계 내용 파일(`D03_PlanAnalysis_Store.plan_state`) —
업체 줄은 조각 번호만.
"""
from __future__ import annotations
@@ -48,15 +50,15 @@ from B03_FileInput.B03_FileInput_Schema import UploadedFileResult
from B03_FileInput.B03_FileInput_Service_WF1 import trigger_wf1_analysis_and_email
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 common_util.common_util_workflow_state import is_analysis_running
from config.config_db import get_db_pool
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
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"])
ROUTE_DIR = "B03_FileInput/input/shp"
DESIGN_SPEEDS = (20, 30, 40) # B05 RouteSolveRequest 가 받는 값
class LinkRequest(BaseModel):
@@ -101,53 +103,42 @@ 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"],
async def piece_route(project_id: str, piece_id: str | None) -> dict[str, Any]:
"""그 조각(파일) — {year, epsg, route_json: {planned, 누가 · 길이}, confirmed_at} ·
조각이 없으면 노선 · 연도 · 확정 None."""
state = await store.plan_state(project_id)
unit = next((u for u in state["units"] if u["id"] == piece_id), None)
if unit is None:
return {
"year": None,
"epsg": state["plan"]["epsg"],
"route_json": None,
"confirmed_at": None,
}
if body.get("planned")
else None
)
return out
return {
"year": unit["year"],
"epsg": unit["epsg"],
"route_json": {
"planned": unit["planned"],
"start_chainage_m": unit["from_m"],
"end_chainage_m": unit["to_m"],
"length_m": unit["length_m"],
},
"confirmed_at": unit["confirmed_at"],
}
async def load_assignee(assignee_id: int) -> dict[str, Any] | None:
"""지정 행 + 예정노선 프로젝트 칸 + 그 조각(옛 = 연도) 노선."""
"""지정 행 + 예정노선 프로젝트 칸 + 그 조각 노선 · 연도 · 확정(파일)."""
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"
"SELECT a.id, a.project_id, a.piece_id, a.company_id, a.linked_project_id,"
" p.name, p.region, p.road_type, p.route_start_m"
" FROM plan_route_assignees a"
" JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL"
f"{PIECE_JOIN}"
" WHERE a.id = %s",
(assignee_id,),
)
return unit_row(row) if row else None
return {**row, **await piece_route(str(row["project_id"]), row["piece_id"])} if row else None
async def load_road_project(project_id: str) -> dict[str, Any] | None:
@@ -157,18 +148,15 @@ async def load_road_project(project_id: str) -> dict[str, Any] | None:
)
async def record_link(
project_id: str, assignee: dict[str, Any], fingerprint: str | None, speed: Any
) -> None:
"""원본 칸 · 연결 표시 — 한 임도 프로젝트는 한 연도에만 이어짐(옛 연결은 끊음)."""
async def record_link(project_id: str, assignee: dict[str, Any], fingerprint: str | None) -> None:
"""원본 칸 · 연결 표시 — 한 임도 프로젝트는 한 조각에만 이어짐(옛 연결은 끊음)."""
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
await connection.begin()
try:
await cursor.execute(
"UPDATE projects SET source_project_id = %s, source_copied_at = NOW(),"
" source_fingerprint = %s, design_speed_kph = COALESCE(design_speed_kph, %s)"
" WHERE id = %s",
(assignee["project_id"], fingerprint, speed, project_id),
" source_fingerprint = %s WHERE id = %s",
(assignee["project_id"], fingerprint, project_id),
)
await cursor.execute(
"UPDATE plan_route_assignees SET linked_project_id = NULL"
@@ -186,7 +174,7 @@ async def record_link(
def year_files(assignee: dict[str, Any]) -> dict[str, bytes]:
"""그 조각(옛 = 연도) shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
"""그 조각 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
body = _parsed(assignee["route_json"]) or {}
route = YearRoute(
year=int(assignee["year"]),
@@ -306,26 +294,11 @@ async def _new_road_project(assignee: dict[str, Any], session: dict[str, Any]) -
return str(created["project_id"])
async def seed_design_speed(project_id: str, speed: Any) -> None:
"""새 임도만 — 예정노선 설계속도를 단계 2(B05) params `options.design_speed_kph` 에.
B05 페이지 설정 · 자동설계 체인(`B03_FileInput_Service_Chain`)이 읽는 자리 · 값만 넣음.
"""
if speed is None or int(speed) not in DESIGN_SPEEDS:
return
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
await connection.begin()
await merge_stage_params(
cursor, project_id, 2, {"options": {"design_speed_kph": int(speed)}}
)
await connection.commit()
@dashboard_router.post("/plan-routes/{assignee_id}/link", response_model=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="지정 업체 행을 찾을 수 없습니다.")
@@ -348,7 +321,6 @@ async def link_plan_route(
project_id = str(payload.project_id)
else:
project_id = await _new_road_project(assignee, session)
await seed_design_speed(project_id, assignee["design_speed_kph"])
try:
result = await import_route_files(
project_id, files, session, payload.confirm_replace, payload.las_free
@@ -358,7 +330,7 @@ async def link_plan_route(
except ValueError as exc:
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"])
await record_link(project_id, assignee, fingerprint)
return {
"status": "success",
"project_id": project_id,
@@ -372,19 +344,18 @@ async def link_plan_route(
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,"
f"{PIECE_COLUMNS}"
" y.epsg, y.route_json, y.confirmed_at"
"SELECT p.source_fingerprint, s.id AS source_id, s.name AS plan_name,"
" a.id AS assignee_id, a.piece_id"
" FROM projects p"
" JOIN projects s ON s.id = p.source_project_id AND s.project_kind = 'PLAN'"
" AND s.deleted_at IS NULL"
" JOIN plan_route_assignees a ON a.linked_project_id = p.id"
f"{PIECE_JOIN}"
" WHERE p.id = %s LIMIT 1",
(str(project_id),),
)
if not row:
return {"status": "success", "linked": False}
row = unit_row(row)
row = {**row, **await piece_route(str(row["source_id"]), row["piece_id"])}
return {
"status": "success",
"linked": True,
D04_PlanConfirm/D04_PlanConfirm_Repository.py
+19 -66
View File
@@ -1,8 +1,10 @@
"""D04 확정 · 업체 지정 DB (Raw SQL · 표는 db_management/020_project_groups.sql)."""
"""D04 확정 · 업체 지정 DB (Raw SQL · 표는 db_management/020_project_groups.sql · 023).
노선 · 조각 · 연도 = 설계 내용 파일(`D03_PlanAnalysis_Store`) · 여기는 업체 줄 · 연도 상태 줄만.
업체 줄 `year` 칸은 읽지 않음(연도 = 파일 조각 연도 · 590-3 에서 지움)."""
from __future__ import annotations
import json
from typing import Any
import aiomysql
@@ -23,24 +25,6 @@ async def get_plan_project(project_id: str) -> dict[str, Any] | None:
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
def line_names(lines: list[dict[str, Any]]) -> list[str]:
"""선 이름 — 갈래 없는 선 = 본선(둘 이상이면 본선 1 · 2 …) · 갈래 = 갈래 1 · 2 …(선 차례).
@@ -57,46 +41,14 @@ def line_names(lines: list[dict[str, Any]]) -> list[str]:
return out
async def list_pieces(project_id: str) -> list[dict[str, Any]]:
"""조각 행(연도 · 선 차례 · 선 위 거리 차례) + `line_name`(`line_names` = 정본 선 차례)."""
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 = (body or {}).get("lines", [])
order = {line["id"]: i for i, line in enumerate(lines)}
names = dict(zip(order, line_names(lines)))
for row in rows:
raw = row["route_json"]
row["route_json"] = json.loads(raw) if isinstance(raw, str | bytes) else raw
row["line_name"] = names.get(row["line_id"], "갈래")
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, piece_id, company_id, company_name, email, notified_at,"
" linked_project_id"
" FROM plan_route_assignees WHERE project_id = %s ORDER BY year, id",
"SELECT id, piece_id, company_id, company_name, email, notified_at, linked_project_id"
" FROM plan_route_assignees WHERE project_id = %s ORDER BY id",
(project_id,),
)
return await cursor.fetchall()
@@ -137,12 +89,11 @@ async def apply_assignee_changes(
for row in inserts:
await cursor.execute(
"INSERT INTO plan_route_assignees"
" (project_id, year, piece_id, company_id, company_name, email)"
" VALUES (%s,%s,%s,%s,%s,%s)",
" (project_id, piece_id, company_id, company_name, email)"
" VALUES (%s,%s,%s,%s,%s)",
(
project_id,
row["year"],
row.get("piece_id"),
row["piece_id"],
row["company_id"],
row["company_name"],
row["email"],
@@ -154,16 +105,18 @@ async def apply_assignee_changes(
raise
async def mark_confirmed(project_id: str, years: list[int]) -> None:
if not years:
async def mark_confirmed(project_id: str, fingerprints: dict[int, str]) -> None:
"""연도 확정 — {연도: 파일 지문} · 상태 줄이 없거나 지문이 낡았으면 새로 맞춤."""
if not fingerprints:
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),
)
for year, fingerprint in fingerprints.items():
await cursor.execute(
"INSERT INTO plan_route_years (project_id, year, fingerprint, confirmed_at)"
" VALUES (%s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE"
" fingerprint = VALUES(fingerprint), confirmed_at = NOW()",
(project_id, year, fingerprint),
)
await connection.commit()
D04_PlanConfirm/D04_PlanConfirm_Router.py
+70 -77
View File
@@ -1,4 +1,4 @@
"""D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · PLAN 531-1 · 531-2 · 563-2).
"""D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · 452 · PLAN 531-1 · 531-2 · 563-2 · 590-1).
- `GET /api/projects/{id}/plan/assignees` — 조각별 지정 업체(지정 회사 = 자기 행만)
- `PUT /api/projects/{id}/plan/assignees` — 조각별 지정 업체 저장(등록 회사 또는 업체명 + 메일)
@@ -6,12 +6,12 @@
- `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` 가 비면
옛 연도 지정 — 그 연도 첫 단위에 붙음.
단위 = 조각(설계 내용 파일 `D03_PlanAnalysis_Store` · 구역 × 갈래 · 요구 391). 확정은 연도 단위
(`plan_route_years.confirmed_at` · 지문이 파일과 같을 때만). 업체 줄 = 조각 번호 ·
연도는 파일 조각에서.
지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이
GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦.
GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 파일 노선을 서버가 다시 센 것으로만.
"""
from __future__ import annotations
@@ -30,6 +30,7 @@ from pydantic import BaseModel
from common_util.common_util_auth import is_forest_service, verify_session
from common_util.common_util_crs import tm_to_lonlat
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
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 (
@@ -46,8 +47,8 @@ router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"])
class AssigneeIn(BaseModel):
year: int
piece_id: str | None = None # 조각 — 비면 옛 연도 지정(조각 없는 프로젝트)
year: int | None = None # 안 씀 — 연도 = 파일 조각 연도
piece_id: str
company_id: int | None = None # 등록 회사 — 없으면 수기 업체
company_name: str | None = None
email: str | None = None
@@ -71,17 +72,17 @@ def _require_forest(session: dict[str, Any]) -> None:
def _key(row: dict[str, Any]) -> tuple:
"""같은 업체 판정 — 같은 조각(옛 = 연도)에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
unit = (row["year"], row.get("piece_id"))
"""같은 업체 판정 — 같은 조각에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
if row.get("company_id") is not None:
return (*unit, "c", row["company_id"])
return (*unit, "e", (row.get("email") or "").strip().lower())
return (row["piece_id"], "c", row["company_id"])
return (row["piece_id"], "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, 고칠 행, 넣을 행).
"""보낸 연도(`years` · 업체 줄 `year` = 파일 조각 연도)만 맞춰 바꿈 —
(지울 id, 고칠 행, 넣을 행).
같은 업체는 행을 살려 두어 발송 시각 · 연결한 프로젝트가 남는다.
"""
@@ -97,7 +98,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}
unit = {"piece_id": item.piece_id}
if item.company_id is not None:
company = await repo.get_company_contact(item.company_id)
if not company:
@@ -124,67 +125,47 @@ 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 = [
def route_units(pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""표 한 줄 단위 = 파일 조각(`store.plan_units` · 연도 차례)."""
return [
{
"piece_id": p["piece_id"],
"line": p["line_id"],
"piece_id": p["id"],
"line": p["line"],
"line_name": p.get("line_name"),
"year": int(p["year"]),
"epsg": int(p["epsg"]),
"planned": (p.get("route_json") or {}).get("planned") or [],
"planned": p["planned"],
"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"])),
"confirmed_at": p.get("confirmed_at"),
}
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 with_years(
assignees: list[dict[str, Any]], units: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""업체 줄에 연도 = 그 조각의 파일 연도(조각이 없으면 None) — 업체 줄 `year` 칸은 안 봄."""
years = {u["piece_id"]: u["year"] for u in units}
return [{**a, "year": years.get(a.get("piece_id"))} for a in assignees]
def unit_key(unit: dict[str, Any]) -> str | int:
"""내보내기 묶음 열쇠 — 조각 id · 옛 연도."""
return unit["piece_id"] or unit["year"]
"""내보내기 묶음 열쇠 — 조각 id."""
return unit["piece_id"]
def unit_label(unit: dict[str, Any]) -> str:
"""zip 폴더 · 첨부 이름 — 「2026」(옛) · 「2026_갈래1_p3」(조각)."""
if not unit["piece_id"]:
return str(unit["year"])
"""zip 폴더 · 첨부 이름 — 「2026_갈래1_p3」."""
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
"""업체 행의 단위 = 그 조각."""
return next((u for u in units if u["piece_id"] == assignee.get("piece_id")), None)
def _saved(unit: dict[str, Any]) -> bool:
@@ -204,9 +185,19 @@ def visible_to(
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)))
"""(연도 줄, 단위) — 파일에서 · 연도 줄 = {year, epsg, route_json, fingerprint,
confirmed_at}."""
state = await store.plan_state(project_id)
epsg = state["plan"]["epsg"]
rows = [
{"year": year, "epsg": epsg, "route_json": row["route"], **row}
for year, row in state["years"].items()
]
return rows, route_units(state["units"])
async def _assignees(project_id: UUID, units: list[dict[str, Any]]) -> list[dict[str, Any]]:
return with_years(await repo.list_assignees(str(project_id)), units)
@router.get("/{project_id}/plan/assignees", response_model=None)
@@ -214,7 +205,8 @@ async def get_plan_assignees(
project_id: UUID, session: dict = Depends(verify_session)
) -> dict[str, Any]:
await _plan_project(project_id)
_, assignees = visible_to(session, [], await repo.list_assignees(str(project_id)))
_, units = await _units(project_id)
_, assignees = visible_to(session, [], await _assignees(project_id, units))
return {"status": "success", "assignees": assignees}
@@ -224,26 +216,26 @@ async def put_plan_assignees(
) -> 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))}
_, units = await _units(project_id)
piece_years = {u["piece_id"]: u["year"] for u in units}
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)
years = {piece_years[a.piece_id] for a in request.assignees} | set(request.years or [])
unknown = sorted(years - set(piece_years.values()))
if unknown:
raise HTTPException(status_code=400, detail=f"예정노선 연도가 아닙니다: {unknown}")
wanted = [await _resolve_assignee(item) for item in request.assignees]
existing = await repo.list_assignees(str(project_id))
wanted = [
{**await _resolve_assignee(item), "year": piece_years[item.piece_id]}
for item in request.assignees
]
existing = await _assignees(project_id, units)
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))}
return {"status": "success", "assignees": await _assignees(project_id, units)}
def unit_routes(units: list[dict[str, Any]]) -> list[YearRoute]:
@@ -280,7 +272,7 @@ async def finalize_plan(
"""연도 노선 확정 → 지정 업체마다 그 조각 zip 을 첨부해 메일 → 보낸 곳에 발송 시각."""
_require_forest(session)
project = await _plan_project(project_id)
_, units = await _units(project_id)
rows, 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)
@@ -289,7 +281,8 @@ async def finalize_plan(
raise HTTPException(
status_code=409, detail=f"분석 [저장]이 안 된 연도입니다: {missing or '없음'}"
)
await repo.mark_confirmed(str(project_id), years)
fingerprints = {r["year"]: r["fingerprint"] for r in rows}
await repo.mark_confirmed(str(project_id), {y: fingerprints[y] for y in years})
# 옛 누가거리가 연도마다 이어지므로 저장된 전 단위로 한 번에 만들고, 보낼 때만 고름.
first = float(project.get("route_start_m") or 0.0)
@@ -297,8 +290,8 @@ async def finalize_plan(
build_export, project["name"], project["region"] or "", routes, first
)
results = []
for assignee in await repo.list_assignees(str(project_id)):
year = assignee["year"]
for assignee in await _assignees(project_id, units):
year = assignee["year"] # 파일 조각 연도(요구 452 ⑤)
if year not in years:
continue
if assignee["notified_at"] and not request.resend:
@@ -359,7 +352,7 @@ def year_summary(row: dict[str, Any]) -> dict[str, Any]:
def unit_summary(unit: dict[str, Any]) -> dict[str, Any]:
"""표 한 줄(조각 · 옛 연도) — 저장 전이면 saved False · label 은 조각만(「본선 0+0~12+10」)."""
"""표 한 줄(조각) — 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"]:
@@ -388,10 +381,10 @@ def piece_label(line_name: str, from_m: float, to_m: float) -> str:
async def get_plan_summary(
project_id: UUID, session: dict = Depends(verify_session)
) -> dict[str, Any]:
"""`years` = 연도 행(확정 고르기) · `pieces` = 표 한 줄 단위(조각 · 옛 연도)."""
"""`years` = 연도 줄(확정 고르기) · `pieces` = 표 한 줄 단위(조각)."""
project = await _plan_project(project_id)
rows, units = await _units(project_id)
units, assignees = visible_to(session, units, await repo.list_assignees(str(project_id)))
units, assignees = visible_to(session, units, await _assignees(project_id, units))
own_years = {u["year"] for u in units}
years = [year_summary(row) for row in rows if int(row["year"]) in own_years]
return {
@@ -404,7 +397,7 @@ async def get_plan_summary(
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 unit in units:
@@ -427,7 +420,7 @@ async def export_plan(project_id: UUID, session: dict = Depends(verify_session))
forest = is_forest_service(session)
picked = [u for u in units if _saved(u)]
if not forest:
own, _ = visible_to(session, picked, await repo.list_assignees(str(project_id)))
own, _ = visible_to(session, picked, await _assignees(project_id, units))
picked = [u for u in own if u["confirmed_at"]]
if not picked:
raise HTTPException(status_code=403, detail="받을 수 있는 조각이 없습니다.")
db_management/023_plan_file.sql
+13
View File
@@ -0,0 +1,13 @@
-- 023_plan_file.sql
-- 예정노선 설계 내용 = 프로젝트 폴더 한 파일 `D03_PlanAnalysis/plan_network.json` (요구 452 · 453 · PLAN 590-1)
-- 업체 줄 = 조각 번호만 · 연도는 파일 조각에서(새 줄은 year 를 비움) · 연도 줄 = 지문 · 확정 상태만
-- 옛 칸(plan_networks · plan_route_pieces · plan_route_years 노선 칸 · projects.plan_base_json · 업체 year)은
-- 남김 — 590-3(전역 깃 싱크 뒤)에서 지움 · 다시 돌려도 같음
USE aislo_db;
ALTER TABLE plan_route_assignees
MODIFY COLUMN year INT NULL COMMENT '안 씀 — 연도 = 설계 내용 파일 조각 연도(590-3 에서 지움)';
ALTER TABLE plan_route_years
COMMENT = '예정노선 연도 상태(지문 · 확정) — 노선 칸은 안 씀(590-3 에서 지움)';
resources/tester/helper_plan_file.py
+56
View File
@@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
"""시험 도움 — 예정노선 설계 내용(파일 꼴 · PLAN 590-1)을 메모리로 꽂음."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
def plan_of(lines: list, marks: list, pieces: list, epsg: int = 5186, points=()) -> dict:
return {
**store.empty_plan(epsg),
"points": list(points),
"lines": lines,
"marks": marks,
"pieces": pieces,
}
def use_plan(monkeypatch: Any, plan: dict, confirmed: dict | None = None) -> dict:
"""`store.load_plan` · `store.year_status` 를 갈아 끼움 — `confirmed` = {연도: 확정 시각}
(지문은 파일과 같게) · 돌려준 dict 의 plan · confirmed 를 고치면 다음 읽기에 반영."""
state = {"plan": plan, "confirmed": dict(confirmed or {})}
async def load(_pid):
return Path("."), state["plan"]
async def status(_pid):
years = network.year_rows(store.plan_units(state["plan"]))
return {
y: {"fingerprint": years[y]["fingerprint"], "confirmed_at": at}
for y, at in state["confirmed"].items()
if y in years
}
monkeypatch.setattr(store, "load_plan", load)
monkeypatch.setattr(store, "year_status", status)
return state
def year_line_plan(parts: dict[int, list]) -> dict:
"""본선 하나 = 연도 꺾은선 이음(앞 끝 = 뒤 시작) · 연도 경계 마크 · 조각 `p26` · `p27` …"""
vertices: list = []
pieces = []
for year, points in sorted(parts.items()):
start = network.draw.line_length(vertices)
vertices += [list(p) for p in (points[1:] if vertices else points)]
end = network.draw.line_length(vertices)
pieces.append(
{"id": f"p{year % 100}", "line": "main", "from_m": start, "to_m": end, "year": year}
)
marks = [{"line": "main", "at_m": p["to_m"]} for p in pieces[:-1]]
return plan_of([{"id": "main", "vertices": vertices, "branch": None}], marks, pieces)
resources/tester/test_529_3_plan_points.py
+36 -22
View File
@@ -1,5 +1,6 @@
"""D02 기준 지정(PLAN 529-3 · 560-2 · 579-1 · 585-1) — 경위도 ↔ TM 변환 · 지점 목록 하나 정본 API ·
옛 시점 · 종점 꼴 이어 붙임 · 옛 한 점 꼴 감쌈 · 옛 연도 행 접기 · 저장된 좌표계 그대로.
"""D02 기준 지정(PLAN 529-3 · 560-2 · 579-1 · 585-1 · 590-1) — 경위도 ↔ TM 변환 · 지점 목록 API ·
지점 = 설계 내용 파일 `points` · 옛 DB 칸(시점 · 종점 꼴 · 한 점 꼴 · 연도 행 접기)은 열 때 옮김 ·
저장된 좌표계 그대로.
./venv/Scripts/python.exe -m pytest resources/tester/test_529_3_plan_points.py -q
"""
@@ -13,6 +14,7 @@ from fastapi.testclient import TestClient
from common_util.common_util_auth import require_company
from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat
from D02_PlanBase import D02_PlanBase_Router as router_mod
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
URL = f"/api/projects/{PID}/plan/points"
@@ -57,21 +59,20 @@ A, B, C = (
@pytest.fixture
def api(monkeypatch):
def api(monkeypatch, tmp_path):
"""파일 = tmp_path 의 진짜 파일 · 없으면 옛 DB 칸(`base_json` · 연도 행)을 옮김."""
state = {
"project": {"project_kind": "PLAN", "project_year": 2026, "project_year_end": 2027},
"rows": {},
"base_json": None,
"rows": [],
"root": tmp_path,
"session": {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1},
}
async def load(project_id):
return state["project"], {y: dict(r) for y, r in state["rows"].items()}
plan = store.read_plan(tmp_path)
return tmp_path, plan or store.migrate(state["base_json"], None, state["rows"])
async def store_base(project_id, base):
state["project"] = {**state["project"], "plan_base_json": json.dumps(base)}
monkeypatch.setattr(router_mod, "_load", load)
monkeypatch.setattr(router_mod, "_store_base", store_base)
monkeypatch.setattr(store, "load_plan", load)
app = FastAPI()
app.include_router(router_mod.router)
app.dependency_overrides[require_company] = lambda: state["session"]
@@ -90,9 +91,9 @@ def test_put_lists(api) -> None:
)
end = data["points"][2]
assert abs(end["lon"] - C["lon"]) < 1e-8 and abs(end["lat"] - C["lat"]) < 1e-8
stored = json.loads(api.state["project"]["plan_base_json"])
assert set(stored) == {"epsg", "points"} and set(stored["points"][1]) == {"x", "y"}
assert api.state["rows"] == {} # 연도 행은 D02 가 안 만듦
stored = json.loads(store.plan_path(api.state["root"]).read_text("utf-8"))
assert stored["version"] == 1 and stored["epsg"] == 5186 and stored["lines"] == []
assert set(stored["points"][1]) == {"x", "y"}
def test_put_partial_and_empty(api) -> None:
@@ -115,7 +116,7 @@ def test_get_joins_old_start_end_lists(api) -> None:
a, b, c = (lonlat_to_tm(p["lon"], p["lat"], 5186) for p in (A, B, C))
old = {"epsg": 5186, "starts": [{"x": a[0], "y": a[1]}, {"x": b[0], "y": b[1]}],
"ends": [{"x": c[0], "y": c[1]}]} # fmt: skip
api.state["project"]["plan_base_json"] = json.dumps(old)
api.state["base_json"] = json.dumps(old)
data = api.get(URL).json()
assert [p["x"] for p in data["points"]] == pytest.approx([a[0], b[0], c[0]]) # 시점 뒤 종점
@@ -123,11 +124,11 @@ def test_get_joins_old_start_end_lists(api) -> None:
def test_get_wraps_old_single_points(api) -> None:
a, c = (lonlat_to_tm(p["lon"], p["lat"], 5186) for p in (A, C))
old = {"epsg": 5186, "start": {"x": a[0], "y": a[1]}, "end": {"x": c[0], "y": c[1]}}
api.state["project"]["plan_base_json"] = json.dumps(old)
api.state["base_json"] = json.dumps(old)
data = api.get(URL).json()
assert [p["x"] for p in data["points"]] == pytest.approx([a[0], c[0]])
old["end"] = None
api.state["project"]["plan_base_json"] = json.dumps(old)
api.state["base_json"] = json.dumps(old)
assert [p["x"] for p in api.get(URL).json()["points"]] == pytest.approx([a[0]])
@@ -142,11 +143,13 @@ def test_get_folds_old_year_rows(api) -> None:
epsg = 5186
a, b, c = (lonlat_to_tm(p["lon"], p["lat"], epsg) for p in (A, B, C))
for year, s, e in ((2026, a, b), (2027, b, c)):
api.state["rows"][year] = {
"year": year,
**{"start_x": s[0], "start_y": s[1], "end_x": e[0], "end_y": e[1]},
"epsg": epsg,
}
api.state["rows"].append(
{
"year": year,
**{"start_x": s[0], "start_y": s[1], "end_x": e[0], "end_y": e[1]},
"epsg": epsg,
}
)
data = api.get(URL).json()
assert data["epsg"] == epsg
assert [p["x"] for p in data["points"]] == pytest.approx([a[0], c[0]])
@@ -154,3 +157,14 @@ def test_get_folds_old_year_rows(api) -> None:
def test_get_empty_new_project(api) -> None:
assert api.get(URL).json() == {"status": "success", "epsg": None, "points": []}
def test_put_keeps_line_epsg(api) -> None:
"""선이 있으면 지점을 다 지워도 좌표계 그대로(파일 머리 한 곳 · 선 좌표가 그 좌표계)."""
api.put(URL, json={"points": [A]})
path = store.plan_path(api.state["root"])
plan = json.loads(path.read_text("utf-8"))
plan["lines"] = [{"id": "main", "vertices": [[0.0, 0.0], [1.0, 0.0]], "branch": None}]
path.write_text(json.dumps(plan), "utf-8")
assert api.put(URL, json={}).json()["points"] == []
assert json.loads(path.read_text("utf-8"))["epsg"] == 5186
resources/tester/test_530_2_plan_analyze.py
+53 -42
View File
@@ -1,5 +1,7 @@
"""D 예정노선 2단계 노선 정본(PLAN 562-2) — 손 그은 선 · 갈래 · 마크 · 조각 → 서버가 다시 셈 ·
연도 행 · 옛 연도 노선을 선으로 · GET · PUT(산림청만 · 틀린 꼴 거절 · 업체 지정된 조각 지움 막음).
"""D 예정노선 2단계 노선 정본(PLAN 562-2 · 590-1) — 손 그은 선 · 갈래 · 마크 · 조각 →
서버가 다시 셈 · 연도 묶음 · 옛 연도 노선을 선으로 · GET · PUT(산림청만 · 틀린 꼴 거절 ·
업체 지정된 조각 지움 막음) ·
[저장] = 설계 내용 파일 + 연도 상태 줄만(옛 DB 설계 칸 안 씀) · 파일 쓰기 실패 = DB 되돌림.
옛 추천 엔진 시험(합성 지형 갈래 넷 · 저장본 · 나란히 셈)은 엔진과 함께 걷음.
@@ -14,6 +16,7 @@ from fastapi.testclient import TestClient
from common_util.common_util_auth import require_company
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
MAIN = [[0.0, 0.0], [30.0, 40.0], [30.0, 100.0], [90.0, 180.0]] # 50 + 60 + 100 = 210 m
@@ -174,22 +177,27 @@ class _Connection:
class _Db:
def __init__(self):
self.log: list = []
self.answers: dict = {"FROM projects": [{"project_year": 2026, "plan_base_json": "{}"}]}
self.answers: dict = {"FROM projects": [(2026,)]}
def acquire(self):
return _Connection(self)
BASE = {"epsg": 5186, "points": [{"x": 0.0, "y": 0.0}, {"x": 90.0, "y": 180.0}]}
@pytest.fixture
def client(monkeypatch):
def client(monkeypatch, tmp_path):
"""파일 = tmp_path 진짜 파일 · 없으면 옛 DB 칸(`old`)을 옮김 · DB = 가짜 기록."""
db = _Db()
monkeypatch.setattr(net, "get_db_pool", lambda: db)
base = {"epsg": 5186, "points": [{"x": 0.0, "y": 0.0}, {"x": 90.0, "y": 180.0}]}
old = {"base": json.dumps(BASE), "network": None, "rows": []}
async def load_base(project_id):
return base
async def load(project_id):
plan = store.read_plan(tmp_path)
return tmp_path, plan or store.migrate(old["base"], old["network"], old["rows"])
monkeypatch.setattr(net.plan_base, "load_base", load_base)
monkeypatch.setattr(store, "load_plan", load)
app = FastAPI()
app.include_router(net.router)
session = {"role": "USER", "email": "a@forest.go.kr"}
@@ -197,62 +205,66 @@ def client(monkeypatch):
test = TestClient(app)
test.db = db
test.session = session
test.base = base
test.old = old
test.file = store.plan_path(tmp_path)
return test
URL = f"/api/projects/{PID}/plan/network"
def test_get_saved_legacy_or_empty(client) -> None:
def test_get_migrates_old_db_or_empty(client) -> None:
assert client.get(URL).json()["network"] is None
client.db.answers["FROM plan_route_years"] = [
client.old["rows"] = [
{"year": 2026, "epsg": 5186, "route_json": '{"planned": [[0, 0], [30, 40]]}'}
]
got = client.get(URL).json()
assert got["legacy"] is True and got["derived"]["pieces"][0]["length_m"] == 50
client.db.answers["FROM plan_networks"] = [{"network_json": json.dumps(_network())}]
assert got["legacy"] is False and got["derived"]["pieces"][0]["length_m"] == 50
client.old["network"] = json.dumps(_network())
got = client.get(URL).json()
assert got["legacy"] is False and len(got["derived"]["pieces"]) == 3
assert got["base"]["points"][1] == {"x": 90.0, "y": 180.0} and got["year_start"] == 2026
assert len(got["derived"]["pieces"]) == 3 and got["network"]["epsg"] == 5186
assert got["base"]["points"][1]["x"] == 90.0 and got["year_start"] == 2026
assert not client.file.exists() # 열기만 = 쓰지 않음
def test_put_stores_network_pieces_years(client) -> None:
def test_put_writes_file_and_year_status_only(client) -> None:
got = client.put(URL, json=_network())
assert got.status_code == 200 and got.json()["derived"]["pieces"][1]["from_m"] == 75.0
assert got.json()["year_start"] == 2026
saved = json.loads(client.file.read_text("utf-8"))
assert saved["version"] == 1 and saved["epsg"] == 5186
assert saved["points"] == BASE["points"] # 옛 지점 칸 → 파일로 옮김
assert [p["id"] for p in saved["pieces"]] == ["p1", "p2", "p3"]
assert "length_m" not in json.dumps(saved) # 파생 저장 안 함
log = client.db.log
assert log[-1][0] == "COMMIT"
pieces = [a for s, a in log if s.startswith("INSERT INTO plan_route_pieces")]
assert [(a[1], a[2], a[3]) for a in pieces] == [
("p1", "main", 2026),
("p2", "main", 2027),
("p3", "b1", 2026),
]
sqls = [s for s, _ in log]
for column in ("plan_route_pieces", "plan_networks", "plan_base_json", "route_json"):
assert not [s for s in sqls if column in s and not s.startswith("SELECT")], column
years = [a for s, a in log if s.startswith("INSERT INTO plan_route_years")]
assert [a[1] for a in years] == [2026, 2027]
assert json.loads(years[0][7])["pieces"][1]["id"] == "p3"
assert [a[1] for a in years] == [2026, 2027] and all(len(a[2]) == 40 for a in years)
assert ("UPDATE projects SET project_year_end = %s WHERE id = %s", (2027, PID)) in log
stale = [a for s, a in log if "year NOT IN" in s]
assert stale == [(PID, 2026, 2027)]
assert not [s for s, _ in log if "plan_base_json = %s" in s] # 정본 칸 있음 = 안 굳힘
assert stale == [(PID, 2026, 2027)] # 없는 연도 줄 지움(빈 줄 남기지 않음)
def test_put_freezes_folded_base_first(client) -> None:
client.db.answers["FROM projects"] = [{"project_year": 2026, "plan_base_json": None}]
assert client.put(URL, json=_network()).status_code == 200
sqls = [(s, a) for s, a in client.db.log if s.startswith("UPDATE projects SET plan_base_json")]
assert len(sqls) == 1 and json.loads(sqls[0][1][0]) == client.base
first_year = next(i for i, (s, _) in enumerate(client.db.log) if "plan_route_years" in s)
assert client.db.log.index(sqls[0]) < first_year
def test_put_reset_clears_year_routes(client) -> None:
def test_put_reset_drops_year_rows(client) -> None:
got = client.put(URL, json={"epsg": 5186, "lines": [], "marks": [], "pieces": []})
assert got.status_code == 200 and got.json()["derived"] == {"lines": [], "pieces": []}
sqls = [s for s, _ in client.db.log]
assert any(s.startswith("UPDATE plan_route_years SET route_json = NULL") for s in sqls)
assert "DELETE FROM plan_route_years WHERE project_id = %s" in sqls
assert not any("project_year_end" in s for s in sqls)
assert json.loads(client.file.read_text("utf-8"))["lines"] == []
def test_put_file_failure_rolls_back(client, monkeypatch) -> None:
def broken(root, plan):
raise OSError("디스크")
monkeypatch.setattr(store, "save_plan", broken)
with pytest.raises(OSError):
client.put(URL, json=_network())
assert ("ROLLBACK", None) in client.db.log and ("COMMIT", None) not in client.db.log
def test_put_rejects(client) -> None:
@@ -261,7 +273,7 @@ def test_put_rejects(client) -> None:
assert (
client.put(URL, json=_network(lines=[{"id": "a b", "vertices": MAIN}])).status_code == 422
)
# 업체가 지정된 조각 · 옛 연도를 지우면 막음
# 업체가 지정된 조각을 지우면 막음 · 파일 안 씀
client.db.answers["piece_id IS NOT NULL"] = [("p2",)]
one = _network(
lines=[{"id": "main", "vertices": MAIN}],
@@ -269,10 +281,9 @@ def test_put_rejects(client) -> None:
pieces=[{"id": "p1", "line": "main", "from_m": 0, "to_m": 210, "year": 2026}],
)
assert client.put(URL, json=one).status_code == 409
client.db.answers["piece_id IS NOT NULL"] = []
client.db.answers["piece_id IS NULL"] = [(2027,)]
assert client.put(URL, json=one).status_code == 409
assert not client.file.exists() and ("ROLLBACK", None) in client.db.log
assert not [s for s, _ in client.db.log if s.startswith("INSERT")]
assert client.put(URL, json=_network(epsg=5187)).status_code == 400 # 기준 지정과 다른 좌표계
client.db.answers["piece_id IS NOT NULL"] = []
assert client.put(URL, json=_network(epsg=5187)).status_code == 400 # 파일 좌표계와 다름
client.session["email"] = "a@gmail.com"
assert client.put(URL, json=_network()).status_code == 403
resources/tester/test_531_3_plan_link.py
+64 -117
View File
@@ -26,6 +26,8 @@ ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from helper_plan_file import use_plan, year_line_plan # noqa: E402
from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata # noqa: E402
from B03_FileInput.B03_FileInput_Router_Helpers import OutputsWouldBeDiscarded # noqa: E402
from common_util.common_util_route_geometry import find_planned_route_file # noqa: E402
@@ -61,12 +63,12 @@ def _assignee(**kw):
"id": 7,
"project_id": PLAN_ID,
"year": 2026,
"piece_id": "p26",
"company_id": 5,
"linked_project_id": None,
"name": "시험 예정노선",
"region": "일월.용화",
"road_type": "work",
"design_speed_kph": 20,
"route_start_m": 0.0,
"epsg": 5186,
"route_json": json.dumps(_route(P2026)),
@@ -158,8 +160,8 @@ def env(monkeypatch, tmp_path):
raise OutputsWouldBeDiscarded(["노선 · 종단 설계"])
return 99
async def record(project_id, assignee, fingerprint, speed):
state["links"].append((project_id, assignee["id"], fingerprint, speed))
async def record(project_id, assignee, fingerprint):
state["links"].append((project_id, assignee["id"], fingerprint))
monkeypatch.setattr(link, "is_analysis_running", false)
monkeypatch.setattr(link, "get_project_storage_relative_path", storage)
@@ -211,7 +213,7 @@ def test_link_writes_route_set_like_upload(env, monkeypatch):
stage_meta = json.loads((env["root"] / "B03_FileInput/metadata.json").read_text("utf-8"))
assert len(stage_meta["files"]) == 5
fp = link.route_fingerprint(_assignee()["route_json"])
assert env["links"] == [(ROAD_ID, 7, fp, 20)] and len(fp) == 40
assert env["links"] == [(ROAD_ID, 7, fp)] and len(fp) == 40
assert any("SUPERSEDED" in sql for sql, _ in env["log"] if isinstance(sql, str))
@@ -278,50 +280,6 @@ def test_new_mode_name_year_once(env, monkeypatch):
assert made["name"] == "2026 홍천 노선 계획"
def _stage2_writes(log):
return [
args
for sql, args in log
if isinstance(sql, str) and "UPDATE project_workflow_stages" in sql
]
def test_new_mode_seeds_b05_design_speed(env, monkeypatch):
"""531-4 — 새 임도 = 예정노선 설계속도가 단계 2 params options 에 · B05 가 그 값을 씀."""
from B05_Profile.B05_Profile_PageSettings import grade_options_of
_use(monkeypatch, _assignee(design_speed_kph=30, road_type="main"))
async def create_project(**kw):
return {"project_id": ROAD_ID}
monkeypatch.setattr(link, "create_project", create_project)
_link(mode="new", project_id=None)
(args,) = _stage2_writes(env["log"])
params = json.loads(args[0])
assert args[1:] == (ROAD_ID, 2) and params == {"options": {"design_speed_kph": 30}}
# 자동설계 체인이 grade_class trunk(기본)로 params 를 다시 써도 이 값을 이어 넘김
chained = {"options": {"grade_class": "trunk", "design_speed_kph": 30}}
assert grade_options_of("main", chained).design_speed_kph == 30
chain = (ROOT / "B03_FileInput/B03_FileInput_Service_Chain.py").read_text("utf-8")
assert "'$.options.design_speed_kph'" in chain and "design_speed_kph=int(seeded_speed)" in chain
def test_existing_and_odd_speed_leave_stage_params(env, monkeypatch):
"""기존 프로젝트는 사용자 값 보존 · 20/30/40 밖 값은 안 넣음."""
_use(monkeypatch, _assignee(design_speed_kph=30), ROAD)
_link()
assert _stage2_writes(env["log"]) == []
async def create_project(**kw):
return {"project_id": ROAD_ID}
monkeypatch.setattr(link, "create_project", create_project)
_use(monkeypatch, _assignee(design_speed_kph=25))
_link(mode="new", project_id=None)
assert _stage2_writes(env["log"]) == []
def test_fingerprint_update_and_revert():
linked = link.route_fingerprint(_route(P2026))
moved = [p[:] for p in P2026]
@@ -338,72 +296,59 @@ def test_fingerprint_update_and_revert():
assert link.update_available(json.dumps(_route(moved)), linked, False) is False
def test_assigned_list_flags_update(monkeypatch):
linked = link.route_fingerprint(_route(P2026))
moved = [[x + 1.0, y] for x, y in P2026]
@pytest.fixture
def plan_file(monkeypatch):
"""설계 내용 파일 — 본선 p26(2026 · 확정) · p27(2027 · 확정 전)."""
return use_plan(monkeypatch, year_line_plan({2026: P2026, 2027: P2027}), {2026: "x"})
def test_assigned_list_flags_update(plan_file, monkeypatch):
moved = link.route_fingerprint(_route([[x + 1.0, y] for x, y in P2026]))
base = {
"project_id": PLAN_ID,
"project_name": "시험",
"linked_project_id": ROAD_ID,
"linked_project_name": "임도",
"source_fingerprint": moved,
}
unlinked = {"linked_project_id": None, "linked_project_name": None, "source_fingerprint": None}
rows = [
{
"assignee_id": 7,
"project_id": PLAN_ID,
"project_name": "시험",
"year": 2026,
"linked_project_id": ROAD_ID,
"linked_project_name": "임도",
"source_fingerprint": linked,
"route_json": json.dumps(_route(moved)),
"confirmed_at": "2026-10-25 12:00:00",
},
{
"assignee_id": 9,
"project_id": PLAN_ID,
"project_name": "시험",
"year": 2028,
"linked_project_id": ROAD_ID,
"linked_project_name": "임도",
"source_fingerprint": linked,
"route_json": json.dumps(_route(moved)),
"confirmed_at": None,
},
{
"assignee_id": 8,
"project_id": PLAN_ID,
"project_name": "시험",
"year": 2027,
"linked_project_id": None,
"linked_project_name": None,
"source_fingerprint": None,
"route_json": json.dumps(_route(P2027)),
"confirmed_at": "2026-10-25 12:00:00",
},
{**base, "assignee_id": 7, "piece_id": "p26"}, # 연결 뒤 노선 바뀜 → 업데이트
{**base, "assignee_id": 9, "piece_id": "p27"}, # 확정 전 → 알림 없음
{**base, "assignee_id": 8, "piece_id": "p26", **unlinked},
{**base, "assignee_id": 10, "piece_id": "p27", **unlinked}, # 확정 전 · 연결 없음 → 뺌
]
log = []
monkeypatch.setattr(d01, "get_db_pool", lambda: FakeConn(log, rows))
monkeypatch.setattr(d01, "get_db_pool", lambda: FakeConn([], rows))
out = asyncio.run(d01.assigned_plan_routes(SESSION))["routes"]
assert [r["update_available"] for r in out] == [True, False, False]
assert "route_json" not in out[0] and out[0]["assignee_id"] == 7
assert "confirmed_at IS NOT NULL" in log[0][0]
assert [(r["assignee_id"], r["year"], r["update_available"]) for r in out] == [
(7, 2026, True),
(8, 2026, False),
(9, 2027, False),
]
assert "route_json" not in out[0] and "source_fingerprint" not in out[0]
def test_plan_source(monkeypatch):
linked = link.route_fingerprint(_route(P2026))
def test_plan_source(plan_file, monkeypatch):
row = {
"source_fingerprint": linked,
"source_fingerprint": link.route_fingerprint(_route(P2026)),
"source_id": PLAN_ID,
"plan_name": "시험",
"assignee_id": 7,
"year": 2026,
"route_json": json.dumps(_route(P2026)),
"confirmed_at": "2026-10-25 12:00:00",
"piece_id": "p26",
}
sqls = []
async def one(sql, args):
sqls.append(sql)
return row
monkeypatch.setattr(link, "_fetch_one", one)
out = asyncio.run(link.plan_source(ROAD_ID))
assert out["linked"] and out["update_available"] is False and out["year"] == 2026
row["route_json"] = json.dumps(_route([[x, y + 2.0] for x, y in P2026]))
assert "s.deleted_at IS NULL" in sqls[0] # 지운 예정노선 띠 안 남김
row["source_fingerprint"] = link.route_fingerprint(_route([[x, y + 2.0] for x, y in P2026]))
assert asyncio.run(link.plan_source(ROAD_ID))["update_available"] is True
row["confirmed_at"] = None # 확정 풀림 → 알림 없음
plan_file["confirmed"].clear() # 확정 풀림 → 알림 없음
assert asyncio.run(link.plan_source(ROAD_ID))["update_available"] is False
async def none(sql, args):
@@ -414,7 +359,7 @@ def test_plan_source(monkeypatch):
@pytest.fixture
def plan_repo(monkeypatch):
def plan_repo(plan_file, monkeypatch):
async def project(_pid):
return {
"id": PLAN_ID,
@@ -424,27 +369,28 @@ def plan_repo(monkeypatch):
"route_start_m": 0.0,
}
async def years(_pid):
async def assignees(_pid): # DB 업체 줄 — 연도 칸 없음(파일 조각에서)
return [
{"year": 2026, "epsg": 5186, "route_json": _route(P2026), "confirmed_at": "x"},
{"year": 2027, "epsg": 5186, "route_json": _route(P2027), "confirmed_at": None},
{"id": 7, "piece_id": "p26", "company_id": 5, "company_name": "가", "email": "a@ga.kr"},
{"id": 8, "piece_id": "p27", "company_id": 5, "company_name": "가", "email": "a@ga.kr"},
{
"id": 9,
"piece_id": "p27",
"company_id": None,
"company_name": "나",
"email": "b@na.kr",
},
{
"id": 10,
"piece_id": "없음",
"company_id": 6,
"company_name": "다",
"email": "c@da.kr",
},
]
async def assignees(_pid):
return [
{"id": 7, "year": 2026, "company_id": 5, "company_name": "가", "email": "a@ga.kr"},
{"id": 8, "year": 2027, "company_id": 5, "company_name": "가", "email": "a@ga.kr"},
{"id": 9, "year": 2027, "company_id": None, "company_name": "나", "email": "b@na.kr"},
{"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):
@@ -454,13 +400,13 @@ def _zip_names(response):
def test_export_forest_gets_all(plan_repo):
forest = {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1}
names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, forest)))
assert {n.split("/")[0] for n in names} == {"2026", "2027", "전체"}
assert {n.split("/")[0] for n in names} == {"2026_본선_p26", "2027_본선_p27", "전체"}
assert sum(n.endswith(".shp") for n in names) == 3
def test_export_shared_gets_own_confirmed_year(plan_repo):
names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, SESSION)))
assert {n.split("/")[0] for n in names} == {"2026"} # 2027 = 확정 전 · 전체 없음
assert {n.split("/")[0] for n in names} == {"2026_본선_p26"} # 2027 = 확정 전 · 전체 없음
with pytest.raises(HTTPException) as err:
asyncio.run(d04.export_plan(PLAN_ID, {**SESSION, "company_id": 9}))
assert err.value.status_code == 403
@@ -475,11 +421,12 @@ def test_summary_year_rows(plan_repo):
first["length_m"]
)
assert out["years"][1]["confirmed_at"] is None and len(out["assignees"]) == 4
assert [a["year"] for a in out["assignees"]] == [2026, 2027, 2027, None]
def test_shared_company_sees_only_own_firm(plan_repo):
"""지정 회사 = 자기 연도 줄 · 자기 행만 — 남의 업체 이름 · 메일은 응답에 없음(요구 351)."""
other = {"role": "USER", "email": "x@ga.kr", "company_id": 6} # 2028 만(노선 없음)
other = {"role": "USER", "email": "x@ga.kr", "company_id": 6} # 없는 조각만
for session, years, ids in ((SESSION, [2026, 2027], [7, 8]), (other, [], [10])):
out = asyncio.run(d04.get_plan_summary(PLAN_ID, session))
assert [y["year"] for y in out["years"]] == years
resources/tester/test_531_d04_export_finalize.py
+20 -39
View File
@@ -26,6 +26,8 @@ ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from helper_plan_file import use_plan, year_line_plan # noqa: E402
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
@@ -244,28 +246,14 @@ FOREST = {"user_id": 1, "email": "a@korea.go.kr", "role": "MEMBER", "company_id"
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)],
"file": use_plan(monkeypatch, year_line_plan({2026: Y2026, 2027: Y2027})),
"assignees": [
{
"id": 1,
"year": 2026,
"piece_id": "p26",
"company_id": None,
"company_name": "가나다건설",
"email": "ga@x.kr",
@@ -274,7 +262,7 @@ def fake_db(monkeypatch):
},
{
"id": 2,
"year": 2026,
"piece_id": "p26",
"company_id": 9,
"company_name": "라마",
"email": "ra@x.kr",
@@ -283,7 +271,7 @@ def fake_db(monkeypatch):
},
{
"id": 3,
"year": 2027,
"piece_id": "p27",
"company_id": None,
"company_name": "메일없음",
"email": None,
@@ -306,9 +294,6 @@ def fake_db(monkeypatch):
"route_start_m": 0.0,
}
async def years(_pid):
return state["years"]
async def assignees(_pid):
return state["assignees"]
@@ -328,13 +313,8 @@ 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)
@@ -367,10 +347,12 @@ def test_finalize_sends_zip_per_assignee(fake_db):
assert fake_db["notified"] == [1]
((to, subject, attachments),) = fake_db["mails"]
assert to == "ga@x.kr" and "2026" in subject
assert set(fake_db["confirmed"]) == {2026, 2027} # 연도 → 파일 지문
name, blob, mime = attachments[0]
assert name == "시험_예정노선_2026.zip" and mime == "application/zip" and len(blob) > 1000
assert name == "시험_예정노선_2026_본선_p26.zip" and mime == "application/zip"
assert len(blob) > 1000
with zipfile.ZipFile(io.BytesIO(blob)) as archive:
assert any(n.endswith("_2026.shp") for n in archive.namelist())
assert any(n.endswith("_2026_본선_p26.shp") for n in archive.namelist())
assert not any("_2027" in n for n in archive.namelist()) # 다른 연도는 안 샘
# 다시 보내기
@@ -380,9 +362,8 @@ def test_finalize_sends_zip_per_assignee(fake_db):
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))
with pytest.raises(HTTPException) as err: # 파일에 없는 연도
asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(years=[2030]), FOREST))
assert err.value.status_code == 409
assert fake_db["confirmed"] == []
@@ -394,10 +375,10 @@ def test_put_assignees_validation_and_diff(fake_db):
)
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),
router.AssigneeIn(piece_id="p26", company_name="수기", email="메일아님"),
router.AssigneeIn(piece_id="p26", email="x@y.kr"),
router.AssigneeIn(piece_id="없음", company_name="없는조각", email="x@y.kr"),
router.AssigneeIn(piece_id="p26", company_id=99),
):
with pytest.raises(HTTPException) as err:
put([bad])
@@ -407,9 +388,9 @@ def test_put_assignees_validation_and_diff(fake_db):
# 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"),
router.AssigneeIn(piece_id="p26", company_name="가나다건설", email="GA@x.kr"),
router.AssigneeIn(piece_id="p26", company_id=9),
router.AssigneeIn(piece_id="p26", company_name="새업체", email="new@x.kr"),
]
)
deleted, updated, inserted = fake_db["applied"]
@@ -418,5 +399,5 @@ def test_put_assignees_validation_and_diff(fake_db):
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")])
put([router.AssigneeIn(piece_id="p26", company_name="새업체", email="new@x.kr")])
assert sorted(fake_db["applied"][0]) == [1, 2] # 빠진 업체 지움
resources/tester/test_547_2_home.py
+38 -1
View File
@@ -93,7 +93,7 @@ class _Cursor:
return {"n": 2, "name": "회사", "member_count": 3, "companies": 4, "users": 9}
async def fetchall(self):
if "temp_upload_batches" in self.sql or "plan_route_years" in self.sql:
if "temp_upload_batches" in self.sql or "plan_route_assignees" in self.sql:
return []
return [{"id": 1, "name": "회사"}]
@@ -150,3 +150,40 @@ def test_home_blocks_by_role(monkeypatch, session, keys, join):
assert ("join_requests" in body["todos"]) is join
if "system" in body:
assert body["system"]["server"] == {"cpu": 1, "memory": 2, "disk": None}
def test_forest_counts_from_plan_file(monkeypatch):
"""산림청 홈(PLAN 590-1) — 분석 전 = 노선 안 그은 프로젝트 수 · 분석됨 · 확정 = 파일 연도 ·
지정 대기 = 확정 연도 중 그 연도 조각에 업체 없음."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from helper_plan_file import year_line_plan
from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
drawn = year_line_plan({2026: [[0.0, 0.0], [100.0, 0.0]], 2027: [[100.0, 0.0], [100.0, 50.0]]})
plans = {"빈": store.empty_plan(), "그음": drawn, "그음2": drawn}
prints = network.year_rows(store.plan_units(drawn))
async def load(pid):
return Path("."), plans[pid]
async def status(pid):
return {2026: {"fingerprint": prints[2026]["fingerprint"], "confirmed_at": "x"}}
monkeypatch.setattr(store, "load_plan", load)
monkeypatch.setattr(store, "year_status", status)
class Cursor(_Cursor):
async def fetchall(self):
return [
{"id": "빈", "piece_id": None},
{"id": "그음", "piece_id": None},
{"id": "그음2", "piece_id": "p26"}, # 확정 2026 에 업체 있음
]
out = asyncio.run(home._forest(Cursor(), 4))
assert out == {"plan_status": {"draft": 1, "analyzed": 2, "confirmed": 2}, "assign_waiting": 1}
resources/tester/test_563_2_plan_pieces.py
+48 -97
View File
@@ -7,13 +7,13 @@
- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘)
- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행
- 연결 · 가져오기 = 조각 꺾은선 shp · 받은 목록 = 조각 이름 · 길이
- 노선 · 조각 · 연도 = 설계 내용 파일(PLAN 590-1 · `helper_plan_file`)
"""
from __future__ import annotations
import asyncio
import io
import json
import sys
import tempfile
import zipfile
@@ -28,6 +28,8 @@ ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from helper_plan_file import plan_of, use_plan # noqa: E402
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
@@ -45,28 +47,21 @@ 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),
LINES = [
{"id": "main", "vertices": [P1[0], P1[1], P2[1]], "branch": None},
{"id": "b1", "vertices": P3, "branch": {"line": "main", "vertex": 1}},
]
MARKS = [{"line": "main", "at_m": 250.0}]
PIECES = [
{"id": "p1", "line": "main", "from_m": 0.0, "to_m": 250.0, "year": 2026},
{"id": "p3", "line": "b1", "from_m": 0.0, "to_m": 130.0, "year": 2026},
{"id": "p2", "line": "main", "from_m": 250.0, "to_m": 250.0 + L2, "year": 2027},
]
PLAN = plan_of(LINES, MARKS, PIECES)
def _assignee(aid, piece, year, company, name, email, notified=None):
"""DB 업체 줄 — `year` 는 낡은 사본(읽지 않음 · 연도 = 파일 조각)."""
return {
"id": aid,
"year": year,
@@ -82,13 +77,10 @@ def _assignee(aid, piece, year, company, name, email, notified=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},
],
"file": use_plan(monkeypatch, PLAN, {2026: "x"}),
"assignees": [
_assignee(7, "p1", 2026, 5, "가", "a@ga.kr"),
_assignee(8, "p3", 2026, 6, "나", "b@na.kr"),
_assignee(8, "p3", 2029, 6, "나", "b@na.kr"), # 낡은 연도 사본(요구 452 ⑤)
],
"mails": [],
"applied": None,
@@ -97,12 +89,6 @@ def plan(monkeypatch):
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"]
@@ -120,8 +106,6 @@ def plan(monkeypatch):
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)
@@ -146,6 +130,7 @@ def test_summary_piece_rows_and_firm_view(plan):
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]
assert [(a["id"], a["year"]) for a in out["assignees"]] == [(7, 2026), (8, 2026)]
# 같은 연도 다른 업체 — 자기 조각 · 자기 행만
for session, piece, aid in ((FIRM5, "p1", 7), (FIRM6, "p3", 8)):
mine = asyncio.run(d04.get_plan_summary(PLAN_ID, session))
@@ -178,7 +163,7 @@ def test_export_piece_folders_and_firm_only_own(plan):
]
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 # 확정 전 = 받을 것 없음
plan["file"]["confirmed"].clear() # 확정 전 = 받을 것 없음
with pytest.raises(HTTPException) as err:
asyncio.run(d04.export_plan(PLAN_ID, FIRM6))
assert err.value.status_code == 403
@@ -218,41 +203,30 @@ def test_put_assignees_by_piece(plan):
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_confirm_drops_when_file_changes(plan, monkeypatch):
"""파일만 바뀌고 연도 상태 지문이 낡으면 확정으로 안 보임(파일 쓰기 · DB 커밋 틈)."""
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
async def stale(_pid):
return {2026: {"fingerprint": "낡음", "confirmed_at": "x"}}
monkeypatch.setattr(store, "year_status", stale)
out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
assert all(r["confirmed_at"] is None for r in out["pieces"])
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)
def test_link_uses_piece_route(plan):
unit = asyncio.run(link.piece_route(PLAN_ID, "p3"))
assert unit["year"] == 2026 and unit["epsg"] == 5186 and unit["confirmed_at"] == "x"
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)
files = link.year_files({**unit, "name": "2026 홍천 노선 계획", "region": "일월.용화"})
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
gone = asyncio.run(link.piece_route(PLAN_ID, "없음"))
assert gone["route_json"] is None and gone["year"] is None
class _Cursor:
@@ -289,43 +263,20 @@ class _Conn:
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
def test_assigned_list_piece_label_length(plan, monkeypatch):
"""연도 · 길이 · 이름 = 파일 조각 · 확정 안 되고 연결도 없는 줄(2027 · 없는 조각)은 뺌."""
row = {
"assignee_id": 8,
"project_id": PLAN_ID,
"project_name": "시험",
"piece_id": "p3",
"linked_project_id": None,
"linked_project_name": None,
"source_fingerprint": None,
}
rows = [row, {**row, "assignee_id": 9, "piece_id": "p2"}, {**row, "piece_id": "없음"}]
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),
assert [(r["piece_id"], r["year"], r["piece_label"], r["length_m"]) for r in out] == [
("p3", 2026, "갈래 1 0+0~6+10", 130.0),
]
resources/tester/test_590_1_plan_file.py
+209
View File
@@ -0,0 +1,209 @@
# -*- coding: utf-8 -*-
"""590-1 예정노선 설계 내용 = 프로젝트 폴더 한 파일(요구 452 · 453) — 지킴 시험.
- 옮김 = 옛 DB 칸(지점 points · starts/ends · start/end · 빈 칸 = 연도 행 접기 · 노선 정본 ·
옛 연도 노선) → 새 꼴 한 벌(좌표계 머리 한 곳 · 빈 연도 없음)
- 원자 쓰기 = 바꿔치기 실패에도 옛 파일 그대로 · 임시 파일 안 남음 · 새 판 파일 거절
- 연도 읽기 = 업체 줄 연도는 파일 조각에서(낡은 DB 연도 무시 · ⑤) · 확정 = 지문 같을 때만
- 잠금 = 같은 프로젝트 [저장] 둘이 겹치지 않음
./venv/Scripts/python.exe -m pytest resources/tester/test_590_1_plan_file.py -q
"""
from __future__ import annotations
import asyncio
import json
import sys
from pathlib import Path
import pytest
from fastapi import HTTPException
sys.path.insert(0, str(Path(__file__).parent))
from helper_plan_file import use_plan, year_line_plan # noqa: E402
from common_util.common_util_crs import lonlat_to_tm # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Router as d04 # noqa: E402
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
A = {"x": 210000.0, "y": 380000.0}
B = {"x": 210300.0, "y": 380400.0}
NETWORK = {
"epsg": 5186,
"lines": [{"id": "main", "vertices": [[0.0, 0.0], [100.0, 0.0]], "branch": None}],
"marks": [{"line": "main", "at_m": 40.0}],
"pieces": [
{"id": "p1", "line": "main", "from_m": 0.0, "to_m": 40.0, "year": 2027},
{"id": "p2", "line": "main", "from_m": 40.0, "to_m": 100.0, "year": 2029},
],
}
# --- 옮김 ---
@pytest.mark.parametrize(
"base",
[
{"epsg": 5186, "points": [A, B]},
{"epsg": 5186, "starts": [A], "ends": [B]},
{"epsg": 5186, "start": A, "end": B},
],
)
def test_migrate_old_point_forms(base):
plan = store.migrate(json.dumps(base), json.dumps(NETWORK), [])
assert plan["points"] == [A, B] and plan["epsg"] == 5186
assert plan["lines"] == NETWORK["lines"] and plan["pieces"] == NETWORK["pieces"]
assert set(plan) == {"version", "epsg", "points", "lines", "marks", "pieces"}
def test_migrate_folds_year_rows_and_legacy_route():
rows = [
{"year": 2027, "epsg": 5186, "route_json": None,
"start_x": 1.0, "start_y": 2.0, "end_x": 3.0, "end_y": 4.0},
{"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]}),
"start_x": 0.0, "start_y": 0.0, "end_x": 1.0, "end_y": 2.0},
{"year": 2028, "epsg": None, "route_json": None,
"start_x": None, "start_y": None, "end_x": None, "end_y": None}, # 빈 연도 줄
] # fmt: skip
plan = store.migrate(None, None, rows)
assert plan["points"] == [{"x": 0.0, "y": 0.0}, {"x": 3.0, "y": 4.0}] # 첫 시점 · 끝 종점
assert [p["year"] for p in plan["pieces"]] == [2026] # 빈 연도는 조각이 아님
assert store.plan_units(plan)[0]["length_m"] == 50.0
assert store.migrate(None, None, []) == store.empty_plan()
def test_migrate_one_epsg_converts_points():
lon_lat = (127.1, 37.5)
x, y = lonlat_to_tm(*lon_lat, 5185)
plan = store.migrate(
json.dumps({"epsg": 5185, "points": [{"x": x, "y": y}]}), json.dumps(NETWORK), []
)
assert plan["epsg"] == 5186
assert (plan["points"][0]["x"], plan["points"][0]["y"]) == pytest.approx(
lonlat_to_tm(*lon_lat, 5186), abs=1e-3
)
# --- 원자 쓰기 · 판 ---
def test_atomic_write_keeps_old_file_on_failure(tmp_path, monkeypatch):
plan = {**store.empty_plan(5186), **NETWORK}
store.save_plan(tmp_path, plan)
path = store.plan_path(tmp_path)
assert (
path.parent.name == "D03_PlanAnalysis"
and json.loads(path.read_text("utf-8"))["version"] == 1
)
assert store.read_plan(tmp_path) == plan
import common_util.common_util_json as cj
def broken(source, target):
raise PermissionError("잡힘")
monkeypatch.setattr(cj, "replace_with_retry", broken)
with pytest.raises(PermissionError):
store.save_plan(tmp_path, {**plan, "lines": [], "marks": [], "pieces": []})
assert store.read_plan(tmp_path) == plan # 옛 파일 그대로
assert [p.name for p in path.parent.iterdir()] == ["plan_network.json"] # 임시 파일 안 남음
def test_newer_version_refused(tmp_path):
path = store.plan_path(tmp_path)
path.parent.mkdir(parents=True)
path.write_text(json.dumps({**store.empty_plan(), "version": 2}), "utf-8")
with pytest.raises(HTTPException) as err:
store.read_plan(tmp_path)
assert err.value.status_code == 409
assert store.read_plan(tmp_path / "없음") is None
# --- 연도 읽기(⑤) ---
def test_assignee_year_comes_from_file(monkeypatch):
use_plan(monkeypatch, {**store.empty_plan(5186), **NETWORK})
_, units = asyncio.run(d04._units(PID))
stale = [{"id": 27, "piece_id": "p1", "year": 2028}, {"id": 28, "piece_id": "p2", "year": 2027}]
assert [a["year"] for a in d04.with_years(stale, units)] == [2027, 2029]
assert d04.with_years([{"id": 1, "piece_id": "없음", "year": 2027}], units)[0]["year"] is None
def test_confirmed_only_when_fingerprint_matches(monkeypatch):
plan = year_line_plan({2026: [[0.0, 0.0], [10.0, 0.0]], 2027: [[10.0, 0.0], [10.0, 10.0]]})
state = use_plan(monkeypatch, plan, {2026: "x"})
years = asyncio.run(store.plan_state(PID))["years"]
assert years[2026]["confirmed_at"] == "x" and years[2027]["confirmed_at"] is None
moved = json.loads(json.dumps(plan))
moved["lines"][0]["vertices"][0] = [20.0, 0.0] # 2026 조각 모양 바뀜(길이 같음)
async def stale(_pid): # DB 지문 = 옛 파일 것
return {2026: {"fingerprint": years[2026]["fingerprint"], "confirmed_at": "x"}}
monkeypatch.setattr(store, "year_status", stale)
state["plan"] = moved
assert asyncio.run(store.plan_state(PID))["years"][2026]["confirmed_at"] is None
# --- 잠금 ---
def test_two_saves_do_not_interleave(monkeypatch, tmp_path):
order: list[str] = []
async def load(_pid):
return tmp_path, store.read_plan(tmp_path) or store.empty_plan()
async def year_start(_pid):
return 2026
async def slow_status(connection, project_id, pieces, years):
order.append(f"start {len(pieces)}")
await asyncio.sleep(0.05)
order.append(f"end {len(pieces)}")
class Conn:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def begin(self):
pass
async def commit(self):
pass
async def rollback(self):
pass
class Pool:
def acquire(self):
return Conn()
monkeypatch.setattr(store, "load_plan", load)
monkeypatch.setattr(net, "_year_start", year_start)
monkeypatch.setattr(net, "_store_status", slow_status)
monkeypatch.setattr(net, "get_db_pool", lambda: Pool())
forest = {"role": "SYSTEM_ADMIN", "email": "a@b.kr"}
one = net.Network(**{**NETWORK, "pieces": NETWORK["pieces"][:1]})
two = net.Network(**NETWORK)
async def both():
from uuid import UUID
await asyncio.gather(
net.put_plan_network(UUID(PID), one, forest),
net.put_plan_network(UUID(PID), two, forest),
)
asyncio.run(both())
assert order == ["start 1", "end 1", "start 2", "end 2"]
assert len(store.read_plan(tmp_path)["pieces"]) == 2 # 뒤 저장이 이김 · 반쪽 없음