632 lines
28 KiB
Python
632 lines
28 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` ·
|
|
좌표는 그림 뒤 버림 · 실패 = 까닭 한 줄
|
|
POST /api/projects/{id}/quantity/milestone/place-image?id= 장소 그림(256-1) — 상자 좌표 ·
|
|
이름 → 작은 지도 캡처 → `place/<상자 id>.png` · 등록 · 수정 · 끌기 뒤 화면이 부름
|
|
GET /api/projects/{id}/quantity/milestone/file?name= 그림 파일(256-1) — `place/<id>.png`
|
|
· `auto/<id>.png` · `<id>.<확장자>`(하위 폴더 둘만 · 경로 넘기 막음)
|
|
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 견적서 파일 길 꼴) · 보이는 그림 = 사용자
|
|
그림 우선. 초기값 층 없음(자동설계가 낳는 값 아님).
|
|
그림은 올리는 즉시 서버에([저장]과 따로) · [저장] 때 없는 경로의 그림은 지움.
|
|
장소 그림(256-1) = `place/<상자 id>.png` · [저장] 뒤 서버가 좌표가 바뀐(또는 그림 없는) 상자만
|
|
다시 캡처(뒤에서 · 지문 = `.prints.json` · 좌표는 해시만) · 없는 상자 그림은 [저장] 때 지움 ·
|
|
GET 답 `place_images` {상자 id: "place/<id>.png"} · `place_times` {상자 id: 바뀐 시각 ms}
|
|
(캐시 깨기).
|
|
경로 그림(256-2) = [저장] 뒤 같은 길로 경로 탭과 같은 전체 경로(거래처 → 기준점 사슬 → 시점 ·
|
|
구간마다 카카오 길 선 · 수기 · 실패 구간 = 곧은 선 · 표지 = `mapMarks` 차례) → `auto/<경로 id>.png`
|
|
· 지문(표지 · 수기) 바뀐 경로만 · 화면 [길찾기] 그림(거래처 구간만)이 들어오면 지문을 지워 다음
|
|
[저장]에 전체로.
|
|
공사지 = 시점(노선 첫 점 · 측점 0) · 중점 기본 = 노선 연장 ÷ 2 — 노선은 설계 노선 정본 차례
|
|
(수정본 → 초기 폴리라인 → 예상노선 → 초기값 · `load_design_route` 와 같은 차례).
|
|
표 합 = `B08_Quantity_Milestone_Calc`(파이썬 · TS 짝).
|
|
기준점 = `bases[]`(254 · 여럿 · to = 시점 또는 다른 기준점) · 경로 `to` — 옛 `base` · `via_base` 는
|
|
읽을 때 · 받을 때 새 꼴로 옮김(`migrate_milestone` · 저장 = 새 꼴).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import threading
|
|
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, BackgroundTasks, 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_Haul import haul_groups, with_haul_routes
|
|
from B08_Quantity.B08_Quantity_Milestone_Calc import (
|
|
SITE,
|
|
base_chain,
|
|
compute_milestone,
|
|
migrate_milestone,
|
|
r3,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Milestone_Capture import make_place_image, make_route_image
|
|
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
|
|
from B08_Quantity.B08_Quantity_Router_Material import material_sheets
|
|
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)
|
|
PLACE = "place" # 장소 그림 하위 폴더(256-1)
|
|
PRINTS = ".prints.json" # 그림 지문 {"place/<id>": 해시} — 바뀐 것만 다시 캡처
|
|
IMAGE_EXT = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp")
|
|
ROUTE_ID = r"^[A-Za-z0-9_-]{1,40}$"
|
|
BOX_IDS = {SITE, "start", "mid"} # 다이어그램 공사지 상자 id — 기준점 · 경로 id 로 못 씀
|
|
|
|
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)
|
|
to: str = Field(default=SITE, pattern=ROUTE_ID) # 254 잇는 곳 = "site"(시점) · 기준점 id
|
|
# 251 운반 묶음 키(총괄자재 · B09 운반 줄과 이음) · 단가연결(dg · sg · logic 꼴 · 정본)
|
|
haul: str | None = Field(default=None, max_length=60, pattern=r"^[A-Za-z0-9_-]+$")
|
|
ref: dict[str, Any] | None = None
|
|
linked: bool = True # 155 False = 아직 선 안 이음(요소 추가 뒤)
|
|
rows: list[Row] = Field(default_factory=list, max_length=50)
|
|
|
|
|
|
class Base(Place):
|
|
"""기준점(254 · 여럿) — to = 시점 또는 다른 기준점 id(돌기는 화면이 막고 셈이 끊음)."""
|
|
|
|
id: str = Field(pattern=ROUTE_ID)
|
|
to: str = Field(default=SITE, pattern=ROUTE_ID)
|
|
linked: bool = True # False = 아직 선 안 이음(요소 추가 뒤)
|
|
|
|
|
|
class LonLat(BaseModel):
|
|
lon: float = Field(ge=-180, le=180)
|
|
lat: float = Field(ge=-90, le=90)
|
|
|
|
|
|
class Site(BaseModel):
|
|
mid_station_m: float | None = Field(default=None, ge=0, le=1_000_000)
|
|
mid_road: Road = "비포장"
|
|
start_pin: LonLat | None = None # 255-2 길찾기용 시점 덮개(노선은 그대로)
|
|
|
|
|
|
class Milestone(BaseModel):
|
|
site: Site = Field(default_factory=Site)
|
|
bases: list[Base] = Field(default_factory=list, max_length=20)
|
|
routes: list[Route] = Field(default_factory=list, max_length=100)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def _migrate(cls, data: Any) -> Any:
|
|
"""옛 꼴(단일 base · via_base) → 새 꼴(254)."""
|
|
return migrate_milestone(data) if isinstance(data, dict) else data
|
|
|
|
@model_validator(mode="after")
|
|
def _unique_ids(self) -> Milestone:
|
|
ids = [b.id for b in self.bases] + [route.id for route in self.routes]
|
|
if len(ids) != len(set(ids)):
|
|
raise ValueError("경로 · 기준점 id 가 겹칩니다.")
|
|
if BOX_IDS & set(ids):
|
|
raise ValueError("site · start · mid 는 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 places(data: dict[str, Any]) -> list[tuple[str, str, float, float]]:
|
|
"""좌표가 정해진 상자 → [(상자 id, 이름, 경도, 위도)] — 거래처 = 경로 id · 기준점 = bases id."""
|
|
boxes = [
|
|
(b["id"], b.get("name") or "", b.get("lon"), b.get("lat"))
|
|
for b in [*data.get("routes", []), *data.get("bases", [])]
|
|
]
|
|
return [(i, n, lon, lat) for i, n, lon, lat in boxes if lon is not None and lat is not None]
|
|
|
|
|
|
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)
|
|
place = images(root, PLACE)
|
|
return {
|
|
"data": data,
|
|
"site": site,
|
|
"table": compute_milestone(data, site["mid_default_m"]),
|
|
"images": {**auto, **images(root)}, # 보이는 그림 = 사용자 그림 우선
|
|
"auto_images": auto,
|
|
"place_images": place,
|
|
"place_times": {
|
|
stem: int((folder(root) / name).stat().st_mtime * 1000) for stem, name in place.items()
|
|
},
|
|
}
|
|
|
|
|
|
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, ids in (("", keep), (AUTO, keep), (PLACE, {p[0] for p in places(data)})):
|
|
for stem, name in images(root, sub).items():
|
|
if stem not in ids:
|
|
(path / name).unlink(missing_ok=True)
|
|
return payload(root, data)
|
|
|
|
|
|
def _print(*parts: Any) -> str:
|
|
"""지문 — 좌표 · 이름 해시(좌표 원문은 안 남김)."""
|
|
return hashlib.sha1(json.dumps(parts, ensure_ascii=False).encode()).hexdigest()[:16]
|
|
|
|
|
|
def _read_prints(root: str | Path) -> dict[str, str]:
|
|
try:
|
|
return json.loads((folder(root) / PRINTS).read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def _set_print(root: str | Path, key: str, value: str | None) -> None:
|
|
"""지문 한 칸 적음 · None = 지움(다음 [저장]에 다시 캡처)."""
|
|
with _prints_lock:
|
|
prints = _read_prints(root)
|
|
if prints.get(key) == value:
|
|
return
|
|
if value is None:
|
|
prints.pop(key, None)
|
|
else:
|
|
prints[key] = value
|
|
atomic_write_json(folder(root) / PRINTS, prints)
|
|
|
|
|
|
_prints_lock = threading.Lock() # ponytail: 서버 전체 한 줄 — 지문 파일 읽고 쓰기 한 번에 하나
|
|
|
|
|
|
def place_image(root: str | Path, place_id: str, name: str, lon: float, lat: float) -> dict:
|
|
"""장소 그림 한 장 캡처 → 지문 적음 → {saved, reason}."""
|
|
got = make_place_image(folder(root) / PLACE, place_id, {"name": name, "lon": lon, "lat": lat})
|
|
if got["saved"]:
|
|
_set_print(root, f"{PLACE}/{place_id}", _print(name, lon, lat))
|
|
return got
|
|
|
|
|
|
def route_chain(data: dict[str, Any], route: dict[str, Any], start: list[float] | None) -> list:
|
|
"""경로 그림 사슬 — [(이름, 경도, 위도, 수기)] · 화면 `mapMarks` 와 같은 차례 ·
|
|
좌표 없는 상자 뺌."""
|
|
owners = [(route.get("name") or "거래처", route)]
|
|
if route.get("linked", True) is not False:
|
|
owners += [
|
|
(b.get("name") or "기준점", b) for b in base_chain(data, route.get("to") or SITE)
|
|
]
|
|
out = [(n, o.get("lon"), o.get("lat"), bool(o.get("manual"))) for n, o in owners]
|
|
if start:
|
|
out.append(("시점", start[0], start[1], False))
|
|
return [c for c in out if c[1] is not None and c[2] is not None]
|
|
|
|
|
|
def route_image(root: str | Path, route_id: str, chain: list) -> dict[str, Any]:
|
|
"""경로 그림 한 장 — 구간마다 카카오 길 선(수기 · 실패 = 곧은 선) → 캡처 → 지문 적음."""
|
|
lines = []
|
|
for (_, lon1, lat1, manual), (_, lon2, lat2, _) in zip(chain, chain[1:]):
|
|
path = []
|
|
if not manual:
|
|
got = directions_km(LonLat(lon=lon1, lat=lat1), LonLat(lon=lon2, lat=lat2))
|
|
path = got.get("path") or []
|
|
lines.append(path or [[lat1, lon1], [lat2, lon2]])
|
|
marks = [{"name": n, "lon": lon, "lat": lat} for n, lon, lat, _ in chain]
|
|
got = make_route_image(folder(root) / AUTO, route_id, lines, marks)
|
|
if got["saved"]:
|
|
_set_print(root, f"{AUTO}/{route_id}", _print(chain))
|
|
return got
|
|
|
|
|
|
def refresh_images(root: str | Path) -> None:
|
|
"""[저장] 뒤(뒤에서) — 지문이 바뀌었거나 그림이 없는 상자만 다시 캡처 ·
|
|
실패 = 다음 [저장]에 다시."""
|
|
data = read_data(root)
|
|
prints = _read_prints(root)
|
|
have = images(root, PLACE)
|
|
for place_id, name, lon, lat in places(data):
|
|
if prints.get(f"{PLACE}/{place_id}") != _print(name, lon, lat) or place_id not in have:
|
|
place_image(root, place_id, name, lon, lat)
|
|
start = site_info(root, None)["start_lonlat"]
|
|
have = images(root, AUTO)
|
|
for route in data["routes"]:
|
|
chain = route_chain(data, route, start)
|
|
key = f"{AUTO}/{route['id']}"
|
|
if route.get("linked", True) is False or len(chain) < 2:
|
|
continue # 안 이은 거래처 = 경로 그림 없음
|
|
if prints.get(key) != _print(chain) or route["id"] not in have:
|
|
route_image(root, route["id"], chain)
|
|
|
|
|
|
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
|
|
|
|
|
|
async def _haul(project_id: UUID) -> list[dict[str, Any]]:
|
|
"""운반 묶음(251) — 자재 집계 못 읽으면 빈 목록(거리이정표는 그대로)."""
|
|
try:
|
|
return haul_groups(await material_sheets(project_id))
|
|
except Exception: # noqa: BLE001 — 자재 집계 없음 · 실패 = 자동 상자 없음
|
|
return []
|
|
|
|
|
|
@router.get(PATH)
|
|
async def milestone_read(project_id: UUID) -> dict[str, Any]:
|
|
root = await _project_root(project_id)
|
|
groups = await _haul(project_id)
|
|
return await asyncio.to_thread(lambda: payload(root, with_haul_routes(read_data(root), groups)))
|
|
|
|
|
|
@router.put(PATH)
|
|
async def milestone_save(
|
|
project_id: UUID, body: Milestone, tasks: BackgroundTasks
|
|
) -> dict[str, Any]:
|
|
root = await _project_root(project_id)
|
|
out = await asyncio.to_thread(save_data, root, body)
|
|
if not os.environ.get(
|
|
"PYTEST_CURRENT_TEST"
|
|
): # 남의 시험 [저장] = 실제 카카오 · 브라우저 안 부름
|
|
tasks.add_task(refresh_images, root) # 답 뒤 캡처(상자마다 2~3초) — 다음 GET 에 보임
|
|
return out
|
|
|
|
|
|
@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)
|
|
if got["saved"]: # 거래처 구간만 그림 — 다음 [저장]에 전체 경로로 다시(256-2)
|
|
await asyncio.to_thread(_set_print, root, f"{AUTO}/{route_id}", None)
|
|
return {"name": f"{AUTO}/{route_id}.png" if got["saved"] else None, "reason": got["reason"]}
|
|
|
|
|
|
FILE_NAME = r"^(?:(?:auto|place)/)?[A-Za-z0-9_-]{1,40}\.(?:png|jpe?g|gif|webp|bmp)$"
|
|
|
|
|
|
@router.post(PATH + "/place-image")
|
|
async def milestone_place_image(project_id: UUID, body: Mark, id: str = Query(...)) -> dict:
|
|
"""장소 그림(256-1) — 등록 · 수정 · 끌기 뒤 화면이 부름 → place/<id>.png · 좌표는 저장 안 함."""
|
|
place_id = _route_id(id)
|
|
root = await _project_root(project_id)
|
|
got = await asyncio.to_thread(place_image, root, place_id, body.name, body.lon, body.lat)
|
|
if not got["saved"]:
|
|
return {"name": None, "time": None, "reason": got["reason"]}
|
|
name = f"{PLACE}/{place_id}.png"
|
|
time_ms = int((folder(root) / name).stat().st_mtime * 1000)
|
|
return {"name": name, "time": time_ms, "reason": None}
|
|
|
|
|
|
@router.get(PATH + "/file")
|
|
async def milestone_file(project_id: UUID, name: str = Query(..., pattern=FILE_NAME)):
|
|
"""그림 파일(256-1) — 하위 폴더 auto · place 만(패턴이 경로 넘기 막음)."""
|
|
root = await _project_root(project_id)
|
|
path = folder(root) / name
|
|
if not path.is_file():
|
|
raise HTTPException(status_code=404, detail="그림이 없습니다.")
|
|
return FileResponse(path)
|
|
|
|
|
|
KAKAO_URL = "https://apis-navi.kakaomobility.com/v1/directions"
|
|
KAKAO_KEY = "KAKAO_REST_API_KEY"
|
|
|
|
|
|
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}
|