"""네이버 블로그 본문 크롤링 모듈.""" import asyncio import logging import re from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlparse import httpx from bs4 import BeautifulSoup logger = logging.getLogger(__name__) _TIMEOUT = 10 # 글당 크롤링 타임아웃 (초) _MAX_CONTENT_LENGTH = 2000 # 본문 최대 길이 # 네이버 블로그 URL 패턴: blog.naver.com/{blogId}/{logNo} _BLOG_URL_RE = re.compile(r"blog\.naver\.com/([^/]+)/(\d+)") def _parse_naver_blog_url(url: str) -> Optional[Tuple[str, str]]: """네이버 블로그 URL에서 blogId, logNo 추출. 실패 시 None.""" match = _BLOG_URL_RE.search(url) if not match: return None return match.group(1), match.group(2) async def _fetch_html(url: str) -> str: """URL에서 HTML을 가져온다.""" async with httpx.AsyncClient(timeout=_TIMEOUT, follow_redirects=True) as client: resp = await client.get(url, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" }) resp.raise_for_status() return resp.text def _extract_text(html: str) -> str: """HTML에서 본문 텍스트를 추출한다.""" soup = BeautifulSoup(html, "html.parser") # 스마트에디터 3 (SE3) container = soup.select_one("div.se-main-container") if not container: # 구 에디터 container = soup.select_one("div#postViewArea") if not container: # 폴백: body 전체 container = soup.body if not container: return "" # 스크립트/스타일 제거 for tag in container.find_all(["script", "style"]): tag.decompose() text = container.get_text(separator="\n", strip=True) return text[:_MAX_CONTENT_LENGTH] async def crawl_blog_content(url: str) -> str: """네이버 블로그 URL에서 본문 텍스트 추출. - 네이버 블로그가 아니면 빈 문자열 - 크롤링 실패 시 빈 문자열 (에러 로그만) - 본문 최대 2,000자 """ parsed = _parse_naver_blog_url(url) if not parsed: return "" blog_id, log_no = parsed # iframe 내부 실제 본문 URL post_url = f"https://blog.naver.com/PostView.naver?blogId={blog_id}&logNo={log_no}" try: html = await _fetch_html(post_url) return _extract_text(html) except Exception as e: logger.warning("블로그 크롤링 실패 (%s): %s", url, e) return "" async def enrich_top_blogs(top_blogs: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """top_blogs 리스트 각 항목에 content 필드를 추가. 개별 크롤링 실패 시 해당 항목의 content를 빈 문자열로 설정하고 나머지 계속 진행. """ result = [] for blog in top_blogs: enriched = dict(blog) try: enriched["content"] = await crawl_blog_content(blog.get("link", "")) except Exception: enriched["content"] = "" result.append(enriched) return result