# -*- coding: utf-8 -*- r""" tools/generate_realtor_naver_map.py 중개사 주소/좌표를 기준으로 네이버 지도 화면을 Playwright headless로 캡처하고, Cafe24 CDN에 업로드한 뒤 blog_realtors.map_* 컬럼을 업데이트한다. 카카오 JS SDK 대신 네이버 지도 웹 화면을 캡처하는 방식이다. - 카카오 Local REST API: 주소 → 좌표 변환용 - 네이버 지도 웹: 실제 지도 렌더링/캡처용 - Cafe24 CDN: 블로그 본문 삽입용 이미지 저장 사용 예: cd D:\honghee\blog_api set KAKAO_REST_API_KEY=카카오_REST_API_키 python tools\generate_realtor_naver_map.py --realtor-id 4 python tools\generate_realtor_naver_map.py --all """ import os import sys import argparse import math import re from pathlib import Path from datetime import datetime from urllib.parse import quote import requests from playwright.sync_api import sync_playwright # STEP347: Windows 콘솔/API 실행 환경에서 한글 로그 깨짐 방지 try: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: pass BASE_DIR = Path(__file__).resolve().parents[1] sys.path.append(str(BASE_DIR)) from db import get_conn from services.cafe24_cdn_service import upload_header_image_to_cafe24 OUTPUT_DIR = BASE_DIR / "storage" / "realtor_maps" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) MAP_WIDTH = 900 MAP_HEIGHT = 540 BROWSER_WIDTH = 1500 BROWSER_HEIGHT = 620 # STEP102-14D: # 네이버지도 c=lng,lat 중심점은 브라우저 viewport 중앙에 위치한다. # 따라서 900px 캡처 영역도 viewport 중앙(1500/2=750)에 맞춰야 한다. # 기존 CLIP_X=520은 캡처 중앙이 970px이라 지도 중심과 핀이 어긋났다. CLIP_X = 300 CLIP_Y = 20 # STEP348: # 네이버지도는 왼쪽 검색/장소 패널이 레이아웃 공간을 차지한다. # URL의 c=lng,lat 중심 좌표는 브라우저 전체 중앙이 아니라 # 왼쪽 패널을 제외한 실제 지도 영역 중앙에 렌더링된다. NAVER_LEFT_PANEL_WIDTH = 390 # STEP350: # 고정 패널 폭/보정값은 주소와 화면 상태에 따라 달라질 수 있으므로 # 실제 지도 렌더링 영역을 브라우저에서 측정한다. # 아래 값은 탐지 실패 시에만 사용하는 fallback이다. NAVER_MAP_CENTER_X_FALLBACK = int(BROWSER_WIDTH / 2) NAVER_MAP_CENTER_Y_FALLBACK = int(BROWSER_HEIGHT / 2) def clean_text(value): return str(value or "").strip() def table_columns(conn, table_name): try: with conn.cursor() as cur: cur.execute(f"SHOW COLUMNS FROM {table_name}") rows = cur.fetchall() return set(row["Field"] for row in rows) except Exception: return set() def fetch_realtor(conn, realtor_id): with conn.cursor() as cur: cur.execute(""" SELECT * FROM blog_realtors WHERE id = %s LIMIT 1 """, (int(realtor_id),)) return cur.fetchone() or {} def fetch_realtors_for_batch(conn): with conn.cursor() as cur: cur.execute(""" SELECT * FROM blog_realtors WHERE COALESCE(office_address, address, '') <> '' AND COALESCE(status, 'active') = 'active' ORDER BY id ASC """) return cur.fetchall() or [] def kakao_geocode(address, rest_api_key): address = clean_text(address) if not address: raise Exception("주소가 비어 있습니다.") if not rest_api_key: raise Exception("KAKAO_REST_API_KEY 값이 없습니다.") res = requests.get( "https://dapi.kakao.com/v2/local/search/address.json", headers={"Authorization": f"KakaoAK {rest_api_key}"}, params={"query": address}, timeout=20, ) if res.status_code >= 400: raise Exception(f"Kakao geocode HTTP {res.status_code}: {res.text[:500]}") data = res.json() docs = data.get("documents") or [] if not docs: raise Exception(f"주소 좌표 변환 결과가 없습니다: {address}") doc = docs[0] return { "lat": float(doc.get("y")), "lng": float(doc.get("x")), "raw": doc, } def normalize_digits(value): return "".join(ch for ch in str(value or "") if ch.isdigit()) def normalize_korean_text(value): value = clean_text(value).lower() value = re.sub(r"[^0-9a-z가-힣]+", "", value) return value def address_tokens(value): """ 주소 비교용 토큰. 시/군/구/동/읍/면/리/로/길/단지/아파트/마을/빌딩 등 위치 식별력이 있는 토큰만 남긴다. """ raw_tokens = re.split(r"[\s,()/]+", clean_text(value)) result = [] for token in raw_tokens: token = token.strip() if len(token) < 2: continue if ( token.endswith(("시", "군", "구", "동", "읍", "면", "리", "로", "길")) or "단지" in token or "아파트" in token or "마을" in token or "빌딩" in token or "타워" in token or "상가" in token ): normalized = normalize_korean_text(token) if normalized and normalized not in result: result.append(normalized) return result def extract_region_hint(address): """ 주소에서 검색 범위를 좁힐 행정구역 힌트를 만든다. 예: 세종특별자치시 새롬동 """ tokens = clean_text(address).replace(",", " ").split() selected = [] for token in tokens: if token.endswith(("특별자치시", "광역시", "특별시", "도", "시", "군", "구", "읍", "면", "동", "리")): selected.append(token) if len(selected) >= 3: break return " ".join(selected) def extract_complex_names(address): """ 주소/상세주소에서 단지명·아파트명·마을명·건물명을 추출한다. """ tokens = clean_text(address).replace(",", " ").split() names = [] for index, token in enumerate(tokens): token = token.strip() if not token: continue is_complex = any( keyword in token for keyword in ("단지", "아파트", "마을", "빌딩", "타워", "프라자", "센터") ) if not is_complex: continue candidates = [token] # "새뜸마을 3단지"처럼 분리된 경우 앞 토큰과 합친다. if index > 0: prev = tokens[index - 1].strip() if prev and len(prev) >= 2: candidates.append(f"{prev} {token}") for candidate in candidates: candidate = clean_text(candidate) if candidate and candidate not in names: names.append(candidate) return names def extract_korean_base_address(address): """ 내부 상세정보가 섞인 주소에서 지번 또는 도로명 건물번호까지만 추출한다. """ address = re.sub(r"\s+", " ", clean_text(address)).strip() if not address: return "" jibun = re.search(r"^(.+?(?:읍|면|동|리))\s+(\d+(?:-\d+)?)\b", address) if jibun: return f"{jibun.group(1)} {jibun.group(2)}".strip() road = re.search(r"^(.+?(?:대로|로|길))\s+(\d+(?:-\d+)?)\b", address) if road: return f"{road.group(1)} {road.group(2)}".strip() return address def build_address_candidates(base_address, detail_address=""): """ STEP347 주소 후보 확장. 우선순위: 1) 지번/도로명 건물번호까지 정제된 주소 2) 기본주소 3) 전체주소 4) 행정동 + 번지 5) 행정동만 6) 단지명/건물명 + 행정구역 7) 단지명/건물명 단독 """ base_address = clean_text(base_address) detail_address = clean_text(detail_address) full_address = clean_text(f"{base_address} {detail_address}") candidates = [] def add(value, source): value = re.sub(r"\s+", " ", clean_text(value)).strip(" ,") if not value: return if any(item["query"] == value for item in candidates): return candidates.append({ "query": value, "source": source, }) normalized_base = extract_korean_base_address(base_address) normalized_full = extract_korean_base_address(full_address) add(normalized_base, "normalized_base") add(normalized_full, "normalized_full") add(base_address, "base_address") add(full_address, "full_address") # 행정동/읍/면/리 + 번지 추출 for source_text in (base_address, full_address): match = re.search( r"((?:[가-힣]+(?:동|읍|면|리)))\s+(\d+(?:-\d+)?)\b", source_text, ) if match: add(f"{match.group(1)} {match.group(2)}", "local_jibun") add(match.group(1), "local_region") region_hint = extract_region_hint(full_address) for complex_name in extract_complex_names(full_address): add(f"{region_hint} {complex_name}", "region_complex") add(complex_name, "complex_name") # '새뜸마을3단지' -> '새뜸마을' simplified = re.sub(r"\d+\s*단지", "", complex_name).strip() if simplified and simplified != complex_name: add(f"{region_hint} {simplified}", "region_complex_simple") add(simplified, "complex_simple") return candidates def haversine_meters(lat1, lng1, lat2, lng2): radius = 6371000.0 p1 = math.radians(float(lat1)) p2 = math.radians(float(lat2)) dp = math.radians(float(lat2) - float(lat1)) dl = math.radians(float(lng2) - float(lng1)) a = ( math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 ) return radius * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def kakao_address_geocode_candidates(base_address, detail_address, rest_api_key): """ 여러 주소 후보를 순차 조회하고 첫 성공 결과를 반환한다. """ errors = [] candidates = build_address_candidates(base_address, detail_address) print("=" * 80, flush=True) print("[STEP347 ADDRESS SEARCH]", flush=True) print(f"[ADDRESS CANDIDATE COUNT] {len(candidates)}", flush=True) print("=" * 80, flush=True) for index, item in enumerate(candidates, start=1): query = item["query"] source = item["source"] print( f"[KAKAO ADDRESS TRY] index={index} source={source} query={query}", flush=True, ) try: result = kakao_geocode(query, rest_api_key) result["query"] = query result["query_source"] = source print( f"[KAKAO ADDRESS OK] index={index} source={source} " f"query={query} lat={result['lat']} lng={result['lng']}", flush=True, ) return result except Exception as exc: errors.append(f"{source}:{query}: {exc}") print( f"[KAKAO ADDRESS FAIL] index={index} source={source} " f"query={query} / {exc}", flush=True, ) raise Exception( "주소 좌표 변환 결과가 없습니다: " + " / ".join(item["query"] for item in candidates) + " / detail=" + "; ".join(errors) ) def kakao_keyword_place_search(realtor, rest_api_key, address_anchor=None): """ STEP346: 상호명뿐 아니라 단지명·건물명·행정동 조합까지 검색한다. 검증 기준: - 상호명 또는 단지명 일치 - 전화번호 일치 - 주소 토큰 일치 - 주소 지오코딩 기준점과의 거리 가장 높은 점수의 후보를 반환하되 위치 근거가 약한 후보는 거절한다. """ if not rest_api_key: return None office_name = clean_text(realtor.get("office_name") or "") office_phone = clean_text(realtor.get("office_phone") or realtor.get("phone") or "") mobile_phone = clean_text(realtor.get("mobile_phone") or realtor.get("mobile") or "") base_address = clean_text(realtor.get("office_address") or realtor.get("address") or "") detail_address = clean_text(realtor.get("office_detail_address") or "") full_address = clean_text(f"{base_address} {detail_address}") region_hint = extract_region_hint(full_address) complex_names = extract_complex_names(full_address) queries = [] def add_query(value, query_type): value = re.sub(r"\s+", " ", clean_text(value)).strip() if not value: return if any(item["query"] == value for item in queries): return queries.append({"query": value, "type": query_type}) # 상호명 검색 add_query(f"{office_name} {region_hint}", "office_region") add_query(f"{office_name} {office_phone}", "office_phone") add_query(f"{office_name} {mobile_phone}", "office_mobile") add_query(office_name, "office") # 단지명/건물명 검색 for complex_name in complex_names: add_query(f"{complex_name} {region_hint}", "complex_region") add_query(complex_name, "complex") # 주소 기반 키워드 검색 add_query(extract_korean_base_address(base_address), "base_address") add_query(base_address, "raw_base_address") expected_phones = { normalize_digits(office_phone), normalize_digits(mobile_phone), } expected_phones.discard("") office_key = normalize_korean_text( office_name .replace("공인중개사사무소", "") .replace("공인중개사", "") .replace("부동산", "") ) expected_complex_keys = { normalize_korean_text(name) for name in complex_names if normalize_korean_text(name) } expected_address_tokens = set(address_tokens(full_address)) candidates = [] for query_item in queries: query = query_item["query"] query_type = query_item["type"] print( f"[KAKAO KEYWORD SEARCH] type={query_type} query={query}", flush=True, ) try: res = requests.get( "https://dapi.kakao.com/v2/local/search/keyword.json", headers={"Authorization": f"KakaoAK {rest_api_key}"}, params={"query": query, "size": 15}, timeout=20, ) if res.status_code >= 400: print( f"[KAKAO KEYWORD HTTP {res.status_code}] {res.text[:300]}", flush=True, ) continue docs = (res.json() or {}).get("documents") or [] for doc in docs: place_name = clean_text(doc.get("place_name")) road_address_name = clean_text(doc.get("road_address_name")) address_name = clean_text(doc.get("address_name")) phone = clean_text(doc.get("phone")) category_name = clean_text(doc.get("category_name")) try: lat = float(doc.get("y")) lng = float(doc.get("x")) except Exception: continue normalized_blob = normalize_korean_text( " ".join([ place_name, road_address_name, address_name, phone, category_name, ]) ) score = 0 reasons = [] # 상호명 office_match = False normalized_office = normalize_korean_text(office_name) if normalized_office and normalized_office in normalized_blob: score += 120 office_match = True reasons.append("office_exact") elif office_key and office_key in normalized_blob: score += 85 office_match = True reasons.append("office_key") # 단지명/건물명 complex_match = False for complex_key in expected_complex_keys: if complex_key and complex_key in normalized_blob: score += 95 complex_match = True reasons.append("complex") break # 부동산 업종 if "부동산" in normalized_blob or "중개" in normalized_blob: score += 20 reasons.append("category") # 전화번호 found_phone = normalize_digits(phone) phone_match = bool(found_phone and found_phone in expected_phones) if phone_match: score += 120 reasons.append("phone") # 주소 토큰 candidate_tokens = set( address_tokens(f"{road_address_name} {address_name}") ) token_matches = expected_address_tokens & candidate_tokens address_match_count = len(token_matches) score += address_match_count * 18 if address_match_count: reasons.append(f"address_tokens:{address_match_count}") # 주소 기준 좌표와 거리 distance_m = None if address_anchor: try: distance_m = haversine_meters( address_anchor["lat"], address_anchor["lng"], lat, lng, ) if distance_m <= 100: score += 100 reasons.append("distance<=100") elif distance_m <= 300: score += 70 reasons.append("distance<=300") elif distance_m <= 700: score += 35 reasons.append("distance<=700") elif distance_m > 3000: score -= 150 reasons.append("distance>3000") except Exception: distance_m = None # 검색 종류별 보정 if query_type.startswith("office") and office_match: score += 20 if query_type.startswith("complex") and complex_match: score += 20 candidate = { "lat": lat, "lng": lng, "score": score, "query": query, "query_type": query_type, "place_name": place_name, "address": road_address_name or address_name, "phone": phone, "phone_match": phone_match, "office_match": office_match, "complex_match": complex_match, "address_match_count": address_match_count, "distance_m": distance_m, "reasons": reasons, "raw": doc, } print( "[KAKAO KEYWORD CANDIDATE] " f"score={score} distance={distance_m} " f"place={place_name} addr={candidate['address']} " f"reasons={','.join(reasons)}", flush=True, ) candidates.append(candidate) except Exception as exc: print( f"[KAKAO KEYWORD ERROR] query={query} / {exc}", flush=True, ) if not candidates: print("[KAKAO KEYWORD PLACE EMPTY]", flush=True) return None candidates.sort( key=lambda item: ( item["score"], -(item["distance_m"] if item["distance_m"] is not None else 999999), ), reverse=True, ) best = candidates[0] has_location_evidence = ( best["phone_match"] or best["address_match_count"] >= 2 or ( best["distance_m"] is not None and best["distance_m"] <= 700 and (best["office_match"] or best["complex_match"]) ) ) if best["score"] >= 110 and has_location_evidence: print("=" * 80, flush=True) print("[STEP347 FINAL KEYWORD RESULT]", flush=True) print( f"[SELECTED PLACE] {best.get('place_name')} / {best.get('address')}", flush=True, ) print( f"[SELECTED SCORE] {best.get('score')} " f"/ distance_m={best.get('distance_m')} " f"/ phone_match={best.get('phone_match')} " f"/ address_match_count={best.get('address_match_count')}", flush=True, ) print( f"[SELECTED REASONS] {', '.join(best.get('reasons') or [])}", flush=True, ) print("=" * 80, flush=True) print(f"[KAKAO KEYWORD PLACE BEST] {best}", flush=True) return best print( "[KAKAO KEYWORD PLACE REJECT] " f"score={best['score']} location_evidence={has_location_evidence} " f"candidate={best}", flush=True, ) return None def build_naver_map_url(lat=None, lng=None, office_name="", address=""): """ 좌표가 있으면 좌표 중심 지도를 연다. 좌표 변환에 실패하면 네이버지도 주소 검색 화면을 직접 연다. """ office_name = clean_text(office_name) address = clean_text(address) if lat is not None and lng is not None: return f"https://map.naver.com/v5/?c={lng},{lat},15,0,0,0,dh" search_query = address or office_name if not search_query: raise Exception("네이버지도 검색어가 없습니다.") return f"https://map.naver.com/p/search/{quote(search_query)}" def render_naver_map_screenshot(lat, lng, office_name, address, phone_text, output_path): output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) url = build_naver_map_url(lat, lng, office_name, address) with sync_playwright() as p: browser = p.chromium.launch( headless=True, args=[ "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage", "--lang=ko-KR", ], ) page = browser.new_page( viewport={"width": BROWSER_WIDTH, "height": BROWSER_HEIGHT}, device_scale_factor=1, locale="ko-KR", ) page.goto(url, wait_until="domcontentloaded", timeout=45000) # 네이버 지도는 타일/검색 결과 로딩이 늦을 수 있다. page.wait_for_timeout(7000 if lat is None or lng is None else 5000) # 주소 검색 fallback에서는 첫 번째 검색 결과를 클릭해 지도 중심을 맞춘다. if lat is None or lng is None: try: selectors = [ 'a[href*="/place/"]', 'li a', '[role="listitem"] a', ] clicked = False for selector in selectors: locator = page.locator(selector) if locator.count() > 0: locator.first.click(timeout=3000) clicked = True break if clicked: page.wait_for_timeout(3500) except Exception as e: print(f"[NAVER SEARCH RESULT CLICK SKIP] {e}", flush=True) cleanup_js = """ () => { const style = document.createElement('style'); style.id = 'real-auto-map-clean-style'; style.textContent = ` input, button, textarea, [role="button"], [class*="search"], [class*="Search"], [class*="panel"], [class*="Panel"], [class*="sidebar"], [class*="Sidebar"], [class*="control"], [class*="Control"], [class*="toolbar"], [class*="Toolbar"], [class*="toast"], [class*="Toast"], [class*="banner"], [class*="Banner"], [class*="promotion"], [class*="Promotion"], [class*="ad_"], [class*="Ad"], iframe { visibility: hidden !important; opacity: 0 !important; pointer-events: none !important; } `; const oldStyle = document.getElementById('real-auto-map-clean-style'); if (oldStyle) oldStyle.remove(); document.head.appendChild(style); const keywords = [ '네이버지도가', '밥값', '맛집이면', '지도앱 업데이트', '실시간 도로', '업데이트 안내', 'SmartAround', '주변', '리뷰' ]; const nodes = Array.from(document.querySelectorAll('div, section, article, aside, nav')); for (const node of nodes) { const text = (node.innerText || '').trim(); if (!text) continue; if (keywords.some(k => text.includes(k))) { let target = node; for (let i = 0; i < 3; i++) { if (target.parentElement && target.getBoundingClientRect().width < 900) { target = target.parentElement; } } target.style.visibility = 'hidden'; target.style.opacity = '0'; target.style.pointerEvents = 'none'; } } } """ page.evaluate(cleanup_js) page.wait_for_timeout(800) page.evaluate(cleanup_js) # STEP350: # 네이버 지도는 화면 상태에 따라 왼쪽 패널/지도 영역 폭이 달라진다. # 고정 X 좌표를 사용하지 않고 실제 렌더링 영역을 탐지한다. map_rect = page.evaluate( """ () => { const viewportWidth = window.innerWidth; const viewportHeight = window.innerHeight; const candidates = []; const selectors = [ 'canvas', '[class*="map"]', '[class*="Map"]', '[id*="map"]', '[id*="Map"]' ]; const seen = new Set(); for (const selector of selectors) { for (const el of document.querySelectorAll(selector)) { if (seen.has(el)) continue; seen.add(el); const style = window.getComputedStyle(el); const rect = el.getBoundingClientRect(); if ( style.display === 'none' || style.visibility === 'hidden' || Number(style.opacity || 1) === 0 ) { continue; } if (rect.width < 400 || rect.height < 300) { continue; } const visibleLeft = Math.max(0, rect.left); const visibleTop = Math.max(0, rect.top); const visibleRight = Math.min(viewportWidth, rect.right); const visibleBottom = Math.min(viewportHeight, rect.bottom); const visibleWidth = Math.max(0, visibleRight - visibleLeft); const visibleHeight = Math.max(0, visibleBottom - visibleTop); const area = visibleWidth * visibleHeight; if (area <= 0) continue; candidates.push({ tag: el.tagName, id: el.id || '', className: String(el.className || '').slice(0, 180), left: visibleLeft, top: visibleTop, right: visibleRight, bottom: visibleBottom, width: visibleWidth, height: visibleHeight, area }); } } candidates.sort((a, b) => b.area - a.area); if (!candidates.length) { return { found: false, left: 0, top: 0, right: viewportWidth, bottom: viewportHeight, width: viewportWidth, height: viewportHeight, centerX: viewportWidth / 2, centerY: viewportHeight / 2 }; } const best = candidates[0]; return { found: true, ...best, centerX: best.left + best.width / 2, centerY: best.top + best.height / 2 }; } """ ) detected_center_x = int( round( float( (map_rect or {}).get( "centerX", NAVER_MAP_CENTER_X_FALLBACK, ) ) ) ) detected_center_y = int( round( float( (map_rect or {}).get( "centerY", NAVER_MAP_CENTER_Y_FALLBACK, ) ) ) ) print( "[MAP RECT DETECTED] " f"found={(map_rect or {}).get('found')} " f"tag={(map_rect or {}).get('tag')} " f"left={(map_rect or {}).get('left')} " f"top={(map_rect or {}).get('top')} " f"width={(map_rect or {}).get('width')} " f"height={(map_rect or {}).get('height')} " f"center_x={detected_center_x} " f"center_y={detected_center_y}", flush=True, ) # 지도 위에 자체 오버레이를 얹는다. # 네이버 실제 지도 중심은 왼쪽 패널을 제외한 지도 영역 중앙이다. overlay_js = """ (args) => { args = args || []; const officeName = args[0] || ''; const address = args[1] || ''; const phoneText = args[2] || ''; const mapCenterX = Number(args[3] || 945); const mapCenterY = Number(args[4] || 310); const old = document.getElementById('real-auto-map-overlay'); if (old) old.remove(); const wrap = document.createElement('div'); wrap.id = 'real-auto-map-overlay'; wrap.style.position = 'fixed'; wrap.style.left = '0'; wrap.style.top = '0'; wrap.style.width = '100vw'; wrap.style.height = '100vh'; wrap.style.pointerEvents = 'none'; wrap.style.zIndex = '999999'; wrap.style.fontFamily = "Arial, 'Malgun Gothic', sans-serif"; const veil = document.createElement('div'); veil.style.position = 'absolute'; veil.style.left = '300px'; veil.style.top = '20px'; veil.style.width = '900px'; veil.style.height = '540px'; veil.style.background = 'rgba(255,255,255,.28)'; wrap.appendChild(veil); const label = document.createElement('div'); label.style.position = 'absolute'; label.style.left = `${mapCenterX}px`; label.style.top = `${mapCenterY - 82}px`; label.style.transform = 'translate(-50%, -100%)'; label.style.minWidth = '410px'; label.style.maxWidth = '620px'; label.style.padding = '16px 22px 18px'; label.style.borderRadius = '18px'; label.style.background = 'rgba(17,24,39,.94)'; label.style.color = '#fff'; label.style.boxShadow = '0 16px 36px rgba(0,0,0,.30)'; label.style.textAlign = 'center'; const title = document.createElement('div'); title.textContent = officeName || '중개사무소'; title.style.fontSize = '25px'; title.style.fontWeight = '900'; title.style.lineHeight = '1.32'; title.style.wordBreak = 'keep-all'; label.appendChild(title); if (address) { const addr = document.createElement('div'); addr.textContent = address; addr.style.marginTop = '8px'; addr.style.fontSize = '14px'; addr.style.lineHeight = '1.55'; addr.style.color = 'rgba(255,255,255,.88)'; addr.style.wordBreak = 'keep-all'; label.appendChild(addr); } if (phoneText) { const phone = document.createElement('div'); phone.textContent = phoneText; phone.style.marginTop = '8px'; phone.style.fontSize = '15px'; phone.style.fontWeight = '900'; phone.style.color = '#bbf7d0'; label.appendChild(phone); } const tail = document.createElement('div'); tail.style.position = 'absolute'; tail.style.left = '50%'; tail.style.bottom = '-20px'; tail.style.transform = 'translateX(-50%)'; tail.style.width = '0'; tail.style.height = '0'; tail.style.borderLeft = '20px solid transparent'; tail.style.borderRight = '20px solid transparent'; tail.style.borderTop = '22px solid rgba(17,24,39,.94)'; label.appendChild(tail); const pin = document.createElement('div'); pin.style.position = 'absolute'; pin.style.left = `${mapCenterX}px`; pin.style.top = `${mapCenterY - 38}px`; pin.style.width = '38px'; pin.style.height = '38px'; pin.style.background = '#ef4444'; pin.style.border = '5px solid #fff'; pin.style.borderRadius = '50% 50% 50% 0'; pin.style.transform = 'translate(-50%, -10%) rotate(-45deg)'; pin.style.boxShadow = '0 8px 20px rgba(0,0,0,.30)'; const dot = document.createElement('div'); dot.style.position = 'absolute'; dot.style.left = '13px'; dot.style.top = '13px'; dot.style.width = '11px'; dot.style.height = '11px'; dot.style.background = '#fff'; dot.style.borderRadius = '50%'; pin.appendChild(dot); wrap.appendChild(label); wrap.appendChild(pin); document.body.appendChild(wrap); } """ page.evaluate( overlay_js, [ office_name, address, phone_text, detected_center_x, detected_center_y, ], ) # 오버레이 반영 대기 page.wait_for_timeout(500) page.screenshot( path=str(output_path), full_page=False, type="png", clip={"x": CLIP_X, "y": CLIP_Y, "width": MAP_WIDTH, "height": MAP_HEIGHT}, ) browser.close() return str(output_path) def update_realtor_map_result(conn, realtor_id, lat=None, lng=None, image_path="", image_url="", error_message=""): columns = table_columns(conn, "blog_realtors") data = {} if "map_lat" in columns: data["map_lat"] = lat if "map_lng" in columns: data["map_lng"] = lng if "map_image_path" in columns: data["map_image_path"] = image_path if "map_image_url" in columns: data["map_image_url"] = image_url if "map_generated_at" in columns: data["map_generated_at"] = datetime.now() if image_url else None if "map_error_message" in columns: data["map_error_message"] = error_message[:2000] if error_message else None if "updated_at" in columns: data["updated_at"] = datetime.now() if not data: print("[MAP UPDATE SKIP] blog_realtors map columns not found") return sets = ", ".join([f"{k} = %s" for k in data.keys()]) values = list(data.values()) values.append(int(realtor_id)) with conn.cursor() as cur: cur.execute(f"UPDATE blog_realtors SET {sets} WHERE id = %s", values) conn.commit() def generate_for_realtor(conn, realtor, rest_api_key): realtor_id = int(realtor.get("id") or 0) office_name = clean_text(realtor.get("office_name") or "중개사무소") office_phone = clean_text(realtor.get("office_phone") or realtor.get("phone") or "") mobile_phone = clean_text(realtor.get("mobile_phone") or realtor.get("mobile") or "") base_address = clean_text(realtor.get("office_address") or realtor.get("address") or "") detail_address = clean_text(realtor.get("office_detail_address") or "") address = clean_text(f"{base_address} {detail_address}") if not realtor_id: raise Exception("realtor_id가 없습니다.") if not address: raise Exception(f"주소가 없습니다. realtor_id={realtor_id}") phone_text = " / ".join([x for x in [office_phone, mobile_phone] if x]) print(f"[MAP START] realtor_id={realtor_id}, office={office_name}") print( "[MAP OVERLAY MODE] dynamic map-rect detection", flush=True, ) # STEP346: # 주소 기준 좌표를 먼저 구해 검색 후보의 거리 검증 기준점으로 사용한다. address_anchor = None try: address_anchor = kakao_address_geocode_candidates( base_address, detail_address, rest_api_key, ) except Exception as address_error: print(f"[MAP ADDRESS ANCHOR FAIL] {address_error}", flush=True) # 상호명/단지명/건물명 검색 후보를 주소 기준점과 비교한다. place = kakao_keyword_place_search( realtor, rest_api_key, address_anchor=address_anchor, ) if place: lat = float(place["lat"]) lng = float(place["lng"]) geo_source = "kakao_verified_keyword" print( f"[MAP GEO SOURCE] {geo_source} " f"lat={lat} lng={lng} place={place.get('place_name')} " f"distance={place.get('distance_m')}", flush=True, ) elif address_anchor: lat = float(address_anchor["lat"]) lng = float(address_anchor["lng"]) geo_source = "kakao_address_anchor" print("=" * 80, flush=True) print("[STEP347 FINAL ADDRESS RESULT]", flush=True) print( f"[MAP GEO SOURCE] {geo_source} " f"lat={lat} lng={lng} " f"query={address_anchor.get('query')} " f"source={address_anchor.get('query_source')}", flush=True, ) print("=" * 80, flush=True) else: raise Exception( "상호명·단지명·주소 검색에서 사용할 수 있는 좌표를 찾지 못했습니다." ) filename = f"realtor_naver_map_{realtor_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}.png" local_path = OUTPUT_DIR / filename image_path = render_naver_map_screenshot( lat=lat, lng=lng, office_name=office_name, address=address, phone_text=phone_text, output_path=local_path, ) cdn = upload_header_image_to_cafe24( conn=conn, article_no=f"realtor_{realtor_id}", local_path=image_path, filename=filename, ) image_url = clean_text((cdn or {}).get("cdn_url") or "") if not image_url: raise Exception("Cafe24 CDN 업로드 결과 URL이 없습니다.") update_realtor_map_result( conn=conn, realtor_id=realtor_id, lat=lat, lng=lng, image_path=image_path, image_url=image_url, error_message="", ) print(f"[MAP DONE] realtor_id={realtor_id}, source={geo_source}, url={image_url}") return { "realtor_id": realtor_id, "lat": lat, "lng": lng, "geo_source": geo_source, "image_path": image_path, "image_url": image_url, } def main(): parser = argparse.ArgumentParser() parser.add_argument("--realtor-id", type=int, default=0) parser.add_argument("--all", action="store_true") parser.add_argument("--kakao-rest-api-key", type=str, default="") args = parser.parse_args() rest_api_key = args.kakao_rest_api_key or os.environ.get("KAKAO_REST_API_KEY", "") conn = get_conn() success = 0 failed = 0 try: if args.all: targets = fetch_realtors_for_batch(conn) else: if args.realtor_id <= 0: raise Exception("--realtor-id 또는 --all 옵션이 필요합니다.") targets = [fetch_realtor(conn, args.realtor_id)] print("=" * 80) print(f"[MAP TARGET COUNT] {len(targets)}") print("=" * 80) for realtor in targets: if not realtor: continue realtor_id = int(realtor.get("id") or 0) try: generate_for_realtor( conn=conn, realtor=realtor, rest_api_key=rest_api_key, ) success += 1 except Exception as e: failed += 1 print(f"[MAP ERROR] realtor_id={realtor_id} / {e}") try: update_realtor_map_result( conn=conn, realtor_id=realtor_id, error_message=str(e), ) except Exception: pass print("=" * 80) print(f"[MAP FINISH] success={success}, failed={failed}") print("=" * 80) if success == 0 and failed > 0: raise SystemExit(1) finally: try: conn.close() except Exception: pass if __name__ == "__main__": main()