diff --git a/apps/pre-processing-service/app/model/schemas.py b/apps/pre-processing-service/app/model/schemas.py index 36bef959..0bfecf88 100644 --- a/apps/pre-processing-service/app/model/schemas.py +++ b/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="포스팅 태그 목록") diff --git a/apps/pre-processing-service/app/service/blog/base_blog_post_service.py b/apps/pre-processing-service/app/service/blog/base_blog_post_service.py index f55bdba0..8bc9c9a8 100644 --- a/apps/pre-processing-service/app/service/blog/base_blog_post_service.py +++ b/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): diff --git a/apps/pre-processing-service/app/service/blog/blog_publish_service.py b/apps/pre-processing-service/app/service/blog/blog_publish_service.py index b7727cce..0848f123 100644 --- a/apps/pre-processing-service/app/service/blog/blog_publish_service.py +++ b/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" ) diff --git a/apps/pre-processing-service/app/service/blog/blog_service_factory.py b/apps/pre-processing-service/app/service/blog/blog_service_factory.py index b6bc6883..4759b5ab 100644 --- a/apps/pre-processing-service/app/service/blog/blog_service_factory.py +++ b/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: diff --git a/apps/pre-processing-service/app/service/blog/blogger_blog_post_adapter.py b/apps/pre-processing-service/app/service/blog/blogger_blog_post_adapter.py index 717a102e..3f4a67e9 100644 --- a/apps/pre-processing-service/app/service/blog/blogger_blog_post_adapter.py +++ b/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 diff --git a/apps/pre-processing-service/app/service/blog/naver_blog_post_service.py b/apps/pre-processing-service/app/service/blog/naver_blog_post_service.py index 0e33a9fd..702211a4 100644 --- a/apps/pre-processing-service/app/service/blog/naver_blog_post_service.py +++ b/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 diff --git a/apps/pre-processing-service/app/service/blog/tistory_blog_post_service.py b/apps/pre-processing-service/app/service/blog/tistory_blog_post_service.py index cc830bac..0b4d98d0 100644 --- a/apps/pre-processing-service/app/service/blog/tistory_blog_post_service.py +++ b/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", "fair_05@nate.com") - 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: