diff --git a/D01_PlanRegister/D01_PlanRegister_Router.py b/D01_PlanRegister/D01_PlanRegister_Router.py index ce81164b8..62df091cc 100644 --- a/D01_PlanRegister/D01_PlanRegister_Router.py +++ b/D01_PlanRegister/D01_PlanRegister_Router.py @@ -68,20 +68,29 @@ async def post_plan_project( async def assigned_plan_routes( session: dict[str, Any] = Depends(require_company), ) -> dict[str, Any]: - """우리 회사에 지정된 예정노선 연도 — 보기 · 연결 상태.""" + """우리 회사에 지정된 예정노선 연도 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트.""" + from D04_PlanConfirm.D04_PlanConfirm_Link import update_available + async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( - """SELECT a.project_id, p.name AS project_name, a.year, a.linked_project_id + """SELECT a.id AS assignee_id, a.project_id, p.name AS project_name, a.year, + l.id AS linked_project_id, l.name AS linked_project_name, + l.source_fingerprint, y.route_json 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 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""", (session.get("company_id"),), ) rows = await cursor.fetchall() - # ponytail: update_available 자리만 — 지문 비교는 531-3. - routes = [{**r, "update_available": False} for r in rows] + routes = [] + for row in rows: + route_json, fingerprint = row.pop("route_json"), row.pop("source_fingerprint") + routes.append({**row, "update_available": update_available(route_json, fingerprint)}) return {"status": "success", "routes": routes} diff --git a/D04_PlanConfirm/D04_PlanConfirm_Link.py b/D04_PlanConfirm/D04_PlanConfirm_Link.py new file mode 100644 index 000000000..966a53fab --- /dev/null +++ b/D04_PlanConfirm/D04_PlanConfirm_Link.py @@ -0,0 +1,323 @@ +"""D 예정노선 → 임도 연결 (요구 345 · 요구서 2-3 · 3-5 · PLAN 531-3). + +- `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/` 에 원래 이름으로 + +`input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길. 입력이 다 차면 +B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림. +연결 시점 복사 · 원본이 바뀌면 알림만(내용 지문 비교 · 자동 반영 없음). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import shutil +import tempfile +from pathlib import Path +from typing import Any, Literal +from uuid import UUID + +import aiomysql +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import JSONResponse +from pydantic import BaseModel + +from B02_ProjRegister.B02_ProjRegister_Repository import create_project +from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata +from B03_FileInput.B03_FileInput_Repository import ( + create_input_file, + get_project_input_readiness, + get_project_storage_relative_path, + supersede_previous_input_files, +) +from B03_FileInput.B03_FileInput_Router_Helpers import ( + _ANALYSIS_RUNNING_MESSAGE, + OutputsWouldBeDiscarded, + _complete_file_input_if_ready, + _confirm_replace_response, + _missing_required_file_types, + _schedule_background_task, + _write_stage_metadata, +) +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 +from config.config_db import get_db_pool +from D04_PlanConfirm.D04_PlanConfirm_Export import build_export +from D04_PlanConfirm.D04_PlanConfirm_Router import year_routes_from_rows + +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" + + +class LinkRequest(BaseModel): + mode: Literal["new", "existing"] + project_id: UUID | None = None + confirm_replace: bool = False + las_free: bool = False # B03 「LAS 없이 설계」(화면 기억값) — 입력 완료 판정에만 + + +def _parsed(route_json: Any) -> dict[str, Any] | None: + value = json.loads(route_json) if isinstance(route_json, str | bytes) else route_json + return value if isinstance(value, dict) else None + + +def route_fingerprint(route_json: Any) -> str | None: + """연도 노선 내용 지문(sha1) — 고른 중심선 점(mm 반올림)만 봄 · 같은 값이면 같은 지문.""" + body = _parsed(route_json) + planned = (body or {}).get("planned") + if not planned: + return None + points = [[round(float(p[0]), 3), round(float(p[1]), 3)] for p in planned] + return hashlib.sha1(json.dumps(points).encode()).hexdigest() + + +def update_available(route_json: Any, source_fingerprint: str | None) -> bool: + """연결 때 지문 ≠ 지금 지문 — 지금 노선이 비면(지점 바뀜 · 분석 전) 알리지 않음.""" + now = route_fingerprint(route_json) + return bool(source_fingerprint) and now is not None and now != source_fingerprint + + +async def _fetch_one(sql: str, args: tuple) -> dict[str, Any] | None: + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute(sql, args) + return await cursor.fetchone() + + +async def load_assignee(assignee_id: int) -> dict[str, Any] | None: + """지정 행 + 예정노선 프로젝트 칸 + 그 연도 노선.""" + return await _fetch_one( + "SELECT a.id, a.project_id, a.year, a.company_id, a.linked_project_id," + " p.name, p.region, p.road_type, p.design_speed_kph, p.route_start_m," + " y.epsg, y.route_json, y.confirmed_at" + " FROM plan_route_assignees a" + " JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL" + " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year" + " WHERE a.id = %s", + (assignee_id,), + ) + + +async def load_road_project(project_id: str) -> dict[str, Any] | None: + return await _fetch_one( + "SELECT id, company_id, project_kind FROM projects WHERE id = %s AND deleted_at IS NULL", + (project_id,), + ) + + +async def record_link( + project_id: str, assignee: dict[str, Any], fingerprint: str | None, speed: Any +) -> 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), + ) + await cursor.execute( + "UPDATE plan_route_assignees SET linked_project_id = NULL" + " WHERE linked_project_id = %s AND id <> %s", + (project_id, assignee["id"]), + ) + await cursor.execute( + "UPDATE plan_route_assignees SET linked_project_id = %s WHERE id = %s", + (project_id, assignee["id"]), + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + + +def year_files(assignee: dict[str, Any]) -> dict[str, bytes]: + """그 연도 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트.""" + row = {**assignee, "route_json": _parsed(assignee["route_json"])} + routes = year_routes_from_rows([row]) + bundle = build_export(assignee["name"], assignee["region"] or "", routes, with_xlsx=False) + return bundle.by_year[int(assignee["year"])] + + +async def _supersede_route_set(connection: aiomysql.Connection, project_id: str) -> None: + """옛 노선 세트(다른 이름 포함) 행을 내림 — 노선 칸은 새 한 벌만 봄.""" + async with connection.cursor() as cursor: + await cursor.execute( + "UPDATE input_files SET status = 'SUPERSEDED' WHERE project_id = %s" + " AND raw_file_path LIKE %s AND status IN ('UPLOADED', 'PROCESSED')", + (project_id, f"%{ROUTE_DIR}/%"), + ) + + +async def import_route_files( + project_id: str, + files: dict[str, bytes], + session: dict[str, Any], + confirm_replace: bool = False, + las_free: bool = False, +) -> dict[str, Any]: + """shp 한 벌 → B03 입력. 지울 결과가 있고 확인이 없으면 `OutputsWouldBeDiscarded`.""" + pid = UUID(project_id) + async with get_db_pool().acquire() as connection: + async with connection.cursor(aiomysql.DictCursor) as cursor: + if await is_analysis_running(cursor, project_id): + raise HTTPException(status_code=409, detail=_ANALYSIS_RUNNING_MESSAGE) + stored = await get_project_storage_relative_path(connection, pid) + project_root = Path(resolve_stored_project_path(stored)) + with tempfile.TemporaryDirectory() as tmp: + for name, data in files.items(): + (Path(tmp) / name).write_bytes(data) + metas = { + name: await asyncio.to_thread(analyze_input_metadata, Path(tmp) / name) + for name in files + } + await connection.begin() + try: + await _supersede_route_set(connection, project_id) + results: list[UploadedFileResult] = [] + for name, data in files.items(): + meta, relative = metas[name], f"{ROUTE_DIR}/{name}" + epsg = meta.get("epsg") + input_id = await create_input_file( + connection, + project_id=pid, + file_type=Path(name).suffix.lower().lstrip("."), + original_filename=name, + relative_path=relative, + file_size_bytes=len(data), + upload_by=int(session["user_id"]), + crs_epsg=int(epsg) if epsg is not None else None, + metadata=meta, + ) + await supersede_previous_input_files(connection, pid, name, input_id) + results.append( + UploadedFileResult( + input_file_id=input_id, + original_filename=name, + file_type=Path(name).suffix.lower().lstrip("."), + relative_path=relative, + size_bytes=len(data), + metadata=meta, + ) + ) + types, _, _ = await get_project_input_readiness(connection, pid) + wf1_input = None + if not _missing_required_file_types(types, las_free): + wf1_input = await _complete_file_input_if_ready( + connection, pid, las_free, confirm_replace + ) + target = project_root / ROUTE_DIR + target.mkdir(parents=True, exist_ok=True) + for name in files: + shutil.move(str(Path(tmp) / name), str(target / name)) + await connection.commit() + except Exception: + await connection.rollback() + raise + _write_stage_metadata(project_root / "B03_FileInput", pid, results) + if wf1_input is not None: + _schedule_background_task( + trigger_wf1_analysis_and_email( + project_id=pid, + input_file_id=wf1_input, + user_role=str(session["role"]), + started_by=session.get("user_id"), + ), + task_name=f"b04-preprocess-auto-{project_id}", + ) + return {"files": [r.original_filename for r in results], "analysis_started": bool(wf1_input)} + + +async def _new_road_project(assignee: dict[str, Any], session: dict[str, Any]) -> str: + """예정노선 이름 · 지역 · 임도 종류 · 그 연도로 임도 프로젝트(B02 등록과 같은 함수).""" + body = _parsed(assignee["route_json"]) or {} + created = await create_project( + user_id=int(session["user_id"]), + company_id=int(session["company_id"]), + name=f"{assignee['name']} {assignee['year']}", + region=assignee["region"] or "", + road_type=assignee["road_type"] or "work", + project_year=int(assignee["year"]), + estimated_length_m=body.get("length_m"), + memo=None, + ) + return str(created["project_id"]) + + +@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 입력으로 — 지정 회사만.""" + assignee = await load_assignee(assignee_id) + if not assignee: + raise HTTPException(status_code=404, detail="지정 업체 행을 찾을 수 없습니다.") + if assignee["company_id"] is None or assignee["company_id"] != session.get("company_id"): + raise HTTPException(status_code=403, detail="이 예정노선을 지정받은 회사가 아닙니다.") + if not assignee["confirmed_at"] or route_fingerprint(assignee["route_json"]) is None: + raise HTTPException(status_code=409, detail="확정된 노선이 없습니다.") + if payload.mode == "existing": + if payload.project_id is None: + raise HTTPException(status_code=400, detail="연결할 임도 프로젝트를 고르세요.") + project = await load_road_project(str(payload.project_id)) + if not project: + raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.") + if project["company_id"] != session.get("company_id"): + raise HTTPException(status_code=403, detail="다른 회사의 프로젝트입니다.") + if project["project_kind"] != "ROAD": + raise HTTPException(status_code=400, detail="임도설계 프로젝트만 연결할 수 있습니다.") + files = await asyncio.to_thread(year_files, assignee) + project_id = ( + str(payload.project_id) + if payload.mode == "existing" + else await _new_road_project(assignee, session) + ) + try: + result = await import_route_files( + project_id, files, session, payload.confirm_replace, payload.las_free + ) + except OutputsWouldBeDiscarded as exc: + return _confirm_replace_response(exc) + 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"]) + return {"status": "success", "project_id": project_id, "year": assignee["year"], **result} + + +@router.get("/{project_id}/plan-source", response_model=None) +async def plan_source(project_id: UUID) -> dict[str, Any]: + """임도 프로젝트가 이어 받은 예정노선(B03 머리 띠) — 없으면 linked False.""" + row = await _fetch_one( + "SELECT p.source_fingerprint, s.name AS plan_name, a.id AS assignee_id, a.year," + " y.route_json" + " FROM projects p" + " JOIN projects s ON s.id = p.source_project_id AND s.project_kind = 'PLAN'" + " JOIN plan_route_assignees a ON a.linked_project_id = p.id" + " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year" + " WHERE p.id = %s LIMIT 1", + (str(project_id),), + ) + if not row: + return {"status": "success", "linked": False} + return { + "status": "success", + "linked": True, + "assignee_id": row["assignee_id"], + "plan_name": row["plan_name"], + "year": row["year"], + "update_available": update_available(row["route_json"], row["source_fingerprint"]), + } diff --git a/D04_PlanConfirm/D04_PlanConfirm_Router.py b/D04_PlanConfirm/D04_PlanConfirm_Router.py index 037101d64..45ba05381 100644 --- a/D04_PlanConfirm/D04_PlanConfirm_Router.py +++ b/D04_PlanConfirm/D04_PlanConfirm_Router.py @@ -3,6 +3,8 @@ - `GET /api/projects/{id}/plan/assignees` — 연도별 지정 업체 - `PUT /api/projects/{id}/plan/assignees` — 연도별 지정 업체 저장(등록 회사 또는 업체명 + 메일) - `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 zip → 메일 첨부 → 발송 시각 +- `GET /api/projects/{id}/plan/summary` — D04 화면 연도 요약(연장 · 시점 · 종점 · 누가 · 확정) +- `GET /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 연도 + 전체 · 공유 = 자기 연도) 지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이 GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦. @@ -11,14 +13,19 @@ GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 from __future__ import annotations import asyncio +import io import logging +import zipfile from typing import Any +from urllib.parse import quote from uuid import UUID from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import Response 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 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 ( @@ -26,6 +33,7 @@ from D04_PlanConfirm.D04_PlanConfirm_Export import ( YearRoute, build_export, company_zip, + safe_name, ) logger = logging.getLogger(__name__) @@ -212,3 +220,87 @@ async def finalize_plan( await repo.mark_notified(assignee["id"]) results.append(_result(assignee, "sent" if sent else "failed")) return {"status": "success", "confirmed_years": years, "notifications": results} + + +def _end_point(point: list[float], epsg: int) -> dict[str, float]: + lon, lat = tm_to_lonlat(float(point[0]), float(point[1]), epsg) + return {"x": float(point[0]), "y": float(point[1]), "lon": lon, "lat": lat} + + +def year_summary(row: dict[str, Any]) -> dict[str, Any]: + """연도 한 줄 — 분석 [저장]이 없으면 saved False 만.""" + body = row.get("route_json") or {} + planned = body.get("planned") or [] + out = {"year": int(row["year"]), "saved": len(planned) >= 2, "confirmed_at": None} + if row.get("confirmed_at"): + out["confirmed_at"] = str(row["confirmed_at"]) + if out["saved"]: + epsg = int(row["epsg"]) + out.update( + length_m=body.get("length_m"), + start_chainage_m=body.get("start_chainage_m"), + end_chainage_m=body.get("end_chainage_m"), + mode=body.get("mode"), + start=_end_point(planned[0], epsg), + end=_end_point(planned[-1], epsg), + ) + return out + + +@router.get("/{project_id}/plan/summary", response_model=None) +async def get_plan_summary(project_id: UUID) -> dict[str, Any]: + project = await _plan_project(project_id) + rows = await repo.list_year_routes(str(project_id)) + return { + "status": "success", + "name": project["name"], + "years": [year_summary(row) for row in rows], + "assignees": await repo.list_assignees(str(project_id)), + } + + +def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes: + """연도마다 `<연도>/` · 전체는 `전체/` — 좌표표 이름이 연도마다 같아 폴더로 가름.""" + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: + for year in years: + for name, data in bundle.by_year[year].items(): + archive.writestr(f"{year}/{name}", data) + if with_all: + for name, data in bundle.all.items(): + archive.writestr(f"전체/{name}", data) + return buffer.getvalue() + + +@router.get("/{project_id}/plan/export", response_model=None) +async def export_plan(project_id: UUID, session: dict = Depends(verify_session)) -> Response: + """산림청 = 저장된 전 연도 + 전체 · 지정 회사 = 확정된 자기 연도만(전체 없음).""" + project = await _plan_project(project_id) + rows = await repo.list_year_routes(str(project_id)) + routes = year_routes_from_rows(rows) + if not routes: + raise HTTPException(status_code=409, detail="분석 [저장]이 된 연도가 없습니다.") + forest = is_forest_service(session) + if forest: + years = [r.year for r in routes] + else: + confirmed = {int(r["year"]) for r in rows if r.get("confirmed_at")} + mine = { + a["year"] + for a in await repo.list_assignees(str(project_id)) + if a["company_id"] is not None and a["company_id"] == session.get("company_id") + } + years = sorted({r.year for r in routes} & confirmed & mine) + if not years: + raise HTTPException(status_code=403, detail="받을 수 있는 연도가 없습니다.") + first = float(project.get("route_start_m") or 0.0) + bundle = await asyncio.to_thread( + build_export, project["name"], project["region"] or "", routes, first + ) + data = await asyncio.to_thread(export_zip, bundle, years, forest) + name = f"{safe_name(project['name'])}_예정노선.zip" + return Response( + content=data, + media_type=MIME_ZIP, + headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(name)}"}, + ) diff --git a/main.py b/main.py index dd14852dd..795369810 100644 --- a/main.py +++ b/main.py @@ -76,6 +76,8 @@ from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_das from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router from D02_PlanBase.D02_PlanBase_Router import router as d02_plan_base_router from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router +from D04_PlanConfirm.D04_PlanConfirm_Link import dashboard_router as d04_link_dashboard_router +from D04_PlanConfirm.D04_PlanConfirm_Link import router as d04_plan_link_router from D04_PlanConfirm.D04_PlanConfirm_Router import router as d04_plan_confirm_router from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router @@ -459,6 +461,7 @@ protected = [Depends(verify_session)] # 신규 그룹 등록(요구 345) — 권한은 길 안에서(산림청 · 회사). app.include_router(d01_plan_register_router, dependencies=protected) app.include_router(d01_dashboard_router, dependencies=protected) +app.include_router(d04_link_dashboard_router, dependencies=protected) app.include_router(e01_stab_register_router, dependencies=protected) # 프로젝트 경로가 붙는 라우터는 그 프로젝트가 내 회사 것인지도 함께 본다 # (2026-09-01 — 지금까지 id만 알면 남의 회사 자료에 닿았다). @@ -482,6 +485,7 @@ app.include_router(b05_route_lifecycle_router, dependencies=protected_with_compa app.include_router(d02_plan_base_router, dependencies=protected_with_company) app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) app.include_router(d04_plan_confirm_router, dependencies=protected_with_company) +app.include_router(d04_plan_link_router, dependencies=protected_with_company) app.include_router(b05_corridor_router, dependencies=protected_with_company) app.include_router(b05_route_replan_router, dependencies=protected_with_company) app.include_router(b05_route_terrain_router, dependencies=protected_with_company) diff --git a/resources/tester/test_531_3_plan_link.py b/resources/tester/test_531_3_plan_link.py new file mode 100644 index 000000000..247f38edb --- /dev/null +++ b/resources/tester/test_531_3_plan_link.py @@ -0,0 +1,389 @@ +# -*- coding: utf-8 -*- +"""531-3 예정노선 → 임도 연결 · 업데이트 있음 · [내보내기] 받기. + +- 연결 → B03 입력 `input/shp/` 파일 5 개 · input_files 행 5 · B03 분석 통과 · 원본 칸 · 연결 표시 +- 입력이 다 찼고 지울 결과가 있으면 확인 409 · 파일 · 행 안 남김 +- 지정 안 받은 회사 403 · 확정 안 된 연도 409 · 남의 회사 임도 403 +- 지문 — 노선 바뀌면 update_available True · 같은 값으로 되돌리면 False +- 받은 예정노선 목록 · plan-source 가 지문으로 판정 +- [내보내기] — 산림청 = 전 연도 + 전체 · 지정 회사 = 확정된 자기 연도 · 없으면 403 +""" + +from __future__ import annotations + +import asyncio +import io +import json +import sys +import zipfile +from pathlib import Path +from uuid import uuid4 + +import pytest +from fastapi import HTTPException + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata # noqa: E402 +from B03_FileInput.B03_FileInput_Router_Helpers import OutputsWouldBeDiscarded # noqa: E402 +from common_util.common_util_route_geometry import find_planned_route_file # noqa: E402 +from D01_PlanRegister import D01_PlanRegister_Router as d01 # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Link as link # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Repository as repo # noqa: E402 +from D04_PlanConfirm import D04_PlanConfirm_Router as d04 # noqa: E402 +from D04_PlanConfirm.D04_PlanConfirm_Export import SHP_SUFFIXES # noqa: E402 + +PLAN_ID = str(uuid4()) +ROAD_ID = str(uuid4()) +# 5186 좌표 두 해 — 2026 끝 = 2027 시작 +P2026 = [[210000.0, 380000.0], [210100.0, 380050.0], [210250.0, 380040.0]] +P2027 = [[210250.0, 380040.0], [210400.0, 380120.0]] + + +def _route(points, start=0.0): + length = sum( + ((b[0] - a[0]) ** 2 + (b[1] - a[1]) ** 2) ** 0.5 + for a, b in zip(points, points[1:], strict=False) + ) + return { + "mode": "balanced", + "planned": points, + "length_m": length, + "start_chainage_m": start, + "end_chainage_m": start + length, + } + + +def _assignee(**kw): + row = { + "id": 7, + "project_id": PLAN_ID, + "year": 2026, + "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)), + "confirmed_at": "2026-10-25 12:00:00", + } + row.update(kw) + return row + + +SESSION = {"user_id": 3, "company_id": 5, "role": "USER", "email": "a@corp.com"} + + +class FakeCursor: + def __init__(self, log, rows=None): + self.log, self.rows = log, rows or [] + + async def execute(self, sql, args=None): + self.log.append((" ".join(sql.split()), args)) + + async def fetchall(self): + return self.rows + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +class FakeConn: + def __init__(self, log, rows=None): + self.log, self.rows = log, rows + + def acquire(self): + return self + + def cursor(self, *a): + return FakeCursor(self.log, self.rows) + + async def begin(self): + self.log.append(("begin", None)) + + async def commit(self): + self.log.append(("commit", None)) + + async def rollback(self): + self.log.append(("rollback", None)) + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +@pytest.fixture +def env(monkeypatch, tmp_path): + """DB 가짜 — input_files 행 · 입력 유형 · 완료 판정 · 연결 기록을 쌓음.""" + state = {"rows": [], "types": set(), "complete": None, "links": [], "log": []} + conn = FakeConn(state["log"]) + monkeypatch.setattr(link, "get_db_pool", lambda: conn) + + async def false(*a, **k): + return False + + async def storage(*a, **k): + return "1/3/x" + + async def create_input_file(connection, **kw): + state["rows"].append(kw) + return len(state["rows"]) + + async def noop(*a, **k): + return 0 + + async def readiness(*a, **k): + types = set(state["types"]) + for row in state["rows"]: + ft = row["file_type"] + types.add("route_prj" if ft == "prj" and "/input/shp/" in row["relative_path"] else ft) + return types, None, None + + async def complete(connection, pid, las_free, confirm): + state["complete"] = confirm + if not confirm: + raise OutputsWouldBeDiscarded(["노선 · 종단 설계"]) + return 99 + + async def record(project_id, assignee, fingerprint, speed): + state["links"].append((project_id, assignee["id"], fingerprint, speed)) + + monkeypatch.setattr(link, "is_analysis_running", false) + monkeypatch.setattr(link, "get_project_storage_relative_path", storage) + monkeypatch.setattr(link, "resolve_stored_project_path", lambda s: str(tmp_path)) + monkeypatch.setattr(link, "create_input_file", create_input_file) + monkeypatch.setattr(link, "supersede_previous_input_files", noop) + monkeypatch.setattr(link, "get_project_input_readiness", readiness) + monkeypatch.setattr(link, "_complete_file_input_if_ready", complete) + monkeypatch.setattr(link, "record_link", record) + monkeypatch.setattr(link, "_schedule_background_task", lambda coro, **k: coro.close()) + state["root"] = tmp_path + return state + + +def _use(monkeypatch, assignee, road=None): + async def load(_id): + return assignee + + async def road_project(_pid): + return road + + monkeypatch.setattr(link, "load_assignee", load) + monkeypatch.setattr(link, "load_road_project", road_project) + + +def _link(mode="existing", **kw): + payload = link.LinkRequest(mode=mode, project_id=kw.pop("project_id", ROAD_ID), **kw) + return asyncio.run(link.link_plan_route(7, payload, SESSION)) + + +ROAD = {"id": ROAD_ID, "company_id": 5, "project_kind": "ROAD"} + + +def test_link_writes_route_set_like_upload(env, monkeypatch): + _use(monkeypatch, _assignee(), ROAD) + result = _link() + shp_dir = env["root"] / "B03_FileInput/input/shp" + names = sorted(p.name for p in shp_dir.iterdir()) + assert len(names) == 5 and {Path(n).suffix for n in names} == set(SHP_SUFFIXES) + assert all(n.startswith("대상지_일월.용화.") and "_2026." in n for n in names) + assert len(env["rows"]) == 5 + assert {r["relative_path"] for r in env["rows"]} == {f"{link.ROUTE_DIR}/{n}" for n in names} + assert env["complete"] is None # 지형 없음 → 완료 판정 안 함 · WF1 없음 + assert result["analysis_started"] is False and result["project_id"] == ROAD_ID + shp = find_planned_route_file(env["root"] / "B03_FileInput/input") + meta = analyze_input_metadata(shp) + assert meta["purpose"] == "planned_route" and meta["epsg"] == 5179 + assert meta["point_count"] == len(P2026) + 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 any("SUPERSEDED" in sql for sql, _ in env["log"] if isinstance(sql, str)) + + +def test_complete_set_asks_before_discarding(env, monkeypatch): + _use(monkeypatch, _assignee(), ROAD) + env["types"] = {"las", "prj", "tfw"} + response = _link() + assert response.status_code == 409 and json.loads(response.body)["confirm"] == "replace_outputs" + assert not (env["root"] / "B03_FileInput/input/shp").exists() + assert ("rollback", None) in env["log"] and env["links"] == [] + env["rows"].clear() + result = _link(confirm_replace=True) + assert env["complete"] is True and result["analysis_started"] is True + assert len(list((env["root"] / "B03_FileInput/input/shp").iterdir())) == 5 + + +def test_other_company_and_unconfirmed_and_foreign_project(env, monkeypatch): + _use(monkeypatch, _assignee(company_id=9), ROAD) + with pytest.raises(HTTPException) as err: + _link() + assert err.value.status_code == 403 + _use(monkeypatch, _assignee(confirmed_at=None), ROAD) + with pytest.raises(HTTPException) as err: + _link() + assert err.value.status_code == 409 + _use(monkeypatch, _assignee(), {**ROAD, "company_id": 9}) + with pytest.raises(HTTPException) as err: + _link() + assert err.value.status_code == 403 + _use(monkeypatch, _assignee(), {**ROAD, "project_kind": "PLAN"}) + with pytest.raises(HTTPException) as err: + _link() + assert err.value.status_code == 400 + assert env["rows"] == [] + + +def test_new_mode_creates_road_project(env, monkeypatch): + _use(monkeypatch, _assignee()) + made = {} + + async def create_project(**kw): + made.update(kw) + return {"project_id": ROAD_ID} + + monkeypatch.setattr(link, "create_project", create_project) + result = _link(mode="new", project_id=None) + assert result["project_id"] == ROAD_ID + assert made["name"] == "시험 예정노선 2026" and made["region"] == "일월.용화" + assert made["road_type"] == "work" and made["project_year"] == 2026 + assert made["company_id"] == 5 and len(env["rows"]) == 5 + + +def test_fingerprint_update_and_revert(): + linked = link.route_fingerprint(_route(P2026)) + moved = [p[:] for p in P2026] + moved[1][0] += 0.5 + assert link.update_available(json.dumps(_route(P2026)), linked) is False + assert link.update_available(json.dumps(_route(moved)), linked) is True + moved[1][0] -= 0.5 # 같은 값으로 되돌림 + assert link.update_available(json.dumps(_route(moved)), linked) is False + assert link.update_available(None, linked) is False # 지점 바뀌어 노선 빔 → 알림 없음 + assert link.update_available(_route(P2026), None) 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] + 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)), + }, + { + "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)), + }, + ] + log = [] + monkeypatch.setattr(d01, "get_db_pool", lambda: FakeConn(log, rows)) + out = asyncio.run(d01.assigned_plan_routes(SESSION))["routes"] + assert [r["update_available"] for r in out] == [True, False] + assert "route_json" not in out[0] and out[0]["assignee_id"] == 7 + assert "confirmed_at IS NOT NULL" in log[0][0] + + +def test_plan_source(monkeypatch): + linked = link.route_fingerprint(_route(P2026)) + row = { + "source_fingerprint": linked, + "plan_name": "시험", + "assignee_id": 7, + "year": 2026, + "route_json": json.dumps(_route(P2026)), + } + + async def one(sql, args): + 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 asyncio.run(link.plan_source(ROAD_ID))["update_available"] is True + + async def none(sql, args): + return None + + monkeypatch.setattr(link, "_fetch_one", none) + assert asyncio.run(link.plan_source(ROAD_ID)) == {"status": "success", "linked": False} + + +@pytest.fixture +def plan_repo(monkeypatch): + async def project(_pid): + return { + "id": PLAN_ID, + "name": "시험 예정노선", + "region": "일월.용화", + "project_kind": "PLAN", + "route_start_m": 0.0, + } + + async def years(_pid): + return [ + {"year": 2026, "epsg": 5186, "route_json": _route(P2026), "confirmed_at": "x"}, + {"year": 2027, "epsg": 5186, "route_json": _route(P2027), "confirmed_at": None}, + ] + + async def assignees(_pid): + return [ + {"id": 7, "year": 2026, "company_id": 5}, + {"id": 8, "year": 2027, "company_id": 5}, + ] + + monkeypatch.setattr(repo, "get_plan_project", project) + monkeypatch.setattr(repo, "list_year_routes", years) + monkeypatch.setattr(repo, "list_assignees", assignees) + + +def _zip_names(response): + return sorted(zipfile.ZipFile(io.BytesIO(response.body)).namelist()) + + +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 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 = 확정 전 · 전체 없음 + with pytest.raises(HTTPException) as err: + asyncio.run(d04.export_plan(PLAN_ID, {**SESSION, "company_id": 9})) + assert err.value.status_code == 403 + + +def test_summary_year_rows(plan_repo): + out = asyncio.run(d04.get_plan_summary(PLAN_ID)) + first = out["years"][0] + assert first["saved"] and first["start"]["x"] == P2026[0][0] and first["confirmed_at"] == "x" + assert 120 < first["start"]["lon"] < 135 and first["end_chainage_m"] == pytest.approx( + first["length_m"] + ) + assert out["years"][1]["confirmed_at"] is None and len(out["assignees"]) == 2