""" blog-writer 출력물 → TistoryPublisher 입력 연결 파이프라인. CLI: python3 scripts/blog/publish_pipeline.py --file --private python3 scripts/blog/publish_pipeline.py --file post.html --blog myblog --category 123 --tags AI,Python HTML 파일 구조 (자동 파싱 지원): 글 제목 또는

글 제목

... HTML 본문 ... """ from __future__ import annotations import argparse import asyncio import logging import sys from html.parser import HTMLParser from pathlib import Path from scripts.blog.tistory_publisher import ( LoginRequiredError, PublishError, SessionExpiredError, TistoryPublisher, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # HTML 파서 # --------------------------------------------------------------------------- class _HtmlMetaParser(HTMLParser): """HTML 파일에서 제목·태그·본문을 추출하는 경량 파서.""" def __init__(self) -> None: super().__init__() self.title: str = "" self.keywords: str = "" self._in_title: bool = False self._in_h1: bool = False self._h1_captured: bool = False # 첫 번째 h1만 사용 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag == "title": self._in_title = True elif tag == "h1" and not self._h1_captured: self._in_h1 = True elif tag == "meta": attr_dict: dict[str, str | None] = {k: v for k, v in attrs} name_val = attr_dict.get("name") or "" if name_val.lower() == "keywords": content_val = attr_dict.get("content") or "" self.keywords = content_val def handle_endtag(self, tag: str) -> None: if tag == "title": self._in_title = False elif tag == "h1": if self._in_h1: self._h1_captured = True self._in_h1 = False def handle_data(self, data: str) -> None: if self._in_title and not self.title: self.title = data.strip() elif self._in_h1 and not self._h1_captured: self.title = self.title or data.strip() def parse_html_file(html_path: Path) -> dict[str, str]: """ HTML 파일을 읽어 제목·태그·본문을 추출한다. Args: html_path: HTML 파일 경로. Returns: dict: {"title": str, "keywords": str, "body": str (전체 HTML)} Raises: FileNotFoundError: 파일이 없을 때. ValueError: HTML 파싱 실패 시. """ if not html_path.exists(): raise FileNotFoundError(f"HTML 파일을 찾을 수 없습니다: {html_path}") try: raw_html = html_path.read_text(encoding="utf-8") except UnicodeDecodeError: # UTF-8 실패 시 EUC-KR 시도 try: raw_html = html_path.read_text(encoding="euc-kr") except Exception as exc: raise ValueError(f"HTML 파일 인코딩을 인식할 수 없습니다: {exc}") from exc parser = _HtmlMetaParser() try: parser.feed(raw_html) except Exception as exc: raise ValueError(f"HTML 파싱 실패: {exc}") from exc return { "title": parser.title, "keywords": parser.keywords, "body": raw_html, } # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="publish_pipeline", description="HTML 파일을 티스토리에 발행하는 파이프라인", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument( "--file", required=True, metavar="HTML_PATH", help="발행할 HTML 파일 경로 (필수)", ) parser.add_argument( "--private", action="store_true", default=True, help="비공개 발행 (기본값, PoC에서는 항상 비공개)", ) parser.add_argument( "--blog", metavar="BLOG_NAME", default=None, help="티스토리 블로그 서브도메인 (기본: TISTORY_BLOG_NAME 환경변수)", ) parser.add_argument( "--category", metavar="CATEGORY_ID", default="0", help="카테고리 ID (기본: 0 = 분류 없음)", ) parser.add_argument( "--tags", metavar="TAG1,TAG2", default=None, help="쉼표 구분 태그 목록 (미지정 시 HTML 에서 추출)", ) parser.add_argument( "--title", metavar="TITLE", default=None, help="글 제목 (미지정 시 HTML 또는 <h1>에서 추출)", ) parser.add_argument( "--login", action="store_true", default=False, help="강제 대화형 로그인 실행 (세션 갱신 시 사용)", ) parser.add_argument( "--verbose", "-v", action="store_true", default=False, help="상세 로그 출력", ) return parser # --------------------------------------------------------------------------- # 파이프라인 메인 로직 # --------------------------------------------------------------------------- async def run_pipeline(args: argparse.Namespace) -> int: """ 파이프라인 실행 함수. Args: args: argparse.Namespace (CLI 인자). Returns: 종료 코드 (0: 성공, 1: 실패). """ html_path = Path(args.file) # 1. HTML 파싱 try: parsed = parse_html_file(html_path) except FileNotFoundError as exc: print(f"[ERROR] {exc}", file=sys.stderr) return 1 except ValueError as exc: print(f"[ERROR] HTML 파싱 실패: {exc}", file=sys.stderr) return 1 # 2. 제목 결정 (CLI 인자 우선 → HTML 추출) title: str = args.title or parsed["title"] if not title: print( "[ERROR] 제목을 결정할 수 없습니다. --title 옵션을 사용하거나 " "HTML 파일에 <title> 또는 <h1> 태그를 추가하세요.", file=sys.stderr, ) return 1 # 3. 태그 결정 (CLI 인자 우선 → HTML meta keywords 추출) tags: list[str] if args.tags: tags = [t.strip() for t in args.tags.split(",") if t.strip()] elif parsed["keywords"]: tags = [t.strip() for t in parsed["keywords"].split(",") if t.strip()] else: tags = [] visibility = "private" # PoC에서는 항상 비공개 print(f"[INFO] 발행 준비") print(f" 파일: {html_path}") print(f" 제목: {title}") print(f" 카테고리: {args.category}") print(f" 태그: {tags}") print(f" 공개여부: {visibility}") # 4. TistoryPublisher 인스턴스 생성 try: publisher = TistoryPublisher(blog_name=args.blog) except ValueError as exc: print(f"[ERROR] 설정 오류: {exc}", file=sys.stderr) return 1 async with publisher: # 5. 강제 로그인 또는 세션 유효성 확인 if args.login: print("[INFO] 대화형 로그인을 시작합니다...") await publisher.login_interactive() print("[INFO] 로그인 완료. 세션이 저장되었습니다.") try: session_valid = await publisher._check_session_valid() except FileNotFoundError as exc: print(f"[ERROR] 세션 파일 없음: {exc}", file=sys.stderr) print("[HINT] --login 옵션으로 먼저 로그인 세션을 생성하세요:", file=sys.stderr) print(f" python3 scripts/blog/publish_pipeline.py --file {args.file} --login", file=sys.stderr) return 1 if not session_valid: print("[ERROR] 세션이 만료되었습니다.", file=sys.stderr) print("[HINT] --login 옵션으로 재로그인하세요:", file=sys.stderr) print(f" python3 scripts/blog/publish_pipeline.py --file {args.file} --login", file=sys.stderr) return 1 # 6. 발행 try: published_url = await publisher.publish_post( title=title, content=parsed["body"], category_id=args.category, tags=tags, visibility=visibility, ) except SessionExpiredError as exc: print(f"[ERROR] 세션 만료: {exc}", file=sys.stderr) return 1 except LoginRequiredError as exc: print(f"[ERROR] 로그인 필요: {exc}", file=sys.stderr) return 1 except PublishError as exc: print(f"[ERROR] 발행 실패: {exc}", file=sys.stderr) return 1 # 7. 결과 출력 print(f"\n[SUCCESS] 글 발행 완료!") print(f" URL: {published_url}") print(f" 상태: {visibility}") return 0 def main() -> None: """CLI 진입점.""" arg_parser = build_arg_parser() args = arg_parser.parse_args() log_level = logging.DEBUG if args.verbose else logging.INFO logging.basicConfig( level=log_level, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) exit_code = asyncio.run(run_pipeline(args)) sys.exit(exit_code) if __name__ == "__main__": main()