# -*- coding: utf-8 -*- """ STEP106-14 - V2 Address Enricher V1 Field Extraction 목적: - V2 초안 생성 전에 단지 주소/학교 정보를 보강한다. - 1순위: services.naver_search_complex_resolver_v2.resolve_complex_address - 2순위: 기존 매물 수집 데이터 fallback - 결과: blog_complex_address_cache 저장 적용 위치: D:/honghee/blog_api/services/v2_complex_address_enricher.py 주의: - V1 publish_worker.py 수정 금지 - 기존 services/naver_search_complex_resolver.py 수정 금지 - 이 모듈은 V2 초안 생성 단계에서만 import """ import json import os import re from dataclasses import asdict from datetime import datetime from typing import Any, Dict, Optional, Tuple try: import pymysql except Exception: pymysql = None from services.naver_search_complex_resolver_v2 import ( clean_complex_search_query, resolve_complex_address, upsert_complex_address_cache, ) def normalize_space(text: Any) -> str: return re.sub(r"\s+", " ", str(text or "")).strip() def normalize_v2_address_text(value: Any) -> str: value = normalize_space(value) if not value: return "" value = value.replace("세종시 세종시 ", "세종시 ") value = value.replace("세종특별자치시 세종시 ", "세종특별자치시 ") value = re.sub(r"((?:[가-힣]+시|[가-힣]+도|세종특별자치시|세종시)(?:\s+[가-힣]+구|\s+[가-힣]+군)?\s+[가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?)\s+[가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?$", r"\1", value) value = re.sub(r"([가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?)\s+\1", r"\1", value) value = re.sub(r"([가-힣0-9]+(?:동|리|가))\s+\1\s+", r"\1 ", value) return value.strip() def normalize_v2_region_name(value: Any) -> str: value = normalize_v2_address_text(value) if not value: return "" m = re.search(r"^(세종특별자치시\s+[가-힣0-9]+(?:동|리|가|읍|면))", value) if m: return m.group(1).strip() m = re.search(r"^(세종시\s+[가-힣0-9]+(?:동|리|가|읍|면))", value) if m: return m.group(1).strip() m = re.search(r"^((?:서울특별시|서울시|부산광역시|부산시|대구광역시|대구시|인천광역시|인천시|광주광역시|광주시|대전광역시|대전시|울산광역시|울산시)\s+[가-힣]+(?:구|군)\s+[가-힣0-9]+(?:동|리|가|읍|면))", value) if m: return m.group(1).strip() m = re.search(r"^([가-힣]+도\s+[가-힣]+(?:시|군)(?:\s+[가-힣]+구)?\s+[가-힣0-9]+(?:동|리|가|읍|면))", value) if m: return m.group(1).strip() m = re.search(r"^(.+?[가-힣0-9]+(?:동|리|가|읍|면))\s+\d", value) if m: return m.group(1).strip() return value def safe_json_loads(value: Any, default=None): if default is None: default = {} if isinstance(value, (dict, list)): return value if value is None: return default try: return json.loads(str(value)) except Exception: return default def get_project_db_config(): """ DB 연결 우선순위: 1) 프로젝트 config.py 의 DB_CONFIG 2) 환경변수 BLOG_DB_* 운영 서버는 Cafe24 DB를 사용하므로 config.DB_CONFIG 우선. """ try: from config import DB_CONFIG if DB_CONFIG and isinstance(DB_CONFIG, dict): return { "host": DB_CONFIG.get("host"), "port": int(DB_CONFIG.get("port", 3306)), "user": DB_CONFIG.get("user"), "password": DB_CONFIG.get("password"), "database": DB_CONFIG.get("database"), "charset": DB_CONFIG.get("charset", "utf8mb4"), } except Exception: pass return { "host": os.getenv("BLOG_DB_HOST", "127.0.0.1"), "port": int(os.getenv("BLOG_DB_PORT", "3306")), "user": os.getenv("BLOG_DB_USER", "root"), "password": os.getenv("BLOG_DB_PASSWORD", ""), "database": os.getenv("BLOG_DB_NAME", "pjjh030551"), "charset": os.getenv("BLOG_DB_CHARSET", "utf8mb4"), } def get_db_conn_from_env(): if pymysql is None: raise RuntimeError("pymysql is not installed") cfg = get_project_db_config() return pymysql.connect( host=cfg["host"], port=int(cfg.get("port", 3306)), user=cfg["user"], password=cfg["password"], database=cfg["database"], charset=cfg.get("charset", "utf8mb4"), autocommit=False, cursorclass=pymysql.cursors.DictCursor, ) def get_table_columns(conn, table_name: str): with conn.cursor() as cur: cur.execute(f"SHOW COLUMNS FROM `{table_name}`") return [r["Field"] for r in cur.fetchall()] def table_exists(conn, table_name: str) -> bool: try: get_table_columns(conn, table_name) return True except Exception: return False def pick_first(*values) -> str: for v in values: s = normalize_space(v) if s: return s return "" def extract_region_hint(article: Dict[str, Any]) -> str: """ article/source_json/detail_json 어디에 있든 region 후보를 최대한 뽑는다. 예: 세종시, 부산시 연제구, 대전 서구 """ source = safe_json_loads(article.get("source_json"), {}) detail = safe_json_loads(article.get("detail_json"), {}) extra = safe_json_loads(article.get("extra_json"), {}) direct = pick_first( article.get("region_name"), article.get("article_address"), article.get("exposure_address"), article.get("address"), article.get("road_address"), article.get("city_name"), article.get("city"), article.get("sido"), source.get("region_name") if isinstance(source, dict) else "", source.get("city_name") if isinstance(source, dict) else "", source.get("city") if isinstance(source, dict) else "", detail.get("region_name") if isinstance(detail, dict) else "", detail.get("city_name") if isinstance(detail, dict) else "", extra.get("region_name") if isinstance(extra, dict) else "", ) if direct: return direct # 주소 후보에서 앞 지역명 추출 for obj in [article, source, detail, extra]: if not isinstance(obj, dict): continue for key, value in obj.items(): if "address" in str(key).lower() or "addr" in str(key).lower() or "소재지" in str(key): txt = normalize_space(value) if not txt: continue m = re.match(r"^((?:서울|부산|대구|인천|광주|대전|울산|세종)[^\s]*)(?:\s+([가-힣]+구|[가-힣]+군))?", txt) if m: return normalize_space(" ".join([x for x in m.groups() if x])) m = re.match(r"^([가-힣]+도)\s+([가-힣]+시|[가-힣]+군)(?:\s+([가-힣]+구))?", txt) if m: return normalize_space(" ".join([x for x in m.groups() if x])) return "" def extract_region_from_resolved_address(address: str) -> str: """ resolved_address에서 캐시용 region_name을 만든다. 예: 세종시 산울동 313-14 -> 세종시 산울동 부산시 연제구 연산동 1602 -> 부산시 연제구 연산동 충청남도 천안시 서북구 성성동 734 -> 충청남도 천안시 서북구 성성동 """ address = normalize_space(address) if not address: return "" tokens = address.split() if not tokens: return "" # 세종시 + 동/리 if tokens[0].startswith("세종"): for i, token in enumerate(tokens[1:], start=1): if token.endswith(("동", "리", "가")): return normalize_space(" ".join(tokens[:i+1])) return tokens[0] # 특별/광역시 + 구/군 + 동/리 if tokens[0].endswith(("시", "특별시", "광역시")): end_idx = 0 for i, token in enumerate(tokens[:5]): if token.endswith(("동", "리", "가")): end_idx = i break if end_idx: return normalize_space(" ".join(tokens[:end_idx+1])) if len(tokens) >= 2 and tokens[1].endswith(("구", "군")): return normalize_space(" ".join(tokens[:2])) return tokens[0] # 도 + 시/군 + 구 + 동/리 if tokens[0].endswith("도"): end_idx = 0 for i, token in enumerate(tokens[:6]): if token.endswith(("동", "리", "가")): end_idx = i break if end_idx: return normalize_space(" ".join(tokens[:end_idx+1])) if len(tokens) >= 3 and tokens[2].endswith("구"): return normalize_space(" ".join(tokens[:3])) if len(tokens) >= 2: return normalize_space(" ".join(tokens[:2])) return tokens[0] # fallback: 동/리 앞까지 for i, token in enumerate(tokens[:5]): if token.endswith(("동", "리", "가")): return normalize_space(" ".join(tokens[:i+1])) return "" def derive_region_name(region_hint: str = "", resolved_address: str = "", road_address: str = "") -> str: """ 캐시 저장용 region_name을 안정적으로 만든다. 우선순위: 1) 기존 region_hint 2) resolved/jibun address에서 시/구/동 추출 3) road address에서 가능한 앞 지역 추출 """ region_hint = normalize_v2_region_name(region_hint) if region_hint: return region_hint derived = normalize_v2_region_name(extract_region_from_resolved_address(resolved_address) or resolved_address) if derived: return derived derived = normalize_v2_region_name(extract_region_from_resolved_address(road_address) or road_address) if derived: return derived return "" def recursive_find_first(obj, key_names): key_names = {str(k).lower() for k in key_names} if isinstance(obj, dict): for k, v in obj.items(): if str(k).lower() in key_names and v not in [None, "", 0, "0", "-", []]: return v for v in obj.values(): found = recursive_find_first(v, key_names) if found not in [None, "", 0, "0", "-", []]: return found elif isinstance(obj, list): for item in obj: found = recursive_find_first(item, key_names) if found not in [None, "", 0, "0", "-", []]: return found return "" def extract_complex_name(article: Dict[str, Any]) -> str: """ 단지명 후보 추출. V1 generate_blog_drafts.py의 원본 row는 blog_realtor_articles 기반이므로 article_name/building_name/raw_json.articleDetail.aptName까지 폭넓게 본다. """ source = safe_json_loads(article.get("source_json"), {}) detail = safe_json_loads(article.get("detail_json"), {}) extra = safe_json_loads(article.get("extra_json"), {}) raw = safe_json_loads(article.get("raw_json"), {}) article_json = safe_json_loads(article.get("article_json"), {}) raw_article_detail = raw.get("articleDetail") if isinstance(raw, dict) else {} raw_article_addition = raw.get("articleAddition") if isinstance(raw, dict) else {} candidates = [ article.get("complex_name"), article.get("article_complex_name"), article.get("complex_title"), article.get("apt_name"), article.get("building_name"), article.get("article_name"), article.get("article_title"), article.get("complexName"), article.get("aptName"), article.get("publish_title"), article.get("draft_title"), article.get("title"), raw_article_detail.get("aptName") if isinstance(raw_article_detail, dict) else "", raw_article_detail.get("complexName") if isinstance(raw_article_detail, dict) else "", raw_article_detail.get("articleName") if isinstance(raw_article_detail, dict) else "", raw_article_addition.get("complexName") if isinstance(raw_article_addition, dict) else "", raw_article_addition.get("aptName") if isinstance(raw_article_addition, dict) else "", raw_article_addition.get("articleName") if isinstance(raw_article_addition, dict) else "", source.get("complex_name") if isinstance(source, dict) else "", source.get("complexName") if isinstance(source, dict) else "", source.get("aptName") if isinstance(source, dict) else "", source.get("articleName") if isinstance(source, dict) else "", detail.get("complex_name") if isinstance(detail, dict) else "", detail.get("complexName") if isinstance(detail, dict) else "", detail.get("aptName") if isinstance(detail, dict) else "", detail.get("articleName") if isinstance(detail, dict) else "", extra.get("complex_name") if isinstance(extra, dict) else "", article_json.get("complexName") if isinstance(article_json, dict) else "", article_json.get("aptName") if isinstance(article_json, dict) else "", ] # 그래도 없으면 원본 JSON 전체에서 재귀 검색 for obj in [raw, source, detail, extra, article_json]: if isinstance(obj, dict): candidates.extend([ recursive_find_first(obj, ["complexName", "aptName", "complex_name", "articleName"]), ]) for raw_name in candidates: raw_name = normalize_space(raw_name) if not raw_name: continue cleaned = clean_complex_search_query(raw_name) if cleaned: return cleaned return "" def build_listing_data(article: Dict[str, Any]) -> Dict[str, Any]: """ resolver_v2 fallback에 넘길 매물 수집 데이터 통합 dict. """ data = dict(article or {}) for key in ["source_json", "detail_json", "extra_json", "raw_json", "article_json"]: parsed = safe_json_loads(article.get(key), {}) if isinstance(parsed, dict): data[key] = parsed # V2 fallback 우선순위 안정화: # V1 detail/source에 이미 매물 주소가 있으면 top-level preferred 키로 복사한다. source = data.get("source_json") if isinstance(data.get("source_json"), dict) else {} source_detail = source.get("detail") if isinstance(source.get("detail"), dict) else {} raw = data.get("raw_json") if isinstance(data.get("raw_json"), dict) else {} article_detail = raw.get("articleDetail") if isinstance(raw.get("articleDetail"), dict) else {} # 지번 숫자가 있는 매물 주소를 우선한다. # articleDetail.exposureAddress는 '세종시 해밀동'처럼 동까지만 오는 경우가 많으므로 뒤로 둔다. preferred_address = pick_first( article.get("complex_resolved_address"), source_detail.get("complex_resolved_address"), source_detail.get("article_address"), source_detail.get("exposure_address"), source_detail.get("address"), article.get("article_address"), article.get("exposure_address"), article.get("address"), article_detail.get("jibunAddress"), article_detail.get("roadAddress"), article_detail.get("exposureAddress"), ) if preferred_address: data["v2_preferred_property_address"] = preferred_address data["complex_resolved_address"] = preferred_address data["article_address"] = preferred_address return data def find_cache(conn, region_name: str, complex_name: str) -> Optional[Dict[str, Any]]: if not table_exists(conn, "blog_complex_address_cache"): return None columns = get_table_columns(conn, "blog_complex_address_cache") select_cols = [c for c in [ "region_name", "complex_name", "resolved_address", "road_address", "jibun_address", "complex_no", "school_name", "school_names", "source", "last_checked_at", "meta_text" ] if c in columns] if not select_cols: return None clean_name = clean_complex_search_query(complex_name) # 1순위: region_name + complex_name 정확 매칭 if normalize_space(region_name): with conn.cursor() as cur: cur.execute( f""" SELECT {", ".join("`"+c+"`" for c in select_cols)} FROM blog_complex_address_cache WHERE region_name=%s AND (complex_name=%s OR complex_name=%s) ORDER BY last_checked_at DESC LIMIT 1 """, [region_name, complex_name, clean_name] ) row = cur.fetchone() if row: return row # 2순위: complex_name만 fallback. 단, 같은 이름이 여러 지역에 있을 수 있으므로 최신 1건만 보조 사용. with conn.cursor() as cur: cur.execute( f""" SELECT {", ".join("`"+c+"`" for c in select_cols)} FROM blog_complex_address_cache WHERE complex_name=%s OR complex_name=%s OR complex_name LIKE %s ORDER BY CASE WHEN region_name IS NULL OR region_name='' THEN 1 ELSE 0 END ASC, last_checked_at DESC LIMIT 1 """, [complex_name, clean_name, f"%{clean_name}%"] ) row = cur.fetchone() return row def resolve_and_cache_complex_address_v2( conn, article: Dict[str, Any], force_refresh: bool = False, headless: bool = True, ) -> Dict[str, Any]: """ V2 초안 생성 전에 호출할 핵심 함수. """ article = article or {} article_no = normalize_space(article.get("article_no") or article.get("articleNo")) complex_name = extract_complex_name(article) region_name = normalize_v2_region_name(extract_region_hint(article)) listing_data = build_listing_data(article) result = { "ok": False, "article_no": article_no, "complex_name": complex_name, "region_name": region_name, "from_cache": False, "cache": None, "resolved": None, "saved": None, "error": None, } if not complex_name: result["error"] = "complex_name_missing" return result if not force_refresh: cache = find_cache(conn, region_name, complex_name) if cache and normalize_space(cache.get("resolved_address")): cache_source = normalize_space(cache.get("source")) cache_meta = normalize_space(cache.get("meta_text")) suspicious_office_cache = ( cache_source == "listing_fallback_after_search_fail" and any(x in cache_meta.lower() for x in [ "articlerealtor.address", "office_address", "realtor_address", "realtor_info.address", ]) ) if suspicious_office_cache: print("[V2 ADDRESS CACHE IGNORE] suspicious office/realtor fallback", complex_name, cache.get("resolved_address")) else: result.update({ "ok": True, "from_cache": True, "cache": cache, "resolved": { "resolved_address": cache.get("jibun_address") or cache.get("resolved_address"), "road_address": cache.get("road_address"), "jibun_address": cache.get("jibun_address") or cache.get("resolved_address"), "complex_no": cache.get("complex_no"), "school_name": cache.get("school_name"), "school_names": cache.get("school_names") or cache.get("school_name"), "source": cache.get("source"), } }) return result try: resolved = resolve_complex_address( complex_name, region_hint=region_name, listing_data=listing_data, headless=headless, ) result["resolved"] = asdict(resolved) if resolved.ok and resolved.resolved_address: resolved.resolved_address = normalize_v2_address_text(resolved.resolved_address) resolved.jibun_address = normalize_v2_address_text(resolved.jibun_address) resolved.road_address = normalize_v2_address_text(resolved.road_address) final_region_name = derive_region_name( region_hint=region_name, resolved_address=resolved.resolved_address, road_address=resolved.road_address, ) result["region_name"] = final_region_name saved = upsert_complex_address_cache(conn, final_region_name, complex_name, resolved) result.update({ "ok": True, "saved": saved, }) return result result["error"] = resolved.error or "resolve_failed" return result except Exception as e: result["error"] = f"resolve_exception:{e}" return result def fetch_article_for_enrich_test(conn, article_no: str = "", draft_id: str = "", queue_id: str = "") -> Optional[Dict[str, Any]]: """ 테스트용으로 기존 테이블에서 article/draft/queue 정보를 최대한 가져온다. 실제 운영 연결은 V2 draft generator에서 현재 article dict를 직접 넘기는 방식 권장. """ article_no = normalize_space(article_no) draft_id = normalize_space(draft_id) queue_id = normalize_space(queue_id) queries = [] if article_no: # 자주 쓰일 법한 테이블명 후보. 없는 테이블은 무시. queries.extend([ ("blog_realestate_article_queue", "article_no=%s", [article_no]), ("blog_article_work_queue", "article_no=%s", [article_no]), ("blog_article_drafts", "article_no=%s", [article_no]), ]) if draft_id: queries.append(("blog_article_drafts", "id=%s", [draft_id])) if queue_id: queries.append(("blog_publish_queue", "id=%s", [queue_id])) for table, where, params in queries: try: if not table_exists(conn, table): continue with conn.cursor() as cur: cur.execute(f"SELECT * FROM `{table}` WHERE {where} LIMIT 1", params) row = cur.fetchone() if row: return row except Exception: continue return None if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--article-no", default="") parser.add_argument("--draft-id", default="") parser.add_argument("--queue-id", default="") parser.add_argument("--complex-name", default="") parser.add_argument("--region", default="") parser.add_argument("--force-refresh", action="store_true") parser.add_argument("--headless", action="store_true") args = parser.parse_args() cfg = get_project_db_config() print("[V2 ADDRESS ENRICHER DB]", { "host": cfg.get("host"), "port": cfg.get("port"), "user": cfg.get("user"), "database": cfg.get("database"), "charset": cfg.get("charset"), "password": "***" if cfg.get("password") else "", }) conn = get_db_conn_from_env() try: article = {} if args.article_no or args.draft_id or args.queue_id: article = fetch_article_for_enrich_test(conn, args.article_no, args.draft_id, args.queue_id) or {} if args.complex_name: article["complex_name"] = args.complex_name if args.region: article["region_name"] = args.region if args.article_no: article["article_no"] = args.article_no out = resolve_and_cache_complex_address_v2( conn, article, force_refresh=args.force_refresh, headless=args.headless, ) print(json.dumps(out, ensure_ascii=False, indent=2, default=str)) finally: conn.close()