refactor: 블로그 포스팅 기능 리펙토링

apps/pre-processing-service/app/model/schemas.py

@@ -193,6 +193,7 @@ class RequestBlogPublish(RequestBase):
     tag: str = Field(..., title="블로그 태그", description="블로그 플랫폼 종류")
     blog_id: str = Field(..., description="블로그 아이디")
     blog_pw: str = Field(..., description="블로그 비밀번호")
+    blog_name: Optional[str] = Field(None, description="블로그 이름")
     post_title: str = Field(..., description="포스팅 제목")
     post_content: str = Field(..., description="포스팅 내용")
     post_tags: List[str] = Field(default_factory=list, description="포스팅 태그 목록")

apps/pre-processing-service/app/service/blog/base_blog_post_service.py

@@ -47,12 +47,13 @@ def _login(self) -> None:
         pass
 
     @abstractmethod
-    def _write_content(self, title: str, content: str, tags: List[str] = None) -> None:
+    def _write_content(self, title: str, content: str, tags: List[str] = None) -> str:
         """
         플랫폼별 포스팅 작성 구현
         :param title: 포스트 제목
         :param content: 포스트 내용
         :param tags: 포스트 태그 리스트
+        :return: 발행된 블로그 포스트 URL
         """
         pass
 
@@ -96,14 +97,15 @@ def post_content(self, title: str, content: str, tags: List[str] = None) -> Dict
         self._login()
 
         # 3. 포스트 작성 및 발행
-        self._write_content(title, content, tags)
+        post_url = self._write_content(title, content, tags)
 
         # 4. 결과 반환
         return {
-            "platform": self._get_platform_name(),
-            "title": title,
-            "content_length": len(content),
+            "tag": self._get_platform_name(),
+            "post_title": title,
             "tags": tags or [],
+            "publish_success": True,
+            "post_url": post_url,
         }
 
     def __del__(self):

apps/pre-processing-service/app/service/blog/blog_publish_service.py

@@ -1,4 +1,4 @@
-from typing import Dict
+from typing import Dict, Optional
 from app.errors.CustomException import CustomException
 from app.model.schemas import RequestBlogPublish
 from app.service.blog.blog_service_factory import BlogServiceFactory
@@ -10,31 +10,37 @@ class BlogPublishService:
     def __init__(self):
         self.factory = BlogServiceFactory()
 
-    def publish_content(self, request: RequestBlogPublish) -> Dict:
+    def publish_content(
+        self,
+        request: RequestBlogPublish,
+    ) -> Dict:
         """
         생성된 블로그 콘텐츠를 배포합니다.
+
+        Args:
+            request: 블로그 발행 요청 데이터
+            blog_id: 블로그 아이디
+            blog_password: 블로그 비밀번호
         """
         try:
             # 팩토리를 통해 적절한 서비스 생성
-            blog_service = self.factory.create_service(request.tag)
+            blog_service = self.factory.create_service(
+                request.tag,
+                blog_id=request.blog_id,
+                blog_password=request.blog_pw,
+                blog_name=request.blog_name,
+            )
 
             # 공통 인터페이스로 포스팅 실행
