import requests
import os
import sqlite3
from bs4 import BeautifulSoup
import logging
# 로깅 설정
logging.basicConfig(level=logging.ERROR)
class BlogPost:
def __init__(self, title, description, link, postdate):
self.title = self.clean_html(title)
self.description = self.clean_html(description)
self.link = link
self.postdate = postdate
def clean_html(self, text):
if text:
return BeautifulSoup(text, "html.parser").get_text()
return ""
def __str__(self):
return f"Title: {self.title}\nDescription: {self.description}\nURL: {self.link}\nPost Date: {self.postdate}\n"
class NaverBlogSearcher:
def __init__(self):
self.client_id = os.getenv("NAVER_CLIENT_ID") # 환경변수에서 Client ID 읽어오기
self.client_secret = os.getenv("NAVER_CLIENT_SECRET") # 환경변수에서 Client Secret 읽어오기
# 환경변수 값이 없을 때 예외 처리
if not self.client_id or not self.client_secret:
logging.error("NAVER_CLIENT_ID and NAVER_CLIENT_SECRET must be set as environment variables.")
raise ValueError("NAVER_CLIENT_ID and NAVER_CLIENT_SECRET must be set as environment variables.")
def get_blog_posts(self, query, display=10):
# 요청 URL 및 헤더 구성
url = f"https://openapi.naver.com/v1/search/blog.json?query={query}&display={display}&sort=date"
headers = {
"X-Naver-Client-Id": self.client_id,
"X-Naver-Client-Secret": self.client_secret,
}
# 네이버 API에 요청 보내기
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
logging.error(f"Network error occurred: {e}")
return []
if response.status_code == 200:
return self.parse_response(response.json())
else:
print(f"Error: {response.status_code}, {response.text}")
return []
def parse_response(self, data):
# 응답 데이터 파싱
blogs = data.get("items", [])
blog_posts = [BlogPost(blog.get("title"), blog.get("description"), blog.get("link"), blog.get("postdate")) for blog in blogs]
return blog_posts
class BlogPostDBHandler:
def __init__(self, db_name="blog_posts.db"):
self.db_name = db_name
self.create_table()
def create_table(self):
# SQLite 데이터베이스 연결
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
# 테이블 생성
cursor.execute('''
CREATE TABLE IF NOT EXISTS BlogPosts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
description TEXT,
link TEXT UNIQUE,
postdate TEXT
)
''')
def save_to_db(self, blog_posts):
# SQLite 데이터베이스 연결
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
# 데이터 삽입 (중복 확인)
for post in blog_posts:
cursor.execute('''
SELECT COUNT(*) FROM BlogPosts WHERE link = ?
''', (post.link,))
if cursor.fetchone()[0] == 0:
cursor.execute('''
INSERT INTO BlogPosts (title, description, link, postdate)
VALUES (?, ?, ?, ?)
''', (post.title, post.description, post.link, post.postdate))
def get_all_posts(self):
# SQLite 데이터베이스 연결
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
# 저장된 블로그 포스팅의 모든 필드 가져오기
cursor.execute('''
SELECT title, description, link, postdate FROM BlogPosts
''')
rows = cursor.fetchall()
# 필드명을 갖는 dict 형태로 변환
posts = [
{"title": row[0], "description": row[1], "link": row[2], "postdate": row[3]}
for row in rows
]
return posts
if __name__ == "__main__":
# "동탄" 검색어로 최신 블로그 정보 가져오기
try:
searcher = NaverBlogSearcher()
blog_posts = searcher.get_blog_posts("동탄")
# 검색 결과 출력
# for post in blog_posts:
# print(post)
# 검색 결과를 SQLite 데이터베이스에 저장
db_handler = BlogPostDBHandler()
db_handler.save_to_db(blog_posts)
# 저장된 블로그 포스팅의 제목 출력
posts = db_handler.get_all_posts()
for post in posts:
print(f"Title: {post['title']}")
except ValueError as e:
print(e)
TABLE이 존재하지 않으면 create 하도록 했으며,
블로그 포스팅을 insert 하기 전에 기존에 있는지 미리 확인한 후에 없을 때에만 insert 하도록 했다.
아직은 PoC 성격의 스크립트 이기에,
SQLite에 저장한 이후에 전체 포스팅 목록을 SQLite로부터 읽어와서 제목들을 출력하도록 했다.
