100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
"""네이버 블로그 본문 크롤링 모듈."""
|
|
|
|
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
|