# -*- 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" 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.cafe24_cdn_service import upload_header_image_to_cafe24 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_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 set_if_empty("address", article_detail.get("exposureAddress"), article_realtor.get("address") if (article_realtor := raw.get("articleRealtor") or {}) 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_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): """ 중개사 매물설명을 사람이 작성한 문단처럼 정리한다. 원문을 억지 문장으로 바꾸지 않고, 주요 생활/단지/커뮤니티 정보를 자연스럽게 묶는다. """ raw = extract_broker_description(detail) if not raw: return { "title": "", "content": "", } text = raw.replace("\r", "\n") 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"\n{2,}", "\n", text) clean_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) <= 12: continue if "위치" in line and len(line) <= 20 and "단지" not in line: continue if line not in clean_lines: clean_lines.append(line) joined = " ".join(clean_lines) paragraphs = [] # 세대수/대단지 if "3100" in joined or "3,100" in joined or ("1990" in joined and "1110" in joined): paragraphs.append( "해밀마을 1단지와 2단지를 포함해 약 3,100세대 규모로 형성된 대단지 아파트입니다." ) elif "대단지" in joined: paragraphs.append( "대단지 아파트 특유의 생활 편의성과 단지 규모감을 함께 기대해볼 수 있습니다." ) # 학교/교통 edu_transport = [] if any(x in joined for x in ["초", "중", "고", "학교"]): edu_transport.append("단지 내 또는 가까운 생활권에 학교시설이 위치해 교육환경을 함께 살펴보시기 좋습니다") if "BRT" in joined.upper() or "정류장" in joined: edu_transport.append("BRT 정류장 이용이 편리한 편입니다") if edu_transport: paragraphs.append(". ".join(edu_transport) + ".") # 상권/생활 if "상권" in joined: paragraphs.append( "주변 상권도 형성되어 있어 일상적인 생활 편의성을 함께 고려해보실 수 있습니다." ) # 커뮤니티 community_items = [] for label in ["수영장", "카페테리아", "헬스장", "골프연습장", "게스트하우스", "공연장", "맘스테이션", "작은도서관"]: if label in joined and label not in community_items: community_items.append(label) if community_items: if len(community_items) >= 4: first = ", ".join(community_items[:4]) rest = ", ".join(community_items[4:]) if rest: paragraphs.append( f"커뮤니티 시설로는 {first} 등을 비롯해 {rest} 등도 함께 확인됩니다." ) else: paragraphs.append( f"커뮤니티 시설로는 {first} 등을 이용하실 수 있습니다." ) else: paragraphs.append( f"커뮤니티 시설로는 {', '.join(community_items)} 등을 참고하실 수 있습니다." ) # fallback if not paragraphs: for line in clean_lines[:5]: line = re.sub(r"\s+", " ", line).strip() if not line: continue if not line.endswith((".", "다", "요")): line = line + "입니다." paragraphs.append(line) content = "\n\n".join(paragraphs[:4]) 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 {} 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 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( detail.get("address") or detail.get("road_address") or "" ), "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 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 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() # 대표이미지 URL 컬럼이 존재하는 경우 별도 업데이트 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 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}, updated_at = NOW() WHERE article_no = %s", values ) conn.commit() except Exception as e: print("[DRAFT HEADER IMAGE URL 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, "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 # --------------------------------------------------------- # 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 apply_complex_resolved_address_to_detail(conn, detail): """ STEP102-13 핵심 보강. build_blog_html / 법정표 생성 전에 detail['address']를 지번 포함 주소로 덮어쓴다. 그래야 핵심정보 표, 법정표, SEO 영역이 모두 같은 주소를 사용한다. """ article_no = clean_text(detail.get("article_no")) keys = _extract_complex_address_keys_for_draft(detail) region_name = keys["region_name"] complex_name = keys["complex_name"] hscp_no = keys["hscp_no"] article_name = keys["article_name"] if not region_name or not complex_name: fallback = clean_text(detail.get("address") or detail.get("road_address") or region_name) print(f"[DRAFT ADDRESS RESOLVED] {article_no} {fallback} fallback_missing_keys") return detail row = _fetch_complex_address_cache_for_draft( conn, region_name=region_name, complex_name=complex_name, hscp_no=hscp_no, ) if not row: try: from services.naver_search_complex_resolver import resolve_and_cache_complex_address row = resolve_and_cache_complex_address( conn, region_name, complex_name, force_refresh=False, hscp_no=hscp_no, article_name=article_name, ) except Exception as e: print(f"[DRAFT ADDRESS RESOLVE ERROR] {article_no} {region_name} / {complex_name} / {e}") row = None resolved_address = clean_text(_row_get(row, "resolved_address")) complex_no = clean_text(_row_get(row, "complex_no") or hscp_no) source = clean_text(_row_get(row, "source") or "complex_address_cache") if resolved_address: 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 if complex_no: detail["complex_no"] = complex_no detail["hscp_no"] = complex_no # 법정표 함수가 raw_json의 complexResolvedAddress를 최우선으로 읽도록 raw_json에도 주입한다. raw = keys.get("raw") or {} if isinstance(raw, dict): raw["complexResolvedAddress"] = resolved_address raw["complexAddressSource"] = source if complex_no: raw["complexNo"] = complex_no try: detail["raw_json"] = json.dumps(raw, ensure_ascii=False, default=json_default) except Exception: pass print(f"[DRAFT ADDRESS RESOLVED] {article_no} {resolved_address} {source}") return detail fallback = clean_text(detail.get("address") or detail.get("road_address") or region_name) print(f"[DRAFT ADDRESS RESOLVED] {article_no} {fallback} fallback_exposureAddress") 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}") # STEP102-13: 초안 생성 전에 소재지 지번 주소를 detail_row 전체에 먼저 반영한다. detail_row = apply_complex_resolved_address_to_detail(conn, 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", ]) 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") 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, ) 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 "")), ) source_json = { "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, } 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("--lock-ttl-minutes", type=int, default=240) parser.add_argument("--no-db-lock", action="store_true") args = parser.parse_args() lock_name = "generate_blog_drafts" lock_owner = None run_id = None success = 0 failed = 0 skipped = 0 total = 0 final_status = "success" final_message = "generate_blog_drafts 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", message="generate_blog_drafts 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, }, ) print("=" * 80) print("[START] generate_blog_drafts") print("TIME:", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) print("=" * 80) print("FORCE:", "ON" if args.force else "OFF") 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") 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, }, ) if lock_owner: release_db_lock(lock_name, lock_owner) if __name__ == "__main__": main()