개인 카카오톡 대화방 내용을 노션DB로 반자동 기록 관리, 시스템 구축하기 🚀

import re
import requests
import logging
from datetime import datetime
from notion_client import Client 
from dateutil import parser
from urllib.parse import urlparse
import time
from typing import Dict, List, Optional
import pytz
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By

NOTION_TOKEN = " 노션 API 연결 "
DATABASE_ID = " 노션 DB ID 연결 "

CONTENT_TYPE_PATTERNS = {
   'youtube.com': '유튜브',
   'youtu.be': '유튜브',
   'blog.naver.com': '블로그',
   'post.naver.com': '블로그', 
   'news.naver.com': '뉴스',
   '.com': '기타',
   '.kr': '기타'
}

class URLProcessor:
   def __init__(self):
       chrome_options = Options()
       chrome_options.add_argument('--headless')
       chrome_options.add_argument('--no-sandbox')
       chrome_options.add_argument('--disable-dev-shm-usage')
       self.driver = webdriver.Chrome(options=chrome_options)

   def __del__(self):
       self.driver.quit()

   def get_page_title(self, url: str) -> str:
       try:
           self.driver.get(url)
           time.sleep(2)
           title = self.driver.title
           if not title:
               title_meta = self.driver.find_elements(By.CSS_SELECTOR, 'meta[property="og:title"]')
               if title_meta:
                   title = title_meta[0].get_attribute('content')
           return title or "제목 없음"
       except Exception as e:
           logging.error(f"제목 추출 실패: {str(e)}")
           return "제목 추출 실패"

   def get_publish_date(self, url: str) -> Optional[datetime]:
       try:
           self.driver.get(url)
           time.sleep(2)
           
           date_selectors = [
               'meta[property="article:published_time"]',
               'meta[property="og:updated_time"]',
               'meta[name="date"]',
               'meta[name="pubdate"]',
               'time[datetime]'
           ]
           
           for selector in date_selectors:
               elements = self.driver.find_elements(By.CSS_SELECTOR, selector)
               for element in elements:
                   date_str = element.get_attribute('content') or element.get_attribute('datetime')
                   if date_str:
                       try:
                           return parser.parse(date_str)
                       except:
                           continue
           return None
       except Exception as e:
           logging.error(f"게시일 추출 실패: {str(e)}")
           return None

   def determine_content_type(self, url: str) -> str:
       domain = urlparse(url).netloc.lower()
       for pattern, content_type in CONTENT_TYPE_PATTERNS.items():
           if pattern in domain:
               return content_type
       return "기타"

class NotionManager:
   def __init__(self, token: str, database_id: str):
       self.notion = Client(auth=token)
       self.database_id = database_id
       self.url_processor = URLProcessor()

   def create_page(self, url: str, chat_date: datetime) -> bool:
       try:
           title = self.url_processor.get_page_title(url)
           content_type = self.url_processor.determine_content_type(url)
           publish_date = self.url_processor.get_publish_date(url)
           current_time = datetime.now(pytz.UTC)

           self.notion.pages.create(
               parent={"database_id": self.database_id},
               properties={
                   "콘텐츠 타입": {"multi_select": [{"name": content_type}]},
                   "제목": {"title": [{"text": {"content": title}}]},
                   "URL": {"url": url},
                   "게시일": {"date": {"start": (publish_date or chat_date).isoformat()}},
                   "업로드 날짜": {"date": {"start": current_time.isoformat()}},
                   "상태": {"multi_select": [{"name": "미확인"}]},
                   "주제별 태깅": {"multi_select": []},
                   "내용 요약(AI)": {"rich_text": [{"text": {"content": ""}}]},
                   "메모": {"rich_text": [{"text": {"content": ""}}]}
               }
           )
           return True
       except Exception as e:
           logging.error(f"페이지 생성 실패: {str(e)}")
           return False

class KakaoTalkProcessor:
   def __init__(self, file_path: str):
       self.file_path = file_path

   def extract_messages(self) -> List[Dict]:
       with open(self.file_path, 'r', encoding='utf-8') as f:
           content = f.read()

       pattern = r'(\d{4}년\s*\d{1,2}월\s*\d{1,2}일\s*(?:오전|오후)\s*\d{1,2}:\d{1,2})[^\n]*\n[^\n]*?(https?://\S+)'
       matches = re.finditer(pattern, content)
       
       messages = []
       for match in matches:
           date_str = match.group(1)
           url = match.group(2).strip()
           
           try:
               date_str = date_str.replace('년 ', '-').replace('월 ', '-').replace('일 ', ' ')
               
               if '오후' in date_str:
                   time_part = date_str.split('오후 ')[1]
                   hour = int(time_part.split(':')[0])
                   if hour < 12:
                       hour += 12
                   date_str = date_str.replace(f'오후 {time_part}', f'{hour}:{time_part.split(":")[1]}')
               elif '오전' in date_str:
                   time_part = date_str.split('오전 ')[1]
                   hour = int(time_part.split(':')[0])
                   if hour == 12:
                       hour = 0
                   date_str = date_str.replace(f'오전 {time_part}', f'{hour:02d}:{time_part.split(":")[1]}')
               
               date = datetime.strptime(date_str, '%Y-%m-%d %H:%M')
               messages.append({
                   'date': date,
                   'url': url
               })
           except Exception as e:
               logging.error(f"날짜 파싱 오류: {date_str} - {str(e)}")
               continue

       return messages

def main():
   logging.basicConfig(
       level=logging.INFO,
       format='%(asctime)s - %(levelname)s - %(message)s',
       filename='notion_upload.log'
   )

   try:
       kakao_processor = KakaoTalkProcessor('카카오톡 대화.eml')
       notion_manager = NotionManager(NOTION_TOKEN, DATABASE_ID)

       messages = kakao_processor.extract_messages()
       logging.info(f"총 {len(messages)}개의 URL 발견")

       success_count = 0
       for message in messages:
           time.sleep(1)
           
           if notion_manager.create_page(message['url'], message['date']):
               success_count += 1
               logging.info(f"업로드 성공: {message['url']}")
           else:
               logging.error(f"업로드 실패: {message['url']}")

       logging.info(f"작업 완료. 성공: {success_count}, 실패: {len(messages) - success_count}")

   except Exception as e:
       logging.error(f"프로그램 실행 중 오류 발생: {str(e)}")
       raise

if __name__ == "__main__":
   main()