# -*- coding: utf-8 -*- import os import sys import json import random import argparse import re import requests import socket import traceback from datetime import datetime BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(BASE_DIR) from db import get_conn # --------------------------------------------------------- # STEP29 OPERATION LOCK / PIPELINE LOG HELPERS # - 기존 성공 흐름은 유지하고, DB 작업락/로그만 보강한다. # - blog_app_locks 구조: lock_name, locked_at, expires_at, owner_token # - JSON 타입 미지원 환경을 고려해 meta_json/context_json에는 문자열 저장 # --------------------------------------------------------- def _safe_json_dumps(data): try: return json.dumps(data or {}, ensure_ascii=False, default=str) except Exception: return "{}" def _worker_token(prefix): try: host = socket.gethostname() except Exception: host = "unknown-host" return f"{prefix}:{host}:{os.getpid()}:{datetime.now().strftime('%Y%m%d%H%M%S')}" def acquire_db_lock(lock_name, ttl_minutes=180, owner_token=None): """ DB 기반 작업락. - 만료된 락은 제거 - 같은 lock_name이 살아 있으면 실행하지 않음 - 실패해도 예외를 밖으로 던지지 않고 False 반환 """ owner_token = owner_token or _worker_token(lock_name) conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute(""" DELETE FROM blog_app_locks WHERE expires_at < NOW() """) cur.execute(""" INSERT INTO blog_app_locks ( lock_name, locked_at, expires_at, owner_token ) VALUES ( %s, NOW(), DATE_ADD(NOW(), INTERVAL %s MINUTE), %s ) """, ( str(lock_name)[:100], int(ttl_minutes), str(owner_token)[:100], )) conn.commit() print(f"[DB LOCK ACQUIRED] {lock_name} owner={owner_token}") return True, owner_token except Exception as e: try: if conn: conn.rollback() except Exception: pass print(f"[DB LOCKED OR ERROR] {lock_name} / {e}") return False, owner_token finally: try: if conn: conn.close() except Exception: pass def release_db_lock(lock_name, owner_token): conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute(""" DELETE FROM blog_app_locks WHERE lock_name = %s AND owner_token = %s """, ( str(lock_name)[:100], str(owner_token)[:100], )) affected = cur.rowcount conn.commit() print(f"[DB LOCK RELEASED] {lock_name} affected={affected}") except Exception as e: print(f"[DB LOCK RELEASE ERROR] {lock_name} / {e}") finally: try: if conn: conn.close() except Exception: pass def create_pipeline_run(run_type, message="", meta=None): conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute(""" INSERT INTO blog_pipeline_runs ( run_type, status, started_at, message, meta_json, created_at ) VALUES ( %s, 'running', NOW(), %s, %s, NOW() ) """, ( str(run_type)[:50], str(message or "")[:2000], _safe_json_dumps(meta or {}), )) run_id = cur.lastrowid conn.commit() print(f"[PIPELINE RUN START] run_id={run_id}, run_type={run_type}") return run_id except Exception as e: print(f"[PIPELINE RUN START ERROR] {run_type} / {e}") return None finally: try: if conn: conn.close() except Exception: pass def finish_pipeline_run(run_id, status, total=0, success=0, failed=0, skipped=0, message="", meta=None): if not run_id: return conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute(""" UPDATE blog_pipeline_runs SET status = %s, finished_at = NOW(), total_count = %s, success_count = %s, fail_count = %s, skipped_count = %s, message = %s, meta_json = %s WHERE id = %s """, ( str(status)[:30], int(total or 0), int(success or 0), int(failed or 0), int(skipped or 0), str(message or "")[:2000], _safe_json_dumps(meta or {}), int(run_id), )) conn.commit() print(f"[PIPELINE RUN FINISH] run_id={run_id}, status={status}") except Exception as e: print(f"[PIPELINE RUN FINISH ERROR] run_id={run_id} / {e}") finally: try: if conn: conn.close() except Exception: pass def add_pipeline_log(run_id=None, level="info", step_name="", message="", article_no=None, realtor_id=None, context=None): conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute(""" INSERT INTO blog_pipeline_logs ( run_id, level, step_name, article_no, realtor_id, message, context_json, created_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, NOW() ) """, ( run_id, str(level or "info")[:20], str(step_name or "")[:100], str(article_no or "")[:30] if article_no else None, realtor_id, str(message or "")[:2000], _safe_json_dumps(context or {}), )) conn.commit() except Exception as e: print(f"[PIPELINE LOG ERROR] {step_name} / {e}") finally: try: if conn: conn.close() except Exception: pass AI_SHORTS_API_URL = "http://61.32.69.107:9000" AI_REALESTATE_API_URL = "http://61.32.69.107:9100" AI_SHORTS_API_KEY = "honghee-shorts-secret-2026" # STEP107-02: V2 주소 강제 재조회 옵션 V2_FORCE_REFRESH_ADDRESS = False V2_PIPELINE_VERSION = "v2" V2_DRAFT_GENERATOR = "generate_blog_drafts_v2" # STEP108-03: 중개사 대표사진 기본 SVG 픽토그램 V2_DEFAULT_REALTOR_AVATAR_REL = "mybox/defaults/default_realtor_avatar.svg" V2_DEFAULT_REALTOR_AVATAR_LOCAL = os.path.join(BASE_DIR, V2_DEFAULT_REALTOR_AVATAR_REL) # STEP108-04: # LOCAL_PUBLIC_BASE는 이 파일에 정의되어 있지 않으므로 사용하지 않는다. # 기존 부동산 이미지 API 공개 경로와 동일하게 AI_REALESTATE_API_URL 기준으로 생성한다. V2_DEFAULT_REALTOR_AVATAR_URL = ( f"{AI_REALESTATE_API_URL.rstrip('/')}/" f"{V2_DEFAULT_REALTOR_AVATAR_REL.replace(os.sep, '/')}" ) try: from services.v2_default_avatar_svg import ensure_default_realtor_avatar_svg except Exception as _avatar_svg_import_error: ensure_default_realtor_avatar_svg = None print("[V2 AVATAR SVG IMPORT WARN]", _avatar_svg_import_error) from services.ollama_writer import generate_ai_sections from services.blog_layout_blocks import build_blog_html from services.extra_image_service import get_extra_images_if_needed from services.hee_policy_service import log_policy_recommendation from services.cafe24_cdn_service import upload_header_image_to_cafe24 # --------------------------------------------------------- # STEP107-01 V2 전용 주소/학교 보강 # - V1 generate_blog_drafts.py가 아니라 복사본 generate_blog_drafts_v2.py에서만 사용 # --------------------------------------------------------- try: from services.v2_complex_address_enricher import resolve_and_cache_complex_address_v2 except Exception as _v2_addr_import_error: resolve_and_cache_complex_address_v2 = None print("[V2 ADDRESS IMPORT WARN]", _v2_addr_import_error) def json_default(obj): if isinstance(obj, datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") return str(obj) def clean_text(value): return str(value or "").strip() def normalize_v2_address_text(value): """ STEP107-07: '세종특별자치시 해밀동 649 해밀동 649'처럼 동+지번이 중복 결합된 주소를 정리한다. """ value = clean_text(value) if not value: return "" value = re.sub(r"\s+", " ", value).strip() value = value.replace("세종시 세종시 ", "세종시 ") value = value.replace("세종특별자치시 세종시 ", "세종특별자치시 ") # 세종특별자치시 해밀동 649 해밀동 649 -> 세종특별자치시 해밀동 649 value = re.sub(r"((?:[가-힣]+시|[가-힣]+도|세종특별자치시|세종시)(?:\s+[가-힣]+구|\s+[가-힣]+군)?\s+[가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?)\s+[가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?$", r"\1", value) # 일반 반복: 해밀동 649 해밀동 649 -> 해밀동 649 value = re.sub(r"([가-힣0-9]+(?:동|리|가)\s+\d+(?:-\d+)?)\s+\1", r"\1", value) # 동명만 반복: 해밀동 해밀동 649 -> 해밀동 649 value = re.sub(r"([가-힣0-9]+(?:동|리|가))\s+\1\s+", r"\1 ", value) return value.strip() def normalize_v2_region_name(value): """ STEP107-09: region_name은 지번 숫자를 제외한 지역 단위까지만 저장한다. 중요: Python 정규식 alternation은 앞 패턴을 먼저 잡으므로 '세종특별자치시|세종시' 순서가 아니라 '세종특별자치시'를 별도 처리한다. 그렇지 않으면 '세종특별자치시 해밀동 649'가 '세종특별자치시'까지만 잘릴 수 있다. """ 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() # 숫자 지번 앞까지만 fallback m = re.search(r"^(.+?[가-힣0-9]+(?:동|리|가|읍|면))\s+\d", value) if m: return m.group(1).strip() return value def choose_v2_region_name_for_detail(*values): """ STEP108-05: payload.region_name이 '세종특별자치시'처럼 시 단위까지만 들어오는 경우가 있다. region_name은 지번 주소에서 동/리까지 추출한 값을 우선 사용한다. """ normalized_candidates = [] for value in values: region = normalize_v2_region_name(value) if not region: continue if region not in normalized_candidates: normalized_candidates.append(region) if not normalized_candidates: return "" def score(region): s = 0 if re.search(r"(동|리|가|읍|면)$", region): s += 100 if re.search(r"(구|군)\s+[가-힣0-9]+(?:동|리|가|읍|면)$", region): s += 30 if "세종특별자치시" in region and len(region.split()) >= 2: s += 30 s += min(len(region), 50) return s normalized_candidates.sort(key=score, reverse=True) return normalized_candidates[0] def ensure_v2_default_realtor_avatar_asset(): """ STEP108-03: 기본 중개사 SVG 아바타 파일을 코드로 생성한다. """ if ensure_default_realtor_avatar_svg is None: print("[V2 DEFAULT REALTOR AVATAR SVG SKIP] import_failed") return { "ok": False, "path": V2_DEFAULT_REALTOR_AVATAR_LOCAL, "url": V2_DEFAULT_REALTOR_AVATAR_URL, "error": "import_failed", } try: result = ensure_default_realtor_avatar_svg( output_path=V2_DEFAULT_REALTOR_AVATAR_LOCAL, size=96, bg_color="#F1F3F5", icon_color="#8A8F98", stroke_color="#D7DCE2", force=False, ) print("[V2 DEFAULT REALTOR AVATAR SVG READY]", result) return { "ok": True, "path": V2_DEFAULT_REALTOR_AVATAR_LOCAL, "url": V2_DEFAULT_REALTOR_AVATAR_URL, "result": result, } except Exception as e: print("[V2 DEFAULT REALTOR AVATAR SVG ERROR]", str(e)) return { "ok": False, "path": V2_DEFAULT_REALTOR_AVATAR_LOCAL, "url": V2_DEFAULT_REALTOR_AVATAR_URL, "error": str(e), } def is_blank_realtor_image_url(value): value = clean_text(value) if not value: return True lowered = value.strip().lower() return lowered in { "none", "null", "undefined", "nan", "noimage", "no_image", "-", } def resolve_v2_realtor_profile_image_url(value): """ 대표사진이 없으면 기본 SVG 아바타 URL 반환. """ if is_blank_realtor_image_url(value): ensure_v2_default_realtor_avatar_asset() return V2_DEFAULT_REALTOR_AVATAR_URL return clean_text(value) def apply_v2_default_realtor_avatar_to_detail(detail): detail = dict(detail or {}) possible_keys = [ "profile_image_url", "realtor_profile_image_url", "representative_image_url", "representative_photo_url", "profileImageUrl", "representativePhotoUrl", "realtor_image_url", "realtor_photo_url", ] found_value = "" found_key = "" for key in possible_keys: value = clean_text(detail.get(key)) if value: found_value = value found_key = key break resolved_url = resolve_v2_realtor_profile_image_url(found_value) for key in possible_keys: detail[key] = resolved_url detail["v2_default_realtor_avatar_used"] = bool(is_blank_realtor_image_url(found_value)) detail["v2_realtor_profile_image_source_key"] = found_key detail["v2_realtor_profile_image_type"] = "svg_default" if detail["v2_default_realtor_avatar_used"] else "source_image" raw = parse_json_safely(detail.get("raw_json")) if isinstance(raw, dict): realtor = raw.get("articleRealtor") if isinstance(realtor, dict): for key in ["profileImageUrl", "representativePhotoUrl", "profile_image_url", "imageUrl"]: if is_blank_realtor_image_url(realtor.get(key)): realtor[key] = resolved_url raw["articleRealtor"] = realtor try: detail["raw_json"] = json.dumps(raw, ensure_ascii=False, default=json_default) except Exception: pass if detail["v2_default_realtor_avatar_used"]: print("[V2 DEFAULT REALTOR AVATAR SVG USED]", resolved_url) else: print("[V2 REALTOR PROFILE IMAGE USED]", resolved_url) return detail def normalize_duplicate_dong_text(value): """ 106동 106동 같은 중복 동 표시를 제거한다. """ value = clean_text(value) if not value: return "" value = re.sub(r"(\b\d{1,4}동)\s+\1\b", r"\1", value) parts = value.split() cleaned = [] for part in parts: if cleaned and cleaned[-1] == part and re.search(r"\d+동$", part): continue cleaned.append(part) return " ".join(cleaned) def remove_trailing_dong_for_sentence(value): """ 본문 첫 문장에서는 106동 같은 동 정보가 반복되면 기계적으로 보이므로 제거한다. 예: 해밀1단지마스터힐스 106동 -> 해밀1단지마스터힐스 """ value = normalize_duplicate_dong_text(value) value = re.sub(r"\s+\d{1,4}동\s*$", "", value).strip() return value def format_floor_text_for_display(value): """ 9/16 -> 9층 / 총 16층 이미 사람이 읽기 좋은 형식이면 그대로 둔다. """ value = clean_text(value) if not value: return "" m = re.match(r"^\s*(\d+)\s*/\s*(\d+)\s*$", value) if m: return f"{m.group(1)}층 / 총 {m.group(2)}층" m = re.match(r"^\s*(\d+)층\s*/\s*(\d+)층\s*$", value) if m: return f"{m.group(1)}층 / 총 {m.group(2)}층" return value.replace("층층", "층") def extract_exclusive_area_text(value): """ 대표이미지에는 공급/전용 전체보다 전용면적만 보여주는 것이 썸네일 가독성이 좋다. 예: 공급 78.71㎡ / 전용 59.4㎡ -> 전용 59.4㎡ """ value = clean_text(value) if not value: return "" m = re.search(r"전용\s*([0-9.]+\s*(?:㎡|m²|m2))", value, flags=re.IGNORECASE) if m: return "전용 " + m.group(1).replace("m2", "㎡").replace("m²", "㎡").strip() m = re.search(r"/\s*([0-9.]+\s*(?:㎡|m²|m2))", value, flags=re.IGNORECASE) if m: return "전용 " + m.group(1).replace("m2", "㎡").replace("m²", "㎡").strip() return value def compact_price_text(value): """ 대표이미지 가격 박스 가독성을 위해 공백을 줄인다. 5억 2,000 -> 5억2,000 """ value = clean_text(value) value = re.sub(r"(\d+억)\s+(\d)", r"\1\2", value) return value def extract_naver_realtor_id_from_anywhere(detail): """ 중개사 전체매물 링크 생성을 위해 naver realtor id를 최대한 찾는다. articleDetail.realtorId / source_json.detail.articleDetail.realtorId 등 대응. """ candidates = [ detail.get("naver_realtor_id"), detail.get("realtor_id_text"), detail.get("naver_realtor_code"), detail.get("realtor_account"), detail.get("realtorId"), ] realtor_info = detail.get("realtor_info") or {} if isinstance(realtor_info, dict): candidates.extend([ realtor_info.get("naver_realtor_id"), realtor_info.get("realtor_id_text"), realtor_info.get("realtorId"), ]) for json_key in ["source_json", "raw_json"]: try: raw = detail.get(json_key) if isinstance(raw, str): raw = json.loads(raw or "{}") if not isinstance(raw, dict): continue # source_json.detail.articleDetail.realtorId source_detail = raw.get("detail") or {} if isinstance(source_detail, dict): article_detail = source_detail.get("articleDetail") or source_detail.get("article_detail") or {} if isinstance(article_detail, dict): candidates.append(article_detail.get("realtorId")) candidates.append(source_detail.get("realtorId")) candidates.append(source_detail.get("naver_realtor_id")) # raw_json.articleDetail.realtorId article_detail = raw.get("articleDetail") or raw.get("article_detail") or {} if isinstance(article_detail, dict): candidates.append(article_detail.get("realtorId")) candidates.append(raw.get("realtorId")) except Exception: pass for value in candidates: value = clean_text(value) if value: return value return "" def select_header_background_from_article_images(images): """ 대표이미지 배경 선택 정책 V2. 운영 정책: - 같은 단지 여러 매물이 발행되므로 수집이미지 후보 중 랜덤 선택 - 단, 평면도/지도/중개사/실내사진은 절대 대표 배경으로 사용하지 않음 - category뿐 아니라 url/local_path/filename까지 검사 - 수집 후보가 없으면 빈 값 반환 → shorts_ai_api.py에서 MYBOX 랜덤 fallback """ if not images: return "" import random include_keywords = [ "ground_gallery", "complex_photo", "complex", "building", "outside", "view", "facility", "landscape", "community", "entrance", "gate", ] exclude_keywords = [ "floorplan", "floor_plan", "floor-plan", "groundplan", "ground_plan", "평면", "평면도", "plan", "layout", "realtor", "profile", "broker", "중개사", "map", "location_map", "지도", "inside", "interior", "room", "article", ] # URL/파일명 기준 제외. category가 잘못 저장된 경우까지 방어. exclude_url_keywords = [ "/floorplan/", "\\floorplan\\", "floorplan", "floor_plan", "groundplan", "ground_plan", "plan_", "_plan", "layout", "/realtor/", "\\realtor\\", "realtor_", "profile", "broker", "/map/", "\\map\\", "map_", "location_map", "/inside/", "\\inside\\", "inside_", "interior", ] def text_of(row): return " ".join([ clean_text(row.get("image_category")), clean_text(row.get("category")), clean_text(row.get("type")), clean_text(row.get("image_type")), clean_text(row.get("filename")), clean_text(row.get("local_path")), clean_text(row.get("image_url")), clean_text(row.get("original_url")), clean_text(row.get("cdn_url")), clean_text(row.get("url")), ]).lower() def get_cat(row): return clean_text( row.get("image_category") or row.get("category") or row.get("type") or row.get("image_type") or "" ).lower() def get_url(row): return clean_text( row.get("image_url") or row.get("original_url") or row.get("cdn_url") or row.get("local_path") or row.get("url") or "" ) weighted_pool = [] seen_urls = set() for row in images: cat = get_cat(row) url = get_url(row) combined = text_of(row) if not url.startswith("http"): continue if any(x in combined for x in exclude_keywords): print("[HEADER BG EXCLUDE]", cat, url) continue if any(x in url.lower() for x in exclude_url_keywords): print("[HEADER BG EXCLUDE URL]", cat, url) continue if not any(x in cat for x in include_keywords): # category가 비어있거나 애매한 이미지는 대표 배경 후보에서 제외 continue if url in seen_urls: continue seen_urls.add(url) weight = 1 if "ground_gallery" in cat: weight = 5 elif "complex_photo" in cat or "complex" in cat: weight = 5 elif "building" in cat or "outside" in cat or "entrance" in cat or "gate" in cat: weight = 4 elif "view" in cat or "landscape" in cat: weight = 3 elif "facility" in cat or "community" in cat: weight = 2 weighted_pool.extend([url] * weight) if not weighted_pool: print("[HEADER BACKGROUND RANDOM POOL] empty -> mybox fallback") return "" selected = random.choice(weighted_pool) print("[HEADER BACKGROUND RANDOM POOL]", len(seen_urls), "unique /", len(weighted_pool), "weighted") print("[HEADER BACKGROUND RANDOM SELECTED]", selected) return selected def normalize_middle_banner_image_url(detail): """ 중간배너는 네이버 포스팅 안정성을 위해 CSS background-image보다 가 안전하다. 이미 생성되어 저장된 banner URL이 있으면 우선 사용한다. 없으면 빈 값 반환 → HTML fallback 사용. """ candidates = [ detail.get("middle_banner_image_url"), detail.get("realtor_banner_image_url"), detail.get("cta_banner_image_url"), detail.get("banner_image_url"), detail.get("middle_banner_cdn_url"), detail.get("realtor_banner_cdn_url"), ] banner_asset = detail.get("banner_asset") or {} if isinstance(banner_asset, dict): candidates.extend([ banner_asset.get("cdn_url"), banner_asset.get("url"), banner_asset.get("image_url"), ]) realtor_info = detail.get("realtor_info") or {} if isinstance(realtor_info, dict): candidates.extend([ realtor_info.get("middle_banner_image_url"), realtor_info.get("realtor_banner_image_url"), realtor_info.get("cta_banner_image_url"), realtor_info.get("banner_image_url"), ]) for url in candidates: url = clean_text(url) if url.startswith("http"): return url return "" def build_realtor_listing_url(detail): """ 중개사 전체매물 CTA 링크 후보를 최대한 찾는다. 직접 fin.land URL이 있으면 우선 사용하고, 없으면 naver_realtor_id 기반 기본 중개사 페이지를 사용한다. """ candidates = [ detail.get("realtor_total_listing_url"), detail.get("total_listing_url"), detail.get("agency_listing_url"), detail.get("fin_land_url"), detail.get("realtor_page_url"), detail.get("office_url"), ] realtor_info = detail.get("realtor_info") or {} if isinstance(realtor_info, dict): candidates.extend([ realtor_info.get("realtor_total_listing_url"), realtor_info.get("total_listing_url"), realtor_info.get("agency_listing_url"), realtor_info.get("fin_land_url"), realtor_info.get("realtor_page_url"), ]) for url in candidates: url = clean_text(url) if url.startswith("http"): return url naver_realtor_id = extract_naver_realtor_id_from_anywhere(detail) if naver_realtor_id: return f"https://m.land.naver.com/agency/info/{naver_realtor_id}" return "" def parse_json_safely(value): if isinstance(value, dict): return value value = str(value or "").strip() if not value: return {} try: return json.loads(value) except Exception: return {} def nested_get_safe(obj, path, default=""): cur = obj for key in path: if not isinstance(cur, dict): return default cur = cur.get(key) if cur is None: return default return cur if cur not in [None] else default def recursive_find_first_safe(obj, key_names): key_names = {str(k).lower() for k in key_names} if isinstance(obj, dict): for key, value in obj.items(): if str(key).lower() in key_names and value not in [None, "", 0, "0", "-", []]: return value for value in obj.values(): found = recursive_find_first_safe(value, key_names) if found not in [None, "", 0, "0", "-", []]: return found elif isinstance(obj, list): for item in obj: found = recursive_find_first_safe(item, key_names) if found not in [None, "", 0, "0", "-", []]: return found return "" def enrich_detail_from_raw_json_for_property_type(detail): """ blog_realtor_articles 컬럼이 비어 있어도 raw_json의 네이버 원본 필드를 이용해 매물유형/설명/면적/주소 등 초안 핵심 정보를 보강한다. """ detail = dict(detail or {}) raw = parse_json_safely(detail.get("raw_json")) if not raw: return detail article_detail = raw.get("articleDetail") or {} article_addition = raw.get("articleAddition") or {} article_space = raw.get("articleSpace") or {} article_ground = raw.get("articleGround") or {} article_facility = raw.get("articleFacility") or {} def set_if_empty(key, *values): current = clean_text(detail.get(key)) if current and current not in ["-", "해당없음"]: return for value in values: text = clean_text(value) if text and text not in ["-", "해당없음"]: detail[key] = text return set_if_empty( "real_estate_type", detail.get("real_estate_type"), article_detail.get("realestateTypeName"), article_addition.get("articleRealEstateTypeName"), article_addition.get("realEstateTypeName"), article_detail.get("buildingTypeName"), article_detail.get("articleName"), ) set_if_empty( "real_estate_type_name", detail.get("real_estate_type_name"), article_detail.get("realestateTypeName"), article_addition.get("articleRealEstateTypeName"), article_addition.get("realEstateTypeName"), ) set_if_empty("trade_type", detail.get("trade_type"), article_detail.get("tradeTypeName"), article_addition.get("tradeTypeName")) set_if_empty("price_text", detail.get("price_text"), article_addition.get("dealOrWarrantPrc")) # 토지 면적 보강 ground_space = article_space.get("groundSpace") if ground_space not in [None, "", 0, "0"]: try: if float(ground_space) > 0: set_if_empty("area_info", f"대지면적 {float(ground_space):g}㎡") except Exception: pass article_realtor = raw.get("articleRealtor") or {} complex_display_address = extract_complex_display_address(raw, article_detail) if complex_display_address: print("[COMPLEX DISPLAY ADDRESS USED]", complex_display_address.replace("\\n", " / ")) set_if_empty( "address", complex_display_address, article_detail.get("exposureAddress"), article_realtor.get("address") if article_realtor else "", ) set_if_empty("building_usage", article_detail.get("currentUsage"), article_detail.get("recommendUsage")) set_if_empty("road_condition", "진입도로 없음" if article_ground.get("roadYN") == "N" else "") set_if_empty("building_allowed", "건축허가 없음" if article_ground.get("buildingAllowedYN") == "N" else "") # 중개사 원문 설명은 최우선 보존 set_if_empty( "article_feature_desc", detail.get("article_feature_desc"), article_detail.get("detailDescription"), article_detail.get("articleFeatureDescription"), article_addition.get("articleFeatureDesc"), ) # API로 넘길 때도 접근 가능하도록 원본 주요 필드 보관 detail["_raw_article_real_estate_type_name"] = clean_text(article_addition.get("articleRealEstateTypeName")) detail["_raw_realestate_type_name"] = clean_text(article_detail.get("realestateTypeName") or article_addition.get("realEstateTypeName")) detail["_raw_trade_building_type_code"] = clean_text(article_detail.get("tradeBuildingTypeCode")) detail["_raw_article_type_code"] = clean_text(article_detail.get("articleTypeCode") or article_addition.get("articleRealEstateTypeCode")) detail["_raw_realestate_type_code"] = clean_text(article_detail.get("realestateTypeCode") or article_addition.get("realEstateTypeCode")) return detail def extract_complex_display_address(raw, article_detail=None): """ 운영 복구 정책: - 매물 소재지는 article exposureAddress보다 단지/complex 주소를 우선 사용한다. - 네이버 단지 정보에서 도로명/지번 주소가 있으면 함께 표기한다. """ article_detail = article_detail or {} raw = raw or {} if isinstance(raw, str): raw = parse_json_safely(raw) if not isinstance(raw, dict): raw = {} road_keys = { "roadAddress", "roadAddressName", "road_address", "road_address_name", "roadNameAddress", "roadAddr", "road_addr", "newAddress", "new_address", "rnAddr", "roadBaseAddress", } jibun_keys = { "jibunAddress", "jibunAddressName", "jibun_address", "jibun_address_name", "address", "addressName", "oldAddress", "old_address", "lotAddress", "lot_address", "dongAddress", "bunjiAddress", } general_keys = { "complexAddress", "complex_address", "fullAddress", "full_address", "address", "addressName", "exposureAddress", "detailAddress", } complex_hint_keys = { "complex", "hscp", "overview", "complexInfo", "complexDetail", "complexOverview", "complexArticle", "danji", "단지" } road = "" jibun = "" general = "" def clean_addr(v): v = clean_text(v) if not v: return "" if "http://" in v or "https://" in v or "<" in v or ">" in v: return "" if len(v) < 4: return "" if v in ["아파트", "오피스텔", "주상복합"]: return "" return v def visit(obj, in_complex=False): nonlocal road, jibun, general if isinstance(obj, dict): local_complex = in_complex or any(str(k) in complex_hint_keys for k in obj.keys()) for k, v in obj.items(): key = str(k) if isinstance(v, (dict, list)): visit(v, local_complex or any(h.lower() in key.lower() for h in complex_hint_keys)) continue if not local_complex: continue value = clean_addr(v) if not value: continue if not road and key in road_keys: road = value elif not jibun and key in jibun_keys: jibun = value elif not general and key in general_keys: general = value elif isinstance(obj, list): for item in obj: visit(item, in_complex) preferred_roots = [ raw.get("complex"), raw.get("complexInfo"), raw.get("complexDetail"), raw.get("complexOverview"), raw.get("complexOverviewResponse"), raw.get("complexArticle"), raw.get("complexAddress"), raw.get("hscp"), raw.get("hscpInfo"), raw.get("overview"), raw.get("articleComplex"), ] for root_obj in preferred_roots: if root_obj: visit(root_obj, True) if not (road or jibun or general): visit(raw, False) if isinstance(article_detail, dict): for key in road_keys: if not road and clean_addr(article_detail.get(key)): road = clean_addr(article_detail.get(key)) for key in jibun_keys: if not jibun and clean_addr(article_detail.get(key)): jibun = clean_addr(article_detail.get(key)) if road and jibun and road != jibun: return f"{road}\n지번 {jibun}" if road: return road if jibun: return jibun if general: return general return "" def extract_broker_description(detail): """ 네이버 부동산의 중개사 매물설명 텍스트를 최대한 찾는다. 사람 작성 느낌을 살리기 위한 핵심 원문이다. """ candidates = [ detail.get("article_feature_desc"), detail.get("articleFeatureDescription"), detail.get("detailDescription"), detail.get("description"), ] raw = parse_json_safely(detail.get("raw_json")) if raw: article_detail = raw.get("articleDetail") or {} candidates.extend([ article_detail.get("articleFeatureDescription"), article_detail.get("detailDescription"), raw.get("detailDescription"), ]) for item in candidates: text = clean_text(item) if not text: continue # 전화번호/상호만 있는 너무 짧은 설명은 제외 if len(text.replace("\n", "").replace(" ", "")) < 15: continue return text return "" def build_broker_description_comment(detail): """ 운영 복구 정책: - 네이버 부동산 중개사가 직접 작성한 매물설명을 최우선 사용한다. - AI/템플릿 문장으로 대체하지 않는다. - 최대 1000자까지 보존한다. - 줄바꿈/기호/전화번호 정도만 정리한다. """ raw = extract_broker_description(detail) if not raw: return { "title": "", "content": "", } text = str(raw or "") text = text.replace("\r\n", "\n").replace("\r", "\n") # 장식 기호 정리 text = re.sub(r"[★☆■□◆◇▶▷●○◎※]+", " ", text) text = re.sub(r"[-=]{3,}", "\n", text) text = re.sub(r"\*+", " ", text) # 연락처는 법정 표시/중개사 정보에 별도 노출되므로 설명 본문에서는 제거 text = re.sub(r"\b0\d{1,2}-\d{3,4}-\d{4}\b", "", text) text = re.sub(r"\b010-\d{3,4}-\d{4}\b", "", text) text = re.sub(r"\b010\d{7,8}\b", "", text) lines = [] for line in text.split("\n"): line = clean_text(line) line = line.strip("-·ㆍ*• ") if not line: continue # 너무 짧은 상호/전화 안내성 줄 제거 if re.search(r"부동산|공인중개사|중개사무소", line) and len(line) <= 14: continue # 동일 줄 반복 제거 if line not in lines: lines.append(line) if not lines: return { "title": "", "content": "", } paragraphs = [] current = "" for line in lines: if len(current) + len(line) + 1 <= 180: current = (current + " " + line).strip() else: if current: paragraphs.append(current) current = line if current: paragraphs.append(current) content_parts = [] total = 0 for p in paragraphs: if total >= 1000: break remain = 1000 - total part = p[:remain].strip() if part: content_parts.append(part) total += len(part) content = "\n\n".join(content_parts).strip() if not content: return { "title": "", "content": "", } print("[BROKER DESCRIPTION RAW USED]", f"len={len(content)}") return { "title": "중개사 매물설명", "content": content, } def pick_one(items): if not items: return "" return random.choice(items) def pick_many(items, count=3): items = list(items or []) random.shuffle(items) return items[:count] def normalize_floor_text(floor_info): floor_info = clean_text(floor_info) if not floor_info: return "" return floor_info.replace("층층", "층") def detect_floor_type(floor_info): floor_info = clean_text(floor_info) if "/" not in floor_info: return "" try: current, total = floor_info.split("/", 1) current = int(str(current).replace("층", "").strip()) total = int(str(total).replace("층", "").strip()) if total <= 0: return "" ratio = current / total if ratio >= 0.7: return "high" if ratio <= 0.3: return "low" return "middle" except Exception: return "" def detect_direction_type(direction): direction = clean_text(direction) if not direction: return "" if "남" in direction: return "south" if "동" in direction: return "east" if "서" in direction: return "west" if "북" in direction: return "north" return "" DIRECTION_MAP = { "E": "동향", "W": "서향", "S": "남향", "N": "북향", "ES": "동남향", "SE": "남동향", "WS": "서남향", "SW": "남서향", "EN": "동북향", "NE": "북동향", "WN": "서북향", "NW": "북서향", } def convert_direction(direction_code): direction_code = clean_text( direction_code ).upper() if not direction_code: return "" return DIRECTION_MAP.get( direction_code, direction_code ) COMMENT_TITLES = [ "현장 설명", "중개사 코멘트", "실제 매물 설명", "현장 체크 포인트", ] COMMENT_OPENERS = [ "중개사 설명 기준으로는", "현장 안내 내용을 보면", "실제 등록된 설명에서는", "대표 설명 기준으로는", ] FEATURE_REWRITE_POOL = { "채광": [ "채광 방향을 중요하게 보시는 분들께 참고가 될 수 있습니다.", "실내 채광 흐름을 함께 체크해보시면 좋겠습니다.", ], "남향": [ "남향 기준이라 일조량 부분도 함께 참고해보시면 좋겠습니다.", "채광과 개방감을 중요하게 보시는 분들께 잘 맞을 수 있습니다.", ], "로얄": [ "선호도 높은 라인과 층 조건을 함께 참고해보시면 좋겠습니다.", ], "뷰": [ "개방감과 전망 부분도 함께 체크해보시면 좋겠습니다.", ], "수리": [ "실내 컨디션과 관리 상태를 함께 참고해보시면 좋겠습니다.", ], "깨끗": [ "실내 관리 상태가 비교적 안정적으로 유지된 느낌입니다.", ], } def split_feature_sentences(text): text = clean_text(text) if not text: return [] separators = [ "\n", ".", ",", "/", "·", ] items = [text] for sep in separators: temp = [] for item in items: temp.extend(item.split(sep)) items = temp result = [] for item in items: item = clean_text(item) if len(item) < 2: continue if item in result: continue result.append(item) return result def rewrite_feature_sentence(sentence): sentence = clean_text(sentence) if not sentence: return "" for keyword, pool in FEATURE_REWRITE_POOL.items(): if keyword in sentence: return pick_one(pool) generic_pool = [ f"{sentence} 부분을 함께 참고해보시면 좋겠습니다.", f"실제 현장에서는 {sentence} 부분도 체크해보시면 좋겠습니다.", f"공간 활용 측면에서 {sentence} 조건도 함께 확인해보시면 좋겠습니다.", ] return pick_one(generic_pool) def build_feature_comment_block(detail): broker_comment = build_broker_description_comment(detail) if broker_comment.get("content"): return broker_comment raw_text = clean_text( detail.get("article_feature_desc") ) if not raw_text: return { "title": "", "content": "", } items = split_feature_sentences(raw_text) if not items: return { "title": "", "content": "", } selected = pick_many(items, 3) lines = [] opener = pick_one(COMMENT_OPENERS) for idx, item in enumerate(selected): line = rewrite_feature_sentence(item) if idx == 0: line = f"{opener} {line}" lines.append(line) return { "title": pick_one(COMMENT_TITLES), "content": " ".join(lines) } def build_intro_text(detail): article_name = ( remove_trailing_dong_for_sentence(detail.get("article_name")) or remove_trailing_dong_for_sentence(detail.get("building_name")) or remove_trailing_dong_for_sentence(detail.get("complex_name")) or "해당 매물" ) trade_type = clean_text(detail.get("trade_type")) real_estate_type = clean_text(detail.get("real_estate_type")) price_text = clean_text(detail.get("price_text")) area_info = clean_text(detail.get("area_info")) floor_info = format_floor_text_for_display(detail.get("floor_info")) direction = convert_direction( detail.get("direction") or detail.get("direction_code") ) openers = [ f"이번에 소개해드리는 매물은 {article_name} 매매 물건입니다.", f"{article_name} 매물을 찾고 계셨다면 참고해보셔도 좋겠습니다.", f"실거주를 고려하시는 분들께 {article_name} 매물을 안내드립니다.", f"생활 편의성과 구조를 함께 보실 수 있는 {article_name} 매물입니다.", f"문의가 꾸준히 이어지는 타입의 {article_name} 매물입니다.", f"조건과 위치를 함께 비교해보기 좋은 {article_name} 매물입니다.", ] second_lines = [] if trade_type and real_estate_type: second_lines += [ f"{real_estate_type} {trade_type} 매물로 현재 조건을 기준으로 안내드립니다.", f"{trade_type} 조건으로 나온 {real_estate_type} 매물입니다.", f"{real_estate_type}을 찾는 분들께 참고가 될 만한 {trade_type} 매물입니다.", ] elif trade_type: second_lines += [ f"{trade_type} 조건으로 확인되는 매물입니다.", f"현재 {trade_type} 기준으로 안내 가능한 매물입니다.", ] if price_text: second_lines += [ f"가격은 {price_text} 기준으로 확인됩니다.", f"현재 안내되는 가격 조건은 {price_text}입니다.", f"예산을 맞춰 비교해보실 때 {price_text} 조건을 기준으로 비교해보시면 좋습니다.", ] if area_info: second_lines += [ f"면적은 {area_info} 기준으로 확인됩니다.", f"{area_info} 구조라 공간 활용을 함께 확인해보시면 좋습니다.", f"실사용 면적과 동선을 함께 보기 좋은 {area_info} 타입입니다.", ] if direction: second_lines += [ f"방향은 {direction} 기준으로 확인됩니다.", f"{direction} 기준이라 채광과 개방감을 중요하게 보시는 분들께 잘 맞을 수 있습니다.", f"{direction} 방향 기준으로 일조량 부분도 함께 참고해보시면 좋겠습니다.", f"{direction} 방향 특성을 고려해 실내 분위기를 함께 체크해보시면 좋겠습니다.", ] lines = [ pick_one(openers), pick_one(second_lines) if second_lines else "", ] return "\n".join([x for x in lines if x]) def build_feature_text(detail): floor_info = format_floor_text_for_display(detail.get("floor_info")) direction = clean_text(detail.get("direction")) area_info = clean_text(detail.get("area_info")) real_estate_type = clean_text(detail.get("real_estate_type")) feature = clean_text(detail.get("article_feature") or detail.get("feature")) address = clean_text(detail.get("address") or detail.get("road_address")) floor_type = detect_floor_type(floor_info) direction_type = detect_direction_type(direction) pool = [ "전체적으로 생활 동선이 무난하게 잡혀 있어 실거주 관점에서 보기 좋은 매물입니다.", "사진과 조건을 함께 비교해보면 기본기가 잘 갖춰진 매물로 볼 수 있습니다.", "처음 집을 보시는 분들도 구조와 조건을 비교하기 쉬운 타입입니다.", "실제 거주를 고려하신다면 채광, 동선, 주변 환경을 함께 보시면 좋습니다.", "가격 조건과 내부 컨디션을 함께 비교해볼 만한 매물입니다.", "같은 단지 안에서도 층수와 방향에 따라 체감이 달라질 수 있어 직접 확인을 추천드립니다.", "실사용 공간을 중요하게 보시는 분들께는 확인해보실 만한 매물입니다.", "주거 안정성과 생활 편의성을 함께 고려하시는 분들께 잘 맞을 수 있습니다.", "입주 일정과 조건만 맞는다면 충분히 문의해보실 만한 매물입니다.", "단순히 가격만 보기보다는 위치, 구조, 관리 상태를 함께 확인하시는 것이 좋습니다.", "실내 동선과 수납, 채광 조건을 함께 비교해보면 판단이 쉬운 매물입니다.", "주변 생활권과 단지 분위기를 함께 고려하면 장점이 더 잘 보이는 매물입니다.", ] if direction_type == "south": pool += [ "남향 계열 방향이라 채광을 중요하게 보시는 분들께 특히 참고가 됩니다.", "햇빛이 들어오는 시간대와 거실 방향을 함께 확인해보시면 만족도가 높을 수 있습니다.", "채광을 선호하시는 분들에게는 장점으로 볼 수 있는 방향 조건입니다.", ] if floor_type == "high": pool += [ "상대적으로 높은 층에 해당해 개방감과 조망감을 기대해볼 수 있습니다.", "고층 선호도가 있는 분들께는 우선적으로 검토해볼 만한 조건입니다.", ] elif floor_type == "middle": pool += [ "중간층에 가까운 조건이라 생활 편의성과 안정감 사이의 균형을 기대할 수 있습니다.", "층수 부담이 크지 않아 실거주용으로 무난하게 검토하기 좋습니다.", ] elif floor_type == "low": pool += [ "저층부에 가까워 이동 편의성을 중요하게 보시는 분들께 참고가 됩니다.", "엘리베이터 대기나 이동 동선을 줄이고 싶은 분들에게 맞을 수 있습니다.", ] if area_info: pool += [ "면적 구성을 보면 가족 구성원 수와 생활 패턴에 맞춰 판단하기 좋습니다.", "공간을 어떻게 나누어 사용할지 상상해보며 보시면 더 도움이 됩니다.", ] if real_estate_type: pool += [ f"{real_estate_type} 특성상 관리 상태와 주변 환경을 함께 확인하는 것이 중요합니다.", f"{real_estate_type}을 찾는 분들이 주로 보는 조건들을 기준으로 정리해볼 만합니다.", ] if feature: pool += [ f"매물 특징으로는 {feature} 부분을 함께 참고하시면 좋습니다.", f"현장 확인 전에는 {feature} 내용을 먼저 체크해보시면 도움이 됩니다.", ] if address: pool += [ "소재지 기준으로 주변 생활권과 이동 동선을 함께 확인해보시는 것을 추천드립니다.", "주소지를 기준으로 주변 편의시설과 교통 흐름을 함께 비교해보시면 좋습니다.", ] return "\n".join(pick_many(pool, 4)) def build_location_text(detail): address = clean_text(detail.get("address") or detail.get("road_address")) complex_name = clean_text(detail.get("complex_name") or detail.get("article_name")) pool = [ "생활권은 실제 거주 만족도에 큰 영향을 주기 때문에 주변 편의시설과 이동 동선을 함께 확인해보시는 것이 좋습니다.", "주변 환경은 사진만으로 판단하기 어려운 부분이 있어 방문 시 단지 분위기와 도로 접근성을 함께 보시면 좋습니다.", "출퇴근 동선, 장보기, 병원, 학원 등 생활에 필요한 시설과의 거리를 함께 비교해보시면 판단이 쉬워집니다.", "단지 주변 분위기와 접근성은 실거주 만족도를 좌우하는 중요한 요소입니다.", "입지만 볼 때는 지도상 거리뿐 아니라 실제 이동 동선도 함께 확인하는 것이 좋습니다.", "주변 생활 인프라가 잘 맞는지 확인하면 장기 거주 관점에서도 도움이 됩니다.", ] if address: pool += [ f"소재지는 {address} 기준으로 확인됩니다.", f"{address} 생활권을 고려하시는 분들께 참고가 될 수 있는 매물입니다.", ] if complex_name: pool += [ f"{complex_name} 주변 환경과 단지 분위기를 함께 비교해보시면 좋습니다.", ] return "\n".join(pick_many(pool, 3)) def build_space_text(detail): area_info = clean_text(detail.get("area_info")) floor_info = clean_text(detail.get("floor_info")) direction = clean_text(detail.get("direction")) pool = [ "공간은 단순 면적보다 실제 동선과 가구 배치가 중요합니다.", "거실과 방의 배치, 수납 공간, 주방 동선을 함께 보시면 실제 사용감이 더 잘 보입니다.", "사진을 보실 때는 창 위치와 채광, 통풍 흐름을 함께 확인해보시면 좋습니다.", "같은 면적이라도 구조에 따라 체감 공간은 달라질 수 있습니다.", "실제 방문 시에는 거실 폭, 방 크기, 주방 동선을 중심으로 확인하시는 것을 추천드립니다.", "공간 활용도는 가족 구성원 수와 생활 패턴에 따라 다르게 느껴질 수 있습니다.", "수납과 동선이 잘 맞는지 확인하면 실제 거주 만족도를 판단하는 데 도움이 됩니다.", ] if area_info: pool += [ f"면적은 {area_info} 기준으로, 실제 사용할 공간을 중심으로 보시면 좋습니다.", f"{area_info} 타입이라 방 구성과 거실 활용도를 함께 체크해보시면 좋습니다.", ] if floor_info: pool += [ f"층수는 {floor_info} 기준으로 확인되며, 채광과 조망을 함께 비교해볼 수 있습니다.", ] if direction: pool += [ f"방향은 {direction}으로 확인되며, 시간대별 채광을 함께 확인해보시면 좋습니다.", ] return "\n".join(pick_many(pool, 4)) def build_price_text_block(detail): price_text = clean_text(detail.get("price_text")) trade_type = clean_text(detail.get("trade_type")) pool = [ "가격은 현재 시장 흐름과 주변 유사 매물 조건을 함께 비교해보시는 것이 좋습니다.", "매물 가격은 층수, 방향, 내부 상태, 입주 가능 시점에 따라 체감 가치가 달라질 수 있습니다.", "단순 가격만 보기보다는 조건 대비 만족도를 함께 확인하는 것이 중요합니다.", "같은 단지 안에서도 동, 층, 방향에 따라 가격 차이가 발생할 수 있습니다.", "예산 범위 안에서 구조와 위치가 잘 맞는지 함께 비교해보시면 좋습니다.", "현재 조건이 본인의 자금 계획과 맞는지 확인한 뒤 현장 방문을 잡는 것이 효율적입니다.", ] if price_text: pool += [ f"현재 가격 조건은 {price_text} 기준으로 확인됩니다.", f"{price_text} 조건을 기준으로 주변 매물과 함께 비교해보시면 좋습니다.", ] if trade_type: pool += [ f"{trade_type} 매물은 조건 변경이 있을 수 있으므로 상담 전 현재 가능 여부를 확인하시는 것이 좋습니다.", ] return "\n".join(pick_many(pool, 3)) def build_school_text(detail): schools = detail.get("schools") or [] if schools: lines = [] for school in schools[:3]: school_name = clean_text(school.get("school_name")) school_type = clean_text(school.get("school_type")) distance = clean_text(school.get("distance_text") or school.get("distance")) if not school_name: continue label = " ".join([school_type, distance]).strip() if label: lines.append(f"인근 교육시설로는 {school_name}({label}) 정보를 참고하실 수 있습니다.") else: lines.append(f"인근 교육시설로는 {school_name} 정보를 참고하실 수 있습니다.") if lines: return "\n".join(lines) pool = [ "학군과 교육환경은 가족 단위 실거주 수요에서 중요한 체크 포인트입니다.", "자녀 계획이 있으시다면 주변 학교와 학원 접근성을 함께 확인해보시면 좋습니다.", "교육환경은 실제 생활 동선과 밀접하므로 방문 전 지도 기준으로 한 번 더 확인을 추천드립니다.", "학교 배정과 통학 동선은 변동 가능성이 있어 상담 시 별도 확인이 필요합니다.", ] return "\n".join(pick_many(pool, 2)) def build_realtor_text(detail): info = detail.get("realtor_info") or {} office_name = clean_text(info.get("office_name")) pool = [ "상담 시에는 현재 매물 가능 여부, 가격 조건, 입주 가능일을 먼저 확인하시면 좋습니다.", "방문 전 유선으로 매물 상태와 조건 변동 여부를 확인하시면 더 정확한 안내를 받을 수 있습니다.", "사진만으로 판단하기 어려운 부분은 현장 방문을 통해 직접 확인하시는 것이 가장 좋습니다.", "실제 매물 여부와 세부 조건은 상담 시점에 다시 확인해드리는 것이 안전합니다.", ] if office_name: pool += [ f"{office_name}에서 현재 확인 가능한 조건을 기준으로 안내드립니다.", f"자세한 상담은 {office_name}을 통해 확인하실 수 있습니다.", ] return "\n".join(pick_many(pool, 3)) def build_closing_text(detail): pool = [ "관심 있으신 분들은 현장 확인과 함께 자세한 상담을 받아보시길 권해드립니다.", "사진과 기본 정보만으로는 판단이 어려운 부분이 있으니 직접 확인해보시는 것을 추천드립니다.", "조건이 맞는 매물은 빠르게 변동될 수 있으니 문의 전 현재 가능 여부를 확인해 주세요.", "방문 상담을 통해 실제 구조와 주변 분위기를 함께 확인해보시면 좋습니다.", "궁금하신 점은 편하게 문의 주시면 현재 확인 가능한 내용 기준으로 안내드리겠습니다.", "매물 조건은 변동될 수 있으므로 상담 시점의 최신 정보를 기준으로 확인해 주세요.", ] return pick_one(pool) def build_human_blog_context(detail_row): intro_text = build_intro_text(detail_row) feature_text = build_feature_text(detail_row) location_text = build_location_text(detail_row) space_text = build_space_text(detail_row) price_text_block = build_price_text_block(detail_row) school_text = build_school_text(detail_row) realtor_text = build_realtor_text(detail_row) outro_text = build_closing_text(detail_row) return { "intro_text": intro_text, "feature_text": feature_text, "location_text": location_text, "space_text": space_text, "price_text_block": price_text_block, "school_text": school_text, "realtor_text": realtor_text, "outro_text": outro_text, } def enrich_ai_sections(ai_sections, human_context): ai_sections = ai_sections or {} mapping = { "intro": "intro_text", "point": "feature_text", "location": "location_text", "space": "space_text", "price": "price_text_block", "school": "school_text", "realtor": "realtor_text", "closing": "outro_text", } bad_phrases = [ "현장에서 확인했을 때 공간감과 구조가 안정적으로 느껴지는 매물입니다", "매물을 소개드립니다", "충분히 관심 있게 볼 수 있는 매물로 판단됩니다", ] for key, human_key in mapping.items(): current = clean_text(ai_sections.get(key)) replacement = clean_text(human_context.get(human_key)) if not replacement: continue if not current: ai_sections[key] = replacement continue if any(phrase in current for phrase in bad_phrases): ai_sections[key] = replacement continue if len(current) < 25: ai_sections[key] = replacement return ai_sections # --------------------------------------------------------------------- # SEO 해시태그 / 검색 노출 보강 # --------------------------------------------------------------------- def clean_seo_keyword(value): value = clean_text(value) value = value.replace("#", "") value = re.sub(r"[\s\t\r\n]+", "", value) value = re.sub(r"[^\w가-힣A-Za-z0-9]", "", value) return value.strip() def add_seo_keyword(items, value): keyword = clean_seo_keyword(value) if not keyword: return if len(keyword) < 2: return if keyword not in items: items.append(keyword) def split_realtor_seo_keywords(value): text = clean_text(value) if not text: return [] parts = re.split(r"[,，\n\r/|]+", text) result = [] for part in parts: keyword = clean_seo_keyword(part) if keyword and keyword not in result: result.append(keyword) return result def is_building_dong_keyword(value): value = clean_text(value) return bool(re.fullmatch(r"\d{1,4}동", value or "")) def extract_location_keywords_for_seo(detail): """ SEO용 지역명만 추출한다. 101동/206동 같은 건물 동 정보는 지역 키워드로 쓰지 않는다. """ merged = " ".join([ clean_text(detail.get("address")), clean_text(detail.get("road_address")), clean_text(detail.get("exposure_address")), clean_text(detail.get("address_info")), clean_text(detail.get("region_name")), clean_text(detail.get("city")), clean_text(detail.get("sigungu")), clean_text(detail.get("dong_name")), clean_text(detail.get("emd_name")), ]) # 주소 컬럼이 빈 경우 raw_json에서 exposureAddress를 보강한다. raw = parse_json_safely(detail.get("raw_json")) if raw: article_detail = raw.get("articleDetail") or {} article_addition = raw.get("articleAddition") or {} merged += " " + " ".join([ clean_text(article_detail.get("exposureAddress")), clean_text(article_detail.get("roadAddress")), clean_text(article_addition.get("lawUsage")), ]) items = [] for pattern in [ r"([가-힣A-Za-z0-9]+시)", r"([가-힣A-Za-z0-9]+군)", r"([가-힣A-Za-z0-9]+구)", r"([가-힣A-Za-z0-9]+동)", r"([가-힣A-Za-z0-9]+읍)", r"([가-힣A-Za-z0-9]+면)", ]: for match in re.findall(pattern, merged): if is_building_dong_keyword(match): continue add_seo_keyword(items, match) return items def extract_region_pair_for_seo(detail): """ 본문 문장용 지역 묶음 반환. 예: 부산진구 양정동 / 세종시 해밀동 """ locations = extract_location_keywords_for_seo(detail) gu = "" dong = "" city = "" for item in locations: if item.endswith("동") and not dong: dong = item elif item.endswith("구") and not gu: gu = item elif item.endswith("시") and not city: city = item if gu and dong: return f"{gu} {dong}" if city and dong: return f"{city} {dong}" if city and gu: return f"{city} {gu}" if locations: return " ".join(locations[:2]) return "" def build_seo_title(detail): article_name = normalize_title_duplicate_dong( clean_text( detail.get("article_name") or detail.get("building_name") or detail.get("complex_name") or "" ) ) base_name = remove_trailing_dong_for_sentence(article_name) or article_name trade_type = clean_text(detail.get("trade_type") or "") real_estate_type = clean_text(detail.get("real_estate_type") or detail.get("real_estate_type_name") or "") region = extract_region_pair_for_seo(detail) parts = [] if base_name: if trade_type: parts.append(f"{base_name} {trade_type}") else: parts.append(base_name) if region and real_estate_type and trade_type: parts.append(f"{region} {real_estate_type} {trade_type}") elif region and real_estate_type: parts.append(f"{region} {real_estate_type}") if not parts: return "매물 검색 안내" return " | ".join(parts[:2])[:120] def build_region_seo_paragraph(detail): article_name = normalize_title_duplicate_dong( clean_text( detail.get("article_name") or detail.get("building_name") or detail.get("complex_name") or "해당 매물" ) ) base_name = remove_trailing_dong_for_sentence(article_name) or article_name trade_type = clean_text(detail.get("trade_type") or "") real_estate_type = clean_text(detail.get("real_estate_type") or detail.get("real_estate_type_name") or "") price_text = clean_text(detail.get("price_text") or "") area_info = clean_text(detail.get("area_info") or "") office_name = clean_text(detail.get("office_name") or "중개사무소") region = extract_region_pair_for_seo(detail) subject = " ".join([x for x in [base_name, trade_type, real_estate_type] if x]).strip() or "해당 매물" lines = [] if region and real_estate_type: lines.append(f"{base_name}은 {region}에서 {real_estate_type}을 찾는 분들이 함께 살펴볼 만한 매물입니다.") else: lines.append(f"{subject} 정보를 찾고 계신 분들께 도움이 되는 매물 안내입니다.") if trade_type and real_estate_type: if region: lines.append(f"{region} {real_estate_type} {trade_type} 조건을 비교하고 계시다면 가격, 면적, 층수, 방향을 함께 확인해보시면 좋습니다.") else: lines.append(f"{real_estate_type} {trade_type} 조건을 비교하고 계시다면 가격, 면적, 층수, 방향을 함께 확인해보시면 좋습니다.") if price_text or area_info: detail_parts = [] if price_text: detail_parts.append(f"가격은 {price_text}") if area_info: detail_parts.append(f"면적은 {area_info}") if detail_parts: lines.append(" / ".join(detail_parts) + " 기준으로 안내되는 매물입니다.") if base_name and trade_type: lines.append(f"{base_name} {trade_type} 문의 전에는 현재 매물 가능 여부와 세부 조건을 다시 확인하는 것이 안전합니다.") lines.append(f"자세한 상담은 {office_name}를 통해 현재 기준의 최신 정보로 안내받으실 수 있습니다.") return "\n".join(lines) def build_seo_keywords(detail): keywords = [] article_name = normalize_title_duplicate_dong( clean_text( detail.get("article_name") or detail.get("building_name") or detail.get("complex_name") or "" ) ) article_name_no_dong = remove_trailing_dong_for_sentence(article_name) real_estate_type = clean_text( detail.get("real_estate_type") or detail.get("real_estate_type_name") or "" ) trade_type = clean_text(detail.get("trade_type") or "") office_name = clean_text(detail.get("office_name") or "") seo_keywords = clean_text(detail.get("seo_keywords") or "") location_keywords = extract_location_keywords_for_seo(detail) # 건물 동까지 붙은 단지명은 해시태그 가치가 낮아 기본 단지명 중심으로 사용한다. add_seo_keyword(keywords, article_name_no_dong) if article_name_no_dong and trade_type: add_seo_keyword(keywords, f"{article_name_no_dong}{trade_type}") if article_name_no_dong and real_estate_type: add_seo_keyword(keywords, f"{article_name_no_dong}{real_estate_type}") for loc in location_keywords: add_seo_keyword(keywords, loc) if real_estate_type: add_seo_keyword(keywords, f"{loc}{real_estate_type}") if trade_type: add_seo_keyword(keywords, f"{loc}{trade_type}") add_seo_keyword(keywords, f"{loc}부동산") if real_estate_type and trade_type: add_seo_keyword(keywords, f"{real_estate_type}{trade_type}") if real_estate_type: add_seo_keyword(keywords, real_estate_type) if trade_type: add_seo_keyword(keywords, trade_type) type_text = f"{article_name} {real_estate_type}".lower() if "apt" in type_text or "아파트" in type_text or re.search(r"(롯데캐슬|자이|래미안|푸르지오|힐스테이트|아이파크|더샵|e편한|단지)", type_text): if trade_type: add_seo_keyword(keywords, f"아파트{trade_type}") for keyword in ["아파트매매", "아파트전세", "아파트월세", "아파트추천"]: add_seo_keyword(keywords, keyword) if "오피스텔" in type_text: for keyword in ["오피스텔매매", "오피스텔전세", "오피스텔월세"]: add_seo_keyword(keywords, keyword) if "상가" in type_text: for keyword in ["상가매매", "상가임대", "상가월세"]: add_seo_keyword(keywords, keyword) if "토지" in type_text or "대지" in type_text or "임야" in type_text or real_estate_type in ["대", "전", "답"]: for keyword in ["토지매매", "대지매매", "토지추천"]: add_seo_keyword(keywords, keyword) if office_name: add_seo_keyword(keywords, office_name) add_seo_keyword(keywords, office_name.replace("공인중개사사무소", "공인중개사")) for keyword in split_realtor_seo_keywords(seo_keywords): add_seo_keyword(keywords, keyword) return keywords[:30] def build_seo_text_block(detail): return build_region_seo_paragraph(detail) def build_realtor_location_html(detail): office_name = clean_text(detail.get("office_name") or "") map_image_url = clean_text(detail.get("map_image_url") or "") if not map_image_url: return "" title = f"{office_name} 위치 안내" if office_name else "중개사무소 위치 안내" return f"""
{title}
{title}
방문 상담 전 매물 가능 여부와 상담 가능 시간을 먼저 확인해 주세요.
""" def insert_realtor_location_before_contact(draft_data, detail): if not draft_data: return draft_data location_html = build_realtor_location_html(detail) if not location_html.strip(): print("[REALTOR MAP SKIP] map_image_url empty") return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] updated = [] for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue if "realestate-realtor-map-image" in html: updated.append(f"{key}:already") continue anchor = "상담 및 중개사 안내" anchor_pos = html.find(anchor) if anchor_pos >= 0: # 상담 섹션 h2 바로 앞에 지도 삽입 h2_pos = html.rfind("= 0 else anchor_pos draft_data[key] = html[:insert_pos] + "\n" + location_html + "\n" + html[insert_pos:] updated.append(key) continue # 상담 섹션을 못 찾으면 최외곽 closing div 앞에 삽입 idx = html.rfind("") if idx >= 0: draft_data[key] = html[:idx] + "\n" + location_html + "\n" + html[idx:] else: draft_data[key] = html + "\n" + location_html updated.append(f"{key}:fallback") print("[REALTOR MAP INSERT DONE]", ",".join(updated)) return draft_data def build_seo_footer_html(detail): keywords = build_seo_keywords(detail) seo_text = build_seo_text_block(detail) hashtags = " ".join([f"#{keyword}" for keyword in keywords]) hashtag_html = "" if hashtags: hashtag_html = f"""
관련 키워드
{hashtags}
""" return f"""
{build_seo_title(detail)}
{seo_text}
{hashtag_html}
""" def append_seo_footer_to_draft_data(draft_data, detail): if not draft_data: return draft_data seo_html = build_seo_footer_html(detail) if not seo_html.strip(): return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] updated = [] for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue if "realestate-seo-footer" in html: updated.append(f"{key}:already") continue # 최외곽 closing div 앞에 넣는 것이 레이아웃 유지에 가장 안전하다. idx = html.rfind("") if idx >= 0: draft_data[key] = html[:idx] + "\n" + seo_html + "\n" + html[idx:] else: draft_data[key] = html + "\n" + seo_html updated.append(key) print("[SEO FOOTER APPEND DONE]", ",".join(updated), "keywords=", len(build_seo_keywords(detail))) return draft_data # --------------------------------------------------------------------- # 법정 표시광고 표준 표 # --------------------------------------------------------------------- def first_non_empty(*values): for value in values: text = clean_text(value) if text: return text return "" def normalize_unknown(value, default="확인 필요"): text = clean_text(value) if not text or text in ["-", "해당없음", "None", "null"]: return default return text def extract_total_floor_for_law(detail): value = clean_text(detail.get("floor_info") or detail.get("floor_info_display") or "") if not value: return "" m = re.search(r"총\s*(\d+)\s*층", value) if m: return f"총 {m.group(1)}층" m = re.search(r"/\s*(\d+)\s*층?", value) if m: return f"총 {m.group(1)}층" return value def extract_raw_value_for_law(detail, keys): for key in keys: value = clean_text(detail.get(key)) if value: return value raw = parse_json_safely(detail.get("raw_json")) if raw: found = recursive_find_first_safe(raw, keys) if found not in [None, "", 0, "0", "-", []]: return clean_text(found) return "" def format_ymd_for_law(value): text = clean_text(value) if not text: return "" if text == "NOW": return "즉시입주" digits = re.sub(r"[^0-9]", "", text) if len(digits) == 8: return f"{digits[:4]}년 {digits[4:6]}월 {digits[6:8]}일" if len(digits) == 6: return f"{digits[:4]}년 {digits[4:6]}월" return text def format_area_for_law(detail, article_space=None): current = clean_text(detail.get("area_info")) if current: return current article_space = article_space or {} supply = "" exclusive = "" if isinstance(article_space, dict): supply = clean_text(article_space.get("supplySpace")) exclusive = clean_text(article_space.get("exclusiveSpace")) if supply and exclusive: return f"공급 {supply}㎡ / 전용 {exclusive}㎡" if exclusive: return f"전용 {exclusive}㎡" if supply: return f"공급 {supply}㎡" return "" def format_parking_for_law(value, per_household=""): text = clean_text(value) per_household = clean_text(per_household) if not text: return "" if per_household: return f"총 {text}대 / 세대당 {per_household}대" return f"총 {text}대" def format_maintenance_for_law(raw, admin_info=None): text = clean_text(raw) if text: return text admin_info = admin_info or {} if isinstance(admin_info, dict) and admin_info: charge_code = clean_text(admin_info.get("chargeCodeType")) unable = admin_info.get("unableCheckDetails") or {} if isinstance(unable, dict) and unable: return "네이버 매물정보 기준 별도 확인 필요" if charge_code: return "네이버 매물정보 기준 별도 확인 필요" return "" def detect_law_target_type(detail): raw = parse_json_safely(detail.get("raw_json")) article_detail = raw.get("articleDetail") if isinstance(raw, dict) else {} article_addition = raw.get("articleAddition") if isinstance(raw, dict) else {} # STEP HOTFIX: # 법정표시사항 매물유형은 raw_json의 네이버 원천값을 최우선으로 신뢰한다. # 한화포레나대전월평공원2단지처럼 article_name에 '대전'이 들어가면 # land_words의 '전'에 걸려 토지로 오판되는 문제를 막는다. raw_type_name = clean_text( article_detail.get("realestateTypeName") if isinstance(article_detail, dict) else "" ) or clean_text( article_addition.get("articleRealEstateTypeName") if isinstance(article_addition, dict) else "" ) or clean_text( article_addition.get("realEstateTypeName") if isinstance(article_addition, dict) else "" ) raw_type_code = clean_text( article_detail.get("articleTypeCode") if isinstance(article_detail, dict) else "" ) or clean_text( article_addition.get("articleRealEstateTypeCode") if isinstance(article_addition, dict) else "" ) or clean_text( article_addition.get("realEstateTypeCode") if isinstance(article_addition, dict) else "" ) if raw_type_name in ["아파트", "오피스텔", "빌라", "연립", "다세대", "주택", "상가", "사무실"] or raw_type_code in ["A01"]: return "building" if raw_type_name in ["토지", "대", "전", "답", "임야", "대지", "잡종지", "공장용지", "창고용지"]: return "land" code_text = " ".join([ clean_text(detail.get("real_estate_type")), clean_text(detail.get("real_estate_type_name")), clean_text(detail.get("_raw_article_type_code")), clean_text(detail.get("_raw_realestate_type_code")), clean_text(article_detail.get("tradeBuildingTypeCode") if isinstance(article_detail, dict) else ""), clean_text(article_detail.get("realestateTypeCode") if isinstance(article_detail, dict) else ""), clean_text(article_addition.get("realEstateTypeCode") if isinstance(article_addition, dict) else ""), clean_text(article_addition.get("articleRealEstateTypeCode") if isinstance(article_addition, dict) else ""), ]).upper() name_text = " ".join([ clean_text(detail.get("real_estate_type")), clean_text(detail.get("real_estate_type_name")), clean_text(detail.get("article_name")), clean_text(article_detail.get("realestateTypeName") if isinstance(article_detail, dict) else ""), clean_text(article_addition.get("realEstateTypeName") if isinstance(article_addition, dict) else ""), clean_text(article_addition.get("articleRealEstateTypeName") if isinstance(article_addition, dict) else ""), ]) land_codes = ["TJ", "LAND", "TOJI"] land_words = ["토지", "대지", "임야", "전", "답", "잡종지", "공장용지", "창고용지"] if any(x in code_text for x in land_codes) or any(x in name_text for x in land_words): return "land" return "building" def format_land_area_for_law(detail, article_space=None, article_addition=None): article_space = article_space or {} article_addition = article_addition or {} ground = "" if isinstance(article_space, dict): ground = clean_text(article_space.get("groundSpace")) if ground: try: if float(ground) > 0: return f"대지면적 {float(ground):g}㎡" except Exception: return f"대지면적 {ground}㎡" area1 = "" if isinstance(article_addition, dict): area1 = clean_text(article_addition.get("area1")) if area1: return f"토지면적 {area1}㎡" current = clean_text(detail.get("area_info")) if current: return current return "" def extract_land_category_for_law(detail, article_detail=None, article_addition=None): article_detail = article_detail or {} article_addition = article_addition or {} candidates = [ detail.get("land_category"), detail.get("land_use"), article_detail.get("articleName") if isinstance(article_detail, dict) else "", article_detail.get("buildingTypeName") if isinstance(article_detail, dict) else "", article_addition.get("articleName") if isinstance(article_addition, dict) else "", extract_raw_value_for_law(detail, ["landCategoryName", "jimokName", "landUseName", "lndcgrCodeNm"]), ] for value in candidates: text = clean_text(value) if text and text not in ["-", "토지", "토지/임야"]: return text return "" def normalize_address_join_for_law(*values): parts = [] for value in values: text = clean_text(value) if not text or text in ["-", "확인 필요", "None", "null"]: continue if parts and text in parts[-1]: continue if any(text == p for p in parts): continue parts.append(text) joined = " ".join(parts) joined = re.sub(r"\s+", " ", joined).strip() joined = re.sub(r"(\S+동)\s+\1", r"\1", joined) return joined def find_address_text_recursive_for_law(obj): address_keys = { "address", "jibunaddress", "jibunaddressname", "roadaddress", "roadaddressname", "locationaddress", "detailaddress", "exposureaddress", "hscpaddress", "complexaddress", "fulladdress", "addr", "jibunaddr", "roadaddr", } candidates = [] def walk(value, path=""): if isinstance(value, dict): for key, child in value.items(): key_l = str(key or "").lower() next_path = f"{path}.{key_l}" if path else key_l if key_l in address_keys and isinstance(child, str): text = clean_text(child) if text: candidates.append((next_path, text)) walk(child, next_path) elif isinstance(value, list): for idx, child in enumerate(value): walk(child, f"{path}[{idx}]") walk(obj) def score(item): path, text = item s = 0 if any(x in text for x in ["시", "군", "구", "동", "읍", "면"]): s += 10 if re.search(r"\d", text): s += 20 if "jibun" in path: s += 8 if "road" in path: s += 5 if "address" in path: s += 3 return s candidates.sort(key=score, reverse=True) if candidates: return candidates[0][1] return "" def resolve_property_address_for_law(detail, article_detail=None, raw=None): article_detail = article_detail or {} raw = raw or {} complex_detail = raw.get("complexDetail") if isinstance(raw, dict) else {} complex_resolved = raw.get("complexResolvedAddress") if isinstance(raw, dict) else "" complex_address = "" if isinstance(complex_detail, dict) and complex_detail: complex_address = find_address_text_recursive_for_law(complex_detail) exposure = first_non_empty( detail.get("article_address"), detail.get("road_address"), detail.get("exposure_address"), article_detail.get("exposureAddress") if isinstance(article_detail, dict) else "", article_detail.get("roadAddress") if isinstance(article_detail, dict) else "", ) detail_addr = first_non_empty( detail.get("detail_address"), article_detail.get("detailAddress") if isinstance(article_detail, dict) else "", ) # 단지주소/complexResolved에 지번이 있으면 최우선 for candidate in [complex_resolved, complex_address]: candidate = clean_text(candidate) if candidate: return candidate # article API에 detailAddress가 있으면 exposureAddress와 결합 joined = normalize_address_join_for_law(exposure, detail_addr) if joined: return joined return "" def build_legal_disclosure_html(detail): """ 공인중개사법 표시광고 명시사항을 중개대상물 종류에 따라 분리해 표기한다. - 중개사무소 명시사항 5개는 공통 - 건축물/토지 유형별 표시 항목을 다르게 구성 - 데이터가 없으면 숨기지 않고 '확인 필요'로 표시 """ office_name = normalize_unknown(detail.get("office_name")) representative_name = normalize_unknown(detail.get("representative_name")) license_number = normalize_unknown(detail.get("license_number")) office_phone = normalize_unknown(detail.get("office_phone")) mobile_phone = clean_text(detail.get("mobile_phone")) raw = parse_json_safely(detail.get("raw_json")) article_detail = raw.get("articleDetail") if isinstance(raw, dict) else {} article_addition = raw.get("articleAddition") if isinstance(raw, dict) else {} article_space = raw.get("articleSpace") if isinstance(raw, dict) else {} article_facility = raw.get("articleFacility") if isinstance(raw, dict) else {} article_floor = raw.get("articleFloor") if isinstance(raw, dict) else {} article_realtor = raw.get("articleRealtor") if isinstance(raw, dict) else {} admin_info = raw.get("administrationCostInfo") if isinstance(raw, dict) else {} realtor_address = normalize_unknown( first_non_empty( detail.get("office_address"), detail.get("office_detail_address"), detail.get("realtor_address"), article_realtor.get("address") if isinstance(article_realtor, dict) else "", article_realtor.get("realtorAddress") if isinstance(article_realtor, dict) else "", ) ) contact = office_phone if mobile_phone and mobile_phone not in contact: contact = f"{office_phone} / {mobile_phone}" if office_phone != "확인 필요" else mobile_phone law_type = detect_law_target_type(detail) property_address = normalize_unknown( resolve_property_address_for_law( detail, article_detail=article_detail, raw=raw, ) ) area_info = normalize_unknown(format_area_for_law(detail, article_space)) price_text = normalize_unknown( first_non_empty( detail.get("price_text"), article_addition.get("dealOrWarrantPrc") if isinstance(article_addition, dict) else "", ) ) real_estate_type = normalize_unknown( first_non_empty( detail.get("real_estate_type"), detail.get("real_estate_type_name"), article_addition.get("articleRealEstateTypeName") if isinstance(article_addition, dict) else "", article_detail.get("realestateTypeName") if isinstance(article_detail, dict) else "", ) ) trade_type = normalize_unknown( first_non_empty( detail.get("trade_type"), article_detail.get("tradeTypeName") if isinstance(article_detail, dict) else "", article_addition.get("tradeTypeName") if isinstance(article_addition, dict) else "", ) ) total_floor = normalize_unknown( first_non_empty( extract_total_floor_for_law(detail), f"총 {article_floor.get('totalFloorCount')}층" if isinstance(article_floor, dict) and clean_text(article_floor.get("totalFloorCount")) else "", ) ) move_in = normalize_unknown( first_non_empty( detail.get("move_in_date"), detail.get("move_in_possible_date"), detail.get("move_in_type"), article_detail.get("moveInTypeName") if isinstance(article_detail, dict) else "", format_ymd_for_law(article_detail.get("moveInPossibleYmd") if isinstance(article_detail, dict) else ""), format_ymd_for_law(extract_raw_value_for_law(detail, ["moveInPossibleYmd"])), ) ) room_count = first_non_empty( detail.get("room_count"), detail.get("room_cnt"), article_detail.get("roomCount") if isinstance(article_detail, dict) else "", extract_raw_value_for_law(detail, ["roomCount", "roomCnt"]), ) bathroom_count = first_non_empty( detail.get("bathroom_count"), detail.get("bath_count"), article_detail.get("bathroomCount") if isinstance(article_detail, dict) else "", extract_raw_value_for_law(detail, ["bathroomCount", "bathCount", "bathroomCnt"]), ) room_bath = "확인 필요" if room_count or bathroom_count: room_bath = f"{room_count or '확인 필요'}개 / {bathroom_count or '확인 필요'}개" approval_date = normalize_unknown( first_non_empty( format_ymd_for_law(detail.get("use_approval_date")), format_ymd_for_law(detail.get("approval_date")), format_ymd_for_law(detail.get("building_approval_date")), format_ymd_for_law(article_detail.get("aptUseApproveYmd") if isinstance(article_detail, dict) else ""), format_ymd_for_law(extract_raw_value_for_law(detail, ["useApproveYmd", "useApprovalDate", "approvalDate", "buildingUseAprvYmd", "aptUseApproveYmd"])), ) ) parking = normalize_unknown( format_parking_for_law( first_non_empty( detail.get("parking_count"), detail.get("parking_available_count"), article_detail.get("parkingCount") if isinstance(article_detail, dict) else "", article_detail.get("aptParkingCount") if isinstance(article_detail, dict) else "", extract_raw_value_for_law(detail, ["parkingCount", "aptParkingCount", "parkingPossibleCount", "parkingTotalCount"]), ), first_non_empty( article_detail.get("parkingPerHouseholdCount") if isinstance(article_detail, dict) else "", article_detail.get("aptParkingCountPerHousehold") if isinstance(article_detail, dict) else "", ) ) ) maintenance = normalize_unknown( format_maintenance_for_law( first_non_empty( detail.get("maintenance_fee"), detail.get("management_fee"), detail.get("management_cost"), extract_raw_value_for_law(detail, ["managementCost", "manageCost", "maintenanceFee", "articleManageCost"]), ), admin_info, ) ) direction = normalize_unknown( first_non_empty( convert_direction(detail.get("direction") or detail.get("direction_code")), article_facility.get("directionTypeName") if isinstance(article_facility, dict) else "", extract_raw_value_for_law(detail, ["direction", "directionBaseTypeName", "directionTypeName"]), ) ) land_category = normalize_unknown( extract_land_category_for_law( detail, article_detail=article_detail, article_addition=article_addition, ) ) article_no = normalize_unknown(detail.get("article_no")) posting_time = datetime.now().strftime("%Y년 %m월 %d일 %H시 %M분") office_rows = [ ("명칭", office_name), ("소재지", realtor_address), ("연락처", contact), ("등록번호", license_number), ("성명(개업공인중개사)", representative_name), ] common_property_rows = [ ("매물번호", article_no), ("소재지", property_address), ("가격", price_text), ("거래형태", trade_type), ] if law_type == "land": property_section_title = "중개대상물 명시사항 - 토지" land_area_info = normalize_unknown( format_land_area_for_law( detail, article_space=article_space, article_addition=article_addition, ) ) property_rows = common_property_rows + [ ("면적", land_area_info), ("지목", land_category), ("포스팅 일시", posting_time), ] else: property_section_title = "중개대상물 명시사항 - 건축물" property_rows = common_property_rows + [ ("면적", area_info), ("중개대상물 종류", real_estate_type), ("총 층수", total_floor), ("입주가능일", move_in), ("방 수/욕실 수", room_bath), ("사용승인일", approval_date), ("주차대수", parking), ("관리비", maintenance), ("방향", direction), ("포스팅 일시", posting_time), ] def render_rows(rows): html = "" for label, value in rows: html += f""" {label} {value} """ return html return f"""
중개대상물 표시 · 광고 명시사항
공인중개사법 제18조의2 및 중개대상물의 표시·광고 명시사항 세부기준에 따라 아래와 같이 안내드립니다.
중개사무소 및 개업공인중개사 명시사항
{render_rows(office_rows)}
{property_section_title}
{render_rows(property_rows)}
본 포스팅의 매물 정보는 포스팅 당시 확인된 진성매물을 기준으로 작성되었습니다. 다만 확인 시점에 따라 거래 완료, 가격 변경, 조건 변경 등이 있을 수 있으므로 방문 또는 상담 전 현재 매물 존재 여부와 상세 조건을 반드시 다시 확인해 주시기 바랍니다.
""" def replace_legal_disclosure_block_in_html(html, detail): html = str(html or "") if not html: return html new_block = build_legal_disclosure_html(detail) if "realestate-legal-disclosure-table" in html: return html title = "중개대상물 표시 · 광고 안내" pos = html.find(title) if pos >= 0: # 기존 법정 안내 블록 시작점을 찾는다. start_candidates = [ html.rfind('
= 0), -1) # 기존 법정 안내 다음에 오는 하단 안내 블록 앞까지를 기존 법정 블록으로 본다. end_markers = [ '
= 0: end = idx break if start >= 0 and end > start: return html[:start] + new_block + "\n" + html[end:] # 기존 블록을 못 찾으면 상담 및 중개사 안내 앞에 삽입 anchor = "상담 및 중개사 안내" anchor_pos = html.find(anchor) if anchor_pos >= 0: start = html.rfind("= 0: return html[:start] + new_block + "\n" + html[start:] # 그래도 못 찾으면 마지막 div 앞에 삽입 idx = html.rfind("
") if idx >= 0: return html[:idx] + "\n" + new_block + "\n" + html[idx:] return html + "\n" + new_block def replace_legal_disclosure_in_draft_data(draft_data, detail): if not draft_data: return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] updated = [] for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue new_html = replace_legal_disclosure_block_in_html(html, detail) if new_html != html: draft_data[key] = new_html updated.append(key) print("[LEGAL DISCLOSURE REPLACED]", ",".join(updated)) return draft_data def ensure_v2_legal_disclosure_final_guard(draft_data, detail): """ STEP107-46: 발행 worker가 본문을 임의 생성/보강하지 않도록, 초안 생성 단계에서 realestate-legal-disclosure-table 존재를 최종 보장한다. """ if not draft_data: return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] report = {} for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue before_has = "realestate-legal-disclosure-table" in html before_len = len(html) if not before_has: html = replace_legal_disclosure_block_in_html(html, detail) after_has = "realestate-legal-disclosure-table" in html if html != str(draft_data.get(key) or ""): draft_data[key] = html report[key] = { "before_has": before_has, "after_has": after_has, "before_len": before_len, "after_len": len(html), } # plain_text는 발행용 HTML 원본은 아니지만 관리자 미리보기/검증에서 누락 오해가 없도록 최소 표식만 보강한다. plain = str(draft_data.get("plain_text") or "") if plain and "중개대상물 표시" not in plain: draft_data["plain_text"] = plain.rstrip() + "\n\n중개대상물 표시 · 광고 명시사항은 본문 하단 표를 기준으로 확인해 주세요." report["plain_text_marker_added"] = True else: report["plain_text_marker_added"] = False print("[V2 LEGAL DISCLOSURE FINAL GUARD]", report) return draft_data def verify_v2_draft_html_required_blocks(draft_data): """ STEP107-46: save 직전 필수 블록 진단 전용. 여기서는 수정하지 않는다. """ draft_data = draft_data or {} report = {} for key in ["draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html"]: html = str(draft_data.get(key) or "") if not html: continue report[key] = { "len": len(html), "has_legal_disclosure": "realestate-legal-disclosure-table" in html, "has_realtor_map": "realestate-realtor-map-image" in html, "has_seo_footer": "realestate-seo-footer" in html, "has_middle_banner": "middle_banner" in html or "middle-banner" in html, } print("[V2 DRAFT REQUIRED BLOCKS VERIFY]", report) return report # --------------------------------------------------------------------- # 대표이미지 자동 생성 / 저장 / 초안 삽입 # --------------------------------------------------------------------- def table_exists(conn, table_name): try: with conn.cursor() as cur: cur.execute("SHOW TABLES LIKE %s", (table_name,)) return cur.fetchone() is not None except Exception: return False 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 clean_header_value(value): value = clean_text(value) value = re.sub(r"\s+", " ", value) return value def normalize_title_duplicate_dong(value): """ 해밀1단지마스터힐스 106동 106동 같은 중복 동 표시를 제거한다. """ value = clean_header_value(value) if not value: return "" value = re.sub(r"(\b\d{1,4}동)\s+\1\b", r"\1", value) parts = value.split() cleaned = [] for part in parts: if cleaned and cleaned[-1] == part and re.search(r"\d+동$", part): continue cleaned.append(part) return " ".join(cleaned) def build_header_display_title(detail, complex_name, dong, ho, dong_ho): """ 대표이미지 API에 보낼 제목. API가 dong/ho를 별도 필드로 받기 때문에 title에는 중복 동을 최대한 제거한다. """ title = clean_header_value( detail.get("article_name") or detail.get("building_name") or complex_name or "부동산 매물 안내" ) title = normalize_title_duplicate_dong(title) if dong: title = re.sub(rf"\s+{re.escape(dong)}$", "", title).strip() if not title: title = complex_name or "부동산 매물 안내" return normalize_title_duplicate_dong(title) def is_dong_like_text(value): value = clean_header_value(value) return bool(re.fullmatch(r"\d{1,4}동", value or "")) def is_ho_like_text(value): value = clean_header_value(value) return bool(re.fullmatch(r"\d{1,5}호", value or "")) def parse_article_name_for_header(detail): article_name = clean_header_value( detail.get("article_name") or detail.get("article_title") or "" ) building_name = clean_header_value( detail.get("building_name") or detail.get("complex_name") or "" ) dong_from_field = clean_header_value( detail.get("dong") or detail.get("building_dong") or detail.get("article_dong") or "" ) complex_name = building_name dong = dong_from_field if article_name: m = re.search(r"(.+?)\s+(\d{1,4}동)\s*$", article_name) if m: if not complex_name or is_dong_like_text(complex_name): complex_name = clean_header_value(m.group(1)) if not dong: dong = clean_header_value(m.group(2)) else: if not complex_name: complex_name = article_name complex_name = re.sub(r"\s*[·|]\s*(매매|전세|월세|분양|임대)\s*[·|].*$", "", complex_name).strip() complex_name = re.sub(r"\s+(매매|전세|월세)\s+.*$", "", complex_name).strip() return { "article_name": article_name, "complex_name": complex_name, "dong": dong, } def choose_complex_name_for_header(detail, parsed_name): parsed_complex = clean_header_value(parsed_name.get("complex_name") or "") candidates = [ parsed_complex, detail.get("complex_name"), detail.get("article_complex_name"), detail.get("complex_title"), detail.get("building_name"), detail.get("article_name"), ] for item in candidates: name = clean_header_value(item) if not name: continue if is_dong_like_text(name) or is_ho_like_text(name): continue m = re.search(r"(.+?)\s+(\d{1,4}동)\s*$", name) if m: name = clean_header_value(m.group(1)) name = re.sub(r"\s*[·|]\s*(매매|전세|월세|분양|임대)\s*[·|].*$", "", name).strip() name = re.sub(r"\s+(매매|전세|월세)\s+.*$", "", name).strip() if name and not is_dong_like_text(name): return name return parsed_complex def detect_property_type_for_header(detail): """ 대표이미지/배너용 매물 유형 판별. LAND 계열은 아파트 키워드보다 먼저 판별한다. 특히 네이버 토지 매물은 real_estate_type 이 '대', '전', '답'처럼 짧게 내려올 수 있다. """ real_type = clean_text( detail.get("real_estate_type") or detail.get("real_estate_type_name") or detail.get("realEstateTypeName") or "" ).strip() land_exact = {"대", "전", "답", "임야", "잡종지", "공장용지", "창고용지", "토지", "대지", "도로", "구거"} if real_type in land_exact: return "land" text = " ".join([ clean_text(detail.get("real_estate_type")), clean_text(detail.get("real_estate_type_name")), clean_text(detail.get("article_name")), clean_text(detail.get("building_name")), clean_text(detail.get("complex_name")), clean_text(detail.get("article_title")), ]).lower() land_words = ["land", "토지", "대지", "임야", "잡종지", "공장용지", "창고용지", "계획관리지역", "자연녹지", "생산녹지"] if any(x in text for x in land_words): return "land" if "officetel" in text or "오피스텔" in text: return "officetel" if "villa" in text or "빌라" in text or "연립" in text or "다세대" in text: return "villa" if "store" in text or "상가" in text or "retail" in text or "사무실" in text: return "store" if "apt" in text or "아파트" in text: return "apt" if re.search(r"(단지|자이|래미안|푸르지오|힐스테이트|아이파크|수자인|더샵|e편한|롯데캐슬|\d{1,4}동)", text): return "apt" return "etc" def extract_dong_ho_for_header(detail): parsed = parse_article_name_for_header(detail) dong = clean_header_value( detail.get("dong") or detail.get("building_dong") or detail.get("article_dong") or parsed.get("dong") or "" ) ho = clean_header_value( detail.get("ho") or detail.get("unit_ho") or detail.get("article_ho") or "" ) dong_ho = clean_header_value( detail.get("dong_ho") or detail.get("dongho") or detail.get("dong_ho_text") or "" ) return dong, ho, dong_ho def call_header_image_api(payload): url = AI_REALESTATE_API_URL.rstrip("/") + "/generate-header-image" res = requests.post( url, json=payload, headers={ "Content-Type": "application/json", "X-API-KEY": AI_SHORTS_API_KEY, }, timeout=180, ) try: data = res.json() except Exception: raise Exception(f"header image api json parse failed: HTTP {res.status_code} / {res.text[:500]}") if res.status_code >= 400 or not data.get("ok"): raise Exception(data.get("error") or data.get("message") or f"header image api failed: HTTP {res.status_code}") return data def generate_header_image_for_article(detail, article_no): parsed_name = parse_article_name_for_header(detail) dong, ho, dong_ho = extract_dong_ho_for_header(detail) realtor_info = detail.get("realtor_info") or {} property_type = detect_property_type_for_header(detail) complex_name = choose_complex_name_for_header(detail, parsed_name) if not dong: possible_title = clean_header_value(detail.get("article_name") or "") m = re.search(r"(.+?)\s+(\d{1,4}동)\s*$", possible_title) if m: dong = clean_header_value(m.group(2)) complex_name = normalize_title_duplicate_dong(complex_name) dong = clean_header_value(dong) ho = clean_header_value(ho) dong_ho = clean_header_value(dong_ho) if property_type == "land": header_template_code = "LAND_PHOTO_LAYOUT" else: header_template_code = random.choice([ "APT_REAL_PHOTO_GLASS", "APT_BRIGHT_CARD", "APT_DARK_REC", "APT_PRICE_BAND", "APT_NOTE_CARD", "APT_RED_POSTER", "APT_CLEAN_WHITE", "APT_GREEN_LABEL", ]) payload = { "article_no": str(article_no), "realtor_id": detail.get("realtor_id") or 0, "template_code": header_template_code, "property_type": property_type, "real_estate_type": clean_header_value(detail.get("real_estate_type") or ""), "real_estate_type_name": clean_header_value(detail.get("real_estate_type_name") or detail.get("real_estate_type") or ""), "article_real_estate_type_name": clean_header_value(detail.get("_raw_article_real_estate_type_name") or ""), "trade_building_type_code": clean_header_value(detail.get("_raw_trade_building_type_code") or ""), "article_type_code": clean_header_value(detail.get("_raw_article_type_code") or ""), "realestate_type_code": clean_header_value(detail.get("_raw_realestate_type_code") or ""), "transaction_type": clean_header_value(detail.get("trade_type") or ""), "complex_name": complex_name, "exclusive_area": extract_exclusive_area_text( detail.get("area_info") or detail.get("exclusive_area") or "" ), # 아파트 대표이미지는 동 정보가 있으면 제목에 함께 노출한다. # 호수는 노출하지 않는다. "dong": dong if property_type == "apt" else "", "ho": "", "dong_ho": dong if property_type == "apt" else "", "title": build_header_display_title( detail=detail, complex_name=complex_name, dong=dong, ho=ho, dong_ho=dong_ho, ), "price": compact_price_text(detail.get("price_text") or ""), "region_name": clean_header_value( choose_v2_region_name_for_detail( detail.get("complex_resolved_address"), detail.get("jibun_address"), detail.get("address"), detail.get("road_address"), detail.get("region_name"), ) ), "realtor_name": clean_header_value( realtor_info.get("office_name") or detail.get("office_name") or "" ), # 정사각형 대표이미지에서는 전화번호를 빼고 중개사무소명만 노출한다. "phone": "", "background_url": detail.get("header_background_url") or detail.get("background_url") or "", "background_image_url": detail.get("header_background_url") or detail.get("background_url") or "", "width": 800, "height": 800, } print("[HEADER IMAGE PROPERTY_TYPE]", property_type) print("[HEADER IMAGE COMPLEX_NAME]", complex_name) print("[HEADER IMAGE DONG/HO]", dong, ho, dong_ho) print("[HEADER IMAGE API PAYLOAD]", json.dumps(payload, ensure_ascii=False, default=json_default)) result = call_header_image_api(payload) print("[HEADER IMAGE CREATED]", result.get("image_url")) print("[HEADER IMAGE BACKGROUND]", result.get("background_path")) print("[HEADER IMAGE THEME]", result.get("theme")) return result def save_mybox_asset(conn, article_no, realtor_id, header_image): if not header_image or not header_image.get("image_url"): return None if not table_exists(conn, "blog_mybox_assets"): print("[MYBOX ASSET SKIP] blog_mybox_assets table not exists") return None columns = table_columns(conn, "blog_mybox_assets") data = { "asset_type": "header_image", "asset_category": "blog_header", "article_no": str(article_no), "realtor_id": int(realtor_id or 0) if realtor_id else None, "title": "블로그 대표이미지", "local_path": header_image.get("image_path") or "", "file_url": header_image.get("image_url") or "", "file_name": header_image.get("filename") or "", "mime_type": "image/png", "file_size": 0, "status": "pending_upload", "is_representative": 1, "created_at": datetime.now(), "updated_at": datetime.now(), } filtered = { key: value for key, value in data.items() if key in columns } if not filtered: print("[MYBOX ASSET SKIP] no matching columns") return None keys = list(filtered.keys()) placeholders = ", ".join(["%s"] * len(keys)) column_sql = ", ".join(keys) values = [filtered[key] for key in keys] update_sql = ", ".join([ f"{key}=VALUES({key})" for key in keys if key not in ["id", "created_at"] ]) sql = f""" INSERT INTO blog_mybox_assets ({column_sql}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {update_sql} """ with conn.cursor() as cur: cur.execute(sql, values) asset_id = cur.lastrowid conn.commit() print(f"[MYBOX ASSET SAVED] article_no={article_no}, asset_id={asset_id}, url={header_image.get('image_url')}") return asset_id def save_header_image_to_article_images(conn, article_no, realtor_id, header_image): if not header_image or not header_image.get("image_url"): return None table_name = "blog_realestate_article_images" if not table_exists(conn, table_name): print("[HEADER ARTICLE IMAGE SKIP] blog_realestate_article_images table not exists") return None columns = table_columns(conn, table_name) try: category_parts = [] if "image_category" in columns: category_parts.append("image_category = 'header_image'") if "image_type" in columns: category_parts.append("image_type = 'header_image'") if category_parts: with conn.cursor() as cur: cur.execute( f"DELETE FROM {table_name} WHERE article_no = %s AND (" + " OR ".join(category_parts) + ")", (str(article_no),) ) conn.commit() except Exception as e: print("[HEADER ARTICLE IMAGE DELETE WARN]", str(e)) image_url = header_image.get("image_url") or "" image_path = header_image.get("image_path") or "" filename = header_image.get("filename") or "" width = int(header_image.get("width") or 0) height = int(header_image.get("height") or 0) data = { "article_no": str(article_no), "realtor_id": int(realtor_id or 0) if realtor_id else None, "image_url": image_url, "source_url": image_url, "original_url": image_url, "cdn_url": image_url, "naver_cdn_url": image_url, "local_path": image_path, "file_path": image_path, "image_path": image_path, "file_name": filename, "filename": filename, "image_type": "header_image", "image_category": "header_image", "asset_type": "header_image", "sort_order": -1000, "display_order": -1000, "is_representative": 1, "is_used": 1, "is_downloaded": 1, "width": width, "height": height, "image_width": width, "image_height": height, "mime_type": "image/png", "status": "active", "created_at": datetime.now(), "updated_at": datetime.now(), } filtered = { key: value for key, value in data.items() if key in columns } if not filtered: print("[HEADER ARTICLE IMAGE SKIP] no matching columns") return None keys = list(filtered.keys()) placeholders = ", ".join(["%s"] * len(keys)) column_sql = ", ".join(keys) values = [filtered[key] for key in keys] sql = f""" INSERT INTO {table_name} ({column_sql}) VALUES ({placeholders}) """ with conn.cursor() as cur: cur.execute(sql, values) image_id = cur.lastrowid conn.commit() print(f"[HEADER ARTICLE IMAGE SAVED] article_no={article_no}, image_id={image_id}, url={image_url}") return image_id def save_cdn_asset(conn, article_no, realtor_id, header_image, cdn_result, status="uploaded", error_message=""): """ Cafe24 CDN 업로드 결과를 blog_cdn_assets에 저장한다. 테이블이 없으면 건너뛴다. """ table_name = "blog_cdn_assets" if not table_exists(conn, table_name): print("[CDN ASSET SKIP] blog_cdn_assets table not exists") return None columns = table_columns(conn, table_name) local_path = "" file_name = "" if header_image: local_path = header_image.get("image_path") or "" file_name = header_image.get("filename") or "" cdn_result = cdn_result or {} data = { "article_no": str(article_no), "realtor_id": int(realtor_id or 0) if realtor_id else None, "asset_type": "header_image", "local_path": local_path, "remote_path": cdn_result.get("remote_path") or "", "public_path": cdn_result.get("public_path") or "", "cdn_url": cdn_result.get("cdn_url") or "", "file_name": file_name, "file_size": cdn_result.get("size") or 0, "mime_type": "image/png", "status": status, "error_message": error_message[:1000] if error_message else None, "uploaded_at": datetime.now() if status == "uploaded" else None, "created_at": datetime.now(), "updated_at": datetime.now(), } filtered = { key: value for key, value in data.items() if key in columns } if not filtered: print("[CDN ASSET SKIP] no matching columns") return None # 같은 매물 대표이미지는 1개만 유지 try: with conn.cursor() as cur: cur.execute( f""" DELETE FROM {table_name} WHERE article_no = %s AND asset_type = 'header_image' """, (str(article_no),) ) conn.commit() except Exception as e: print("[CDN ASSET DELETE WARN]", str(e)) keys = list(filtered.keys()) placeholders = ", ".join(["%s"] * len(keys)) column_sql = ", ".join(keys) values = [filtered[key] for key in keys] sql = f""" INSERT INTO {table_name} ({column_sql}) VALUES ({placeholders}) """ with conn.cursor() as cur: cur.execute(sql, values) asset_id = cur.lastrowid conn.commit() print( f"[CDN ASSET SAVED] article_no={article_no}, " f"asset_id={asset_id}, status={status}, url={cdn_result.get('cdn_url') or ''}" ) return asset_id def replace_header_image_url(header_image, cdn_result): """ header_image의 image_url을 Cafe24 CDN URL로 교체한다. 이후 prepend_header_image_to_draft_data()가 CDN URL을 본문 첫 이미지로 넣는다. """ if not header_image or not cdn_result: return header_image cdn_url = cdn_result.get("cdn_url") or "" if not cdn_url: return header_image header_image = dict(header_image) header_image["original_image_url"] = header_image.get("image_url", "") header_image["image_url"] = cdn_url header_image["cdn_url"] = cdn_url header_image["cdn_remote_path"] = cdn_result.get("remote_path", "") header_image["cdn_public_path"] = cdn_result.get("public_path", "") return header_image def build_header_image_html(header_image): if not header_image or not header_image.get("image_url"): return "" image_url = clean_text(header_image.get("image_url")) if not image_url: return "" return f"""
대표이미지
""" def prepend_header_image_to_draft_data(draft_data, header_image): """ 대표이미지를 draft_data 최상단에 삽입한다. step8 안정 구조 유지: build_blog_html -> header 생성 -> prepend -> save_draft """ if not draft_data: return draft_data if not header_image: print("[REP PREPEND SKIP] header_image empty") return draft_data image_url = clean_text( header_image.get("image_url") or header_image.get("cdn_url") or header_image.get("file_url") or header_image.get("url") or "" ) if not image_url: print("[REP PREPEND SKIP] image_url empty") return draft_data header_image = dict(header_image) header_image["image_url"] = image_url header_html = build_header_image_html(header_image) if not header_html: print("[REP PREPEND SKIP] header_html empty") return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] inserted = [] for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue if "realestate-representative-image" in html: inserted.append(f"{key}:already") continue draft_data[key] = header_html + "\n\n" + html inserted.append(key) draft_data["header_image_url"] = image_url draft_data["representative_image_url"] = image_url draft_data["thumbnail_url"] = image_url print("[REP PREPEND DONE]", image_url, "keys=", ",".join(inserted)) return draft_data # --------------------------------------------------------------------- # 중간배너 이미지 생성 / CDN 업로드 # --------------------------------------------------------------------- def call_middle_banner_image_api(payload): """ 중간배너 이미지 생성 API. shorts_ai_api.py에 /generate-middle-banner-image가 있으면 사용한다. 없거나 실패하면 None을 반환하여 기존 텍스트 fallback을 유지한다. """ url = AI_REALESTATE_API_URL.rstrip("/") + "/generate-middle-banner-image" try: res = requests.post( url, json=payload, headers={ "Content-Type": "application/json", "X-API-KEY": AI_SHORTS_API_KEY, }, timeout=180, ) try: data = res.json() except Exception: print("[MIDDLE BANNER API JSON ERROR]", res.status_code, res.text[:300]) return None if res.status_code >= 400 or not data.get("ok"): print("[MIDDLE BANNER API FAIL]", data.get("error") or data.get("message") or res.status_code) return None return data except Exception as e: print("[MIDDLE BANNER API ERROR]", str(e)) return None def generate_middle_banner_image_for_article(detail, article_no): """ 중간 CTA 배너용 600x200 이미지 생성. 생성 실패 시 None 반환 → 본문은 안전 텍스트 배너 fallback 사용. """ realtor_info = detail.get("realtor_info") or {} office_name = clean_header_value( realtor_info.get("office_name") or detail.get("office_name") or "중개사무소" ) phone = clean_header_value( realtor_info.get("office_phone") or realtor_info.get("mobile_phone") or detail.get("office_phone") or detail.get("mobile_phone") or detail.get("phone") or "" ) payload = { "article_no": str(article_no), "realtor_id": detail.get("realtor_id") or 0, "property_type": detect_property_type_for_header(detail), "real_estate_type": clean_header_value(detail.get("real_estate_type") or ""), "real_estate_type_name": clean_header_value(detail.get("real_estate_type_name") or detail.get("real_estate_type") or ""), "article_real_estate_type_name": clean_header_value(detail.get("_raw_article_real_estate_type_name") or ""), "trade_building_type_code": clean_header_value(detail.get("_raw_trade_building_type_code") or ""), "article_type_code": clean_header_value(detail.get("_raw_article_type_code") or ""), "realestate_type_code": clean_header_value(detail.get("_raw_realestate_type_code") or ""), "office_name": office_name, "realtor_name": office_name, "phone": phone, "button_text": "전체매물 보기", "width": 600, "height": 200, } print("[MIDDLE BANNER API PAYLOAD]", json.dumps(payload, ensure_ascii=False, default=json_default)) result = call_middle_banner_image_api(payload) if not result: print("[MIDDLE BANNER CREATED] skipped/fallback") return None print("[MIDDLE BANNER CREATED]", result.get("image_url")) print("[MIDDLE BANNER BACKGROUND]", result.get("background_path")) print("[MIDDLE BANNER THEME]", result.get("theme")) return result def upload_middle_banner_to_cafe24(conn, article_no, banner_image): """ 중간배너도 대표이미지와 동일한 Cafe24 CDN으로 업로드한다. 기존 upload_header_image_to_cafe24() 함수를 재사용하되 filename에 middle_banner가 들어가도록 한다. """ if not banner_image: return None local_path = banner_image.get("image_path") or "" filename = banner_image.get("filename") or "" if not local_path: print("[MIDDLE BANNER CDN SKIP] local_path empty") return None if not filename: filename = f"middle_banner_{article_no}.png" try: cdn_result = upload_header_image_to_cafe24( conn=conn, article_no=article_no, local_path=local_path, filename=filename, ) if cdn_result and cdn_result.get("cdn_url"): print("[MIDDLE BANNER CDN UPLOAD OK]", cdn_result.get("cdn_url")) return cdn_result print("[MIDDLE BANNER CDN UPLOAD EMPTY]") return None except Exception as e: print("[MIDDLE BANNER CDN UPLOAD ERROR]", str(e)) return None def replace_middle_banner_url(banner_image, cdn_result): if not banner_image: return banner_image banner_image = dict(banner_image) if cdn_result and cdn_result.get("cdn_url"): banner_image["original_image_url"] = banner_image.get("image_url", "") banner_image["image_url"] = cdn_result.get("cdn_url") banner_image["cdn_url"] = cdn_result.get("cdn_url") banner_image["cdn_remote_path"] = cdn_result.get("remote_path", "") banner_image["cdn_public_path"] = cdn_result.get("public_path", "") return banner_image def get_realtor_draft_buffer_target(row, default_target=45): """ 중개사별 초안 버퍼 목표값. blog_realtors.draft_buffer_target이 있으면 우선 사용하고, 없거나 0 이하이면 기본값을 사용한다. """ try: value = int((row or {}).get("draft_buffer_target") or 0) if value > 0: return value except Exception: pass try: default_target = int(default_target or 45) except Exception: default_target = 45 return max(1, default_target) def count_active_draft_buffer(conn, realtor_id): """ 중개사별 현재 사용 가능한 초안/발행대기 재고 수. - published/private_done 등 완료성 큐는 재고로 보지 않는다. - draft만 있고 queue가 없는 경우도 재고로 본다. """ if not table_exists(conn, "blog_article_drafts"): return 0 has_queue = table_exists(conn, "blog_publish_queue") if not has_queue: with conn.cursor() as cur: cur.execute(""" SELECT COUNT(*) AS cnt FROM blog_article_drafts d WHERE d.realtor_id = %s AND COALESCE(d.status, 'ready') IN ('ready', 'approved', 'pending') """, (int(realtor_id),)) row = cur.fetchone() or {} return int(row.get("cnt") or 0) with conn.cursor() as cur: cur.execute(""" SELECT COUNT(DISTINCT d.id) AS cnt FROM blog_article_drafts d LEFT JOIN blog_publish_queue q ON ( (q.draft_id IS NOT NULL AND q.draft_id = d.id) OR ( CONVERT(q.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci = CONVERT(d.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci ) ) WHERE d.realtor_id = %s AND COALESCE(d.status, 'ready') IN ('ready', 'approved', 'pending') AND ( q.id IS NULL OR COALESCE(q.queue_status, 'waiting') IN ( 'waiting', 'pending', 'ready', 'processing', 'failed', 'hold' ) ) """, (int(realtor_id),)) row = cur.fetchone() or {} return int(row.get("cnt") or 0) def is_valid_collected_article_for_draft(row): """ STEP89-B 초안 생성 방어 조건. 초안/대표이미지는 상세수집 완료 데이터로만 만든다. 후보 상태의 placeholder 매물은 여기서 제외한다. """ if not row: return False, "empty row" article_no = str(row.get("article_no") or "").strip() article_name = str(row.get("article_name") or "").strip() price_text = str(row.get("price_text") or "").strip() collect_status = str(row.get("collect_status") or "").strip() bad_titles = { "", "네이버 부동산 후보 매물", "네이버 부동산 현재 매물", "네이버 부동산 매물", "부동산 매물", "현재 매물", "추천 매물", } try: detail_collected = int(row.get("detail_collected") or 0) except Exception: detail_collected = 0 if not article_no: return False, "article_no empty" if collect_status and collect_status != "collected": return False, f"collect_status not collected: {collect_status}" if detail_collected != 1: return False, f"detail_collected not 1: {detail_collected}" if article_name in bad_titles: return False, f"bad article_name: {article_name}" if not price_text: return False, "price_text empty" return True, "ok" def fetch_realtors_for_draft_buffer(conn, realtor_id=None, limit_realtors=100): """ 초안 보충 대상 중개사 목록. detail_collected가 완료되고 이미지가 있는 매물을 가진 중개사만 대상으로 한다. """ columns = table_columns(conn, "blog_realtors") has_buffer_col = "draft_buffer_target" in columns select_buffer = "r.draft_buffer_target" if has_buffer_col else "NULL AS draft_buffer_target" sql = f""" SELECT r.id, r.office_name, {select_buffer} FROM blog_realtors r WHERE EXISTS ( SELECT 1 FROM blog_realtor_articles a WHERE a.realtor_id = r.id AND COALESCE(a.collect_status, '') = 'collected' AND COALESCE(a.detail_collected, 0) = 1 AND COALESCE(a.article_name, '') NOT IN ( '', '네이버 부동산 후보 매물', '네이버 부동산 현재 매물', '네이버 부동산 매물', '부동산 매물', '현재 매물', '추천 매물' ) AND COALESCE(a.price_text, '') <> '' AND EXISTS ( SELECT 1 FROM blog_realestate_article_images i WHERE CONVERT(i.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci = CONVERT(a.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci ) ) """ params = [] if realtor_id: sql += " AND r.id = %s " params.append(int(realtor_id)) sql += """ ORDER BY r.id ASC LIMIT %s """ params.append(int(limit_realtors or 100)) with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() def fetch_target_articles_for_realtor( conn, realtor_id, limit=10, force=False ): sql = """ SELECT a.*, r.office_name, r.representative_name, r.license_number, r.business_number, r.office_phone, r.mobile_phone, r.office_address, r.office_detail_address, r.address AS realtor_address, r.map_image_url, r.seo_keywords FROM blog_realtor_articles a LEFT JOIN blog_realtors r ON a.realtor_id = r.id WHERE a.realtor_id = %s AND COALESCE(a.collect_status, '') = 'collected' AND COALESCE(a.detail_collected, 0) = 1 AND COALESCE(a.article_name, '') NOT IN ( '', '네이버 부동산 후보 매물', '네이버 부동산 현재 매물', '네이버 부동산 매물', '부동산 매물', '현재 매물', '추천 매물' ) AND COALESCE(a.price_text, '') <> '' AND a.article_no IN ( SELECT DISTINCT article_no FROM blog_realestate_article_images ) """ params = [int(realtor_id)] if not force: sql += """ AND CONVERT(a.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci NOT IN ( SELECT CONVERT(article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci FROM blog_article_drafts ) """ if table_exists(conn, "blog_publish_queue"): sql += """ AND CONVERT(a.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci NOT IN ( SELECT CONVERT(article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci FROM blog_publish_queue WHERE article_no IS NOT NULL AND COALESCE(queue_status, '') IN ( 'waiting', 'pending', 'ready', 'processing', 'published', 'success', 'done' ) ) """ sql += """ ORDER BY COALESCE(a.first_posted_at, a.created_at) DESC, a.id DESC LIMIT %s """ params.append(int(limit)) with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() def fetch_target_articles( conn, realtor_id=None, article_no=None, limit=10, force=False, draft_buffer=45, replenish_count=5, limit_realtors=100, ): """ 초안 생성 대상 조회 STEP84. 기본 운영: - 중개사별 초안 재고가 draft_buffer_target 미만이면 부족분만 보충한다. - 한 번에 너무 많이 만들지 않도록 replenish_count로 중개사별 보충 상한을 둔다. - 전체 생성량은 --limit로 제한한다. 테스트/특수: - --article-no 또는 --force는 기존 방식에 가깝게 동작한다. """ if article_no: sql = """ SELECT a.*, r.office_name, r.representative_name, r.license_number, r.business_number, r.office_phone, r.mobile_phone, r.office_address, r.office_detail_address, r.address AS realtor_address, r.map_image_url, r.seo_keywords FROM blog_realtor_articles a LEFT JOIN blog_realtors r ON a.realtor_id = r.id WHERE a.article_no = %s AND COALESCE(a.collect_status, '') = 'collected' AND COALESCE(a.detail_collected, 0) = 1 AND COALESCE(a.article_name, '') NOT IN ( '', '네이버 부동산 후보 매물', '네이버 부동산 현재 매물', '네이버 부동산 매물', '부동산 매물', '현재 매물', '추천 매물' ) AND COALESCE(a.price_text, '') <> '' AND a.article_no IN ( SELECT DISTINCT article_no FROM blog_realestate_article_images ) LIMIT %s """ params = [str(article_no), int(limit)] with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() if force: # 강제 재생성은 지정 중개사 또는 전체에서 limit만큼 기존 방식으로 처리 sql = """ SELECT a.*, r.office_name, r.representative_name, r.license_number, r.business_number, r.office_phone, r.mobile_phone, r.office_address, r.office_detail_address, r.address AS realtor_address, r.map_image_url, r.seo_keywords FROM blog_realtor_articles a LEFT JOIN blog_realtors r ON a.realtor_id = r.id WHERE COALESCE(a.collect_status, '') = 'collected' AND COALESCE(a.detail_collected, 0) = 1 AND COALESCE(a.article_name, '') NOT IN ( '', '네이버 부동산 후보 매물', '네이버 부동산 현재 매물', '네이버 부동산 매물', '부동산 매물', '현재 매물', '추천 매물' ) AND COALESCE(a.price_text, '') <> '' AND a.article_no IN ( SELECT DISTINCT article_no FROM blog_realestate_article_images ) """ params = [] if realtor_id: sql += " AND a.realtor_id = %s " params.append(int(realtor_id)) sql += """ ORDER BY COALESCE(a.first_posted_at, a.created_at) DESC, a.id DESC LIMIT %s """ params.append(int(limit)) with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() total_limit = int(limit or 0) if total_limit <= 0: total_limit = 1 try: replenish_count = int(replenish_count or 0) except Exception: replenish_count = 5 if replenish_count <= 0: replenish_count = total_limit articles = [] realtors = fetch_realtors_for_draft_buffer( conn=conn, realtor_id=realtor_id, limit_realtors=limit_realtors, ) print(f"[DRAFT BUFFER REALTORS] {len(realtors)}") for realtor in realtors: rid = int(realtor.get("id") or 0) if not rid: continue target = get_realtor_draft_buffer_target( realtor, default_target=draft_buffer, ) current = count_active_draft_buffer(conn, rid) need = max(0, int(target) - int(current)) if need <= 0: print( f"[DRAFT BUFFER OK] realtor_id={rid}, " f"current={current}, target={target}" ) continue remaining_global = total_limit - len(articles) if remaining_global <= 0: break take = min(need, replenish_count, remaining_global) print( f"[DRAFT BUFFER NEED] realtor_id={rid}, " f"current={current}, target={target}, need={need}, take={take}" ) rows = fetch_target_articles_for_realtor( conn=conn, realtor_id=rid, limit=take, force=False, ) print( f"[DRAFT BUFFER TARGETS] realtor_id={rid}, " f"rows={len(rows)}" ) articles.extend(rows) if len(articles) >= total_limit: break return articles # --------------------------------------------------------- # STEP111-13 HEE Layout Recommendation - Log Only # --------------------------------------------------------- def _hee_clean_text_for_layout(value): import re from html import unescape value = "" if value is None else str(value) value = unescape(value) value = re.sub(r"<[^>]+>", " ", value) value = re.sub(r"\s+", " ", value) return value.strip() def _hee_detect_property_class_for_layout(detail): raw = parse_json_safely(detail.get("raw_json")) article_detail = raw.get("articleDetail") if isinstance(raw, dict) else {} article_addition = raw.get("articleAddition") if isinstance(raw, dict) else {} type_name = _hee_clean_text_for_layout( detail.get("real_estate_type") or detail.get("real_estate_type_name") or (article_detail.get("realestateTypeName") if isinstance(article_detail, dict) else "") or (article_addition.get("articleRealEstateTypeName") if isinstance(article_addition, dict) else "") or (article_addition.get("realEstateTypeName") if isinstance(article_addition, dict) else "") ) type_code = _hee_clean_text_for_layout( (article_detail.get("articleTypeCode") if isinstance(article_detail, dict) else "") or (article_detail.get("realestateTypeCode") if isinstance(article_detail, dict) else "") or (article_addition.get("articleRealEstateTypeCode") if isinstance(article_addition, dict) else "") or (article_addition.get("realEstateTypeCode") if isinstance(article_addition, dict) else "") ).upper() if type_name in ["아파트"] or type_code in ["A01", "APT"]: return "apartment" if type_name in ["오피스텔"]: return "officetel" if type_name in ["빌라", "연립", "다세대", "주택"]: return "villa_house" if type_name in ["상가", "상가점포"]: return "store" if type_name in ["사무실"]: return "office" if type_name in ["공장", "창고"]: return "factory_warehouse" if type_name in ["토지", "대", "전", "답", "임야", "대지", "잡종지", "공장용지", "창고용지"] or type_code in ["TJ", "LAND", "TOJI", "E03"]: return "land" if type_name in ["분양권", "입주권"]: return "presale_right" return "unknown" def _hee_layout_mode_from_detail(detail, images, ai_sections=None): property_class = _hee_detect_property_class_for_layout(detail) try: image_count = len(images or []) except Exception: image_count = 0 broker_desc = _hee_clean_text_for_layout( detail.get("article_feature_desc") or detail.get("article_desc") or detail.get("article_description") or "" ) broker_desc_len = len(broker_desc) if broker_desc_len <= 0 and ai_sections: broker_desc_len = len(_hee_clean_text_for_layout((ai_sections or {}).get("intro") or "")) if image_count <= 0 and broker_desc_len < 20: base = "minimal" elif image_count <= 1 or broker_desc_len < 40: base = "compact" elif image_count <= 4 or broker_desc_len < 80: base = "simple" elif image_count <= 10: base = "balanced" else: base = "rich" if property_class == "land": layout_mode = f"land_{base}" elif property_class == "factory_warehouse": layout_mode = f"factory_{base}" elif property_class in ["store", "office"]: layout_mode = f"business_{base}" elif property_class in ["apartment", "officetel", "villa_house", "presale_right"]: layout_mode = f"living_{base}" else: layout_mode = f"generic_{base}" return { "property_class": property_class, "image_count": image_count, "broker_desc_len": broker_desc_len, "layout_mode": layout_mode, } def _hee_recommended_layout_type(layout_mode, realtor_id=None, article_no=None): import random candidates_map = { "living_rich": ["storytelling", "magazine_style", "local_life", "family_recommend", "premium_report", "gallery_story"], "living_balanced": ["story_first", "location_focus", "recommendation_focus", "storytelling", "local_life", "photo_grid"], "living_simple": ["summary_first", "minimal_modern", "short_review_style", "checklist_style"], "living_compact": ["summary_first", "minimal_modern", "checklist_style"], "living_minimal": ["summary_first", "minimal_modern"], "business_rich": ["office_focus", "location_focus", "broker_expert", "premium_report"], "business_balanced": ["office_focus", "location_focus", "summary_first", "checklist_style"], "business_simple": ["office_focus", "summary_first", "minimal_modern"], "business_compact": ["summary_first", "minimal_modern"], "business_minimal": ["summary_first", "minimal_modern"], "factory_rich": ["office_focus", "broker_expert", "checklist_style", "summary_first"], "factory_balanced": ["office_focus", "summary_first", "checklist_style"], "factory_simple": ["summary_first", "checklist_style", "minimal_modern"], "factory_compact": ["summary_first", "minimal_modern"], "factory_minimal": ["summary_first", "minimal_modern"], "land_rich": ["land_focus"], "land_balanced": ["land_focus"], "land_simple": ["land_focus"], "land_compact": ["land_focus"], "land_minimal": ["land_focus"], "generic_rich": ["story_first", "summary_first", "checklist_style"], "generic_balanced": ["summary_first", "checklist_style", "minimal_modern"], "generic_simple": ["summary_first", "minimal_modern"], "generic_compact": ["summary_first", "minimal_modern"], "generic_minimal": ["summary_first", "minimal_modern"], } candidates = candidates_map.get(layout_mode) or candidates_map.get("generic_simple") seed_text = f"{realtor_id or ''}:{article_no or ''}:{layout_mode}" rnd = random.Random(seed_text) return rnd.choice(candidates), candidates def hee_log_layout_recommendation(detail, images, ai_sections, current_layout_type): try: info = _hee_layout_mode_from_detail(detail, images, ai_sections) recommended, candidates = _hee_recommended_layout_type( info.get("layout_mode"), realtor_id=detail.get("realtor_id"), article_no=detail.get("article_no"), ) print( "[HEE LAYOUT RECOMMEND] " f"property_class={info.get('property_class')}, " f"image_count={info.get('image_count')}, " f"broker_desc_len={info.get('broker_desc_len')}, " f"layout_mode={info.get('layout_mode')}, " f"recommended={recommended}, " f"current={current_layout_type}, " f"apply=False" ) except Exception as e: print("[HEE LAYOUT RECOMMEND ERROR]", str(e)) def fetch_article_images(conn, article_no): with conn.cursor() as cur: cur.execute(""" SELECT * FROM blog_realestate_article_images WHERE article_no = %s ORDER BY CASE WHEN image_category = 'inside' THEN 1 WHEN image_category = 'article' THEN 2 WHEN image_category = 'building' THEN 3 WHEN image_category = 'floorplan' THEN 4 WHEN image_category = 'birdview' THEN 5 WHEN image_category = 'realtor' THEN 9 ELSE 6 END, sort_order ASC, id ASC """, (article_no,)) return cur.fetchall() def fetch_article_prices(conn, article_no): with conn.cursor() as cur: cur.execute(""" SELECT * FROM blog_realestate_article_prices WHERE article_no = %s ORDER BY id ASC LIMIT 20 """, (article_no,)) return cur.fetchall() def fetch_article_schools(conn, article_no): with conn.cursor() as cur: cur.execute(""" SELECT * FROM blog_realestate_article_schools WHERE article_no = %s ORDER BY id ASC LIMIT 20 """, (article_no,)) return cur.fetchall() def fetch_existing_draft_source_detail(conn, article_no): """ STEP107-04: 기존 draft의 source_json.detail 안에 이미 보정된 지번 주소가 있는 경우가 있다. 예: source_json.detail.article_address = 세종시 해밀동 649 V2 재생성 시 blog_realtor_articles에는 동까지만 있을 수 있으므로 이를 병합한다. """ if not article_no: return {} if not table_exists(conn, "blog_article_drafts"): return {} try: with conn.cursor() as cur: cur.execute(""" SELECT source_json FROM blog_article_drafts WHERE article_no = %s ORDER BY id DESC LIMIT 1 """, (str(article_no),)) row = cur.fetchone() or {} source_json = parse_json_safely(row.get("source_json")) detail = source_json.get("detail") if isinstance(source_json, dict) else {} if isinstance(detail, dict): return detail except Exception as e: print("[V2 EXISTING DRAFT SOURCE MERGE WARN]", article_no, str(e)) return {} def merge_existing_draft_source_address(detail_row, existing_detail): """ 기존 draft source_json.detail에서 매물 주소/학교/complex 정보를 보강한다. 단, 중개사 주소 계열은 사용하지 않는다. """ detail_row = dict(detail_row or {}) existing_detail = dict(existing_detail or {}) def has_number_address(value): value = clean_text(value) return bool(value and re.search(r"\d", value) and re.search(r"(시|군|구|동|리)", value)) address_keys = [ "complex_resolved_address", "article_address", "exposure_address", "address", "jibun_address", "road_address", ] for key in address_keys: old = clean_text(detail_row.get(key)) new = normalize_v2_address_text(existing_detail.get(key)) # 숫자 지번이 있는 기존 detail 주소를 우선 병합 if has_number_address(new) and not has_number_address(old): detail_row[key] = new print(f"[V2 EXISTING DRAFT ADDRESS MERGE] {key}={new}") # detail_row.address가 동까지만 있으면 article_address/complex_resolved_address의 지번 주소로 교체 for key in ["complex_resolved_address", "article_address", "exposure_address", "address"]: new = normalize_v2_address_text(detail_row.get(key)) if has_number_address(new): detail_row["address"] = new detail_row["article_address"] = new detail_row["exposure_address"] = new detail_row["complex_resolved_address"] = new break # 학교/단지 보조 정보 병합 for key in [ "school_name", "school_names", "complex_school_name", "complex_school_names", "complex_no", "hscp_no", "complex_name" ]: if not clean_text(detail_row.get(key)) and clean_text(existing_detail.get(key)): detail_row[key] = existing_detail.get(key) # source_json.detail 전체를 보존해 v2 resolver listing fallback 후보로 쓰게 한다. source_json = parse_json_safely(detail_row.get("source_json")) if not isinstance(source_json, dict): source_json = {} source_detail = source_json.get("detail") if not isinstance(source_detail, dict): source_detail = {} merged_detail = dict(existing_detail) merged_detail.update(source_detail) for key in address_keys: if clean_text(detail_row.get(key)): merged_detail[key] = detail_row.get(key) source_json["detail"] = merged_detail try: detail_row["source_json"] = json.dumps(source_json, ensure_ascii=False, default=json_default) except Exception: pass return detail_row def save_draft( conn, realtor_id, article_no, draft_data, ai_sections, source_json, ): draft_title = draft_data.get("draft_title", "")[:255] layout_type = draft_data.get("layout_type", "default") draft_html = draft_data.get("draft_html", "") clipboard_html = draft_data.get("clipboard_html", "") plain_text = draft_data.get("plain_text", "") with conn.cursor() as cur: cur.execute(""" INSERT INTO blog_article_drafts ( realtor_id, article_no, draft_title, layout_type, draft_html, clipboard_html, plain_text, ai_summary, ai_sections, source_json, status, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'ready', NOW(), NOW() ) ON DUPLICATE KEY UPDATE draft_title = VALUES(draft_title), layout_type = VALUES(layout_type), draft_html = VALUES(draft_html), clipboard_html = VALUES(clipboard_html), plain_text = VALUES(plain_text), ai_summary = VALUES(ai_summary), ai_sections = VALUES(ai_sections), source_json = VALUES(source_json), status = 'ready', updated_at = NOW() """, ( realtor_id, article_no, draft_title, layout_type, draft_html, clipboard_html, plain_text, ai_sections.get("intro", "")[:1000], json.dumps(ai_sections, ensure_ascii=False, default=json_default), json.dumps(source_json, ensure_ascii=False, default=json_default), )) conn.commit() # 대표이미지/V2 메타 컬럼이 존재하는 경우 별도 업데이트 try: columns = table_columns(conn, "blog_article_drafts") update_data = {} if "header_image_url" in columns: update_data["header_image_url"] = draft_data.get("header_image_url", "") if "representative_image_url" in columns: update_data["representative_image_url"] = draft_data.get("representative_image_url", "") if "pipeline_version" in columns: update_data["pipeline_version"] = V2_PIPELINE_VERSION if "source" in columns: update_data["source"] = V2_PIPELINE_VERSION if "draft_generator" in columns: update_data["draft_generator"] = V2_DRAFT_GENERATOR if "pipeline_version" in columns: update_data["pipeline_version"] = V2_PIPELINE_VERSION if "source" in columns: update_data["source"] = V2_PIPELINE_VERSION if "worker_version" in columns: update_data["worker_version"] = V2_PIPELINE_VERSION if "publish_worker" in columns: update_data["publish_worker"] = "v2_publish_worker" if "updated_at" in columns: update_data["updated_at"] = datetime.now() if update_data: sets = ", ".join([f"{key} = %s" for key in update_data.keys()]) values = list(update_data.values()) values.append(article_no) with conn.cursor() as cur: cur.execute( f"UPDATE blog_article_drafts SET {sets} WHERE article_no = %s", values ) conn.commit() print("[V2 DRAFT META UPDATE]", article_no, ",".join(update_data.keys())) except Exception as e: print("[DRAFT V2 META UPDATE WARN]", str(e)) def create_publish_queue(conn, realtor_id, article_no, draft_id): """ blog_article_drafts 생성 후 발행 대기 큐를 만든다. 이미 같은 draft_id 또는 article_no의 pending/processing/published 큐가 있으면 중복 생성하지 않는다. failed/hold 큐가 있으면 pending으로 복구한다. """ if not table_exists(conn, "blog_publish_queue"): print("[PUBLISH QUEUE SKIP] blog_publish_queue table not exists") return None columns = table_columns(conn, "blog_publish_queue") # 1) 기존 큐 확인 where_parts = [] params = [] if "draft_id" in columns: where_parts.append("draft_id = %s") params.append(int(draft_id)) if "article_no" in columns: where_parts.append("article_no = %s") params.append(str(article_no)) existing = None if where_parts: sql = f""" SELECT * FROM blog_publish_queue WHERE ({' OR '.join(where_parts)}) ORDER BY id DESC LIMIT 1 """ with conn.cursor() as cur: cur.execute(sql, params) existing = cur.fetchone() if existing: existing_id = existing.get("id") existing_status = str(existing.get("queue_status") or existing.get("status") or "").strip() if existing_status in ["waiting", "pending", "ready", "processing", "published"]: print( f"[PUBLISH QUEUE EXISTS] queue_id={existing_id}, " f"status={existing_status}, article_no={article_no}" ) return existing_id # hold/failed 등은 테스트와 재생성 편의를 위해 pending 복구 update_data = {} if "queue_status" in columns: update_data["queue_status"] = "pending" elif "status" in columns: update_data["status"] = "pending" if "error_message" in columns: update_data["error_message"] = None if "started_at" in columns: update_data["started_at"] = None if "finished_at" in columns: update_data["finished_at"] = None if "updated_at" in columns: update_data["updated_at"] = datetime.now() if update_data: sets = ", ".join([f"{key} = %s" for key in update_data.keys()]) values = list(update_data.values()) values.append(existing_id) with conn.cursor() as cur: cur.execute( f"UPDATE blog_publish_queue SET {sets} WHERE id = %s", values ) conn.commit() print( f"[PUBLISH QUEUE RESET] queue_id={existing_id}, " f"status={existing_status} -> pending, article_no={article_no}" ) return existing_id # 2) 신규 큐 생성 data = { "realtor_id": int(realtor_id or 0), "article_no": str(article_no), "draft_id": int(draft_id), "queue_status": "pending", "status": "pending", "priority": 0, "error_message": None, "pipeline_version": V2_PIPELINE_VERSION, "source": V2_PIPELINE_VERSION, "worker_version": V2_PIPELINE_VERSION, "publish_worker": "v2_publish_worker", "created_at": datetime.now(), "updated_at": datetime.now(), } filtered = { key: value for key, value in data.items() if key in columns } if not filtered: print("[PUBLISH QUEUE SKIP] no matching columns") return None keys = list(filtered.keys()) column_sql = ", ".join(keys) placeholders = ", ".join(["%s"] * len(keys)) values = [filtered[key] for key in keys] sql = f""" INSERT INTO blog_publish_queue ({column_sql}) VALUES ({placeholders}) """ with conn.cursor() as cur: cur.execute(sql, values) queue_id = cur.lastrowid conn.commit() print( f"[PUBLISH QUEUE CREATED] queue_id={queue_id}, " f"draft_id={draft_id}, realtor_id={realtor_id}, article_no={article_no}" ) return queue_id # --------------------------------------------------------- # STEP107-01: V2 배정학교 HTML / 본문 삽입 헬퍼 # --------------------------------------------------------- def split_v2_school_names(school_names): school_names = clean_text(school_names) if not school_names: return [] result = [] for item in re.split(r"[,/·|]", school_names): item = clean_text(item) if item and item not in result: result.append(item) return result def build_v2_school_table_html_from_detail(detail): """ V2 Resolver에서 확보한 school_names가 있을 때 본문 삽입용 HTML table을 생성한다. 네이버 검색 단지카드의 배정학교 정보만 사용한다. """ school_names = clean_text( detail.get("school_names") or detail.get("v2_school_names") or detail.get("complex_school_names") or detail.get("school_name") or "" ) schools = split_v2_school_names(school_names) # STEP107-03: # V2 Resolver에서 school_names가 비어도 fetch_article_schools() 결과가 있으면 사용한다. # 로그상 [SCHOOLS] 1인데 school_html이 비는 문제 보정. if not schools: for row in (detail.get("schools") or []): if not isinstance(row, dict): continue name = clean_text( row.get("school_name") or row.get("name") or row.get("schoolName") or row.get("assigned_school") or row.get("배정학교") ) if name and name not in schools: schools.append(name) if not schools: return "" school_text = ", ".join(schools) return f"""
배정학교 정보
배정학교 {school_text}
확인 기준 네이버 검색 단지정보
""".strip() def insert_v2_school_html_to_draft_data(draft_data, detail): """ school_html을 본문에 삽입한다. 위치는 법정표 위 / 상담 안내 위 / 마지막 div 앞 순서로 시도한다. """ if not draft_data: return draft_data school_html = clean_text( detail.get("school_html") or detail.get("v2_school_html") or "" ) if not school_html: school_html = build_v2_school_table_html_from_detail(detail) if not school_html: print("[V2 SCHOOL HTML SKIP] school_names empty") return draft_data html_keys = [ "draft_html", "clipboard_html", "preview_html", "content_html", "body_html", "html", ] updated = [] for key in html_keys: html = str(draft_data.get(key) or "") if not html: continue if "v2-school-info-table" in html: updated.append(f"{key}:already") continue insert_pos = -1 # 법정 표시광고 표 위에 삽입하면 소재지/생활 정보 흐름상 자연스럽다. anchors = [ "중개대상물 표시 · 광고", "중개대상물 표시", "realestate-legal-disclosure-table", "상담 및 중개사 안내", "realestate-realtor-map-image", "realestate-seo-footer", ] for anchor in anchors: pos = html.find(anchor) if pos >= 0: h_pos = html.rfind("= 0 else div_pos if div_pos >= 0 else pos break if insert_pos >= 0: draft_data[key] = html[:insert_pos] + "\n" + school_html + "\n" + html[insert_pos:] updated.append(key) continue idx = html.rfind("
") if idx >= 0: draft_data[key] = html[:idx] + "\n" + school_html + "\n" + html[idx:] updated.append(f"{key}:fallback") else: draft_data[key] = html + "\n" + school_html updated.append(f"{key}:append") print("[V2 SCHOOL HTML INSERT DONE]", ",".join(updated)) return draft_data def _v2_pick_resolved_payload(address_result): """ resolve_and_cache_complex_address_v2 결과에서 detail에 주입할 주소/학교 값을 정리한다. """ 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 "" ) return { "resolved_address": normalize_v2_address_text(resolved_address), "road_address": normalize_v2_address_text(resolved.get("road_address") or cache.get("road_address") or ""), "jibun_address": normalize_v2_address_text(resolved.get("jibun_address") or cache.get("jibun_address") or resolved_address), "complex_no": clean_text(resolved.get("complex_no") or cache.get("complex_no") or ""), "school_name": clean_text(resolved.get("school_name") or cache.get("school_name") or ""), "school_names": clean_text(resolved.get("school_names") or cache.get("school_names") or resolved.get("school_name") or cache.get("school_name") or ""), "source": clean_text(resolved.get("source") or cache.get("source") or ""), "region_name": choose_v2_region_name_for_detail( resolved.get("jibun_address") or cache.get("jibun_address") or resolved_address, resolved.get("resolved_address") or cache.get("resolved_address"), address_result.get("region_name") or cache.get("region_name") or "", ), "complex_name": clean_text(address_result.get("complex_name") or cache.get("complex_name") or resolved.get("complex_name") or ""), "from_cache": bool(address_result.get("from_cache")), } # --------------------------------------------------------- # STEP102-13: 초안 생성 전 소재지 지번 주소 강제 보강 # --------------------------------------------------------- def _extract_complex_address_keys_for_draft(detail): """ raw_json / detail_row에서 주소 보강에 필요한 값을 최대한 추출한다. - hscpNo/complexNo가 있으면 캐시 조회는 complex_no 우선 - aptName/building_name/article_name을 단지명 후보로 사용 - exposureAddress를 region_name으로 사용 """ raw = parse_json_safely(detail.get("raw_json")) article_detail = raw.get("articleDetail") if isinstance(raw, dict) else {} article_addition = raw.get("articleAddition") if isinstance(raw, dict) else {} hscp_no = first_non_empty( detail.get("complex_no"), detail.get("hscp_no"), article_detail.get("hscpNo") if isinstance(article_detail, dict) else "", article_detail.get("complexNo") if isinstance(article_detail, dict) else "", article_addition.get("hscpNo") if isinstance(article_addition, dict) else "", raw.get("hscpNo") if isinstance(raw, dict) else "", raw.get("complexNo") if isinstance(raw, dict) else "", ) region_name = first_non_empty( detail.get("article_address"), detail.get("exposure_address"), article_detail.get("exposureAddress") if isinstance(article_detail, dict) else "", detail.get("address"), detail.get("road_address"), ) complex_name = first_non_empty( detail.get("complex_name"), detail.get("building_name"), article_detail.get("aptName") if isinstance(article_detail, dict) else "", article_detail.get("complexName") if isinstance(article_detail, dict) else "", article_addition.get("complexName") if isinstance(article_addition, dict) else "", detail.get("article_name"), article_detail.get("articleName") if isinstance(article_detail, dict) else "", ) article_name = first_non_empty( detail.get("article_name"), article_detail.get("articleName") if isinstance(article_detail, dict) else "", detail.get("building_name"), complex_name, ) return { "raw": raw, "article_detail": article_detail if isinstance(article_detail, dict) else {}, "hscp_no": clean_text(hscp_no), "region_name": clean_text(region_name), "complex_name": clean_text(complex_name), "article_name": clean_text(article_name), } def _fetch_complex_address_cache_for_draft(conn, region_name, complex_name, hscp_no=""): """ 초안 생성 단계에서는 단지명보다 complex_no/hscpNo를 우선한다. 같은 단지명이 여러 지역에 있을 수 있으므로 complex_no가 있으면 가장 안정적이다. """ region_name = clean_text(region_name) complex_name = clean_text(complex_name) hscp_no = clean_text(hscp_no) with conn.cursor() as cur: if hscp_no: cur.execute(""" SELECT * FROM blog_complex_address_cache WHERE complex_no = %s ORDER BY last_checked_at DESC, id DESC LIMIT 1 """, (hscp_no,)) row = cur.fetchone() if row: return row if region_name and complex_name: compact = re.sub(r"[\\s\\(\\)\\[\\]\\-_/·ㆍ,]+", "", complex_name) compact_no_apt = re.sub(r"아파트$", "", compact) names = [] for v in [complex_name, compact, compact_no_apt, compact_no_apt + "아파트" if compact_no_apt else ""]: v = clean_text(v) if v and v not in names: names.append(v) if names: placeholders = ",".join(["%s"] * len(names)) cur.execute(f""" SELECT * FROM blog_complex_address_cache WHERE region_name = %s AND complex_name IN ({placeholders}) ORDER BY last_checked_at DESC, id DESC LIMIT 1 """, [region_name, *names]) row = cur.fetchone() if row: return row return None def _row_get(row, key, default=""): if not row: return default if isinstance(row, dict): return row.get(key, default) return default def build_v2_draft_metadata(detail, draft_data=None, header_image=None, middle_banner_image=None): """ STEP107-06: V2 초안/큐 추적용 메타데이터. source_json에 저장되어 관리자/디버깅/발행 워커에서 확인 가능하다. """ detail = detail or {} draft_data = draft_data or {} return { "pipeline_version": V2_PIPELINE_VERSION, "draft_generator": V2_DRAFT_GENERATOR, "address": { "resolved_address": normalize_v2_address_text(detail.get("complex_resolved_address") or detail.get("address")), "road_address": normalize_v2_address_text(detail.get("road_address")), "jibun_address": normalize_v2_address_text(detail.get("jibun_address") or detail.get("complex_resolved_address") or detail.get("address")), "region_name": choose_v2_region_name_for_detail( detail.get("complex_resolved_address"), detail.get("jibun_address"), detail.get("address"), detail.get("region_name"), ), "complex_no": clean_text(detail.get("complex_no") or detail.get("hscp_no")), "source": clean_text(detail.get("complex_address_source") or detail.get("v2_address_source")), }, "school": { "school_name": clean_text(detail.get("school_name") or detail.get("complex_school_name")), "school_names": clean_text(detail.get("school_names") or detail.get("complex_school_names") or detail.get("v2_school_names")), "has_school_html": bool(clean_text(detail.get("school_html") or detail.get("v2_school_html"))), }, "draft": { "layout_type": clean_text((draft_data or {}).get("layout_type")), "draft_html_len": len(str((draft_data or {}).get("draft_html") or "")), "clipboard_html_len": len(str((draft_data or {}).get("clipboard_html") or "")), "plain_text_len": len(str((draft_data or {}).get("plain_text") or "")), }, "header_image": { "url": clean_text((header_image or {}).get("cdn_url") or (header_image or {}).get("image_url")), "has_header_image": bool(clean_text((header_image or {}).get("cdn_url") or (header_image or {}).get("image_url"))), }, "middle_banner": { "url": clean_text((middle_banner_image or {}).get("cdn_url") or (middle_banner_image or {}).get("image_url")), "has_middle_banner": bool(clean_text((middle_banner_image or {}).get("cdn_url") or (middle_banner_image or {}).get("image_url"))), }, "realtor_profile": { "image_url": clean_text(detail.get("profile_image_url") or detail.get("realtor_profile_image_url")), "default_avatar_used": bool(detail.get("v2_default_realtor_avatar_used")), "image_type": clean_text(detail.get("v2_realtor_profile_image_type")), "source_key": clean_text(detail.get("v2_realtor_profile_image_source_key")), }, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } def apply_complex_resolved_address_to_detail(conn, detail): """ STEP107-01 V2 주소 보강. 기존 V1 resolver 대신 services/v2_complex_address_enricher.py를 사용한다. 핵심: - 네이버 검색 단지카드 우선 - 실패 시 매물 주소 fallback - 중개사 주소는 최후 fallback - school_names가 있으면 school_html 생성 """ detail = dict(detail or {}) article_no = clean_text(detail.get("article_no")) if resolve_and_cache_complex_address_v2 is None: print(f"[V2 DRAFT ADDRESS SKIP] {article_no} resolver_import_failed") return detail try: address_result = resolve_and_cache_complex_address_v2( conn, detail, force_refresh=bool(V2_FORCE_REFRESH_ADDRESS), headless=True, ) except Exception as e: print(f"[V2 DRAFT ADDRESS ERROR] {article_no} {e}") return detail payload = _v2_pick_resolved_payload(address_result) resolved_address = normalize_v2_address_text(payload.get("resolved_address")) if payload.get("jibun_address"): payload["jibun_address"] = normalize_v2_address_text(payload.get("jibun_address")) if payload.get("road_address"): payload["road_address"] = normalize_v2_address_text(payload.get("road_address")) source = clean_text(payload.get("source") or "v2_address_resolver") if not resolved_address: fallback = clean_text(detail.get("address") or detail.get("road_address") or detail.get("article_address") or "") print(f"[V2 DRAFT ADDRESS RESOLVED] {article_no} {fallback} fallback_empty_result") return detail # 본문/법정표/SEO가 모두 같은 주소를 쓰도록 주요 필드에 일괄 반영한다. detail["address"] = resolved_address detail["article_address"] = resolved_address detail["exposure_address"] = resolved_address detail["complex_resolved_address"] = resolved_address detail["complex_address_source"] = source detail["v2_address_source"] = source if payload.get("road_address"): detail["road_address"] = payload.get("road_address") if payload.get("jibun_address"): detail["jibun_address"] = payload.get("jibun_address") if payload.get("complex_no"): detail["complex_no"] = payload.get("complex_no") detail["hscp_no"] = payload.get("complex_no") normalized_region_name = choose_v2_region_name_for_detail( payload.get("jibun_address"), payload.get("resolved_address"), payload.get("road_address"), payload.get("region_name"), detail.get("region_name"), ) if normalized_region_name: detail["region_name"] = normalized_region_name if payload.get("school_name"): detail["school_name"] = payload.get("school_name") detail["complex_school_name"] = payload.get("school_name") if payload.get("school_names"): detail["school_names"] = payload.get("school_names") detail["complex_school_names"] = payload.get("school_names") detail["v2_school_names"] = payload.get("school_names") school_html = build_v2_school_table_html_from_detail(detail) if school_html: detail["school_html"] = school_html detail["v2_school_html"] = school_html # source_json/detail에서 확인 가능하도록 raw_json에도 주입한다. raw = parse_json_safely(detail.get("raw_json")) if isinstance(raw, dict): raw["complexResolvedAddress"] = resolved_address raw["complexAddressSource"] = source raw["v2AddressResolved"] = True raw["v2AddressSource"] = source if payload.get("complex_no"): raw["complexNo"] = payload.get("complex_no") if payload.get("school_names"): raw["complexSchoolNames"] = payload.get("school_names") try: detail["raw_json"] = json.dumps(raw, ensure_ascii=False, default=json_default) except Exception: pass print( f"[V2 DRAFT ADDRESS RESOLVED] {article_no} " f"{resolved_address} {source} " f"road={payload.get('road_address') or ''} " f"jibun={payload.get('jibun_address') or ''} " f"complex_no={payload.get('complex_no') or ''} " f"school_names={payload.get('school_names') or ''}" ) return detail def process_article(conn, detail_row): article_no = str(detail_row.get("article_no")) realtor_id = detail_row.get("realtor_id") or 0 print("-" * 80) print(f"[DRAFT START] {article_no}") # STEP107-04: # 기존 draft source_json.detail에 지번 주소가 있으면 V2 재생성 전에 먼저 병합한다. existing_source_detail = fetch_existing_draft_source_detail(conn, article_no) if existing_source_detail: detail_row = merge_existing_draft_source_address(detail_row, existing_source_detail) # STEP107: 초안 생성 전에 V2 소재지 주소를 detail_row 전체에 먼저 반영한다. detail_row = apply_complex_resolved_address_to_detail(conn, detail_row) # STEP108-03: 중개사 대표사진이 없으면 코드 생성 SVG 픽토그램으로 보강한다. detail_row = apply_v2_default_realtor_avatar_to_detail(detail_row) images = fetch_article_images(conn, article_no) prices = fetch_article_prices(conn, article_no) schools = fetch_article_schools(conn, article_no) feature_comment = build_feature_comment_block( detail_row ) if not images: raise Exception("draft images empty") naver_realtor_id = clean_text( detail_row.get("naver_realtor_id") or detail_row.get("realtor_id_text") or detail_row.get("naver_realtor_code") or detail_row.get("realtor_account") or "" ) total_listing_url = clean_text( detail_row.get("realtor_total_listing_url") or detail_row.get("total_listing_url") or detail_row.get("agency_listing_url") or detail_row.get("fin_land_url") or "" ) if not total_listing_url and naver_realtor_id: total_listing_url = f"https://m.land.naver.com/agency/info/{naver_realtor_id}" detail_row["realtor_total_listing_url"] = total_listing_url detail_row["total_listing_url"] = total_listing_url # CTA 배너용 전체매물 링크를 한 번 더 보강한다. fallback_listing_url = build_realtor_listing_url(detail_row) if fallback_listing_url: detail_row["realtor_total_listing_url"] = fallback_listing_url detail_row["total_listing_url"] = fallback_listing_url total_listing_url = fallback_listing_url detail_row["middle_banner_image_url"] = normalize_middle_banner_image_url(detail_row) detail_row["realtor_info"] = { "office_name": detail_row.get("office_name", ""), "representative_name": detail_row.get("representative_name", ""), "license_number": detail_row.get("license_number", ""), "business_number": detail_row.get("business_number", ""), "office_phone": detail_row.get("office_phone", ""), "mobile_phone": detail_row.get("mobile_phone", ""), "office_address": detail_row.get("office_address", ""), "office_detail_address": detail_row.get("office_detail_address", ""), "address": detail_row.get("realtor_address", ""), "map_image_url": detail_row.get("map_image_url", ""), "seo_keywords": detail_row.get("seo_keywords", ""), "naver_realtor_id": naver_realtor_id, "realtor_total_listing_url": total_listing_url, "total_listing_url": total_listing_url, "realtor_listing_url": total_listing_url, } detail_row["market_prices"] = prices detail_row["schools"] = schools detail_row["floor_info_display"] = format_floor_text_for_display(detail_row.get("floor_info")) if detail_row["floor_info_display"]: detail_row["floor_info"] = detail_row["floor_info_display"] human_context = build_human_blog_context(detail_row) ai_sections = generate_ai_sections({ **detail_row, **human_context, }) ai_sections = enrich_ai_sections( ai_sections, human_context, ) layout_type = random.choice([ "story_first", "photo_first", "summary_first", "gallery_story", "location_focus", "recommendation_focus", "magazine_style", "interview_style", "checklist_style", "quiet_luxury", "short_review_style", "wide_visual", "photo_grid", "minimal_modern", "storytelling", "local_life", "broker_expert", "premium_report", ]) # STEP111-15 HEE Policy Service - Log Only # 실제 layout_type은 아직 변경하지 않는다. 운영 안전을 위해 apply=False 유지. log_policy_recommendation( detail=detail_row, images=images, ai_sections=ai_sections, current_layout_type=layout_type, apply=False, ) extra_images = get_extra_images_if_needed( conn=conn, realtor_id=realtor_id, article_no=article_no, images=images, ) print( f"[EXTRA IMAGES] " f"original={len(images)}, " f"extra={len(extra_images)}" ) header_background_url = select_header_background_from_article_images(images) if header_background_url: detail_row["header_background_url"] = header_background_url detail_row["background_url"] = header_background_url print("[HEADER BACKGROUND SOURCE] collected random:", header_background_url) else: print("[HEADER BACKGROUND SOURCE] mybox fallback") # ------------------------------------------------- # 중간배너 이미지 생성 → CDN 업로드 → 본문 배너로 사용 # ------------------------------------------------- middle_banner_image = None try: middle_banner_image = generate_middle_banner_image_for_article( detail=detail_row, article_no=article_no, ) if middle_banner_image and middle_banner_image.get("image_url"): middle_banner_cdn = upload_middle_banner_to_cafe24( conn=conn, article_no=article_no, banner_image=middle_banner_image, ) middle_banner_image = replace_middle_banner_url( banner_image=middle_banner_image, cdn_result=middle_banner_cdn, ) middle_banner_url = clean_text( middle_banner_image.get("cdn_url") or middle_banner_image.get("image_url") or "" ) if middle_banner_url: detail_row["middle_banner_image_url"] = middle_banner_url detail_row["realtor_banner_image_url"] = middle_banner_url detail_row["banner_image_url"] = middle_banner_url print("[MIDDLE BANNER FINAL URL]", middle_banner_url) if isinstance(detail_row.get("realtor_info"), dict): detail_row["realtor_info"]["middle_banner_image_url"] = middle_banner_url detail_row["realtor_info"]["realtor_banner_image_url"] = middle_banner_url except Exception as e: print("[MIDDLE BANNER ERROR]", str(e)) draft_data = build_blog_html( detail=detail_row, images=images, ai_sections=ai_sections, layout_type=layout_type, feature_comment=feature_comment, extra_images=extra_images, ) if not draft_data: raise Exception("draft_data empty") # STEP107-01: V2 배정학교 정보가 있으면 본문 HTML에 테이블로 삽입한다. draft_data = insert_v2_school_html_to_draft_data( draft_data=draft_data, detail=detail_row, ) print( "[DRAFT LENGTH AFTER BUILD]", "draft_html=", len(str(draft_data.get("draft_html") or "")), "clipboard_html=", len(str(draft_data.get("clipboard_html") or "")), "plain_text=", len(str(draft_data.get("plain_text") or "")), ) # ------------------------------------------------- # 대표이미지 V3 생성 → DB 저장 → 본문 첫 이미지로 삽입 # ------------------------------------------------- header_image = None header_image_error = "" try: header_image = generate_header_image_for_article( detail=detail_row, article_no=article_no, ) if header_image and header_image.get("image_url"): save_mybox_asset( conn=conn, article_no=article_no, realtor_id=realtor_id, header_image=header_image, ) save_header_image_to_article_images( conn=conn, article_no=article_no, realtor_id=realtor_id, header_image=header_image, ) # Cafe24 CDN 업로드 후 대표이미지 URL을 CDN URL로 교체 cdn_result = None try: cdn_result = upload_header_image_to_cafe24( conn=conn, article_no=article_no, local_path=header_image.get("image_path") or "", filename=header_image.get("filename") or "", ) save_cdn_asset( conn=conn, article_no=article_no, realtor_id=realtor_id, header_image=header_image, cdn_result=cdn_result, status="uploaded", ) header_image = replace_header_image_url( header_image=header_image, cdn_result=cdn_result, ) print("[CDN UPLOAD OK]", cdn_result.get("cdn_url")) except Exception as e: print("[CDN UPLOAD ERROR]", str(e)) save_cdn_asset( conn=conn, article_no=article_no, realtor_id=realtor_id, header_image=header_image, cdn_result={}, status="failed", error_message=str(e), ) # CDN 실패 시에도 초안 생성은 중단하지 않는다. # 단, 이 경우 Python 서버 URL이 들어가므로 운영 발행 전 확인 필요. draft_data = prepend_header_image_to_draft_data( draft_data=draft_data, header_image=header_image, ) except Exception as e: header_image_error = str(e) print("[HEADER IMAGE ERROR]", header_image_error) # ------------------------------------------------- # 법정 표시광고 표준 표 교체 # ------------------------------------------------- draft_data = replace_legal_disclosure_in_draft_data( draft_data=draft_data, detail=detail_row, ) # ------------------------------------------------- # 중개사무소 위치 지도 이미지 삽입 # 법정 표시광고 아래 / 상담 및 중개사 안내 위에 배치 # ------------------------------------------------- draft_data = insert_realtor_location_before_contact( draft_data=draft_data, detail=detail_row, ) # ------------------------------------------------- # SEO 해시태그 / 검색 노출용 문단 하단 삽입 # ------------------------------------------------- draft_data = append_seo_footer_to_draft_data( draft_data=draft_data, detail=detail_row, ) # STEP107-46: 발행 worker는 붙여넣기만 하도록, 법정/중개사 안내 블록은 초안 생성 단계에서 최종 보장한다. draft_data = ensure_v2_legal_disclosure_final_guard( draft_data=draft_data, detail=detail_row, ) verify_v2_draft_html_required_blocks(draft_data) print( "[DRAFT LENGTH BEFORE SAVE]", "draft_html=", len(str(draft_data.get("draft_html") or "")), "clipboard_html=", len(str(draft_data.get("clipboard_html") or "")), "plain_text=", len(str(draft_data.get("plain_text") or "")), ) v2_metadata = build_v2_draft_metadata( detail=detail_row, draft_data=draft_data, header_image=header_image, middle_banner_image=middle_banner_image if 'middle_banner_image' in locals() else None, ) source_json = { "pipeline_version": V2_PIPELINE_VERSION, "draft_generator": V2_DRAFT_GENERATOR, "v2_metadata": v2_metadata, "v2_address": v2_metadata.get("address"), "v2_school": v2_metadata.get("school"), "v2_school_html": detail_row.get("v2_school_html") or detail_row.get("school_html") or "", "detail": detail_row, "images": images, "prices": prices, "schools": schools, "human_context": human_context, "feature_comment": feature_comment, "header_image": header_image, "header_image_error": header_image_error, "middle_banner_image": middle_banner_image if 'middle_banner_image' in locals() else None, } print("[V2 SOURCE META]", json.dumps(v2_metadata, ensure_ascii=False, default=json_default)) save_draft( conn=conn, realtor_id=realtor_id, article_no=article_no, draft_data=draft_data, ai_sections=ai_sections, source_json=source_json, ) with conn.cursor() as cur: cur.execute(""" SELECT id FROM blog_article_drafts WHERE article_no = %s LIMIT 1 """, (article_no,)) saved = cur.fetchone() if not saved: raise Exception(f"draft save failed: {article_no}") draft_id = saved.get("id") create_publish_queue( conn=conn, realtor_id=realtor_id, article_no=article_no, draft_id=draft_id, ) print(f"[DRAFT SAVED] {article_no}") print(f"[LAYOUT] {layout_type}") print(f"[IMAGES] {len(images)}") print(f"[PRICES] {len(prices)}") print(f"[SCHOOLS] {len(schools)}") def main(): parser = argparse.ArgumentParser() parser.add_argument("--realtor-id", type=int, default=None) parser.add_argument("--article-no", type=str, default=None) parser.add_argument("--limit", type=int, default=9999) parser.add_argument("--draft-buffer", type=int, default=45, help="중개사별 초안 재고 목표 기본값") parser.add_argument("--replenish-count", type=int, default=5, help="중개사별 1회 최대 초안 보충 수") parser.add_argument("--limit-realtors", type=int, default=100, help="초안 보충 대상 중개사 수 제한") parser.add_argument("--force", action="store_true", help="기존 초안이 있어도 강제로 재생성") parser.add_argument("--force-refresh-address", action="store_true", help="V2 주소 캐시를 무시하고 네이버 검색/매물주소 재조회") parser.add_argument("--lock-ttl-minutes", type=int, default=240) parser.add_argument("--no-db-lock", action="store_true") args = parser.parse_args() global V2_FORCE_REFRESH_ADDRESS V2_FORCE_REFRESH_ADDRESS = bool(args.force_refresh_address) lock_name = "generate_blog_drafts_v2" lock_owner = None run_id = None success = 0 failed = 0 skipped = 0 total = 0 final_status = "success" final_message = "generate_blog_drafts_v2 completed" if not args.no_db_lock: locked, lock_owner = acquire_db_lock( lock_name, ttl_minutes=args.lock_ttl_minutes, ) if not locked: print("[SKIP] generate_blog_drafts already running") return conn = get_conn() try: run_id = create_pipeline_run( "generate_blog_drafts_v2", message="generate_blog_drafts_v2 started", meta={ "realtor_id": args.realtor_id, "article_no": args.article_no, "limit": args.limit, "draft_buffer": args.draft_buffer, "replenish_count": args.replenish_count, "limit_realtors": args.limit_realtors, "force": args.force, "force_refresh_address": args.force_refresh_address, }, ) print("=" * 80) print("[START] generate_blog_drafts_v2") print("TIME:", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) print("=" * 80) print("FORCE:", "ON" if args.force else "OFF") print("V2_FORCE_REFRESH_ADDRESS:", "ON" if args.force_refresh_address else "OFF") print("V2_DEFAULT_REALTOR_AVATAR_URL:", V2_DEFAULT_REALTOR_AVATAR_URL) articles = fetch_target_articles( conn=conn, realtor_id=args.realtor_id, article_no=args.article_no, limit=args.limit, force=args.force, draft_buffer=args.draft_buffer, replenish_count=args.replenish_count, limit_realtors=args.limit_realtors, ) total = len(articles) print(f"[TARGET ARTICLES] {total}") add_pipeline_log( run_id=run_id, level="info", step_name="target_articles", message=f"target articles: {total}", article_no=args.article_no, realtor_id=args.realtor_id, context={"target_count": total}, ) for row in articles: valid_article, valid_reason = is_valid_collected_article_for_draft(row) if not valid_article: print( "[DRAFT SKIP INVALID DETAIL]", row.get("article_no"), valid_reason ) continue article_no = str(row.get("article_no") or "") realtor_id = row.get("realtor_id") try: process_article(conn, row) success += 1 add_pipeline_log( run_id=run_id, level="info", step_name="draft_success", message=f"draft saved: {article_no}", article_no=article_no, realtor_id=realtor_id, ) except Exception as e: failed += 1 try: conn.rollback() except Exception: pass print(f"[DRAFT ERROR] article_no={article_no} / {str(e)}") add_pipeline_log( run_id=run_id, level="error", step_name="draft_error", message=str(e), article_no=article_no, realtor_id=realtor_id, context={"traceback": traceback.format_exc()}, ) print("=" * 80) print("[DONE] generate_blog_drafts_v2") print("=" * 80) if failed > 0: final_status = "failed" final_message = f"completed with failures: success={success}, failed={failed}" except Exception as e: final_status = "failed" final_message = str(e)[:2000] print("[FATAL]", final_message) traceback.print_exc() add_pipeline_log( run_id=run_id, level="error", step_name="fatal", message=final_message, context={"traceback": traceback.format_exc()}, ) finally: try: conn.close() except Exception: pass finish_pipeline_run( run_id, final_status, total=total, success=success, failed=failed, skipped=skipped, message=final_message, meta={ "realtor_id": args.realtor_id, "article_no": args.article_no, "limit": args.limit, "draft_buffer": args.draft_buffer, "replenish_count": args.replenish_count, "limit_realtors": args.limit_realtors, "force": args.force, "force_refresh_address": args.force_refresh_address, }, ) if lock_owner: release_db_lock(lock_name, lock_owner) if __name__ == "__main__": main()