""" Playwright 기반 티스토리 자동 발행 모듈. 티스토리 Open API v1이 2024년 2월 완전 폐지됨에 따라 Playwright 브라우저 자동화 + storage_state 세션 관리 방식을 사용. 세션 워크플로우: 1. 최초 1회: login_interactive() → 수동 카카오 로그인 → storage_state 저장 2. 이후 실행: storage_state 복원 → 세션 유효성 확인 → 자동 발행 3. 세션 만료 시: login_interactive() 재실행 안내 """ from __future__ import annotations import logging import os from pathlib import Path from typing import Any from playwright.async_api import Browser, BrowserContext, Page, async_playwright logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 커스텀 예외 # --------------------------------------------------------------------------- class TistoryError(Exception): """티스토리 발행 모듈 기본 예외.""" class SessionExpiredError(TistoryError): """세션이 만료되었거나 유효하지 않을 때.""" class LoginRequiredError(TistoryError): """로그인이 필요한 상태일 때.""" class PublishError(TistoryError): """글 발행/저장에 실패했을 때.""" # --------------------------------------------------------------------------- # .env.keys 파서 (python-dotenv 미사용) # --------------------------------------------------------------------------- def _load_env_file(env_path: str) -> dict[str, str]: """ .env.keys 형식의 파일을 파싱하여 dict 반환. 지원 형식: export KEY=value KEY=value # 주석 """ result: dict[str, str] = {} path = Path(env_path) if not path.exists(): return result for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[len("export ") :] if "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") result[key] = value return result def _resolve_config(key: str, constructor_value: str | None, env_file_path: str = "") -> str | None: """환경변수 → .env.keys → 생성자 인자 순서로 설정값 해석.""" # 1. 프로세스 환경변수 우선 env_val = os.environ.get(key) if env_val: return env_val # 2. .env.keys 파일 if env_file_path: file_env = _load_env_file(env_file_path) if key in file_env: return file_env[key] # 3. 생성자 인자 폴백 return constructor_value # --------------------------------------------------------------------------- # TistoryPublisher # --------------------------------------------------------------------------- _DEFAULT_ENV_FILE = os.path.join(os.path.expanduser("~"), ".env.keys") _TISTORY_LOGIN_URL = "https://www.tistory.com/auth/login" _MANAGE_URL_PATTERN = "*/manage*" class TistoryPublisher: """Playwright 기반 티스토리 자동 발행 클래스.""" def __init__( self, blog_name: str | None = None, session_path: str | None = None, headless: bool = True, env_file: str = _DEFAULT_ENV_FILE, ) -> None: """ Args: blog_name: 티스토리 블로그 서브도메인 (예: "myblog" → myblog.tistory.com). 미입력 시 환경변수 TISTORY_BLOG_NAME 사용. session_path: storage_state JSON 파일 경로. 미입력 시 환경변수 TISTORY_SESSION_PATH 사용. headless: 브라우저 헤드리스 모드 여부. env_file: 환경 변수 파일 경로 (기본: ~/.env.keys). """ resolved_blog = _resolve_config("TISTORY_BLOG_NAME", blog_name, env_file) resolved_session = _resolve_config("TISTORY_SESSION_PATH", session_path, env_file) if not resolved_blog: raise ValueError("blog_name이 지정되지 않았습니다. 인자 또는 TISTORY_BLOG_NAME 환경변수를 설정하세요.") if not resolved_session: raise ValueError( "session_path가 지정되지 않았습니다. 인자 또는 TISTORY_SESSION_PATH 환경변수를 설정하세요." ) self.blog_name: str = resolved_blog self.session_path: str = resolved_session self.headless: bool = headless self._browser: Browser | None = None self._context: BrowserContext | None = None self._page: Page | None = None logger.debug("TistoryPublisher 초기화: blog=%s, session=%s", self.blog_name, self.session_path) # ------------------------------------------------------------------ # 내부 URL 헬퍼 # ------------------------------------------------------------------ @property def _base_url(self) -> str: return f"https://{self.blog_name}.tistory.com" @property def _manage_url(self) -> str: return f"{self._base_url}/manage" @property def _newpost_url(self) -> str: return f"{self._base_url}/manage/newpost/" def _edit_url(self, post_id: str | int) -> str: return f"{self._base_url}/manage/post/{post_id}" # ------------------------------------------------------------------ # 세션 관리 # ------------------------------------------------------------------ async def login_interactive(self) -> None: """ Headed 모드로 브라우저를 열고 사용자가 직접 로그인하도록 안내. 카카오 2FA를 포함한 수동 로그인이 완료되면 storage_state를 session_path에 저장한다. """ logger.info("대화형 로그인 시작 (headed 모드)") print(f"\n[TistoryPublisher] 브라우저가 열립니다.") print(f" → 카카오 계정으로 티스토리에 로그인하세요 (2FA 포함).") print(f" → 로그인 완료 후 관리 페이지로 이동되면 자동으로 세션이 저장됩니다.\n") async with async_playwright() as pw: browser = await pw.chromium.launch(headless=False) context = await browser.new_context() page = await context.new_page() await page.goto(_TISTORY_LOGIN_URL) # 관리 페이지 URL 패턴 감지: /manage 로 이동되면 로그인 완료로 판단 # 실제 셀렉터 확인 필요: 로그인 완료 후 리다이렉트 URL이 다를 수 있음 logger.info("로그인 완료 대기 중... 관리 페이지 URL 패턴을 감지합니다.") await page.wait_for_url( _MANAGE_URL_PATTERN, timeout=300_000, # 5분 대기 ) logger.info("로그인 감지됨. storage_state 저장 중: %s", self.session_path) Path(self.session_path).parent.mkdir(parents=True, exist_ok=True) await context.storage_state(path=self.session_path) print(f"[TistoryPublisher] 세션 저장 완료: {self.session_path}") await browser.close() async def _load_session(self) -> None: """session_path에서 storage_state를 로드하여 브라우저 컨텍스트를 초기화.""" if not Path(self.session_path).exists(): raise FileNotFoundError( f"세션 파일을 찾을 수 없습니다: {self.session_path}\n" "login_interactive()를 먼저 실행하여 세션을 생성하세요." ) logger.debug("세션 로드: %s", self.session_path) async def _ensure_browser(self, storage_state: str | None = None) -> tuple[Browser, BrowserContext, Page]: """ 브라우저 / 컨텍스트 / 페이지를 생성하거나 기존 것을 반환. Args: storage_state: storage_state JSON 파일 경로 (없으면 익명 세션). """ if self._browser is None: pw = await async_playwright().start() self._browser = await pw.chromium.launch(headless=self.headless) if self._context is None: kwargs: dict[str, Any] = {} if storage_state and Path(storage_state).exists(): kwargs["storage_state"] = storage_state self._context = await self._browser.new_context(**kwargs) if self._page is None: self._page = await self._context.new_page() return self._browser, self._context, self._page async def _check_session_valid(self) -> bool: """ 저장된 세션으로 관리 페이지에 접근하여 유효성 검증. Returns: True: 세션 유효, False: 세션 만료 또는 로그인 페이지로 리다이렉트. """ await self._load_session() try: _, _, page = await self._ensure_browser(storage_state=self.session_path) await page.goto(self._manage_url, wait_until="domcontentloaded", timeout=30_000) current_url = page.url if "auth/login" in current_url or "login" in current_url: logger.warning("세션 만료: 로그인 페이지로 리다이렉트됨 (%s)", current_url) return False logger.debug("세션 유효: %s", current_url) return True except Exception as exc: logger.warning("세션 유효성 확인 중 오류: %s", exc) return False # ------------------------------------------------------------------ # 글 발행 # ------------------------------------------------------------------ async def publish_post( self, title: str, content: str, category_id: str = "0", tags: list[str] | None = None, visibility: str = "private", ) -> str: """ 글쓰기 페이지에서 새 글을 작성하고 발행한다. Args: title: 글 제목. content: HTML 본문. category_id: 카테고리 ID (기본 "0" = 분류 없음). tags: 태그 목록. visibility: "private" (비공개) 또는 "public" (공개). PoC에서는 private만 사용. Returns: 발행된 글 URL. Raises: SessionExpiredError: 세션 만료 시. PublishError: 발행 실패 시. """ if tags is None: tags = [] logger.info("글 발행 시작: title=%r, visibility=%s", title, visibility) valid = await self._check_session_valid() if not valid: raise SessionExpiredError("세션이 만료되었습니다. login_interactive()를 실행하여 재로그인하세요.") _, _, page = await self._ensure_browser(storage_state=self.session_path) try: # 글쓰기 페이지 이동 await page.goto(self._newpost_url, wait_until="networkidle", timeout=30_000) logger.debug("글쓰기 페이지 로드 완료: %s", self._newpost_url) # --- 제목 입력 --- # 실제 셀렉터 확인 필요: 티스토리 글쓰기 제목 입력 필드 title_selector = "#post-title-inp" await page.wait_for_selector(title_selector, timeout=15_000) await page.fill(title_selector, title) logger.debug("제목 입력 완료") # --- HTML 모드 전환 --- # 실제 셀렉터 확인 필요: 에디터 모드 전환 버튼 (기본은 WYSIWYG) # 일반적으로 "HTML" 버튼 또는 탭이 존재 html_mode_btn_selector = "button.btn-html, .editor-mode-btn[data-mode='html'], button:has-text('HTML')" try: await page.click(html_mode_btn_selector, timeout=5_000) logger.debug("HTML 모드 전환 완료") except Exception: # 이미 HTML 모드이거나 버튼이 다른 위치에 있을 수 있음 # 실제 셀렉터 확인 필요 logger.warning("HTML 모드 전환 버튼을 찾지 못했습니다. 셀렉터를 확인하세요.") # --- HTML 본문 입력 --- # 실제 셀렉터 확인 필요: HTML 에디터 textarea 또는 contenteditable div content_selector = "textarea#content, .CodeMirror textarea, .html-editor textarea" try: await page.wait_for_selector(content_selector, timeout=10_000) await page.fill(content_selector, content) logger.debug("HTML 본문 입력 완료 (textarea)") except Exception: # CodeMirror 등 커스텀 에디터일 경우 evaluate로 직접 주입 # 실제 셀렉터 확인 필요 logger.warning("본문 textarea를 찾지 못했습니다. JavaScript 주입 시도") await page.evaluate( """(content) => { const cm = document.querySelector('.CodeMirror'); if (cm && cm.CodeMirror) { cm.CodeMirror.setValue(content); } }""", content, ) # --- 카테고리 설정 --- if category_id != "0": # 실제 셀렉터 확인 필요: 카테고리 선택 드롭다운 category_selector = "select#category, select[name='category']" try: await page.select_option(category_selector, value=category_id) logger.debug("카테고리 설정: %s", category_id) except Exception: logger.warning("카테고리 선택 실패: category_id=%s. 셀렉터를 확인하세요.", category_id) # --- 태그 입력 --- if tags: # 실제 셀렉터 확인 필요: 태그 입력 필드 tag_selector = "input#tag, input[name='tag'], .tag-input input" try: await page.fill(tag_selector, ", ".join(tags)) logger.debug("태그 입력 완료: %s", tags) except Exception: logger.warning("태그 입력 실패. 셀렉터를 확인하세요.") # --- 공개 설정 (비공개) --- if visibility == "private": # 실제 셀렉터 확인 필요: 비공개 라디오버튼 또는 선택 옵션 private_selector = ( "input[value='3'][name='visibility'], " "label:has-text('비공개'), " ".visibility-private" ) try: await page.click(private_selector, timeout=5_000) logger.debug("비공개 설정 완료") except Exception: logger.warning("비공개 설정 버튼을 찾지 못했습니다. 셀렉터를 확인하세요.") # --- 발행 버튼 클릭 --- # 실제 셀렉터 확인 필요: 발행/완료 버튼 publish_selector = "button#publish-layer-btn, button.btn-publish, button:has-text('발행')" await page.click(publish_selector, timeout=10_000) logger.debug("발행 버튼 클릭") # 발행 완료 후 URL 변경 대기 # 실제 셀렉터 확인 필요: 발행 완료 후 이동되는 URL 패턴 await page.wait_for_url(f"*/{self.blog_name}.tistory.com/*", timeout=15_000) published_url = page.url logger.info("글 발행 완료: %s", published_url) return published_url except SessionExpiredError: raise except Exception as exc: logger.exception("글 발행 중 오류 발생") raise PublishError(f"글 발행에 실패했습니다: {exc}") from exc # ------------------------------------------------------------------ # 글 수정 # ------------------------------------------------------------------ async def edit_post( self, post_id: str | int, title: str | None = None, content: str | None = None, ) -> None: """ 기존 글을 수정한다. Args: post_id: 수정할 글의 ID. title: 새 제목 (None이면 변경 안 함). content: 새 HTML 본문 (None이면 변경 안 함). Raises: SessionExpiredError: 세션 만료 시. PublishError: 저장 실패 시. """ logger.info("글 수정 시작: post_id=%s", post_id) valid = await self._check_session_valid() if not valid: raise SessionExpiredError("세션이 만료되었습니다. login_interactive()를 실행하여 재로그인하세요.") _, _, page = await self._ensure_browser(storage_state=self.session_path) try: await page.goto(self._edit_url(post_id), wait_until="networkidle", timeout=30_000) if title is not None: # 실제 셀렉터 확인 필요 title_selector = "#post-title-inp" await page.fill(title_selector, title) logger.debug("제목 수정 완료") if content is not None: # 실제 셀렉터 확인 필요: HTML 에디터 content_selector = "textarea#content, .CodeMirror textarea" try: await page.fill(content_selector, content) except Exception: await page.evaluate( """(content) => { const cm = document.querySelector('.CodeMirror'); if (cm && cm.CodeMirror) { cm.CodeMirror.setValue(content); } }""", content, ) logger.debug("본문 수정 완료") # 저장 버튼 클릭 # 실제 셀렉터 확인 필요: 저장/수정 버튼 save_selector = ( "button#publish-layer-btn, button.btn-save, button:has-text('저장'), button:has-text('수정')" ) await page.click(save_selector, timeout=10_000) logger.info("글 수정 저장 완료: post_id=%s", post_id) except SessionExpiredError: raise except Exception as exc: logger.exception("글 수정 중 오류 발생") raise PublishError(f"글 수정에 실패했습니다: {exc}") from exc # ------------------------------------------------------------------ # 글 상태 조회 # ------------------------------------------------------------------ async def get_post_status(self, post_id: str | int) -> dict[str, str]: """ 관리 페이지에서 특정 글의 상태를 확인한다. Args: post_id: 조회할 글의 ID. Returns: dict: {"post_id": str, "title": str, "status": "private"|"public", "url": str} Raises: SessionExpiredError: 세션 만료 시. TistoryError: 글 정보를 파싱할 수 없을 때. """ logger.info("글 상태 조회: post_id=%s", post_id) valid = await self._check_session_valid() if not valid: raise SessionExpiredError("세션이 만료되었습니다. login_interactive()를 실행하여 재로그인하세요.") _, _, page = await self._ensure_browser(storage_state=self.session_path) try: # 글 관리 목록 페이지 # 실제 셀렉터 확인 필요: 관리 페이지 글 목록에서 해당 post_id 행 확인 manage_post_url = f"{self._manage_url}/posts" await page.goto(manage_post_url, wait_until="networkidle", timeout=30_000) # post_id에 해당하는 행에서 제목·상태·URL 추출 # 실제 셀렉터 확인 필요 row_selector = f"tr[data-post-id='{post_id}'], li[data-post-id='{post_id}']" row = page.locator(row_selector) title = await row.locator(".post-title, .title").inner_text(timeout=5_000) status_text = await row.locator(".post-status, .status").inner_text(timeout=5_000) status = "private" if "비공개" in status_text else "public" post_url = f"{self._base_url}/{post_id}" result: dict[str, str] = { "post_id": str(post_id), "title": title.strip(), "status": status, "url": post_url, } logger.debug("글 상태 조회 결과: %s", result) return result except SessionExpiredError: raise except Exception as exc: logger.exception("글 상태 조회 중 오류 발생") raise TistoryError(f"글 상태를 확인할 수 없습니다: {exc}") from exc # ------------------------------------------------------------------ # 이미지 업로드 # ------------------------------------------------------------------ async def upload_image(self, image_path: str) -> str: """ 글쓰기 페이지의 이미지 업로드 기능을 사용하여 이미지를 업로드한다. Args: image_path: 업로드할 이미지 파일의 로컬 경로. Returns: 업로드된 이미지의 URL. Raises: FileNotFoundError: 이미지 파일이 없을 때. SessionExpiredError: 세션 만료 시. PublishError: 업로드 실패 시. """ if not Path(image_path).exists(): raise FileNotFoundError(f"이미지 파일을 찾을 수 없습니다: {image_path}") logger.info("이미지 업로드 시작: %s", image_path) valid = await self._check_session_valid() if not valid: raise SessionExpiredError("세션이 만료되었습니다. login_interactive()를 실행하여 재로그인하세요.") _, _, page = await self._ensure_browser(storage_state=self.session_path) try: # 글쓰기 페이지로 이동 (이미지 업로드 UI가 있는 페이지) await page.goto(self._newpost_url, wait_until="networkidle", timeout=30_000) # 이미지 업로드 input 트리거 # 실제 셀렉터 확인 필요: 파일 업로드 input (hidden일 수 있음) file_input_selector = "input[type='file'][accept*='image'], #imageUploadInput" file_input = page.locator(file_input_selector) async with page.expect_response( lambda r: "upload" in r.url or "image" in r.url, timeout=30_000, ) as response_info: await file_input.set_input_files(image_path) response = await response_info.value response_json: dict[str, Any] = await response.json() # 실제 응답 구조 확인 필요: 업로드 결과에서 URL 추출 image_url: str = ( response_json.get("url") or response_json.get("imageUrl") or response_json.get("data", {}).get("url", "") ) if not image_url: raise PublishError("이미지 업로드 응답에서 URL을 파싱할 수 없습니다.") logger.info("이미지 업로드 완료: %s", image_url) return image_url except (SessionExpiredError, FileNotFoundError, PublishError): raise except Exception as exc: logger.exception("이미지 업로드 중 오류 발생") raise PublishError(f"이미지 업로드에 실패했습니다: {exc}") from exc # ------------------------------------------------------------------ # 리소스 정리 # ------------------------------------------------------------------ async def close(self) -> None: """브라우저 리소스를 정리한다.""" if self._page is not None: await self._page.close() self._page = None if self._context is not None: await self._context.close() self._context = None if self._browser is not None: await self._browser.close() self._browser = None logger.debug("브라우저 리소스 정리 완료") # ------------------------------------------------------------------ # 컨텍스트 매니저 지원 # ------------------------------------------------------------------ async def __aenter__(self) -> "TistoryPublisher": return self async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: await self.close()