o2o-infinith-demo/scripts/fetch_medical_tourism.py

#!/usr/bin/env python3
"""의료관광 연결 데이터 수집 (한국관광공사 TourAPI 다국어).

병원 좌표를 중심으로 숙박·음식점·관광지·문화시설·쇼핑을 모으고, 진행 중인 축제를
거리와 함께 붙여 프론트가 쓸 JSON 하나로 낸다.

왜 사전 수집인가
  영문 서비스는 강남 일대 커버리지가 얇고(5km 음식점 18건, 국문은 338건) 관광지의
  79%가 의료기관이라 필터링이 많이 필요하다. 그 처리를 수집 시점에 끝내고 프론트는
  정제된 결과만 그린다. 미팅 데모 중 외부 API 상태에 화면이 종속되지도 않는다.

무엇을 만들지 않는가
  없는 값을 채우지 않는다. 평점·리뷰·영업시간은 TourAPI 가 주지 않으므로 비워 두고,
  Google Places 승인 후 별도 단계에서 붙인다. 식이 적합성은 병원 입력이며 여기서
  판정하지 않는다.

  python3 scripts/fetch_medical_tourism.py
  python3 scripts/fetch_medical_tourism.py --radius 5000 --out src/data/medicalTourism.json
"""
import argparse, io, json, math, os, ssl, sys, time, urllib.parse, urllib.request

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# 영문 관광지(76)의 대부분은 관광지가 아니라 의료관광 등록업소다. 반경 5km 100건 중 92건이
# 분류코드 A02020500 = "Medical Tourism Sites"(categoryCode2 로 확인)이며 병원과 유치 에이전시가
# 여기 들어 있다. 제목 키워드로는 "하이안과", "닥파인더코리아" 같은 것을 놓치므로 코드로 거른다.
CAT_MEDICAL_TOURISM = "A02020500"
# 스파·찜질방은 수술 후 회복 여정에 맞으므로 관광이 아니라 별도 범주로 뺀다.
CAT_WELLNESS = ("A02020300", "A02020400")  # Hot Springs & Spa, Jjimjilbang

# 다국어 ContentTypeId. 국문과 코드가 다르다.
# 반경은 목적에 따라 다르다. 회복기에 매일 가는 곳(숙박·식사·스파)은 가깝게, 관광은 넓게 본다.
TYPES = {
    "76": ("attraction", "10000"),
    "78": ("culture", "10000"),
    "79": ("shopping", "10000"),
    "80": ("stay", "5000"),
    "82": ("restaurant", "5000"),
}


def ssl_ctx():
    try:
        import certifi; return ssl.create_default_context(cafile=certifi.where())
    except Exception:
        return ssl.create_default_context()


def load_env(path):
    d = {}
    for line in io.open(path, encoding="utf-8"):
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1); d[k.strip()] = v.strip()
    return d


def haversine_km(lat1, lng1, lat2, lng2):
    r = 6371.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp, dl = math.radians(lat2 - lat1), math.radians(lng2 - lng1)
    a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 2 * r * math.asin(math.sqrt(a))


def call(env, op, extra, rows=100, retries=3, page=1):
    base, svc, key = env["TOUR_API_BASE"], env["TOUR_API_SERVICE"], env["TOUR_API_KEY"]
    p = {"MobileOS": "ETC", "MobileApp": "INFINITH", "_type": "json",
         "numOfRows": str(rows), "pageNo": str(page), **extra}
    url = f"{base}/{svc}/{op}?serviceKey={key}&" + urllib.parse.urlencode(p)
    for i in range(retries):
        try:
            with urllib.request.urlopen(url, timeout=30, context=ssl_ctx()) as r:
                body = r.read().decode("utf-8", "replace")
            if body.lstrip().startswith("<"):
                raise RuntimeError(f"XML 응답(인증·한도 확인): {body[:160]}")
            j = json.loads(body)["response"]
            if j["header"].get("resultCode") != "0000":
                raise RuntimeError(f"{j['header'].get('resultCode')} {j['header'].get('resultMsg')}")
            b = j.get("body", {}); items = b.get("items")
            return b.get("totalCount", 0), ([] if not items or items == "" else items["item"])
        except Exception as e:
            if i == retries - 1: raise
            time.sleep(1.5 * (i + 1))


def call_all(env, op, extra, max_pages=6, rows=100):
    """전 페이지를 모은다. 관광지는 거리순 상위 100건이 거의 다 의료관광업소라 한 페이지로는 부족하다."""
    got, total = [], None
    for page in range(1, max_pages + 1):
        total, items = call(env, op, extra, rows=rows, page=page)
        got += items
        if not items or len(got) >= (total or 0): break
        time.sleep(0.2)
    return total, got


def bucket(it):
    """분류코드로 범주를 정한다. None 이면 버린다."""
    cat3 = (it.get("cat3") or "").strip()
    if cat3 == CAT_MEDICAL_TOURISM: return None
    if cat3 in CAT_WELLNESS: return "wellness"
    return "keep"


