#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""상세수집 완료 매물을 독립 웹진 기사로 생성한다.

기존 blog_ 테이블은 읽기만 하며 기존 블로그 초안·발행 Queue를 변경하지 않는다.
정보가 부족해도 기사는 공개하고, 헤드라인 적합성만 별도로 판정한다.
"""

import argparse
import difflib
import hashlib
import html
import json
import os
import re
import sys
import traceback
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parents[1]
if str(BASE_DIR) not in sys.path:
    sys.path.insert(0, str(BASE_DIR))

from db import get_conn
from update_site_area_data import refresh_site_area_snapshots
from update_complex_public_data import refresh_complex_public_data
from update_policy_feeds import refresh_policy_feeds


BAD_TEXT = {"", "-", "null", "none", "undefined", "해당없음"}

PROPERTY_TYPE_MAP = {
    "A01": "아파트", "A02": "오피스텔", "A03": "주상복합",
    "A05": "연립·다세대",
    "B01": "상가·사무실", "B02": "상가·사무실", "B03": "상가·사무실",
    "C01": "단독·다가구", "C02": "단독·다가구", "C03": "단독·다가구",
    "D01": "토지", "D02": "건물", "D03": "공장·창고",
    "APT": "아파트", "ABYG": "분양권", "OPST": "오피스텔",
    "OBYG": "분양권", "YR": "연립·다세대", "VL": "연립·다세대",
    "DDDGG": "단독·다가구", "JWJT": "단독·다가구",
    "SMS": "상가·사무실", "SG": "상가·사무실", "GJCG": "상가·사무실",
    "GY": "공장·창고", "TJ": "토지", "GM": "건물",
}

TRADE_TYPE_MAP = {"A1": "매매", "B1": "전세", "B2": "월세", "B3": "단기임대"}


def clean(value):
    text = re.sub(r"\s+", " ", str(value or "")).strip()
    return "" if text.lower() in BAD_TEXT else text


def parse_json(value):
    if isinstance(value, dict):
        return value
    try:
        data = json.loads(value or "{}")
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def recursive_find(obj, names):
    wanted = {str(x).lower() for x in names}
    if isinstance(obj, dict):
        for key, value in obj.items():
            if str(key).lower() in wanted and clean(value):
                return value
        for value in obj.values():
            found = recursive_find(value, wanted)
            if clean(found):
                return found
    elif isinstance(obj, list):
        for item in obj:
            found = recursive_find(item, wanted)
            if clean(found):
                return found
    return ""


def normalize_display_code(value, mappings):
    value = clean(value)
    if not value:
        return ""
    mapped = mappings.get(value.upper())
    if mapped:
        return mapped
    return "" if re.fullmatch(r"[A-Z]+\d+", value.upper()) else value


def format_won(value):
    try:
        amount = int(float(str(value).replace(",", "").strip()))
    except Exception:
        return ""
    return f"{amount:,}원" if amount > 0 else ""


def normalize_maintenance_fee(value):
    """월별 관리비 JSON/코드를 사용자용 한 줄 금액으로 변환한다."""
    if isinstance(value, (dict, list)):
        parsed = value
    else:
        text = str(value or "").strip()
        if not text:
            return ""
        if text[:1] in "[{":
            try:
                parsed = json.loads(text)
            except Exception:
                # DB에 Python dict 문자열 형태(작은따옴표)로 저장된 과거 데이터도 처리한다.
                amount_match = re.search(r"['\"]?(?:totalPrice|totalCost|maintenanceFee|managementCost|amount)['\"]?\s*:\s*['\"]?([0-9,]+)", text, flags=re.I)
                month_match = re.search(r"['\"]?(?:basisYearMonth|yearMonth)['\"]?\s*:\s*['\"]?(\d{6})", text, flags=re.I)
                display = format_won(amount_match.group(1)) if amount_match else ""
                if display:
                    return f"{month_match.group(1)} 기준 {display}" if month_match else display
                return "관리비 별도 확인"
        else:
            if len(text) > 120 or "costsByDate" in text or "totalPrice" in text:
                return "관리비 별도 확인"
            numeric = format_won(text)
            return numeric or clean(text)

    candidates = []
    if isinstance(parsed, dict):
        monthly = parsed.get("costsByDate") or parsed.get("monthlyCosts") or parsed.get("costList") or []
        if isinstance(monthly, list):
            candidates.extend(monthly)
        candidates.append(parsed)
    elif isinstance(parsed, list):
        candidates.extend(parsed)

    for item in candidates:
        if not isinstance(item, dict):
            continue
        for key in ("totalPrice", "totalCost", "maintenanceFee", "managementCost", "amount"):
            display = format_won(item.get(key))
            if display:
                month = clean(item.get("basisYearMonth") or item.get("yearMonth") or item.get("month"))
                return f"{month} 기준 {display}" if re.fullmatch(r"\d{6}", month) else display
    return "관리비 별도 확인"


def normalize_adjacent_title_duplicates(value):
    """초안 제목의 연속 단어/구절 중복만 보수적으로 제거한다."""
    text = clean(value)
    if not text:
        return ""
    text = re.sub(r"(?:네이버\s*부동산\s*)?후보\s*매물", "", text, flags=re.I)
    parts = [clean(part) for part in re.split(r"\s*·\s*", text) if clean(part)]
    normalized_parts = []
    for part in parts:
        tokens = part.split()
        changed = True
        while changed and len(tokens) > 1:
            changed = False
            for span in range(min(8, len(tokens) // 2), 0, -1):
                index = 0
                while index + span * 2 <= len(tokens):
                    if tokens[index:index + span] == tokens[index + span:index + span * 2]:
                        del tokens[index + span:index + span * 2]
                        changed = True
                    else:
                        index += 1
        part = " ".join(tokens)
        if part and (not normalized_parts or part != normalized_parts[-1]):
            normalized_parts.append(part)
    return " · ".join(normalized_parts)


def normalize_address(value):
    """반복 수집된 시·도/시·군·구 주소 토큰을 한 번만 남긴다."""
    text = clean(value)
    if not text:
        return ""
    result, seen_admin = [], set()
    for token in text.split():
        key = token.strip(",()")
        is_admin = bool(re.search(r"(?:특별자치도|특별자치시|특별시|광역시|도|시|군|구|읍|면|동|리)$", key))
        if is_admin and key in seen_admin:
            continue
        if is_admin:
            seen_admin.add(key)
        result.append(token)
    return " ".join(result)


def enriched(row):
    data = dict(row or {})
    raw = parse_json(data.get("raw_json"))
    draft_source = parse_json(data.get("article_draft_source_json"))
    if not draft_source:
        draft_source = parse_json(data.get("realestate_draft_source_json"))
    sources = [raw, draft_source]

    def fill(key, *names):
        if clean(data.get(key)):
            return
        for source in sources:
            value = recursive_find(source, names)
            if clean(value):
                data[key] = clean(value)
                return

    fill("article_name", "articleName", "complexName", "buildingName")
    fill("building_name", "buildingName", "complexName", "articleName")
    fill("complex_name", "aptName", "complexName", "complexNameForMap", "complexTitle", "articleComplexName")
    fill("trade_type", "tradeTypeName", "tradeType")
    fill("real_estate_type", "realestateTypeName", "articleRealEstateTypeName")
    fill("price_text", "dealOrWarrantPrc", "priceText")
    fill("area_info", "areaInfo", "exclusiveArea", "supplyArea")
    fill("floor_info", "floorInfo", "correspondingFloorCount")
    fill("direction_code", "directionName", "direction")
    fill("article_feature_desc", "detailDescription", "articleFeatureDescription", "articleFeatureDesc")
    fill("address_text", "exposureAddress", "roadAddress", "address")
    fill("room_count", "roomCount", "roomCnt")
    fill("bathroom_count", "bathroomCount", "bathroomCnt")
    fill("maintenance_fee", "maintenanceFee", "maintenanceCost", "monthlyManagementCost")
    fill("move_in_date", "moveInDate", "moveInTypeName", "moveInPossibleDate")
    fill("parking_count", "parkingCount", "totalParkingCount")
    fill("approval_date", "useApproveYmd", "approvalDate", "useApprovalDate")
    fill("total_floor", "totalFloorCount", "totalFloor")
    fill("building_usage", "buildingUseName", "buildingUsage")

    # DB의 real_estate_type에는 과거 articleTypeCode(A01/B01 등)가 섞여 있다.
    # 네이버 상세 원문의 실제 매물종류명을 발견하면 이를 항상 우선한다.
    raw_property_name = clean(recursive_find(raw, (
        "realestateTypeName", "realEstateTypeName", "articleRealEstateTypeName",
    )))
    if raw_property_name:
        data["real_estate_type"] = raw_property_name

    if clean(data.get("cached_complex_name")) and (
            not clean(data.get("complex_name")) or re.fullmatch(r"\d{1,4}동", clean(data.get("complex_name")))):
        data["complex_name"] = clean(data.get("cached_complex_name"))

    data["real_estate_type"] = normalize_display_code(data.get("real_estate_type"), PROPERTY_TYPE_MAP)
    data["trade_type"] = normalize_display_code(data.get("trade_type"), TRADE_TYPE_MAP)
    data["maintenance_fee"] = normalize_maintenance_fee(data.get("maintenance_fee"))
    data["address_text"] = normalize_address(data.get("address_text"))

    html_keys = (
        "article_draft_clipboard_html", "article_draft_html",
        "article_draft_content_html", "article_draft_body_html",
        "realestate_draft_clipboard_html", "realestate_draft_html",
        "realestate_draft_content_html", "realestate_draft_body_html",
        "blog_draft_html", "article_draft_plain_text", "realestate_draft_plain_text",
    )
    html_candidates = [str(data.get(key) or "").strip() for key in html_keys]
    html_candidates = [value for value in html_candidates if value]
    if html_candidates:
        def html_quality(value):
            lower = value.lower()
            return (
                len(value)
                + (12000 if re.search(r"<(?:p|div|section|h[1-6]|table)\b", lower) else 0)
                + (20000 if "<table" in lower else 0)
                + (18000 if "중개대상물 명시사항" in value else 0)
                + (5000 if re.search(r"<h[2-4]\b", lower) else 0)
                + (3000 if "관리비" in value else 0)
            )
        data["blog_draft_html"] = max(html_candidates, key=html_quality)

    if not clean(data.get("complex_name")):
        draft_title = normalize_adjacent_title_duplicates(data.get("blog_draft_title"))
        first_part = clean(draft_title.split("·", 1)[0])
        building = clean(data.get("building_name"))
        if building and first_part.endswith(building):
            first_part = clean(first_part[:-len(building)])
        if first_part and first_part not in {"아파트", "오피스텔", "단독·다가구", "상가", "토지", "건물"}:
            data["complex_name"] = first_part
    return data


def slugify(value, article_no):
    value = clean(value).lower()
    value = re.sub(r"[^0-9a-z가-힣]+", "-", value).strip("-")
    value = value[:170].strip("-") or "property"
    return f"{value}-{clean(article_no)}"[:220]


def category_code(row):
    payload = parse_json(row.get("raw_json"))
    raw_name = clean(recursive_find(payload, (
        "realestateTypeName", "realEstateTypeName", "articleRealEstateTypeName",
    )))
    raw_code = clean(recursive_find(payload, (
        "realestateTypeCode", "realEstateTypeCode", "articleRealEstateTypeCode",
    )))
    value = (
        raw_name
        or normalize_display_code(raw_code, PROPERTY_TYPE_MAP)
        or normalize_display_code(row.get("real_estate_type"), PROPERTY_TYPE_MAP)
        or clean(row.get("real_estate_type"))
    )
    search_text = " ".join(clean(row.get(key)) for key in (
        "title", "blog_draft_title", "article_name", "building_name",
    ))
    value = f"{value} {search_text}".strip()
    mappings = (
        ("아파트분양권", "분양권"), ("오피스텔분양권", "분양권"),
        ("분양권", "분양권"), ("아파트", "아파트"),
        ("오피스텔", "오피스텔"), ("단독", "단독·다가구"),
        ("다가구", "단독·다가구"), ("연립", "연립·다세대"),
        ("다세대", "연립·다세대"), ("빌라", "연립·다세대"),
        ("상가", "상가·사무실"), ("사무실", "상가·사무실"),
        ("토지", "토지"), ("공장", "공장·창고"),
        ("창고", "공장·창고"), ("건물", "건물"),
    )
    for keyword, label in mappings:
        if keyword in value:
            return label
    return "기타"


def build_title(row):
    building = clean(row.get("building_name"))
    complex_name = clean(row.get("complex_name"))
    invalid_complex_words = ("네이버 부동산 후보 매물", "후보 매물", "후보매물")
    if (not complex_name or len(complex_name) > 100 or "·" in complex_name
            or any(word in complex_name for word in invalid_complex_words)
            or complex_name in {"아파트", "오피스텔", "단독", "다가구", "상가", "토지", "건물"}):
        complex_name = ""

    estate = normalize_display_code(row.get("real_estate_type"), PROPERTY_TYPE_MAP)
    trade = normalize_display_code(row.get("trade_type"), TRADE_TYPE_MAP)
    price = clean(row.get("price_text"))

    # generate_blog_drafts.py가 만든 제목을 최우선으로 사용한다.
    draft_title = normalize_adjacent_title_duplicates(row.get("blog_draft_title"))
    if draft_title and not any(word in draft_title for word in invalid_complex_words):
        title_parts = []
        for part in (clean(item) for item in draft_title.split("·")):
            part = normalize_display_code(part, PROPERTY_TYPE_MAP) or normalize_display_code(part, TRADE_TYPE_MAP)
            if part and part not in title_parts:
                title_parts.append(part)
        if title_parts and complex_name:
            first = title_parts[0]
            if re.fullmatch(r"\d{1,4}동", first) or (building and first == building):
                title_parts[0] = clean(f"{complex_name} {building or first}")
        if title_parts:
            return " · ".join(title_parts)[:300]

    if complex_name:
        property_name = complex_name
        if building and building not in property_name:
            property_name = f"{property_name} {building}"
        return " · ".join(value for value in (property_name, estate, trade, price) if value)[:300]

    article_name = clean(row.get("article_name"))
    if any(word in article_name for word in invalid_complex_words):
        article_name = ""
    name = building or article_name
    office = clean(row.get("office_name")) or "중개사"
    parts = []
    for value in (name, estate, trade, price):
        if value and value not in parts:
            parts.append(value)
    if not parts:
        return f"{office} 신규 부동산 매물 안내 · 매물번호 {clean(row.get('article_no'))}"
    return " · ".join(parts)[:300]


def build_natural_summary(row):
    complex_name = clean(row.get("complex_name"))
    building = clean(row.get("building_name"))
    name_parts = []
    if complex_name:
        name_parts.append(complex_name)
    if building and building not in complex_name:
        name_parts.append(building)
    name = clean(" ".join(name_parts)) or building or clean(row.get("article_name")) or "부동산 매물"
    estate = clean(row.get("real_estate_type"))
    trade = clean(row.get("trade_type"))
    price = clean(row.get("price_text")); area = clean(row.get("area_info"))
    floor = clean(row.get("floor_info")); direction = clean(row.get("direction_code"))
    address = clean(row.get("address_text")) or clean(row.get("region"))
    subject = " ".join(x for x in (name, estate, trade) if x)
    sentences = [f"{subject} 매물을 안내해 드립니다."]
    conditions = []
    if price: conditions.append(f"가격은 {price}")
    if area: conditions.append(f"면적은 {area}")
    if floor: conditions.append(f"층 정보는 {floor}")
    if direction: conditions.append(f"방향은 {direction}")
    if conditions: sentences.append(", ".join(conditions) + "입니다.")
    if address: sentences.append(f"소재지는 {address}입니다. 자세한 조건과 방문 가능 일정은 상담을 통해 안내해 드리겠습니다.")
    else: sentences.append("세부 조건과 방문 가능 일정은 상담을 통해 안내해 드리겠습니다.")
    return " ".join(sentences)[:500]


def extract_listing_description(value):
    """블로그 초안의 '중개사 매물설명' 본문을 목록용 원문으로 복원한다."""
    source = str(value or "").strip()
    if not source:
        return ""
    candidates = []
    for pattern in (
        r'<p\b[^>]*class=["\'][^"\']*listing-description[^"\']*["\'][^>]*>(.*?)</p>',
        r'<(?:h[1-6]|div|p|strong|b)\b[^>]*>\s*중개사\s*매물\s*설명\s*</(?:h[1-6]|div|p|strong|b)>\s*(.*?)(?=<(?:h[1-6]|div|p|strong|b)\b[^>]*>\s*(?:매물\s*핵심\s*정보|확인\s*포인트|추천\s*포인트|관리비|중개대상물\s*명시사항|평면도|상담)|중개대상물\s*명시사항|$)',
        r'중개사\s*매물\s*설명\s*(.*?)(?=매물\s*핵심\s*정보|확인\s*포인트|추천\s*포인트|중개대상물\s*명시사항|평면도|상담|$)',
    ):
        match = re.search(pattern, source, flags=re.I | re.S)
        if match:
            candidates.append(match.group(1))
    for candidate in candidates:
        candidate = re.sub(r"<(script|style)[^>]*>.*?</\1>", " ", candidate, flags=re.I | re.S)
        candidate = re.sub(r"<br\s*/?>", "\n", candidate, flags=re.I)
        candidate = re.sub(r"</p\s*>", "\n", candidate, flags=re.I)
        text = html.unescape(re.sub(r"<[^>]+>", " ", candidate))
        lines = [clean(line) for line in re.split(r"[\r\n]+", text) if clean(line)]
        text = " ".join(lines)
        text = re.sub(r"[❤♥💕💖💗💙💚💛🧡💜⭐🌟✨]+", "", text)
        text = re.sub(r"(?<![0-9A-Za-z가-힣])#[0-9A-Za-z가-힣_·~-]+", "", text)
        text = clean(text)
        if len(text) >= 20:
            return text[:1000].rstrip()
    return ""


def build_summary(row):
    # 1순위는 초안에 포함된 중개사의 실제 매물설명이다.
    description = extract_listing_description(row.get("blog_draft_html"))
    if description:
        return description[:700].rstrip()
    # 실제 매물설명이 없을 때만 짧은 특징 설명을 사용한다.
    description = clean(html.unescape(re.sub(r"<[^>]+>", " ", str(row.get("article_feature_desc") or ""))))
    description = re.sub(r"[❤♥💕💖💗💙💚💛🧡💜⭐🌟✨]+", "", description)
    if description:
        # 목록에서는 중개사가 직접 작성한 매물설명을 가장 먼저 사용한다.
        return description[:500].rstrip()
    return build_natural_summary(row)


def clean_draft_html(value):
    source = str(value or "").strip()
    if not source:
        return ""
    source = re.sub(r"<!--.*?-->", "", source, flags=re.S)
    source = re.sub(
        r"<(?:h[1-6]|div|p|strong|b)\b[^>]*>\s*(중개대상물\s*명시사항[^<]*)</(?:h[1-6]|div|p|strong|b)>",
        lambda match: "<h2>__MULTI_LEGAL_NOTICE__" + clean(match.group(1).replace("중개대상물", "").replace("명시사항", "")) + "</h2>",
        source, count=1, flags=re.I | re.S,
    )
    source = re.sub(r"중개대상물\s*명시사항", "__MULTI_LEGAL_NOTICE__", source, count=1, flags=re.I)
    source = re.sub(r"<(script|style|svg|iframe|video|audio|form|button)[^>]*>.*?</\1\s*>", "", source, flags=re.I | re.S)
    source = re.sub(r"<img\b[^>]*>", "", source, flags=re.I | re.S)
    source = re.sub(r"<(?:h[1-6]|div|p|strong|b)\b[^>]*>\s*중개사\s*매물\s*설명\s*</(?:h[1-6]|div|p|strong|b)>", "", source, flags=re.I | re.S)
    source = re.sub(r"<h[1-4][^>]*>\s*평면도\s*및\s*구조\s*확인\s*</h[1-4]>.*?(?=<h[1-4]\b|__MULTI_LEGAL_NOTICE__|$)", "", source, flags=re.I | re.S)
    source = re.sub(r"<h[1-4][^>]*>\s*추천\s*포인트\s*</h[1-4]>.*?(?=<h[1-4]\b|$)", "", source, flags=re.I | re.S)
    source = re.sub(
        r"<h[1-4][^>]*>\s*상담(?:\s*및\s*중개사)?\s*안내\s*</h[1-4]>.*?(?=__MULTI_LEGAL_NOTICE__|$)",
        "", source, flags=re.I | re.S,
    )
    source = re.sub(r"[^<]*(?:에서\s*안내하는\s*최신\s*부동산\s*매물(?:입니다)?\.?)[^<]*", "", source, flags=re.I)
    source = re.sub(
        r"<p\b[^>]*>\s*[^<]{0,300}?매물(?:\s*정보)?(?:이|가)?\s*등록(?:되었|됐|되었습니|됐습니)다\.?\s*</p>",
        "", source, count=1, flags=re.I | re.S,
    )
    source = re.sub(r"[❤♥💕💖💗💙💚💛🧡💜⭐🌟✨]+", "", source)
    source = re.sub(r"(?<![0-9A-Za-z가-힣])#[0-9A-Za-z가-힣_·~-]+", "", source)
    source = re.sub(r"(?:[·|]\s*)?관련\s*키워드\s*$", "", source, flags=re.I)
    source = re.sub(r"관련\s*키워드(?=\s*</(?:p|div|span|strong|b)>)", "", source, flags=re.I)
    source = source.replace("__MULTI_LEGAL_NOTICE__", "중개대상물 명시사항")
    source = re.sub(r"\s+", " ", source).strip()
    visible = clean(re.sub(r"<[^>]+>", " ", source))
    return source if len(visible) >= 80 else ""


def replace_html_table_value(source, label, value):
    if not source or not clean(value):
        return source
    pattern = re.compile(
        rf"(<tr\b[^>]*>\s*<(?:th|td)\b[^>]*>\s*{re.escape(label)}\s*</(?:th|td)>\s*<(?:th|td)\b[^>]*>)(.*?)(</(?:th|td)>)",
        flags=re.I | re.S,
    )
    return pattern.sub(lambda match: match.group(1) + html.escape(clean(value)) + match.group(3), source)


def build_legal_notice(row):
    esc = lambda value: html.escape(clean(value), quote=True)
    rows = []
    for label, value in (
        ("매물번호", row.get("article_no")),
        ("소재지", row.get("address_text") or row.get("region")),
        ("거래형태", row.get("trade_type")),
        ("가격", row.get("price_text")),
        ("중개대상물 종류", row.get("real_estate_type")),
        ("면적", row.get("area_info")),
        ("층", row.get("floor_info")),
        ("총 층수", row.get("total_floor")),
        ("방향", row.get("direction_code")),
        ("방 수/욕실 수", " / ".join(filter(None, (clean(row.get("room_count")), clean(row.get("bathroom_count")))))),
        ("관리비", row.get("maintenance_fee")),
        ("입주 가능일", row.get("move_in_date")),
        ("주차", row.get("parking_count")),
        ("사용승인일", row.get("approval_date")),
        ("건축물 용도", row.get("building_usage")),
    ):
        if clean(value):
            rows.append(f"<tr><th>{html.escape(label)}</th><td>{esc(value)}</td></tr>")
    if not rows:
        return ""
    kind = clean(row.get("real_estate_type")) or "매물"
    return f"<h2>중개대상물 명시사항 - {html.escape(kind)}</h2><table><tbody>{''.join(rows)}</tbody></table>"


def build_body(row, title, summary):
    prepared = clean_draft_html(row.get("blog_draft_html"))
    if prepared:
        prepared = replace_html_table_value(prepared, "관리비", row.get("maintenance_fee"))
        prepared = replace_html_table_value(prepared, "소재지", row.get("address_text") or row.get("region"))
        prepared = replace_html_table_value(prepared, "주소", row.get("address_text") or row.get("region"))
        if "중개대상물 명시사항" not in prepared:
            prepared += build_legal_notice(row)
        return prepared
    esc = lambda value: html.escape(clean(value), quote=True)
    facts = []
    for label, key in (
        ("매물종류", "real_estate_type"), ("거래유형", "trade_type"),
        ("가격", "price_text"), ("면적", "area_info"),
        ("층", "floor_info"), ("방향", "direction_code"),
        ("소재지", "address_text"),
    ):
        value = clean(row.get(key))
        if value:
            facts.append(f"<tr><th>{html.escape(label)}</th><td>{esc(value)}</td></tr>")

    description = str(row.get("article_feature_desc") or "").strip()
    description_lines = [clean(x) for x in re.split(r"[\r\n]+", description) if clean(x)]
    if len(description_lines) == 1:
        # 원문이 한 줄이어도 문장 단위로 읽기 좋게 나눈다.
        description_lines = [clean(x) for x in re.split(r"(?<=[.!?。]|[다요])\s+", description_lines[0]) if clean(x)]
    if description_lines:
        listing_text = "<br>".join(html.escape(x) for x in description_lines)
        paragraphs = f'<p class="listing-description">{listing_text}</p>'
    else:
        paragraphs = ""

    return (
        f"<p>{html.escape(summary)}</p>"
        f"<h2>{html.escape(title)} 핵심정보</h2>"
        f"<table><tbody>{''.join(facts)}</tbody></table>" if facts else f"<p>{html.escape(summary)}</p>"
    ) + (
        f"{('<h2>매물 안내</h2>' + paragraphs) if paragraphs else ''}{build_legal_notice(row)}"
    )


def extract_hash_tags(value):
    text = html.unescape(re.sub(r"<[^>]+>", " ", str(value or "")))
    return [clean(x) for x in re.findall(r"#([0-9A-Za-z가-힣_·~-]{2,100})", text)]


def build_tags(row):
    """검색 의도가 분명한 지역·단지·조건 키워드만 만든다."""
    complex_name = clean(row.get("complex_name"))
    building = clean(row.get("building_name"))
    property_name = clean(" ".join(x for x in (complex_name, building if building and building not in complex_name else "") if x)) or building
    estate = normalize_display_code(row.get("real_estate_type"), PROPERTY_TYPE_MAP)
    trade = normalize_display_code(row.get("trade_type"), TRADE_TYPE_MAP)
    address = clean(row.get("address_text")) or clean(row.get("region"))
    regions = []
    for token in re.split(r"\s+", address):
        token = token.strip(",()")
        if re.search(r"(?:특별시|광역시|특별자치시|도|시|군|구|읍|면|동|가|리)$", token) and token not in regions:
            regions.append(token)

    candidates = [complex_name, property_name]
    if complex_name and trade:
        candidates.append(f"{complex_name} {trade}")
    if complex_name and estate:
        candidates.append(f"{complex_name} {estate}")
    if regions and complex_name:
        candidates.append(f"{regions[-1]} {complex_name}")
    if regions and estate and trade:
        candidates.append(f"{regions[-1]} {estate} {trade}")
    if estate and trade:
        candidates.append(f"{estate} {trade}")

    feature_text = " ".join(clean(row.get(key)) for key in (
        "article_feature_desc", "blog_draft_title", "article_name",
        "direction_code", "floor_info",
    ))
    for feature in (
        "역세권", "초역세권", "학세권", "숲세권", "대단지", "남향", "남동향", "남서향",
        "로얄층", "고층", "저층", "신축", "리모델링", "풀옵션", "즉시입주", "입주가능",
        "주차가능", "조망권", "채광", "테라스",
    ):
        if feature in feature_text:
            candidates.append(feature)

    allowed_features = {
        "역세권", "초역세권", "학세권", "숲세권", "대단지", "남향", "남동향", "남서향",
        "로얄층", "고층", "저층", "신축", "리모델링", "풀옵션", "즉시입주", "입주가능",
        "주차가능", "조망권", "채광", "테라스",
    }
    candidates.extend(x for x in re.split(r"[,\n]+", clean(row.get("seo_keywords"))) if clean(x) in allowed_features)
    candidates.extend(x for x in extract_hash_tags(row.get("blog_draft_html")) if clean(x) in allowed_features)
    tags, normalized = [], set()
    blocked = {"네이버부동산후보매물", "네이버후보매물", "후보매물", "네이버부동산", "부동산매물"}
    for value in candidates:
        value = re.sub(r"^#+|#+$", "", clean(value)).strip()
        key = re.sub(r"[^0-9a-z가-힣㎡평]+", "", value.lower())
        if (not value or len(value) < 2 or len(value) > 45 or key in blocked
                or "후보매물" in key or key in normalized):
            continue
        tags.append(value)
        normalized.add(key)
        if len(tags) >= 10:
            break
    return tags


def extract_table_value(source, labels):
    source = str(source or "")
    wanted = {re.sub(r"\s+", "", label) for label in labels}
    for table_row in re.findall(r"<tr\b[^>]*>(.*?)</tr>", source, flags=re.I | re.S):
        cells = re.findall(r"<(?:th|td)\b[^>]*>(.*?)</(?:th|td)>", table_row, flags=re.I | re.S)
        if len(cells) < 2:
            continue
        label = re.sub(r"\s+", "", clean(html.unescape(re.sub(r"<[^>]+>", " ", cells[0]))))
        if label in wanted:
            value = clean(html.unescape(re.sub(r"<[^>]+>", " ", cells[1])))
            if value:
                return value
    return ""


def extract_legal_table_value(source, labels):
    """중개사 정보표가 아닌 중개대상물 법적 명시표에서만 값을 찾는다."""
    source = str(source or "")
    wanted = {re.sub(r"\s+", "", label) for label in labels}
    best_value = ""
    best_score = -1
    for table in re.findall(r"<table\b[^>]*>.*?</table>", source, flags=re.I | re.S):
        rows = re.findall(r"<tr\b[^>]*>(.*?)</tr>", table, flags=re.I | re.S)
        parsed = {}
        for table_row in rows:
            cells = re.findall(r"<(?:th|td)\b[^>]*>(.*?)</(?:th|td)>", table_row, flags=re.I | re.S)
            if len(cells) < 2:
                continue
            key = re.sub(r"\s+", "", clean(html.unescape(re.sub(r"<[^>]+>", " ", cells[0]))))
            value = clean(html.unescape(re.sub(r"<[^>]+>", " ", cells[1])))
            if key and value:
                parsed[key] = value
        score = sum(1 for marker in ("매물번호", "가격", "거래형태", "중개대상물종류", "면적", "방향") if marker in parsed)
        value = next((parsed[key] for key in wanted if key in parsed), "")
        if value and score > best_score:
            best_value, best_score = value, score
    return best_value if best_score >= 2 else ""


def location_address(row):
    draft_address = extract_legal_table_value(row.get("blog_draft_html"), ("소재지", "주소", "매물주소"))
    if draft_address and re.search(r"(?:로|길|동|리)\s*\d|산\s*\d", draft_address):
        return draft_address[:500], "legal_notice"
    raw = parse_json(row.get("raw_json"))
    for names in (("roadAddress", "roadAddressName"), ("jibunAddress", "addressName"), ("exposureAddress", "address")):
        value = clean(recursive_find(raw, names))
        if value and re.search(r"(?:로|길|동|리)\s*\d|산\s*\d", value):
            return value[:500], "source_json"
    address = clean(row.get("address_text"))
    if address and re.search(r"(?:로|길|동|리)\s*\d|산\s*\d", address):
        return address[:500], "source_json"
    complex_name = clean(row.get("complex_name"))
    if not complex_name or "후보 매물" in complex_name or "후보매물" in complex_name:
        complex_name = ""
    region_hint = clean(row.get("address_text")) or clean(row.get("region"))
    if region_hint:
        # '세종시 세종시 고운동'처럼 반복된 행정구역을 제거한다.
        unique_tokens = []
        seen_tokens = set()
        for token in region_hint.split():
            key = re.sub(r"\s+", "", token)
            if key and key not in seen_tokens:
                unique_tokens.append(token)
                seen_tokens.add(key)
        # 지도 단지 검색은 광역 지역명(+구/군)과 실제 단지명 조합이 가장 안정적이다.
        broad_tokens = [
            token for token in unique_tokens
            if re.search(r"(?:특별자치시|특별시|광역시|도|시|구|군)$", token)
        ]
        if broad_tokens:
            primary = broad_tokens[0]
            region_tokens = [primary]
            if not re.search(r"(?:특별자치시|특별시|광역시)$", primary):
                district = next((token for token in broad_tokens[1:] if re.search(r"(?:구|군)$", token)), "")
                if district:
                    region_tokens.append(district)
            region_hint = " ".join(region_tokens)
        else:
            region_hint = " ".join(unique_tokens[:1])
    fallback = " ".join(x for x in (region_hint, complex_name) if x) if complex_name else ""
    return fallback[:500], "region_building" if fallback else "missing"


def realtor_profile_url(row):
    raw = parse_json(row.get("raw_json"))
    value = clean(recursive_find(raw, (
        "_maemulhome_realtor_profile_image_url", "profileImageUrl", "profileFullImageUrl",
        "realtorImageUrl", "representativeImageUrl", "photoUrl",
    )))
    value = html.unescape(value)
    if value.startswith("//"):
        value = "https:" + value
    elif value.startswith("/"):
        value = "https://landthumb-phinf.pstatic.net" + value
    return value[:700] if value.startswith(("http://", "https://")) else ""


def kakao_get(path, params):
    proxy_url = clean(os.environ.get("MULTI_KAKAO_PROXY_URL")) or "http://multi.hongheemarketing.com/multi/api/kakao_local.php"
    payload = json.dumps({"path": path, "params": params}, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(proxy_url, data=payload, headers={
        "Content-Type": "application/json; charset=utf-8",
        "X-Multi-Proxy-Token": "vYCs9sRojdJzo3dBzuOokwjWs_9rZlDetOqP5rFsoUs",
        "User-Agent": "MaemulHOME-MultiWebzine/1.0",
    }, method="POST")
    with urllib.request.urlopen(request, timeout=12) as response:
        result = json.loads(response.read().decode("utf-8"))
    if not result.get("ok"):
        raise RuntimeError(clean(result.get("message")) or "kakao_proxy_failed")
    return result.get("data") or {}


def validate_region_building_result(query, complex_name, document):
    """지역+단지 검색 결과가 다른 단지를 가리키면 지도를 표시하지 않는다."""
    compact = lambda value: re.sub(r"[^0-9a-z가-힣]", "", clean(value).lower())
    expected = compact(complex_name)
    actual_name = compact(document.get("place_name"))
    if expected and actual_name:
        expected_core = expected.replace("아파트", "")
        actual_core = actual_name.replace("아파트", "")
        similarity = difflib.SequenceMatcher(None, expected_core, actual_core).ratio()
        if expected_core not in actual_core and actual_core not in expected_core and similarity < 0.45:
            return False

    result_address = clean(document.get("road_address_name")) or clean(document.get("address_name"))
    region_terms = re.findall(r"[가-힣0-9]+(?:구|동|읍|면|리)", clean(query))
    specific_terms = [term for term in region_terms if not re.fullmatch(r"\d{1,4}동", term)]
    if specific_terms and result_address and not any(term in result_address for term in specific_terms):
        return False
    return bool(result_address or actual_name)


def save_region_code(conn, article_id, longitude, latitude):
    """좌표의 법정동/행정동 코드를 별도 저장한다. 테이블 미적용 시 기사 생성을 막지 않는다."""
    try:
        data = kakao_get("/v2/local/geo/coord2regioncode.json", {
            "x": longitude, "y": latitude,
        })
        documents = data.get("documents") or []
        legal = next((item for item in documents if item.get("region_type") == "B"), {})
        admin = next((item for item in documents if item.get("region_type") == "H"), {})
        base = legal or admin
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO multi_article_region_codes
                (article_id,b_code,h_code,province_name,district_name,neighborhood_name,
                 normalized_region,latitude,longitude,collected_at,created_at,updated_at)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW(),NOW(),NOW())
                ON DUPLICATE KEY UPDATE b_code=VALUES(b_code),h_code=VALUES(h_code),
                 province_name=VALUES(province_name),district_name=VALUES(district_name),
                 neighborhood_name=VALUES(neighborhood_name),normalized_region=VALUES(normalized_region),
                 latitude=VALUES(latitude),longitude=VALUES(longitude),collected_at=NOW(),updated_at=NOW()
            """, (
                article_id, clean(legal.get("code"))[:20] or None, clean(admin.get("code"))[:20] or None,
                clean(base.get("region_1depth_name"))[:80] or None,
                clean(base.get("region_2depth_name"))[:100] or None,
                clean(base.get("region_3depth_name"))[:100] or None,
                " ".join(filter(None, [clean(base.get("region_1depth_name")), clean(base.get("region_2depth_name")), clean(base.get("region_3depth_name"))]))[:255] or None,
                latitude, longitude,
            ))
        conn.commit()
    except Exception as exc:
        conn.rollback()
        print(f"[MULTI REGION CODE WARNING] article_id={article_id} error={exc}")


