[FastAPI/Crawling] 네이버 뉴스 API를 활용해 야구 뉴스 데이터 수집하기

토이 프로젝트로 야구장 관련 홈페이지를 만들고 있는데, KBO 야구장을 소개하는 것이다보니 해당 뉴스를 함께 노출할 수 있었으면 좋겠다는 생각에 야구 뉴스만을 수집하는 크롤러를 제작해보았다.

 

실시간으로 뉴스 업데이트를 해주는 것이 목표이기 때문에, 기본적으로 비동기적 실행이 가능하도록 제작했다.

 

우선 '야구'라는 키워드를 가진 뉴스를 모두 수집한 다음 link라는 칼럼(네이버 뉴스 상의 하이퍼링크)에서 'kbaseball'이라는 키워드가 포함된 것만 DB에 저장할 리스트에 추가하는 방식으로 설계했다.

 

import aiohttp
import asyncio
from config import get_secret


class NaverNewsScraper:

    NAVER_API_NEWS = "https://openapi.naver.com/v1/search/news.json"
    # API ID와 PW은 자신의 것을 직접 입력하면 된다.
    NAVER_API_ID = get_secret("NAVER_API_ID")
    NAVER_API_SECRET = get_secret("NAVER_API_SECRET")

	# session을 열어 json으로 리턴되는 값 중 ['items']칼럼에 해당하는 값만 돌려받는다.
    @staticmethod
    async def fetch(session, url, headers):
        async with session.get(url, headers=headers) as response:
            if response.status == 200:
                result = await response.json()
                return result["items"]

	# start(시작할 부분), display(한 번에 수집(노출)할 개수)를 설정해준다.
    def unit_url(self, keyword, start):
        return {
            "url": f"{self.NAVER_API_NEWS}?query={keyword}&display=10&start={start}",
            "headers": {
                "X-Naver-Client-Id": self.NAVER_API_ID,
                "X-Naver-Client-Secret": self.NAVER_API_SECRET,
            },
        }

    async def search(self, keyword, total_page):
        # 스크래핑할 url apis list에 입력
        apis = [self.unit_url(keyword, 1 + i * 10) for i in range(total_page)]
        # 데이터를 수집하는 부분
        async with aiohttp.ClientSession() as session:
            all_data = await asyncio.gather(
                *[NaverNewsScraper.fetch(session, api["url"], api["headers"]) for api in apis]
            )
            # DB에 input하기 좋은 형식으로 수정해준다.
            result = []
            for data in all_data:
                if data is not None:
                    for news in data:
                    	# link 내 'kbaseball'이라는 문자열이 포함될 때만 DB에 저장할 뉴스 리스트에 추가
                        if 'kbaseball' in news['link']:
                            result.append(news)
                        else:
                            pass
            return result

    def run(self, keyword, total_page):
        return asyncio.run(self.search(keyword, total_page))

# 이 부분은 해당 파일만 실행하여 테스트할 목적이다.
if __name__ == "__main__":
    scraper = NaverNewsScraper()
    results = scraper.run("야구", 3)
    print(results)