# -*- coding: utf-8 -*-
# V15_STEP112_EXISTING_DRAFT_POPUP_HOTFIX
# Generated from V1.5 publish_worker with V2 popup/blog_id/session flow fixes.
# ---------------------------------------------------------
# STABLE RENDERED COPY VERSION
# - set_input_files 이미지 업로드 사용 안 함
# - clipboard HTML source paste 사용 안 함
# - preview page 렌더링 후 Ctrl+A/C 복사 → 네이버 editor Ctrl+V
# ---------------------------------------------------------
import os
import sys
import traceback
import json
import argparse
import time
import re
import socket
import random
from datetime import datetime, timedelta
from urllib.parse import urlparse, parse_qs, unquote, quote
from playwright.sync_api import sync_playwright
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.dirname(CURRENT_DIR)
if ROOT_DIR not in sys.path:
sys.path.append(ROOT_DIR)
from config import STORAGE_DIR
try:
from config import APP_CRYPTO_KEY
except Exception:
APP_CRYPTO_KEY = ""
try:
from Crypto.Cipher import AES
CRYPTO_AVAILABLE = True
except Exception:
AES = None
CRYPTO_AVAILABLE = False
from db import get_conn
SESSION_DIR = os.path.join(STORAGE_DIR, "naver_sessions")
os.makedirs(SESSION_DIR, exist_ok=True)
def normalize_aes_key(key_text):
key = str(key_text or "").encode("utf-8")
if len(key) >= 32:
return key[:32]
return key.ljust(32, b"\0")
def php_openssl_decrypt(encrypted_text):
"""PHP openssl_encrypt(AES-256-CBC) 저장값 복호화."""
if not encrypted_text:
return ""
if not CRYPTO_AVAILABLE:
raise Exception("pycryptodome 설치 필요: pip install pycryptodome")
raw = str(encrypted_text or "").strip()
try:
import base64
raw_bytes = base64.b64decode(raw, validate=True)
except Exception:
return ""
if len(raw_bytes) <= 16:
return ""
iv = raw_bytes[:16]
cipher_text = raw_bytes[16:]
cipher = AES.new(
normalize_aes_key(APP_CRYPTO_KEY),
AES.MODE_CBC,
iv
)
decrypted = cipher.decrypt(cipher_text)
pad_len = decrypted[-1]
if pad_len < 1 or pad_len > 16:
return ""
return decrypted[:-pad_len].decode("utf-8", errors="ignore")
def get_naver_account_credentials(realtor_id):
"""자동 재로그인용 네이버 계정 정보 조회."""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT
naver_id,
naver_password_enc,
consent_status,
login_status
FROM blog_naver_accounts
WHERE realtor_id = %s
LIMIT 1
""", (realtor_id,))
row = cur.fetchone()
if not row:
return None, None, "등록된 네이버 계정 정보가 없습니다."
if str(row.get("consent_status") or "").strip() != "agreed":
return None, None, "네이버 자동발행 동의 상태가 아닙니다."
naver_id = str(row.get("naver_id") or "").strip()
encrypted_pw = str(row.get("naver_password_enc") or "").strip()
if not naver_id:
return None, None, "네이버 ID가 비어 있습니다."
if not encrypted_pw:
return None, None, "네이버 비밀번호 암호화 값이 비어 있습니다."
try:
naver_pw = php_openssl_decrypt(encrypted_pw)
except Exception as e:
return None, None, f"네이버 비밀번호 복호화 실패: {e}"
if not naver_pw:
return None, None, "네이버 비밀번호 복호화 결과가 비어 있습니다."
return naver_id, naver_pw, ""
finally:
conn.close()
def get_naver_blog_id(realtor_id):
"""블로그 주소용 blog_id 조회. 없으면 naver_id fallback."""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT
naver_id,
blog_id
FROM blog_naver_accounts
WHERE realtor_id = %s
LIMIT 1
""", (realtor_id,))
row = cur.fetchone()
if not row:
return ""
naver_id = str(row.get("naver_id") or "").strip()
blog_id = str(row.get("blog_id") or "").strip()
return blog_id or naver_id
except Exception as e:
print("[BLOG ID LOAD ERROR]", realtor_id, str(e))
return ""
finally:
conn.close()
def detect_login_fail_reason(page):
"""로그인 실패 원인 분류."""
try:
page_text = page.locator("body").inner_text(timeout=5000)
except Exception:
page_text = ""
page_text_lower = page_text.lower()
current_url = str(page.url or "")
if (
"2단계 인증" in page_text
or "2차 인증" in page_text
or "otp" in page_text_lower
or "인증번호" in page_text
):
return "2차 인증 필요"
if (
"새로운 환경에서 로그인" in page_text
or "본인 확인" in page_text
or "새 기기" in page_text
):
return "새 기기 로그인 인증 필요"
if (
"자동입력 방지문자" in page_text
or "captcha" in page_text_lower
or "보안문자" in page_text
):
return "캡차 발생"
if (
"비밀번호가 일치하지 않습니다" in page_text
or "아이디 또는 비밀번호" in page_text
):
return "아이디 또는 비밀번호 오류"
if (
"해외" in page_text
or "지역" in page_text
or "로그인 차단" in page_text
):
return "타지역 로그인 제한"
if "nid.naver.com" in current_url.lower():
return f"로그인 미완료 / current_url={current_url}"
return f"로그인 실패 원인 미확인 / current_url={current_url}"
def classify_naver_login_status(message, default_status="expired"):
"""
STEP96:
publish_worker에서 자동 재로그인 실패 원인을 운영 상태값으로 분류한다.
렌더링 복사/대표이미지/본문 입력 로직은 건드리지 않는다.
"""
text = str(message or "")
if (
"아이디 또는 비밀번호" in text
or "비밀번호가 일치" in text
or "비밀번호 오류" in text
or "비밀번호 복호화" in text
or "비밀번호 암호화 값" in text
or "비밀번호 복호화 결과" in text
):
return "password_error"
if (
"2차 인증" in text
or "2단계 인증" in text
or "otp" in text.lower()
or "인증번호" in text
or "본인 확인" in text
or "새 기기" in text
or "새로운 환경" in text
or "캡차" in text
or "captcha" in text.lower()
or "보안문자" in text
or "타지역" in text
or "로그인 차단" in text
or "보호조치" in text
):
return "naver_block"
return default_status
def auto_relogin_naver(page, context, realtor_id, session_file, reason="", target_url=""):
"""
세션 만료 감지 시 기존 브라우저 컨텍스트에서 네이버 자동 재로그인을 1회 시도한다.
기존 렌더링 복사/대표이미지 업로드 방식은 건드리지 않는다.
"""
target_url = normalize_naver_login_target_url(target_url) or "https://blog.naver.com/GoBlogWrite.naver"
print("[AUTO RELOGIN START]", f"realtor_id={realtor_id}", f"reason={reason}", f"target_url={target_url}")
naver_id, naver_pw, cred_error = get_naver_account_credentials(realtor_id)
if cred_error:
classified_status = classify_naver_login_status(cred_error, default_status="expired")
update_naver_session_status(
realtor_id,
classified_status,
f"자동 재로그인 불가: {cred_error}",
session_file=session_file
)
print("[AUTO RELOGIN CREDENTIAL ERROR]", classified_status, cred_error)
return False, cred_error
try:
if not is_login_page_url(page.url):
login_url = "https://nid.naver.com/nidlogin.login"
if target_url:
login_url += "?mode=form&url=" + quote(target_url, safe="")
page.goto(login_url, wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(1200)
print("[AUTO RELOGIN PAGE]", page.url)
page.evaluate("""
([naverId, naverPw]) => {
const id = document.querySelector('#id');
const pw = document.querySelector('#pw');
if (id) {
id.focus();
id.value = naverId;
id.dispatchEvent(new Event('input', { bubbles: true }));
id.dispatchEvent(new Event('change', { bubbles: true }));
}
if (pw) {
pw.focus();
pw.value = naverPw;
pw.dispatchEvent(new Event('input', { bubbles: true }));
pw.dispatchEvent(new Event('change', { bubbles: true }));
}
}
""", [naver_id, naver_pw])
page.wait_for_timeout(600)
clicked = False
for selector in ["#log\\.login", "button:has-text('로그인')", "input[type='submit']"]:
try:
loc = page.locator(selector).first
if loc.count() > 0:
loc.click(force=True, timeout=5000)
clicked = True
break
except Exception:
continue
if not clicked:
try:
page.keyboard.press("Enter")
clicked = True
except Exception:
pass
if not clicked:
raise Exception("로그인 버튼을 찾지 못했습니다.")
login_success = False
for i in range(90):
page.wait_for_timeout(1000)
try:
cookies = context.cookies()
cookie_names = [
c.get("name", "")
for c in cookies
if "naver.com" in c.get("domain", "")
]
if "NID_AUT" in cookie_names or "NID_SES" in cookie_names:
login_success = True
break
except Exception:
pass
if i in [5, 15, 30, 60]:
print(f"[AUTO RELOGIN WAIT] {i}/90 url={page.url}")
if not login_success:
fail_reason = detect_login_fail_reason(page)
classified_status = classify_naver_login_status(fail_reason, default_status="expired")
update_naver_session_status(
realtor_id,
classified_status,
f"자동 재로그인 실패: {fail_reason}",
session_file=session_file
)
print("[AUTO RELOGIN FAIL]", classified_status, fail_reason)
return False, fail_reason
context.storage_state(path=session_file)
update_naver_session_status(
realtor_id,
"linked",
"자동 재로그인 성공 / 세션 재저장",
session_file=session_file
)
print("[AUTO RELOGIN SUCCESS] session saved:", session_file)
if target_url:
try:
page.goto(target_url, wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(2500)
print("[AUTO RELOGIN TARGET GOTO]", page.url)
# 네이버가 로그인 URL에 머무는 경우가 있어 한 번 더 목적지로 강제 이동
if is_login_page_url(page.url):
normalized_target = normalize_naver_login_target_url(page.url) or target_url
print("[AUTO RELOGIN TARGET RETRY]", normalized_target)
page.goto(normalized_target, wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(2500)
print("[AUTO RELOGIN TARGET RETRY URL]", page.url)
except Exception as e:
print("[AUTO RELOGIN TARGET GOTO ERROR]", str(e))
return True, "자동 재로그인 성공"
except Exception as e:
message = f"자동 재로그인 예외: {e}"
classified_status = classify_naver_login_status(message, default_status="expired")
update_naver_session_status(
realtor_id,
classified_status,
message,
session_file=session_file
)
print("[AUTO RELOGIN ERROR]", classified_status, message)
return False, message
def normalize_naver_login_target_url(target_url):
"""
nid.naver.com 로그인 URL이 target_url로 들어온 경우,
query string의 실제 url 값을 꺼내 글쓰기 목적지로 복원한다.
예:
https://nid.naver.com/nidlogin.login?mode=form&url=https://blog.naver.com/GoBlogWrite.naver
-> https://blog.naver.com/GoBlogWrite.naver
"""
raw = str(target_url or "").strip()
if not raw:
return ""
if not is_login_page_url(raw):
return raw
try:
parsed = urlparse(raw)
qs = parse_qs(parsed.query or "")
real_url = ""
if "url" in qs and qs["url"]:
real_url = str(qs["url"][0] or "").strip()
if real_url:
real_url = unquote(real_url)
if real_url and not is_login_page_url(real_url):
print("[LOGIN TARGET NORMALIZED]", raw, "=>", real_url)
return real_url
except Exception as e:
print("[LOGIN TARGET NORMALIZE ERROR]", str(e))
# 로그인 URL에서 실제 목적지를 못 꺼내면 글쓰기 페이지로 fallback
if "GoBlogWrite" in raw or "PostWriteForm" in raw:
fallback = "https://blog.naver.com/GoBlogWrite.naver"
print("[LOGIN TARGET FALLBACK]", fallback)
return fallback
return ""
def recover_login_page_once(page, context, realtor_id, session_file, reason, target_url=""):
"""현재 URL이 로그인 페이지이면 자동 재로그인을 1회 시도한다."""
if not is_login_page_url(page.url):
return True, "login page 아님"
target_url = normalize_naver_login_target_url(target_url) or normalize_naver_login_target_url(page.url)
ok, message = auto_relogin_naver(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=reason,
target_url=target_url,
)
if ok and not is_login_page_url(page.url):
return True, message
return False, message
def extract_article_no_from_html(html):
text = str(html or "")
patterns = [
r"매물번호\s*\s*:\s*([0-9]{6,20})",
r"매물번호[^0-9]{0,20}([0-9]{6,20})",
r"articleNo[=:/ ]+([0-9]{6,20})",
]
for pattern in patterns:
m = re.search(pattern, text)
if m:
return m.group(1).strip()
return ""
def validate_queue_draft_integrity(queue):
"""발행 직전 queue.article_no, draft.article_no, HTML 내부 매물번호를 검증한다."""
queue_id = queue.get("id")
draft_id = queue.get("draft_id")
queue_article_no = str(
queue.get("queue_article_no")
or queue.get("article_no")
or ""
).strip()
draft_article_no = str(
queue.get("draft_article_no")
or ""
).strip()
if not queue_article_no:
return False, f"[INTEGRITY_HOLD] queue article_no empty queue_id={queue_id}, draft_id={draft_id}"
if not draft_article_no:
return False, f"[INTEGRITY_HOLD] draft article_no empty queue_id={queue_id}, draft_id={draft_id}"
if queue_article_no != draft_article_no:
return False, (
f"[INTEGRITY_HOLD] queue/draft article_no mismatch "
f"queue_id={queue_id}, draft_id={draft_id}, "
f"queue_article_no={queue_article_no}, draft_article_no={draft_article_no}"
)
html = str(queue.get("clipboard_html") or queue.get("draft_html") or "")
html_article_no = extract_article_no_from_html(html)
if html_article_no and html_article_no != draft_article_no:
return False, (
f"[INTEGRITY_HOLD] html article_no mismatch "
f"queue_id={queue_id}, draft_id={draft_id}, "
f"draft_article_no={draft_article_no}, html_article_no={html_article_no}"
)
try:
source = json.loads(queue.get("source_json") or "{}")
except Exception:
source = {}
source_detail = source.get("detail") or {}
source_article_no = str(
source_detail.get("article_no")
or source_detail.get("articleNo")
or ""
).strip()
if source_article_no and source_article_no != draft_article_no:
return False, (
f"[INTEGRITY_HOLD] source_json article_no mismatch "
f"queue_id={queue_id}, draft_id={draft_id}, "
f"draft_article_no={draft_article_no}, source_article_no={source_article_no}"
)
return True, f"[INTEGRITY OK] queue_id={queue_id}, draft_id={draft_id}, article_no={draft_article_no}"
def get_pending_queue(
realtor_id=None,
per_realtor_daily_limit=1,
skip_realtor_ids=None,
):
"""
STEP87 운영형 발행 대상 조회.
핵심:
- 중개사별 하루 발행 제한 적용. 기본 1건.
- 오늘 이미 published 된 중개사는 제외.
- 발행 순서는 queue 생성순이 아니라 네이버부동산 등록일 최신순.
기준: blog_realtor_articles.first_posted_at > created_at > q.created_at
- 같은 worker 실행 중 이미 시도한 중개사는 skip_realtor_ids로 제외하여
세션 오류/글쓰기 오류가 반복되는 것을 방지한다.
"""
conn = get_conn()
try:
try:
per_realtor_daily_limit = int(per_realtor_daily_limit or 1)
except Exception:
per_realtor_daily_limit = 1
if per_realtor_daily_limit <= 0:
per_realtor_daily_limit = 1
skip_realtor_ids = [
int(x)
for x in (skip_realtor_ids or [])
if str(x or "").strip()
]
sql = """
SELECT
q.*,
d.draft_title,
d.draft_html,
d.clipboard_html,
d.plain_text,
d.article_no AS draft_article_no,
d.source_json,
q.article_no AS queue_article_no,
CHAR_LENGTH(d.clipboard_html) AS clipboard_len,
CHAR_LENGTH(d.draft_html) AS draft_len,
CHAR_LENGTH(d.plain_text) AS plain_len,
r.office_name,
ra.first_posted_at AS article_first_posted_at,
ra.created_at AS article_cache_created_at,
COALESCE(
ra.first_posted_at,
ra.created_at,
q.created_at
) AS publish_sort_at,
(
SELECT COUNT(*)
FROM blog_publish_queue qp
WHERE qp.realtor_id = q.realtor_id
AND qp.queue_status = 'published'
AND qp.finished_at >= CURDATE()
AND qp.finished_at < DATE_ADD(CURDATE(), INTERVAL 1 DAY)
) AS today_published_count
FROM blog_publish_queue q
INNER JOIN blog_article_drafts d
ON q.draft_id = d.id
INNER JOIN blog_realtors r
ON q.realtor_id = r.id
LEFT JOIN blog_naver_accounts na
ON na.realtor_id = q.realtor_id
LEFT JOIN blog_publish_settings ps
ON ps.realtor_id = q.realtor_id
LEFT JOIN blog_realtor_articles ra
ON ra.realtor_id = q.realtor_id
AND CONVERT(ra.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci
=
CONVERT(q.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci
WHERE q.queue_status = 'pending'
AND COALESCE(r.status, 'active') = 'active'
AND COALESCE(ps.auto_publish_enabled, 0) = 1
AND COALESCE(na.consent_status, '') = 'agreed'
AND COALESCE(na.login_status, '') NOT IN ('password_error', 'naver_block', 'disconnected')
AND (q.scheduled_at IS NULL OR q.scheduled_at <= NOW())
AND (q.next_retry_at IS NULL OR q.next_retry_at <= NOW())
AND (
SELECT COUNT(*)
FROM blog_publish_queue qp
WHERE qp.realtor_id = q.realtor_id
AND qp.queue_status = 'published'
AND qp.finished_at >= CURDATE()
AND qp.finished_at < DATE_ADD(CURDATE(), INTERVAL 1 DAY)
) < %s
"""
params = [int(per_realtor_daily_limit)]
if realtor_id:
sql += " AND q.realtor_id = %s "
params.append(int(realtor_id))
if skip_realtor_ids:
placeholders = ", ".join(["%s"] * len(skip_realtor_ids))
sql += f" AND q.realtor_id NOT IN ({placeholders}) "
params.extend(skip_realtor_ids)
sql += """
ORDER BY
publish_sort_at DESC,
q.id DESC
LIMIT 1
"""
with conn.cursor() as cur:
cur.execute(sql, params)
row = cur.fetchone()
if row:
print(
"[PUBLISH TARGET SELECTED]",
f"queue_id={row.get('id')}",
f"realtor_id={row.get('realtor_id')}",
f"article_no={row.get('queue_article_no') or row.get('article_no')}",
f"sort_at={row.get('publish_sort_at')}",
f"today_published_count={row.get('today_published_count')}",
)
else:
print(
"[PUBLISH TARGET NONE]",
f"realtor_id={realtor_id or 'ALL'}",
f"daily_limit={per_realtor_daily_limit}",
f"skip_realtor_ids={skip_realtor_ids}",
)
return row
finally:
conn.close()
def reset_stale_processing(minutes=90):
"""
worker가 중간에 죽어서 processing 상태로 멈춘 큐를 pending으로 복구.
자동 야간 발행에서는 필수 안전장치.
"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_publish_queue
SET
queue_status = 'pending',
error_message = CONCAT('[AUTO RESET] stale processing over ', %s, ' minutes'),
updated_at = NOW()
WHERE queue_status = 'processing'
AND started_at IS NOT NULL
AND started_at < DATE_SUB(NOW(), INTERVAL %s MINUTE)
""", (int(minutes), int(minutes)))
affected = cur.rowcount
conn.commit()
if affected:
print(f"[STALE RESET] {affected} rows reset to pending")
return affected
finally:
conn.close()
def reset_session_hold_queues(realtor_id=None, limit=20):
"""
네이버 재연동 후 SESSION_HOLD 상태의 큐를 pending으로 복구한다.
일반 실패/무결성 오류/글쓰기 오류는 건드리지 않고,
세션 만료로 보류된 큐만 복구한다.
"""
conn = get_conn()
try:
sql = """
UPDATE blog_publish_queue
SET
queue_status = 'pending',
error_message = CONCAT('[SESSION HOLD RESET] ', IFNULL(error_message, '')),
started_at = NULL,
finished_at = NULL,
updated_at = NOW()
WHERE queue_status = 'hold'
AND error_message LIKE '%%[SESSION_HOLD]%%'
"""
params = []
if realtor_id:
sql += " AND realtor_id = %s "
params.append(int(realtor_id))
sql += " ORDER BY id DESC LIMIT %s "
params.append(int(limit))
with conn.cursor() as cur:
cur.execute(sql, params)
affected = cur.rowcount
conn.commit()
print(f"[SESSION HOLD RESET] affected={affected}")
return affected
finally:
conn.close()
def count_session_hold_queues(realtor_id=None):
conn = get_conn()
try:
sql = """
SELECT COUNT(*) AS cnt
FROM blog_publish_queue
WHERE queue_status = 'hold'
AND error_message LIKE '%%[SESSION_HOLD]%%'
"""
params = []
if realtor_id:
sql += " AND realtor_id = %s "
params.append(int(realtor_id))
with conn.cursor() as cur:
cur.execute(sql, params)
row = cur.fetchone()
return int((row or {}).get("cnt") or 0)
finally:
conn.close()
def get_worker_server_name():
"""관리자 화면에서 어느 서버/프로세스가 처리했는지 확인하기 위한 값."""
try:
return socket.gethostname()
except Exception:
return "unknown-worker"
def safe_json_dumps(data):
try:
return json.dumps(data or {}, ensure_ascii=False, default=str)
except Exception:
return "{}"
def classify_publish_error(message, fallback="unknown"):
"""큐 실패/보류 사유를 관리자 필터용으로 짧게 분류한다."""
text = str(message or "")
if "[SESSION_HOLD]" in text or "세션" in text or "로그인" in text:
return "session_hold"
if "[WRITE_HOLD]" in text or "iframe" in text or "글쓰기" in text:
return "write_hold"
if "[INTEGRITY_HOLD]" in text or "article_no mismatch" in text:
return "integrity_hold"
if "본문" in text or "에디터 본문" in text or "BODY INPUT" in text:
return "content_failed"
if "대표이미지" in text or "REP" in text:
return "representative_image_failed"
if "발행 URL 검증" in text:
return "publish_result_failed"
if "발행 버튼" in text or "발행하기" in text:
return "publish_click_failed"
if "timeout" in text.lower() or "Timeout" in text or "일시" in text:
return "transient_error"
return fallback
def is_within_publish_window(start_hour=22, end_hour=8):
"""발행 허용 시간인지 확인. 기본값은 22:00~08:00."""
now_hour = datetime.now().hour
start_hour = int(start_hour)
end_hour = int(end_hour)
if start_hour == end_hour:
return True
# 예: 22~08처럼 자정을 넘기는 윈도우
if start_hour > end_hour:
return now_hour >= start_hour or now_hour < end_hour
# 예: 09~18처럼 같은 날짜 안의 윈도우
return start_hour <= now_hour < end_hour
def next_publish_window_start(start_hour=22):
"""다음 발행 가능 시작 시각을 DATETIME 문자열로 반환."""
now = datetime.now()
start_hour = int(start_hour)
target = now.replace(hour=start_hour, minute=0, second=0, microsecond=0)
if now >= target:
target = target + timedelta(days=1)
return target.strftime("%Y-%m-%d %H:%M:%S")
def get_publish_window_end_datetime(start_hour=22, end_hour=8):
"""현재 발행 윈도우의 종료 시각을 datetime으로 반환한다."""
now = datetime.now()
start_hour = int(start_hour)
end_hour = int(end_hour)
# 24시간 허용 모드
if start_hour == end_hour:
return now + timedelta(hours=24)
# 예: 22:00~08:00처럼 자정을 넘기는 윈도우
if start_hour > end_hour:
if now.hour >= start_hour:
return (now + timedelta(days=1)).replace(hour=end_hour, minute=0, second=0, microsecond=0)
if now.hour < end_hour:
return now.replace(hour=end_hour, minute=0, second=0, microsecond=0)
return next_publish_window_start(start_hour)
# 예: 09:00~18:00처럼 같은 날짜 안의 윈도우
if start_hour <= now.hour < end_hour:
return now.replace(hour=end_hour, minute=0, second=0, microsecond=0)
return next_publish_window_start(start_hour)
def seconds_until_publish_window_end(start_hour=22, end_hour=8):
"""현재 시각부터 발행 윈도우 종료까지 남은 초를 반환한다."""
end_dt = get_publish_window_end_datetime(start_hour, end_hour)
if isinstance(end_dt, str):
try:
end_dt = datetime.strptime(end_dt, "%Y-%m-%d %H:%M:%S")
except Exception:
return 0
return max(0, int((end_dt - datetime.now()).total_seconds()))
def count_available_pending_queues(realtor_id=None, per_realtor_daily_limit=1):
"""
현재 발행 가능한 중개사 수를 계산한다.
STEP87 이후에는 pending 큐 총량이 아니라,
'오늘 아직 발행하지 않았고 발행 가능한 중개사 수'가 운영상 더 중요하다.
"""
conn = get_conn()
try:
try:
per_realtor_daily_limit = int(per_realtor_daily_limit or 1)
except Exception:
per_realtor_daily_limit = 1
if per_realtor_daily_limit <= 0:
per_realtor_daily_limit = 1
sql = """
SELECT COUNT(DISTINCT q.realtor_id) AS cnt
FROM blog_publish_queue q
INNER JOIN blog_realtors r
ON r.id = q.realtor_id
LEFT JOIN blog_naver_accounts na
ON na.realtor_id = q.realtor_id
LEFT JOIN blog_publish_settings ps
ON ps.realtor_id = q.realtor_id
WHERE q.queue_status = 'pending'
AND COALESCE(r.status, 'active') = 'active'
AND COALESCE(ps.auto_publish_enabled, 0) = 1
AND COALESCE(na.consent_status, '') = 'agreed'
AND COALESCE(na.login_status, '') NOT IN ('password_error', 'naver_block', 'disconnected')
AND (q.scheduled_at IS NULL OR q.scheduled_at <= NOW())
AND (q.next_retry_at IS NULL OR q.next_retry_at <= NOW())
AND (
SELECT COUNT(*)
FROM blog_publish_queue qp
WHERE qp.realtor_id = q.realtor_id
AND qp.queue_status = 'published'
AND qp.finished_at >= CURDATE()
AND qp.finished_at < DATE_ADD(CURDATE(), INTERVAL 1 DAY)
) < %s
"""
params = [int(per_realtor_daily_limit)]
if realtor_id:
sql += " AND q.realtor_id = %s "
params.append(int(realtor_id))
with conn.cursor() as cur:
cur.execute(sql, params)
row = cur.fetchone()
return int((row or {}).get("cnt") or 0)
finally:
conn.close()
def count_available_pending_private_queues(realtor_id=None):
"""현재 처리 가능한 private_queue pending 수를 계산한다."""
conn = get_conn()
try:
sql = """
SELECT COUNT(*) AS cnt
FROM blog_post_private_queue
WHERE queue_status = 'pending'
AND (scheduled_at IS NULL OR scheduled_at <= NOW())
AND (next_retry_at IS NULL OR next_retry_at <= NOW())
"""
params = []
if realtor_id:
sql += " AND realtor_id = %s "
params.append(int(realtor_id))
with conn.cursor() as cur:
cur.execute(sql, params)
row = cur.fetchone()
return int((row or {}).get("cnt") or 0)
finally:
conn.close()
def print_remaining_queue_summary(realtor_id=None):
"""worker 종료 시 관리자/로그 확인용 남은 큐 요약."""
try:
publish_pending = count_available_pending_queues(realtor_id=realtor_id)
except Exception as e:
publish_pending = -1
print("[REMAINING PUBLISH QUEUE COUNT ERROR]", str(e))
try:
private_pending = count_available_pending_private_queues(realtor_id=realtor_id)
except Exception as e:
private_pending = -1
print("[REMAINING PRIVATE QUEUE COUNT ERROR]", str(e))
print(
"[REMAINING QUEUE]",
f"realtor_id={realtor_id or 'ALL'}",
f"publish_pending={publish_pending}",
f"private_pending={private_pending}",
)
return {
"publish_pending": publish_pending,
"private_pending": private_pending,
}
def calculate_smart_sleep_seconds(
processed,
limit,
realtor_id=None,
window_start_hour=22,
window_end_hour=8,
until_time=None,
avg_publish_seconds=None,
estimated_publish_seconds=600,
min_sleep=20,
max_sleep=900,
sleep_jitter=0.5,
):
"""
남은 시간, 남은 목표 건수, 실제 평균 발행시간을 기준으로 다음 발행 전 대기시간을 계산한다.
- 400건/10시간처럼 빠듯하면 대기시간이 거의 0에 가까워진다.
- 100건/10시간처럼 여유가 있으면 몇 분 단위로 랜덤 분산한다.
"""
try:
processed = int(processed)
limit = int(limit)
except Exception:
return 0
remaining_by_limit = max(0, limit - processed)
if remaining_by_limit <= 0:
return 0
try:
pending_count = count_available_pending_queues(realtor_id=realtor_id)
except Exception as e:
print("[SMART SLEEP COUNT ERROR]", str(e))
pending_count = remaining_by_limit
remaining_count = min(remaining_by_limit, max(0, int(pending_count)))
if remaining_count <= 0:
return 0
if until_time:
remaining_seconds = seconds_until_time(until_time)
else:
remaining_seconds = seconds_until_publish_window_end(window_start_hour, window_end_hour)
if remaining_seconds <= 0:
return 0
try:
publish_seconds = float(avg_publish_seconds or 0)
except Exception:
publish_seconds = 0
if publish_seconds <= 0:
publish_seconds = float(estimated_publish_seconds or 90)
target_cycle_seconds = remaining_seconds / max(1, remaining_count)
base_sleep = target_cycle_seconds - publish_seconds
# 시간이 빠듯하면 억지로 쉬지 않는다.
if base_sleep <= 0:
print(
"[SMART SLEEP] no sleep /",
f"remaining_seconds={remaining_seconds}",
f"remaining_count={remaining_count}",
f"target_cycle={target_cycle_seconds:.1f}",
f"avg_publish={publish_seconds:.1f}",
)
return 0
try:
min_sleep = max(0, int(min_sleep))
max_sleep = max(min_sleep, int(max_sleep))
sleep_jitter = max(0.0, min(0.9, float(sleep_jitter)))
except Exception:
min_sleep = 20
max_sleep = 900
sleep_jitter = 0.5
low = max(float(min_sleep), base_sleep * (1.0 - sleep_jitter))
high = min(float(max_sleep), base_sleep * (1.0 + sleep_jitter))
if high < low:
high = low
sleep_seconds = int(random.uniform(low, high))
print(
"[SMART SLEEP CALC]",
f"remaining_seconds={remaining_seconds}",
f"remaining_count={remaining_count}",
f"target_cycle={target_cycle_seconds:.1f}",
f"avg_publish={publish_seconds:.1f}",
f"base_sleep={base_sleep:.1f}",
f"range={int(low)}~{int(high)}",
f"selected={sleep_seconds}",
)
return max(0, sleep_seconds)
def is_retryable_error_type(error_type):
"""다음 야간에 자동 재시도해도 되는 실패 유형."""
error_type = str(error_type or "")
return error_type in [
"content_failed",
"representative_image_failed",
"publish_result_failed",
"publish_click_failed",
"transient_error",
"failed",
]
def reset_retryable_failed_queues(realtor_id=None, max_retry=2, limit=50):
"""
다음 야간 재시도 시간이 지난 failed 큐를 pending으로 복구한다.
SESSION_HOLD / WRITE_HOLD / INTEGRITY_HOLD는 관리자 확인이 필요하므로 건드리지 않는다.
"""
conn = get_conn()
try:
sql = """
UPDATE blog_publish_queue
SET
queue_status = 'pending',
worker_status = 'retry_ready',
error_message = CONCAT('[AUTO RETRY READY] ', IFNULL(error_message, '')),
started_at = NULL,
finished_at = NULL,
updated_at = NOW()
WHERE queue_status = 'failed'
AND next_retry_at IS NOT NULL
AND next_retry_at <= NOW()
AND retry_count < %s
AND IFNULL(last_error_type, '') IN (
'content_failed',
'representative_image_failed',
'publish_result_failed',
'publish_click_failed',
'transient_error',
'failed'
)
"""
params = [int(max_retry)]
if realtor_id:
sql += " AND realtor_id = %s "
params.append(int(realtor_id))
sql += " ORDER BY id ASC LIMIT %s "
params.append(int(limit))
with conn.cursor() as cur:
cur.execute(sql, params)
affected = cur.rowcount
conn.commit()
if affected:
print(f"[AUTO RETRY RESET] failed -> pending affected={affected}")
return affected
finally:
conn.close()
def create_pipeline_run(run_type="publish_worker", meta=None):
"""blog_pipeline_runs 기록. 실패해도 발행 자체는 막지 않는다."""
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()
)
""", (
run_type,
"publish_worker started",
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("[PIPELINE RUN START ERROR]", str(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):
"""blog_pipeline_runs 종료 기록. 실패해도 발행 자체는 막지 않는다."""
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
""", (
status,
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 {}),
run_id
))
conn.commit()
print(f"[PIPELINE RUN FINISH] run_id={run_id}, status={status}")
except Exception as e:
print("[PIPELINE RUN FINISH ERROR]", str(e))
finally:
try:
if conn:
conn.close()
except Exception:
pass
def add_pipeline_log(run_id=None, level="info", step_name="publish_worker", message="", article_no=None, realtor_id=None, context=None):
"""blog_pipeline_logs 기록. 실패해도 발행 자체는 막지 않는다."""
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 "publish_worker")[: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("[PIPELINE LOG ERROR]", str(e))
finally:
try:
if conn:
conn.close()
except Exception:
pass
# ---------------------------------------------------------
# STEP29 DB LOCK HELPERS
# - 기존 파일락은 유지하고, 관리자/배치 상태 확인을 위한 DB 락을 추가한다.
# - blog_app_locks 구조: lock_name, locked_at, expires_at, owner_token
# ---------------------------------------------------------
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=720, owner_token=None):
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 parse_naver_blog_result_url(result_url):
"""
네이버 블로그 발행 결과 URL에서 blog_id / blog_post_no를 추출한다.
예:
https://blog.naver.com/animone/224300936609
https://blog.naver.com/PostView.naver?blogId=animone&logNo=224300936609
"""
url = str(result_url or "").strip()
if not url:
return "", ""
m = re.search(r"blog\.naver\.com/([^/?#]+)/([0-9]{6,30})", url, flags=re.IGNORECASE)
if m:
return str(m.group(1) or "").strip(), str(m.group(2) or "").strip()
m = re.search(r"[?&]blogId=([^&#]+).*?[?&]logNo=([0-9]{6,30})", url, flags=re.IGNORECASE)
if m:
return str(m.group(1) or "").strip(), str(m.group(2) or "").strip()
m = re.search(r"[?&]logNo=([0-9]{6,30}).*?[?&]blogId=([^&#]+)", url, flags=re.IGNORECASE)
if m:
return str(m.group(2) or "").strip(), str(m.group(1) or "").strip()
return "", ""
def update_queue_status(queue_id, status, error_message=None, result_url=None, worker_status=None, last_error_type=None, next_retry_at=None, blog_id=None, blog_post_no=None):
conn = get_conn()
try:
with conn.cursor() as cur:
sql = """
UPDATE blog_publish_queue
SET queue_status = %s,
worker_status = %s,
worker_server = %s,
updated_at = NOW()
"""
params = [
status,
worker_status or status,
get_worker_server_name()
]
if status == "processing":
sql += ", started_at = NOW()"
if status in ["published", "failed", "hold"]:
sql += ", finished_at = NOW()"
if error_message is not None:
sql += ", error_message = %s"
params.append(str(error_message)[:2000])
if result_url is not None:
sql += ", publish_result_url = %s"
params.append(result_url)
parsed_blog_id, parsed_blog_post_no = parse_naver_blog_result_url(result_url)
if blog_id is None:
blog_id = parsed_blog_id
if blog_post_no is None:
blog_post_no = parsed_blog_post_no
if blog_id is not None:
sql += ", blog_id = %s"
params.append(str(blog_id or "")[:100])
if blog_post_no is not None:
sql += ", blog_post_no = %s"
params.append(str(blog_post_no or "")[:50])
if last_error_type is None and status in ["failed", "hold"]:
last_error_type = classify_publish_error(error_message, fallback=status)
if status == "failed":
sql += ", retry_count = retry_count + 1"
if next_retry_at is None and is_retryable_error_type(last_error_type):
next_retry_at = next_publish_window_start(start_hour=22)
if status in ["processing", "published"]:
sql += ", next_retry_at = NULL"
if last_error_type is not None:
sql += ", last_error_type = %s"
params.append(str(last_error_type)[:50])
if next_retry_at is not None:
sql += ", next_retry_at = %s"
params.append(next_retry_at)
sql += " WHERE id = %s"
params.append(queue_id)
cur.execute(sql, params)
conn.commit()
finally:
conn.close()
def update_draft_status(draft_id, status):
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_article_drafts
SET draft_status = %s,
updated_at = NOW()
WHERE id = %s
""", (status, draft_id))
conn.commit()
finally:
conn.close()
def update_draft_publish_result(draft_id, result_url):
"""
발행 성공 후 초안 테이블에 최종 블로그 URL / blog_id / blog_post_no / published_at을 저장한다.
삭제매물 비공개 처리의 기준 데이터가 된다.
"""
blog_id, blog_post_no = parse_naver_blog_result_url(result_url)
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_article_drafts
SET
status = 'published',
draft_status = 'published',
blog_url = %s,
blog_id = %s,
blog_post_no = %s,
published_at = NOW(),
updated_at = NOW()
WHERE id = %s
""", (
str(result_url or "")[:1000],
str(blog_id or "")[:100],
str(blog_post_no or "")[:50],
draft_id
))
conn.commit()
print(
"[DRAFT PUBLISH RESULT SAVED]",
f"draft_id={draft_id}",
f"blog_id={blog_id}",
f"blog_post_no={blog_post_no}",
f"url={result_url}"
)
return blog_id, blog_post_no
finally:
conn.close()
def get_pending_private_queues(realtor_id, limit=3):
"""
같은 중개사 세션을 연 상태에서 처리할 비공개 큐 조회.
대량 처리는 나중에 옵션화하고, 1차는 중개사별 소량 처리로 안전하게 간다.
"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT
id,
realtor_id,
article_no,
draft_id,
blog_url,
blog_id,
blog_post_no,
reason,
retry_count
FROM blog_post_private_queue
WHERE realtor_id = %s
AND queue_status = 'pending'
AND (scheduled_at IS NULL OR scheduled_at <= NOW())
AND (next_retry_at IS NULL OR next_retry_at <= NOW())
AND COALESCE(blog_id, '') <> ''
AND COALESCE(blog_post_no, '') <> ''
ORDER BY id ASC
LIMIT %s
""", (
int(realtor_id),
int(limit),
))
return cur.fetchall() or []
finally:
conn.close()
def update_private_queue_status(private_queue_id, status, worker_status=None, error_message=None, last_error_type=None):
conn = get_conn()
try:
with conn.cursor() as cur:
sql = """
UPDATE blog_post_private_queue
SET
queue_status = %s,
worker_status = %s,
worker_server = %s,
updated_at = NOW()
"""
params = [
str(status or "")[:30],
str(worker_status or status or "")[:50],
get_worker_server_name(),
]
if status == "processing":
sql += ", started_at = NOW()"
if status in ["completed", "failed", "hold"]:
sql += ", finished_at = NOW()"
if status == "failed":
# private_queue는 일시적인 네이버 에디터/세션 오류가 있을 수 있으므로
# 5회 미만은 다음 야간에 자동 재시도 가능하도록 pending으로 되돌린다.
sql += """,
retry_count = retry_count + 1,
next_retry_at = %s,
queue_status = CASE
WHEN retry_count + 1 < 5 THEN 'pending'
ELSE 'failed'
END,
worker_status = CASE
WHEN retry_count + 1 < 5 THEN 'retry_scheduled'
ELSE worker_status
END
"""
params.append(next_publish_window_start(start_hour=22))
if status == "completed":
sql += ", next_retry_at = NULL, last_error_type = NULL"
if error_message is not None:
sql += ", error_message = %s"
params.append(str(error_message or "")[:2000])
if last_error_type is not None:
sql += ", last_error_type = %s"
params.append(str(last_error_type or "")[:50])
sql += " WHERE id = %s"
params.append(int(private_queue_id))
cur.execute(sql, params)
conn.commit()
finally:
conn.close()
def mark_realtor_article_private_done(realtor_id, article_no):
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_realtor_articles
SET
visibility_status = 'private_done',
updated_at = NOW()
WHERE realtor_id = %s
AND article_no = %s
""", (
int(realtor_id),
str(article_no or ""),
))
conn.commit()
finally:
conn.close()
def build_post_edit_candidate_urls(blog_id, blog_post_no):
blog_id = str(blog_id or "").strip()
blog_post_no = str(blog_post_no or "").strip()
return [
f"https://blog.naver.com/PostUpdateForm.naver?blogId={blog_id}&logNo={blog_post_no}",
f"https://blog.naver.com/PostWriteForm.naver?blogId={blog_id}&logNo={blog_post_no}",
f"https://blog.naver.com/{blog_id}/{blog_post_no}",
]
def is_private_success_text(page):
try:
body_text = page.locator("body").inner_text(timeout=5000)
except Exception:
body_text = ""
current_url = str(page.url or "")
if (
"blog.naver.com/" in current_url
and "PostWriteForm" not in current_url
and "PostUpdateForm" not in current_url
and "Redirect=Write" not in current_url
):
return True
if "비공개" in body_text and ("수정" in body_text or "발행" in body_text):
return True
return False
def click_private_option(page):
"""
발행/수정 패널에서 비공개 옵션을 선택한다.
네이버 에디터 UI 변경 가능성이 있어 텍스트/라벨/좌표를 단계적으로 시도한다.
"""
selectors = [
"label:has-text('비공개')",
"button:has-text('비공개')",
"span:has-text('비공개')",
"div:has-text('비공개')",
"input[value='private']",
]
for selector in selectors:
try:
loc = page.locator(selector)
count = loc.count()
if count <= 0:
continue
for idx in range(min(count, 5)):
try:
target = loc.nth(idx)
box = target.bounding_box()
if box:
# 너무 큰 컨테이너는 제외
if box.get("width", 0) > 500 or box.get("height", 0) > 220:
continue
target.click(force=True, timeout=3000)
page.wait_for_timeout(800)
print("[PRIVATE OPTION CLICKED]", selector, "idx=", idx)
return True
except Exception as e:
print("[PRIVATE OPTION CLICK ERROR]", selector, idx, str(e))
except Exception as e:
print("[PRIVATE OPTION SCAN ERROR]", selector, str(e))
# 좌표 fallback: 발행 패널 내 공개/비공개 영역 추정. 테스트 후 조정 가능.
try:
print("[PRIVATE OPTION COORD CLICK TRY]")
page.mouse.click(1050, 350)
page.wait_for_timeout(800)
return True
except Exception as e:
print("[PRIVATE OPTION COORD CLICK ERROR]", str(e))
return False
def click_update_complete_button(page):
"""
수정 글의 최종 완료/발행 버튼 클릭.
"""
selectors = [
"button:has-text('발행하기')",
"button:has-text('수정')",
"button:has-text('확인')",
"button:has-text('완료')",
".confirm_btn__WEaBq",
"button.confirm_btn__WEaBq",
]
for selector in selectors:
try:
loc = page.locator(selector)
count = loc.count()
if count <= 0:
continue
for idx in range(min(count, 5)):
try:
target = loc.nth(idx)
box = target.bounding_box()
if box:
if box.get("width", 0) > 500 or box.get("height", 0) > 220:
continue
target.click(force=True, timeout=5000)
page.wait_for_timeout(8000)
print("[PRIVATE FINAL CLICKED]", selector, "idx=", idx)
return True
except Exception as e:
print("[PRIVATE FINAL CLICK ERROR]", selector, idx, str(e))
except Exception as e:
print("[PRIVATE FINAL SCAN ERROR]", selector, str(e))
try:
print("[PRIVATE FINAL COORD CLICK TRY]")
page.mouse.click(1294, 556)
page.wait_for_timeout(8000)
return True
except Exception as e:
print("[PRIVATE FINAL COORD CLICK ERROR]", str(e))
return False
def open_post_edit_page(page, private_queue):
blog_id = str(private_queue.get("blog_id") or "").strip()
blog_post_no = str(private_queue.get("blog_post_no") or "").strip()
for url in build_post_edit_candidate_urls(blog_id, blog_post_no):
try:
print("[PRIVATE EDIT OPEN]", url)
page.goto(
url,
wait_until="domcontentloaded",
timeout=60000
)
page.wait_for_timeout(2500)
print("[PRIVATE EDIT CURRENT URL]", page.url)
if is_login_page_url(page.url):
return False, "login_required"
# 수정/글쓰기 폼으로 들어갔거나, 본문 페이지라도 이후 버튼 탐색 가능
return True, ""
except Exception as e:
print("[PRIVATE EDIT OPEN ERROR]", url, str(e))
return False, "edit_open_failed"
def click_post_edit_button_if_needed(page):
"""
일반 PostView 페이지로 열린 경우 '수정' 버튼을 찾아 진입한다.
"""
if "PostUpdateForm" in str(page.url) or "PostWriteForm" in str(page.url):
return True
selectors = [
"a:has-text('수정')",
"button:has-text('수정')",
"a[href*='PostUpdateForm']",
"a[href*='PostWriteForm']",
"span:has-text('수정')",
]
for selector in selectors:
try:
loc = page.locator(selector)
if loc.count() <= 0:
continue
print("[PRIVATE EDIT BUTTON]", selector)
href = loc.first.get_attribute("href")
if href:
page.goto(href, wait_until="domcontentloaded", timeout=60000)
else:
loc.first.click(force=True, timeout=5000)
page.wait_for_timeout(2500)
print("[PRIVATE EDIT AFTER BUTTON URL]", page.url)
return True
except Exception as e:
print("[PRIVATE EDIT BUTTON ERROR]", selector, str(e))
return False
def make_post_private(page, context, realtor_id, session_file, private_queue):
private_queue_id = private_queue.get("id")
article_no = str(private_queue.get("article_no") or "")
blog_id = str(private_queue.get("blog_id") or "")
blog_post_no = str(private_queue.get("blog_post_no") or "")
print(
"[PRIVATE START]",
f"private_queue_id={private_queue_id}",
f"realtor_id={realtor_id}",
f"article_no={article_no}",
f"blog_id={blog_id}",
f"blog_post_no={blog_post_no}"
)
update_private_queue_status(
private_queue_id,
"processing",
worker_status="processing"
)
opened, open_message = open_post_edit_page(page, private_queue)
if not opened:
if open_message == "login_required":
recovered, recover_message = recover_login_page_once(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=f"비공개 수정 진입 중 로그인 페이지로 이동: {page.url}",
target_url=str(private_queue.get("blog_url") or ""),
)
if not recovered:
update_private_queue_status(
private_queue_id,
"hold",
worker_status="session_hold",
error_message=f"[PRIVATE SESSION_HOLD] {recover_message}",
last_error_type="session_hold"
)
return False
opened, open_message = open_post_edit_page(page, private_queue)
if not opened:
update_private_queue_status(
private_queue_id,
"failed",
worker_status="edit_open_failed",
error_message=f"비공개 수정 진입 실패 after relogin: {open_message}",
last_error_type="edit_open_failed"
)
return False
else:
update_private_queue_status(
private_queue_id,
"failed",
worker_status="edit_open_failed",
error_message=f"비공개 수정 진입 실패: {open_message}",
last_error_type="edit_open_failed"
)
return False
if not click_post_edit_button_if_needed(page):
update_private_queue_status(
private_queue_id,
"failed",
worker_status="edit_button_failed",
error_message="수정 버튼 또는 수정 화면 진입 실패",
last_error_type="edit_button_failed"
)
return False
# 에디터 iframe 로딩 대기
write_frame = None
for i in range(20):
page.wait_for_timeout(1000)
print(f"[PRIVATE WAIT EDITOR] {i + 1}/20 url={page.url}")
if is_login_page_url(page.url):
recovered, recover_message = recover_login_page_once(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=f"비공개 에디터 대기 중 로그인 페이지로 이동: {page.url}",
target_url=str(page.url or ""),
)
if not recovered:
update_private_queue_status(
private_queue_id,
"hold",
worker_status="session_hold",
error_message=f"[PRIVATE SESSION_HOLD] {recover_message}",
last_error_type="session_hold"
)
return False
write_frame = find_write_frame(page)
if write_frame is not None:
break
# 에디터 iframe이 없어도 패널 조작이 가능할 수 있으므로 바로 실패시키지 않고 진행한다.
if write_frame:
print("[PRIVATE WRITE FRAME FOUND]", write_frame.url)
try:
remove_right_help_panel(write_frame)
except Exception:
pass
else:
print("[PRIVATE WRITE FRAME NOT FOUND] continue with page selectors")
close_guides(page)
# 상단 발행/수정 패널 열기
if not click_top_publish_button(page):
update_private_queue_status(
private_queue_id,
"failed",
worker_status="top_publish_failed",
error_message="비공개 처리: 상단 발행/수정 버튼 클릭 실패",
last_error_type="top_publish_failed"
)
return False
page.wait_for_timeout(1500)
if not click_private_option(page):
update_private_queue_status(
private_queue_id,
"failed",
worker_status="private_option_failed",
error_message="비공개 옵션 선택 실패",
last_error_type="private_option_failed"
)
return False
if not click_update_complete_button(page):
update_private_queue_status(
private_queue_id,
"failed",
worker_status="private_final_failed",
error_message="비공개 최종 완료 버튼 클릭 실패",
last_error_type="private_final_failed"
)
return False
page.wait_for_timeout(3000)
if not is_private_success_text(page):
print("[PRIVATE RESULT WARN] success text/url not clearly verified", page.url)
update_private_queue_status(
private_queue_id,
"completed",
worker_status="completed",
error_message="비공개 처리 완료",
last_error_type=None
)
mark_realtor_article_private_done(
realtor_id=realtor_id,
article_no=article_no
)
print("[PRIVATE DONE]", f"private_queue_id={private_queue_id}", f"article_no={article_no}")
return True
def process_private_queues_for_realtor(page, context, realtor_id, session_file, limit=3):
queues = get_pending_private_queues(
realtor_id=realtor_id,
limit=limit,
)
if not queues:
print("[PRIVATE QUEUE NONE]", f"realtor_id={realtor_id}")
return {
"processed": 0,
"success": 0,
"failed": 0,
}
print("[PRIVATE QUEUE COUNT]", f"realtor_id={realtor_id}", f"count={len(queues)}")
success = 0
failed = 0
for private_queue in queues:
ok = make_post_private(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
private_queue=private_queue,
)
if ok:
success += 1
else:
failed += 1
return {
"processed": len(queues),
"success": success,
"failed": failed,
}
def update_naver_session_status(realtor_id, status, message=None, session_file=None):
conn = get_conn()
try:
session_json = ""
if session_file and os.path.exists(session_file):
try:
with open(session_file, "r", encoding="utf-8") as f:
session_json = f.read()
except Exception:
session_json = ""
with conn.cursor() as cur:
cur.execute("""
SELECT naver_id
FROM blog_naver_accounts
WHERE realtor_id = %s
LIMIT 1
""", (realtor_id,))
row = cur.fetchone()
naver_id = row.get("naver_id") if row else None
if not naver_id:
cur.execute("""
SELECT naver_id
FROM blog_naver_sessions
WHERE realtor_id = %s
ORDER BY id DESC
LIMIT 1
""", (realtor_id,))
row = cur.fetchone()
naver_id = row.get("naver_id") if row else None
if status == "linked":
cur.execute("""
INSERT INTO blog_naver_sessions
(
realtor_id, naver_id, session_status, session_data,
issued_at, last_verified_at, fail_count,
created_at, updated_at
)
VALUES
(
%s, %s, 'linked', %s,
NOW(), NOW(), 0,
NOW(), NOW()
)
""", (realtor_id, naver_id, session_json))
cur.execute("""
UPDATE blog_publish_settings
SET session_status = 'linked',
session_last_checked_at = NOW(),
updated_at = NOW()
WHERE realtor_id = %s
""", (realtor_id,))
cur.execute("""
UPDATE blog_naver_accounts
SET login_status = 'linked',
last_login_at = NOW(),
last_login_error = NULL,
updated_at = NOW()
WHERE realtor_id = %s
""", (realtor_id,))
else:
cur.execute("""
INSERT INTO blog_naver_sessions
(
realtor_id, naver_id, session_status, session_data,
last_verified_at, fail_count,
created_at, updated_at
)
VALUES
(
%s, %s, %s, %s,
NOW(), 1,
NOW(), NOW()
)
""", (realtor_id, naver_id, status, session_json))
cur.execute("""
UPDATE blog_publish_settings
SET session_status = %s,
session_last_checked_at = NOW(),
updated_at = NOW()
WHERE realtor_id = %s
""", (status, realtor_id))
cur.execute("""
UPDATE blog_naver_accounts
SET login_status = %s,
last_login_error = %s,
updated_at = NOW()
WHERE realtor_id = %s
""", (status, str(message or "")[:1000], realtor_id))
conn.commit()
finally:
conn.close()
def get_session_file(realtor_id):
return os.path.join(
SESSION_DIR,
f"naver_session_{realtor_id}.json"
)
def is_login_page_url(url):
url = str(url or "").lower()
if "nid.naver.com" in url:
return True
if "nidlogin.login" in url:
return True
if "mode=form" in url and "login" in url:
return True
return False
def is_write_editor_url(url):
url = str(url or "")
if "PostWriteForm" in url:
return True
if "Redirect=Write" in url:
return True
return False
def mark_session_hold(queue_id, realtor_id, message):
full_message = f"[SESSION_HOLD] realtor_id={realtor_id} / {message}"
update_queue_status(
queue_id,
"hold",
full_message[:1000],
worker_status="session_hold",
last_error_type="session_hold"
)
print("[SESSION HOLD]", full_message)
def mark_write_hold(queue_id, realtor_id, message):
full_message = f"[WRITE_HOLD] realtor_id={realtor_id} / {message}"
update_queue_status(
queue_id,
"hold",
full_message[:1000],
worker_status="write_hold",
last_error_type="write_hold"
)
print("[WRITE HOLD]", full_message)
def close_guides(page):
selectors = [
'button[aria-label="닫기"]',
'button:has-text("닫기")',
'button:has-text("확인")',
'a:has-text("닫기")',
".btn_close",
".popup_close",
]
for selector in selectors:
try:
locator = page.locator(selector)
if locator.count() > 0:
print(f"[GUIDE CLOSE] {selector} / count={locator.count()}")
locator.first.click(
force=True,
timeout=2000
)
page.wait_for_timeout(800)
except Exception:
pass
def install_naver_dialog_handler(page):
"""
V1.5 안전장치:
네이버가 DOM 팝업이 아니라 browser dialog/alert로 띄우는 경우도 대비한다.
DOM 팝업(.se-popup-alert-confirm)은 handle_existing_draft_popup()에서 처리한다.
"""
try:
def _on_dialog(dialog):
try:
message = dialog.message
except Exception:
message = ""
print("[NAVER DIALOG DETECTED]", message)
try:
# 작성중 글/임시글 계열은 새 글 작성을 위해 dismiss 우선
dialog.dismiss()
print("[NAVER DIALOG DISMISSED]", message)
except Exception as exc:
try:
dialog.accept()
print("[NAVER DIALOG ACCEPTED]", message)
except Exception as exc2:
print("[NAVER DIALOG HANDLE ERROR]", f"{type(exc).__name__}: {exc}", f"{type(exc2).__name__}: {exc2}")
page.on("dialog", _on_dialog)
print("[NAVER DIALOG HANDLER INSTALLED]")
return True
except Exception as exc:
print("[NAVER DIALOG HANDLER INSTALL ERROR]", f"{type(exc).__name__}: {exc}")
return False
def _click_existing_draft_popup_in_root(root, root_name="root"):
"""
네이버 스마트에디터 '작성 중인 글이 있습니다.' 팝업을 DOM 기준으로 직접 닫는다.
F12 확인 구조:
div.se-popup-alert-confirm[data-group="popupLayer"]
strong.se-popup-title = 작성 중인 글이 있습니다.
button.se-popup-button-cancel > span = 취소
핵심:
- page / write_frame / 모든 frame에서 탐색
- JS click + MouseEvent + 좌표 클릭 fallback까지 수행
"""
try:
result = root.evaluate("""
(rootName) => {
const norm = (s) => String(s || '').replace(/\\s+/g, ' ').trim();
const selectors = [
'.se-popup-alert-confirm',
'.se-popup.__se-sentry.se-popup-alert.se-popup-alert-confirm',
'.se-popup[data-group="popupLayer"]',
'div[data-name*="se-popup-alert-confirm"]'
];
let popup = null;
let candidates = [];
for (const selector of selectors) {
candidates = candidates.concat(Array.from(document.querySelectorAll(selector)));
}
// 중복 제거
candidates = Array.from(new Set(candidates));
for (const el of candidates) {
const text = norm(el.innerText || el.textContent || '');
const title = norm((el.querySelector('.se-popup-title') || {}).innerText || '');
if (
title.includes('작성 중인 글이 있습니다') ||
text.includes('작성 중인 글이 있습니다') ||
text.includes('이어서 작성하시겠습니까')
) {
popup = el;
break;
}
}
if (!popup) {
return {
found: false,
clicked: false,
root: rootName,
popup_count: candidates.length
};
}
const sample = norm(popup.innerText || popup.textContent || '').slice(0, 300);
let button =
popup.querySelector('button.se-popup-button-cancel') ||
popup.querySelector('.se-popup-button-cancel') ||
popup.querySelector('button[class*="cancel"]');
if (!button) {
const buttons = Array.from(popup.querySelectorAll('button, a, [role="button"], span'));
button = buttons.find((btn) => norm(btn.innerText || btn.textContent || '') === '취소')
|| buttons.find((btn) => norm(btn.innerText || btn.textContent || '').includes('취소'));
}
if (!button) {
return {
found: true,
clicked: false,
root: rootName,
reason: 'cancel_button_not_found',
sample
};
}
const target = button.closest('button, a, [role="button"]') || button;
try { target.scrollIntoView({block:'center', inline:'center'}); } catch(e) {}
const rect = target.getBoundingClientRect();
const x = rect.left + rect.width / 2;
const y = rect.top + rect.height / 2;
const evOpts = {bubbles:true, cancelable:true, view:window, clientX:x, clientY:y};
let click_error = '';
try {
target.focus && target.focus();
target.click();
} catch(e) {
click_error = String(e && e.message || e || '');
}
try {
target.dispatchEvent(new MouseEvent('mouseover', evOpts));
target.dispatchEvent(new MouseEvent('mousemove', evOpts));
target.dispatchEvent(new MouseEvent('mousedown', evOpts));
target.dispatchEvent(new MouseEvent('mouseup', evOpts));
target.dispatchEvent(new MouseEvent('click', evOpts));
} catch(e) {
click_error = click_error || String(e && e.message || e || '');
}
return {
found: true,
clicked: true,
root: rootName,
button_text: norm(target.innerText || target.textContent || ''),
sample,
rect: {x, y, width: rect.width, height: rect.height},
click_error
};
}
""", root_name)
return result
except Exception as e:
return {
"found": False,
"clicked": False,
"root": root_name,
"error": f"{type(e).__name__}: {e}",
}
def handle_existing_draft_popup(page, write_frame=None, label="", max_rounds=6):
"""
글쓰기 진입 후 뜨는 '작성 중인 글이 있습니다.' 팝업을 page + 모든 frame에서 반복 탐색한다.
frame이 이미 잡힌 뒤에도 팝업이 살아있을 수 있으므로 제목 입력 전 반드시 호출한다.
"""
handled = False
last_results = []
for round_idx in range(1, int(max_rounds or 6) + 1):
roots = []
if write_frame is not None:
roots.append(("write_frame", write_frame))
roots.append(("page", page))
try:
for idx, fr in enumerate(page.frames):
if write_frame is not None and fr == write_frame:
continue
name = fr.name or f"frame_{idx}"
roots.append((name, fr))
except Exception:
pass
clicked_this_round = False
for root_name, root in roots:
result = _click_existing_draft_popup_in_root(root, root_name=root_name)
if result.get("found"):
print("[EXISTING DRAFT POPUP FOUND]", f"label={label}", f"round={round_idx}", result)
last_results.append(result)
if result.get("clicked"):
print("[EXISTING DRAFT POPUP CANCEL CLICKED]", f"label={label}", f"round={round_idx}", result)
handled = True
clicked_this_round = True
try:
page.wait_for_timeout(900)
except Exception:
pass
break
if clicked_this_round:
continue
if handled:
print("[EXISTING DRAFT POPUP HANDLED]", f"label={label}", f"rounds={round_idx}")
else:
if round_idx == 1:
print("[EXISTING DRAFT POPUP NONE]", f"label={label}")
return {
"handled": handled,
"label": label,
"rounds": round_idx,
"results": last_results[-5:],
}
print("[EXISTING DRAFT POPUP HANDLE FINISH]", f"label={label}", f"handled={handled}")
return {
"handled": handled,
"label": label,
"rounds": int(max_rounds or 6),
"results": last_results[-5:],
}
def find_write_frame(page):
for frame in page.frames:
if "PostWriteForm.naver" in frame.url or "PostWriteForm" in frame.url:
return frame
return None
def remove_right_help_panel(write_frame):
try:
print("[RIGHT PANEL REMOVE START]")
remove_script = """
(() => {
const keywords = [
"What’s New",
"도움말",
"간편해진 태그 입력",
"간결해진 에디터 화면",
"기본 인터페이스 안내"
];
const all = Array.from(
document.querySelectorAll("div, section, aside, article")
);
all.forEach(el => {
try {
const text = el.innerText || "";
const rect = el.getBoundingClientRect();
if (
rect.width >= 250 &&
rect.height >= 200 &&
rect.left >= 900 &&
keywords.some(k => text.includes(k))
) {
el.remove();
}
} catch(e) {}
});
})();
"""
write_frame.evaluate(remove_script)
print("[RIGHT PANEL REMOVE END]")
except Exception as e:
print("[RIGHT PANEL REMOVE ERROR]", str(e))
def click_top_publish_button(page):
selectors = [
'button:has-text("발행")',
'button.publish_btn__m9KHH',
'.publish_btn__m9KHH',
]
for selector in selectors:
try:
locator = page.locator(selector)
if locator.count() <= 0:
continue
print("[PUBLISH TOP SELECTOR CLICK]", selector)
locator.first.click(
force=True,
timeout=5000
)
page.wait_for_timeout(1200)
print("[PUBLISH TOP SELECTOR CLICK SUCCESS]")
return True
except Exception as e:
print("[PUBLISH TOP SELECTOR CLICK ERROR]", selector, str(e))
try:
print("[PUBLISH TOP COORD CLICK START]")
page.mouse.click(1335, 22)
page.wait_for_timeout(1200)
print("[PUBLISH TOP COORD CLICK SUCCESS]")
return True
except Exception as e:
print("[PUBLISH TOP COORD CLICK ERROR]", str(e))
return False
def click_final_publish_button(page):
"""
V1.5 FINAL PUBLISH:
발행 설정창 하단의 실제 최종 '발행' 버튼을 우선 클릭한다.
중요:
- 상단 고정바의 '발행' 버튼(y가 작은 버튼)은 제외
- '발행 설정 닫기' 같은 닫기 버튼은 제외
- 현재 네이버 UI에서는 하단 최종 버튼 텍스트가 '발행하기'가 아니라 '발행'일 수 있음
"""
print("[FINAL PUBLISH CLICK START]")
roots = [("page", page)]
try:
for idx, frame in enumerate(page.frames):
roots.append((f"frame_{idx}", frame))
except Exception:
pass
selectors = [
'button:has-text("발행")',
'[role="button"]:has-text("발행")',
'button:has-text("발행하기")',
'[role="button"]:has-text("발행하기")',
'button.confirm_btn__WEaBq',
'.confirm_btn__WEaBq',
]
def is_good_final_button(text, box):
text = str(text or "").strip()
if not box:
return False
x = float(box.get("x") or 0)
y = float(box.get("y") or 0)
w = float(box.get("width") or 0)
h = float(box.get("height") or 0)
if y < 180:
return False
if w < 45 or h < 24:
return False
if "닫기" in text or "설정 닫기" in text or "취소" in text:
return False
if text in ["발행", "발행하기"]:
return True
# 버튼 내부에 줄바꿈/부가텍스트가 붙는 경우 대응
compact = text.replace("\n", " ").strip()
if compact in ["발행", "발행하기"]:
return True
return False
for root_name, root in roots:
for selector in selectors:
try:
loc = root.locator(selector)
count = loc.count()
if count <= 0:
continue
print(f"[FINAL PUBLISH SELECTOR FOUND] root={root_name} selector={selector} count={count}")
# 하단 최종 버튼은 보통 뒤쪽 index에 있으므로 역순 우선
for idx in reversed(range(min(count, 10))):
try:
target = loc.nth(idx)
try:
text = target.inner_text(timeout=1000)
except Exception:
text = ""
try:
box = target.bounding_box()
except Exception:
box = None
print(
"[FINAL PUBLISH CANDIDATE]",
f"root={root_name}",
f"selector={selector}",
f"idx={idx}",
f"text={str(text or '').strip()}",
f"box={box}"
)
if not is_good_final_button(text, box):
continue
target.scroll_into_view_if_needed(timeout=3000)
page.wait_for_timeout(300)
target.click(force=True, timeout=5000)
page.wait_for_timeout(10000)
print(
"[FINAL PUBLISH CLICK SUCCESS]",
f"root={root_name}",
f"selector={selector}",
f"idx={idx}",
f"text={str(text or '').strip()}"
)
return True
except Exception as e:
print("[FINAL PUBLISH CANDIDATE ERROR]", root_name, selector, idx, str(e))
except Exception as e:
print("[FINAL PUBLISH SELECTOR ERROR]", root_name, selector, str(e))
# 마지막 fallback: 로그에서 확인된 발행 설정창 하단 버튼 좌표 주변
for x, y in [(1294, 585), (1294, 556), (1239, 565)]:
try:
print("[FINAL PUBLISH COORD CLICK TRY]", x, y)
page.mouse.click(x, y)
page.wait_for_timeout(10000)
print("[FINAL PUBLISH COORD CLICK SUCCESS]", x, y)
return True
except Exception as e:
print("[FINAL PUBLISH COORD CLICK ERROR]", x, y, str(e))
print("[FINAL PUBLISH CLICK FAILED]")
return False
def wait_publish_result_url(page):
result_url = page.url
for i in range(30):
page.wait_for_timeout(1000)
current_url = page.url
print(f"[WAIT PUBLISH RESULT] {i + 1}/30 url={current_url}")
result_url = current_url
if (
"blog.naver.com/" in current_url
and "Redirect=Write" not in current_url
and "PostWriteForm" not in current_url
):
break
return result_url
def validate_publish_result(result_url):
if not result_url:
return False
result_url = result_url.strip()
if (
"blog.naver.com/" in result_url
and "Redirect=Write" not in result_url
and "PostWriteForm" not in result_url
):
return True
return False
def human_wait_before_publish(page, min_seconds=300, max_seconds=600):
"""
STEP103-01:
네이버 글쓰기 진입 후 최종 발행까지 수초만에 끝나는 패턴을 피하기 위한
발행 직전 검토 대기.
기본 5~10분 랜덤 대기.
"""
try:
min_seconds = int(min_seconds or 300)
max_seconds = int(max_seconds or 600)
except Exception:
min_seconds = 300
max_seconds = 600
if min_seconds < 0:
min_seconds = 0
if max_seconds < min_seconds:
max_seconds = min_seconds
wait_seconds = random.randint(min_seconds, max_seconds)
if wait_seconds <= 0:
print("[HUMAN WAIT BEFORE PUBLISH SKIP] wait_seconds=0")
return 0
print(
"[HUMAN WAIT BEFORE PUBLISH START]",
f"wait_seconds={wait_seconds}",
f"range={min_seconds}~{max_seconds}"
)
elapsed = 0
while elapsed < wait_seconds:
try:
if random.random() < 0.55:
page.mouse.wheel(0, random.randint(-450, 750))
else:
page.mouse.move(
random.randint(420, 1180),
random.randint(180, 760),
steps=random.randint(8, 22)
)
except Exception as e:
print("[HUMAN WAIT ACTION WARN]", str(e))
chunk = random.randint(25, 70)
chunk = min(chunk, wait_seconds - elapsed)
time.sleep(chunk)
elapsed += chunk
print(
"[HUMAN WAIT BEFORE PUBLISH PROGRESS]",
f"elapsed={elapsed}",
f"remain={max(0, wait_seconds - elapsed)}"
)
print("[HUMAN WAIT BEFORE PUBLISH DONE]", f"elapsed={elapsed}")
return wait_seconds
def click_write_button_or_goto(page, blog_id=None):
blog_id = str(blog_id or "").strip()
if blog_id:
write_url = f"https://blog.naver.com/{blog_id}?Redirect=Write&categoryNo=2"
print("[WRITE DIRECT BLOG_ID GOTO]", write_url)
page.goto(
write_url,
wait_until="domcontentloaded",
timeout=60000
)
page.wait_for_timeout(1200)
# 글쓰기 URL 진입 직후 작성중 글 DOM 팝업이 뜨는 경우가 가장 많다.
try:
handle_existing_draft_popup(page=page, write_frame=None, label="inside_click_write_blog_id_after_goto", max_rounds=8)
except Exception as exc:
print("[EXISTING DRAFT POPUP INSIDE WRITE WARN]", f"{type(exc).__name__}: {exc}")
return True
write_selectors = [
"a:has-text('글쓰기')",
"button:has-text('글쓰기')",
"a[href*='GoBlogWrite']",
"a[href*='PostWriteForm']",
"a[href*='Write']",
]
for selector in write_selectors:
try:
locator = page.locator(selector)
if locator.count() <= 0:
continue
print("[WRITE CLICK]", selector)
href = locator.first.get_attribute("href")
if href:
print("[WRITE DIRECT GOTO]", href)
page.goto(
href,
wait_until="domcontentloaded",
timeout=60000
)
page.wait_for_timeout(1200)
return True
try:
with page.context.expect_page(timeout=5000) as new_page_info:
locator.first.click(
force=True,
timeout=5000
)
new_page = new_page_info.value
new_page.wait_for_load_state(
"domcontentloaded",
timeout=60000
)
page = new_page
except Exception:
locator.first.click(
force=True,
timeout=5000
)
page.wait_for_timeout(1200)
return True
except Exception as e:
print("[WRITE CLICK FAIL]", selector, str(e))
return False
def input_title_like_old_worker(page, write_frame, publish_title):
"""
기존 성공하던 방식 유지:
- placeholder 클릭
- bounding_box 좌표 클릭
- keyboard.type
이 방식이 제목/본문 분리 안정성이 더 좋았음.
"""
try:
title_placeholder = write_frame.locator(
".se-section-documentTitle .se-placeholder"
).first
title_placeholder.click(force=True)
page.wait_for_timeout(1000)
box = title_placeholder.bounding_box()
print("[TITLE PLACEHOLDER BOX]", box)
if box:
page.mouse.click(
box["x"] + 30,
box["y"] + 20
)
page.wait_for_timeout(1000)
page.keyboard.type(
publish_title[:80],
delay=80
)
page.wait_for_timeout(1200)
print("[TITLE INPUT SUCCESS]")
return True
except Exception as e:
print("[TITLE INPUT ERROR]", str(e))
return False
def build_preview_html(clipboard_html):
html = str(clipboard_html or "")
return f"""
Naver Blog Preview Copy
{html}
"""
def copy_rendered_preview_to_clipboard(page, clipboard_html):
"""
HTML 소스를 직접 clipboard에 쓰지 않고,
preview 페이지에서 렌더링된 본문을 Ctrl+A/C로 복사한다.
"""
preview_page = None
try:
print("[PREVIEW COPY START]")
preview_page = page.context.new_page()
preview_page.set_content(
build_preview_html(clipboard_html),
wait_until="load",
timeout=60000
)
try:
image_count = preview_page.locator("img").count()
except Exception:
image_count = 0
print(f"[PREVIEW IMAGE COUNT] {image_count}")
preview_page.wait_for_timeout(2500)
try:
preview_page.wait_for_function(
"""
() => Array.from(document.images).every(img => img.complete)
""",
timeout=10000
)
print("[PREVIEW IMAGE LOAD OK]")
except Exception as e:
print("[PREVIEW IMAGE LOAD TIMEOUT]", str(e))
try:
broken_count = preview_page.evaluate("""
() => Array.from(document.images).filter(img => !img.complete || img.naturalWidth === 0).length
""")
print(f"[PREVIEW BROKEN IMAGE COUNT] {broken_count}")
except Exception as e:
print("[PREVIEW BROKEN IMAGE CHECK ERROR]", str(e))
preview_page.locator("#copy-root").click(
force=True,
timeout=5000
)
preview_page.wait_for_timeout(500)
preview_page.keyboard.press("Control+A")
preview_page.wait_for_timeout(500)
preview_page.keyboard.press("Control+C")
preview_page.wait_for_timeout(1200)
print("[PREVIEW COPY DONE]")
return True
except Exception as e:
print("[PREVIEW COPY ERROR]", str(e))
return False
finally:
try:
if preview_page:
preview_page.close()
except Exception:
pass
def save_editor_debug_snapshot(write_frame, queue_id=None, article_no="", label="after_paste"):
"""
네이버 에디터 내부에 실제로 남은 HTML/TEXT를 저장한다.
"""
try:
debug_dir = os.path.join(STORAGE_DIR, "debug", "publish_editor")
os.makedirs(debug_dir, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_queue = str(queue_id or "unknown")
safe_article = re.sub(r"[^0-9A-Za-z_-]+", "_", str(article_no or "")) or "article"
html_path = os.path.join(debug_dir, f"editor_{safe_queue}_{safe_article}_{label}_{ts}.html")
txt_path = os.path.join(debug_dir, f"editor_{safe_queue}_{safe_article}_{label}_{ts}.txt")
editor_html = write_frame.evaluate("() => document.body ? document.body.innerHTML : ''")
try:
editor_text = write_frame.locator("body").inner_text(timeout=5000)
except Exception:
editor_text = ""
with open(html_path, "w", encoding="utf-8") as f:
f.write(str(editor_html or ""))
with open(txt_path, "w", encoding="utf-8") as f:
f.write(str(editor_text or ""))
print("[EDITOR DEBUG SAVED]", html_path.replace("\\", "/"))
print("[EDITOR DEBUG TEXT SAVED]", txt_path.replace("\\", "/"))
print("[EDITOR DEBUG LEN]", "html=", len(str(editor_html or "")), "text=", len(str(editor_text or "")))
return {
"html_path": html_path,
"txt_path": txt_path,
"html_len": len(str(editor_html or "")),
"text_len": len(str(editor_text or "")),
"text": str(editor_text or ""),
}
except Exception as e:
print("[EDITOR DEBUG SAVE ERROR]", str(e))
return {
"html_path": "",
"txt_path": "",
"html_len": 0,
"text_len": 0,
"text": "",
"error": str(e),
}
def validate_editor_content_after_paste(write_frame, queue_id=None, article_no=""):
"""
Ctrl+V 이후 네이버 에디터가 실제 본문을 받아들였는지 검증한다.
"""
snap = save_editor_debug_snapshot(
write_frame=write_frame,
queue_id=queue_id,
article_no=article_no,
label="after_paste"
)
text = str(snap.get("text") or "")
html_len = int(snap.get("html_len") or 0)
text_len = int(snap.get("text_len") or 0)
markers = [
"매물 핵심 정보",
"매물 첫인상",
"추천 포인트",
"중개대상물 표시",
"상담 및 중개사 안내",
]
hit_count = sum(1 for marker in markers if marker in text)
print("[EDITOR VALIDATE]", f"html_len={html_len}", f"text_len={text_len}", f"hit_count={hit_count}")
if text_len < 300:
return False, f"에디터 본문 길이 부족 text_len={text_len}, html_len={html_len}"
if hit_count < 2:
return False, f"에디터 본문 핵심문구 부족 hit_count={hit_count}, text_len={text_len}"
return True, "에디터 본문 검증 성공"
def move_cursor_after_representative_image(page, write_frame):
"""
대표이미지 직접 업로드 후 본문 붙여넣기 위치를 이미지 아래로 옮긴다.
"""
try:
print("[CURSOR AFTER REP MOVE START]")
page.keyboard.press("Escape")
page.wait_for_timeout(500)
try:
page.mouse.wheel(0, 700)
page.wait_for_timeout(500)
except Exception:
pass
for x, y in [(520, 760), (520, 820), (620, 760)]:
try:
page.mouse.click(x, y)
page.wait_for_timeout(400)
break
except Exception:
continue
for _ in range(2):
try:
page.keyboard.press("Enter")
page.wait_for_timeout(250)
except Exception:
pass
print("[CURSOR AFTER REP MOVE END]")
return True
except Exception as e:
print("[CURSOR AFTER REP MOVE ERROR]", str(e))
return False
def strip_representative_image_from_html(html):
"""
대표이미지는 네이버 에디터에 직접 업로드하여 대표 체크 대상이 되게 한다.
따라서 HTML 본문에 들어간 CDN 대표이미지 블록은 중복 방지를 위해 제거한다.
generate_blog_drafts.py의 build_header_image_html() 구조를 우선 제거하고,
예외적으로 첫 realestate-representative-image 블록도 제거한다.
"""
html = str(html or "")
if not html:
return html
patterns = [
r']*class=["\']realestate-representative-image["\'][\s\S]*?\s*',
r']*realestate-representative-image[^>]*>[\s\S]*?\s*',
]
cleaned = html
for pattern in patterns:
new_cleaned, count = re.subn(
pattern,
"",
cleaned,
count=1,
flags=re.IGNORECASE
)
if count > 0:
print(f"[REP HTML STRIP] representative image block removed count={count}")
cleaned = new_cleaned
break
return cleaned.strip()
def extract_representative_image_path_from_queue(queue):
"""
source_json.header_image.image_path에서 로컬 대표이미지 파일 경로를 찾는다.
CDN URL이 아니라 D:/honghee/blog_api/mybox/generated/... PNG 파일이 필요하다.
"""
try:
source = json.loads(queue.get("source_json") or "{}")
except Exception:
source = {}
header_image = source.get("header_image") or {}
candidates = [
header_image.get("image_path"),
header_image.get("local_path"),
header_image.get("original_image_path"),
source.get("header_image_path"),
source.get("representative_image_path"),
]
for path in candidates:
path = str(path or "").strip().replace("\\", "/")
if path and os.path.exists(path):
return path
return ""
def try_set_representative_checkbox(page, write_frame):
"""
네이버 에디터에서 대표이미지 체크를 안정적으로 시도한다.
핵심 로그:
[REP BUTTON FOUND]
[REP BUTTON CLICKED]
[REP THUMBNAIL CONFIRMED]
"""
print("[REP CHECK TRY]")
for wait_i in range(8):
page.wait_for_timeout(1000)
candidates = [
("frame", write_frame, "label:has-text('대표')"),
("frame", write_frame, "button:has-text('대표')"),
("frame", write_frame, "span:has-text('대표')"),
("frame", write_frame, "div[role='button']:has-text('대표')"),
("page", page, "label:has-text('대표')"),
("page", page, "button:has-text('대표')"),
("page", page, "span:has-text('대표')"),
("page", page, "div[role='button']:has-text('대표')"),
]
for root_name, root, selector in candidates:
try:
loc = root.locator(selector)
count = loc.count()
if count <= 0:
continue
print(f"[REP BUTTON FOUND] wait={wait_i + 1}, root={root_name}, selector={selector}, count={count}")
for idx in range(min(count, 5)):
try:
target = loc.nth(idx)
box = target.bounding_box()
if box:
if box.get("width", 0) > 420 or box.get("height", 0) > 170:
continue
target.click(force=True, timeout=3000)
page.wait_for_timeout(1000)
print(f"[REP BUTTON CLICKED] root={root_name}, selector={selector}, index={idx}")
try:
checked_count = root.locator("input[type='checkbox']:checked").count()
if checked_count > 0:
print(f"[REP THUMBNAIL CONFIRMED] checked_count={checked_count}")
return True
except Exception:
pass
print("[REP THUMBNAIL CONFIRMED] representative click attempted")
return True
except Exception as e:
print(f"[REP BUTTON CLICK ERROR] root={root_name}, selector={selector}, index={idx}, error={e}")
except Exception as e:
print(f"[REP BUTTON SCAN ERROR] root={root_name}, selector={selector}, error={e}")
for root_name, root in [("frame", write_frame), ("page", page)]:
try:
checks = root.locator("input[type='checkbox']")
count = checks.count()
if count <= 0:
continue
print(f"[REP CHECKBOX FOUND] wait={wait_i + 1}, root={root_name}, count={count}")
for idx in range(min(count, 5)):
try:
cb = checks.nth(idx)
try:
if cb.is_checked(timeout=1000):
print(f"[REP THUMBNAIL CONFIRMED] already checked root={root_name}, index={idx}")
return True
except Exception:
pass
cb.click(force=True, timeout=3000)
page.wait_for_timeout(800)
print(f"[REP BUTTON CLICKED] checkbox root={root_name}, index={idx}")
try:
if cb.is_checked(timeout=1000):
print(f"[REP THUMBNAIL CONFIRMED] checkbox checked root={root_name}, index={idx}")
return True
except Exception:
print("[REP THUMBNAIL CONFIRMED] checkbox click attempted")
return True
except Exception as e:
print(f"[REP CHECKBOX CLICK ERROR] root={root_name}, index={idx}, error={e}")
except Exception as e:
print(f"[REP CHECKBOX SCAN ERROR] root={root_name}, error={e}")
print("[REP CHECK NOT FOUND]")
return False
def upload_representative_image_to_editor(page, write_frame, image_path):
"""
네이버 블로그 대표이미지만 에디터에 직접 업로드한다.
직접 업로드된 이미지는 네이버 에디터의 대표 체크 대상이 된다.
성공 기준:
- file chooser 또는 input[type=file] set_input_files 성공
- 업로드 후 일정 시간 대기
"""
image_path = str(image_path or "").strip().replace("\\", "/")
if not image_path or not os.path.exists(image_path):
print("[REP UPLOAD SKIP] representative image path missing")
return False
print("[REP UPLOAD START]", image_path)
# 본문 위치 클릭
try:
page.mouse.click(520, 430)
page.wait_for_timeout(800)
except Exception:
pass
image_button_selectors = [
'button:has-text("사진")',
'button:has-text("이미지")',
'button:has-text("사진 첨부")',
'button:has-text("첨부")',
'a:has-text("사진")',
'a:has-text("이미지")',
'span:has-text("사진")',
'span:has-text("이미지")',
'[aria-label*="사진"]',
'[aria-label*="이미지"]',
'[title*="사진"]',
'[title*="이미지"]',
'.se-toolbar-item-image button',
'.se-toolbar-item-image',
]
# 1) 이미 존재하는 file input 직접 처리
for root_name, root in [("frame", write_frame), ("page", page)]:
try:
inputs = root.locator("input[type='file']")
count = inputs.count()
print(f"[REP FILE INPUT COUNT] root={root_name}, count={count}")
for i in range(count):
try:
inputs.nth(i).set_input_files(image_path, timeout=5000)
page.wait_for_timeout(8000)
print(f"[REP UPLOAD OK] direct input root={root_name}, index={i}")
print("[REP IMAGE UPLOADED] direct input")
try_set_representative_checkbox(page, write_frame)
return True
except Exception as e:
print(f"[REP FILE INPUT ERROR] root={root_name}, index={i}, error={e}")
except Exception as e:
print(f"[REP FILE INPUT SCAN ERROR] root={root_name}, error={e}")
# 2) 사진/이미지 버튼 클릭 후 file chooser 처리
last_error = ""
for root_name, root in [("frame", write_frame), ("page", page)]:
for selector in image_button_selectors:
try:
loc = root.locator(selector).first
if loc.count() <= 0:
continue
print(f"[REP IMAGE BUTTON TRY] root={root_name}, selector={selector}")
with page.expect_file_chooser(timeout=7000) as fc_info:
loc.click(force=True, timeout=5000)
chooser = fc_info.value
chooser.set_files(image_path)
page.wait_for_timeout(10000)
print(f"[REP UPLOAD OK] file chooser root={root_name}, selector={selector}")
print("[REP IMAGE UPLOADED] file chooser")
try_set_representative_checkbox(page, write_frame)
return True
except Exception as e:
last_error = f"root={root_name}, selector={selector}, error={e}"
print("[REP IMAGE BUTTON ERROR]", last_error)
continue
# 3) 버튼 클릭 후 생성된 input[type=file] 처리
for root_name, root in [("frame", write_frame), ("page", page)]:
for selector in image_button_selectors:
try:
loc = root.locator(selector).first
if loc.count() <= 0:
continue
loc.click(force=True, timeout=5000)
page.wait_for_timeout(1500)
inputs = root.locator("input[type='file']")
count = inputs.count()
for i in range(count):
try:
inputs.nth(i).set_input_files(image_path, timeout=5000)
page.wait_for_timeout(10000)
print(f"[REP UPLOAD OK] post-click input root={root_name}, selector={selector}, index={i}")
print("[REP IMAGE UPLOADED] post-click input")
try_set_representative_checkbox(page, write_frame)
return True
except Exception as e:
last_error = f"root={root_name}, selector={selector}, input={i}, error={e}"
print("[REP POST CLICK INPUT ERROR]", last_error)
except Exception as e:
last_error = f"root={root_name}, selector={selector}, error={e}"
print("[REP POST CLICK ERROR]", last_error)
print("[REP UPLOAD FAIL]", last_error)
return False
def input_body_like_old_worker(page, write_frame, clipboard_html, plain_text, queue_id=None, article_no=''):
"""
기존 성공하던 방식 유지:
- 본문 placeholder 탐색
- 제목 placeholder 제외
- 본문 bounding_box 좌표 클릭
- HTML clipboard paste
"""
try:
body_placeholders = write_frame.locator(
".se-component-content .se-placeholder"
)
body_count = body_placeholders.count()
print("[BODY PLACEHOLDER COUNT]", body_count)
clicked_body = False
for i in range(body_count):
placeholder = body_placeholders.nth(i)
try:
text = placeholder.inner_text(timeout=3000)
except Exception:
text = ""
if "제목" in text:
continue
box = placeholder.bounding_box()
print(f"[BODY PLACEHOLDER {i}] text={text} box={box}")
if not box:
continue
placeholder.click(force=True)
page.wait_for_timeout(800)
page.mouse.click(
box["x"] + 50,
box["y"] + 20
)
page.wait_for_timeout(800)
clicked_body = True
break
if not clicked_body:
page.mouse.click(520, 430)
page.wait_for_timeout(1000)
if clipboard_html:
print("[BODY INPUT MODE] rendered_preview_copy")
copied = copy_rendered_preview_to_clipboard(
page,
clipboard_html
)
if not copied:
raise Exception("preview rendered copy failed")
page.bring_to_front()
page.wait_for_timeout(800)
try:
if clicked_body and box:
page.mouse.click(
box["x"] + 50,
box["y"] + 20
)
else:
page.mouse.click(520, 430)
page.wait_for_timeout(800)
except Exception:
page.mouse.click(520, 430)
page.wait_for_timeout(800)
print("[RENDERED PASTE START]")
page.keyboard.press("Control+V")
page.wait_for_timeout(5000)
print("[RENDERED PASTE SUCCESS]")
validate_ok, validate_message = validate_editor_content_after_paste(
write_frame=write_frame,
queue_id=queue_id,
article_no=article_no,
)
print("[EDITOR VALIDATE RESULT]", validate_ok, validate_message)
if not validate_ok:
raise Exception("네이버 에디터 본문 검증 실패: " + validate_message)
else:
print("[BODY INPUT MODE] plain_text")
if not plain_text:
plain_text = "매물 상세 내용은 상담을 통해 안내드리겠습니다."
page.keyboard.type(
plain_text[:2500],
delay=20
)
page.wait_for_timeout(1200)
print("[BODY INPUT SUCCESS]")
return True
except Exception as e:
print("[BODY INPUT ERROR]", str(e))
return False
def publish_one(queue):
queue_id = queue["id"]
realtor_id = queue["realtor_id"]
draft_id = queue["draft_id"]
article_no = str(
queue.get("queue_article_no")
or queue.get("draft_article_no")
or queue.get("article_no")
or ""
)
clipboard_len = int(queue.get("clipboard_len") or 0)
draft_len = int(queue.get("draft_len") or 0)
plain_len = int(queue.get("plain_len") or 0)
print("=" * 80)
print(
f"[QUEUE START] queue_id={queue_id}, "
f"realtor_id={realtor_id}, draft_id={draft_id}, article_no={article_no}"
)
print(
f"[DRAFT HTML CHECK] clipboard_len={clipboard_len}, "
f"draft_len={draft_len}, plain_len={plain_len}"
)
print("=" * 80)
integrity_ok, integrity_message = validate_queue_draft_integrity(queue)
print(integrity_message)
if not integrity_ok:
update_queue_status(
queue_id,
"hold",
integrity_message[:1000]
)
return False
session_file = get_session_file(realtor_id)
blog_id = get_naver_blog_id(realtor_id)
print("[BLOG ID]", blog_id or "")
print("[SESSION FILE]", session_file)
if not os.path.exists(session_file):
update_naver_session_status(
realtor_id,
"missing",
f"세션 파일 없음: {session_file}",
session_file=session_file
)
mark_session_hold(
queue_id,
realtor_id,
f"세션 파일 없음: {session_file}"
)
return False
update_queue_status(queue_id, "processing")
browser = None
try:
with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
]
)
context = browser.new_context(
storage_state=session_file,
locale="ko-KR",
viewport={
"width": 1400,
"height": 1000
}
)
page = context.new_page()
print("[V15 STEP112 HOTFIX ACTIVE] existing_draft_popup/blog_id/session")
install_naver_dialog_handler(page)
print("[OPEN BLOG HOME] https://blog.naver.com")
page.goto(
"https://blog.naver.com",
wait_until="domcontentloaded",
timeout=60000
)
page.wait_for_timeout(1200)
print("[CURRENT URL]", page.url)
if is_login_page_url(page.url):
recovered, recover_message = recover_login_page_once(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=f"블로그 홈 접속 중 로그인 페이지로 이동: {page.url}",
target_url="https://blog.naver.com",
)
if not recovered:
try:
browser.close()
except Exception:
pass
mark_session_hold(
queue_id,
realtor_id,
f"블로그 홈 접속 중 로그인 페이지로 이동 / 자동복구 실패: {recover_message}"
)
return False
print("[SESSION RECOVERED] blog home")
try:
context.storage_state(path=session_file)
update_naver_session_status(
realtor_id,
"linked",
"발행 전 blog.naver.com 접속 성공 / 세션 재저장",
session_file=session_file
)
print("[SESSION REFRESHED] storage_state 재저장 완료")
except Exception as e:
print("[SESSION REFRESH ERROR]", str(e))
print("[LOGIN OK] 네이버 로그인 상태 확인 완료")
close_guides(page)
clicked = click_write_button_or_goto(page, blog_id=blog_id)
if clicked:
handle_existing_draft_popup(
page=page,
write_frame=None,
label="after_write_goto",
max_rounds=8,
)
if not clicked:
browser.close()
update_queue_status(
queue_id,
"failed",
"글쓰기 버튼 클릭 실패"
)
update_draft_status(
draft_id,
"failed"
)
print("[FAILED] 글쓰기 버튼 클릭 실패")
return False
if is_login_page_url(page.url):
recovered, recover_message = recover_login_page_once(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=f"글쓰기 이동 중 로그인 페이지로 이동: {page.url}",
target_url=f"https://blog.naver.com/{blog_id}?Redirect=Write&categoryNo=2",
)
if recovered:
print("[SESSION RECOVERED] write redirect")
clicked = click_write_button_or_goto(page, blog_id=blog_id)
if clicked:
handle_existing_draft_popup(
page=page,
write_frame=None,
label="after_relogin_write_goto",
max_rounds=8,
)
if not clicked:
print("[SESSION RECOVERED BUT WRITE RETRY FAILED]")
else:
try:
browser.close()
except Exception:
pass
mark_session_hold(
queue_id,
realtor_id,
f"글쓰기 이동 중 로그인 페이지로 이동 / 자동복구 실패: {recover_message}"
)
return False
write_frame = None
for i in range(30):
page.wait_for_timeout(1000)
print(f"[WAIT EDITOR] {i + 1}/30 url={page.url}")
handle_existing_draft_popup(
page=page,
write_frame=None,
label=f"wait_editor_{i + 1}",
max_rounds=3,
)
if is_login_page_url(page.url):
recovered, recover_message = recover_login_page_once(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
reason=f"에디터 대기 중 로그인 페이지로 이동: {page.url}",
target_url=f"https://blog.naver.com/{blog_id}?Redirect=Write&categoryNo=2",
)
if recovered:
print("[SESSION RECOVERED] editor wait")
clicked = click_write_button_or_goto(page, blog_id=blog_id)
if clicked:
handle_existing_draft_popup(
page=page,
write_frame=None,
label="after_relogin_editor_wait_write_goto",
max_rounds=8,
)
page.wait_for_timeout(1500)
continue
try:
browser.close()
except Exception:
pass
mark_session_hold(
queue_id,
realtor_id,
f"에디터 대기 중 로그인 페이지로 이동 / 자동복구 실패: {recover_message}"
)
return False
write_frame = find_write_frame(page)
if write_frame is not None:
break
print("[WRITE FINAL URL]", page.url)
write_frame = find_write_frame(page)
if write_frame is None:
final_url = str(page.url or "")
try:
browser.close()
except Exception:
pass
if is_login_page_url(final_url):
update_naver_session_status(
realtor_id,
"expired",
f"글쓰기 iframe 미발견 / 로그인 페이지 감지: {final_url}",
session_file=session_file
)
mark_session_hold(
queue_id,
realtor_id,
f"글쓰기 iframe 미발견 / 로그인 페이지 감지: {final_url}"
)
else:
mark_write_hold(
queue_id,
realtor_id,
f"글쓰기 iframe 미발견 / 최종 URL: {final_url}"
)
return False
print("[WRITE FRAME FOUND]", write_frame.url)
handle_existing_draft_popup(
page=page,
write_frame=write_frame,
label="after_write_frame_found_before_title",
max_rounds=10,
)
# 팝업 처리 뒤 frame 객체가 바뀌었을 수 있어 한 번 더 잡는다.
write_frame = find_write_frame(page) or write_frame
# -------------------------------------------------
# 제목 입력
# -------------------------------------------------
publish_title = str(
queue.get("draft_title")
or queue.get("publish_title")
or "부동산 매물 안내"
).strip()
title_ok = input_title_like_old_worker(
page,
write_frame,
publish_title
)
if not title_ok:
browser.close()
update_queue_status(
queue_id,
"failed",
"제목 입력 실패"
)
update_draft_status(
draft_id,
"failed"
)
return False
# -------------------------------------------------
# 본문 입력
# -------------------------------------------------
# 핵심: preview 렌더링 복사 방식 고정.
# clipboard_html이 비어 있으면 draft_html로 fallback.
# input_body_like_old_worker 내부에서는 queue를 절대 참조하지 않는다.
clipboard_html = str(
queue.get("clipboard_html")
or queue.get("draft_html")
or ""
).strip()
plain_text = str(
queue.get("plain_text") or ""
).strip()
# -------------------------------------------------
# 대표이미지 직접 업로드
# -------------------------------------------------
# 외부 CDN 이미지는 본문에는 보이지만 네이버 에디터의 '대표' 체크 대상이
# 안 될 수 있다. 따라서 대표이미지만 로컬 PNG 파일로 직접 업로드하고,
# HTML 본문에서는 CDN 대표이미지 블록을 제거해 중복을 방지한다.
handle_existing_draft_popup(
page=page,
write_frame=write_frame,
label="before_representative_upload",
max_rounds=5,
)
representative_image_path = extract_representative_image_path_from_queue(queue)
if representative_image_path:
rep_uploaded = upload_representative_image_to_editor(
page,
write_frame,
representative_image_path
)
print("[REP UPLOAD RESULT]", rep_uploaded)
if rep_uploaded:
clipboard_html = strip_representative_image_from_html(clipboard_html)
move_cursor_after_representative_image(page, write_frame)
else:
print("[REP UPLOAD WARN] 대표이미지 직접 업로드 실패. CDN 본문 이미지를 유지합니다.")
else:
print("[REP UPLOAD SKIP] source_json.header_image.image_path 없음")
handle_existing_draft_popup(
page=page,
write_frame=write_frame,
label="before_body_input",
max_rounds=5,
)
print(
"[BODY HTML CHECK BEFORE INPUT]",
"clipboard_len=",
len(clipboard_html),
"plain_len=",
len(plain_text),
)
body_ok = input_body_like_old_worker(
page,
write_frame,
clipboard_html,
plain_text,
queue_id=queue_id,
article_no=article_no,
)
if not body_ok:
browser.close()
update_queue_status(
queue_id,
"failed",
"본문 입력 실패"
)
update_draft_status(
draft_id,
"failed"
)
return False
# -------------------------------------------------
# 에디터 포커스 제거
# -------------------------------------------------
try:
print("[EDITOR BLUR START]")
page.keyboard.press("Escape")
page.wait_for_timeout(1000)
page.mouse.click(1200, 200)
page.wait_for_timeout(1000)
page.keyboard.press("Escape")
page.wait_for_timeout(1000)
print("[EDITOR BLUR END]")
except Exception as e:
print("[EDITOR BLUR ERROR]", str(e))
remove_right_help_panel(write_frame)
page.wait_for_timeout(1000)
# -------------------------------------------------
# 사람처럼 검토 시간
# -------------------------------------------------
# 중요:
# 발행 설정창을 먼저 열어둔 채 5~10분 기다리면 패널 상태가 바뀌거나
# 최종 버튼 클릭이 빗나갈 수 있다.
# 따라서 사람처럼 검토 대기는 글쓰기 화면에서 먼저 진행하고,
# 그 다음 상단 발행 버튼 → 최종 발행 버튼 순서로 즉시 처리한다.
human_wait_before_publish(
page,
min_seconds=300,
max_seconds=600
)
# -------------------------------------------------
# 상단 발행
# -------------------------------------------------
top_publish_clicked = click_top_publish_button(page)
if not top_publish_clicked:
browser.close()
update_queue_status(
queue_id,
"failed",
"상단 발행 버튼 클릭 실패"
)
update_draft_status(
draft_id,
"failed"
)
return False
page.wait_for_timeout(1800)
# -------------------------------------------------
# 최종 발행
# -------------------------------------------------
final_publish_clicked = click_final_publish_button(page)
if not final_publish_clicked:
browser.close()
update_queue_status(
queue_id,
"failed",
"최종 발행 버튼 클릭 실패"
)
update_draft_status(
draft_id,
"failed"
)
return False
# -------------------------------------------------
# 발행 결과
# -------------------------------------------------
result_url = wait_publish_result_url(page)
print("[RESULT URL CHECK]", result_url)
is_valid_publish = validate_publish_result(result_url)
if not is_valid_publish:
browser.close()
update_queue_status(
queue_id,
"failed",
"발행 URL 검증 실패",
result_url
)
update_draft_status(
draft_id,
"failed"
)
return False
private_stats = process_private_queues_for_realtor(
page=page,
context=context,
realtor_id=realtor_id,
session_file=session_file,
limit=3
)
print("[PRIVATE QUEUE RESULT]", private_stats)
browser.close()
blog_id, blog_post_no = parse_naver_blog_result_url(result_url)
update_queue_status(
queue_id,
"published",
error_message="발행 완료",
result_url=result_url,
blog_id=blog_id,
blog_post_no=blog_post_no
)
update_draft_publish_result(
draft_id,
result_url
)
print("[DONE] 네이버 블로그 발행 완료")
print("[RESULT URL]", result_url)
print("[BLOG RESULT PARSED]", f"blog_id={blog_id}", f"blog_post_no={blog_post_no}")
return True
except Exception as e:
traceback.print_exc()
try:
if browser:
browser.close()
except Exception:
pass
update_queue_status(
queue_id,
"failed",
str(e)[:1000]
)
update_draft_status(
draft_id,
"failed"
)
print("[FAILED]", str(e))
return False
def should_stop_by_hour(until_hour):
if until_hour is None:
return False
now_hour = datetime.now().hour
return now_hour >= int(until_hour)
def parse_hhmm_time(value):
"""
HH:MM 형식 문자열을 (hour, minute)로 변환한다.
예: 07:30, 7:30, 0730
"""
raw = str(value or "").strip()
if not raw:
return None
try:
if ":" in raw:
hour_text, minute_text = raw.split(":", 1)
elif len(raw) in [3, 4] and raw.isdigit():
hour_text = raw[:-2]
minute_text = raw[-2:]
else:
raise ValueError("HH:MM 형식이 아닙니다.")
hour = int(hour_text)
minute = int(minute_text)
if hour < 0 or hour > 23 or minute < 0 or minute > 59:
raise ValueError("시간 범위 오류")
return hour, minute
except Exception as e:
raise argparse.ArgumentTypeError(f"--until-time 값 오류: {value} / 예: 07:30 / {e}")
def get_next_until_datetime(until_time):
"""
현재 발행 윈도우에서 until_time 종료 시각을 datetime으로 반환한다.
야간 발행 기준으로 07:30 같은 다음날 새벽 시간을 자연스럽게 처리한다.
"""
parsed = parse_hhmm_time(until_time)
if not parsed:
return None
hour, minute = parsed
now = datetime.now()
target = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
# 현재가 12:00~23:59인데 종료시각이 07:30이면 다음날 07:30으로 본다.
if target <= now and now.hour >= 12:
target = target + timedelta(days=1)
return target
def seconds_until_time(until_time):
target = get_next_until_datetime(until_time)
if not target:
return 0
return max(0, int((target - datetime.now()).total_seconds()))
def should_stop_by_time(until_time):
if not until_time:
return False
parsed = parse_hhmm_time(until_time)
if not parsed:
return False
hour, minute = parsed
now = datetime.now()
now_minutes = now.hour * 60 + now.minute
until_minutes = hour * 60 + minute
# 야간 발행 종료시각은 보통 00:00~11:59 사이.
# 오전 시간대에는 해당 시각 도달 여부로 중지 판단.
if now.hour < 12:
return now_minutes >= until_minutes
# 낮/저녁에는 아직 다음날 종료시각 전이므로 중지하지 않는다.
return False
def run_publish_worker(limit=1, realtor_id=None, sleep_seconds=45, until_hour=None, until_time=None, stale_minutes=90, retry_session_hold=False, only_night_window=False, window_start_hour=22, window_end_hour=8, max_retry=2, auto_reset_retry_failed=True, smart_sleep=False, min_sleep=20, max_sleep=900, sleep_jitter=0.5, estimated_publish_seconds=90, per_realtor_daily_limit=1):
"""
pending 큐를 반복 처리한다.
기존 발행 흐름은 유지하고, STEP26 운영 안정화용 실행 기록만 추가한다.
- blog_pipeline_runs: worker 시작/종료 요약
- blog_pipeline_logs: 큐별 성공/실패 로그
- blog_publish_queue.worker_status / last_error_type / worker_server 기록
"""
run_id = create_pipeline_run(
run_type="publish_worker",
meta={
"limit": limit,
"realtor_id": realtor_id,
"sleep_seconds": sleep_seconds,
"until_hour": until_hour,
"until_time": until_time,
"stale_minutes": stale_minutes,
"retry_session_hold": retry_session_hold,
"only_night_window": only_night_window,
"window_start_hour": window_start_hour,
"window_end_hour": window_end_hour,
"max_retry": max_retry,
"auto_reset_retry_failed": auto_reset_retry_failed,
"smart_sleep": smart_sleep,
"min_sleep": min_sleep,
"max_sleep": max_sleep,
"sleep_jitter": sleep_jitter,
"estimated_publish_seconds": estimated_publish_seconds,
"per_realtor_daily_limit": per_realtor_daily_limit,
"worker_server": get_worker_server_name(),
}
)
success = 0
failed = 0
processed = 0
skipped = 0
final_status = "success"
final_message = "publish_worker completed"
total_publish_seconds = 0.0
attempted_realtor_ids = set()
try:
if only_night_window and not is_within_publish_window(window_start_hour, window_end_hour):
final_status = "skipped"
final_message = (
f"outside publish window: {window_start_hour}:00~{window_end_hour}:00 / "
f"next_start={next_publish_window_start(window_start_hour)}"
)
print("[PUBLISH WINDOW SKIP]", final_message)
add_pipeline_log(
run_id=run_id,
level="info",
step_name="publish_window_skip",
message=final_message,
realtor_id=realtor_id,
context={
"window_start_hour": window_start_hour,
"window_end_hour": window_end_hour,
"next_start": next_publish_window_start(window_start_hour),
}
)
return
reset_stale_processing(minutes=stale_minutes)
if auto_reset_retry_failed:
retry_reset_count = reset_retryable_failed_queues(
realtor_id=realtor_id,
max_retry=max_retry,
limit=max(int(limit), 20)
)
if retry_reset_count:
add_pipeline_log(
run_id=run_id,
level="info",
step_name="retry_failed_reset",
message=f"retryable failed queues reset: {retry_reset_count}",
realtor_id=realtor_id,
context={"retry_reset_count": retry_reset_count, "max_retry": max_retry}
)
if retry_session_hold:
hold_count = count_session_hold_queues(realtor_id=realtor_id)
print(f"[SESSION HOLD COUNT BEFORE RESET] {hold_count}")
add_pipeline_log(
run_id=run_id,
level="info",
step_name="session_hold_count",
message=f"SESSION_HOLD count before reset: {hold_count}",
realtor_id=realtor_id,
context={"hold_count": hold_count}
)
if hold_count > 0:
reset_count = reset_session_hold_queues(
realtor_id=realtor_id,
limit=max(int(limit), 20)
)
add_pipeline_log(
run_id=run_id,
level="info",
step_name="session_hold_reset",
message=f"SESSION_HOLD reset affected: {reset_count}",
realtor_id=realtor_id,
context={"reset_count": reset_count}
)
while processed < int(limit):
if only_night_window and not is_within_publish_window(window_start_hour, window_end_hour):
final_status = "stopped"
final_message = f"outside publish window during loop: {window_start_hour}:00~{window_end_hour}:00"
print("[PUBLISH WINDOW STOP]", final_message)
add_pipeline_log(
run_id=run_id,
level="info",
step_name="publish_window_stop",
message=final_message,
context={
"window_start_hour": window_start_hour,
"window_end_hour": window_end_hour,
"processed": processed,
}
)
break
if should_stop_by_time(until_time):
final_status = "stopped"
final_message = f"until_time reached: {until_time}"
print(f"[STOP] until_time reached: {until_time}")
add_pipeline_log(
run_id=run_id,
level="info",
step_name="time_window_stop",
message=final_message,
context={"until_time": until_time, "processed": processed}
)
break
if should_stop_by_hour(until_hour):
final_status = "stopped"
final_message = f"until_hour reached: {until_hour}"
print(f"[STOP] until_hour reached: {until_hour}")
add_pipeline_log(
run_id=run_id,
level="info",
step_name="time_window_stop",
message=final_message,
context={"until_hour": until_hour, "processed": processed}
)
break
queue = get_pending_queue(
realtor_id=realtor_id,
per_realtor_daily_limit=per_realtor_daily_limit,
skip_realtor_ids=attempted_realtor_ids,
)
if not queue:
final_message = "pending Queue 없음"
print("[INFO] pending Queue 없음")
add_pipeline_log(
run_id=run_id,
level="info",
step_name="no_pending_queue",
message=final_message,
realtor_id=realtor_id,
)
break
queue_id = queue.get("id")
queue_article_no = str(
queue.get("queue_article_no")
or queue.get("draft_article_no")
or queue.get("article_no")
or ""
)
queue_realtor_id = queue.get("realtor_id")
if queue_realtor_id:
attempted_realtor_ids.add(int(queue_realtor_id))
add_pipeline_log(
run_id=run_id,
level="info",
step_name="queue_start",
message=f"queue start id={queue_id}",
article_no=queue_article_no,
realtor_id=queue_realtor_id,
context={"queue_id": queue_id, "draft_id": queue.get("draft_id")}
)
publish_started_at = time.time()
ok = publish_one(queue)
publish_elapsed = max(0.0, time.time() - publish_started_at)
processed += 1
total_publish_seconds += publish_elapsed
avg_publish_seconds = total_publish_seconds / max(1, processed)
print(
"[PUBLISH ELAPSED]",
f"queue_id={queue_id}",
f"elapsed={publish_elapsed:.1f}s",
f"avg={avg_publish_seconds:.1f}s"
)
if ok:
success += 1
add_pipeline_log(
run_id=run_id,
level="info",
step_name="queue_success",
message=f"queue success id={queue_id}",
article_no=queue_article_no,
realtor_id=queue_realtor_id,
context={"queue_id": queue_id}
)
else:
failed += 1
add_pipeline_log(
run_id=run_id,
level="warning",
step_name="queue_failed_or_hold",
message=f"queue failed or hold id={queue_id}",
article_no=queue_article_no,
realtor_id=queue_realtor_id,
context={"queue_id": queue_id}
)
if processed >= int(limit):
break
next_sleep_seconds = 0
if smart_sleep:
next_sleep_seconds = calculate_smart_sleep_seconds(
processed=processed,
limit=limit,
realtor_id=realtor_id,
window_start_hour=window_start_hour,
window_end_hour=window_end_hour,
until_time=until_time,
avg_publish_seconds=(total_publish_seconds / max(1, processed)),
estimated_publish_seconds=estimated_publish_seconds,
min_sleep=min_sleep,
max_sleep=max_sleep,
sleep_jitter=sleep_jitter,
)
elif sleep_seconds and int(sleep_seconds) > 0:
next_sleep_seconds = int(sleep_seconds)
if next_sleep_seconds and int(next_sleep_seconds) > 0:
print(f"[SLEEP] {next_sleep_seconds}s before next queue")
time.sleep(int(next_sleep_seconds))
except Exception as e:
final_status = "failed"
final_message = str(e)[:2000]
failed += 1
traceback.print_exc()
add_pipeline_log(
run_id=run_id,
level="error",
step_name="worker_exception",
message=final_message,
realtor_id=realtor_id,
context={"error": str(e)}
)
finally:
if final_status == "success" and failed > 0:
final_status = "failed"
final_message = f"completed with failures: success={success}, failed={failed}"
remaining_summary = print_remaining_queue_summary(realtor_id=realtor_id)
finish_pipeline_run(
run_id=run_id,
status=final_status,
total=processed,
success=success,
failed=failed,
skipped=skipped,
message=final_message,
meta={
"limit": limit,
"realtor_id": realtor_id,
"processed": processed,
"success": success,
"failed": failed,
"skipped": skipped,
"until_hour": until_hour,
"until_time": until_time,
"only_night_window": only_night_window,
"window_start_hour": window_start_hour,
"window_end_hour": window_end_hour,
"max_retry": max_retry,
"auto_reset_retry_failed": auto_reset_retry_failed,
"smart_sleep": smart_sleep,
"min_sleep": min_sleep,
"max_sleep": max_sleep,
"sleep_jitter": sleep_jitter,
"estimated_publish_seconds": estimated_publish_seconds,
"per_realtor_daily_limit": per_realtor_daily_limit,
"attempted_realtor_ids": sorted(list(attempted_realtor_ids)),
"avg_publish_seconds": (total_publish_seconds / max(1, processed)) if processed else 0,
"remaining_publish_pending": remaining_summary.get("publish_pending"),
"remaining_private_pending": remaining_summary.get("private_pending"),
"worker_server": get_worker_server_name(),
}
)
print("=" * 80)
print("[PUBLISH WORKER DONE]")
print("[PROCESSED]", processed)
print("[SUCCESS]", success)
print("[FAILED]", failed)
print("=" * 80)
def acquire_lock(lock_path):
try:
fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
os.write(fd, str(os.getpid()).encode("utf-8"))
os.close(fd)
return True
except FileExistsError:
return False
def release_lock(lock_path):
try:
if os.path.exists(lock_path):
os.remove(lock_path)
except Exception:
pass
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--limit", type=int, default=1)
parser.add_argument("--realtor-id", type=int, default=None)
parser.add_argument("--sleep", type=int, default=300)
parser.add_argument("--until-hour", type=int, default=None)
parser.add_argument("--until-time", type=str, default=None, help="HH:MM 형식 종료 시각. 예: 07:30")
parser.add_argument("--stale-minutes", type=int, default=90)
parser.add_argument("--retry-session-hold", action="store_true", help="재연동 후 SESSION_HOLD 큐를 pending으로 복구")
parser.add_argument("--only-night-window", action="store_true", help="22:00~08:00 발행 윈도우 안에서만 실행")
parser.add_argument("--window-start-hour", type=int, default=22, help="발행 시작 시각. 기본 22")
parser.add_argument("--window-end-hour", type=int, default=8, help="발행 종료 시각. 기본 8")
parser.add_argument("--max-retry", type=int, default=2, help="자동 재시도 최대 횟수")
parser.add_argument("--no-auto-reset-retry-failed", action="store_true", help="재시도 시간이 지난 failed 큐 자동 pending 복구 끄기")
parser.add_argument("--smart-sleep", action="store_true", help="남은 시간/남은 건수/평균 발행시간 기준으로 발행 간격을 자동 랜덤 조절")
parser.add_argument("--min-sleep", type=int, default=20, help="smart-sleep 최소 대기 초. 기본 20")
parser.add_argument("--max-sleep", type=int, default=900, help="smart-sleep 최대 대기 초. 기본 900=15분")
parser.add_argument("--sleep-jitter", type=float, default=0.5, help="smart-sleep 랜덤 흔들림 비율. 기본 0.5=±50%%")
parser.add_argument("--estimated-publish-seconds", type=int, default=600, help="초기 평균 발행시간 추정값. 기본 600초=10분")
parser.add_argument("--per-realtor-daily-limit", type=int, default=1, help="중개사별 하루 실제 발행 제한. 기본 1건")
parser.add_argument("--lock-ttl-minutes", type=int, default=720, help="DB 작업락 유지 시간. 기본 720분")
parser.add_argument("--no-db-lock", action="store_true", help="DB 작업락 사용 안 함")
parser.add_argument("--no-lock", action="store_true")
args = parser.parse_args()
lock_path = os.path.join(STORAGE_DIR, "publish_worker.lock")
db_lock_name = "publish_worker"
db_lock_owner = None
if not args.no_lock:
if not acquire_lock(lock_path):
print("[LOCKED] publish_worker already running")
print("[LOCK FILE]", lock_path)
return
if not args.no_db_lock:
db_locked, db_lock_owner = acquire_db_lock(
db_lock_name,
ttl_minutes=args.lock_ttl_minutes,
)
if not db_locked:
print("[DB LOCKED] publish_worker already running or stale lock not expired")
if not args.no_lock:
release_lock(lock_path)
return
try:
run_publish_worker(
limit=args.limit,
realtor_id=args.realtor_id,
sleep_seconds=args.sleep,
until_hour=args.until_hour,
until_time=args.until_time,
stale_minutes=args.stale_minutes,
retry_session_hold=args.retry_session_hold,
only_night_window=args.only_night_window,
window_start_hour=args.window_start_hour,
window_end_hour=args.window_end_hour,
max_retry=args.max_retry,
auto_reset_retry_failed=not args.no_auto_reset_retry_failed,
smart_sleep=args.smart_sleep,
min_sleep=args.min_sleep,
max_sleep=args.max_sleep,
sleep_jitter=args.sleep_jitter,
estimated_publish_seconds=args.estimated_publish_seconds,
per_realtor_daily_limit=args.per_realtor_daily_limit,
)
finally:
if db_lock_owner:
release_db_lock(db_lock_name, db_lock_owner)
if not args.no_lock:
release_lock(lock_path)
if __name__ == "__main__":
main()