def clean(it, lat, lng):
    """TourAPI 원본에서 화면이 쓸 필드만 남긴다. mapx=경도, mapy=위도 (이름과 반대라 자주 틀린다)."""
    try:
        ilng, ilat = float(it.get("mapx") or 0), float(it.get("mapy") or 0)
    except ValueError:
        ilng = ilat = 0.0
    dist = it.get("dist")
    return {
        "id": it.get("contentid"),
        "title": (it.get("title") or "").strip(),
        "address": (it.get("addr1") or "").strip(),
        # TourAPI 는 이미지 URL 을 http 로 준다. https 사이트에서 혼합 콘텐츠로 차단되므로 올린다.
        # tong.visitkorea.or.kr 은 https 로도 같은 파일을 준다(실측 확인).
        "image": (lambda u: u.replace("http://", "https://", 1) if u else None)(
            it.get("firstimage") or it.get("firstimage2") or None),
        "tel": (it.get("tel") or "").strip() or None,
        "lat": ilat or None, "lng": ilng or None,
        "distanceM": round(float(dist)) if dist else (
            round(haversine_km(lat, lng, ilat, ilng) * 1000) if ilat and ilng else None),
        # TourAPI 가 주지 않는 값. Google Places 승인 후 채운다. 지금은 비워 둔다.
        "rating": None, "reviewCount": None, "openingHours": None,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--festival-km", type=float, default=60.0, help="이 거리 안의 축제만 담는다")
    ap.add_argument("--per-type", type=int, default=24, help="타입별 최대 건수")
    ap.add_argument("--out", default="src/data/medicalTourism.json")
    a = ap.parse_args()

    env = load_env(os.path.join(ROOT, ".env"))
    for k in ("TOUR_API_KEY", "TOUR_API_BASE", "TOUR_API_SERVICE", "TOUR_API_ORIGIN_LAT", "TOUR_API_ORIGIN_LNG"):
        if not env.get(k): sys.exit(f".env 에 {k} 가 비어 있습니다")
    lat, lng = float(env["TOUR_API_ORIGIN_LAT"]), float(env["TOUR_API_ORIGIN_LNG"])

    out = {"places": {}, "festivals": [], "meta": {}}
    filtered_log = {}

    wellness = []
    for ct, (name, radius) in TYPES.items():
        total, rows = call_all(env, "locationBasedList2",
                               {"mapX": str(lng), "mapY": str(lat), "radius": radius,
                                "contentTypeId": ct, "arrange": "E"})
        kept, dropped = [], []
        for it in rows:
            b = bucket(it)
            if b is None:
                dropped.append((it.get("title") or "").strip()); continue
            (wellness if b == "wellness" else kept).append(clean(it, lat, lng))
        kept.sort(key=lambda x: x["distanceM"] if x["distanceM"] is not None else 10 ** 9)
        out["places"][name] = kept[:a.per_type]
        filtered_log[name] = {"apiTotal": total, "fetched": len(rows), "radiusM": int(radius),
                              "droppedMedicalTourism": len(dropped), "kept": len(kept),
                              "shown": len(out["places"][name]), "droppedSample": dropped[:5]}
        print(f"  {name:<11} r={radius:>5}m · API {total:>4} · 받음 {len(rows):>3} · 의료관광업소 제외 {len(dropped):>3} · 담음 {len(out['places'][name])}")
    wellness.sort(key=lambda x: x["distanceM"] if x["distanceM"] is not None else 10 ** 9)
    out["places"]["wellness"] = wellness[:a.per_type]
    filtered_log["wellness"] = {"note": "관광지(76)에서 스파·찜질방 분류만 분리", "shown": len(out["places"]["wellness"])}
    print(f"  {'wellness':<11} 스파·찜질방 분리 · 담음 {len(out['places']['wellness'])}")

    # 영문 축제는 areacode 필드가 비어 있어 지역 필터를 쓸 수 없다. 좌표로 거리를 계산해 거른다.
    today = time.strftime("%Y%m%d")
    total, rows = call(env, "searchFestival2", {"eventStartDate": today, "arrange": "A"}, rows=300)
    near, nocoord = [], 0
    for it in rows:
        c = clean(it, lat, lng)
        if c["lat"] is None: nocoord += 1; continue
        km = haversine_km(lat, lng, c["lat"], c["lng"])
        if km > a.festival_km: continue
        c.update(startDate=it.get("eventstartdate"), endDate=it.get("eventenddate"),
                 distanceKm=round(km, 1))
        near.append(c)
    near.sort(key=lambda x: x["startDate"] or "")
    out["festivals"] = near[:a.per_type]
    print(f"  festival    API {total:>4}건 · 받음 {len(rows):>3} · 좌표없음 {nocoord} · {a.festival_km:.0f}km 이내 {len(near)}")

    out["meta"] = {
        "source": "한국관광공사 TourAPI " + env["TOUR_API_SERVICE"],
        "fetchedAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "origin": {"lat": lat, "lng": lng, "label": "뷰성형외과의원 (봉은사로 107)",
                   "coordSource": "Naver 지역검색 실측 2026-09-09"},
        "radiusByType": {n: int(r) for n, r in TYPES.values()}, "festivalRadiusKm": a.festival_km,
        "counts": filtered_log,
        # 화면이 한계를 그대로 표시할 수 있게 데이터에 적어 둔다. 없는 값을 채우지 않는다.
        "limits": [
            "영문 서비스는 강남 일대 숙박·음식점 커버리지가 얇다(반경 5km 숙박 4건·음식점 18건, 국문은 43건·338건).",
            "관광지 항목의 대부분(반경 5km 100건 중 92건)이 분류코드 A02020500 \"Medical Tourism Sites\" 였고 제외했다.",
            "평점·리뷰·영업시간은 TourAPI 가 제공하지 않는다. Google Places 승인 후 채운다.",
            "영문 축제는 areacode 필드가 비어 있어 좌표 거리로 걸렀다.",
            "수술 전후 식이 적합성은 병원이 입력한 가이드로만 표시한다. 이 파일에는 없다.",
        ],
    }
    p = os.path.join(ROOT, a.out)
    os.makedirs(os.path.dirname(p), exist_ok=True)
    io.open(p, "w", encoding="utf-8").write(json.dumps(out, ensure_ascii=False, indent=2))
    print(f"\n→ {a.out}")


if __name__ == "__main__":
    main()