# -*- coding: utf-8 -*-
"""
STEP102-12
services/naver_search_complex_resolver.py
네이버 검색 부동산 단지 카드에서 단지 주소/지번을 추출해
blog_complex_address_cache 에 저장한다.
핵심 개선:
1) 짧은 단지명 대응
예: "에덴" -> "구미시 옥계동 에덴아파트" 후보 검색
2) 지번 우선 추출
예: 도로명 "경상북도 구미시 흥안로 46" 보다
지번 "옥계동 543"을 우선 사용해
"경상북도 구미시 옥계동 543" 으로 저장
3) 원래 요청한 complex_name으로 캐시 저장
예: 요청 complex_name="에덴"
검색 성공 query="구미시 옥계동 에덴아파트"
저장 complex_name="에덴"
이렇게 해야 generate_blog_drafts.py가 기존 단지명으로 캐시를 다시 찾을 수 있다.
4) complex_no/hscp_no 보조 지원
resolve_and_cache_complex_address(conn, region, complex, hscp_no="24425")
처럼 호출 가능. 기존 3개 인자 호출도 그대로 호환된다.
사용 예:
cd /d D:\honghee\blog_api
python -c "from db import get_conn; from services.naver_search_complex_resolver import resolve_and_cache_complex_address; c=get_conn(); r=resolve_and_cache_complex_address(c,'경상북도 구미시 옥계동','에덴',force_refresh=True); print(r); c.close()"
python -c "from db import get_conn; from services.naver_search_complex_resolver import resolve_and_cache_complex_address; c=get_conn(); r=resolve_and_cache_complex_address(c,'경상북도 구미시 옥계동','에덴',hscp_no='24425',force_refresh=True); print(r); c.close()"
"""
from __future__ import annotations
import html
import re
import time
from typing import Any, Dict, List, Optional, Tuple
import requests
NAVER_SEARCH_URL = "https://search.naver.com/search.naver"
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
"Referer": "https://search.naver.com/",
}
def clean_text(value: Any) -> str:
value = "" if value is None else str(value)
value = html.unescape(value)
value = re.sub(r"<[^>]+>", " ", value)
value = value.replace("\ufeff", "")
value = re.sub(r"\s+", " ", value)
return value.strip()
def normalize_space(value: Any) -> str:
return re.sub(r"\s+", " ", str(value or "")).strip()
def strip_dong_suffix(name: str) -> str:
"""단지 주소 검색용으로 끝의 동/호 정보를 반복 제거한다."""
name = clean_text(name)
prev = None
while prev != name:
prev = name
name = re.sub(r"\s*\d{1,4}\s*동\s*$", "", name).strip()
name = re.sub(r"\s*\d{1,4}\s*호\s*$", "", name).strip()
name = re.sub(r"\s*[A-Za-z가-힣]\s*동\s*$", "", name).strip()
name = re.sub(r"\s+\d{1,4}\s*$", "", name).strip()
return name
def strip_property_prefix(name: str) -> str:
name = clean_text(name)
name = re.sub(r"^(아파트|오피스텔|빌라|주상복합|상가|토지)\s+", "", name).strip()
return name
def normalize_complex_name(name: str) -> str:
name = strip_property_prefix(strip_dong_suffix(name))
name = re.sub(r"\s+", "", name)
return name.strip()
def compact_name(name: str) -> str:
return re.sub(r"[\s\(\)\[\]\-_/·ㆍ,]+", "", clean_text(name))
def without_apt_suffix(name: str) -> str:
name = clean_text(name)
for suffix in ["아파트", "APT", "apt"]:
if name.endswith(suffix):
return name[: -len(suffix)].strip()
return name
def has_building_type_suffix(name: str) -> bool:
name = clean_text(name)
suffixes = [
"아파트", "오피스텔", "주상복합", "빌라", "타운", "하우스", "맨션",
"자이", "래미안", "힐스테이트", "푸르지오", "더샵", "롯데캐슬", "아이파크",
]
return any(name.endswith(s) for s in suffixes)
def split_region(region_name: str) -> Dict[str, str]:
parts = clean_text(sanitize_region_for_search(region_name)).split()
sido = parts[0] if len(parts) >= 1 else ""
sigungu = parts[1] if len(parts) >= 2 else ""
dong = parts[-1] if len(parts) >= 1 else ""
short_map = {
"경상북도": "경북", "경상남도": "경남", "전라북도": "전북", "전라남도": "전남",
"충청북도": "충북", "충청남도": "충남", "강원특별자치도": "강원", "강원도": "강원",
"제주특별자치도": "제주", "서울특별시": "서울", "부산광역시": "부산", "대구광역시": "대구",
"인천광역시": "인천", "광주광역시": "광주", "대전광역시": "대전", "울산광역시": "울산",
"세종특별자치시": "세종",
}
short_sido = short_map.get(sido, sido.replace("특별시", "").replace("광역시", ""))
sigungu_short = re.sub(r"(시|군|구)$", "", sigungu)
return {
"sido": sido,
"short_sido": short_sido,
"sigungu": sigungu,
"sigungu_short": sigungu_short,
"dong": dong,
"region": clean_text(region_name),
"prefix_without_dong": " ".join(parts[:-1]) if len(parts) >= 2 else clean_text(region_name),
}
def sanitize_region_for_search(region_name: str) -> str:
"""상가동/호수/번지를 제거하고 시·군·구·읍·면·동까지만 검색에 사용한다."""
value = clean_text(region_name)
if not value:
return ""
province_short = {
"서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
"경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주",
}
result = []
for raw_token in value.split():
token = raw_token.strip(",()[]")
if not token:
continue
if token in province_short or re.search(
r"(?:특별자치도|특별자치시|특별시|광역시|도|시|군|구)$", token
):
result.append(token)
continue
locality = re.match(r"^([가-힣A-Za-z0-9]+?(?:읍|면|동))(?:\d.*)?$", token)
if locality:
name = locality.group(1)
if name not in {"상가", "상가동"}:
result.append(name)
break
if re.search(r"\d|(?:로|길)$|(?:상가|상가동|호)$", token):
break
return clean_text(" ".join(result))
def unique_keep_order(items: List[str]) -> List[str]:
seen = set()
result = []
for item in items:
item = normalize_space(item)
if not item:
continue
key = item.lower()
if key in seen:
continue
seen.add(key)
result.append(item)
return result
def build_search_queries(region_name: str, complex_name: str, hscp_no: str = "", article_name: str = "") -> List[str]:
region_name = sanitize_region_for_search(region_name)
region = split_region(region_name)
base = normalize_complex_name(complex_name)
article_base = normalize_complex_name(article_name)
names = [base]
if article_base and article_base not in names:
names.append(article_base)
expanded_names = []
for name in names:
if not name:
continue
expanded_names.append(name)
if not has_building_type_suffix(name):
expanded_names.append(f"{name}아파트")
no_apt = without_apt_suffix(name)
if no_apt and no_apt != name:
expanded_names.append(no_apt)
expanded_names = unique_keep_order(expanded_names)
queries = []
for name in expanded_names:
queries += [
f"{region['region']} {name}",
f"{region['sigungu']} {region['dong']} {name}",
f"{region['sigungu_short']} {region['dong']} {name}",
f"{region['short_sido']} {region['sigungu_short']} {region['dong']} {name}",
name,
]
if hscp_no:
queries += [
f"fin.land.naver.com/complexes/{hscp_no}",
f"네이버 부동산 {hscp_no}",
]
return unique_keep_order(queries)
def fetch_naver_search_html(query: str, timeout: int = 12) -> str:
params = {
"where": "nexearch",
"sm": "top_hty",
"fbm": "0",
"ie": "utf8",
"query": query,
}
res = requests.get(NAVER_SEARCH_URL, params=params, headers=HEADERS, timeout=timeout)
res.raise_for_status()
return res.text
def extract_complex_no(page_html: str) -> str:
for pattern in [r"fin\.land\.naver\.com/complexes/(\d+)", r"new\.land\.naver\.com/complexes/(\d+)", r"/complexes/(\d+)"]:
m = re.search(pattern, page_html)
if m:
return m.group(1)
return ""
def extract_card_name(page_html: str) -> str:
patterns = [
r'\s*]*>(.*?)\s*',
r']+class="[^"]*\bname\b[^"]*"[^>]*>\s*]*>(.*?)',
]
for pattern in patterns:
m = re.search(pattern, page_html, re.S)
if m:
return clean_text(m.group(1))
return ""
def extract_school_name(page_html: str) -> str:
for pattern in [r'anchorTo=complex-school-info[^>]*>(.*?)', r"배정학교\s*-\s*]*>(.*?)"]:
m = re.search(pattern, page_html, re.S)
if m:
return clean_text(m.group(1))
return ""
def extract_map_coords(page_html: str) -> Tuple[str, str]:
for pattern in [r"center=([0-9.\-]+),([0-9.\-]+)", r"pos:([0-9.\-]+)\+([0-9.\-]+)", r"/directions/-/([0-9.\-]+)%2C([0-9.\-]+)%2C"]:
m = re.search(pattern, page_html)
if m:
return m.group(1), m.group(2)
return "", ""
def extract_jibun_address(page_html: str) -> str:
patterns = [
r']*class="[^"]*\blabel_address\b[^"]*"[^>]*>\s*지번\s*\s*]*class="[^"]*\baddr\b[^"]*"[^>]*>(.*?)',
r"지번\s*\s*]*class=[\"'][^\"']*addr[^\"']*[\"'][^>]*>(.*?)",
]
for pattern in patterns:
m = re.search(pattern, page_html, re.S)
if m:
return clean_text(m.group(1))
return ""
def extract_first_address(page_html: str) -> str:
m = re.search(r']*class="[^"]*\baddr\b[^"]*"[^>]*>(.*?)', page_html, re.S)
if m:
return clean_text(m.group(1))
return ""
def combine_region_and_jibun(region_name: str, jibun: str) -> str:
region = split_region(region_name)
jibun = clean_text(jibun)
if not jibun:
return ""
if region["sido"] and region["sido"] in jibun:
return jibun
if region["short_sido"] and region["short_sido"] in jibun and region["sigungu_short"] in jibun:
return jibun
if region["dong"] and jibun.startswith(region["dong"]):
return normalize_space(f"{region['prefix_without_dong']} {jibun}")
if re.match(r"^(산\s*)?\d", jibun):
return normalize_space(f"{region['region']} {jibun}")
if "동" not in jibun and "읍" not in jibun and "면" not in jibun:
return normalize_space(f"{region['region']} {jibun}")
return normalize_space(f"{region['prefix_without_dong']} {jibun}")
def address_matches_region(region_name: str, address: str) -> bool:
region = split_region(region_name)
address = clean_text(address)
if not address:
return False
if region["dong"] and region["dong"] in address:
return True
if region["sigungu"] and region["sigungu"] in address:
return True
if region["sigungu_short"] and region["sigungu_short"] in address:
return True
return False
def parse_complex_card(page_html: str, region_name: str, requested_complex_name: str, expected_complex_no: str = "") -> Optional[Dict[str, Any]]:
if "sp_nland" not in page_html and "lan_clc" not in page_html and "fin.land.naver.com/complexes" not in page_html:
return None
complex_no = extract_complex_no(page_html)
if expected_complex_no and complex_no and str(complex_no) != str(expected_complex_no):
print(f"[NAVER_SEARCH_COMPLEX_NO_MISMATCH] expected={expected_complex_no} found={complex_no}", flush=True)
return None
card_name = extract_card_name(page_html)
if card_name:
req_without_apt = compact_name(without_apt_suffix(requested_complex_name))
card_without_apt = compact_name(without_apt_suffix(card_name))
if req_without_apt and card_without_apt:
if req_without_apt not in card_without_apt and card_without_apt not in req_without_apt:
if not expected_complex_no:
print(f"[NAVER_SEARCH_NAME_MISMATCH] requested={requested_complex_name} card={card_name}", flush=True)
return None
jibun = extract_jibun_address(page_html)
first_addr = extract_first_address(page_html)
resolved_address = combine_region_and_jibun(region_name, jibun) if jibun else clean_text(first_addr)
if not resolved_address:
return None
if re.search(
r"상가\s*(?:제?\s*\d+\s*동|동)?\s*제?\s*\d+\s*호",
resolved_address,
):
print(
f"[NAVER_SEARCH_ADDR_REALTOR_SUITE_REJECTED] address={resolved_address}",
flush=True,
)
return None
if not address_matches_region(region_name, resolved_address):
print(f"[NAVER_SEARCH_ADDR_REGION_MISMATCH] region={region_name} address={resolved_address}", flush=True)
return None
map_lng, map_lat = extract_map_coords(page_html)
school_name = extract_school_name(page_html)
return {
"region_name": clean_text(region_name),
# 요청 단지명 그대로 저장한다. generate_blog_drafts.py의 기존 조회와 맞추기 위함.
"complex_name": clean_text(requested_complex_name),
"display_complex_name": card_name,
"complex_no": complex_no or str(expected_complex_no or ""),
"resolved_address": resolved_address,
"school_name": school_name,
"map_lng": map_lng,
"map_lat": map_lat,
"source": "naver_search_complex_card",
"raw_html": page_html,
}
def fetch_cached_complex_address(conn, region_name: str, complex_name: str, hscp_no: str = "") -> Optional[Dict[str, Any]]:
region_name = clean_text(region_name)
complex_name = clean_text(complex_name)
base = normalize_complex_name(complex_name)
base_no_apt = without_apt_suffix(base)
names = unique_keep_order([
complex_name,
base,
base_no_apt,
f"{base_no_apt}아파트" if base_no_apt and not has_building_type_suffix(base_no_apt) else "",
])
with conn.cursor() as cur:
if hscp_no:
cur.execute(
"""
SELECT *
FROM blog_complex_address_cache
WHERE complex_no = %s
OR (region_name = %s AND complex_name IN ({}))
ORDER BY CASE WHEN complex_no = %s THEN 0 ELSE 1 END, last_checked_at DESC, id DESC
LIMIT 1
""".format(",".join(["%s"] * len(names))),
[str(hscp_no), region_name, *names, str(hscp_no)],
)
else:
cur.execute(
"""
SELECT *
FROM blog_complex_address_cache
WHERE region_name = %s
AND complex_name IN ({})
ORDER BY last_checked_at DESC, id DESC
LIMIT 1
""".format(",".join(["%s"] * len(names))),
[region_name, *names],
)
return cur.fetchone()
def save_complex_address_cache(conn, data: Dict[str, Any]) -> Dict[str, Any]:
region_name = clean_text(data.get("region_name"))
complex_name = clean_text(data.get("complex_name"))
complex_no = clean_text(data.get("complex_no"))
resolved_address = clean_text(data.get("resolved_address"))
if not region_name or not complex_name or not resolved_address:
raise ValueError("region_name, complex_name, resolved_address are required")
with conn.cursor() as cur:
cur.execute(
"""
SELECT id
FROM blog_complex_address_cache
WHERE region_name = %s
AND complex_name = %s
LIMIT 1
""",
(region_name, complex_name),
)
row = cur.fetchone()
if row:
cache_id = row.get("id") if isinstance(row, dict) else row[0]
cur.execute(
"""
UPDATE blog_complex_address_cache
SET complex_no = %s,
resolved_address = %s,
school_name = %s,
map_lng = %s,
map_lat = %s,
source = %s,
raw_html = %s,
last_checked_at = NOW(),
updated_at = NOW()
WHERE id = %s
""",
(
complex_no or None,
resolved_address,
clean_text(data.get("school_name")) or None,
clean_text(data.get("map_lng")) or None,
clean_text(data.get("map_lat")) or None,
clean_text(data.get("source")) or "naver_search_complex_card",
data.get("raw_html") or "",
cache_id,
),
)
else:
cur.execute(
"""
INSERT INTO blog_complex_address_cache
(region_name, complex_name, complex_no, resolved_address, school_name, map_lng, map_lat, source, raw_html, last_checked_at, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW())
""",
(
region_name,
complex_name,
complex_no or None,
resolved_address,
clean_text(data.get("school_name")) or None,
clean_text(data.get("map_lng")) or None,
clean_text(data.get("map_lat")) or None,
clean_text(data.get("source")) or "naver_search_complex_card",
data.get("raw_html") or "",
),
)
conn.commit()
print(f"[COMPLEX_ADDRESS_CACHE_SAVED] {region_name} / {complex_name} -> {resolved_address}", flush=True)
return data
def resolve_complex_address_from_naver_search(region_name: str, complex_name: str, hscp_no: str = "", article_name: str = "", sleep_sec: float = 0.4) -> Optional[Dict[str, Any]]:
queries = build_search_queries(region_name=region_name, complex_name=complex_name, hscp_no=hscp_no, article_name=article_name)
print(f"[COMPLEX_ADDRESS_SEARCH_QUERIES] region={region_name} complex={complex_name} queries={len(queries)}", flush=True)
last_error = None
for idx, query in enumerate(queries, start=1):
try:
print(f"[NAVER_SEARCH_QUERY] {idx}/{len(queries)} {query}", flush=True)
page_html = fetch_naver_search_html(query)
parsed = parse_complex_card(page_html, region_name, complex_name, expected_complex_no=str(hscp_no or ""))
if parsed:
parsed["source_query"] = query
print(f"[NAVER_SEARCH_COMPLEX_CARD_OK] query={query} address={parsed.get('resolved_address')} complex_no={parsed.get('complex_no')}", flush=True)
return parsed
print(f"[NAVER_SEARCH_NO_CARD] query={query}", flush=True)
except Exception as e:
last_error = e
print(f"[NAVER_SEARCH_ERROR] query={query} err={e}", flush=True)
if sleep_sec:
time.sleep(float(sleep_sec))
if last_error:
print(f"[COMPLEX_ADDRESS_RESOLVE_LAST_ERROR] {last_error}", flush=True)
return None
def resolve_and_cache_complex_address(conn, region_name: str, complex_name: str, force_refresh: bool = False, hscp_no: str = "", article_name: str = "") -> Optional[Dict[str, Any]]:
"""
기존 호출 호환:
resolve_and_cache_complex_address(conn, region_name, complex_name)
resolve_and_cache_complex_address(conn, region_name, complex_name, force_refresh=True)
신규 호출:
resolve_and_cache_complex_address(conn, region_name, complex_name, hscp_no="24425")
resolve_and_cache_complex_address(conn, region_name, complex_name, article_name="에덴 104동")
"""
region_name = clean_text(region_name)
complex_name = clean_text(complex_name)
if not region_name or not complex_name:
return None
if not force_refresh:
cached = fetch_cached_complex_address(conn, region_name=region_name, complex_name=complex_name, hscp_no=str(hscp_no or ""))
if cached:
addr = cached.get("resolved_address") if isinstance(cached, dict) else ""
if clean_text(addr):
print(f"[COMPLEX_ADDRESS_CACHE_HIT] {region_name} / {complex_name}", flush=True)
return cached
resolved = resolve_complex_address_from_naver_search(
region_name=region_name,
complex_name=complex_name,
hscp_no=str(hscp_no or ""),
article_name=article_name,
)
if not resolved:
print(f"[COMPLEX_ADDRESS_RESOLVE_EMPTY] {region_name} / {complex_name}", flush=True)
return None
return save_complex_address_cache(conn, resolved)
def main():
try:
from db import get_conn
except Exception as e:
print(f"[IMPORT_DB_ERROR] {e}")
return
conn = get_conn()
try:
tests = [
("부산시 연제구 연산동", "시청역해모로센티아", ""),
("경상북도 구미시 옥계동", "에덴", "24425"),
]
for region_name, complex_name, hscp_no in tests:
print("=" * 80)
result = resolve_and_cache_complex_address(conn, region_name, complex_name, force_refresh=True, hscp_no=hscp_no)
print(result)
finally:
conn.close()
if __name__ == "__main__":
main()