diff --git a/B01_Dashboard/B01_Dashboard_Map.py b/B01_Dashboard/B01_Dashboard_Map.py index a692c3c6c..dbba6349e 100644 --- a/B01_Dashboard/B01_Dashboard_Map.py +++ b/B01_Dashboard/B01_Dashboard_Map.py @@ -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( diff --git a/common_util/common_util_map_router.py b/common_util/common_util_map_router.py index 76ef2b420..77bc724ce 100644 --- a/common_util/common_util_map_router.py +++ b/common_util/common_util_map_router.py @@ -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} diff --git a/resources/tester/test_563_2_plan_pieces.py b/resources/tester/test_563_2_plan_pieces.py index 3fd859add..113d2909b 100644 --- a/resources/tester/test_563_2_plan_pieces.py +++ b/resources/tester/test_563_2_plan_pieces.py @@ -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", "용화본선", "용화본선"] diff --git a/resources/tester/test_594_4_map_search.py b/resources/tester/test_594_4_map_search.py new file mode 100644 index 000000000..f11ef0f90 --- /dev/null +++ b/resources/tester/test_594_4_map_search.py @@ -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"] == []