377 lines
15 KiB
Python
377 lines
15 KiB
Python
"""
|
|
Facebook OAuth 2.0 API 클라이언트
|
|
|
|
Facebook Graph API를 통한 OAuth 2.0 인증 흐름을 처리하는 클라이언트입니다.
|
|
|
|
인증 흐름:
|
|
1. get_authorization_url()로 Facebook 로그인 페이지 URL 획득
|
|
2. 사용자가 Facebook에서 로그인 후 인가 코드(code) 발급
|
|
3. get_access_token()으로 인가 코드를 단기 액세스 토큰으로 교환
|
|
4. exchange_long_lived_token()으로 단기 토큰을 장기 토큰(약 60일)으로 교환
|
|
5. get_user_info()로 사용자 정보 조회
|
|
6. get_user_pages()로 관리 페이지 목록 조회 (선택)
|
|
|
|
Example:
|
|
```python
|
|
client = FacebookOAuthClient()
|
|
auth_url = client.get_authorization_url(state="csrf_token")
|
|
token_data = await client.get_access_token(code="auth_code")
|
|
long_token = await client.exchange_long_lived_token(token_data["access_token"])
|
|
user_info = await client.get_user_info(long_token["access_token"])
|
|
```
|
|
"""
|
|
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.utils.logger import get_logger
|
|
from config import facebook_settings
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# =============================================================================
|
|
# Facebook OAuth 예외 클래스 정의
|
|
# =============================================================================
|
|
class FacebookOAuthException(HTTPException):
|
|
"""Facebook OAuth 관련 기본 예외"""
|
|
|
|
def __init__(self, status_code: int, code: str, message: str):
|
|
super().__init__(status_code=status_code, detail={"code": code, "message": message})
|
|
|
|
|
|
class FacebookAuthFailedError(FacebookOAuthException):
|
|
"""Facebook 인증 실패"""
|
|
|
|
def __init__(self, message: str = "Facebook 인증에 실패했습니다."):
|
|
super().__init__(status.HTTP_400_BAD_REQUEST, "FACEBOOK_AUTH_FAILED", message)
|
|
|
|
|
|
class FacebookAPIError(FacebookOAuthException):
|
|
"""Facebook API 호출 오류"""
|
|
|
|
def __init__(self, message: str = "Facebook API 호출 중 오류가 발생했습니다."):
|
|
super().__init__(status.HTTP_500_INTERNAL_SERVER_ERROR, "FACEBOOK_API_ERROR", message)
|
|
|
|
|
|
class FacebookTokenExpiredError(FacebookOAuthException):
|
|
"""Facebook 토큰 만료"""
|
|
|
|
def __init__(self, message: str = "Facebook 토큰이 만료되었습니다. 재연동이 필요합니다."):
|
|
super().__init__(status.HTTP_401_UNAUTHORIZED, "FACEBOOK_TOKEN_EXPIRED", message)
|
|
|
|
|
|
class FacebookOAuthClient:
|
|
"""
|
|
Facebook OAuth 2.0 API 클라이언트
|
|
|
|
Facebook Graph API를 통한 OAuth 인증 흐름을 처리합니다.
|
|
모든 설정값은 config.py의 FacebookSettings에서 로드됩니다.
|
|
|
|
인증 흐름:
|
|
1. get_authorization_url() → Facebook 로그인 페이지 URL 생성
|
|
2. get_access_token() → 인가 코드를 단기 토큰으로 교환
|
|
3. exchange_long_lived_token() → 단기 토큰을 장기 토큰(~60일)으로 교환
|
|
4. get_user_info() → 사용자 프로필 조회
|
|
5. get_user_pages() → 관리 페이지 목록 조회
|
|
"""
|
|
|
|
# Facebook OAuth/Graph API URL 템플릿
|
|
AUTH_URL_TEMPLATE = "https://www.facebook.com/{version}/dialog/oauth"
|
|
TOKEN_URL_TEMPLATE = "https://graph.facebook.com/{version}/oauth/access_token"
|
|
USER_INFO_URL_TEMPLATE = "https://graph.facebook.com/{version}/me"
|
|
PAGES_URL_TEMPLATE = "https://graph.facebook.com/{version}/{user_id}/accounts"
|
|
|
|
def __init__(self) -> None:
|
|
# FacebookSettings에서 설정값 로드
|
|
self.client_id = facebook_settings.FACEBOOK_APP_ID
|
|
self.client_secret = facebook_settings.FACEBOOK_APP_SECRET
|
|
self.redirect_uri = facebook_settings.FACEBOOK_REDIRECT_URI
|
|
self.api_version = facebook_settings.FACEBOOK_GRAPH_API_VERSION
|
|
self.scope = facebook_settings.FACEBOOK_OAUTH_SCOPE
|
|
|
|
logger.debug(
|
|
f"[FACEBOOK] OAuth 클라이언트 초기화 - "
|
|
f"api_version: {self.api_version}, redirect_uri: {self.redirect_uri}"
|
|
)
|
|
|
|
def get_authorization_url(self, state: str) -> str:
|
|
"""
|
|
Facebook 로그인 페이지 URL 생성
|
|
|
|
Args:
|
|
state: CSRF 방지용 state 토큰
|
|
|
|
Returns:
|
|
Facebook OAuth 인증 페이지 URL
|
|
"""
|
|
# Facebook 인증 페이지 URL 조합
|
|
base_url = self.AUTH_URL_TEMPLATE.format(version=self.api_version)
|
|
params = {
|
|
"client_id": self.client_id,
|
|
"redirect_uri": self.redirect_uri,
|
|
"state": state,
|
|
"scope": self.scope,
|
|
"response_type": "code",
|
|
}
|
|
auth_url = f"{base_url}?{urlencode(params)}"
|
|
|
|
logger.info(f"[FACEBOOK] 인증 URL 생성 - redirect_uri: {self.redirect_uri}")
|
|
logger.debug(f"[FACEBOOK] 인증 URL 상세 - state: {state[:20]}..., scope: {self.scope}")
|
|
|
|
return auth_url
|
|
|
|
async def get_access_token(self, code: str) -> dict:
|
|
"""
|
|
인가 코드를 단기 액세스 토큰으로 교환
|
|
|
|
Args:
|
|
code: Facebook 로그인 후 발급받은 인가 코드
|
|
|
|
Returns:
|
|
dict: {access_token, token_type, expires_in}
|
|
|
|
Raises:
|
|
FacebookAuthFailedError: 토큰 발급 실패 시
|
|
FacebookAPIError: API 호출 오류 시
|
|
"""
|
|
logger.info(f"[FACEBOOK] 액세스 토큰 요청 시작 - code: {code[:20]}...")
|
|
|
|
# Facebook Graph API - 인가 코드를 액세스 토큰으로 교환
|
|
token_url = self.TOKEN_URL_TEMPLATE.format(version=self.api_version)
|
|
params = {
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"code": code,
|
|
"redirect_uri": self.redirect_uri,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
logger.debug(f"[FACEBOOK] 토큰 요청 URL: {token_url}")
|
|
response = await client.get(token_url, params=params)
|
|
result = response.json()
|
|
|
|
logger.debug(f"[FACEBOOK] 토큰 응답 상태 - status: {response.status_code}")
|
|
|
|
# 에러 응답 처리
|
|
if "error" in result:
|
|
error_msg = result.get("error", {})
|
|
error_message = error_msg.get("message", "알 수 없는 오류")
|
|
logger.error(
|
|
f"[FACEBOOK] 토큰 발급 실패 - "
|
|
f"type: {error_msg.get('type')}, message: {error_message}"
|
|
)
|
|
raise FacebookAuthFailedError(f"Facebook 토큰 발급 실패: {error_message}")
|
|
|
|
logger.info("[FACEBOOK] 단기 액세스 토큰 발급 성공")
|
|
logger.debug(
|
|
f"[FACEBOOK] 토큰 정보 - "
|
|
f"token_type: {result.get('token_type')}, "
|
|
f"expires_in: {result.get('expires_in')}"
|
|
)
|
|
|
|
return result
|
|
|
|
except FacebookAuthFailedError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[FACEBOOK] API 호출 오류 - error: {str(e)}")
|
|
raise FacebookAPIError(f"Facebook API 호출 중 오류 발생: {str(e)}")
|
|
|
|
async def exchange_long_lived_token(self, short_lived_token: str) -> dict:
|
|
"""
|
|
단기 토큰을 장기 토큰으로 교환 (약 60일 유효)
|
|
|
|
Args:
|
|
short_lived_token: 단기 액세스 토큰
|
|
|
|
Returns:
|
|
dict: {access_token, token_type, expires_in}
|
|
|
|
Raises:
|
|
FacebookAuthFailedError: 토큰 교환 실패 시
|
|
FacebookAPIError: API 호출 오류 시
|
|
"""
|
|
logger.info("[FACEBOOK] 장기 토큰 교환 시작")
|
|
|
|
# Facebook Graph API - 단기 토큰을 장기 토큰으로 교환
|
|
token_url = self.TOKEN_URL_TEMPLATE.format(version=self.api_version)
|
|
params = {
|
|
"grant_type": "fb_exchange_token",
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"fb_exchange_token": short_lived_token,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
logger.debug(f"[FACEBOOK] 장기 토큰 교환 URL: {token_url}")
|
|
response = await client.get(token_url, params=params)
|
|
result = response.json()
|
|
|
|
logger.debug(f"[FACEBOOK] 장기 토큰 교환 응답 상태 - status: {response.status_code}")
|
|
|
|
# 에러 응답 처리
|
|
if "error" in result:
|
|
error_msg = result.get("error", {})
|
|
error_message = error_msg.get("message", "알 수 없는 오류")
|
|
logger.error(
|
|
f"[FACEBOOK] 장기 토큰 교환 실패 - "
|
|
f"type: {error_msg.get('type')}, message: {error_message}"
|
|
)
|
|
raise FacebookAuthFailedError(f"Facebook 장기 토큰 교환 실패: {error_message}")
|
|
|
|
expires_in = result.get("expires_in", 0)
|
|
logger.info(f"[FACEBOOK] 장기 토큰 교환 성공 - expires_in: {expires_in}초 (약 {expires_in // 86400}일)")
|
|
|
|
return result
|
|
|
|
except FacebookAuthFailedError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[FACEBOOK] API 호출 오류 - error: {str(e)}")
|
|
raise FacebookAPIError(f"Facebook API 호출 중 오류 발생: {str(e)}")
|
|
|
|
async def get_user_info(self, access_token: str) -> dict:
|
|
"""
|
|
액세스 토큰으로 사용자 정보 조회
|
|
|
|
Args:
|
|
access_token: Facebook 액세스 토큰
|
|
|
|
Returns:
|
|
dict: {id, name, email, picture}
|
|
|
|
Raises:
|
|
FacebookAuthFailedError: 사용자 정보 조회 실패 시
|
|
FacebookAPIError: API 호출 오류 시
|
|
"""
|
|
logger.info("[FACEBOOK] 사용자 정보 조회 시작")
|
|
|
|
# Facebook Graph API - 사용자 프로필 조회
|
|
user_info_url = self.USER_INFO_URL_TEMPLATE.format(version=self.api_version)
|
|
params = {
|
|
"fields": "id,name,email,picture",
|
|
"access_token": access_token,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
logger.debug(f"[FACEBOOK] 사용자 정보 요청 URL: {user_info_url}")
|
|
response = await client.get(user_info_url, params=params)
|
|
result = response.json()
|
|
|
|
logger.debug(f"[FACEBOOK] 사용자 정보 응답 상태 - status: {response.status_code}")
|
|
|
|
# 에러 응답 처리
|
|
if "error" in result:
|
|
error_msg = result.get("error", {})
|
|
error_message = error_msg.get("message", "알 수 없는 오류")
|
|
error_code = error_msg.get("code")
|
|
logger.error(
|
|
f"[FACEBOOK] 사용자 정보 조회 실패 - "
|
|
f"code: {error_code}, message: {error_message}"
|
|
)
|
|
# 토큰 만료 에러 (code=190)
|
|
if error_code == 190:
|
|
raise FacebookTokenExpiredError()
|
|
raise FacebookAuthFailedError(f"Facebook 사용자 정보 조회 실패: {error_message}")
|
|
|
|
# 필수 필드(id) 확인
|
|
if "id" not in result:
|
|
logger.error(f"[FACEBOOK] 사용자 정보에 id 없음 - response: {result}")
|
|
raise FacebookAuthFailedError("Facebook 사용자 정보를 가져올 수 없습니다.")
|
|
|
|
logger.info(f"[FACEBOOK] 사용자 정보 조회 성공 - id: {result.get('id')}")
|
|
logger.debug(
|
|
f"[FACEBOOK] 사용자 상세 정보 - "
|
|
f"name: {result.get('name')}, email: {result.get('email')}"
|
|
)
|
|
|
|
return result
|
|
|
|
except (FacebookAuthFailedError, FacebookTokenExpiredError):
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[FACEBOOK] API 호출 오류 - error: {str(e)}")
|
|
raise FacebookAPIError(f"Facebook API 호출 중 오류 발생: {str(e)}")
|
|
|
|
async def get_user_pages(self, user_id: str, access_token: str) -> list[dict]:
|
|
"""
|
|
사용자가 관리하는 Facebook 페이지 목록 조회
|
|
|
|
Args:
|
|
user_id: Facebook 사용자 ID
|
|
access_token: Facebook 액세스 토큰
|
|
|
|
Returns:
|
|
list[dict]: [{id, name, access_token, category}, ...]
|
|
|
|
Raises:
|
|
FacebookAPIError: API 호출 오류 시
|
|
"""
|
|
logger.info(f"[FACEBOOK] 페이지 목록 조회 시작 - user_id: {user_id}")
|
|
|
|
# Facebook Graph API - 사용자 관리 페이지 목록 조회
|
|
pages_url = self.PAGES_URL_TEMPLATE.format(
|
|
version=self.api_version, user_id=user_id
|
|
)
|
|
params = {"access_token": access_token}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
logger.debug(f"[FACEBOOK] 페이지 목록 요청 URL: {pages_url}")
|
|
response = await client.get(pages_url, params=params)
|
|
result = response.json()
|
|
|
|
logger.debug(f"[FACEBOOK] 페이지 목록 응답 상태 - status: {response.status_code}")
|
|
|
|
# 에러 응답 처리
|
|
if "error" in result:
|
|
error_msg = result.get("error", {})
|
|
error_message = error_msg.get("message", "알 수 없는 오류")
|
|
logger.error(f"[FACEBOOK] 페이지 목록 조회 실패 - message: {error_message}")
|
|
raise FacebookAPIError(f"Facebook 페이지 목록 조회 실패: {error_message}")
|
|
|
|
pages = result.get("data", [])
|
|
logger.info(f"[FACEBOOK] 페이지 목록 조회 성공 - 페이지 수: {len(pages)}")
|
|
logger.debug(
|
|
f"[FACEBOOK] 페이지 상세 - "
|
|
f"pages: {[{'id': p.get('id'), 'name': p.get('name')} for p in pages]}"
|
|
)
|
|
|
|
return pages
|
|
|
|
except FacebookAPIError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[FACEBOOK] API 호출 오류 - error: {str(e)}")
|
|
raise FacebookAPIError(f"Facebook API 호출 중 오류 발생: {str(e)}")
|
|
|
|
@staticmethod
|
|
def is_token_expired(token_expires_at) -> bool:
|
|
"""
|
|
토큰 만료 여부 확인 (만료 7일 전부터 True 반환)
|
|
|
|
Args:
|
|
token_expires_at: 토큰 만료 일시 (aware datetime)
|
|
|
|
Returns:
|
|
True: 토큰이 만료되었거나 7일 이내 만료 예정
|
|
"""
|
|
from datetime import timedelta
|
|
|
|
from app.utils.timezone import now
|
|
|
|
# 만료 7일 전부터 재연동 필요로 판단 (aware datetime 사용)
|
|
threshold = now() + timedelta(days=7)
|
|
is_expired = token_expires_at <= threshold
|
|
logger.debug(
|
|
f"[FACEBOOK] 토큰 만료 확인 - "
|
|
f"expires_at: {token_expires_at}, threshold: {threshold}, expired: {is_expired}"
|
|
)
|
|
return is_expired
|