feat(공용 지도): 주소 · 장소 찾기 길 GET /api/map/search · 선 이름 내보내기 이음 시험 (PLAN 594-2 · 594-4 · 요구 463)

- /api/map/search?q= — VWorld 주소(도로명 → 지번) 먼저 · 장소(POI) 뒤 · 같은 이름 · 주소 줄 하나 · 키는 서버 · 실패 = 빈 목록
- B01_Dashboard_Map.search_place 더함(있는 search_address 그대로)
- 시험 594_4 새로(VWorld 가짜) · 563_2 붙인 선 이름 = 조각 이름표 · zip 폴더 · 좌표표 xlsx 「선」 칸

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-30 14:42:38 +09:00
co-authored by Claude Opus 5.5
parent bf77b70be9
commit 52a9fc4784
4 changed files with 189 additions and 1 deletions
+45
View File
@@ -116,6 +116,51 @@ async def search_address(query: str, size: int = 10) -> list[dict[str, Any]]:
return await asyncio.to_thread(_search_sync, query.strip(), size)
def _search_place_sync(query: str, size: int) -> list[dict[str, Any]]:
"""장소 이름(POI) 찾기(594-4) — {title, road, parcel, lon, lat} · 실패 = 빈 목록."""
params = urllib.parse.urlencode(
{
"service": "search",
"request": "search",
"version": "2.0",
"crs": "EPSG:4326",
"size": str(size),
"page": "1",
"query": query,
"type": "place",
"format": "json",
"errorformat": "json",
"key": VWORLD_API_KEY,
}
)
try:
body = json.loads(_fetch(f"{_SEARCH_URL}?{params}", timeout=8).decode("utf-8"))
except Exception:
return []
found = []
for item in ((body.get("response") or {}).get("result") or {}).get("items") or []:
address = item.get("address") or {}
point = item.get("point") or {}
try:
lon, lat = float(point["x"]), float(point["y"])
except (KeyError, TypeError, ValueError):
continue
found.append(
{
"title": item.get("title") or "",
"road": address.get("road") or "",
"parcel": address.get("parcel") or "",
"lon": lon,
"lat": lat,
}
)
return found
async def search_place(query: str, size: int = 10) -> list[dict[str, Any]]:
return await asyncio.to_thread(_search_place_sync, query.strip(), size)
def _address_items(lon: float, lat: float) -> dict[str, dict[str, Any]]:
"""좌표 → VWorld getAddress 결과(종류 road · parcel → 한 칸) · 실패 = 빈 dict."""
query = urllib.parse.urlencode(
+36 -1
View File
@@ -2,16 +2,19 @@
GET /api/map/kakao-key {key} — 공개용(도메인 등록으로 막힘) · .env KAKAO_JS_KEY · 없으면 null
GET /api/map/address {address, region, source} — VWorld(B08 과 같은 길) · 지번 먼저 · 없음 null
GET /api/map/search?q= {items: [{kind address|place, title, address, lon, lat}], source} — 594-4 ·
등록(D01) · 예정노선 위치 칸이 같이 씀 · VWorld 주소 + 장소
"""
from __future__ import annotations
import asyncio
import os
from typing import Any
from fastapi import APIRouter, Query
from B01_Dashboard.B01_Dashboard_Map import reverse_place
from B01_Dashboard.B01_Dashboard_Map import reverse_place, search_address, search_place
router = APIRouter(prefix="/api/map", tags=["map"])
@@ -31,3 +34,35 @@ async def map_address(
) -> dict[str, Any]:
"""좌표 → 주소 · 「시군구 읍면」(사업 지역 채움) · 출처."""
return {**await reverse_place(lon, lat), "source": ADDRESS_SOURCE}
@router.get("/search")
async def map_search(
q: str = Query(..., min_length=1, max_length=100), size: int = Query(10, ge=1, le=30)
) -> dict[str, Any]:
"""주소 · 장소 찾기(594-4) — 주소(도로명 → 지번) 먼저 · 장소 뒤 · 같은 줄 하나 · 키는 서버."""
query = q.strip()
if not query:
return {"items": [], "source": ADDRESS_SOURCE}
addresses, places = await asyncio.gather(search_address(query, size), search_place(query, size))
items = [
{
"kind": "address",
"title": hit[hit["category"]] or hit["road"] or hit["parcel"],
"address": hit["parcel"] if hit["category"] == "road" else hit["road"],
"lon": hit["lon"],
"lat": hit["lat"],
}
for hit in addresses
] + [
{
"kind": "place",
"title": hit["title"],
"address": hit["road"] or hit["parcel"],
"lon": hit["lon"],
"lat": hit["lat"],
}
for hit in places
]
unique = {(item["title"], item["address"]): item for item in items} # 같은 줄 하나(차례 그대로)
return {"items": list(unique.values()), "source": ADDRESS_SOURCE}
@@ -3,6 +3,7 @@
견본 = 같은 연도 조각 둘(본선 p1 · 갈래 p3, 2026) + 본선 p2(2027) · 업체 둘(p1 = 회사 5 · p3 = 회사 6).
- D04 요약 = 조각 줄 · 이름 「본선 0+0~12+10」 · 지정 회사 = 자기 조각 · 자기 행만
- 594-2 붙인 선 이름 = 조각 이름표 · zip 폴더 · 좌표표 xlsx 「선」 칸
- 내보내기 = 조각마다 폴더 · 전체 = 선마다 레코드 · 지정 회사 = 확정된 자기 조각만
- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘)
- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행
@@ -21,6 +22,7 @@ from pathlib import Path
from uuid import uuid4
import geopandas as gpd
import openpyxl
import pytest
from fastapi import HTTPException
@@ -280,3 +282,27 @@ def test_assigned_list_piece_label_length(plan, monkeypatch):
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),
]
def test_line_name_flows_to_labels_zip_xlsx(plan):
"""594-2 붙인 선 이름 = 조각 이름표 · zip 폴더 · 좌표표 xlsx 「선」 칸(빈칸 뺌)."""
named = [{**LINES[0], "name": "용화 본선"}, LINES[1]]
plan["file"]["plan"] = plan_of(named, MARKS, PIECES)
rows = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))["pieces"]
assert [r["label"] for r in rows] == [
"용화 본선 0+0~12+10",
"갈래 1 0+0~6+10",
"용화 본선 12+10~25+19.3",
]
body = asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body
with zipfile.ZipFile(io.BytesIO(body)) as z:
assert {n.split("/")[0] for n in z.namelist()} == {
"2026_용화본선_p1",
"2026_갈래1_p3",
"2027_용화본선_p2",
"전체",
}
book = next(n for n in z.namelist() if n.startswith("전체/") and n.endswith(".xlsx"))
sheet = openpyxl.load_workbook(io.BytesIO(z.read(book))).active
lines = [row[14] for row in sheet.iter_rows(min_row=3, values_only=True)]
assert sorted(lines) == ["갈래1", "용화본선", "용화본선"]
+82
View File
@@ -0,0 +1,82 @@
"""594-4 공용 주소 · 장소 찾기 길(`GET /api/map/search`) — VWorld 가짜 응답.
- 주소(도로명 → 지번) 먼저 · 장소 뒤 · 같은 이름 · 주소 줄 하나 · 키는 서버(응답에 없음).
- 빈 값 = 빈 목록 · 100 자 넘음 = 422 · VWorld 실패 = 빈 목록.
./venv/Scripts/python.exe -m pytest resources/tester/test_594_4_map_search.py -q
"""
from __future__ import annotations
import json
from urllib.parse import parse_qs, urlparse
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from B01_Dashboard import B01_Dashboard_Map as vmap
from common_util import common_util_map_router as map_router
ADDRESS = {
"address": {"zipcode": "36509", "road": "", "parcel": "경상북도 영양군 일월면 용화리 1"},
"point": {"x": "129.165", "y": "36.815"},
}
PLACE = {
"title": "용화보건진료소",
"address": {"road": "경상북도 영양군 일월면 영양로 3773", "parcel": "… 용화리 225-1"},
"point": {"x": "129.150", "y": "36.789"},
}
@pytest.fixture
def api(monkeypatch):
asked = []
def fetch(url, timeout=5):
query = {k: v[0] for k, v in parse_qs(urlparse(url).query).items()}
asked.append(query)
if query.get("query") == "고장":
raise OSError("down")
if query["type"] == "address":
items = [ADDRESS] if query["category"] == "parcel" else []
else:
items = [PLACE, PLACE] # VWorld 가 같은 장소를 두 번 줌
return json.dumps({"response": {"status": "OK", "result": {"items": items}}}).encode()
monkeypatch.setattr(vmap, "_fetch", fetch)
app = FastAPI()
app.include_router(map_router.router)
client = TestClient(app)
client.asked = asked
return client
def test_address_then_place(api) -> None:
got = api.get("/api/map/search", params={"q": " 용화리 "})
assert got.status_code == 200
items = got.json()["items"]
assert items == [
{
"kind": "address",
"title": "경상북도 영양군 일월면 용화리 1",
"address": "",
"lon": 129.165,
"lat": 36.815,
},
{
"kind": "place",
"title": "용화보건진료소",
"address": "경상북도 영양군 일월면 영양로 3773",
"lon": 129.15,
"lat": 36.789,
},
]
assert {q["query"] for q in api.asked} == {"용화리"} # 앞뒤 빈칸 뺌
assert "key" not in got.text # 키는 서버만
def test_empty_long_and_failure(api) -> None:
assert api.get("/api/map/search", params={"q": " "}).json()["items"] == []
assert api.get("/api/map/search", params={"q": "가" * 101}).status_code == 422
assert api.get("/api/map/search", params={"q": "고장"}).json()["items"] == []