-            blog_service.post_content(
+            response_data = blog_service.post_content(
                 title=request.post_title,
                 content=request.post_content,
                 tags=request.post_tags,
             )
 
-            # 올바른 응답 데이터를 직접 구성
-            response_data = {
-                "tag": request.tag,
-                "post_title": request.post_title,
-                "publish_success": True,  # 포스팅 성공 가정
-            }
-
             if not response_data:
                 raise CustomException(
-                    f"{request.tag} 블로그 포스팅에 실패했습니다.", status_code=500
+                    500, f"{request.tag} 블로그 포스팅에 실패했습니다.", "POSTING_FAIL"
                 )
 
             return response_data
@@ -45,5 +51,5 @@ def publish_content(self, request: RequestBlogPublish) -> Dict:
         except Exception as e:
             # 예상치 못한 예외 처리
             raise CustomException(
-                f"블로그 포스팅 중 오류가 발생했습니다: {str(e)}", status_code=500
+                500, f"블로그 포스팅 중 오류가 발생했습니다: {str(e)}", "ERROR"
             )

apps/pre-processing-service/app/service/blog/blog_service_factory.py

@@ -1,4 +1,4 @@
-from typing import Dict, Type
+from typing import Dict, Type, Optional
 from app.service.blog.base_blog_post_service import BaseBlogPostService
 from app.service.blog.naver_blog_post_service import NaverBlogPostService
 from app.service.blog.tistory_blog_post_service import TistoryBlogPostService
@@ -11,15 +11,26 @@ class BlogServiceFactory:
 
     # 서비스 타입별 클래스 매핑
     _services: Dict[str, Type[BaseBlogPostService]] = {
-        "naver": NaverBlogPostService,
-        "tistory": TistoryBlogPostService,
+        "naver_blog": NaverBlogPostService,
+        "tistory_blog": TistoryBlogPostService,
         "blogger": BloggerBlogPostAdapter,
     }
 
     @classmethod
-    def create_service(cls, platform: str) -> BaseBlogPostService:
+    def create_service(
+        cls,
+        platform: str,
+        blog_id: str,
+        blog_password: str,
+        blog_name: Optional[str] = None,
+    ) -> BaseBlogPostService:
         """
         플랫폼에 따른 블로그 서비스 인스턴스 생성
+
+        Args:
+            platform: 블로그 플랫폼 (naver, tistory, blogger)
+            blog_id: 블로그 아이디
+            blog_password: 블로그 비밀번호
         """
         service_class = cls._services.get(platform.lower())
 
@@ -30,7 +41,18 @@ def create_service(cls, platform: str) -> BaseBlogPostService:
                 status_code=400,
             )
 
-        return service_class()
+        # 각 서비스의 설정을 의존성 주입
+        if platform.lower() == "tistory_blog":
+            if not blog_name:
+                raise CustomException(
+                    200,
+                    "티스토리 블로그가 존재하지않습니다.",
+                    "NOT_FOUND_BLOG",
+                )
+            return service_class(blog_id, blog_password, blog_name)
+        if platform.lower() == "blogger":
+            return service_class()
+        return service_class(blog_id, blog_password)
 
     @classmethod
     def get_supported_platforms(cls) -> list:

apps/pre-processing-service/app/service/blog/blogger_blog_post_adapter.py

@@ -55,7 +55,7 @@ def _write_content(self, title: str, content: str, tags: List[str] = None) -> No
 
     def _get_platform_name(self) -> str:
         """플랫폼 이름 반환"""
