ChatGPT로 얻은 정보를 매번 수동으로 복사해서 Notion에 붙여넣는 작업이 번거로우신가요? 파이썬을 활용하면 이 과정을 완전히 자동화할 수 있어요. OpenAI API와 Notion API를 연결해서 ChatGPT 응답을 자동으로 요약하고 Notion 데이터베이스에 정리하는 시스템을 만들어볼게요.
필요한 API 키 준비하기
먼저 OpenAI API 키와 Notion Integration 토큰이 필요해요. OpenAI는 platform.openai.com에서 발급받을 수 있고, Notion은 notion.so/my-integrations에서 새로운 통합을 만들어 토큰을 받아요.
# macOS 터미널에서 환경변수 설정
echo 'export OPENAI_API_KEY="sk-..."' >> ~/.zshrc
echo 'export NOTION_TOKEN="secret_..."' >> ~/.zshrc
source ~/.zshrc
기본 라이브러리 설치하기
macOS 터미널을 열고 필요한 패키지들을 설치해요. pip가 없다면 먼저
python3 -m ensurepip
로 설치하세요.
pip3 install openai requests python-dotenv
ChatGPT 응답 요약 함수 만들기
OpenAI API를 사용해서 긴 텍스트를 요약하는 함수를 만들어요. 프롬프트 엔지니어링을 통해 원하는 형태의 요약을 얻을 수 있어요.
import openai
import os
from dotenv import load_dotenv
load_dotenv()
def generate_summary(text, max_length=3):
"""ChatGPT를 사용해 텍스트를 요약하는 함수"""
openai.api_key = os.getenv('OPENAI_API_KEY')
try:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": "당신은 핵심만 간단히 정리하는 요약 전문가입니다."
},
{
"role": "user",
"content": f"다음 내용을 {max_length}줄로 요약해주세요:\n\n{text}"
}
],
temperature=0.3,
max_tokens=500
)
return response['choices'][0]['message']['content']
except Exception as e:
print(f"요약 중 오류 발생: {e}")
return None
Notion API 연동 설정하기
Notion 데이터베이스에 요약된 내용을 저장하는 함수를 만들어요. Notion API는 특정한 JSON 구조를 요구하므로 정확히 맞춰줘야 해요.
import requests
import json
from datetime import datetime
def add_to_notion(title, summary, original_text=""):
"""Notion 데이터베이스에 요약 내용을 추가하는 함수"""
notion_token = os.getenv('NOTION_TOKEN')
database_id = os.getenv('NOTION_DATABASE_ID')
headers = {
"Authorization": f"Bearer {notion_token}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28"
}
# Notion 페이지 데이터 구조
data = {
"parent": {"database_id": database_id},
"properties": {
"제목": {
"title": [{
"text": {"content": title}
}]
},
"요약": {
"rich_text": [{
"text": {"content": summary}
}]
},
"날짜": {
"date": {
"start": datetime.now().isoformat()
}
},
"원문": {
"rich_text": [{
"text": {"content": original_text[:2000]} # 2000자 제한
}]
}
}
}
try:
response = requests.post(
"https://api.notion.com/v1/pages",
headers=headers,
data=json.dumps(data)
)
if response.status_code == 200:
print("✅ Notion에 성공적으로 저장되었습니다!")
return True
else:
print(f"❌ 오류 발생: {response.status_code}")
print(response.text)
return False
except Exception as e:
print(f"❌ 요청 중 오류: {e}")
return False
에러 처리와 재시도 로직 추가하기
API 호출은 네트워크 문제나 속도 제한으로 실패할 수 있어요. 안정적인 시스템을 위해 재시도 로직을 추가해요.
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=2):
"""API 호출 실패 시 자동 재시도하는 데코레이터"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
result = func(*args, **kwargs)
if result is not None:
return result
except Exception as e:
if attempt < max_retries - 1:
print(f"재시도 {attempt + 1}/{max_retries}...")
time.sleep(delay * (attempt + 1))
else:
raise e
return None
return wrapper
return decorator
# 데코레이터 적용
@retry_on_failure(max_retries=3, delay=2)
def generate_summary_with_retry(text, max_length=3):
return generate_summary(text, max_length)
전체 자동화 워크플로우 구현하기
이제 모든 요소를 합쳐서 완전한 자동화 시스템을 만들어요. 파일이나 텍스트를 읽어서 요약하고 Notion에 저장하는 전체 프로세스예요.
class ChatGPTNotionAutomation:
def __init__(self):
load_dotenv()
self.setup_apis()
def setup_apis(self):
"""API 설정 확인"""
required_vars = ['OPENAI_API_KEY', 'NOTION_TOKEN', 'NOTION_DATABASE_ID']
for var in required_vars:
if not os.getenv(var):
raise ValueError(f"{var}가 설정되지 않았습니다.")
def process_text(self, text, title="ChatGPT 요약"):
"""텍스트를 처리하여 요약 후 Notion에 저장"""
# 긴 텍스트 분할 처리
if len(text) > 3000:
chunks = self.split_text(text, 3000)
summaries = []
for i, chunk in enumerate(chunks):
print(f"청크 {i+1}/{len(chunks)} 처리 중...")
summary = generate_summary_with_retry(chunk)
if summary:
summaries.append(summary)
final_summary = "\n\n".join(summaries)
else:
final_summary = generate_summary_with_retry(text)
# Notion에 저장
if final_summary:
success = add_to_notion(title, final_summary, text)
return success
return False
def split_text(self, text, chunk_size):
"""긴 텍스트를 적절한 크기로 분할"""
words = text.split()
chunks = []
current_chunk = []
current_size = 0
for word in words:
current_size += len(word) + 1
if current_size > chunk_size:
chunks.append(' '.join(current_chunk))
current_chunk = [word]
current_size = len(word)
else:
current_chunk.append(word)
if current_chunk:
chunks.append(' '.join(current_chunk))
return chunks
실제 사용 예제
만든 자동화 시스템을 실제로 사용하는 방법이에요. 파일에서 텍스트를 읽거나 직접 입력한 내용을 처리할 수 있어요.
# 자동화 시스템 초기화
automation = ChatGPTNotionAutomation()
# 1. 텍스트 파일에서 읽어서 처리
with open('article.txt', 'r', encoding='utf-8') as f:
content = f.read()
automation.process_text(content, "기술 문서 요약")
# 2. ChatGPT 대화 내용 직접 처리
chat_response = """
여기에 ChatGPT의 긴 응답 내용이 들어갑니다.
여러 주제에 대한 상세한 설명이 포함되어 있을 수 있습니다.
"""
automation.process_text(chat_response, "ChatGPT 대화 요약")
# 3. 여러 파일 일괄 처리
import glob
for file_path in glob.glob('documents/*.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
file_name = os.path.basename(file_path)
automation.process_text(content, f"요약: {file_name}")
Notion 데이터베이스 구조 설정하기
Notion에서 데이터베이스를 만들 때는 다음과 같은 속성들을 추가해요. 각 속성의 타입을 정확히 맞춰야 API가 제대로 작동해요.
- 제목 (Title): 기본 제목 속성
- 요약 (Text): 요약된 내용 저장
- 날짜 (Date): 생성 시간 자동 기록
- 원문 (Text): 원본 텍스트 일부 저장
- 태그 (Multi-select): 분류용 태그
- 출처 (URL): 원본 링크나 참조
보안과 성능 최적화하기
실무에서 사용할 때는 보안과 성능을 더욱 신경써야 해요. API 키를 안전하게 관리하고 대용량 처리를 위한 최적화가 필요해요.
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
class OptimizedAutomation(ChatGPTNotionAutomation):
def __init__(self):
super().__init__()
self.executor = ThreadPoolExecutor(max_workers=5)
async def process_multiple_texts(self, text_list):
"""여러 텍스트를 비동기로 동시 처리"""
tasks = []
async with aiohttp.ClientSession() as session:
for text, title in text_list:
task = self.async_process(session, text, title)
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
async def async_process(self, session, text, title):
"""비동기 처리 함수"""
# OpenAI API는 동기 호출이므로 스레드풀 사용
loop = asyncio.get_event_loop()
summary = await loop.run_in_executor(
self.executor,
generate_summary_with_retry,
text
)
if summary:
# Notion API 호출
success = await self.async_notion_upload(session, title, summary, text)
return success
return False
실무 적용 시 추가 고려사항
자동화 시스템을 실제 업무에 적용할 때는 몇 가지 추가 기능이 필요해요. 로깅 시스템을 구축하고 모니터링 체계를 만들어야 안정적으로 운영할 수 있어요.
import logging
from datetime import datetime
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('automation.log'),
logging.StreamHandler()
]
)
class ProductionAutomation(OptimizedAutomation):
def __init__(self):
super().__init__()
self.stats = {
'processed': 0,
'failed': 0,
'total_chars': 0
}
def process_with_logging(self, text, title):
"""로깅과 통계를 포함한 처리"""
start_time = datetime.now()
try:
logging.info(f"처리 시작: {title}")
success = self.process_text(text, title)
if success:
self.stats['processed'] += 1
logging.info(f"처리 완료: {title}")
else:
self.stats['failed'] += 1
logging.error(f"처리 실패: {title}")
self.stats['total_chars'] += len(text)
# 처리 시간 기록
duration = (datetime.now() - start_time).total_seconds()
logging.info(f"처리 시간: {duration:.2f}초")
return success
except Exception as e:
logging.error(f"예외 발생: {title} - {str(e)}")
self.stats['failed'] += 1
return False
이렇게 구축한 자동화 시스템은 단순히 ChatGPT 요약에만 국한되지 않아요. 슬랙 메시지, 이메일, 회의록 등 다양한 텍스트 데이터를 자동으로 정리하는 데 활용할 수 있어요. 중요한 건 각 API의 특성을 이해하고 에러 처리를 꼼꼼히 하는 거예요. 특히 Notion API는 데이터 구조가 복잡하니 공식 문서를 참고하면서 개발하는 게 좋아요.