# -*- coding: utf-8 -*- """ STEP106-11 - V2 Draft Address Apply School Table HTML 목적: - services/v2_complex_address_enricher.py 결과를 V2 초안 생성 데이터에 주입한다. - 기존 V1 draft generator를 직접 수정하지 않는다. - V2 draft generator에서만 import해서 사용한다. 적용 위치: D:/honghee/blog_api/services/v2_draft_address_apply.py 권장 호출 위치: V2 draft HTML 생성 직전, article/source_json/context dict가 준비된 시점 사용 예: from services.v2_draft_address_apply import apply_v2_address_to_article_context article, address_result = apply_v2_address_to_article_context(conn, article, force_refresh=False) # 이후 article["resolved_address"], article["school_names"] 등을 사용해 draft 생성 """ import json import re from copy import deepcopy from typing import Any, Dict, Tuple from services.v2_complex_address_enricher import resolve_and_cache_complex_address_v2 def normalize_space(text: Any) -> str: return re.sub(r"\s+", " ", str(text or "")).strip() 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 safe_json_dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, default=str) def extract_resolved_payload(address_result: Dict[str, Any]) -> Dict[str, Any]: """ resolve_and_cache_complex_address_v2 결과에서 draft에 필요한 필드만 평탄화한다. """ address_result = address_result or {} resolved = address_result.get("resolved") or {} cache = address_result.get("cache") or {} resolved_address = ( resolved.get("resolved_address") or resolved.get("jibun_address") or cache.get("jibun_address") or cache.get("resolved_address") or "" ) road_address = resolved.get("road_address") or cache.get("road_address") or "" jibun_address = resolved.get("jibun_address") or cache.get("jibun_address") or resolved_address school_names = resolved.get("school_names") or cache.get("school_names") or resolved.get("school_name") or cache.get("school_name") or "" school_name = resolved.get("school_name") or cache.get("school_name") or (school_names.split(",")[0].strip() if school_names else "") complex_no = str(resolved.get("complex_no") or cache.get("complex_no") or "") source = resolved.get("source") or cache.get("source") or "" address_region_name = ( address_result.get("region_name") or cache.get("region_name") or resolved.get("region_name") or "" ) address_complex_name = ( address_result.get("complex_name") or cache.get("complex_name") or resolved.get("complex_name") or "" ) payload = { "resolved_address": normalize_space(resolved_address), "road_address": normalize_space(road_address), "jibun_address": normalize_space(jibun_address), "school_name": normalize_space(school_name), "school_names": normalize_space(school_names), "complex_no": normalize_space(complex_no), "address_source": normalize_space(source), "address_ok": bool(address_result.get("ok") and normalize_space(resolved_address)), "address_from_cache": bool(address_result.get("from_cache")), "address_region_name": normalize_space(address_region_name), "address_complex_name": normalize_space(address_complex_name), } payload["school_html"] = build_school_table_html(payload) return payload def split_school_names(school_names: str): school_names = normalize_space(school_names) if not school_names: return [] parts = [] for x in re.split(r"[,/·|]", school_names): x = normalize_space(x) if x and x not in parts: parts.append(x) return parts def build_school_table_html(payload: Dict[str, Any]) -> str: """ 배정학교 정보가 있을 때 본문 삽입용 HTML table을 생성한다. - 네이버 검색 단지카드의 배정학교만 사용 - 없으면 빈 문자열 반환 """ school_names = normalize_space(payload.get("school_names") or payload.get("school_name")) if not school_names: return "" schools = split_school_names(school_names) if not schools: return "" school_text = ", ".join(schools) return f"""
배정학교 정보
배정학교 {school_text}
확인 기준 네이버 검색 단지정보
""".strip() def merge_address_into_source_json(source_json: Dict[str, Any], payload: Dict[str, Any]) -> Dict[str, Any]: source_json = deepcopy(source_json or {}) source_json["v2_address"] = { "ok": payload.get("address_ok"), "resolved_address": payload.get("resolved_address"), "road_address": payload.get("road_address"), "jibun_address": payload.get("jibun_address"), "school_name": payload.get("school_name"), "school_names": payload.get("school_names"), "school_html": payload.get("school_html"), "complex_no": payload.get("complex_no"), "source": payload.get("address_source"), "from_cache": payload.get("address_from_cache"), "region_name": payload.get("address_region_name"), "complex_name": payload.get("address_complex_name"), } # 기존 draft builder가 다양한 키를 볼 수 있으므로 공통 후보도 같이 주입한다. source_json["resolved_address"] = payload.get("resolved_address") source_json["road_address"] = payload.get("road_address") source_json["jibun_address"] = payload.get("jibun_address") source_json["school_name"] = payload.get("school_name") source_json["school_names"] = payload.get("school_names") source_json["school_html"] = payload.get("school_html") source_json["v2_school_html"] = payload.get("school_html") source_json["complex_no"] = payload.get("complex_no") source_json["address_source"] = payload.get("address_source") return source_json def apply_v2_address_to_article_context( conn, article: Dict[str, Any], force_refresh: bool = False, headless: bool = True, ) -> Tuple[Dict[str, Any], Dict[str, Any]]: """ V2 draft generator에서 호출할 함수. 반환: updated_article, address_result """ article = deepcopy(article or {}) address_result = resolve_and_cache_complex_address_v2( conn, article, force_refresh=force_refresh, headless=headless, ) payload = extract_resolved_payload(address_result) article["v2_address_result"] = address_result article["v2_address_payload"] = payload # article top-level에 주입 for k, v in payload.items(): article[k] = v # source_json / meta_json에도 주입 for json_col in ["source_json", "meta_json", "extra_json"]: parsed = safe_json_loads(article.get(json_col), {}) if isinstance(parsed, dict): merged = merge_address_into_source_json(parsed, payload) article[json_col] = safe_json_dumps(merged) return article, address_result def build_v2_address_template_context(article: Dict[str, Any]) -> Dict[str, Any]: """ HTML 템플릿 또는 draft builder에 넘길 최소 context. """ article = article or {} payload = article.get("v2_address_payload") or {} def val(key): return normalize_space(article.get(key) or payload.get(key)) return { "resolved_address": val("resolved_address"), "road_address": val("road_address"), "jibun_address": val("jibun_address"), "school_name": val("school_name"), "school_names": val("school_names"), "school_html": article.get("school_html") or payload.get("school_html") or "", "complex_no": val("complex_no"), "address_source": val("address_source"), } def replace_address_in_existing_html(html: str, article: Dict[str, Any]) -> str: """ 임시 호환용: 이미 만들어진 HTML에 '소재지' 행이 있고 값이 비어 있거나 부정확할 때 교체한다. 최종적으로는 draft 생성 전 context 주입 방식이 우선이다. """ html = str(html or "") ctx = build_v2_address_template_context(article) address = ctx.get("jibun_address") or ctx.get("resolved_address") if not address: return html # '소재지' 셀 다음 td 값을 보정 patterns = [ (r"(]*>\s*소재지\s*\s*]*>)(.*?)()", rf"\1{address}\3"), (r"(]*>\s*소재지\s*\s*]*>)(.*?)()", rf"\1{address}\3"), (r"(소재지\s*]+>\s*<[^>]+>)(.*?)(]+>)", rf"\1{address}\3"), ] new_html = html for pat, repl in patterns: new_html2 = re.sub(pat, repl, new_html, flags=re.I | re.S) if new_html2 != new_html: return new_html2 return html if __name__ == "__main__": import argparse from services.v2_complex_address_enricher import get_db_conn_from_env, fetch_article_for_enrich_test 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() 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 updated, result = apply_v2_address_to_article_context( conn, article, force_refresh=args.force_refresh, headless=args.headless, ) print(json.dumps({ "address_result": result, "template_context": build_v2_address_template_context(updated), "source_json_sample": safe_json_loads(updated.get("source_json"), {}), }, ensure_ascii=False, indent=2, default=str)) finally: conn.close()