-        return "Blogger"
+        return "BLOGGER"
 
     def _validate_content(
         self, title: str, content: str, tags: Optional[List[str]] = None

apps/pre-processing-service/app/service/blog/naver_blog_post_service.py

@@ -1,4 +1,3 @@
-import os
 import time
 import pyperclip
 
@@ -15,13 +14,28 @@
 class NaverBlogPostService(BaseBlogPostService):
     """네이버 블로그 포스팅 서비스 구현"""
 
+    def __init__(self, blog_id: str, blog_password: str, use_webdriver=True):
+        """네이버 블로그 서비스 초기화
+
+        Args:
+            blog_id: 네이버 아이디
+            blog_password: 네이버 비밀번호
+            use_webdriver: 웹드라이버 사용 여부
+        """
+        self.blog_id = blog_id
+        self.blog_password = blog_password
+        print(blog_id)
+        print(blog_password)
+        super().__init__(use_webdriver)
+
     def _load_config(self) -> None:
         """네이버 블로그 설정 로드"""
-
-        self.id = os.getenv("NAVER_ID", "all2641")
-        self.password = os.getenv("NAVER_PASSWORD", "cjh83520*")
+        self.id = self.blog_id
+        self.password = self.blog_password
         self.login_url = "https://nid.naver.com/nidlogin.login"
         self.post_content_url = f"https://blog.naver.com/PostWriteForm.naver?blogId={self.id}&Redirect=Write&redirect=Write&widgetTypeCall=true&noTrackingCode=true&directAccess=false"
+        # print(self.id)
+        # print(self.password)
 
     def _get_platform_name(self) -> str:
         return "NAVER_BLOG"
@@ -93,7 +107,7 @@ def _login(self) -> None:
         except Exception as e:
             raise BlogLoginException("네이버 블로그", f"예상치 못한 오류: {str(e)}")
 
-    def _write_content(self, title: str, content: str, tags: List[str] = None) -> None:
+    def _write_content(self, title: str, content: str, tags: List[str] = None) -> str:
         """네이버 블로그 포스팅 작성 구현"""
         from selenium.webdriver.common.by import By
         from selenium.webdriver.support import expected_conditions as EC
@@ -193,8 +207,10 @@ def _write_content(self, title: str, content: str, tags: List[str] = None) -> No
                     self.web_driver.execute_script("arguments[0].click();", final_btn)
             except TimeoutException:
                 raise BlogElementInteractionException("최종 발행 버튼", "버튼 클릭")
+            time.sleep(5)
 
-            # 발행 완료 확인
+            # 발행 완료 확인 및 URL 가져오기
+            blog_url = None
             try:
                 self.wait_driver.until(
                     EC.any_of(
@@ -204,8 +220,36 @@ def _write_content(self, title: str, content: str, tags: List[str] = None) -> No
                         EC.url_contains("entry.naver"),
                     )
                 )
+                # 현재 URL 가져오기
+                current_url = self.web_driver.current_url
+
+                # PostView URL인 경우 해당 URL을 반환
+                if "PostView.naver" in current_url or "entry.naver" in current_url:
+                    blog_url = current_url
+                # postList인 경우 가장 최근 포스트 URL 찾기
+                elif "postList" in current_url:
+                    try:
+                        # 가장 최근 포스트 링크 찾기
+                        recent_post = self.wait_driver.until(
+                            EC.element_to_be_clickable(
+                                (
+                                    By.CSS_SELECTOR,
+                                    ".post_area .post_item:first-child .title_area a",
+                                )
+                            )
+                        )
+                        blog_url = recent_post.get_attribute("href")
+                    except TimeoutException:
+                        # 대안으로 현재 URL 사용
+                        blog_url = current_url
+                else:
+                    blog_url = current_url
+
             except TimeoutException:
-                pass
+                # 발행 완료를 확인할 수 없는 경우 현재 URL 사용
+                blog_url = self.web_driver.current_url
+            print(f"blog_url: {blog_url}")
+            return blog_url
 
         except (BlogElementInteractionException, BlogPostPublishException):
             raise

apps/pre-processing-service/app/service/blog/tistory_blog_post_service.py

@@ -1,5 +1,8 @@
 import os
 import time
+import json
+import requests
+from datetime import datetime
 
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support import expected_conditions as EC
@@ -13,12 +16,27 @@
 class TistoryBlogPostService(BaseBlogPostService):
     """티스토리 블로그 포스팅 서비스"""
 
+    def __init__(
+        self, blog_id: str, blog_password: str, blog_name: str, use_webdriver=True
+    ):
+        """네이버 블로그 서비스 초기화
+
+        Args:
+            blog_id: 네이버 아이디
+            blog_password: 네이버 비밀번호
+            use_webdriver: 웹드라이버 사용 여부
+        """
+        self.blog_id = blog_id
+        self.blog_password = blog_password
+        self.blog_name = blog_name
+        super().__init__(use_webdriver)
+
     def _load_config(self) -> None:
         """티스토리 블로그 설정 로드"""
 
-        self.blog_name = os.getenv("TISTORY_BLOG_NAME", "hoons2641")
-        self.id = os.getenv("TISTORY_ID", "[email protected]")
-        self.password = os.getenv("TISTORY_PASSWORD", "kdyn264105*")
+        self.blog_name = self.blog_name
+        self.id = self.blog_id
+        self.password = self.blog_password
         self.login_url = "https://accounts.kakao.com/login/?continue=https%3A%2F%2Fkauth.kakao.com%2Foauth%2Fauthorize%3Fclient_id%3D3e6ddd834b023f24221217e370daed18%26state%3DaHR0cHM6Ly93d3cudGlzdG9yeS5jb20v%26redirect_uri%3Dhttps%253A%252F%252Fwww.tistory.com%252Fauth%252Fkakao%252Fredirect%26response_type%3Dcode%26auth_tran_id%3Dslj3F.mFC~2JNOiCOGi5HdGPKOA.Pce4l5tiS~3fZkInLGuEG3tMq~xZkxx4%26ka%3Dsdk%252F2.7.3%2520os%252Fjavascript%2520sdk_type%252Fjavascript%2520lang%252Fko-KR%2520device%252FMacIntel%2520origin%252Fhttps%25253A%25252F%25252Fwww.tistory.com%26is_popup%3Dfalse%26through_account%3Dtrue&talk_login=hidden#login"
         self.post_content_url = f"https://{self.blog_name}.tistory.com/manage/newpost"
 
@@ -90,7 +108,60 @@ def _login(self) -> None:
         except Exception as e:
             raise BlogLoginException("티스토리 블로그", f"예상치 못한 오류: {str(e)}")
 
-    def _write_content(self, title: str, content: str, tags: List[str] = None) -> None:
+    def _get_post_url_from_api(self, title: str) -> str:
+        """API를 통해 제목이 일치하는 가장 최근 포스트의 URL을 가져옴"""
+        try:
+            # 현재 세션의 쿠키를 가져와서 API 요청에 사용
+            cookies = self.web_driver.get_cookies()
+            session_cookies = {}
+            for cookie in cookies:
+                session_cookies[cookie["name"]] = cookie["value"]
+
+            # 포스트 목록 API 호출
+            api_url = f"https://{self.blog_name}.tistory.com/manage/posts.json"
+            params = {
+                "category": "-3",
+                "page": "1",
+                "searchKeyword": "",
+                "searchType": "title",
+                "visibility": "all",
+            }
+
+            response = requests.get(api_url, params=params, cookies=session_cookies)
+
+            if response.status_code == 200:
+                data = response.json()
+                items = data.get("items", [])
+
+                # 제목이 일치하는 포스트들 찾기
+                matching_posts = [item for item in items if item["title"] == title]
+
+                if matching_posts:
+                    # created 시간으로 정렬하여 가장 최근 포스트 찾기
+                    latest_post = max(
+                        matching_posts,
+                        key=lambda x: datetime.strptime(x["created"], "%Y-%m-%d %H:%M"),
+                    )
+                    return latest_post["permalink"]
+                else:
+                    # 매칭되는 포스트가 없으면 가장 최근 포스트 반환
+                    if items:
+                        latest_post = max(
+                            items,
+                            key=lambda x: datetime.strptime(
+                                x["created"], "%Y-%m-%d %H:%M"
+                            ),
+                        )
+                        return latest_post["permalink"]
+
+            # API 호출 실패 시 블로그 메인 URL 반환
+            return f"https://{self.blog_name}.tistory.com"
+
+        except Exception:
+            # 오류 발생 시 블로그 메인 URL 반환
+            return f"https://{self.blog_name}.tistory.com"
+
+    def _write_content(self, title: str, content: str, tags: List[str] = None) -> str:
         """티스토리 블로그 포스팅 작성 구현"""
 
         try:
@@ -231,6 +302,11 @@ def _write_content(self, title: str, content: str, tags: List[str] = None) -> No
                         "티스토리 블로그", "발행 과정에서 오류가 발생했습니다"
                     )
 
+            # 발행 완료 확인 및 URL 가져오기
+            time.sleep(3)  # 발행 완료 대기
+            blog_url = self._get_post_url_from_api(title)
+            return blog_url
+
         except (BlogElementInteractionException, BlogPostPublishException):
             raise
         except TimeoutException: