Python
[Python] FastAPI로 구현하는 카카오톡 메시지 전송 시스템 - 2
dev-grace
2025. 6. 15. 14:27
이전 포스트에서 "카카오톡 메시지 알림" 테스트 애플리케이션을 구현해보았습니다.
한 단계 더 나아가 친구목록 조회와 개별/단체 메시지 전송이 가능한 카카오톡 API 시스템을 구현해보겠습니다.
⚠️ 테스트용 애플리케이션 안내
해당 예제는 카카오 개발자 센터의 테스트 앱 기반으로 구현됩니다.
팀원 등록된 사용자에게만 메시지 전송이 가능하며, 실제 서비스 운영을 위해서는 비즈 앱 전환이 필요합니다.
목차
1. 개발 환경 설정
2. 카카오 개발자 설정
3. 애플리케이션 화면 및 기능 설명
4. 시스템 엔드포인트 구조
5. 전체 구현 코드
6. 실행 및 테스트 방법
1. 개발 환경 설정
카카오톡 API 시스템을 구현하기 위해 다음 패키지들이 필요합니다.
pip install fastapi uvicorn httpx
- FastAPI: 빠른 API 서버 구축을 위한 프레임워크
- Uvicorn: ASGI 서버로 FastAPI 앱 실행
- HTTPX: 비동기 HTTP 요청을 위한 라이브러리
2. 카카오 개발자 설정
카카오 개발자 센터에서 다음 순서대로 설정을 진행합니다.
2-1. 기본 설정
기본 설정은 이전 포스트에서의 방식과 동일합니다.
- 카카오 개발자 계정 및 애플리케이션 등록
- 웹 플랫폼 등록 (`http://localhost:8000`)
- 카카오 로그인 활성화
- Redirect URI 등록 (`http://localhost:8000/oauth/callback`)
2-2. 권한 설정
"카카오 로그인" > "동의항목" 탭에서 다음 권한들을 설정합니다.
▶ 필수 설정 항목
1. 개인 정보 : 카카오톡 서비스 내 친구 목록 (friends)
- 동의 단계: "이용 중 동의" 선택
- 용도: 친구목록 조회용
2. 접근 권한 : 카카오톡 메시지 전송 (talk_message)
- 동의 단계: "선택 동의" 선택
- 용도: 친구에게 메시지 전송용
2-3. 팀원 등록 (테스트 앱)
▶ 테스트 앱 vs 비즈 앱
- 테스트 앱 (개발용): 팀원만 조회/전송 가능, 개발 테스트 목적
- 비즈 앱 (실서비스용): 실제 사용자 대상 서비스 가능, 사업자 등록 필요
※ 팀원 초대 방법
1. "앱 설정" > "팀 관리" 메뉴 클릭
2. "팀원 초대" 버튼 클릭
3. 초대할 팀원의 이메일 주소 입력
4. 앱 권한 설정 (EDITOR / EDITOR_MESSAGE_TEMPLATE / READER)
5. 팀원은 카카오톡 계정 이메일에서 팀원 초대 메일 확인
※ 핵심 사항
- ✅ 카카오 개발자 센터에 팀원으로 등록 (상태 : 초대 메일 발송됨 > 활성)
- ✅ 팀원은 해당 테스트 앱에 1회 이상 로그인 필요 → 필수 권한에 동의 체크
⚠️ 일반적인 카카오톡 친구는 API에서 조회되지 않습니다.
3. 애플리케이션 화면 및 기능 설명
3-1. 메인 화면 (홈)
▶ 로그인 전 화면
- 내용: 간단한 소개 문구와 카카오 로그인 옵션이 표시됩니다.
- 기능:
- `카카오 로그인`: 일반 로그인 진행
- `강제 재인증`: 계정 선택 화면을 표시하여 다른 계정으로 로그인
- 상태 표시: "연결 필요" 상태로 표시
▶ 로그인 후 화면
- 내용: 로그인 완료 메시지와 함께 주요 기능 메뉴가 표시됩니다.
- 기능:
- `친구목록 조회`: 카카오톡 친구들의 정보와 UUID 확인
- `메시지 전송`: 개별 또는 단체 메시지 전송 페이지로 이동
- `내 정보`: 현재 로그인한 계정의 프로필 정보 확인
- `로그아웃`: 현재 세션에서 안전하게 로그아웃
- 상태 표시: "연결됨" 상태로 표시
3-2. 친구목록 화면
- 친구목록 표시: 총 친구 수와 함께 전체 친구목록이 표시됩니다.
- 친구 정보: 각 친구별로 다음 정보가 표시됩니다.
- 프로필 이미지 (없으면 생략)
- 친구 닉네임
- UUID (개발자용 식별자)
- 검색 기능: 상단 검색창에서 친구 이름이나 UUID로 실시간 검색 가능
- 개별 기능: 각 친구별로 제공되는 기능
- "메시지" 버튼: 해당 친구에게 바로 메시지 전송 (팝업으로 메시지 입력)
- "복사" 버튼: UUID를 클립보드에 복사
3-3. 메시지 전송 화면
- 전송 대상 선택: 드롭다운 메뉴로 전송 방식을 선택할 수 있습니다.
- 모든 친구에게 전송: 친구목록의 모든 사람에게 동시 전송
- 특정 친구 (UUID 입력): 한 명의 친구에게만 개별 전송
- 여러 친구 (UUID 목록): 쉼표로 구분된 여러 UUID에 전송
- UUID 입력 필드: 전송 방식에 따라 조건부로 표시되는 입력 필드
- 메시지 작성: 여러 줄 텍스트 입력이 가능한 메시지 작성 영역
- 전송 기능:
- 전송 전 확인 다이얼로그 표시
- 전송 대상 수 확인
- 전송 버튼 클릭 시 새 창에서 결과 표시
3-4. 전송 결과 화면
- 전송 상태 표시:
- 성공 시: 녹색 체크마크와 "전송 성공!" 메시지
- 실패 시: 빨간색 X마크와 "전송 실패" 메시지
- 전송 정보 상세:
- 전송 타입 (전체/개별/다중)
- 대상 친구 수
- UUID 목록 (다중 전송인 경우)
- 전송된 메시지 내용 (박스 형태로 표시)
- 전송 시간
- 전송 상태 (성공/실패)
- API 응답 상세: 개발자를 위한 전체 API 응답을 JSON 형식으로 표시
3-5. 내 정보 화면
- 기본 정보 표시:
- 사용자 ID
- 닉네임
- 카카오 계정 연결 시간
- 프로필 이미지 유무
- 프로필 이미지: 프로필 이미지가 있는 경우 별도 카드로 표시
- 전체 API 응답: 개발 참고용으로 전체 JSON 응답 데이터를 코드 블록으로 표시
4. 시스템 엔드포인트 구조
4-1. API 엔드포인트 구조
- 웹 페이지 엔드포인트
- `/`: 메인 대시보드 (로그인 상태별 분기 처리)
- `/friends-list`: 친구목록 페이지
- `/send-message`: 메시지 전송 페이지
- `/send-result`: 통합 전송 결과 페이지
- `/user-info`: 사용자 정보 페이지
- JSON API 엔드포인트
- `/api/friends-uuid`: 페이징된 친구목록 JSON 데이터
- `/api/friends-uuid/all`: 전체 친구목록 JSON 데이터
- `/api/send-test-alert`: 전체 친구 대상 테스트 알림
- `/api/debug/user-info`: 디버그용 사용자 정보 조회
5. 전체 구현 코드
아래는 위에서 설명한 화면과 기능을 구현한 코드 `kakao_friends_api.py`입니다.
from fastapi import FastAPI, HTTPException, Query, Form
from fastapi.responses import RedirectResponse, HTMLResponse
import httpx
import json
import uvicorn
from datetime import datetime
from typing import Optional, List, Dict
import urllib.parse
# FastAPI 앱 생성
app = FastAPI(title="카카오톡 친구목록 API", version="2.0.0")
# 카카오 설정값 (실제 키로 변경 필요)
KAKAO_APP_KEY = "YREST_API_KEY" # 실제 REST API 키로 교체
KAKAO_REDIRECT_URI = "http://localhost:8000/oauth/callback"
KAKAO_CHANNEL_ID = "CHANNEL_ID" # 채널 ID (선택사항)
# 토큰 저장용 전역 변수
ACCESS_TOKEN = None
REFRESH_TOKEN = None
# ================================
# CSS 스타일
# ================================
SIMPLE_STYLE = """
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', system-ui, sans-serif;
background: #f8f9fa;
color: #333;
line-height: 1.6;
padding: 20px;
}
.container {
max-width: 800px;
margin: 0 auto;
background: white;
border-radius: 8px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
overflow: hidden;
}
.header {
text-align: center;
padding: 30px 20px;
background: #FEE500;
color: #333;
}
.header h1 {
font-size: 1.8em;
margin-bottom: 8px;
font-weight: 600;
}
.header .subtitle {
font-size: 0.95em;
opacity: 0.8;
}
.status {
display: inline-block;
padding: 4px 12px;
border-radius: 20px;
font-size: 0.85em;
font-weight: 500;
margin-top: 10px;
}
.status.connected {
background: #d4edda;
color: #155724;
}
.status.disconnected {
background: #f8d7da;
color: #721c24;
}
.content {
padding: 30px;
}
.card {
border: 1px solid #e9ecef;
border-radius: 6px;
padding: 20px;
margin-bottom: 20px;
background: white;
}
.card:hover {
border-color: #dee2e6;
}
.card h3 {
font-size: 1.1em;
margin-bottom: 8px;
color: #333;
font-weight: 500;
}
.card p {
color: #666;
font-size: 0.9em;
margin-bottom: 15px;
}
.btn {
display: inline-block;
padding: 8px 16px;
border: none;
border-radius: 4px;
text-decoration: none;
font-size: 0.9em;
font-weight: 500;
cursor: pointer;
transition: all 0.2s ease;
text-align: center;
}
.btn-primary {
background: #007bff;
color: white;
}
.btn-primary:hover {
background: #0056b3;
}
.btn-secondary {
background: #6c757d;
color: white;
}
.btn-secondary:hover {
background: #545b62;
}
.btn-success {
background: #28a745;
color: white;
}
.btn-success:hover {
background: #1e7e34;
}
.btn-warning {
background: #ffc107;
color: #212529;
}
.btn-warning:hover {
background: #e0a800;
}
.btn-danger {
background: #dc3545;
color: white;
}
.btn-danger:hover {
background: #c82333;
}
.btn-block {
display: block;
width: 100%;
}
.actions {
display: flex;
gap: 10px;
flex-wrap: wrap;
margin-top: 20px;
justify-content: center;
}
.form-group {
margin-bottom: 20px;
}
.form-group label {
display: block;
margin-bottom: 6px;
font-weight: 500;
color: #333;
}
.form-group input,
.form-group textarea,
.form-group select {
width: 100%;
padding: 10px;
border: 1px solid #ced4da;
border-radius: 4px;
font-size: 0.9em;
}
.form-group input:focus,
.form-group textarea:focus,
.form-group select:focus {
outline: none;
border-color: #80bdff;
box-shadow: 0 0 0 2px rgba(0,123,255,0.25);
}
.form-group textarea {
height: 100px;
resize: vertical;
}
.friends-list {
max-height: 400px;
overflow-y: auto;
border: 1px solid #e9ecef;
border-radius: 4px;
}
.friend-item {
display: flex;
align-items: center;
padding: 12px 15px;
border-bottom: 1px solid #f8f9fa;
}
.friend-item:last-child {
border-bottom: none;
}
.friend-item:hover {
background: #f8f9fa;
}
.friend-avatar {
width: 40px;
height: 40px;
border-radius: 50%;
margin-right: 12px;
object-fit: cover;
}
.friend-info {
flex: 1;
}
.friend-name {
font-weight: 500;
margin-bottom: 2px;
}
.friend-uuid {
font-size: 0.8em;
color: #666;
font-family: monospace;
}
.friend-actions {
display: flex;
gap: 5px;
}
.search-box {
margin-bottom: 20px;
}
.search-box input {
width: 100%;
padding: 10px;
border: 1px solid #ced4da;
border-radius: 4px;
font-size: 0.9em;
}
.back-link {
display: inline-block;
margin-bottom: 20px;
color: #007bff;
text-decoration: none;
font-size: 0.9em;
}
.back-link:hover {
text-decoration: underline;
}
.result-box {
padding: 20px;
border-radius: 4px;
margin-bottom: 20px;
text-align: center;
}
.result-success {
background: #d4edda;
border: 1px solid #c3e6cb;
color: #155724;
}
.result-error {
background: #f8d7da;
border: 1px solid #f5c6cb;
color: #721c24;
}
.json-output {
background: #f8f9fa;
border: 1px solid #e9ecef;
border-radius: 4px;
padding: 15px;
font-family: monospace;
font-size: 0.85em;
white-space: pre-wrap;
max-height: 300px;
overflow-y: auto;
}
@media (max-width: 600px) {
.container {
margin: 0;
border-radius: 0;
}
.actions {
flex-direction: column;
}
.btn {
width: 100%;
}
.friend-item {
flex-direction: column;
text-align: center;
padding: 15px;
}
.friend-avatar {
margin: 0 0 10px 0;
}
.friend-actions {
margin-top: 10px;
justify-content: center;
}
}
</style>
"""
# ================================
# 유틸리티 함수
# ================================
def check_access_token() -> str:
"""액세스 토큰 확인"""
if not ACCESS_TOKEN:
raise HTTPException(
status_code=401,
detail="인증이 필요합니다. 카카오 로그인을 진행하세요.",
)
return ACCESS_TOKEN
def create_auth_url(force_login: bool = False) -> str:
"""카카오 인증 URL 생성"""
base_url = "https://kauth.kakao.com/oauth/authorize"
params = {
"client_id": KAKAO_APP_KEY,
"redirect_uri": KAKAO_REDIRECT_URI,
"response_type": "code",
"scope": "talk_message friends",
}
if force_login:
params["prompt"] = "login"
return f"{base_url}?{urllib.parse.urlencode(params)}"
# ================================
# 메인 화면
# ================================
@app.get("/", response_class=HTMLResponse)
async def root():
"""메인 대시보드"""
global ACCESS_TOKEN
if ACCESS_TOKEN:
# 로그인된 상태의 대시보드
return f"""
<!DOCTYPE html>
<html>
<head>
<title>카카오톡 API</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>카카오톡 API</h1>
<p class="subtitle">친구목록 조회 및 메시지 전송</p>
<div class="status connected">연결됨</div>
</div>
<div class="content">
<div class="card">
<h3>친구목록 조회</h3>
<p>카카오톡 친구들의 정보를 확인하고 UUID를 조회할 수 있습니다</p>
<a href="/friends-list" class="btn btn-primary">친구목록 보기</a>
</div>
<div class="card">
<h3>메시지 전송</h3>
<p>친구들에게 개별 또는 단체 메시지를 전송할 수 있습니다</p>
<a href="/send-message" class="btn btn-primary">메시지 보내기</a>
</div>
<div class="card">
<h3>내 정보</h3>
<p>현재 로그인한 카카오 계정의 프로필 정보를 확인합니다</p>
<a href="/user-info" class="btn btn-primary">내 정보 보기</a>
</div>
<div class="actions">
<a href="/logout" class="btn btn-secondary">로그아웃</a>
</div>
</div>
</div>
</body>
</html>
"""
else:
# 로그인 화면
return f"""
<!DOCTYPE html>
<html>
<head>
<title>카카오톡 API - 로그인</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>카카오톡 API</h1>
<p class="subtitle">친구목록 조회 및 메시지 전송 서비스</p>
<div class="status disconnected">연결 필요</div>
</div>
<div class="content">
<div class="card">
<h3>카카오 로그인</h3>
<p>카카오 계정으로 로그인하여 API 기능을 사용할 수 있습니다</p>
<a href="/auth/login" class="btn btn-primary btn-block">로그인</a>
</div>
<div class="card">
<h3>강제 재인증</h3>
<p>계정 선택 화면을 표시하여 다른 계정으로 로그인할 수 있습니다</p>
<a href="/auth/login/force" class="btn btn-secondary btn-block">강제 로그인</a>
</div>
</div>
</div>
</body>
</html>
"""
# ================================
# OAuth 인증
# ================================
@app.get("/auth/login")
async def kakao_login():
"""카카오 로그인 페이지로 리디렉션"""
auth_url = create_auth_url(force_login=False)
return RedirectResponse(url=auth_url)
@app.get("/auth/login/force")
async def kakao_login_force():
"""강제 재인증 로그인"""
auth_url = create_auth_url(force_login=True)
return RedirectResponse(url=auth_url)
@app.get("/oauth/callback")
async def oauth_callback(code: str = Query(...)):
"""카카오 OAuth 인증 콜백"""
global ACCESS_TOKEN, REFRESH_TOKEN
token_url = "https://kauth.kakao.com/oauth/token"
data = {
"grant_type": "authorization_code",
"client_id": KAKAO_APP_KEY,
"redirect_uri": KAKAO_REDIRECT_URI,
"code": code,
}
async with httpx.AsyncClient() as client:
try:
response = await client.post(token_url, data=data)
if response.status_code == 200:
token_info = response.json()
ACCESS_TOKEN = token_info.get("access_token")
REFRESH_TOKEN = token_info.get("refresh_token")
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>인증 성공</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>인증 성공</h1>
<p class="subtitle">카카오 로그인이 완료되었습니다</p>
</div>
<div class="content">
<div class="result-box result-success">
<h3>로그인 완료</h3>
<p>이제 모든 API 기능을 사용할 수 있습니다</p>
<a href="/" class="btn btn-primary">대시보드로 이동</a>
</div>
</div>
</div>
<script>
setTimeout(function() {{
window.location.href = '/';
}}, 3000);
</script>
</body>
</html>
"""
)
else:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>인증 실패</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>인증 실패</h1>
<p class="subtitle">로그인 과정에서 오류가 발생했습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<h3>토큰 발급 실패</h3>
<p>{response.text}</p>
<a href="/" class="btn btn-primary">다시 시도</a>
</div>
</div>
</div>
</body>
</html>
"""
)
except Exception as e:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>인증 오류</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>인증 오류</h1>
<p class="subtitle">시스템 오류가 발생했습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<h3>시스템 오류</h3>
<p>{str(e)}</p>
<a href="/" class="btn btn-primary">홈으로 돌아가기</a>
</div>
</div>
</div>
</body>
</html>
"""
)
@app.get("/logout")
async def logout():
"""로그아웃"""
global ACCESS_TOKEN, REFRESH_TOKEN
ACCESS_TOKEN = None
REFRESH_TOKEN = None
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>로그아웃 완료</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>로그아웃 완료</h1>
<p class="subtitle">안전하게 로그아웃되었습니다</p>
</div>
<div class="content">
<div class="result-box result-success">
<h3>세션 종료</h3>
<p>다시 사용하려면 로그인해 주세요</p>
<a href="/" class="btn btn-primary">홈으로 이동</a>
</div>
</div>
</div>
<script>
setTimeout(function() {{
window.location.href = '/';
}}, 2000);
</script>
</body>
</html>
"""
)
# ================================
# 기존 API 함수들 (변경 없음)
# ================================
async def get_friends_list(limit: int = 100, offset: int = 0) -> Dict:
"""카카오톡 친구목록을 조회하는 함수"""
access_token = check_access_token()
url = "https://kapi.kakao.com/v1/api/talk/friends"
headers = {"Authorization": f"Bearer {access_token}"}
params = {"limit": min(limit, 100), "offset": offset}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, params=params)
return response.json()
except Exception as e:
return {"error": f"친구목록 조회 중 오류 발생: {e}"}
async def get_all_friends() -> List[Dict]:
"""모든 친구목록을 조회하는 함수"""
all_friends = []
offset = 0
limit = 100
while True:
result = await get_friends_list(limit=limit, offset=offset)
if "error" in result:
return [{"error": result["error"]}]
if "elements" in result and result["elements"]:
all_friends.extend(result["elements"])
if len(result["elements"]) < limit:
break
offset += limit
else:
break
return all_friends
async def send_message_to_friends(
message: str, receiver_uuids: Optional[List[str]] = None
) -> Dict:
"""카카오톡 메시지 전송"""
access_token = check_access_token()
url = "https://kapi.kakao.com/v1/api/talk/friends/message/default/send"
headers = {"Authorization": f"Bearer {access_token}"}
template_object = {
"object_type": "text",
"text": message,
"link": {
"web_url": "https://developers.kakao.com",
"mobile_web_url": "https://developers.kakao.com",
},
}
data = {"template_object": json.dumps(template_object)}
# 개별 전송 처리 방식 수정
if receiver_uuids and len(receiver_uuids) == 1:
data["receiver_uuids"] = f'["{receiver_uuids[0]}"]'
elif receiver_uuids and len(receiver_uuids) > 1:
data["receiver_uuids"] = json.dumps(receiver_uuids)
async with httpx.AsyncClient() as client:
try:
response = await client.post(url, headers=headers, data=data)
result = response.json()
result["status_code"] = response.status_code
return result
except Exception as e:
return {"error": f"메시지 전송 중 오류 발생: {e}", "status_code": 500}
# ================================
# 웹 UI 페이지
# ================================
@app.get("/friends-list", response_class=HTMLResponse)
async def friends_list_page():
"""친구목록 웹 페이지"""
try:
friends = await get_all_friends()
if friends and "error" in friends[0]:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>친구목록 오류</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>오류 발생</h1>
<p class="subtitle">{friends[0]['error']}</p>
</div>
<div class="content">
<a href="/" class="btn btn-primary">홈으로 돌아가기</a>
</div>
</div>
</body>
</html>
"""
)
friends_html = ""
for friend in friends:
friends_html += f"""
<div class="friend-item">
<img src="{friend.get('profile_thumbnail_image', '/static/default-avatar.png')}"
alt="프로필" class="friend-avatar" loading="lazy"
onerror="this.style.display='none'">
<div class="friend-info">
<div class="friend-name">{friend.get('profile_nickname', '이름 없음')}</div>
<div class="friend-uuid">{friend.get('uuid', 'N/A')}</div>
</div>
<div class="friend-actions">
<button onclick="sendMessage('{friend.get('uuid')}')" class="btn btn-primary">메시지</button>
<button onclick="copyUUID('{friend.get('uuid')}')" class="btn btn-secondary">복사</button>
</div>
</div>
"""
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>친구목록</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>친구목록</h1>
<p class="subtitle">총 {len(friends)}명의 친구</p>
</div>
<div class="content">
<a href="/" class="back-link">← 대시보드로</a>
<div class="search-box">
<input type="text" id="searchInput" placeholder="친구 검색..."
onkeyup="searchFriends()">
</div>
<div class="friends-list" id="friendsList">
{friends_html}
</div>
</div>
</div>
<script>
function sendMessage(uuid) {{
const message = prompt('보낼 메시지를 입력하세요:', '안녕하세요!');
if (message && message.trim()) {{
const url = `/send-result?uuid=${{encodeURIComponent(uuid)}}&message=${{encodeURIComponent(message)}}`;
window.open(url, '_blank');
}}
}}
function copyUUID(uuid) {{
navigator.clipboard.writeText(uuid).then(function() {{
alert('UUID가 복사되었습니다!');
}}).catch(function() {{
alert('UUID 복사에 실패했습니다: ' + uuid);
}});
}}
function searchFriends() {{
const input = document.getElementById('searchInput');
const filter = input.value.toLowerCase();
const friendsList = document.getElementById('friendsList');
const friendItems = friendsList.getElementsByClassName('friend-item');
for (let i = 0; i < friendItems.length; i++) {{
const friendInfo = friendItems[i].getElementsByClassName('friend-info')[0];
const nickname = friendInfo.querySelector('.friend-name').textContent.toLowerCase();
const uuid = friendInfo.querySelector('.friend-uuid').textContent.toLowerCase();
if (nickname.includes(filter) || uuid.includes(filter)) {{
friendItems[i].style.display = '';
}} else {{
friendItems[i].style.display = 'none';
}}
}}
}}
</script>
</body>
</html>
"""
)
except Exception as e:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>시스템 오류</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>시스템 오류</h1>
<p class="subtitle">친구목록을 불러오는 중 오류가 발생했습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<p>{str(e)}</p>
<a href="/" class="btn btn-primary">대시보드로 돌아가기</a>
</div>
</div>
</div>
</body>
</html>
"""
)
@app.get("/send-message", response_class=HTMLResponse)
async def send_message_page():
"""메시지 전송 페이지"""
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>메시지 전송</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>메시지 전송</h1>
<p class="subtitle">친구들에게 카카오톡 메시지를 전송하세요</p>
</div>
<div class="content">
<a href="/" class="back-link">← 대시보드로</a>
<form id="messageForm">
<div class="form-group">
<label for="messageType">전송 대상</label>
<select id="messageType" name="messageType" onchange="toggleUuidInput()">
<option value="all">모든 친구에게 전송</option>
<option value="individual">특정 친구 (UUID 입력)</option>
<option value="multiple">여러 친구 (UUID 목록)</option>
</select>
</div>
<div class="form-group" id="uuidGroup" style="display: none;">
<label for="uuids">UUID 입력</label>
<input type="text" id="uuids" name="uuids"
placeholder="여러 개는 쉼표로 구분: uuid1, uuid2, uuid3">
</div>
<div class="form-group">
<label for="message">메시지 내용</label>
<textarea id="message" name="message"
placeholder="전송할 메시지를 입력하세요..."></textarea>
</div>
<div class="actions">
<button type="button" onclick="sendMessage()" class="btn btn-primary">메시지 전송</button>
<a href="/friends-list" class="btn btn-success">친구목록 보기</a>
</div>
</form>
</div>
</div>
<script>
function toggleUuidInput() {{
const messageType = document.getElementById('messageType').value;
const uuidGroup = document.getElementById('uuidGroup');
if (messageType === 'individual' || messageType === 'multiple') {{
uuidGroup.style.display = 'block';
}} else {{
uuidGroup.style.display = 'none';
}}
}}
function sendMessage() {{
const messageType = document.getElementById('messageType').value;
const message = document.getElementById('message').value;
const uuids = document.getElementById('uuids').value;
if (!message.trim()) {{
alert('메시지 내용을 입력해주세요.');
return;
}}
let url = '';
let confirmMessage = '';
if (messageType === 'all') {{
url = `/send-result?send_all=true&message=${{encodeURIComponent(message)}}`;
confirmMessage = '모든 친구에게 메시지를 전송하시겠습니까?';
}} else if (messageType === 'individual') {{
if (!uuids.trim()) {{
alert('UUID를 입력해주세요.');
return;
}}
url = `/send-result?uuid=${{encodeURIComponent(uuids.trim())}}&message=${{encodeURIComponent(message)}}`;
confirmMessage = '선택한 친구에게 메시지를 전송하시겠습니까?';
}} else if (messageType === 'multiple') {{
if (!uuids.trim()) {{
alert('UUID 목록을 입력해주세요.');
return;
}}
const uuidList = uuids.split(',').map(u => u.trim()).filter(u => u);
if (uuidList.length === 0) {{
alert('유효한 UUID를 입력해주세요.');
return;
}}
const uuidParams = uuidList.map(u => `uuids=${{encodeURIComponent(u)}}`).join('&');
url = `/send-result?${{uuidParams}}&message=${{encodeURIComponent(message)}}`;
confirmMessage = `${{uuidList.length}}명의 친구에게 메시지를 전송하시겠습니까?`;
}}
if (confirm(confirmMessage)) {{
window.open(url, '_blank');
}}
}}
// 기본 메시지 설정
document.addEventListener('DOMContentLoaded', function() {{
const messageInput = document.getElementById('message');
if (!messageInput.value) {{
messageInput.value = '안녕하세요!\\n\\n카카오톡 API 테스트 메시지입니다.\\n좋은 하루 되세요!';
}}
}});
</script>
</body>
</html>
"""
)
@app.get("/user-info", response_class=HTMLResponse)
async def user_info_page():
"""사용자 정보 페이지"""
try:
access_token = check_access_token()
url = "https://kapi.kakao.com/v2/user/me"
headers = {"Authorization": f"Bearer {access_token}"}
async with httpx.AsyncClient() as client:
response = await client.get(url, headers=headers)
if response.status_code == 200:
user_info = response.json()
# 사용자 정보 파싱
user_id = user_info.get("id", "N/A")
connected_at = user_info.get("connected_at", "N/A")
properties = user_info.get("properties", {})
nickname = properties.get("nickname", "N/A")
profile_image = properties.get("profile_image", "")
thumbnail_image = properties.get("thumbnail_image", "")
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>내 정보</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>내 카카오 정보</h1>
<p class="subtitle">{nickname}</p>
</div>
<div class="content">
<a href="/" class="back-link">← 대시보드로</a>
<div class="card">
<h3>기본 정보</h3>
<p><strong>사용자 ID:</strong> {user_id}</p>
<p><strong>닉네임:</strong> {nickname}</p>
<p><strong>연결 시간:</strong> {connected_at}</p>
{"<p><strong>프로필 이미지:</strong> 있음</p>" if thumbnail_image else "<p><strong>프로필 이미지:</strong> 없음</p>"}
</div>
{"<div class='card'><img src='" + thumbnail_image + "' alt='프로필' style='max-width: 200px; border-radius: 8px;'></div>" if thumbnail_image else ""}
<div class="card">
<h3>전체 API 응답</h3>
<div class="json-output">{json.dumps(user_info, indent=2, ensure_ascii=False)}</div>
</div>
</div>
</div>
</body>
</html>
"""
)
else:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>정보 조회 실패</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>정보 조회 실패</h1>
<p class="subtitle">사용자 정보를 가져올 수 없습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<p>HTTP {response.status_code}: {response.text}</p>
<a href="/" class="btn btn-primary">대시보드로 돌아가기</a>
</div>
</div>
</div>
</body>
</html>
"""
)
except HTTPException as e:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>인증 필요</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>인증 필요</h1>
<p class="subtitle">이 기능을 사용하려면 로그인이 필요합니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<p>{e.detail}</p>
<a href="/auth/login" class="btn btn-primary">카카오 로그인</a>
</div>
</div>
</div>
</body>
</html>
"""
)
except Exception as e:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>시스템 오류</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>시스템 오류</h1>
<p class="subtitle">사용자 정보 조회 중 오류가 발생했습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<p>{str(e)}</p>
<a href="/" class="btn btn-primary">대시보드로 돌아가기</a>
</div>
</div>
</div>
</body>
</html>
"""
)
@app.get("/send-result", response_class=HTMLResponse)
async def send_message_result(
message: str = Query(...),
uuid: Optional[str] = Query(None),
uuids: Optional[List[str]] = Query(None),
send_all: bool = Query(False),
):
"""통합 메시지 전송 결과 페이지"""
try:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
full_message = f"{message}\n\n발송 시간: {current_time}"
# 전송 타입과 대상 결정
if send_all:
# 모든 친구에게 전송
friends = await get_all_friends()
if friends and "error" not in friends[0]:
target_uuids = [
friend.get("uuid") for friend in friends if friend.get("uuid")
]
result = await send_message_to_friends(
full_message, receiver_uuids=target_uuids
)
send_type = "전체"
target_info = f"모든 친구 ({len(target_uuids)}명)"
target_display = f"{len(target_uuids)}명의 모든 친구"
else:
result = {"error": "친구목록 조회 실패", "status_code": 500}
send_type = "전체"
target_info = "친구목록 조회 실패"
target_display = "알 수 없음"
elif uuid:
# 개별 친구에게 전송
result = await send_message_to_friends(full_message, receiver_uuids=[uuid])
send_type = "개별"
target_info = f"UUID: {uuid}"
target_display = "1명의 친구"
elif uuids:
# 여러 친구에게 전송
result = await send_message_to_friends(full_message, receiver_uuids=uuids)
send_type = "다중"
target_info = f"{len(uuids)}개의 UUID"
target_display = f"{len(uuids)}명의 친구"
else:
# 잘못된 요청
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>잘못된 요청</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>잘못된 요청</h1>
<p class="subtitle">전송 대상이 지정되지 않았습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<p>전송 대상을 지정해주세요.</p>
<a href="/send-message" class="btn btn-primary">메시지 전송으로</a>
</div>
</div>
</div>
</body>
</html>
"""
)
success = result.get("status_code") == 200
# UUID 목록 표시용 (다중 전송일 때만)
uuid_list_html = ""
if uuids and len(uuids) > 1:
uuid_list_html = f"""
<p><strong>UUID 목록:</strong></p>
<div style="font-family: monospace; font-size: 0.8em; background: #f8f9fa; padding: 10px; border-radius: 4px; margin: 10px 0; word-break: break-all;">
{', '.join(uuids)}
</div>
"""
elif uuid:
uuid_list_html = f"""
<p><strong>대상 UUID:</strong> <span style="font-family: monospace; background: #f8f9fa; padding: 2px 6px; border-radius: 3px;">{uuid}</span></p>
"""
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>메시지 전송 결과</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>메시지 전송 결과</h1>
<p class="subtitle">{send_type} 전송 - {'전송 성공' if success else '전송 실패'}</p>
</div>
<div class="content">
<a href="/send-message" class="back-link">← 메시지 전송으로</a>
<div class="result-box {'result-success' if success else 'result-error'}">
<h3>{'✅ 전송 성공!' if success else '❌ 전송 실패'}</h3>
<p>{f'{target_display}에게 메시지가 성공적으로 전송되었습니다.' if success else '메시지 전송 중 오류가 발생했습니다.'}</p>
</div>
<div class="card">
<h3>전송 정보</h3>
<p><strong>전송 타입:</strong> {send_type} 전송</p>
<p><strong>대상:</strong> {target_display}</p>
{uuid_list_html}
<p><strong>메시지:</strong></p>
<div style="background: #f8f9fa; padding: 15px; border-radius: 4px; margin: 10px 0; white-space: pre-wrap; border-left: 4px solid #007bff;">
{message}
</div>
<p><strong>전송 시간:</strong> {current_time}</p>
<p><strong>상태:</strong> <span style="color: {'#28a745' if success else '#dc3545'}; font-weight: 500;">{'성공' if success else '실패'}</span></p>
</div>
<div class="card">
<h3>API 응답 상세</h3>
<div class="json-output">{json.dumps(result, indent=2, ensure_ascii=False)}</div>
</div>
<div class="actions">
<a href="/send-message" class="btn btn-primary">다른 메시지 작성</a>
<a href="/friends-list" class="btn btn-success">친구목록으로</a>
<a href="/" class="btn btn-secondary">대시보드로</a>
</div>
</div>
</div>
</body>
</html>
"""
)
except Exception as e:
return HTMLResponse(
content=f"""
<!DOCTYPE html>
<html>
<head>
<title>전송 오류</title>
<meta charset="UTF-8">
{SIMPLE_STYLE}
</head>
<body>
<div class="container">
<div class="header">
<h1>전송 오류</h1>
<p class="subtitle">메시지 전송 중 시스템 오류가 발생했습니다</p>
</div>
<div class="content">
<div class="result-box result-error">
<h3>시스템 오류</h3>
<p>{str(e)}</p>
<a href="/send-message" class="btn btn-primary">다시 시도</a>
</div>
</div>
</div>
</body>
</html>
"""
)
# ================================
# API 엔드포인트들 (기존 로직 유지)
# ================================
@app.get("/api/friends-uuid")
async def friends_uuid_api(limit: int = 10, offset: int = 0):
"""친구목록 UUID 조회 API"""
result = await get_friends_list(limit=limit, offset=offset)
if "error" in result:
return {"status": "오류", "error": result["error"]}
friends_info = []
if "elements" in result and result["elements"]:
for friend in result["elements"]:
friends_info.append(
{
"uuid": friend.get("uuid"),
"id": friend.get("id"),
"profile_nickname": friend.get("profile_nickname"),
"profile_thumbnail_image": friend.get("profile_thumbnail_image"),
"favorite": friend.get("favorite", False),
}
)
return {
"status": "성공",
"total_count": result.get("total_count", 0),
"friends": friends_info,
"before_url": result.get("before_url"),
"after_url": result.get("after_url"),
}
@app.get("/api/friends-uuid/all")
async def all_friends_uuid_api():
"""모든 친구목록 UUID 조회 API"""
friends = await get_all_friends()
if friends and "error" in friends[0]:
return {"status": "오류", "error": friends[0]["error"]}
friends_info = []
for friend in friends:
friends_info.append(
{
"uuid": friend.get("uuid"),
"id": friend.get("id"),
"profile_nickname": friend.get("profile_nickname"),
"profile_thumbnail_image": friend.get("profile_thumbnail_image"),
"favorite": friend.get("favorite", False),
}
)
return {"status": "성공", "total_count": len(friends_info), "friends": friends_info}
@app.get("/api/send-test-alert")
async def send_test_alert():
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
message = f"⚠️ 테스트 알림 ⚠️\n\n안녕하세요! 이것은 테스트 알림입니다.\n발송 시간: {current_time}"
# 모든 친구 UUID 조회
friends = await get_all_friends()
if friends and "error" not in friends[0]:
friend_uuids = [friend.get("uuid") for friend in friends if friend.get("uuid")]
result = await send_message_to_friends(message, receiver_uuids=friend_uuids)
else:
result = {"error": "친구목록 조회 실패"}
return {"status": "요청 완료", "result": result}
@app.get("/api/send-to-uuid/{uuid}")
async def send_to_specific_uuid(
uuid: str, message: str = Query(default="안녕하세요! 개별 메시지입니다.")
):
"""특정 UUID에게 메시지 전송 API"""
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
full_message = f"{message}\n\n발송 시간: {current_time}"
result = await send_message_to_friends(full_message, receiver_uuids=[uuid])
return {
"status": "요청 완료",
"target_uuid": uuid,
"message": full_message,
"result": result,
}
@app.get("/api/send-to-multiple-uuids")
async def send_to_multiple_uuids(
uuids: List[str] = Query(...),
message: str = Query("안녕하세요! 단체 메시지입니다."),
):
"""여러 UUID에게 메시지 전송 API"""
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
full_message = f"{message}\n\n발송 시간: {current_time}"
result = await send_message_to_friends(full_message, receiver_uuids=uuids)
return {
"status": "요청 완료",
"target_uuids": uuids,
"message": full_message,
"result": result,
}
# ================================
# 디버그 API
# ================================
@app.get("/api/debug/user-info")
async def debug_user_info():
"""현재 로그인한 사용자 정보 조회"""
access_token = check_access_token()
url = "https://kapi.kakao.com/v2/user/me"
headers = {"Authorization": f"Bearer {access_token}"}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers)
if response.status_code == 200:
user_info = response.json()
return {
"status": "성공",
"user_id": user_info.get("id"),
"connected_at": user_info.get("connected_at"),
"properties": user_info.get("properties", {}),
"kakao_account": user_info.get("kakao_account", {}),
}
else:
return {"status": "오류", "error": response.text}
except Exception as e:
return {"status": "오류", "error": f"사용자 정보 조회 중 오류: {e}"}
@app.get("/api/debug/config")
async def debug_config():
"""설정값 확인"""
global ACCESS_TOKEN
return {
"app_key_set": KAKAO_APP_KEY != "YOUR_REST_API_KEY_HERE",
"app_key_length": len(KAKAO_APP_KEY) if KAKAO_APP_KEY else 0,
"redirect_uri": KAKAO_REDIRECT_URI,
"has_access_token": ACCESS_TOKEN is not None,
"access_token_preview": ACCESS_TOKEN[:20] + "..." if ACCESS_TOKEN else None,
}
# ================================
# 기타 엔드포인트
# ================================
@app.get("/favicon.ico")
async def favicon():
"""Favicon 요청 처리"""
return {"message": "No favicon"}
# ================================
# 서버 실행
# ================================
if __name__ == "__main__":
print("📌 사용 순서:")
print(" 1. KAKAO_APP_KEY를 실제 REST API 키로 교체")
print(" 2. http://localhost:8000/ 접속")
print(" 3. '로그인' 버튼 클릭")
print(" 4. 인증 완료 후 기능 사용")
print("📖 API 문서: http://localhost:8000/docs")
uvicorn.run(app, host="0.0.0.0", port=8000, reload=False)
6. 실행 및 테스트 방법
6-1. 코드 실행 준비
- 위 코드를 `kakao_friends_api.py` 파일로 저장합니다.
- KAKAO_APP_KEY를 실제 REST API 키로 변경합니다.
- 터미널에서 다음 명령어를 실행합니다.
python kakao_friends_api.py
6-2. 테스트 순서
- 브라우저에서 http://localhost:8000 접속
- "카카오 로그인" 버튼 클릭하여 인증 진행 *메시지 전송과 친구 목록 접근 권한 동의 필요
- 로그인 완료 후 기능 사용
6-3. API 테스트
- /docs 접속하여 Swagger UI에서 각 API 엔드포인트 테스트
- JSON 형식의 친구목록 데이터 확인
- 디버그 정보로 설정 상태 점검
6-4. 실제 테스트 결과 화면
다양한 서비스에 알림 기능을 추가하고 싶을 때 해당 코드를 기반으로 확장하여 사용할 수 있습니다.
참고 자료