def backfill_region_codes(conn, realtor_id=None, limit=300):
    """기사 생성 대상이 0건이어도 기존 좌표의 법정동 코드를 순차 보완한다."""
    sql = """
        SELECT ma.id AS article_id,l.longitude,l.latitude
          FROM multi_articles ma
          INNER JOIN multi_article_locations l ON l.article_id=ma.id AND l.geocode_status='success'
          LEFT JOIN multi_article_region_codes rc ON rc.article_id=ma.id
         WHERE ma.status='published' AND rc.id IS NULL
    """
    params = []
    if realtor_id:
        sql += " AND ma.realtor_id=%s"
        params.append(int(realtor_id))
    sql += " ORDER BY ma.published_at DESC,ma.id DESC LIMIT %s"
    params.append(max(1, int(limit)))
    try:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            rows = cur.fetchall() or []
    except Exception as exc:
        conn.rollback()
        print(f"[MULTI REGION CODE SKIP] {exc}")
        return 0
    for row in rows:
        save_region_code(conn, int(row["article_id"]), str(row["longitude"]), str(row["latitude"]))
    if rows:
        print(f"[MULTI REGION CODE BACKFILL] count={len(rows)}")
    return len(rows)


def save_location(conn, article_id, row):
    address, address_source = location_address(row)
    if not address:
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    UPDATE multi_article_locations
                       SET geocode_status='failed', error_message='safe_address_missing', updated_at=NOW()
                     WHERE article_id=%s
                """, (article_id,))
            conn.commit()
        except Exception:
            conn.rollback()
        return
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM multi_article_locations WHERE lookup_address=%s AND geocode_status='success' ORDER BY id DESC LIMIT 1", (address,))
            cached = cur.fetchone()
        if cached and address_source != "region_building":
            longitude, latitude = str(cached["longitude"]), str(cached["latitude"])
            normalized = clean(cached.get("normalized_address")) or address
        else:
            if address_source == "region_building":
                cached = None
            if address_source == "region_building":
                result = kakao_get("/v2/local/search/keyword.json", {"query": address, "size": 1})
            else:
                result = kakao_get("/v2/local/search/address.json", {"query": address, "size": 1})
            documents = result.get("documents") or []
            if not documents:
                raise RuntimeError("address_not_found")
            doc = documents[0]
            if address_source == "region_building" and not validate_region_building_result(address, row.get("complex_name"), doc):
                raise RuntimeError("region_building_result_mismatch")
            longitude, latitude = str(doc.get("x") or ""), str(doc.get("y") or "")
            normalized = clean(doc.get("road_address_name")) or clean(doc.get("address_name")) or address
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO multi_article_locations
                (article_id,lookup_address,address_source,normalized_address,latitude,longitude,geocode_status,collected_at,created_at,updated_at)
                VALUES (%s,%s,%s,%s,%s,%s,'success',NOW(),NOW(),NOW())
                ON DUPLICATE KEY UPDATE lookup_address=VALUES(lookup_address),address_source=VALUES(address_source),normalized_address=VALUES(normalized_address),latitude=VALUES(latitude),longitude=VALUES(longitude),geocode_status='success',error_message=NULL,collected_at=NOW(),updated_at=NOW()
            """, (article_id, address, address_source, normalized, latitude, longitude))
            cached_article_id = int(cached.get("article_id") or 0) if cached else 0
            if not cached or cached_article_id != int(article_id):
                cur.execute("DELETE FROM multi_article_nearby_places WHERE article_id=%s", (article_id,))
            if cached and cached_article_id != int(article_id):
                cur.execute("""
                    INSERT IGNORE INTO multi_article_nearby_places
                    (article_id,place_group,place_id,place_name,category_name,address_name,road_address_name,distance_m,place_url,latitude,longitude,created_at)
                    SELECT %s,place_group,place_id,place_name,category_name,address_name,road_address_name,distance_m,place_url,latitude,longitude,NOW()
                    FROM multi_article_nearby_places WHERE article_id=%s
                """, (article_id, cached_article_id))
        if cached:
            conn.commit()
            save_region_code(conn, article_id, longitude, latitude)
            print(f"[MULTI LOCATION CACHED] article_id={article_id} source={address_source} address={normalized}")
            return
        groups = (("school", "SC4"), ("subway", "SW8"), ("hospital", "HP8"), ("mart", "MT1"))
        for group_name, category in groups:
            data = kakao_get("/v2/local/search/category.json", {"category_group_code": category, "x": longitude, "y": latitude, "radius": 2000, "sort": "distance", "size": 5})
            with conn.cursor() as cur:
                for place in data.get("documents") or []:
                    cur.execute("""
                        INSERT IGNORE INTO multi_article_nearby_places
                        (article_id,place_group,place_id,place_name,category_name,address_name,road_address_name,distance_m,place_url,latitude,longitude,created_at)
                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW())
                    """, (article_id, group_name, clean(place.get("id")), clean(place.get("place_name"))[:255], clean(place.get("category_name"))[:500], clean(place.get("address_name"))[:500], clean(place.get("road_address_name"))[:500], int(place.get("distance") or 0), clean(place.get("place_url"))[:700], clean(place.get("y")), clean(place.get("x"))))
        conn.commit()
        save_region_code(conn, article_id, longitude, latitude)
        print(f"[MULTI LOCATION SAVED] article_id={article_id} source={address_source} address={normalized}")
    except Exception as exc:
        conn.rollback()
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO multi_article_locations
                    (article_id,lookup_address,address_source,geocode_status,error_message,created_at,updated_at)
                    VALUES (%s,%s,%s,'failed',%s,NOW(),NOW())
                    ON DUPLICATE KEY UPDATE lookup_address=VALUES(lookup_address),address_source=VALUES(address_source),geocode_status='failed',error_message=VALUES(error_message),updated_at=NOW()
                """, (article_id, address, address_source, str(exc)[:1000]))
            conn.commit()
        except Exception:
            conn.rollback()
        print(f"[MULTI LOCATION WARNING] article_id={article_id} error={exc}")


def image_url(row):
    for key in ("public_url", "image_url", "original_url", "local_file_path", "local_path"):
        value = clean(row.get(key))
        if value.startswith("http://") or value.startswith("https://"):
            return value
    return ""


def embedded_image_rows(row):
    """매물HOME 초안 HTML과 네이버 상세 JSON에서도 원본 이미지 URL을 복원한다."""
    found = []
    html_source = str(row.get("blog_draft_html") or "")
    for index, match in enumerate(re.finditer(r"<img\b[^>]*?src=[\"']([^\"']+)[\"'][^>]*>", html_source, flags=re.I | re.S)):
        tag = match.group(0)
        found.append({
            "image_url": html.unescape(match.group(1)),
            "image_type": "draft_html " + clean(tag),
            "sort_order": 1000 + index,
            "image_source": "draft_html",
        })

    def walk(value, path=""):
        if isinstance(value, dict):
            for key, child in value.items():
                walk(child, f"{path} {key}".strip())
        elif isinstance(value, list):
            for index, child in enumerate(value):
                walk(child, f"{path} {index}".strip())
        elif isinstance(value, str):
            decoded = html.unescape(value).strip()
            if decoded.startswith("//"):
                decoded = "https:" + decoded
            if not decoded.startswith(("http://", "https://")):
                return
            lower = decoded.lower().split("?", 1)[0]
            image_like = bool(re.search(r"\.(?:jpe?g|png|webp|gif|avif)$", lower)) or any(
                host in lower for host in ("pstatic.net", "cafe24img.com", "naver.net")
            )
            if image_like:
                found.append({
                    "image_url": decoded,
                    "image_type": path,
                    "sort_order": 2000 + len(found),
                    "image_source": "source_json",
                })

    for key in ("raw_json", "article_draft_source_json", "realestate_draft_source_json"):
        walk(parse_json(row.get(key)), key)
    return found


def fetch_images(conn, article_no, article_row=None):
    with conn.cursor() as cur:
        cur.execute("""
            SELECT *
            FROM blog_realestate_article_images
            WHERE article_no = %s
            ORDER BY sort_order ASC, id ASC
        """, (str(article_no),))
        rows = cur.fetchall() or []
        try:
            cur.execute("""
                SELECT image_id AS source_image_id, article_no, image_url, local_path,
                       image_type, sort_order, 'blog_draft' AS image_source
                FROM blog_realestate_blog_draft_images
                WHERE BINARY article_no = BINARY %s
                  AND COALESCE(is_used, 1) = 1
                ORDER BY sort_order ASC, id ASC
            """, (str(article_no),))
            draft_rows = cur.fetchall() or []
        except Exception:
            draft_rows = []
        try:
            cur.execute("""
                SELECT NULL AS source_image_id,mai.image_url,mai.image_role AS image_type,
                       mai.sort_order,'existing_webzine' AS image_source
                  FROM multi_article_images mai
                  INNER JOIN multi_articles ma ON ma.id=mai.article_id
                 WHERE BINARY ma.source_article_no=BINARY %s
                 ORDER BY mai.sort_order,mai.id
            """, (str(article_no),))
            existing_rows = cur.fetchall() or []
        except Exception:
            existing_rows = []
    for row in rows:
        row["source_image_id"] = row.get("id")
    rows.extend(draft_rows)
    rows.extend(existing_rows)
    if article_row:
        rows.extend(embedded_image_rows(article_row))
    output = []
    seen_urls = set()
    for row in rows:
        url = image_url(row)
        text = " ".join(clean(row.get(k)).lower() for k in (
            "image_type", "image_category", "file_name", "local_path",
            "local_file_path", "public_url", "image_url", "image_source", "raw_json",
        ))
        excluded = (
            "profile", "realtor", "banner", "map", "header.png",
            "/header_images/", "composite", "generated_header",
            "대표이미지", "합성",
        )
        normalized_url = html.unescape(url).strip()
        if not normalized_url or normalized_url in seen_urls or any(x in text for x in excluded):
            continue
        seen_urls.add(normalized_url)
        item = dict(row)
        item["resolved_url"] = normalized_url
        if any(x in text for x in ("floorplan", "floor_plan", "평면", "도면")):
            item["webzine_priority"] = 2
            item["webzine_section"] = "floorplan"
        elif any(x in text for x in ("outside", "external", "building", "exterior", "complex", "외부", "건물", "전경", "단지", "입지")):
            item["webzine_priority"] = 0
            item["webzine_section"] = "location_reference"
        elif any(x in text for x in ("view", "inside", "interior", "living", "kitchen", "내부", "거실", "주방", "조망", "방", "욕실")):
            item["webzine_priority"] = 0
            item["webzine_section"] = "property_photo"
        else:
            item["webzine_priority"] = 1
            item["webzine_section"] = "property_photo"
        output.append(item)
    priority_order = {0: 0, 2: 1, 1: 2}  # 외부사진 → 평면도 → 기타사진
    output.sort(key=lambda x: (priority_order.get(x.get("webzine_priority", 1), 2), int(x.get("sort_order") or 0), int(x.get("id") or 0)))
    return output[:50]


def table_columns(conn, table_name):
    """운영 DB 버전별 컬럼 차이를 안전하게 흡수한다."""
    allowed = {"blog_article_drafts", "blog_realestate_blog_drafts", "blog_complex_address_cache"}
    if table_name not in allowed:
        return set()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SHOW COLUMNS FROM `{table_name}`")
            columns = set()
            for row in (cur.fetchall() or []):
                value = row.get("Field") if isinstance(row, dict) else (row[0] if row else "")
                if clean(value):
                    columns.add(clean(value))
            return columns
    except Exception:
        return set()


def draft_subquery(columns, table_name, alias, column, realtor_condition=True):
    if column not in columns:
        return "NULL"
    realtor_sql = f" AND {alias}.realtor_id = a.realtor_id" if realtor_condition and "realtor_id" in columns else ""
    return (
        f"(SELECT NULLIF(TRIM({alias}.`{column}`), '') FROM `{table_name}` {alias} "
        f"WHERE BINARY {alias}.article_no = BINARY a.article_no{realtor_sql} "
        f"ORDER BY {alias}.id DESC LIMIT 1)"
    )


def fetch_targets(conn, realtor_id=None, limit=1, article_no=None, include_existing=False, multi_article_id=None, all_current=False):
    article_columns = table_columns(conn, "blog_article_drafts")
    realestate_columns = table_columns(conn, "blog_realestate_blog_drafts")
    cache_columns = table_columns(conn, "blog_complex_address_cache")
    article_value = lambda column: draft_subquery(article_columns, "blog_article_drafts", "d", column)
    realestate_value = lambda column: draft_subquery(realestate_columns, "blog_realestate_blog_drafts", "bd", column)
    if {"complex_no", "complex_name"}.issubset(cache_columns):
        cache_order = "cac.last_checked_at DESC, cac.id DESC" if "last_checked_at" in cache_columns else "cac.id DESC"
        cached_complex_name = f"""
            (SELECT NULLIF(TRIM(cac.complex_name), '')
               FROM blog_complex_address_cache cac
              WHERE BINARY cac.complex_no = BINARY a.complex_no
              ORDER BY {cache_order} LIMIT 1)
        """
    else:
        cached_complex_name = "NULL"
    sql = f"""
        SELECT a.*, r.office_name, r.representative_name, r.office_phone,
               r.mobile_phone, r.phone, r.mobile, r.region, r.address,
               r.office_address, r.seo_keywords,
               s.id AS site_id, s.slug AS site_slug, s.site_title,
               s.is_shorts_enabled,
               {cached_complex_name} AS cached_complex_name,
               COALESCE(
                 {article_value('draft_title')},
                 {realestate_value('draft_title')},
                 (SELECT q.publish_title
                    FROM blog_publish_queue q
                   WHERE q.realtor_id = a.realtor_id
                     AND BINARY q.article_no = BINARY a.article_no
                     AND NULLIF(TRIM(q.publish_title), '') IS NOT NULL
                   ORDER BY q.id DESC LIMIT 1)
               ) AS blog_draft_title,
               {article_value('draft_html')} AS article_draft_html,
               {article_value('clipboard_html')} AS article_draft_clipboard_html,
               {article_value('plain_text')} AS article_draft_plain_text,
               {article_value('content_html')} AS article_draft_content_html,
               {article_value('body_html')} AS article_draft_body_html,
               {article_value('source_json')} AS article_draft_source_json,
               {realestate_value('draft_html')} AS realestate_draft_html,
               {realestate_value('clipboard_html')} AS realestate_draft_clipboard_html,
               {realestate_value('plain_text')} AS realestate_draft_plain_text,
               {realestate_value('content_html')} AS realestate_draft_content_html,
               {realestate_value('body_html')} AS realestate_draft_body_html,
               {realestate_value('source_json')} AS realestate_draft_source_json
        FROM blog_realtor_articles a
        INNER JOIN blog_realtors r ON r.id = a.realtor_id
        INNER JOIN multi_sites s ON s.realtor_id = a.realtor_id AND s.status = 'active'
        WHERE r.status = 'active'
          AND COALESCE(r.is_deleted, 0) = 0
          AND COALESCE(a.article_status, 'active') NOT IN ('removed', 'closed')
          AND COALESCE(a.detail_collected, 0) = 1
    """
    params = []
    if realtor_id:
        sql += " AND a.realtor_id = %s"
        params.append(int(realtor_id))
    if article_no:
        sql += " AND BINARY a.article_no = BINARY %s"
        params.append(str(article_no))
    if multi_article_id:
        sql += """
            AND EXISTS (
                SELECT 1 FROM multi_articles selected_ma
                WHERE selected_ma.id = %s
                  AND selected_ma.site_id = s.id
                  AND BINARY selected_ma.source_article_no = BINARY a.article_no
            )
        """
        params.append(int(multi_article_id))
    if not include_existing:
        sql += """
          AND NOT EXISTS (
              SELECT 1 FROM multi_articles ma
              WHERE ma.site_id = s.id
                AND BINARY ma.source_article_no = BINARY a.article_no
          )
        """
    sql += " ORDER BY COALESCE(a.first_posted_at, a.detail_collected_at, a.created_at) DESC, a.id DESC"
    if not all_current:
        sql += " LIMIT %s"
        params.append(max(1, min(1000, int(limit))))
    with conn.cursor() as cur:
        cur.execute(sql, params)
        return cur.fetchall() or []


def resolve_realtor_and_site(conn, realtor_id=None, group_id=None, initialize_site=False):
    """단체아이디로 중개사를 찾고 필요하면 같은 주소 별칭의 웹진을 만든다."""
    group_id = clean(group_id)
    with conn.cursor() as cur:
        if group_id:
            cur.execute("""
                SELECT id,office_name,group_id FROM blog_realtors
                 WHERE TRIM(group_id)=%s AND status='active' AND COALESCE(is_deleted,0)=0
                 ORDER BY id
            """, (group_id,))
            matches = cur.fetchall() or []
            if not matches:
                raise RuntimeError(f"단체아이디 {group_id}에 해당하는 활성 중개사가 없습니다.")
            if len(matches) > 1:
                ids = ",".join(str(row["id"]) for row in matches)
                raise RuntimeError(f"단체아이디 {group_id}가 중복되어 있습니다. realtor_id={ids}")
            resolved = matches[0]
            if realtor_id and int(realtor_id) != int(resolved["id"]):
                raise RuntimeError("--realtor-id와 --group-id가 서로 다른 중개사를 가리킵니다.")
            realtor_id = int(resolved["id"])
        elif realtor_id:
            cur.execute("""
                SELECT id,office_name,group_id FROM blog_realtors
                 WHERE id=%s AND status='active' AND COALESCE(is_deleted,0)=0 LIMIT 1
            """, (int(realtor_id),))
            resolved = cur.fetchone()
            if not resolved:
                raise RuntimeError(f"realtor_id={realtor_id} 활성 중개사를 찾을 수 없습니다.")
            group_id = clean(resolved.get("group_id"))
        else:
            return None, None

        if initialize_site:
            if not group_id or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,99}", group_id):
                raise RuntimeError("단체아이디가 없거나 웹진 주소로 사용할 수 없는 형식입니다.")
            cur.execute("SELECT id,realtor_id FROM multi_sites WHERE slug=%s LIMIT 1", (group_id,))
            slug_site = cur.fetchone()
            if slug_site and int(slug_site["realtor_id"]) != int(realtor_id):
                raise RuntimeError(f"웹진 주소 {group_id}가 다른 중개사에 사용 중입니다.")
            cur.execute("SELECT id FROM multi_sites WHERE realtor_id=%s LIMIT 1", (realtor_id,))
            realtor_site = cur.fetchone()
            if realtor_site:
                site_id = realtor_site["id"]
                cur.execute("UPDATE multi_sites SET slug=%s,status='active',updated_at=NOW() WHERE id=%s", (group_id,site_id))
            else:
                office_name = clean(resolved.get("office_name")) or f"중개사 {realtor_id}"
                cur.execute("""
                    INSERT INTO multi_sites
                    (realtor_id,slug,site_title,description,template_code,primary_color,
                     secondary_color,is_shorts_enabled,status,published_at,created_at,updated_at)
                    VALUES (%s,%s,%s,%s,'news_default','#123a70','#f2f5f9',0,'active',NOW(),NOW(),NOW())
                """, (realtor_id,group_id,f"{office_name} 웹진",f"{office_name}의 최신 부동산 매물과 지역 소식입니다."))
                site_id = cur.lastrowid
            try:
                cur.execute("""
                    INSERT INTO multi_admin_users
                    (site_id,username,password_hash,password_algo,must_change_password,status,created_at,updated_at)
                    VALUES (%s,%s,SHA2(CONCAT(%s,'1234'),256),'sha256_initial',1,'active',NOW(),NOW())
                    ON DUPLICATE KEY UPDATE username=VALUES(username),status='active',updated_at=NOW()
                """, (site_id,group_id,group_id))
            except Exception as exc:
                print(f"[MULTI ADMIN WARNING] 관리자 계정 생성 생략: {exc}")
        conn.commit()
    print(f"[MULTI REALTOR] realtor_id={realtor_id} group_id={group_id}")
    return int(realtor_id), group_id


def seo_score(row, images):
    checks = [
        clean(row.get("building_name")) or clean(row.get("article_name")),
        clean(row.get("real_estate_type")), clean(row.get("trade_type")),
        clean(row.get("price_text")), clean(row.get("area_info")),
        clean(row.get("article_feature_desc")), bool(images),
    ]
    return int(sum(bool(x) for x in checks) / len(checks) * 100)


def parse_naver_ymd(value):
    text = str(value or "").strip()
    for fmt in ("%Y%m%d", "%Y-%m-%d", "%Y.%m.%d"):
        try:
            return datetime.strptime(text[:10] if fmt != "%Y%m%d" else text[:8], fmt)
        except ValueError:
            continue
    return None


def source_published_at(row):
    """네이버 등록일을 우선 사용하고, 없으면 최초 수집일로 기사화한다."""
    if row.get("first_posted_at"):
        return row["first_posted_at"]
    raw = row.get("raw_json")
    try:
        payload = json.loads(raw) if isinstance(raw, str) else raw
    except (TypeError, ValueError):
        payload = None
    if isinstance(payload, dict):
        sections = [payload.get("articleDetail"), payload.get("articleAddition"), payload]
        for key in ("articleConfirmYMD", "exposeStartYMD"):
            for section in sections:
                if isinstance(section, dict):
                    parsed = parse_naver_ymd(section.get(key))
                    if parsed:
                        return parsed
    # 목록에 존재하는 현재 매물은 상세 응답이나 등록일이 없어도 기사화한다.
    # created_at은 이 매물이 시스템에 최초로 들어온 시점이므로 정렬 기준으로
    # 사용할 수 있고, 이 값마저 없는 예외 행만 현재 시각을 사용한다.
    return row.get("created_at") or row.get("updated_at") or datetime.now()


def reselect_headline(conn, site_id):
    with conn.cursor() as cur:
        # 중개사가 직접 고정한 헤드라인은 기사 갱신 때 자동 선택으로 덮어쓰지 않는다.
        try:
            cur.execute("SELECT headline_mode,manual_headline_article_id FROM multi_sites WHERE id=%s", (site_id,))
            setting = cur.fetchone() or {}
            if setting.get("headline_mode") == "manual" and setting.get("manual_headline_article_id"):
                cur.execute("SELECT id FROM multi_articles WHERE id=%s AND site_id=%s AND status='published'", (setting["manual_headline_article_id"], site_id))
                if cur.fetchone():
                    cur.execute("UPDATE multi_articles SET is_headline=IF(id=%s,1,0) WHERE site_id=%s", (setting["manual_headline_article_id"], site_id))
                    return
        except Exception:
            # 마이그레이션 전 DB에서는 기존 자동 선택을 유지한다.
            pass
        cur.execute("UPDATE multi_articles SET is_headline=0 WHERE site_id=%s", (site_id,))
        cur.execute("""
            UPDATE multi_articles
               SET is_headline=1
             WHERE id = (
                SELECT picked.id FROM (
                    SELECT ma.id FROM multi_articles ma
                     WHERE ma.site_id=%s AND ma.status='published' AND ma.headline_eligible=1
                     ORDER BY CASE
                                WHEN NULLIF(TRIM(ma.primary_image_url),'') IS NOT NULL
                                 AND EXISTS (
                                     SELECT 1 FROM blog_realtor_articles bra
                                      WHERE bra.realtor_id=ma.realtor_id
                                        AND BINARY bra.article_no=BINARY ma.source_article_no
                                        AND NULLIF(TRIM(bra.article_feature_desc),'') IS NOT NULL
                                 ) THEN 0
                                ELSE 1
                              END,
                              ma.published_at DESC, ma.id DESC LIMIT 1
                ) picked
             )
        """, (site_id,))


def reconcile_inactive_articles(conn, realtor_id=None):
    """원본 매물이 종료/삭제된 경우 웹진 기사도 공개 목록에서 제외한다.

    숏츠 비공개 연동은 별도 작업에서 처리하며 여기서는 웹진 기사 상태만 바꾼다.
    """
    sql = """
        UPDATE multi_articles ma
        INNER JOIN blog_realtor_articles a
          ON a.realtor_id=ma.realtor_id
         AND BINARY a.article_no=BINARY ma.source_article_no
           SET ma.status='archived', ma.is_headline=0, ma.updated_at=NOW()
         WHERE ma.status='published'
           AND COALESCE(a.article_status,'active') IN ('removed','closed')
    """
    params = []
    if realtor_id:
        sql += " AND ma.realtor_id=%s"
        params.append(int(realtor_id))
    with conn.cursor() as cur:
        affected = cur.execute(sql, params)
    conn.commit()
    if affected:
        print(f"[MULTI INACTIVE RECONCILED] count={affected}")
    return int(affected or 0)


def reconcile_unready_articles(conn, realtor_id=None):
    """상세수집 대기 기사는 공개 상태를 유지한다.

    상세수집은 하루 회차별로 최대 20건씩 진행되므로 ``detail_collected=0``은
    종료 매물이 아니라 보강 대기 상태이다. 이를 archived 처리하면 회차가 끝날
    때마다 기사 수가 감소하므로 더 이상 공개 상태를 변경하지 않는다.
    실제 종료·삭제 매물은 ``reconcile_inactive_articles()``에서만 처리한다.
    """
    print(
        f"[MULTI UNREADY KEEP] realtor_id={realtor_id or 'all'} "
        "action=keep_published_until_detail_sync"
    )
    return 0


def save_article(conn, row, dry_run=False, refresh_existing=False):
    row = enriched(row)
    article_no = clean(row.get("article_no"))
    title = build_title(row)
    summary = build_summary(row)
    body_html = build_body(row, title, summary)
    images = fetch_images(conn, article_no, row)
    article_category = category_code(row)
    body_text = clean(html.unescape(re.sub(r"<[^>]+>", " ", body_html)))
    quality_ready = bool(
        article_category != "기타"
        and len(title) >= 8
        and len(body_text) >= 180
        and images
    )
    # 네이버 현재매물은 사진·설명 품질과 관계없이 기사로 공개한다.
    # quality_ready는 헤드라인/추천 선정 품질 판단에만 사용한다.
    publish_status = "published"
    score = seo_score(row, images)
    headline_eligible = int(
        quality_ready
        and
        bool(clean(row.get("building_name")) or clean(row.get("article_name")))
        and bool(clean(row.get("trade_type")))
        and bool(clean(row.get("price_text")))
        and clean(row.get("article_status")).lower() not in {"removed", "closed"}
    )
    tags = build_tags(row)
    slug = slugify(title, article_no)
    # 메인·목록 대표이미지는 외관/전경/내부/조망 실사진만 허용한다.
    # 평면도는 상세기사 갤러리에서만 사용하고, 실사진이 없으면 텍스트 기사로 표시한다.
    primary_candidate = next((item for item in images if item.get("webzine_priority") == 0), None)
    if not primary_candidate:
        primary_candidate = next((item for item in images if item.get("webzine_priority") != 2), None)
    primary_image = primary_candidate["resolved_url"] if primary_candidate else ""
    seo_title = f"{title}｜{clean(row.get('site_title'))}"[:300]
    # 상세기사 제목 아래에는 중개사 매물설명을 반복하지 않고 별도 조건 요약을 사용한다.
    seo_description = build_natural_summary(row)[:500]
    seo_keywords = ", ".join(tags)
    published_at = source_published_at(row)
    profile_url = realtor_profile_url(row)

    print(f"[MULTI ARTICLE] realtor_id={row.get('realtor_id')} article_no={article_no} score={score} status={publish_status} images={len(images)} headline={headline_eligible}")
    if dry_run:
        return None

    try:
        with conn.cursor() as cur:
            # 다음 실행부터도 동일한 네이버 등록일을 사용하도록 원천 필드를 보정한다.
            cur.execute(
                """UPDATE blog_realtor_articles
                      SET first_posted_at=%s,updated_at=updated_at
                    WHERE realtor_id=%s AND BINARY article_no=BINARY %s""",
                (published_at, row["realtor_id"], article_no),
            )
            if profile_url:
                cur.execute("UPDATE multi_sites SET profile_image_url=COALESCE(NULLIF(profile_image_url,''),%s),updated_at=NOW() WHERE id=%s", (profile_url, row["site_id"]))
            existing_id = None
            if refresh_existing:
                cur.execute("""
                    SELECT id FROM multi_articles
                    WHERE site_id=%s AND BINARY source_article_no=BINARY %s
                    LIMIT 1
                """, (row["site_id"], article_no))
                existing = cur.fetchone()
                existing_id = int(existing["id"]) if existing else None

            if existing_id:
                cur.execute("""
                    UPDATE multi_articles
                    SET category_code=%s, title=%s, summary=%s,
                        body_html=%s, seo_title=%s, seo_description=%s,
                        seo_keywords=%s, primary_image_url=%s, status=%s,
                        is_headline=IF(%s=1, is_headline, 0),
                        headline_eligible=%s, seo_quality_score=%s,
                        search_status='crawl_pending', source_updated_at=%s, published_at=%s,
                        updated_at=NOW()
                    WHERE id=%s
                """, (
                    article_category, title, summary, body_html,
                    seo_title, seo_description, seo_keywords, primary_image or None,
                    publish_status, headline_eligible, headline_eligible, score, row.get("updated_at"), published_at, existing_id,
                ))
                multi_article_id = existing_id
                cur.execute("DELETE FROM multi_article_images WHERE article_id=%s", (multi_article_id,))
                cur.execute("DELETE FROM multi_article_tag_map WHERE article_id=%s", (multi_article_id,))
            else:
                cur.execute("""
                INSERT INTO multi_articles
                (site_id, realtor_id, source_article_no, article_slug, category_code,
                 title, summary, body_html, seo_title, seo_description, seo_keywords,
                 primary_image_url, status, is_headline, headline_eligible,
                 seo_quality_score, search_status, source_updated_at,
                 published_at, created_at, updated_at)
                VALUES
                (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,
                 %s,0,%s,%s,'crawl_pending',%s,%s,NOW(),NOW())
                """, (
                    row["site_id"], row["realtor_id"], article_no, slug, article_category,
                    title, summary, body_html, seo_title, seo_description, seo_keywords,
                    primary_image or None, publish_status, headline_eligible, score, row.get("updated_at"), published_at,
                ))
                multi_article_id = cur.lastrowid

            for index, image_row in enumerate(images):
                if image_row.get("webzine_priority") == 2:
                    role = "floorplan"
                elif primary_image and image_row["resolved_url"] == primary_image:
                    role = image_row.get("webzine_section") or "headline"
                elif image_row.get("webzine_section"):
                    role = image_row.get("webzine_section")
                else:
                    role = clean(image_row.get("image_type")) or "gallery"
                complex_name = clean(row.get('complex_name'))
                building_name = clean(row.get('building_name'))
                image_subject = complex_name or building_name or clean(row.get('article_name')) or clean(row.get('address_text')) or '부동산 매물'
                if complex_name and building_name and building_name not in complex_name:
                    image_subject = f"{complex_name} {building_name}"
                role_label = {
                    'property_photo': '매물 내부',
                    'headline': '대표 매물',
                    'gallery': '매물',
                    'floorplan': '평면도',
                    'location_reference': '단지 및 주변환경',
                    'generated_header': '대표 매물',
                }.get(role, role)
                image_context = " ".join(filter(None, [
                    image_subject,
                    normalize_display_code(row.get('real_estate_type'), PROPERTY_TYPE_MAP),
                    normalize_display_code(row.get('trade_type'), TRADE_TYPE_MAP),
                    f"{role_label} 이미지",
                    f"{index + 1}번",
                ]))
                alt_text = re.sub(r"\s+", " ", image_context).strip()
                cur.execute("""
                    INSERT INTO multi_article_images
                    (article_id, source_image_id, image_url, image_role, alt_text, sort_order, created_at)
                    VALUES (%s,%s,%s,%s,%s,%s,NOW())
                """, (multi_article_id, image_row.get("source_image_id"), image_row["resolved_url"], role[:30], alt_text[:300], index))

            for tag_name in tags:
                tag_slug = re.sub(r"[^0-9a-z가-힣]+", "-", tag_name.lower()).strip("-")[:120]
                if not tag_slug:
                    tag_slug = hashlib.sha1(tag_name.encode("utf-8")).hexdigest()[:20]
                cur.execute("""
                    INSERT INTO multi_tags (tag_name, tag_slug, created_at)
                    VALUES (%s,%s,NOW())
                    ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id), tag_name=VALUES(tag_name)
                """, (tag_name[:100], tag_slug))
                tag_id = cur.lastrowid
                cur.execute("INSERT IGNORE INTO multi_article_tag_map (article_id, tag_id, created_at) VALUES (%s,%s,NOW())", (multi_article_id, tag_id))

            reselect_headline(conn, row["site_id"])

            article_path = f"/{clean(row.get('site_slug'))}/article/{multi_article_id}"
            if publish_status == "published":
                if existing_id:
                    cur.execute("""
                        UPDATE multi_search_submission_jobs
                        SET queue_status='pending', retry_count=0, error_message=NULL, updated_at=NOW()
                        WHERE article_id=%s
                    """, (multi_article_id,))
                for engine in ("naver", "google", "daum", "bing"):
                    cur.execute("""
                        INSERT IGNORE INTO multi_search_submission_jobs
                        (article_id, site_id, realtor_id, search_engine, article_url,
                         queue_status, retry_count, created_at, updated_at)
                        VALUES (%s,%s,%s,%s,%s,'pending',0,NOW(),NOW())
                    """, (multi_article_id, row["site_id"], row["realtor_id"], engine, article_path))
            elif existing_id:
                cur.execute("""
                    UPDATE multi_search_submission_jobs
                       SET queue_status='cancelled',error_message='quality_not_ready',updated_at=NOW()
                     WHERE article_id=%s AND queue_status IN ('pending','retry_wait')
                """, (multi_article_id,))

        conn.commit()
        save_location(conn, multi_article_id, row)
        action = "REFRESHED" if existing_id else "SAVED"
        print(f"[MULTI ARTICLE {action}] id={multi_article_id} status={publish_status} images={len(images)} tags={len(tags)}")
        return multi_article_id
    except Exception:
        conn.rollback()
        raise


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--realtor-id", type=int)
    parser.add_argument("--group-id", help="중개사 단체아이디(웹진 주소 별칭으로 사용)")
    parser.add_argument("--article-no")
    parser.add_argument("--multi-article-id", type=int)
    parser.add_argument("--limit", type=int, default=1)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--refresh-existing", action="store_true")
    parser.add_argument(
        "--sync-all-current", action="store_true",
        help="현재 활성 매물 전체를 신규 생성/갱신하고 종료 매물을 웹진에서 제외",
    )
    parser.add_argument(
        "--skip-policy-feeds", action="store_true",
        help="정책·생활정보 수집을 생략(중개사별 전체 갱신 시 API 중복 호출 방지)",
    )
    parser.add_argument("--initialize-site", action="store_true", help="해당 중개사의 현재 전체 매물을 최초 기사화")
    parser.add_argument("--initial-max", type=int, default=60, help="홈페이지 최초 생성 시 상세 완료 기사 최대 건수")
    parser.add_argument("--build-job-id", type=int, help="관리자에서 등록한 최초 기사 생성 작업 ID")
    args = parser.parse_args()

    conn = get_conn()
    success = failed = 0
    try:
        if args.initialize_site and not (args.realtor_id or args.group_id):
            parser.error("--initialize-site에는 --realtor-id 또는 --group-id가 필요합니다.")
        args.realtor_id, resolved_group_id = resolve_realtor_and_site(
            conn, args.realtor_id, args.group_id, args.initialize_site
        )
        targets = fetch_targets(
            conn, args.realtor_id, args.limit, args.article_no,
            args.refresh_existing or args.sync_all_current or bool(args.multi_article_id),
            args.multi_article_id,
            all_current=args.initialize_site or args.sync_all_current,
        )
        if args.initialize_site:
            targets = targets[:max(1, min(200, int(args.initial_max)))]
        build_job_id = None
        if args.initialize_site and not args.dry_run:
            with conn.cursor() as cur:
                cur.execute("SELECT id AS site_id FROM multi_sites WHERE realtor_id=%s AND status='active' LIMIT 1", (args.realtor_id,))
                site_row = cur.fetchone()
                if not site_row:
                    raise RuntimeError("활성화된 multi_sites가 없습니다.")
                if args.build_job_id:
                    cur.execute("""UPDATE multi_site_build_jobs
                                      SET status='running',total_count=%s,completed_count=0,
                                          success_count=0,failed_count=0,started_at=NOW(),
                                          finished_at=NULL,updated_at=NOW()
                                    WHERE id=%s AND site_id=%s AND realtor_id=%s
                                      AND build_type='initial_articles'""",
                                (len(targets), args.build_job_id, site_row["site_id"], args.realtor_id))
                    if cur.rowcount < 1:
                        raise RuntimeError(f"최초 기사 생성 작업을 찾을 수 없습니다: {args.build_job_id}")
                    build_job_id = args.build_job_id
                else:
                    cur.execute("""
                        INSERT INTO multi_site_build_jobs
                        (site_id,realtor_id,build_type,status,total_count,completed_count,success_count,failed_count,started_at,created_at,updated_at)
                        VALUES (%s,%s,'initial_articles','running',%s,0,0,0,NOW(),NOW(),NOW())
                    """, (site_row["site_id"], args.realtor_id, len(targets)))
                    build_job_id = cur.lastrowid
            conn.commit()
        print(f"[MULTI TARGET COUNT] {len(targets)}")
        for row in targets:
            try:
                save_article(
                    conn, row, dry_run=args.dry_run,
                    refresh_existing=(
                        args.refresh_existing or args.sync_all_current
                        or bool(args.multi_article_id)
                    ),
                )
                success += 1
            except Exception as exc:
                failed += 1
                print(f"[MULTI ARTICLE ERROR] article_no={row.get('article_no')} error={exc}")
                traceback.print_exc()
            if build_job_id:
                with conn.cursor() as cur:
                    cur.execute("""
                        UPDATE multi_site_build_jobs
                           SET completed_count=%s, success_count=%s, failed_count=%s,
                               current_article_no=%s, updated_at=NOW()
                         WHERE id=%s
                    """, (success + failed, success, failed, row.get("article_no"), build_job_id))
                conn.commit()
        if build_job_id:
            with conn.cursor() as cur:
                cur.execute("""
                    UPDATE multi_site_build_jobs
                       SET status=%s, current_article_no=NULL, finished_at=NOW(), updated_at=NOW()
                     WHERE id=%s
                """, ('completed' if failed == 0 else 'partial', build_job_id))
            conn.commit()
        if not args.dry_run:
            reconcile_inactive_articles(conn, realtor_id=args.realtor_id)
            reconcile_unready_articles(conn, realtor_id=args.realtor_id)
            # 기사 생성과 별도로 현재 공개 매물을 다시 집계한다.
            # 중개사별 실패는 기존 스냅샷을 유지하며 다른 중개사의 갱신을 막지 않는다.
            backfill_region_codes(conn, realtor_id=args.realtor_id)
            refresh_site_area_snapshots(conn, realtor_id=args.realtor_id)
            # 공공 API 실패/미설정은 기사 생성 성공 여부에 영향을 주지 않는다.
            try:
                refresh_complex_public_data(conn, realtor_id=args.realtor_id, limit=20)
            except Exception as exc:
                conn.rollback()
                print(f"[MULTI PUBLIC API WARNING] {exc}")
            # 정책·생활정보는 별도 수집기가 출처별 20시간 캐시를 적용한다.
            if args.skip_policy_feeds:
                print("[MULTI POLICY FEED SKIP] command_option")
            else:
                try:
                    refresh_policy_feeds(conn, realtor_id=args.realtor_id)
                except Exception as exc:
                    conn.rollback()
                    print(f"[MULTI POLICY FEED WARNING] {exc}")
    finally:
        conn.close()
    print(f"[MULTI DONE] success={success} failed={failed}")
    raise SystemExit(1 if failed else 0)


if __name__ == "__main__":
    main()