""" task_718_1_text_post.py ----------------------- ThreadAuto 5단계 파이프라인으로 텍스트 전용 포스트 1건을 생성하고 Threads에 업로드하는 스크립트. 담당: 토르 (개발2팀 백엔드) 태스크: task-718.1 """ from __future__ import annotations import asyncio import json import logging import re import sys import traceback # ThreadAuto 프로젝트 경로 주입 sys.path.insert(0, "/home/jay/projects/ThreadAuto") # --------------------------------------------------------------------------- # 로깅 설정 # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s — %(message)s", datefmt="%Y-%m-%d %H:%M:%S", stream=sys.stderr, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 검증 헬퍼 # --------------------------------------------------------------------------- # text_insight 자수 제한 (줄바꿈 제외) CHAR_MIN = 100 CHAR_MAX = 300 # 한국어 종결어미 패턴 (문장 끝에 있어야 함) ENDING_PATTERN = re.compile(r"[다요죠음임][\.\!\?]?\s*$", re.MULTILINE) def validate_text(text: str) -> dict: """텍스트 검증 결과를 반환한다. Returns: { "pure_char_count": int, 줄바꿈 제외 순수 글자수 "char_range_ok": bool, 100~300자 범위 여부 "ending_ok": bool, 한국어 종결어미로 끝나는지 여부 "last_char": str, 마지막 비공백 문자 "issues": [str], 문제 목록 } """ issues: list[str] = [] pure_text = text.replace("\n", "") pure_len = len(pure_text) char_range_ok = CHAR_MIN <= pure_len <= CHAR_MAX if not char_range_ok: issues.append(f"글자수 범위 위반: {pure_len}자 (허용: {CHAR_MIN}~{CHAR_MAX}자)") # 마지막 비공백 문자 추출 stripped = text.rstrip() last_char = stripped[-1] if stripped else "" # 한국어 종결어미로 끝나는지 확인 ending_ok = bool(ENDING_PATTERN.search(stripped)) if not ending_ok: issues.append(f"종결어미 미준수: 마지막 문자='{last_char}' " "(다/요/죠/음/임 등으로 끝나야 함)") return { "pure_char_count": pure_len, "char_range_ok": char_range_ok, "ending_ok": ending_ok, "last_char": last_char, "issues": issues, } # --------------------------------------------------------------------------- # 메인 비동기 실행 함수 # --------------------------------------------------------------------------- async def main() -> None: # ------------------------------------------------------------------ # Step 1: 토픽 선택 (업계동향 카테고리 — text_insight에 적합) # ------------------------------------------------------------------ logger.info("=== Step 1: 토픽 선택 (카테고리: 업계동향) ===") from content.topic_selector import select_single_topic topic = select_single_topic(category="업계동향") logger.info("선택된 토픽:\n%s", json.dumps(topic, ensure_ascii=False, indent=2)) # ------------------------------------------------------------------ # Step 2: 5단계 파이프라인으로 텍스트 생성 # ------------------------------------------------------------------ logger.info("=== Step 2: FiveStagePipeline 콘텐츠 생성 (content_type=text_insight) ===") from content.five_stage_pipeline import FiveStagePipeline pipeline = FiveStagePipeline() result = pipeline.generate(topic=topic, content_type="text_insight") text: str = result.get("text", "") hashtags: list[str] = result.get("hashtags", []) review_score = result.get("pipeline_metadata", {}).get("review_score", "N/A") logger.info("--- 생성된 텍스트 ---\n%s", text) logger.info("--- 해시태그 ---\n%s", hashtags) logger.info("--- 검수 점수 ---\n%s", review_score) # ------------------------------------------------------------------ # Step 3: 텍스트 검증 # ------------------------------------------------------------------ logger.info("=== Step 3: 텍스트 검증 ===") validation = validate_text(text) logger.info("순수 글자수 (줄바꿈 제외): %d자", validation["pure_char_count"]) logger.info("글자수 범위 OK: %s", validation["char_range_ok"]) logger.info("종결어미 OK: %s (마지막 문자: '%s')", validation["ending_ok"], validation["last_char"]) if validation["issues"]: logger.warning("검증 이슈 발견:") for issue in validation["issues"]: logger.warning(" - %s", issue) else: logger.info("검증 통과: 모든 조건 충족") # ------------------------------------------------------------------ # Step 4: 유효한 토큰 취득 # ------------------------------------------------------------------ logger.info("=== Step 4: 액세스 토큰 취득 ===") from auth.token_store import get_valid_token token = get_valid_token() if not token: raise RuntimeError("유효한 액세스 토큰이 없습니다. 먼저 인증을 완료하세요.") logger.info("액세스 토큰 획득 완료 (길이: %d자)", len(token)) # ------------------------------------------------------------------ # Step 5: Threads 텍스트 포스트 업로드 # ------------------------------------------------------------------ logger.info("=== Step 5: Threads 텍스트 포스트 업로드 ===") from api.client import ThreadsClient client = ThreadsClient(token) post_id = await client.post_text(text) logger.info("=" * 60) logger.info("텍스트 포스트 업로드 성공!") logger.info("Threads Post ID: %s", post_id) logger.info("=" * 60) # ------------------------------------------------------------------ # 최종 결과 JSON 출력 (stdout) # ------------------------------------------------------------------ output = { "post_id": post_id, "text": text, "char_count": validation["pure_char_count"], "topic_title": topic.get("title", ""), "topic_category": topic.get("category", ""), "hashtags": hashtags, "review_score": review_score, "validation": { "char_range_ok": validation["char_range_ok"], "ending_ok": validation["ending_ok"], "issues": validation["issues"], }, } print(json.dumps(output, ensure_ascii=False, indent=2)) # --------------------------------------------------------------------------- # 진입점 # --------------------------------------------------------------------------- if __name__ == "__main__": try: asyncio.run(main()) except Exception: logger.error("실행 중 예외 발생 — 전체 traceback:") traceback.print_exc(file=sys.stderr) sys.exit(1)