eomsangdonandClaude Opus 5.5
2d5d357bf3
- 공사지 · 기준점 [+] 없앰 → [자동 정렬] 옆 [요소 추가] - 등록 = 상자만(linked false) · 연결 점으로 이음 · 안 이은 거래처 = 구간 km 안 셈(파이썬 · TS 짝) - 연결선 모달 [수기 입력] = 카카오 길찾기 연결 끊음(manual) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LLCk2y5xE8YGVgjbTntWsY
445 lines
19 KiB
Python
445 lines
19 KiB
Python
"""B08 거리이정표 길(PLAN 244-1) — 공사지 · 기준점 · 경로(업체) · 표 · 경로 그림. DB 표 없음.
|
|
|
|
GET /api/projects/{id}/quantity/milestone {data, site, table, images}
|
|
PUT /api/projects/{id}/quantity/milestone [저장] — 입력만 · 표는 서버가 셈
|
|
POST /api/projects/{id}/quantity/milestone/image?route= 경로 그림 올리기(하나 · 교체)
|
|
GET /api/projects/{id}/quantity/milestone/image?route= 그림 내려받기
|
|
DELETE /api/projects/{id}/quantity/milestone/image?route= 그림 지우기(자동 그림까지) ·
|
|
&user_only=true = 사용자 그림만(「자동 그림으로 되돌리기」 · 244-11)
|
|
POST /api/projects/{id}/quantity/milestone/image/auto?route= 경로 그림 자동(244-11) — 길 선 ·
|
|
표지 → 서버 헤드리스 캡처(`B08_Quantity_Milestone_Capture`) → `auto/<경로 id>.png` ·
|
|
좌표는 그림 뒤 버림 · 실패 = 까닭 한 줄
|
|
GET /api/projects/{id}/quantity/milestone/site-address 시점 모달(244-4) — 등록 주소 ·
|
|
시점 · 중점 VWorld 주소(좌표 → 주소 · 서버 단독 · 저장 안 함) · ?mid_m= 다이어그램
|
|
지금 중점 측점(244-12 · 상자 작은 글씨)
|
|
POST /api/projects/{id}/quantity/milestone/directions [길찾기](244-7) — 두 좌표 → 카카오
|
|
길찾기 → km(소수 1자리) · 실패 · 키 없음 = 까닭 한 줄(서버 단독 · 키 서버만 · 저장 안 함)
|
|
+ 경로 좌표 [위도, 경도] 목록(244-9 · 카카오 지도 길 선 · 그때그때 · 저장 안 함)
|
|
GET /api/projects/{id}/quantity/milestone/reverse?lon=&lat= 지도에서 고른 소재지(244-13) →
|
|
VWorld 주소 한 줄(서버 단독 · 저장 안 함 · 못 찾음 = null)
|
|
GET /api/projects/{id}/quantity/milestone/map-key 카카오 지도 JavaScript 키(244-9 ·
|
|
공개용 · .env KAKAO_JS_KEY · 없으면 null)
|
|
|
|
정본 = 프로젝트 storage `B08_Quantity/milestone/`(`milestone.json` + `<경로 id>.<확장자>` 사용자
|
|
그림 · `auto/<경로 id>.png` 자동 그림 · 폴더가 목록 · B09 견적서 파일 길 꼴) · 보이는 그림 = 사용자
|
|
그림 우선. 초기값 층 없음(자동설계가 낳는 값 아님).
|
|
그림은 올리는 즉시 서버에([저장]과 따로) · [저장] 때 없는 경로의 그림은 지움.
|
|
공사지 = 시점(노선 첫 점 · 측점 0) · 중점 기본 = 노선 연장 ÷ 2 — 노선은 설계 노선 정본 차례
|
|
(수정본 → 초기 폴리라인 → 예상노선 → 초기값 · `load_design_route` 와 같은 차례).
|
|
표 합 = `B08_Quantity_Milestone_Calc`(파이썬 · TS 짝).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Annotated, Any, Literal
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, File, HTTPException, Query, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, Field, ValidationError, model_validator
|
|
|
|
from B01_Dashboard.B01_Dashboard_Map import reverse_geocode
|
|
from B02_ProjRegister.B02_ProjRegister_Repository import get_project_by_id
|
|
from B03_FileInput.B03_FileInput_Engine import save_upload_stream
|
|
from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone, r3
|
|
from B08_Quantity.B08_Quantity_Milestone_Capture import make_route_image
|
|
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
|
|
from common_util.common_util_initial_snapshot import design_route_csv_path
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_route_geometry import (
|
|
expected_route_csv_path,
|
|
interpolate_vertex,
|
|
planned_route_initial_path,
|
|
planned_route_working_path,
|
|
read_planned_route_csv,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/projects", tags=["B08 Milestone"])
|
|
PATH = "/{project_id}/quantity/milestone"
|
|
FOLDER = Path("B08_Quantity") / "milestone"
|
|
DATA_FILE = "milestone.json"
|
|
AUTO = "auto" # 자동 그림 하위 폴더(244-11)
|
|
IMAGE_EXT = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp")
|
|
ROUTE_ID = r"^[A-Za-z0-9_-]{1,40}$"
|
|
|
|
Road = Literal["2차포장", "1차포장", "포장", "비포장"] # 포장 = 기준점 안 거친 구간(244-13)
|
|
Dir = Literal["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
|
|
|
|
|
|
class Leg(BaseModel):
|
|
road: Road | None = "2차포장" # None = 사용자가 안 고름(244-7 · 표에 「도로 구분 고르기」)
|
|
km: float | None = Field(default=None, ge=0, le=10000) # 사용자 값 — 있으면 우선
|
|
auto_km: float | None = Field(default=None, ge=0, le=10000) # [길찾기] 자동 값(km 만 저장)
|
|
|
|
|
|
class Row(BaseModel):
|
|
item: str = Field(default="", max_length=100)
|
|
dest: Literal["start", "mid"] = "start"
|
|
note: str = Field(default="", max_length=200)
|
|
|
|
|
|
class Place(BaseModel):
|
|
name: str = Field(default="", max_length=100)
|
|
address: str = Field(default="", max_length=200)
|
|
dir: Dir | None = None
|
|
lon: float | None = Field(default=None, ge=-180, le=180)
|
|
lat: float | None = Field(default=None, ge=-90, le=90)
|
|
# 153 다이어그램에 끌어 놓은 자리(왼 위 · 그림 좌표) — 없으면 방위 자동 자리
|
|
pos: tuple[float, float] | None = None
|
|
manual: bool = False # 155 수기 입력 — 카카오 길찾기 안 함
|
|
legs: list[Leg] = Field(default_factory=list, max_length=20)
|
|
|
|
|
|
class Route(Place):
|
|
id: str = Field(pattern=ROUTE_ID)
|
|
via_base: bool = True
|
|
linked: bool = True # 155 False = 아직 선 안 이음(요소 추가 뒤)
|
|
rows: list[Row] = Field(default_factory=list, max_length=50)
|
|
|
|
|
|
class Base(Place):
|
|
enabled: bool = False
|
|
|
|
|
|
class Site(BaseModel):
|
|
mid_station_m: float | None = Field(default=None, ge=0, le=1_000_000)
|
|
mid_road: Road = "비포장"
|
|
|
|
|
|
class Milestone(BaseModel):
|
|
site: Site = Field(default_factory=Site)
|
|
base: Base = Field(default_factory=Base)
|
|
routes: list[Route] = Field(default_factory=list, max_length=100)
|
|
|
|
@model_validator(mode="after")
|
|
def _unique_ids(self) -> Milestone:
|
|
ids = [route.id for route in self.routes]
|
|
if len(ids) != len(set(ids)):
|
|
raise ValueError("경로 id 가 겹칩니다.")
|
|
return self
|
|
|
|
|
|
def folder(root: str | Path) -> Path:
|
|
return Path(root) / FOLDER
|
|
|
|
|
|
def read_data(root: str | Path) -> dict[str, Any]:
|
|
"""저장본 — 없거나 깨졌으면 빈 꼴."""
|
|
path = folder(root) / DATA_FILE
|
|
try:
|
|
return Milestone.model_validate_json(path.read_text(encoding="utf-8")).model_dump()
|
|
except (OSError, ValidationError, json.JSONDecodeError):
|
|
return Milestone().model_dump()
|
|
|
|
|
|
def _route_vertices(root: Path) -> list[Any]:
|
|
for path in (
|
|
planned_route_working_path(root),
|
|
planned_route_initial_path(root),
|
|
expected_route_csv_path(root),
|
|
design_route_csv_path(root),
|
|
):
|
|
route = read_planned_route_csv(path) if path.is_file() else None
|
|
if route is not None and len(route.vertices) >= 2:
|
|
return route.vertices
|
|
return []
|
|
|
|
|
|
def _lonlat(root: Path, x: float, y: float) -> list[float] | None:
|
|
"""작업 좌표계 → 경도 · 위도(다이어그램 방위 · 244-3 지도) — 못 하면 None."""
|
|
try:
|
|
from pyproj import Transformer
|
|
|
|
from B04_PreProcess.B04_PreProcess_Engine_Extent import project_epsg_from_prj
|
|
|
|
crs = project_epsg_from_prj(root)
|
|
lon, lat = Transformer.from_crs(crs, "EPSG:4326", always_xy=True).transform(x, y)
|
|
return [round(lon, 7), round(lat, 7)]
|
|
except Exception: # 좌표계 못 읽음 = 좌표 없음(방위는 사용자 8방)
|
|
return None
|
|
|
|
|
|
def site_info(root: str | Path, mid_station_m: float | None) -> dict[str, Any]:
|
|
"""노선 연장 · 중점 기본(연장 ÷ 2) · 시점 · 중점 경위도."""
|
|
root = Path(root)
|
|
vertices = _route_vertices(root)
|
|
if not vertices:
|
|
return {"length_m": None, "mid_default_m": None, "start_lonlat": None, "mid_lonlat": None}
|
|
length = vertices[-1].chainage_m
|
|
mid_default = r3(length / 2)
|
|
mx, my, _ = interpolate_vertex(
|
|
vertices, mid_default if mid_station_m is None else mid_station_m
|
|
)
|
|
return {
|
|
"length_m": r3(length),
|
|
"mid_default_m": mid_default,
|
|
"start_lonlat": _lonlat(root, vertices[0].x, vertices[0].y),
|
|
"mid_lonlat": _lonlat(root, mx, my),
|
|
}
|
|
|
|
|
|
def route_images(root: str | Path, route_id: str, sub: str = "") -> list[Path]:
|
|
path = folder(root) / sub
|
|
if not path.is_dir():
|
|
return []
|
|
return sorted(
|
|
f for f in path.glob(f"{route_id}.*") if f.is_file() and f.suffix.lower() in IMAGE_EXT
|
|
)
|
|
|
|
|
|
def images(root: str | Path, sub: str = "") -> dict[str, str]:
|
|
"""경로 id → 그림 파일 이름(폴더가 목록 · sub = 하위 폴더 · 이름에 붙음)."""
|
|
path = folder(root) / sub
|
|
if not path.is_dir():
|
|
return {}
|
|
return {
|
|
f.stem: f"{sub}/{f.name}" if sub else f.name
|
|
for f in sorted(path.iterdir())
|
|
if f.is_file() and f.suffix.lower() in IMAGE_EXT and not f.name.startswith(".")
|
|
}
|
|
|
|
|
|
def payload(root: str | Path, data: dict[str, Any]) -> dict[str, Any]:
|
|
site = site_info(root, data["site"]["mid_station_m"])
|
|
auto = images(root, AUTO)
|
|
return {
|
|
"data": data,
|
|
"site": site,
|
|
"table": compute_milestone(data, site["mid_default_m"]),
|
|
"images": {**auto, **images(root)}, # 보이는 그림 = 사용자 그림 우선
|
|
"auto_images": auto,
|
|
}
|
|
|
|
|
|
def save_data(root: str | Path, body: Milestone) -> dict[str, Any]:
|
|
data = body.model_dump()
|
|
path = folder(root)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(path / DATA_FILE, data)
|
|
keep = {route["id"] for route in data["routes"]}
|
|
for sub in ("", AUTO):
|
|
for stem, name in images(root, sub).items():
|
|
if stem not in keep:
|
|
(path / name).unlink(missing_ok=True)
|
|
return payload(root, data)
|
|
|
|
|
|
def _route_id(route: str) -> str:
|
|
try:
|
|
return Route(id=route).id
|
|
except ValidationError as error:
|
|
raise HTTPException(status_code=400, detail="경로 id 가 바르지 않습니다.") from error
|
|
|
|
|
|
@router.get(PATH)
|
|
async def milestone_read(project_id: UUID) -> dict[str, Any]:
|
|
root = await _project_root(project_id)
|
|
return await asyncio.to_thread(lambda: payload(root, read_data(root)))
|
|
|
|
|
|
@router.put(PATH)
|
|
async def milestone_save(project_id: UUID, body: Milestone) -> dict[str, Any]:
|
|
root = await _project_root(project_id)
|
|
return await asyncio.to_thread(save_data, root, body)
|
|
|
|
|
|
@router.get(PATH + "/site-address")
|
|
async def milestone_site_address(
|
|
project_id: UUID, mid_m: Annotated[float | None, Query(ge=0)] = None
|
|
) -> dict[str, Any]:
|
|
"""시점 모달 — 프로젝트 등록 때 사업지역 · 시점 · 중점 경위도와 VWorld 주소.
|
|
mid_m = 다이어그램 지금 중점 측점(캐시 · 244-12) · 없으면 저장본."""
|
|
root = await _project_root(project_id)
|
|
data = await asyncio.to_thread(read_data, root)
|
|
mid = data["site"]["mid_station_m"] if mid_m is None else mid_m
|
|
site = await asyncio.to_thread(site_info, root, mid)
|
|
project = await get_project_by_id(str(project_id)) or {}
|
|
|
|
async def where(lonlat: list[float] | None) -> str | None:
|
|
return await reverse_geocode(*lonlat) if lonlat else None
|
|
|
|
start, mid = await asyncio.gather(where(site["start_lonlat"]), where(site["mid_lonlat"]))
|
|
return {
|
|
"region": project.get("region") or "",
|
|
"start_lonlat": site["start_lonlat"],
|
|
"mid_lonlat": site["mid_lonlat"],
|
|
"start_address": start,
|
|
"mid_address": mid,
|
|
}
|
|
|
|
|
|
@router.get(PATH + "/reverse")
|
|
async def milestone_reverse(
|
|
project_id: UUID,
|
|
lon: Annotated[float, Query(ge=-180, le=180)],
|
|
lat: Annotated[float, Query(ge=-90, le=90)],
|
|
) -> dict[str, Any]:
|
|
"""지도에서 고른 소재지(244-13) — 좌표 → VWorld 주소(못 찾음 = null)."""
|
|
return {"address": await reverse_geocode(lon, lat)}
|
|
|
|
|
|
@router.post(PATH + "/image")
|
|
async def milestone_image_upload(
|
|
project_id: UUID, route: str = Query(...), file: UploadFile = File(...)
|
|
) -> dict[str, Any]:
|
|
route_id = _route_id(route)
|
|
ext = Path(file.filename or "").suffix.lower()
|
|
if ext not in IMAGE_EXT:
|
|
raise HTTPException(status_code=400, detail="그림 파일(png · jpg · gif · webp · bmp)만.")
|
|
root = await _project_root(project_id)
|
|
path = folder(root)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
await save_upload_stream(file, path / f".{route_id}{ext}.new")
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=400, detail=str(error)) from error
|
|
for old in route_images(root, route_id):
|
|
old.unlink(missing_ok=True)
|
|
(path / f".{route_id}{ext}.new").replace(path / f"{route_id}{ext}")
|
|
return {"name": f"{route_id}{ext}"}
|
|
|
|
|
|
@router.get(PATH + "/image")
|
|
async def milestone_image(project_id: UUID, route: str = Query(...)):
|
|
route_id = _route_id(route)
|
|
root = await _project_root(project_id)
|
|
found = route_images(root, route_id) + route_images(root, route_id, AUTO)
|
|
if not found:
|
|
raise HTTPException(status_code=404, detail="경로 그림이 없습니다.")
|
|
return FileResponse(found[0]) # 사용자 그림 우선
|
|
|
|
|
|
@router.delete(PATH + "/image")
|
|
async def milestone_image_delete(
|
|
project_id: UUID, route: str = Query(...), user_only: bool = False
|
|
) -> dict[str, Any]:
|
|
route_id = _route_id(route)
|
|
root = await _project_root(project_id)
|
|
auto = route_images(root, route_id, AUTO)
|
|
for old in route_images(root, route_id) + ([] if user_only else auto):
|
|
old.unlink(missing_ok=True)
|
|
return {"name": f"{AUTO}/{auto[0].name}" if user_only and auto else None}
|
|
|
|
|
|
class Mark(BaseModel):
|
|
name: str = Field(default="", max_length=100)
|
|
lon: float = Field(ge=-180, le=180)
|
|
lat: float = Field(ge=-90, le=90)
|
|
|
|
|
|
Lat = Annotated[float, Field(ge=-90, le=90)]
|
|
Lon = Annotated[float, Field(ge=-180, le=180)]
|
|
|
|
|
|
class AutoImage(BaseModel):
|
|
"""길 선 = 구간마다 [위도, 경도] 목록(마지막 [길찾기] 응답) · 표지 = 거래처 · 기준점 · 시점."""
|
|
|
|
lines: list[Annotated[list[tuple[Lat, Lon]], Field(max_length=20000)]] = Field(
|
|
default_factory=list, max_length=20
|
|
)
|
|
marks: list[Mark] = Field(default_factory=list, max_length=10)
|
|
|
|
|
|
@router.post(PATH + "/image/auto")
|
|
async def milestone_image_auto(
|
|
project_id: UUID, body: AutoImage, route: str = Query(...)
|
|
) -> dict[str, Any]:
|
|
"""경로 그림 자동(244-11) — 서버 헤드리스 캡처 → auto/<경로 id>.png · 좌표는 저장 안 함."""
|
|
route_id = _route_id(route)
|
|
root = await _project_root(project_id)
|
|
lines = [[list(p) for p in line] for line in body.lines]
|
|
marks = [m.model_dump() for m in body.marks]
|
|
got = await asyncio.to_thread(make_route_image, folder(root) / AUTO, route_id, lines, marks)
|
|
return {"name": f"{AUTO}/{route_id}.png" if got["saved"] else None, "reason": got["reason"]}
|
|
|
|
|
|
KAKAO_URL = "https://apis-navi.kakaomobility.com/v1/directions"
|
|
KAKAO_KEY = "KAKAO_REST_API_KEY"
|
|
|
|
|
|
class LonLat(BaseModel):
|
|
lon: float = Field(ge=-180, le=180)
|
|
lat: float = Field(ge=-90, le=90)
|
|
|
|
|
|
class Directions(BaseModel):
|
|
origin: LonLat
|
|
destination: LonLat
|
|
|
|
|
|
def _kakao_open(request: urllib.request.Request) -> bytes:
|
|
with urllib.request.urlopen(request, timeout=8) as response:
|
|
return response.read()
|
|
|
|
|
|
def directions_km(origin: LonLat, destination: LonLat) -> dict[str, Any]:
|
|
"""카카오 길찾기 → {km, reason, path} — km = 총 거리 ÷ 1000 소수 1자리 · 실패 = reason 한 줄 ·
|
|
path = 경로 좌표 [위도, 경도] 목록(성공 때만 · 지도 길 선 · 저장 안 함).
|
|
키 · 응답 원문은 돌려주지도 적지도 않음(까닭 = 고정 글 · 카카오 result_msg 만)."""
|
|
key = os.environ.get(KAKAO_KEY, "").strip()
|
|
if not key:
|
|
return {
|
|
"km": None,
|
|
"reason": f"카카오 길찾기 키가 없습니다(.env {KAKAO_KEY}) — km 를 손으로 넣으세요.",
|
|
}
|
|
query = urllib.parse.urlencode(
|
|
{
|
|
"origin": f"{origin.lon},{origin.lat}",
|
|
"destination": f"{destination.lon},{destination.lat}",
|
|
"priority": "RECOMMEND",
|
|
}
|
|
)
|
|
request = urllib.request.Request(
|
|
f"{KAKAO_URL}?{query}", headers={"Authorization": f"KakaoAK {key}"}
|
|
)
|
|
try:
|
|
body = json.loads(_kakao_open(request).decode("utf-8"))
|
|
except urllib.error.HTTPError as error:
|
|
reason = {
|
|
401: "카카오가 키를 받지 않았습니다(키 · 앱 설정 확인)",
|
|
403: "카카오가 길찾기를 막았습니다(앱 · 권한 확인)",
|
|
429: "오늘 카카오 길찾기 요청 수를 넘었습니다",
|
|
}.get(error.code, f"카카오 길찾기 응답 오류(HTTP {error.code})")
|
|
return {"km": None, "reason": f"{reason} — km 를 손으로 넣으세요."}
|
|
except (
|
|
Exception
|
|
): # 끊김 · 시간 넘김 · 깨진 응답 — 원문 · 예외 글은 내지 않음(키가 섞일 수 있음)
|
|
return {"km": None, "reason": "카카오 길찾기에 닿지 못했습니다 — km 를 손으로 넣으세요."}
|
|
route = ((body or {}).get("routes") or [{}])[0] if isinstance(body, dict) else {}
|
|
distance = (route.get("summary") or {}).get("distance")
|
|
if route.get("result_code") != 0 or not isinstance(distance, (int, float)):
|
|
msg = str(route.get("result_msg") or "결과 없음")[:80]
|
|
return {"km": None, "reason": f"길찾기 실패 — {msg} · km 를 손으로 넣으세요."}
|
|
path: list[list[float]] = []
|
|
try: # vertexes = 경도 · 위도 번갈아(도로마다)
|
|
for section in route.get("sections") or []:
|
|
for road in section.get("roads") or []:
|
|
v = [float(n) for n in road.get("vertexes") or []]
|
|
path += [[lat, lon] for lon, lat in zip(v[0::2], v[1::2])]
|
|
except (TypeError, ValueError, AttributeError):
|
|
path = [] # 깨진 좌표 = 길 선만 없음(km 는 그대로)
|
|
return {"km": math.floor(distance / 100 + 0.5) / 10, "reason": None, "path": path}
|
|
|
|
|
|
@router.post(PATH + "/directions")
|
|
async def milestone_directions(project_id: UUID, body: Directions) -> dict[str, Any]:
|
|
"""[길찾기] 한 구간 — 단추 누를 때만(조작 중 자동 호출 없음)."""
|
|
return await asyncio.to_thread(directions_km, body.origin, body.destination)
|
|
|
|
|
|
KAKAO_JS = "KAKAO_JS_KEY"
|
|
|
|
|
|
@router.get(PATH + "/map-key")
|
|
async def milestone_map_key(project_id: UUID) -> dict[str, Any]:
|
|
"""카카오 지도 JavaScript 키(공개용 · 도메인 등록으로 막힘) — 없으면 null = 화면 안내 한 줄."""
|
|
return {"key": os.environ.get(KAKAO_JS, "").strip() or None}
|