# -*- coding: utf-8 -*- """ STEP105-13 V2 Final Publish Action 역할: - 발행 설정창의 최종 '발행' 버튼을 처리한다. - execute=False이면 버튼 확인만 하고 클릭하지 않는다. - execute=True이면 실제 최종 발행 버튼을 클릭한다. - 클릭 후 URL 변화/PostView/게시글 URL 후보를 수집한다. 중요: - 이 모듈은 V2 전용이다. - V1 workers/publish_worker.py는 수정하지 않는다. """ class V2FinalPublishAction: def __init__(self, verbose=True): self.verbose = verbose def run(self, page, write_frame, dialog_result=None, execute=False, logger=None, session=None): def _run(): return self._run_impl( page=page, write_frame=write_frame, dialog_result=dialog_result, execute=execute, ) if session and logger: with session.step( "publish.final", logger=logger, meta={ "execute": bool(execute), }, ): return _run() return _run() def _run_impl(self, page, write_frame, dialog_result=None, execute=False): button_info = self._resolve_final_button(write_frame, dialog_result=dialog_result) if not button_info.get("found"): raise RuntimeError("최종 발행 버튼을 찾지 못했습니다.") print("[V2 FINAL PUBLISH BUTTON]", button_info) if not execute: return { "status": "ready", "execute": False, "message": "최종 발행 버튼 확인 완료. execute=False라 클릭하지 않음.", "button": button_info, "url_before": page.url, } selector = button_info["selector"] index = int(button_info.get("index") or 0) button = write_frame.locator(selector).nth(index) if not button.is_visible(timeout=5000): raise RuntimeError("최종 발행 버튼이 visible 상태가 아닙니다.") if not button.is_enabled(timeout=5000): raise RuntimeError("최종 발행 버튼이 enabled 상태가 아닙니다.") url_before = page.url print("[V2 FINAL PUBLISH CLICK]", { "selector": selector, "index": index, "url_before": url_before, }) button.click(force=True, timeout=5000) result = self._wait_after_publish(page=page, write_frame=write_frame, url_before=url_before) result["status"] = "clicked" result["execute"] = True result["button"] = button_info result["url_before"] = url_before result["url_after"] = page.url print("[V2 FINAL PUBLISH RESULT]", result) return result def _resolve_final_button(self, write_frame, dialog_result=None): """ dialog_result에 저장된 best를 우선 사용하되, 실제 DOM에서 다시 검증한다. """ best = ((dialog_result or {}).get("final_button") or {}).get("best") or {} if best: selector = best.get("selector") index = int(best.get("index") or 0) if selector: try: locator = write_frame.locator(selector) count = locator.count() if count > index: item = locator.nth(index) visible = item.is_visible(timeout=1000) enabled = item.is_enabled(timeout=1000) text = str(item.inner_text(timeout=1000) or "").strip() cls = str(item.get_attribute("class") or "").strip() if visible and enabled and text == "발행": return { "found": True, "selector": selector, "index": index, "count": count, "visible": visible, "enabled": enabled, "text": text, "class": cls, "source": "dialog_result", } except Exception as e: if self.verbose: print("[V2 FINAL PUBLISH BEST VERIFY FAIL]", str(e)) selectors = [ "[class*='layer_btn_area'] button:has-text('발행')", "button[class*='confirm_btn']", "[class*='layer_publish'] button[class*='confirm_btn']", "[class*='layer_publish'] button:has-text('발행')", ] candidates = [] for selector in selectors: try: locator = write_frame.locator(selector) count = locator.count() if self.verbose: print("[V2 FINAL PUBLISH CHECK]", selector, count) if count <= 0: continue for idx in range(min(count, 10)): item = locator.nth(idx) info = { "found": True, "selector": selector, "index": idx, "count": count, "visible": False, "enabled": False, "text": "", "class": "", "source": "fallback_scan", } try: info["visible"] = item.is_visible(timeout=1000) except Exception: pass try: info["enabled"] = item.is_enabled(timeout=1000) except Exception: pass try: info["text"] = str(item.inner_text(timeout=1000) or "").strip() except Exception: pass try: info["class"] = str(item.get_attribute("class") or "").strip() except Exception: pass if info["visible"] and info["enabled"] and info["text"] == "발행": candidates.append(info) except Exception as e: if self.verbose: print("[V2 FINAL PUBLISH CHECK ERROR]", selector, str(e)) if not candidates: return { "found": False, } candidates = sorted( candidates, key=lambda x: ( 1 if "confirm_btn" in str(x.get("class") or "") else 0, 1 if "layer_btn_area" in str(x.get("selector") or "") else 0, ), reverse=True, ) return candidates[0] def _wait_after_publish(self, page, write_frame, url_before): """ 실제 발행 클릭 후 결과를 기다린다. 네이버는 계정/상태에 따라 같은 탭 URL 이동, 새 탭, 레이어, PostView 이동이 다를 수 있으므로 1차로 현재 page URL과 텍스트/링크 후보를 수집한다. """ result = { "waited_ms": 0, "url_changed": False, "post_url": "", "post_url_candidates": [], "success_like": False, "signals": {}, } # 최대 25초 대기 for i in range(25): page.wait_for_timeout(1000) result["waited_ms"] = (i + 1) * 1000 current_url = page.url if current_url != url_before: result["url_changed"] = True if self._is_post_url(current_url): result["post_url"] = current_url result["success_like"] = True return result # frame/page 내에서 post url 후보 수집 candidates = self._collect_post_url_candidates(page=page, write_frame=write_frame) if candidates: result["post_url_candidates"] = candidates # 가장 그럴듯한 후보 1개 for url in candidates: if self._is_post_url(url): result["post_url"] = url result["success_like"] = True return result # 성공 신호 텍스트 signals = self._collect_success_signals(page=page, write_frame=write_frame) result["signals"] = signals if signals.get("success_text_found"): result["success_like"] = True return result return result def _is_post_url(self, url): url = str(url or "") return ( "blog.naver.com" in url and ( "PostView" in url or "/PostView.naver" in url or url.rstrip("/").split("/")[-1].isdigit() ) ) def _collect_post_url_candidates(self, page, write_frame): urls = [] targets = [page, write_frame] for target in targets: try: hrefs = target.locator("a[href*='blog.naver.com']").evaluate_all( """els => els.map(a => a.href).filter(Boolean).slice(0, 20)""" ) for href in hrefs: if href and href not in urls: urls.append(href) except Exception: pass return urls[:20] def _collect_success_signals(self, page, write_frame): texts = [ "발행되었습니다", "게시되었습니다", "등록되었습니다", "내 블로그", "게시글", ] result = { "success_text_found": False, "matches": [], } for target_name, target in [("page", page), ("write_frame", write_frame)]: for text in texts: try: count = target.locator(f"text={text}").count() if count > 0: result["success_text_found"] = True result["matches"].append({ "target": target_name, "text": text, "count": count, }) except Exception: pass return result def final_publish(page, write_frame, dialog_result=None, execute=False, logger=None, session=None, verbose=True): worker = V2FinalPublishAction(verbose=verbose) return worker.run( page=page, write_frame=write_frame, dialog_result=dialog_result, execute=execute, logger=logger, session=session, )