o2o-infinith-demo/scripts/fetch_medical_tourism.py

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

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

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

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

출력은 서포터즈 사이트(Astro)의 데이터다. 화면은 supporters/src/pages/en/stay.astro 다.
INFINITH 제품 화면(/discovery)이 아니라 병원 서포터즈 사이트에 실린다. 외국인 환자가 보는 곳이다.

  python3 scripts/fetch_medical_tourism.py
  python3 scripts/fetch_medical_tourism.py --out supporters/src/data/medicalTourism.json
"""
import argparse, io, json, math, os, re, 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. 국문과 코드가 다르다.
# 반경은 목적에 따라 다르다. 회복기에 매일 가는 곳(숙박·식사·스파)은 가깝게, 관광은 넓게 본다.
SEASON = {3: "spring", 4: "spring", 5: "spring", 6: "summer", 7: "summer", 8: "summer",
          9: "autumn", 10: "autumn", 11: "autumn", 12: "winter", 1: "winter", 2: "winter"}

# ContentTypeId 는 다국어와 국문이 다르다. 같은 코드로 부르면 엉뚱한 것이 온다.
TYPES_EN = {
    "76": ("attraction", "10000"),
    "78": ("culture", "10000"),
    "79": ("shopping", "10000"),
    "80": ("stay", "5000"),
    "82": ("restaurant", "5000"),
}
TYPES_KO = {
    "12": ("attraction", "10000"),
    "14": ("culture", "10000"),
    "38": ("shopping", "10000"),
    "32": ("stay", "5000"),
    "39": ("restaurant", "5000"),
}
TYPES = TYPES_EN  # main() 에서 --lang 에 맞게 바꾼다


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 walk_min(m):
    """도보 분. 80m/분(4.8km/h) 기준. 참고 사이트의 표기를 역산해 맞춘 값이다
    (135m→2분, 248m→3분, 297m→4분, 566m→7분)."""
    return max(1, round(m / 80))


def drive_min(km):
    """차 이동 분. 직선거리 km x 1.2. 역시 역산값이다(25km→30분, 33km→40분)."""
    return max(1, round(km * 1.2))


def travel(m):
    """이동 수단과 소요 시간.
    2km 까지는 걸을 수 있는 거리로 본다(약 25분). 그 위는 차·지하철이다.
    짧은 거리의 차 시간은 주차와 신호 때문에 의미가 없으므로 최소 5분으로 둔다."""
    if m is None: return None
    if m <= 2000: return {"mode": "walk", "minutes": walk_min(m), "label": f"{walk_min(m)} min walk"}
    mins = max(5, drive_min(m / 1000))
    return {"mode": "drive", "minutes": mins, "label": f"{mins} min by car"}


def fetch_overview(env, content_id, cache):
    """detailCommon2 의 overview(설명문)와 homepage. locationBasedList2 는 주지 않는다.
    설명문이 없으면 카드가 이름과 주소뿐이라 사람이 고를 수 없다."""
    if content_id in cache: return cache[content_id]
    try:
        _, rows = call(env, "detailCommon2", {"contentId": content_id}, rows=1)
    except Exception:
        cache[content_id] = ("", ""); return cache[content_id]
    if not rows: cache[content_id] = ("", ""); return cache[content_id]
    r = rows[0]
    ov = re.sub(r"<[^>]+>", " ", r.get("overview") or "")
    ov = re.sub(r"\s+", " ", ov).strip()
    hp = re.sub(r"<[^>]+>", " ", r.get("homepage") or "").strip()
    hp = (re.search(r"https?://\S+", hp) or [None])[0] if hp else ""
    cache[content_id] = (ov, hp or "")
    return cache[content_id]


def fetch_image(env, content_id, cache):
    """locationBasedList2 의 firstimage 가 비어 있을 때 detailImage2 로 한 번 더 찾는다.
    이미지 전용 오퍼레이션이라 목록 API 에 없는 사진이 여기 있다(도산공원 7장, 몽마르뜨공원 5장).
    실측: 사진 없던 38건 중 15건을 여기서 살렸다. 나머지 23건은 TourAPI 에 사진이 없다."""
    key = ("img", content_id)
    if key in cache: return cache[key]
    try:
        _, rows = call(env, "detailImage2", {"contentId": content_id, "imageYN": "Y"}, rows=5)
    except Exception:
        cache[key] = ""; return ""
    url = ""
    for r in rows or []:
        u = (r.get("originimgurl") or r.get("smallimageurl") or "").strip()
        if u: url = u.replace("http://", "https://", 1); break
    cache[key] = url
    return url


def call_kor(env, op, extra, rows=10):
    """국문 서비스(KorService2). 영문 항목에 사진이 없어도 국문 항목에는 있는 경우가 있다.
    같은 한국관광공사 사진이라 출처가 달라지지 않는다."""
    p = {"MobileOS": "ETC", "MobileApp": "INFINITH", "_type": "json",
         "numOfRows": str(rows), "pageNo": "1", **extra}
    url = f"{env['TOUR_API_BASE']}/KorService2/{op}?serviceKey={env['TOUR_API_KEY']}&" + urllib.parse.urlencode(p)
    try:
        with urllib.request.urlopen(url, timeout=25, context=ssl_ctx()) as r:
            body = r.read().decode("utf-8", "replace")
        if body.lstrip().startswith("<"): return []
        j = json.loads(body)["response"]
        if j["header"].get("resultCode") != "0000": return []
        it = j["body"].get("items")
        return [] if not it or it == "" else it["item"]
    except Exception:
        return []


def korean_name(title):
    m = re.search(r"\(([^)]+)\)\s*$", title or "")
    return (m.group(1) if m else (title or "")).strip()


def fetch_image_kor(env, p):
    """국문 서비스에서 같은 이름의 항목을 찾아 사진을 가져온다. 실측 23건 중 10건이 여기서 나왔다."""
    ko = korean_name(p.get("title"))
    if not re.search(r"[가-힣]", ko): return ""
    rows = []
    if p.get("lat"):
        rows = call_kor(env, "locationBasedList2",
                        {"mapX": str(p["lng"]), "mapY": str(p["lat"]), "radius": "300", "keyword": ko})
    if not rows: rows = call_kor(env, "searchKeyword2", {"keyword": ko})
    key = ko.replace(" ", "")
    for r in rows:
        if key in (r.get("title") or "").replace(" ", ""):
            u = (r.get("firstimage") or r.get("firstimage2") or "").strip()
            if u: return u.replace("http://", "https://", 1)
    return ""


def fetch_image_og(url):
    """업체 공식 홈페이지의 대표 이미지(og:image). 업체가 스스로 올린 사진이라 목록에 쓰기에 무리가 없다.
    검색 결과 이미지를 긁어 쓰면 식당·블로그·언론사의 저작물이라 환자용 페이지에 올릴 수 없다."""
    if not url or not url.startswith("http"): return ""
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=12, context=ssl_ctx()) as r:
            head = r.read(120000).decode("utf-8", "replace")
            base = r.geturl()
    except Exception:
        return ""
    m = re.search(r'<meta[^>]+(?:property|name)=["\']og:image["\'][^>]*content=["\']([^"\']+)', head, re.I) \
        or re.search(r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+(?:property|name)=["\']og:image', head, re.I)
    if not m: return ""
    u = m.group(1).strip()
    if u.startswith("//"): u = "https:" + u
    elif u.startswith("/"):
        from urllib.parse import urlparse
        pr = urlparse(base); u = f"{pr.scheme}://{pr.netloc}{u}"
    return u if u.startswith("https://") else ""


def usable_image(url):
    """실제로 이미지가 오는지 열어 본다. og:image 가 이미지가 아니거나(빈 content-type)
    인스타 CDN 처럼 핫링크를 403 으로 막는 경우가 있어 화면에서 빈 칸이 된다."""
    if not url: return False
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=10, context=ssl_ctx()) as r:
            return r.status == 200 and (r.headers.get("content-type") or "").startswith("image")
    except Exception:
        return False


def fetch_image_naver(env, p):
    """마지막 폴백. Naver 이미지 검색 API.
    주의: 결과는 제3자 저작물이다. 내부 시연 목적으로만 쓰고, 공개 배포 전에는
    imageSource 가 'naver' 인 항목을 걷어내거나 라이선스가 있는 사진으로 교체한다.
    구글 HTML 크롤링 대신 이걸 쓰는 이유는 공식 API 라 차단되지 않고 결과가 일정해서다."""
    cid, sec = env.get("NAVER_CLIENT_ID"), env.get("NAVER_CLIENT_SECRET")
    if not (cid and sec): return ""
    ko = korean_name(p.get("title"))
    if not ko: return ""
    # 지역을 붙이면 정확도가 오르지만 결과가 아예 없는 경우가 있다. 넓혀 가며 시도한다.
    plain = re.sub(r"\s*\(.*\)\s*", "", p.get("title") or "").strip()
    for q in (f"{ko} 서울", ko, plain):
        if not q: continue
        qs = urllib.parse.urlencode({"query": q, "display": "3", "sort": "sim"})
        req = urllib.request.Request("https://openapi.naver.com/v1/search/image.json?" + qs,
                                     headers={"X-Naver-Client-Id": cid, "X-Naver-Client-Secret": sec})
        try:
            with urllib.request.urlopen(req, timeout=15, context=ssl_ctx()) as r:
                items = json.loads(r.read().decode()).get("items", [])
        except Exception:
            continue
        for it in items:
            u = (it.get("link") or "").strip()
            if u.startswith("https://"): return u
    return ""


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),
        "imageSource": "tourapi" if (it.get("firstimage") or it.get("firstimage2")) else "",
        "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),
        "travel": travel(round(float(dist)) if dist else (
            round(haversine_km(lat, lng, ilat, ilng) * 1000) if ilat and ilng else None)),
        # detailCommon2 로 따로 채운다(locationBasedList2 는 설명문을 주지 않는다).
        "overview": "", "homepage": "",
        # 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("--lang", choices=["en", "ko"], default="en",
                    help="en 은 EngService2(외국인 환자용 /en/stay), ko 는 KorService2(/stay)")
    ap.add_argument("--out", default="")
    a = ap.parse_args()

    global TYPES
    env = load_env(os.path.join(ROOT, ".env"))
    if a.lang == "ko":
        TYPES = TYPES_KO
        env = dict(env, TOUR_API_SERVICE="KorService2")
    if not a.out:
        a.out = ("supporters/src/data/medicalTourism.json" if a.lang == "en"
                 else "supporters/src/data/medicalTourismKo.json")
    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
        sd = it.get("eventstartdate") or ""
        c.update(startDate=sd, endDate=it.get("eventenddate"),
                 distanceKm=round(km, 1), month=(int(sd[4:6]) if len(sd) == 8 else None),
                 season=SEASON.get(int(sd[4:6]), "") if len(sd) == 8 else "")
        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)}")

    # 설명문은 항목마다 detailCommon2 를 한 번씩 부른다. 카드에 이름과 주소만 있으면 고를 수 없다.
    cache: dict = {}
    targets = [p for v in out["places"].values() for p in v] + out["festivals"]
    print(f"\n  설명문 수집 {len(targets)}건", end="", flush=True)
    for i, p in enumerate(targets):
        if not p.get("id"): continue
        ov, hp = fetch_overview(env, p["id"], cache)
        p["overview"] = ov[:600]
        p["homepage"] = hp
        if not p.get("image"):
            # 1) 영문 detailImage2 → 2) 국문 서비스 → 3) 업체 공식 홈페이지 og:image
            # 1) 영문 detailImage2 → 2) 국문 서비스 → 3) 업체 홈페이지 og:image → 4) Naver 이미지 검색
            src = ""
            img = fetch_image(env, p["id"], cache)
            if img: src = "tourapi"
            if not img and a.lang == "en":
                img = fetch_image_kor(env, p); src = "tourapi-kor" if img else src
            if not img:
                cand = fetch_image_og(p.get("homepage") or "")
                if cand and usable_image(cand): img, src = cand, "homepage"
            if not img:
                img = fetch_image_naver(env, p); src = "naver" if img else src
            p["image"] = img or None
            p["imageSource"] = src
        if i % 20 == 0: print(".", end="", flush=True)
        time.sleep(0.12)
    got = sum(1 for p in targets if p["overview"])
    pic = sum(1 for p in targets if p.get("image"))
    print(f" 완료 · 설명문 {got}/{len(targets)} · 사진 {pic}/{len(targets)}")

    # 도보 시간 구간별 건수. 화면의 필터 칩이 이 수를 쓴다.
    allp = [p for v in out["places"].values() for p in v]
    # 누적 기준이다. "15분 이내"는 5분 이내도 포함한다. 사람이 고르는 방식에 맞춘다.
    def walk_upto(mins):
        return sum(1 for p in allp if p.get("travel") and p["travel"]["mode"] == "walk"
                   and p["travel"]["minutes"] <= mins)
    out["filters"] = {
        "total": len(allp),
        "walk5": walk_upto(5),
        "walk15": walk_upto(15),
        "walkAll": sum(1 for p in allp if p.get("travel") and p["travel"]["mode"] == "walk"),
        "drive": sum(1 for p in allp if not p.get("travel") or p["travel"]["mode"] != "walk"),
    }

    out["meta"] = {
        "lang": a.lang,
        "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()