# -*- coding: utf-8 -*- import re import time from urllib.parse import quote from playwright.sync_api import sync_playwright def normalize_digits(value): return re.sub(r"[^0-9]", "", str(value or "")) def uniq_keep_order(items): seen = set() result = [] for item in items: item = str(item or "").strip() if not item or item in seen: continue seen.add(item) result.append(item) return result def build_playwright_keywords(realtor): office_name = str(realtor.get("office_name") or "").strip() representative_name = str(realtor.get("representative_name") or "").strip() license_number = str(realtor.get("license_number") or "").strip() business_number = str(realtor.get("business_number") or "").strip() office_phone = str( realtor.get("office_phone") or realtor.get("phone") or "" ).strip() mobile_phone = str( realtor.get("mobile_phone") or realtor.get("mobile") or "" ).strip() office_phone_digits = normalize_digits(office_phone) mobile_phone_digits = normalize_digits(mobile_phone) keywords = [] # 복합 키워드 우선: 실제 네이버 검색 정확도 강화 if office_name and office_phone: keywords.append(f"{office_name} {office_phone}") if office_name and mobile_phone: keywords.append(f"{office_name} {mobile_phone}") if office_name and license_number: keywords.append(f"{office_name} {license_number}") if office_name and business_number: keywords.append(f"{office_name} {business_number}") if office_name and representative_name: keywords.append(f"{office_name} {representative_name}") # 전화번호 단독 검색 if office_phone: keywords.append(office_phone) if office_phone_digits: keywords.append(office_phone_digits) if mobile_phone: keywords.append(mobile_phone) if mobile_phone_digits: keywords.append(mobile_phone_digits) # 등록번호/사업자번호 단독 검색 if license_number: keywords.append(license_number) keywords.append(f"{license_number} 네이버부동산") keywords.append(f"{license_number} 매물") if business_number: keywords.append(business_number) # 중개업소명 단독 검색 if office_name: keywords.append(office_name) keywords.append(f"{office_name} 네이버부동산") keywords.append(f"{office_name} 매물") if representative_name: keywords.append(representative_name) return uniq_keep_order(keywords) def extract_article_nos(text): text = str(text or "") patterns = [ r"articleNo[=:\"'\s]+([0-9]{8,})", r"articleNumber[=:\"'\s]+([0-9]{8,})", r"/articles/([0-9]{8,})", r"article_no[=:\"'\s]+([0-9]{8,})", r"매물번호[^0-9]*([0-9]{8,})", ] results = [] for pattern in patterns: results.extend(re.findall(pattern, text, flags=re.IGNORECASE)) return uniq_keep_order(results) def collect_from_page(page, keyword, url): network_texts = [] def handle_response(response): try: response_url = response.url if ( "land" not in response_url and "article" not in response_url and "new.land.naver.com" not in response_url and "fin.land.naver.com" not in response_url ): return content_type = response.headers.get("content-type", "") if ( "application/json" in content_type or "text/html" in content_type or "text/plain" in content_type ): text = response.text() if text: network_texts.append(text) except Exception: pass page.on("response", handle_response) page.goto(url, wait_until="domcontentloaded", timeout=45000) try: page.wait_for_load_state("networkidle", timeout=15000) except Exception: pass page.wait_for_timeout(5000) html = page.content() current_url = page.url combined_text = html + "\n" + current_url + "\n" + "\n".join(network_texts) article_nos = extract_article_nos(combined_text) try: links = page.locator("a").evaluate_all( "(els) => els.map(a => a.href).filter(Boolean)" ) for href in links: article_nos.extend(extract_article_nos(href)) except Exception: links = [] article_nos = uniq_keep_order(article_nos) try: page.remove_listener("response", handle_response) except Exception: pass return { "keyword": keyword, "url": url, "final_url": current_url, "html_length": len(html), "network_response_count": len(network_texts), "network_text_length": sum(len(x) for x in network_texts), "article_nos": article_nos, "links_count": len(links), } def find_article_candidates_for_realtor_playwright(realtor, max_keywords=6, headless=True): keywords = build_playwright_keywords(realtor)[:max_keywords] candidates = [] debug_sources = [] with sync_playwright() as p: browser = p.chromium.launch( headless=headless, args=[ "--disable-blink-features=AutomationControlled", "--no-sandbox", ] ) context = browser.new_context( locale="ko-KR", viewport={"width": 1365, "height": 900}, user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/136.0.0.0 Safari/537.36" ) ) page = context.new_page() for keyword in keywords: q = quote(keyword) urls = [ f"https://search.naver.com/search.naver?query={q}", f"https://new.land.naver.com/search?query={q}", f"https://new.land.naver.com/offices?query={q}", f"https://fin.land.naver.com/search?keyword={q}", ] for url in urls: try: source = collect_from_page(page, keyword, url) debug_sources.append(source) for article_no in source.get("article_nos", []): candidates.append({ "article_no": article_no, "source_keyword": keyword, "source_url": source.get("final_url") or url, "source_status_code": 200, "collector": "playwright", }) except Exception as e: debug_sources.append({ "keyword": keyword, "url": url, "final_url": "", "html_length": 0, "article_nos": [], "links_count": 0, "error": str(e), }) time.sleep(1.2) context.close() browser.close() unique = {} for item in candidates: article_no = str(item.get("article_no") or "").strip() if article_no and article_no not in unique: unique[article_no] = item return { "keywords": keywords, "candidates": list(unique.values()), "debug_sources": debug_sources, }