Compare commits

...

6 Commits

Author SHA1 Message Date
Dohyun Lim c6fc164c41 Merge branch 'facebook' of https://gitea.o2o.kr/castad/o2o-castad-backend into facebook
마지막 main 머지를 facebook 브렌치로 백업, 저장한 버전
2026-03-13 15:49:38 +09:00
Dohyun Lim 4fbfbf92a6 add exception error cases 2026-03-13 11:22:55 +09:00
Dohyun Lim 50796ac743 Merge facebook into main 2026-03-13 11:04:25 +09:00
Dohyun Lim 9c3f616f37 finish facebook login 2026-03-13 10:59:03 +09:00
Dohyun Lim 0dd0c8595f add fc 2026-03-09 13:54:10 +09:00
Dohyun Lim 6d86aaa1be first commit 2026-03-05 13:29:33 +09:00
12 changed files with 2046 additions and 16 deletions

674
WORK_PLAN_FACEBOOK_OAUTH.md Normal file
View File

@ -0,0 +1,674 @@
# Facebook OAuth 구현 작업 계획서
> 작성일: 2026-03-03
> 프로젝트: O2O Castad Backend
> 대상 모듈: `app/sns/` (Facebook OAuth 연동)
---
## 1. 개요
Facebook OAuth 2.0 로그인 기능을 구현합니다.
Facebook Graph API를 통해 사용자 인증, 토큰 교환, 장기 토큰 발급, 페이지 토큰 조회 기능을 제공합니다.
### 참조 공식 문서
- Facebook Manual Login Flow: https://developers.facebook.com/docs/facebook-login/guides/advanced/manual-flow
- Facebook Graph API: https://developers.facebook.com/docs/graph-api
- Long-Lived Token Exchange: https://developers.facebook.com/docs/facebook-login/guides/access-tokens/get-long-lived
### Facebook OAuth 2.0 핵심 흐름
```
1. 사용자 → Facebook 인증 페이지 리다이렉트
URL: https://www.facebook.com/v21.0/dialog/oauth
Params: client_id, redirect_uri, state, scope, response_type=code
2. Facebook → 콜백 URL로 인가 코드(code) 전달
3. 서버 → 인가 코드를 액세스 토큰으로 교환
URL: https://graph.facebook.com/v21.0/oauth/access_token
Params: client_id, client_secret, code, redirect_uri
4. 서버 → 단기 토큰을 장기 토큰으로 교환 (약 60일)
URL: https://graph.facebook.com/v21.0/oauth/access_token
Params: grant_type=fb_exchange_token, client_id, client_secret, fb_exchange_token
5. 서버 → 사용자 정보 조회
URL: https://graph.facebook.com/v21.0/me
Params: fields=id,name,email,picture
6. 서버 → 페이지 목록 및 페이지 토큰 조회 (선택)
URL: https://graph.facebook.com/v21.0/{user-id}/accounts
```
---
## 2. 파일 구조 및 역할 분담
```
app/
├── utils/
│ └── facebook_oauth.py # [신규] Facebook OAuth 클래스 (API 통신 전담)
├── sns/
│ ├── api/routers/v1/
│ │ └── oauth.py # [구현] 엔드포인트 정의 (prefix: /sns)
│ ├── services/
│ │ └── facebook.py # [구현] 비즈니스 로직 (facebook_oauth.py 클래스 호출)
│ ├── models.py # [검토/수정] SNSUploadTask 필드 추가 여부 확인
│ ├── dependency.py # [구현] SNS 전용 Depends 정의
│ └── schemas/
│ └── facebook_schema.py # [신규] Facebook OAuth 스키마 정의
├── config.py # [수정] FacebookSettings 독립 클래스 신설 + 인스턴스 생성
├── .env # [수정] FACEBOOK_ 환경변수 추가
└── main.py # [수정] 라우터 등록 및 Scalar 문서 업데이트
```
---
## 3. 상세 작업 단계
### Phase 1: 설정 및 기반 작업
#### Step 1.1: config.py - FacebookSettings 독립 클래스 신설
- **파일**: `config.py`
- **작업**: `FacebookSettings` 독립 클래스를 신규 생성 (기존 `SocialOAuthSettings` 내 주석 코드는 삭제)
- **네이밍 패턴**: 기존 `KakaoSettings`, `JWTSettings` 등과 동일한 패턴 준수
- **환경변수**: 실제 값은 `.env` 파일에서 `FACEBOOK_` prefix 변수로 관리
**config.py에 추가할 클래스** (KakaoSettings 바로 아래에 배치):
```python
class FacebookSettings(BaseSettings):
"""Facebook OAuth 설정
Facebook Graph API를 통한 OAuth 2.0 인증 설정입니다.
Meta for Developers (https://developers.facebook.com/)에서 앱을 생성하고
App ID/Secret을 발급받아야 합니다.
"""
FACEBOOK_APP_ID: str = Field(
default="",
description="Facebook App ID (Meta 개발자 콘솔에서 발급)",
)
FACEBOOK_APP_SECRET: str = Field(
default="",
description="Facebook App Secret",
)
FACEBOOK_REDIRECT_URI: str = Field(
default="http://localhost:8000/sns/facebook/callback",
description="Facebook OAuth 콜백 URI",
)
FACEBOOK_GRAPH_API_VERSION: str = Field(
default="v21.0",
description="Facebook Graph API 버전",
)
FACEBOOK_OAUTH_SCOPE: str = Field(
default="public_profile,email,pages_show_list,pages_read_engagement,pages_manage_posts",
description="Facebook OAuth 요청 권한 범위 (쉼표 구분)",
)
model_config = _base_config
```
**인스턴스 생성** (파일 하단 인스턴스 블록에 추가):
```python
facebook_settings = FacebookSettings()
```
**SocialOAuthSettings 내 기존 Facebook 주석 코드 처리**:
- `SocialOAuthSettings` 내의 Facebook 관련 주석 처리된 코드 블록(524~530행)은 삭제
- 해당 책임이 `FacebookSettings` 클래스로 완전 이관됨
**.env 파일에 추가할 환경변수**:
```env
# ============================================================
# Facebook OAuth 설정
# ============================================================
FACEBOOK_APP_ID=your-facebook-app-id
FACEBOOK_APP_SECRET=your-facebook-app-secret
FACEBOOK_REDIRECT_URI=http://localhost:8000/sns/facebook/callback
FACEBOOK_GRAPH_API_VERSION=v21.0
FACEBOOK_OAUTH_SCOPE=public_profile,email,pages_show_list,pages_read_engagement,pages_manage_posts
```
- **주의**: redirect_uri는 sns 프리픽스 기준으로 설정 (`/sns/facebook/callback`)
---
### Phase 2: OAuth 클라이언트 구현
#### Step 2.1: app/utils/facebook_oauth.py - FacebookOAuthClient 클래스 구현
- **파일**: `app/utils/facebook_oauth.py` (신규)
- **클래스명**: `FacebookOAuthClient` (oauth 단어 포함 필수)
- **역할**: Facebook Graph API와의 HTTP 통신 전담
- **패턴**: `KakaoOAuthClient` (app/user/services/kakao.py) 패턴 준수
- **HTTP 클라이언트**: `httpx.AsyncClient` (기존 Instagram 패턴 사용)
- **Graph API 버전**: v21.0
**클래스 구조**:
```python
from config import facebook_settings
class FacebookOAuthClient:
"""Facebook OAuth 2.0 API 클라이언트"""
# URL 템플릿 (API 버전은 facebook_settings에서 동적 로드)
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):
# facebook_settings에서 설정값 로드
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
def get_authorization_url(self, state: str) -> str:
# Facebook 인증 페이지 URL 생성
# scope: facebook_settings.FACEBOOK_OAUTH_SCOPE에서 로드
async def get_access_token(self, code: str) -> dict:
# 인가 코드 → 단기 액세스 토큰 교환
# 반환: {access_token, token_type, expires_in}
async def exchange_long_lived_token(self, short_lived_token: str) -> dict:
# 단기 토큰 → 장기 토큰 교환 (약 60일)
# 반환: {access_token, token_type, expires_in}
async def get_user_info(self, access_token: str) -> dict:
# 사용자 프로필 조회
# fields: id, name, email, picture
# 반환: {id, name, email, picture}
async def get_user_pages(self, user_id: str, access_token: str) -> list[dict]:
# 사용자가 관리하는 Facebook 페이지 목록 조회
# 반환: [{id, name, access_token, category}, ...]
```
**예외 클래스** (같은 파일 내 정의):
```python
class FacebookOAuthException(HTTPException):
"""Facebook OAuth 기본 예외"""
class FacebookAuthFailedError(FacebookOAuthException):
"""인증 실패 (400)"""
class FacebookAPIError(FacebookOAuthException):
"""API 호출 오류 (500)"""
class FacebookTokenExpiredError(FacebookOAuthException):
"""토큰 만료 (401)"""
```
**필수 사항**:
- 모든 메서드에 `logger.debug()` 로 입력값, 호출 URL, 응답 상태 기록
- 중요 함수 호출 부분에 한글 주석 추가
- 모듈 로거: `get_logger(__name__)`
---
### Phase 3: 스키마 정의
#### Step 3.1: app/sns/schemas/facebook_schema.py - Facebook 스키마 구현
- **파일**: `app/sns/schemas/facebook_schema.py` (신규)
- **패턴**: 기존 `sns_schema.py` 패턴 준수
**스키마 정의**:
```python
class FacebookConnectResponse(BaseModel):
"""Facebook OAuth 연동 시작 응답"""
auth_url: str # Facebook 인증 페이지 URL
state: str # CSRF 방지용 state 토큰
class FacebookCallbackRequest(BaseModel):
"""Facebook OAuth 콜백 파라미터"""
code: str # 인가 코드
state: str # CSRF state 토큰
class FacebookTokenResponse(BaseModel):
"""Facebook 토큰 교환 응답"""
access_token: str # 액세스 토큰
token_type: str # Bearer
expires_in: int # 만료 시간 (초)
class FacebookUserInfo(BaseModel):
"""Facebook 사용자 정보"""
id: str # Facebook 사용자 ID
name: str # 이름
email: Optional[str] # 이메일 (선택)
picture: Optional[dict] # 프로필 사진 (선택)
class FacebookPageInfo(BaseModel):
"""Facebook 페이지 정보"""
id: str # 페이지 ID
name: str # 페이지 이름
access_token: str # 페이지 액세스 토큰
category: Optional[str] # 카테고리
class FacebookAccountResponse(BaseModel):
"""Facebook 연동 완료 응답"""
success: bool
message: str
account_id: int # SocialAccount.id
platform_user_id: str # Facebook 사용자 ID
platform_username: str # Facebook 사용자 이름
pages: Optional[list[FacebookPageInfo]] # 관리 가능한 페이지 목록
```
---
### Phase 4: 의존성 정의
#### Step 4.1: app/sns/dependency.py - SNS 전용 Depends 구현
- **파일**: `app/sns/dependency.py`
- **역할**: SNS 모듈에서만 사용하는 FastAPI 의존성
**정의 내용**:
```python
async def get_facebook_oauth_client() -> FacebookOAuthClient:
"""FacebookOAuthClient 인스턴스를 제공하는 의존성"""
async def get_facebook_social_account(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> SocialAccount:
"""현재 사용자의 활성 Facebook 소셜 계정을 조회하는 의존성"""
# SocialAccount에서 platform=facebook, is_active=True, is_deleted=False 조회
# 없으면 SocialAccountNotFoundError 발생
async def validate_oauth_state(state: str) -> str:
"""OAuth state 토큰 유효성 검증 의존성"""
# Redis에서 state 조회 및 검증
# 유효하지 않으면 예외 발생
```
---
### Phase 5: 서비스 레이어 구현
#### Step 5.1: app/sns/services/facebook.py - 비즈니스 로직 구현
- **파일**: `app/sns/services/facebook.py`
- **역할**: 엔드포인트와 OAuth 클라이언트 사이의 비즈니스 로직
- **핵심 원칙**: 모든 Facebook API 호출은 `FacebookOAuthClient` 클래스를 통해서만 수행
**서비스 함수 구조**:
```python
class FacebookService:
def __init__(self):
self.oauth_client = FacebookOAuthClient()
async def start_connect(self, user_uuid: str) -> FacebookConnectResponse:
"""Facebook 연동 시작"""
# 1. CSRF state 토큰 생성 (secrets.token_urlsafe)
# 2. Redis에 state:user_uuid 매핑 저장 (TTL: OAUTH_STATE_TTL_SECONDS)
# 3. oauth_client.get_authorization_url(state) 호출
# 4. FacebookConnectResponse 반환
async def handle_callback(
self, code: str, state: str, session: AsyncSession
) -> FacebookAccountResponse:
"""OAuth 콜백 처리"""
# 1. Redis에서 state로 user_uuid 조회 및 검증
# 2. oauth_client.get_access_token(code) → 단기 토큰 획득
# 3. oauth_client.exchange_long_lived_token() → 장기 토큰 교환
# 4. oauth_client.get_user_info() → 사용자 정보 조회
# 5. oauth_client.get_user_pages() → 페이지 목록 조회
# 6. SocialAccount 생성 또는 업데이트 (DB 저장)
# - platform: facebook
# - access_token: 장기 토큰
# - platform_user_id: Facebook 사용자 ID
# - platform_username: Facebook 사용자 이름
# - platform_data: {pages: [...], picture_url: "..."}
# - token_expires_at: 현재시간 + expires_in
# - scope: 요청한 scope 문자열
# 7. FacebookAccountResponse 반환
async def disconnect(
self, user_uuid: str, session: AsyncSession
) -> None:
"""Facebook 연동 해제"""
# SocialAccount 소프트 삭제 (is_deleted=True, is_active=False)
facebook_service = FacebookService()
```
**필수 사항**:
- 모든 주요 단계에서 `logger.debug()` 호출
- 주요 함수 호출 부분에 한글 주석 추가
- 에러 발생 시 `logger.error()` 로 상세 에러 정보 기록
---
### Phase 6: 라우터 구현
#### Step 6.1: app/sns/api/routers/v1/oauth.py - 엔드포인트 구현
- **파일**: `app/sns/api/routers/v1/oauth.py`
- **prefix**: `/sns` (항목 1번 요구사항 - sns 프리픽스 필수)
- **tags**: `["SNS OAuth"]`
**엔드포인트 구조**:
```python
router = APIRouter(prefix="/sns", tags=["SNS OAuth"])
@router.get("/facebook/connect")
async def facebook_connect(
current_user: User = Depends(get_current_user),
) -> FacebookConnectResponse:
"""Facebook OAuth 연동 시작"""
# facebook_service.start_connect() 호출
@router.get("/facebook/callback")
async def facebook_callback(
code: str | None = Query(None),
state: str | None = Query(None),
error: str | None = Query(None),
error_description: str | None = Query(None),
session: AsyncSession = Depends(get_session),
) -> RedirectResponse:
"""Facebook OAuth 콜백 처리"""
# 에러/취소 처리
# facebook_service.handle_callback() 호출
# 성공/실패에 따라 프론트엔드로 리다이렉트
@router.delete("/facebook/disconnect")
async def facebook_disconnect(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> dict:
"""Facebook 계정 연동 해제"""
# facebook_service.disconnect() 호출
```
---
### Phase 7: 모델 검토 및 수정
#### Step 7.1: SNSUploadTask 모델 필드 검토
- **파일**: `app/sns/models.py`
- **검토 항목**: Facebook 토큰 정보 저장을 위한 추가 필드 필요 여부
**분석 결과**:
- `SNSUploadTask`는 업로드 작업 관리용 모델이며, 토큰 저장 역할이 아님
- Facebook 토큰 정보는 기존 `SocialAccount` 모델에 이미 적절한 필드가 존재:
- `access_token` (Text): 장기 토큰 저장
- `refresh_token` (Text, nullable): Facebook은 refresh_token 미지원이므로 NULL
- `token_expires_at` (DateTime): 장기 토큰 만료 시간
- `platform_data` (JSON): 페이지 토큰, 페이지 ID 등 추가 정보
- `scope` (Text): OAuth scope 저장
- `SNSUploadTask`에 Facebook 업로드 시 필요한 필드 추가 검토:
- `platform` 필드 추가 (String(20)): 어떤 플랫폼에 업로드하는지 구분 (현재 Instagram 전용 구조)
- `platform_post_id` 필드 추가 (String(255), nullable): 업로드 후 플랫폼에서 반환한 게시물 ID
- `platform_post_url` 필드 추가 (String(2048), nullable): 업로드 후 게시물 URL
---
### Phase 8: main.py 라우터 등록 및 Scalar 문서 업데이트
#### Step 8.1: main.py 수정
- **파일**: `main.py`
**변경 내용**:
1. **라우터 import 추가**:
```python
from app.sns.api.routers.v1.oauth import router as sns_oauth_router
```
2. **라우터 등록**:
```python
app.include_router(sns_oauth_router) # SNS OAuth 라우터 (Facebook)
```
- 주의: oauth.py 내부 router에 이미 `/sns` prefix가 있으므로 main.py에서는 prefix 추가 불필요
3. **tags_metadata 업데이트** - SNS 태그 설명에 Facebook 추가:
```python
{
"name": "SNS OAuth",
"description": """SNS OAuth API - Facebook 계정 연동
**인증: 필요** - `Authorization: Bearer {access_token}` 헤더 필수
## Facebook OAuth 연동 흐름
1. `GET /sns/facebook/connect` - Facebook OAuth 인증 URL 획득
2. 사용자를 auth_url로 리다이렉트 → Facebook 로그인 및 권한 승인
3. Facebook에서 `/sns/facebook/callback`으로 인가 코드 전달
4. 서버에서 토큰 교환, 장기 토큰 발급, 사용자 정보 조회
5. 연동 완료 후 프론트엔드로 리다이렉트
## 계정 관리
- `DELETE /sns/facebook/disconnect` - Facebook 계정 연동 해제
""",
},
```
4. **공개 엔드포인트 추가** (인증 불필요):
```python
public_endpoints = [
...
"/sns/facebook/callback", # Facebook OAuth 콜백
]
```
5. **기존 SNS 태그 설명 업데이트**:
- SNS 태그 description에 Facebook 업로드 관련 엔드포인트 추가
---
## 4. 작업 순서 (실행 순서)
| 순서 | Phase | 작업 내용 | 대상 파일 | 의존성 |
|------|-------|----------|----------|--------|
| 1 | Phase 1 | config.py FacebookSettings 클래스 신설 + .env 변수 추가 | `config.py`, `.env` | 없음 |
| 2 | Phase 3 | Facebook 스키마 정의 | `app/sns/schemas/facebook_schema.py` | 없음 |
| 3 | Phase 2 | FacebookOAuthClient 클래스 구현 | `app/utils/facebook_oauth.py` | Step 1 (config) |
| 4 | Phase 4 | SNS 전용 Depends 구현 | `app/sns/dependency.py` | Step 3 |
| 5 | Phase 5 | Facebook 서비스 레이어 구현 | `app/sns/services/facebook.py` | Step 2, 3, 4 |
| 6 | Phase 6 | OAuth 라우터 엔드포인트 구현 | `app/sns/api/routers/v1/oauth.py` | Step 2, 5 |
| 7 | Phase 7 | SNSUploadTask 모델 필드 추가 | `app/sns/models.py` | Step 1~6 완료 후 검토 |
| 8 | Phase 8 | main.py 라우터 등록 및 Scalar 문서 | `main.py` | Step 6 |
| 9 | - | 코드리뷰 수행 (`/review`) | 전체 변경 파일 | Step 1~8 전체 |
| 10 | - | 코드리뷰 결과 기반 개선 | 해당 파일 | Step 9 |
---
## 5. 품질 기준 (전 단계 공통)
### 5.1 로깅 기준 (항목 7)
- **모든 엔드포인트 진입점**: `logger.info()` 로 요청 시작 기록
- **외부 API 호출 전/후**: `logger.debug()` 로 URL, 파라미터, 응답 상태 기록
- **중요 변수 할당**: `logger.debug()` 로 변수값 확인
- **에러 발생 시**: `logger.error()` 로 상세 에러 정보 기록
- **로거 패턴**: `from app.utils.logger import get_logger; logger = get_logger(__name__)`
### 5.2 주석 기준 (항목 8)
- 각 메서드의 주요 단계별 한글 주석 (예: `# 인가 코드를 액세스 토큰으로 교환`)
- 외부 API 호출 시 호출 대상 명시 (예: `# Facebook Graph API - 사용자 정보 조회`)
- 복잡한 분기 로직에 판단 기준 설명
### 5.3 코드 품질 기준
- 타입 힌트 필수
- async/await 패턴 준수
- Pydantic v2 스키마 사용
- 서비스 레이어 패턴 (router → service → oauth_client)
---
## 6. Scalar 문서 설정 (항목 9)
### 변경 파일: `main.py`
- `tags_metadata`에 "SNS OAuth" 태그 추가
- `custom_openapi()` 함수의 `public_endpoints`에 Facebook 콜백 경로 추가
- 기존 "SNS" 태그 description에 Facebook 업로드 안내 추가 (향후 확장)
---
## 7. 코드리뷰 수행 (항목 10, 11)
### 리뷰 범위
모든 구현 완료 후 1회 코드리뷰 수행:
1. **설계 검증**
- 레이어 분리 적정성 (router → service → oauth_client)
- 의존성 방향 확인 (순환 의존 없음)
- 예외 처리 체계 일관성
2. **코드 정의 검증**
- 타입 힌트 누락 확인
- async/await 적정 사용
- 로거 사용 패턴 일관성
- 주석 존재 여부
3. **보안 검증**
- CSRF state 토큰 검증 로직
- 토큰 노출 방지 (로그에 토큰 전체 미출력)
- redirect_uri 고정 검증
4. **기능 검증**
- OAuth 흐름 완전성
- 에러 케이스 처리 (취소, 코드 만료, 토큰 실패)
- DB 저장 정합성
### 리뷰 결과 처리 (항목 11)
- 설계적 오류: 구조 변경 및 수정
- 코드 정의 오류: 즉시 수정 반영
---
## 8. 검수 1차 - 작업 계획 완전성 검증
### 검수 항목별 준수 여부
| # | 요구사항 | 준수 여부 | 근거 |
|---|---------|----------|------|
| 1 | oauth.py 라우터에 sns 프리픽스 | **O** | `router = APIRouter(prefix="/sns")` - Phase 6 |
| 2 | utils/facebook_oauth.py에 단일 클래스, 이름에 oauth 포함 | **O** | `FacebookOAuthClient` 클래스 - Phase 2 |
| 3 | 비즈니스 로직은 facebook.py, facebook_oauth.py 클래스 사용 | **O** | `FacebookService``FacebookOAuthClient`를 호출 - Phase 5 |
| 4 | SNSUploadTask 모델 참조 후 필요 필드 추가 | **O** | Phase 7에서 platform, platform_post_id, platform_post_url 추가 검토 |
| 5 | SNS 전용 Depends는 sns/dependency.py | **O** | Phase 4에서 구현 |
| 6 | 스키마는 sns/schemas에 정의 | **O** | `facebook_schema.py` - Phase 3 |
| 7 | 중요 입력/호출/변수 logger.debug 출력 | **O** | 품질 기준 5.1 적용 |
| 8 | 중요 함수 호출 주석 | **O** | 품질 기준 5.2 적용 |
| 9 | Scalar 문서 설정 업데이트 | **O** | Phase 8에서 tags_metadata, public_endpoints 수정 |
| 10 | 1회 코드리뷰 수행 | **O** | Step 9에서 `/review` 수행 |
| 11 | 리뷰 후 설계적/코드 정의 오류 개선 | **O** | Step 10에서 수정 반영 |
### 1차 검수 발견 사항
**발견 1: redirect_uri 불일치 위험**
- config.py의 `SocialOAuthSettings` 내 기존 주석은 `/social/facebook/callback`으로 되어 있으나, 본 계획에서는 `/sns/facebook/callback`으로 설정
- **해결**: `FacebookSettings` 독립 클래스에서 redirect_uri를 `/sns/facebook/callback`으로 명확히 설정하고, `SocialOAuthSettings` 내 Facebook 주석 코드는 삭제
**발견 2: Redis 의존성 미확인**
- OAuth state 토큰 저장에 Redis 사용 예정이나, Redis 클라이언트 획득 방법 미명시
- **해결**: `app/database/session.py`의 Redis 연결 활용 또는 `dependency.py`에 Redis 의존성 추가 명시
**발견 3: 기존 SNS 라우터와의 prefix 충돌 가능성**
- 기존 `app/sns/api/routers/v1/sns.py`에 이미 `prefix="/sns"`가 설정되어 있음
- 새로운 `oauth.py`에도 `prefix="/sns"` 설정 시 main.py에서 중복 prefix 발생 가능
- **해결**: main.py에서 기존 sns_router는 prefix 없이 등록 (`app.include_router(sns_router)`), 새 oauth_router도 prefix 없이 등록하되 router 내부에 `/sns` prefix 유지
**발견 4: schemas 디렉토리 __init__.py 확인 필요**
- `app/sns/schemas/` 디렉토리에 `__init__.py`가 있는지 확인하여 import 가능하도록 보장
---
## 9. 검수 2차 - 설계 개선 검토
### 설계 관점 검토
**검토 1: 레이어 분리 적정성** - **적정**
```
Router (oauth.py)
↓ 호출
Service (facebook.py - FacebookService)
↓ 호출
OAuth Client (facebook_oauth.py - FacebookOAuthClient)
↓ HTTP 요청
Facebook Graph API
```
- 각 레이어의 책임이 명확히 분리됨
- OAuth 클라이언트는 HTTP 통신만, 서비스는 비즈니스 로직만, 라우터는 HTTP 요청/응답만 담당
**검토 2: 기존 social 모듈과의 관계** - **개선 필요 없음**
- `app/social/` 모듈은 YouTube OAuth 및 범용 업로드 담당
- `app/sns/` 모듈은 Instagram/Facebook 등 SNS 특화 기능 담당
- 역할이 명확히 분리되어 있으므로 현행 구조 유지
**검토 3: 토큰 갱신 전략** - **보완 필요**
- Facebook은 refresh_token을 미지원하며, 장기 토큰도 약 60일 만료
- **보완**: `FacebookOAuthClient`에 토큰 만료 확인 유틸리티 메서드 추가
```python
def is_token_expired(self, token_expires_at: datetime) -> bool:
"""토큰 만료 여부 확인 (만료 7일 전부터 True 반환)"""
```
- 서비스 레이어에서 API 호출 전 토큰 만료 확인 → 만료 시 재연동 안내 예외 발생
**검토 4: 페이지 토큰 관리** - **적정**
- 페이지 토큰은 `SocialAccount.platform_data` JSON 필드에 저장
- 이 방식은 기존 YouTube의 channel_id 저장 패턴과 일관됨
**검토 5: scope 범위** - **해결 완료**
- 기본 scope: `public_profile, email` (Facebook 앱 리뷰 없이 사용 가능)
- 페이지 관리 scope: `pages_show_list, pages_read_engagement, pages_manage_posts` (앱 리뷰 필요)
- 비디오 업로드 scope: `publish_video` (향후 확장 시)
- **해결**: `FacebookSettings.FACEBOOK_OAUTH_SCOPE` 필드에서 .env 변수로 관리
- 기본값: `public_profile,email,pages_show_list,pages_read_engagement,pages_manage_posts`
- 향후 확장 시 .env 파일에서 scope 변경만으로 대응 가능
**검토 6: 에러 코드 체계** - **적정**
- Facebook OAuth 예외 클래스가 기존 Kakao 패턴과 동일 구조
- HTTP 상태 코드 + 내부 에러 코드 + 메시지 3단계 체계
### 2차 검수 발견 사항
**발견 1: state 토큰 저장소 구현 세부사항 보완**
- Redis 키 패턴 명확화 필요: `facebook_oauth_state:{state}` → value: `user_uuid`
- TTL 설정: `social_oauth_settings.OAUTH_STATE_TTL_SECONDS` (기본 300초)
- **반영**: Phase 5의 `start_connect()` 구현 시 Redis 키 패턴 명시
**발견 2: 기존 SocialAccount 중복 체크 로직**
- 동일 Facebook 사용자가 재연동할 경우, 기존 레코드를 업데이트해야 함
- **반영**: `handle_callback()`에서 `(user_uuid, platform, platform_user_id)` 기준으로 UPSERT 로직 구현
**발견 3: 향후 확장성 확보**
- 현재 Facebook 전용이지만 TikTok 등 추가 시 패턴 재사용 가능한 구조
- `dependency.py``get_facebook_social_account()`는 플랫폼 파라미터화하여 범용화 가능
- **판단**: 현재 단계에서는 Facebook 전용으로 유지 (YAGNI 원칙)
---
## 10. 최종 작업 실행 순서 (확정)
```
1. config.py 수정 (FacebookSettings 독립 클래스 신설 + SocialOAuthSettings 내 Facebook 주석 삭제)
2. .env 파일에 FACEBOOK_ 환경변수 추가
3. app/sns/schemas/facebook_schema.py 생성 (스키마 정의)
4. app/utils/facebook_oauth.py 생성 (FacebookOAuthClient 클래스 - facebook_settings 참조)
5. app/sns/dependency.py 구현 (SNS 전용 Depends)
6. app/sns/services/facebook.py 구현 (비즈니스 로직)
7. app/sns/api/routers/v1/oauth.py 구현 (엔드포인트)
8. app/sns/models.py 검토 및 필드 추가
9. main.py 수정 (라우터 등록 + Scalar 문서 업데이트)
10. 코드리뷰 수행 (/review)
11. 코드리뷰 결과 기반 개선 수행
```
---
## 부록: 참조 소스
### 기존 코드 패턴 참조 파일
| 패턴 | 참조 파일 | 설명 |
|------|----------|------|
| OAuth 클래스 | `app/user/services/kakao.py` | KakaoOAuthClient 구조 |
| HTTP 클라이언트 | `app/utils/instagram.py` | httpx.AsyncClient 패턴 |
| SNS 라우터 | `app/sns/api/routers/v1/sns.py` | prefix, 예외 처리 패턴 |
| Social OAuth 라우터 | `app/social/api/routers/v1/oauth.py` | 콜백, 리다이렉트 패턴 |
| 모델 | `app/user/models.py` | SocialAccount, Platform enum |
| 스키마 | `app/sns/schemas/sns_schema.py` | Pydantic v2 패턴 |
| 설정 (독립 클래스) | `config.py` | KakaoSettings 패턴 → FacebookSettings |
| 설정 (OAuth 공통) | `config.py` | SocialOAuthSettings (state TTL, 프론트엔드 URL 등) |
| 로거 | `app/utils/logger.py` | get_logger 패턴 |

View File

@ -24,10 +24,12 @@ engine = create_async_engine(
max_overflow=20, # 추가 연결: 20 (총 최대 40) max_overflow=20, # 추가 연결: 20 (총 최대 40)
pool_timeout=30, # 풀에서 연결 대기 시간 (초) pool_timeout=30, # 풀에서 연결 대기 시간 (초)
pool_recycle=280, # MySQL wait_timeout(기본 28800s, 클라우드는 보통 300s) 보다 짧게 설정 pool_recycle=280, # MySQL wait_timeout(기본 28800s, 클라우드는 보통 300s) 보다 짧게 설정
pool_use_lifo=True,
pool_pre_ping=True, # 연결 유효성 검사 (죽은 연결 자동 재연결) pool_pre_ping=True, # 연결 유효성 검사 (죽은 연결 자동 재연결)
pool_reset_on_return="rollback", # 반환 시 롤백으로 초기화 pool_reset_on_return="rollback", # 반환 시 롤백으로 초기화
connect_args={ connect_args={
"connect_timeout": 10, # DB 연결 타임아웃 "connect_timeout": 10, # DB 연결 타임아웃
"read_timeout": 30,
"charset": "utf8mb4", "charset": "utf8mb4",
}, },
) )
@ -51,10 +53,12 @@ background_engine = create_async_engine(
max_overflow=10, # 추가 연결: 10 (총 최대 20) max_overflow=10, # 추가 연결: 10 (총 최대 20)
pool_timeout=60, # 백그라운드는 대기 시간 여유있게 pool_timeout=60, # 백그라운드는 대기 시간 여유있게
pool_recycle=280, # MySQL wait_timeout 보다 짧게 설정 pool_recycle=280, # MySQL wait_timeout 보다 짧게 설정
pool_use_lifo=True,
pool_pre_ping=True, # 연결 유효성 검사 (죽은 연결 자동 재연결) pool_pre_ping=True, # 연결 유효성 검사 (죽은 연결 자동 재연결)
pool_reset_on_return="rollback", pool_reset_on_return="rollback",
connect_args={ connect_args={
"connect_timeout": 10, "connect_timeout": 10,
"read_timeout": 30,
"charset": "utf8mb4", "charset": "utf8mb4",
}, },
) )
@ -128,13 +132,22 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]:
try: try:
yield session yield session
except Exception as e: except Exception as e:
import traceback from fastapi import HTTPException
await session.rollback() await session.rollback()
logger.error(traceback.format_exc()) duration = (time.perf_counter() - start_time) * 1000
logger.error( # 클라이언트 에러(4xx)는 WARNING, 서버 에러(5xx)는 ERROR로 구분
f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, " if isinstance(e, HTTPException) and e.status_code < 500:
f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms" logger.warning(
) f"[get_session] ROLLBACK ({e.status_code}) - "
f"error: {type(e).__name__}: {e}, duration: {duration:.1f}ms"
)
else:
import traceback
logger.error(traceback.format_exc())
logger.error(
f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
f"duration: {duration:.1f}ms"
)
raise e raise e
finally: finally:
total_time = time.perf_counter() - start_time total_time = time.perf_counter() - start_time

View File

@ -0,0 +1,209 @@
"""
SNS OAuth API 라우터
Facebook OAuth 연동 관련 엔드포인트를 제공합니다.
"""
from urllib.parse import urlencode
from fastapi import APIRouter, Depends, Query
from fastapi.responses import RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.session import get_session
from app.sns.schemas.facebook_schema import FacebookConnectResponse
from app.sns.services.facebook import facebook_service
from app.user.dependencies.auth import get_current_user
from app.user.models import User
from app.utils.logger import get_logger
from config import social_oauth_settings
logger = get_logger(__name__)
router = APIRouter(prefix="/sns", tags=["SNS OAuth"])
def _build_redirect_url(is_success: bool, params: dict) -> str:
"""OAuth 완료 후 프론트엔드 리다이렉트 URL 생성"""
base_url = social_oauth_settings.OAUTH_FRONTEND_URL.rstrip("/")
path = (
social_oauth_settings.OAUTH_SUCCESS_PATH
if is_success
else social_oauth_settings.OAUTH_ERROR_PATH
)
return f"{base_url}{path}?{urlencode(params)}"
@router.get(
"/facebook/connect",
response_model=FacebookConnectResponse,
summary="Facebook OAuth 연동 시작",
description="""
## 개요
Facebook OAuth 2.0 인증을 시작합니다.
## 플로우
1. 엔드포인트를 호출하여 `auth_url` `state` 받음
2. 프론트엔드에서 `auth_url` 사용자를 리다이렉트
3. 사용자가 Facebook에서 로그인 권한 승인
4. Facebook이 `/sns/facebook/callback` 엔드포인트로 리다이렉트
5. 연동 완료 프론트엔드로 리다이렉트
## 인증
- Bearer 토큰 필요 (Authorization: Bearer <token>)
""",
responses={
200: {"description": "인증 URL 반환 성공"},
401: {"description": "인증 실패"},
},
)
async def facebook_connect(
current_user: User = Depends(get_current_user),
) -> FacebookConnectResponse:
"""Facebook OAuth 연동을 시작합니다."""
logger.info(f"[SNS_OAUTH] Facebook 연동 시작 - user_uuid: {current_user.user_uuid}")
# FacebookService를 통해 연동 시작
response = await facebook_service.start_connect(user_uuid=current_user.user_uuid)
logger.info("[SNS_OAUTH] Facebook 연동 URL 생성 완료")
return response
@router.get(
"/facebook/callback",
summary="Facebook OAuth 콜백",
description="""
## 개요
Facebook OAuth 콜백을 처리합니다.
엔드포인트는 Facebook에서 직접 호출되며,
처리 완료 프론트엔드로 리다이렉트합니다.
## 파라미터
- **code**: Facebook에서 발급한 인가 코드
- **state**: CSRF 방지용 state 토큰
- **error**: OAuth 에러 코드 (사용자 취소 )
""",
responses={
302: {"description": "프론트엔드로 리다이렉트"},
},
)
async def facebook_callback(
code: str | None = Query(None, description="Facebook 인가 코드"),
state: str | None = Query(None, description="CSRF 방지용 state 토큰"),
error: str | None = Query(None, description="OAuth 에러 코드"),
error_description: str | None = Query(None, description="OAuth 에러 설명"),
session: AsyncSession = Depends(get_session),
) -> RedirectResponse:
"""Facebook OAuth 콜백을 처리합니다."""
# 사용자가 취소하거나 에러가 발생한 경우
if error:
logger.info(
f"[SNS_OAUTH] Facebook 콜백 에러/취소 - "
f"error: {error}, description: {error_description}"
)
# 에러 메시지 분기
if error == "access_denied":
error_message = "사용자가 Facebook 연동을 취소했습니다."
else:
error_message = error_description or error
redirect_url = _build_redirect_url(
is_success=False,
params={
"platform": "facebook",
"error": error_message,
"cancelled": "true" if error == "access_denied" else "false",
},
)
return RedirectResponse(url=redirect_url, status_code=302)
# code 또는 state가 없는 경우
if not code or not state:
logger.warning(
f"[SNS_OAUTH] Facebook 콜백 파라미터 누락 - "
f"code: {bool(code)}, state: {bool(state)}"
)
redirect_url = _build_redirect_url(
is_success=False,
params={
"platform": "facebook",
"error": "잘못된 요청입니다. 다시 시도해주세요.",
},
)
return RedirectResponse(url=redirect_url, status_code=302)
logger.info(f"[SNS_OAUTH] Facebook 콜백 수신 - code: {code[:20]}...")
try:
# FacebookService를 통해 콜백 처리
account_response = await facebook_service.handle_callback(
code=code,
state=state,
session=session,
)
# 성공 시 프론트엔드로 리다이렉트
redirect_url = _build_redirect_url(
is_success=True,
params={
"platform": "facebook",
"account_id": account_response.account_id,
"channel_name": account_response.platform_username,
},
)
logger.info("[SNS_OAUTH] Facebook 연동 성공, 리다이렉트")
return RedirectResponse(url=redirect_url, status_code=302)
except Exception as e:
logger.error(f"[SNS_OAUTH] Facebook 콜백 처리 실패 - error: {e}")
# 실패 시 에러 페이지로 리다이렉트
redirect_url = _build_redirect_url(
is_success=False,
params={
"platform": "facebook",
"error": str(e),
},
)
return RedirectResponse(url=redirect_url, status_code=302)
@router.delete(
"/facebook/disconnect",
summary="Facebook 계정 연동 해제",
description="""
## 개요
Facebook 계정 연동을 해제합니다.
## 연동 해제 시
- Facebook으로의 업로드가 불가능해집니다
- 기존 업로드 기록은 유지됩니다
- 재연동 다시 권한 승인이 필요합니다
## 인증
- Bearer 토큰 필요 (Authorization: Bearer <token>)
""",
responses={
200: {"description": "연동 해제 성공"},
401: {"description": "인증 실패"},
404: {"description": "연동된 Facebook 계정 없음"},
},
)
async def facebook_disconnect(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> dict:
"""Facebook 계정 연동을 해제합니다."""
logger.info(f"[SNS_OAUTH] Facebook 연동 해제 - user_uuid: {current_user.user_uuid}")
# FacebookService를 통해 연동 해제
await facebook_service.disconnect(
user_uuid=current_user.user_uuid,
session=session,
)
logger.info("[SNS_OAUTH] Facebook 연동 해제 완료")
return {"success": True, "message": "Facebook 계정 연동이 해제되었습니다."}

View File

@ -0,0 +1,72 @@
"""
SNS 모듈 전용 FastAPI 의존성
SNS 모듈에서만 사용하는 의존성을 정의합니다.
Facebook OAuth 관련 의존성을 포함합니다.
"""
from fastapi import Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.session import get_session
from app.user.dependencies.auth import get_current_user
from app.user.models import Platform, SocialAccount, User
from app.utils.facebook_oauth import FacebookOAuthClient
from app.utils.logger import get_logger
logger = get_logger(__name__)
def get_facebook_oauth_client() -> FacebookOAuthClient:
"""
FacebookOAuthClient 인스턴스를 제공하는 의존성
Returns:
FacebookOAuthClient: Facebook OAuth API 클라이언트 인스턴스
"""
return FacebookOAuthClient()
async def get_facebook_social_account(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> SocialAccount:
"""
현재 사용자의 활성 Facebook 소셜 계정을 조회하는 의존성
Args:
current_user: 현재 인증된 사용자
session: DB 세션
Returns:
SocialAccount: 활성 Facebook 소셜 계정
Raises:
HTTPException: Facebook 소셜 계정이 없는 경우 404 반환
"""
logger.debug(f"[SNS_DEP] Facebook 소셜 계정 조회 - user_uuid: {current_user.user_uuid}")
# SocialAccount에서 Facebook 활성 계정 조회
result = await session.execute(
select(SocialAccount).where(
SocialAccount.user_uuid == current_user.user_uuid,
SocialAccount.platform == Platform.FACEBOOK,
SocialAccount.is_active == True, # noqa: E712
SocialAccount.is_deleted == False, # noqa: E712
)
)
social_account = result.scalar_one_or_none()
if social_account is None:
logger.warning(f"[SNS_DEP] Facebook 계정 없음 - user_uuid: {current_user.user_uuid}")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"code": "FACEBOOK_ACCOUNT_NOT_FOUND",
"message": "연동된 Facebook 계정을 찾을 수 없습니다.",
},
)
logger.debug(f"[SNS_DEP] Facebook 소셜 계정 확인 - account_id: {social_account.id}")
return social_account

View File

@ -52,6 +52,7 @@ class SNSUploadTask(Base):
Index("idx_sns_upload_task_user_uuid", "user_uuid"), Index("idx_sns_upload_task_user_uuid", "user_uuid"),
Index("idx_sns_upload_task_task_id", "task_id"), Index("idx_sns_upload_task_task_id", "task_id"),
Index("idx_sns_upload_task_social_account_id", "social_account_id"), Index("idx_sns_upload_task_social_account_id", "social_account_id"),
Index("idx_sns_upload_task_platform", "platform"),
Index("idx_sns_upload_task_status", "status"), Index("idx_sns_upload_task_status", "status"),
Index("idx_sns_upload_task_is_scheduled", "is_scheduled"), Index("idx_sns_upload_task_is_scheduled", "is_scheduled"),
Index("idx_sns_upload_task_scheduled_at", "scheduled_at"), Index("idx_sns_upload_task_scheduled_at", "scheduled_at"),
@ -116,6 +117,12 @@ class SNSUploadTask(Base):
comment="소셜 계정 외래키 (SocialAccount.id 참조)", comment="소셜 계정 외래키 (SocialAccount.id 참조)",
) )
platform: Mapped[Optional[str]] = mapped_column(
String(20),
nullable=True,
comment="업로드 대상 플랫폼 (instagram, facebook 등)",
)
# ========================================================================== # ==========================================================================
# 업로드 콘텐츠 # 업로드 콘텐츠
# ========================================================================== # ==========================================================================
@ -131,6 +138,18 @@ class SNSUploadTask(Base):
comment="게시물 캡션/설명", comment="게시물 캡션/설명",
) )
platform_post_id: Mapped[Optional[str]] = mapped_column(
String(255),
nullable=True,
comment="업로드 후 플랫폼에서 반환한 게시물 ID",
)
platform_post_url: Mapped[Optional[str]] = mapped_column(
String(2048),
nullable=True,
comment="업로드 후 게시물 URL (permalink)",
)
# ========================================================================== # ==========================================================================
# 발행 상태 # 발행 상태
# ========================================================================== # ==========================================================================

View File

@ -0,0 +1,122 @@
"""
Facebook OAuth API Schemas
Facebook OAuth 연동 관련 Pydantic 스키마를 정의합니다.
"""
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class FacebookConnectResponse(BaseModel):
"""Facebook OAuth 연동 시작 응답
Usage:
GET /sns/facebook/connect
Facebook OAuth 인증 URL과 CSRF state 토큰을 반환합니다.
Example Response:
{
"auth_url": "https://www.facebook.com/v21.0/dialog/oauth?client_id=...",
"state": "abc123xyz"
}
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"auth_url": "https://www.facebook.com/v21.0/dialog/oauth?client_id=123&redirect_uri=...",
"state": "abc123xyz456",
}
}
)
auth_url: str = Field(..., description="Facebook 인증 페이지 URL")
state: str = Field(..., description="CSRF 방지용 state 토큰")
class FacebookTokenResponse(BaseModel):
"""Facebook 토큰 교환 응답
Facebook Graph API에서 반환하는 액세스 토큰 정보입니다.
"""
access_token: str = Field(..., description="액세스 토큰")
token_type: str = Field(default="bearer", description="토큰 타입")
expires_in: int = Field(..., description="만료 시간 (초)")
class FacebookUserInfo(BaseModel):
"""Facebook 사용자 정보
Graph API /me 엔드포인트에서 반환하는 사용자 프로필 정보입니다.
"""
id: str = Field(..., description="Facebook 사용자 ID")
name: str = Field(..., description="사용자 이름")
email: Optional[str] = Field(default=None, description="이메일 (동의 시 제공)")
picture: Optional[dict] = Field(default=None, description="프로필 사진 정보")
class FacebookPageInfo(BaseModel):
"""Facebook 페이지 정보
사용자가 관리하는 Facebook 페이지 정보입니다.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "123456789",
"name": "My Business Page",
"access_token": "EAA...",
"category": "Business",
}
}
)
id: str = Field(..., description="페이지 ID")
name: str = Field(..., description="페이지 이름")
access_token: str = Field(..., description="페이지 액세스 토큰")
category: Optional[str] = Field(default=None, description="페이지 카테고리")
class FacebookAccountResponse(BaseModel):
"""Facebook 연동 완료 응답
Usage:
Facebook OAuth 콜백 처리 완료 반환되는 연동 결과입니다.
Example Response:
{
"success": true,
"message": "Facebook 계정 연동이 완료되었습니다.",
"account_id": 1,
"platform_user_id": "123456789",
"platform_username": "홍길동",
"pages": [...]
}
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"success": True,
"message": "Facebook 계정 연동이 완료되었습니다.",
"account_id": 1,
"platform_user_id": "123456789",
"platform_username": "홍길동",
"pages": [],
}
}
)
success: bool = Field(..., description="연동 성공 여부")
message: str = Field(..., description="결과 메시지")
account_id: int = Field(..., description="SocialAccount ID")
platform_user_id: str = Field(..., description="Facebook 사용자 ID")
platform_username: str = Field(..., description="Facebook 사용자 이름")
pages: Optional[list[FacebookPageInfo]] = Field(
default=None, description="관리 가능한 Facebook 페이지 목록"
)

View File

@ -0,0 +1,287 @@
"""
Facebook OAuth 서비스
Facebook OAuth 연동 관련 비즈니스 로직을 처리합니다.
모든 Facebook API 호출은 FacebookOAuthClient를 통해서만 수행됩니다.
"""
import json
import secrets
from datetime import timedelta
from redis.asyncio import Redis
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.sns.schemas.facebook_schema import (
FacebookAccountResponse,
FacebookConnectResponse,
FacebookPageInfo,
)
from app.user.models import Platform, SocialAccount
from app.utils.facebook_oauth import FacebookOAuthClient
from app.utils.logger import get_logger
from app.utils.timezone import now
from config import db_settings, social_oauth_settings
logger = get_logger(__name__)
# Facebook OAuth용 Redis 클라이언트 (DB 3 사용 - social과 분리)
_redis_client = Redis(
host=db_settings.REDIS_HOST,
port=db_settings.REDIS_PORT,
db=3,
decode_responses=True,
)
class FacebookService:
"""
Facebook OAuth 연동 서비스
OAuth 인증 시작, 콜백 처리, 연동 해제 기능을 제공합니다.
모든 Facebook Graph API 호출은 FacebookOAuthClient를 통해 수행됩니다.
"""
# Redis key prefix for Facebook OAuth state
STATE_KEY_PREFIX = "facebook:oauth:state:"
def __init__(self) -> None:
# FacebookOAuthClient 인스턴스 생성
self.oauth_client = FacebookOAuthClient()
async def start_connect(self, user_uuid: str) -> FacebookConnectResponse:
"""
Facebook OAuth 연동 시작
CSRF state 토큰을 생성하고 Redis에 저장한 ,
Facebook 인증 페이지 URL을 반환합니다.
Args:
user_uuid: 사용자 UUID
Returns:
FacebookConnectResponse: 인증 URL state 토큰
"""
logger.info(f"[FACEBOOK_SVC] 연동 시작 - user_uuid: {user_uuid}")
# 1. CSRF state 토큰 생성
state = secrets.token_urlsafe(32)
logger.debug(f"[FACEBOOK_SVC] state 토큰 생성 - state: {state[:20]}...")
# 2. Redis에 state:user_uuid 매핑 저장 (TTL 적용)
state_key = f"{self.STATE_KEY_PREFIX}{state}"
state_data = json.dumps({"user_uuid": user_uuid})
await _redis_client.setex(
state_key,
social_oauth_settings.OAUTH_STATE_TTL_SECONDS,
state_data,
)
logger.debug(
f"[FACEBOOK_SVC] Redis state 저장 - "
f"key: {state_key}, ttl: {social_oauth_settings.OAUTH_STATE_TTL_SECONDS}"
)
# 3. Facebook 인증 페이지 URL 생성
auth_url = self.oauth_client.get_authorization_url(state)
logger.info("[FACEBOOK_SVC] 연동 시작 완료 - 인증 URL 생성됨")
return FacebookConnectResponse(auth_url=auth_url, state=state)
async def handle_callback(
self, code: str, state: str, session: AsyncSession
) -> FacebookAccountResponse:
"""
Facebook OAuth 콜백 처리
인가 코드를 토큰으로 교환하고, 장기 토큰 발급
사용자 정보를 조회하여 SocialAccount에 저장합니다.
Args:
code: Facebook OAuth 인가 코드
state: CSRF 방지용 state 토큰
session: DB 세션
Returns:
FacebookAccountResponse: 연동 완료 정보
Raises:
HTTPException: state 무효, 토큰 교환 실패
"""
logger.info(f"[FACEBOOK_SVC] 콜백 처리 시작 - state: {state[:20]}...")
# 1. Redis에서 state로 user_uuid 조회 및 검증
state_key = f"{self.STATE_KEY_PREFIX}{state}"
state_data_str = await _redis_client.get(state_key)
if state_data_str is None:
logger.warning(f"[FACEBOOK_SVC] state 토큰 없음 또는 만료 - state: {state[:20]}...")
from fastapi import HTTPException, status
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"code": "FACEBOOK_STATE_EXPIRED",
"message": "인증 세션이 만료되었습니다. 다시 시도해주세요.",
},
)
# state 데이터 파싱 및 삭제 (일회성)
state_data = json.loads(state_data_str)
user_uuid = state_data["user_uuid"]
await _redis_client.delete(state_key)
logger.debug(f"[FACEBOOK_SVC] state 검증 완료 및 삭제 - user_uuid: {user_uuid}")
# 2. 인가 코드를 단기 액세스 토큰으로 교환
short_token_data = await self.oauth_client.get_access_token(code)
short_lived_token = short_token_data["access_token"]
logger.debug("[FACEBOOK_SVC] 단기 토큰 획득 완료")
# 3. 단기 토큰을 장기 토큰으로 교환 (약 60일)
long_token_data = await self.oauth_client.exchange_long_lived_token(short_lived_token)
long_lived_token = long_token_data["access_token"]
expires_in = long_token_data.get("expires_in", 5184000) # 기본 60일
logger.debug(f"[FACEBOOK_SVC] 장기 토큰 교환 완료 - expires_in: {expires_in}")
# 4. 사용자 정보 조회
user_info = await self.oauth_client.get_user_info(long_lived_token)
facebook_user_id = user_info["id"]
facebook_user_name = user_info.get("name", "")
facebook_picture = user_info.get("picture", {})
logger.debug(
f"[FACEBOOK_SVC] 사용자 정보 조회 완료 - "
f"id: {facebook_user_id}, name: {facebook_user_name}"
)
# 5. 사용자 관리 페이지 목록 조회
pages = []
try:
pages_data = await self.oauth_client.get_user_pages(facebook_user_id, long_lived_token)
pages = [
FacebookPageInfo(
id=page["id"],
name=page.get("name", ""),
access_token=page.get("access_token", ""),
category=page.get("category"),
)
for page in pages_data
]
logger.debug(f"[FACEBOOK_SVC] 페이지 목록 조회 완료 - 페이지 수: {len(pages)}")
except Exception as e:
# 페이지 조회 실패는 연동 실패로 처리하지 않음
logger.warning(f"[FACEBOOK_SVC] 페이지 목록 조회 실패 (무시) - error: {str(e)}")
# 6. SocialAccount 생성 또는 업데이트 (UPSERT)
token_expires_at = now() + timedelta(seconds=expires_in)
platform_data = {
"picture_url": facebook_picture.get("data", {}).get("url") if isinstance(facebook_picture, dict) else None,
"pages": [page.model_dump() for page in pages],
}
# 기존 계정 조회 (user_uuid + platform + platform_user_id 기준)
existing_result = await session.execute(
select(SocialAccount).where(
SocialAccount.user_uuid == user_uuid,
SocialAccount.platform == Platform.FACEBOOK,
SocialAccount.platform_user_id == facebook_user_id,
)
)
existing_account = existing_result.scalar_one_or_none()
if existing_account:
# 기존 계정 업데이트 (토큰 갱신 + 재활성화)
logger.info(f"[FACEBOOK_SVC] 기존 계정 업데이트 - account_id: {existing_account.id}")
existing_account.access_token = long_lived_token
existing_account.token_expires_at = token_expires_at
existing_account.platform_username = facebook_user_name
existing_account.platform_data = platform_data
existing_account.scope = self.oauth_client.scope
existing_account.is_active = True
existing_account.is_deleted = False
existing_account.connected_at = now()
await session.commit()
await session.refresh(existing_account)
account = existing_account
else:
# 새 소셜 계정 생성
logger.info(f"[FACEBOOK_SVC] 새 계정 생성 - user_uuid: {user_uuid}")
account = SocialAccount(
user_uuid=user_uuid,
platform=Platform.FACEBOOK,
access_token=long_lived_token,
refresh_token=None, # Facebook은 refresh_token 미지원
token_expires_at=token_expires_at,
scope=self.oauth_client.scope,
platform_user_id=facebook_user_id,
platform_username=facebook_user_name,
platform_data=platform_data,
is_active=True,
is_deleted=False,
)
session.add(account)
await session.commit()
await session.refresh(account)
logger.info(
f"[FACEBOOK_SVC] 연동 완료 - "
f"account_id: {account.id}, platform_user_id: {facebook_user_id}"
)
return FacebookAccountResponse(
success=True,
message="Facebook 계정 연동이 완료되었습니다.",
account_id=account.id,
platform_user_id=facebook_user_id,
platform_username=facebook_user_name,
pages=pages if pages else None,
)
async def disconnect(self, user_uuid: str, session: AsyncSession) -> None:
"""
Facebook 계정 연동 해제
SocialAccount를 소프트 삭제 처리합니다.
Args:
user_uuid: 사용자 UUID
session: DB 세션
Raises:
HTTPException: 연동된 Facebook 계정이 없는 경우
"""
logger.info(f"[FACEBOOK_SVC] 연동 해제 시작 - user_uuid: {user_uuid}")
# 활성 Facebook 계정 조회
result = await session.execute(
select(SocialAccount).where(
SocialAccount.user_uuid == user_uuid,
SocialAccount.platform == Platform.FACEBOOK,
SocialAccount.is_active == True, # noqa: E712
SocialAccount.is_deleted == False, # noqa: E712
)
)
account = result.scalar_one_or_none()
if account is None:
logger.warning(f"[FACEBOOK_SVC] 연동 해제 대상 없음 - user_uuid: {user_uuid}")
from fastapi import HTTPException, status
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"code": "FACEBOOK_ACCOUNT_NOT_FOUND",
"message": "연동된 Facebook 계정을 찾을 수 없습니다.",
},
)
# 소프트 삭제 처리
account.is_active = False
account.is_deleted = True
await session.commit()
logger.info(f"[FACEBOOK_SVC] 연동 해제 완료 - account_id: {account.id}")
# 모듈 레벨 싱글턴 인스턴스
facebook_service = FacebookService()

376
app/utils/facebook_oauth.py Normal file
View File

@ -0,0 +1,376 @@
"""
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

View File

@ -258,6 +258,38 @@ class KakaoSettings(BaseSettings):
model_config = _base_config model_config = _base_config
class FacebookSettings(BaseSettings):
"""Facebook OAuth 설정
Facebook Graph API를 통한 OAuth 2.0 인증 설정입니다.
Meta for Developers (https://developers.facebook.com/)에서 앱을 생성하고
App ID/Secret을 발급받아야 합니다.
"""
FACEBOOK_APP_ID: str = Field(
default="",
description="Facebook App ID (Meta 개발자 콘솔에서 발급)",
)
FACEBOOK_APP_SECRET: str = Field(
default="",
description="Facebook App Secret",
)
FACEBOOK_REDIRECT_URI: str = Field(
default="http://localhost:8000/sns/facebook/callback",
description="Facebook OAuth 콜백 URI",
)
FACEBOOK_GRAPH_API_VERSION: str = Field(
default="v21.0",
description="Facebook Graph API 버전",
)
FACEBOOK_OAUTH_SCOPE: str = Field(
default="public_profile,email,pages_show_list,pages_read_engagement,pages_manage_posts",
description="Facebook OAuth 요청 권한 범위 (쉼표 구분)",
)
model_config = _base_config
class JWTSettings(BaseSettings): class JWTSettings(BaseSettings):
"""JWT 토큰 설정""" """JWT 토큰 설정"""
@ -519,16 +551,6 @@ class SocialOAuthSettings(BaseSettings):
# description="Instagram OAuth 콜백 URI", # description="Instagram OAuth 콜백 URI",
# ) # )
# ============================================================
# Facebook (Meta OAuth) - 추후 구현
# ============================================================
# FACEBOOK_APP_ID: str = Field(default="", description="Facebook App ID")
# FACEBOOK_APP_SECRET: str = Field(default="", description="Facebook App Secret")
# FACEBOOK_REDIRECT_URI: str = Field(
# default="http://localhost:8000/social/facebook/callback",
# description="Facebook OAuth 콜백 URI",
# )
# ============================================================ # ============================================================
# TikTok - 추후 구현 # TikTok - 추후 구현
# ============================================================ # ============================================================
@ -621,6 +643,7 @@ creatomate_settings = CreatomateSettings()
prompt_settings = PromptSettings() prompt_settings = PromptSettings()
log_settings = LogSettings() log_settings = LogSettings()
kakao_settings = KakaoSettings() kakao_settings = KakaoSettings()
facebook_settings = FacebookSettings()
jwt_settings = JWTSettings() jwt_settings = JWTSettings()
recovery_settings = RecoverySettings() recovery_settings = RecoverySettings()
social_oauth_settings = SocialOAuthSettings() social_oauth_settings = SocialOAuthSettings()

207
front_plan.md Normal file
View File

@ -0,0 +1,207 @@
# Facebook 로그인 연동 - 프론트엔드 개발계획서
## 1. 개요
CastAD 서비스에 Facebook 계정 연동 기능을 구현한다. 백엔드에 OAuth 2.0 Authorization Code Flow가 이미 구현되어 있으며, 프론트엔드는 이 플로우의 시작(연동 요청)과 끝(결과 수신)을 담당한다.
기술 스택: React 기반 SPA를 전제로 한다.
## 2. OAuth 2.0 전체 플로우
Facebook OAuth 연동은 총 8단계로 이루어진다. 프론트엔드가 직접 관여하는 단계는 1, 2, 8번이며, 3~7번은 Facebook과 백엔드 사이에서 자동으로 처리된다.
**1단계 (프론트 → 백엔드)**: 프론트엔드가 `GET /sns/facebook/connect`를 호출한다. Authorization 헤더에 사용자의 Bearer 토큰을 포함해야 한다. 백엔드는 `auth_url`(Facebook 인증 페이지 URL)과 `state`(CSRF 방지 토큰)를 JSON으로 응답한다.
**2단계 (프론트 → Facebook)**: 프론트엔드가 응답받은 `auth_url`을 팝업 윈도우로 연다. 사용자는 이 팝업에서 Facebook 로그인과 권한 승인을 수행한다.
**3단계 (Facebook → 백엔드)**: 사용자가 승인을 완료하면 Facebook이 백엔드의 `redirect_uri`(`/sns/facebook/callback`)로 리다이렉트한다. 이때 query parameter로 `code`(인가 코드)와 `state`를 전달한다. 프론트엔드는 이 단계에 관여하지 않는다.
**4~6단계 (백엔드 ↔ Facebook)**: 백엔드가 `code`를 Facebook API에 전달하여 access token을 획득하고, 장기 토큰으로 교환한 뒤, 사용자 정보를 조회하여 DB에 저장한다. 프론트엔드는 이 단계에 관여하지 않는다.
**7단계 (백엔드 → 프론트)**: 백엔드가 처리를 완료한 뒤 302 리다이렉트로 프론트엔드 URL로 보낸다. 성공 시 `/social/connect/success`로, 실패 시 `/social/connect/error`로 리다이렉트하며 결과 정보를 query parameter에 포함한다.
**8단계 (프론트)**: 리다이렉트된 페이지(팝업 내부)에서 query parameter를 파싱하고, `window.opener.postMessage()`로 부모 창에 결과를 전달한 뒤 팝업을 닫는다. 부모 창은 메시지를 수신하여 UI를 업데이트한다.
## 3. 백엔드 API 명세
### 3-1. Facebook 연동 시작
- 엔드포인트: `GET /sns/facebook/connect`
- 인증: 필수. `Authorization: Bearer <access_token>` 헤더를 포함해야 한다.
- 성공 응답 (200): `auth_url` 필드에 Facebook 인증 페이지 URL, `state` 필드에 CSRF 방지 토큰이 포함된 JSON 객체를 반환한다.
```json
{
"auth_url": "https://www.facebook.com/v21.0/dialog/oauth?client_id=...&redirect_uri=...&state=...&scope=...&response_type=code",
"state": "5PcHtcvJdQskHw5aG-MIOmv5CCX3L9v3eqQOqyZlKsc"
}
```
- 에러 응답: 401 (인증 실패, 토큰 없음 또는 만료)
### 3-2. Facebook OAuth 콜백
- 엔드포인트: `GET /sns/facebook/callback`
- 인증: 불필요. 이 엔드포인트는 Facebook이 직접 호출하며, 프론트엔드가 직접 호출하지 않는다.
- query parameter: `code` (Facebook 인가 코드), `state` (CSRF 토큰), `error` (선택, OAuth 에러 코드), `error_description` (선택, 에러 설명)
- 응답: JSON이 아닌 302 리다이렉트로 응답한다. 프론트엔드의 성공 또는 에러 페이지로 리다이렉트한다.
성공 시 리다이렉트 URL: `{OAUTH_FRONTEND_URL}/social/connect/success?platform=facebook&account_id={id}&channel_name={name}`
실패 또는 취소 시 리다이렉트 URL: `{OAUTH_FRONTEND_URL}/social/connect/error?platform=facebook&error={message}&cancelled=true|false`
`OAUTH_FRONTEND_URL`은 백엔드 환경변수로 설정되며, 개발 환경 기본값은 `http://localhost:3000`이다. 프론트엔드 배포 도메인과 일치해야 하므로 백엔드 팀과 사전 확인이 필요하다.
### 3-3. Facebook 연동 해제
- 엔드포인트: `DELETE /sns/facebook/disconnect`
- 인증: 필수. `Authorization: Bearer <access_token>` 헤더를 포함해야 한다.
- 성공 응답 (200): `{"success": true, "message": "Facebook 계정 연동이 해제되었습니다."}`
- 에러 응답: 401 (인증 실패), 404 (연동된 Facebook 계정 없음, 에러 코드 `FACEBOOK_ACCOUNT_NOT_FOUND`)
## 4. 프론트엔드 구현 항목
### 4-1. 라우트 등록
React Router에 다음 2개의 라우트를 추가해야 한다. 이 페이지들은 백엔드 콜백의 302 리다이렉트 대상이므로 반드시 존재해야 한다.
- `/social/connect/success`: OAuth 연동 성공 처리 페이지
- `/social/connect/error`: OAuth 연동 실패/취소 처리 페이지
이 두 페이지는 팝업 윈도우 안에서 렌더링되며, 독립적으로 접근 가능해야 한다(SPA 라우팅이 아닌 직접 URL 접근으로도 동작해야 한다).
### 4-2. Facebook 연동 버튼 컴포넌트
소셜 계정 관리 페이지 또는 설정 페이지에 배치한다.
동작 흐름:
1. 사용자가 "Facebook 연동하기" 버튼을 클릭한다.
2. 버튼을 로딩(비활성) 상태로 전환한다.
3. `GET /sns/facebook/connect`를 호출하여 `auth_url``state`를 받는다.
4. `window.open(auth_url, 'facebook-oauth', 'width=600,height=700,scrollbars=yes')`로 팝업을 연다.
5. `window.open()`의 반환값이 `null`이면 브라우저가 팝업을 차단한 것이다. 이 경우 `window.location.href = auth_url`로 현재 창에서 리다이렉트하는 폴백을 적용한다.
6. 팝업에서 `postMessage`를 수신할 `message` 이벤트 리스너를 등록한다.
팝업 방식을 권장하는 이유: 현재 페이지의 상태(폼 입력값, 스크롤 위치 등)를 유지하면서 OAuth 인증을 진행할 수 있다.
### 4-3. OAuth 성공 페이지 (`/social/connect/success`)
이 페이지는 팝업 윈도우 안에서 렌더링된다.
동작 흐름:
1. `URLSearchParams`로 query parameter를 파싱한다. 사용 가능한 파라미터는 `platform`("facebook"), `account_id`(연동된 계정 ID, 숫자), `channel_name`(Facebook 사용자 이름)이다.
2. `window.opener`가 존재하는지 확인한다(팝업으로 열렸는지 판별).
3. 팝업인 경우: `window.opener.postMessage()`로 부모 창에 성공 결과를 전달하고, `window.close()`로 팝업을 닫는다. postMessage의 `targetOrigin`은 보안을 위해 `window.location.origin`으로 지정한다.
4. 팝업이 아닌 경우(폴백 리다이렉트): 소셜 계정 관리 페이지로 `navigate()`한다.
postMessage 데이터 구조:
```json
{
"type": "FACEBOOK_OAUTH_RESULT",
"success": true,
"platform": "facebook",
"accountId": "연동된 계정 ID",
"channelName": "Facebook 사용자 이름"
}
```
### 4-4. OAuth 에러 페이지 (`/social/connect/error`)
이 페이지도 팝업 윈도우 안에서 렌더링된다.
동작 흐름:
1. `URLSearchParams`로 query parameter를 파싱한다. 사용 가능한 파라미터는 `platform`("facebook"), `error`(에러 메시지 문자열), `cancelled`("true" 또는 "false")이다.
2. `cancelled`가 "true"이면 사용자가 Facebook에서 직접 취소한 것이다. "false"이면 시스템 에러이다.
3. 팝업인 경우: `window.opener.postMessage()`로 부모 창에 에러 결과를 전달하고, `window.close()`로 팝업을 닫는다.
4. 팝업이 아닌 경우: 에러 메시지를 화면에 표시하고 "다시 시도" 버튼과 소셜 관리 페이지로 돌아가는 링크를 제공한다.
postMessage 데이터 구조:
```json
{
"type": "FACEBOOK_OAUTH_RESULT",
"success": false,
"platform": "facebook",
"error": "에러 메시지",
"cancelled": true
}
```
### 4-5. 부모 창 메시지 수신 처리
연동 버튼이 있는 컴포넌트에서 `window.addEventListener('message', handler)`로 메시지를 수신한다.
처리 로직:
1. `event.origin`이 현재 페이지의 origin과 일치하는지 검증한다(보안).
2. `event.data.type``'FACEBOOK_OAUTH_RESULT'`인지 확인한다.
3. `event.data.success``true`이면 연동 성공 UI를 표시한다(토스트 메시지, 연동된 계정 정보 갱신 등).
4. `event.data.success``false`이면 `event.data.cancelled` 여부에 따라 취소 메시지 또는 에러 메시지를 표시한다.
5. 컴포넌트 언마운트 시 이벤트 리스너를 반드시 해제한다.
### 4-6. 연동 해제 기능
동작 흐름:
1. 연동된 계정 정보 옆에 "연동 해제" 버튼을 배치한다.
2. 클릭 시 확인 다이얼로그를 표시한다("Facebook 연동을 해제하시겠습니까?").
3. 확인 시 `DELETE /sns/facebook/disconnect`를 호출한다. Authorization 헤더를 포함해야 한다.
4. 성공 응답(200) 수신 시 UI를 미연동 상태로 업데이트한다.
5. 404 응답 시 이미 연동 해제된 상태이므로 UI를 미연동 상태로 업데이트한다.
### 4-7. 연동 상태 표시
연동 전 상태: Facebook 아이콘과 "연동하기" 버튼을 표시한다.
연동 후 상태: Facebook 아이콘, 연동된 사용자 이름(channel_name), "연동 해제" 버튼을 표시한다.
참고: 현재 백엔드에 사용자의 Facebook 연동 상태를 조회하는 전용 API가 없다. 페이지 진입 시 연동 여부를 표시하려면, 기존 사용자 프로필 API에 소셜 연동 정보가 포함되어 있는지 확인하거나, 백엔드에 상태 조회 API 추가를 요청해야 한다. 연동 직후에는 postMessage로 받은 `accountId``channelName`을 컴포넌트 상태에 저장하여 표시할 수 있다.
## 5. 에러 처리
프론트엔드에서 처리해야 하는 에러 시나리오와 대응 방법은 다음과 같다.
**사용자 취소 (`cancelled=true`)**: 사용자가 Facebook 인증 화면에서 "취소"를 눌렀다. "Facebook 연동이 취소되었습니다." 메시지를 표시한다. 재시도 가능하므로 연동 버튼을 다시 활성화한다.
**state 만료 (`FACEBOOK_STATE_EXPIRED`)**: 사용자가 Facebook 로그인을 너무 오래 지체하여 백엔드의 state 토큰이 만료되었다(기본 TTL: 300초, 약 5분). "인증 세션이 만료되었습니다. 다시 시도해주세요." 메시지를 표시한다.
**토큰 교환 실패 (`FACEBOOK_AUTH_FAILED`)**: Facebook에서 받은 인가 코드로 access token 교환이 실패했다. "Facebook 인증에 실패했습니다. 다시 시도해주세요." 메시지를 표시한다.
**Facebook API 오류 (`FACEBOOK_API_ERROR`)**: Facebook Graph API 호출 중 서버 오류가 발생했다. "일시적인 오류가 발생했습니다. 잠시 후 다시 시도해주세요." 메시지를 표시한다.
**연동 해제 시 계정 없음 (`FACEBOOK_ACCOUNT_NOT_FOUND`)**: 이미 연동 해제된 상태에서 다시 해제를 시도했다. UI를 미연동 상태로 갱신한다.
**인증 실패 (401)**: 서비스 로그인 토큰이 만료되었다. 로그인 페이지로 이동시킨다.
## 6. 폴백 처리 (팝업 차단 시)
브라우저가 팝업을 차단하는 경우 `window.open()``null`을 반환한다. 이때는 현재 창에서 `window.location.href = auth_url`로 직접 이동시키는 폴백을 적용한다.
폴백 모드에서는 OAuth 완료 후 `/social/connect/success` 또는 `/social/connect/error` 페이지로 돌아왔을 때 `window.opener``null`이다. 이 경우 postMessage 대신 다음과 같이 처리한다:
1. query parameter에서 결과를 파싱한다.
2. 결과를 `sessionStorage`에 저장한다(키 예시: `facebook_oauth_result`).
3. 소셜 계정 관리 페이지로 `navigate()`한다.
4. 소셜 계정 관리 페이지 마운트 시 `sessionStorage`에서 결과를 확인하고, 있으면 UI에 반영한 뒤 `sessionStorage`에서 삭제한다.
이 폴백이 필요한 이유: 현재 창 리다이렉트 방식에서는 OAuth 전에 있던 페이지 상태가 사라지므로, 결과를 임시 저장하여 원래 페이지로 복귀 후 처리해야 한다.
## 7. 환경 설정 확인 사항
프론트엔드 개발 시작 전에 백엔드 팀과 다음 값들을 확인해야 한다.
1. `OAUTH_FRONTEND_URL`: 백엔드가 콜백 처리 후 프론트엔드로 302 리다이렉트할 때 사용하는 URL이다. 개발 환경에서는 `http://localhost:3000`, 운영 환경에서는 실제 프론트엔드 도메인이 설정되어야 한다. 이 값이 잘못되면 OAuth 완료 후 엉뚱한 URL로 리다이렉트된다.
2. `OAUTH_SUCCESS_PATH`: 성공 시 리다이렉트 경로. 기본값은 `/social/connect/success`이다.
3. `OAUTH_ERROR_PATH`: 실패 시 리다이렉트 경로. 기본값은 `/social/connect/error`이다.
4. `FACEBOOK_REDIRECT_URI`: Facebook이 인증 완료 후 호출하는 백엔드 콜백 URL이다. 프론트엔드가 변경할 수 없으며, Facebook 개발자 콘솔에 등록된 URI와 정확히 일치해야 한다. 현재 설정값은 `http://dev.castad.net/sns/facebook/callback`이다.
## 8. 구현 순서
다음 순서로 구현을 권장한다.
1. React Router에 `/social/connect/success``/social/connect/error` 라우트를 등록한다.
2. 성공 페이지를 구현한다. query parameter 파싱과 postMessage 전송 로직을 작성한다.
3. 에러 페이지를 구현한다. 취소와 에러를 분기 처리하고 postMessage 전송 로직을 작성한다.
4. Facebook 연동 버튼 컴포넌트를 구현한다. API 호출, 팝업 열기, postMessage 수신을 작성한다.
5. 팝업 차단 시 폴백 로직을 구현한다(sessionStorage 기반).
6. 연동 해제 기능을 구현한다.
7. 연동 상태 표시 UI를 구현한다(상태 조회 API 확인 후).

22
main.py
View File

@ -19,6 +19,7 @@ from app.user.api.routers.v1.social_account import router as social_account_rout
from app.lyric.api.routers.v1.lyric import router as lyric_router from app.lyric.api.routers.v1.lyric import router as lyric_router
from app.song.api.routers.v1.song import router as song_router from app.song.api.routers.v1.song import router as song_router
from app.sns.api.routers.v1.sns import router as sns_router from app.sns.api.routers.v1.sns import router as sns_router
from app.sns.api.routers.v1.oauth import router as sns_oauth_router
from app.video.api.routers.v1.video import router as video_router from app.video.api.routers.v1.video import router as video_router
from app.social.api.routers.v1.oauth import router as social_oauth_router from app.social.api.routers.v1.oauth import router as social_oauth_router
from app.social.api.routers.v1.upload import router as social_upload_router from app.social.api.routers.v1.upload import router as social_upload_router
@ -267,6 +268,25 @@ tags_metadata = [
1. 사용자의 Instagram 계정이 연동되어 있어야 합니다 (Social Account API 참조) 1. 사용자의 Instagram 계정이 연동되어 있어야 합니다 (Social Account API 참조)
2. task_id에 해당하는 비디오가 생성 완료 상태(result_movie_url 존재)여야 합니다 2. task_id에 해당하는 비디오가 생성 완료 상태(result_movie_url 존재)여야 합니다
3. 업로드 성공 Instagram media_id와 permalink 반환 3. 업로드 성공 Instagram media_id와 permalink 반환
""",
},
{
"name": "SNS OAuth",
"description": """SNS OAuth API - Facebook 계정 연동
**인증: 필요** - `Authorization: Bearer {access_token}` 헤더 필수 (콜백 제외)
## Facebook OAuth 연동 흐름
1. `GET /sns/facebook/connect` - Facebook OAuth 인증 URL 획득
2. 사용자를 auth_url로 리다이렉트 Facebook 로그인 권한 승인
3. Facebook에서 `/sns/facebook/callback`으로 인가 코드 전달
4. 서버에서 토큰 교환, 장기 토큰 발급, 사용자 정보 조회
5. 연동 완료 프론트엔드로 리다이렉트
## 계정 관리
- `DELETE /sns/facebook/disconnect` - Facebook 계정 연동 해제
""", """,
}, },
] ]
@ -335,6 +355,7 @@ def custom_openapi():
"/autocomplete", "/autocomplete",
"/search", # 숙박 검색 자동완성 "/search", # 숙박 검색 자동완성
"/social/oauth/youtube/callback", # OAuth 콜백 (플랫폼에서 직접 호출) "/social/oauth/youtube/callback", # OAuth 콜백 (플랫폼에서 직접 호출)
"/sns/facebook/callback", # Facebook OAuth 콜백
] ]
# 보안이 필요한 엔드포인트에 security 적용 # 보안이 필요한 엔드포인트에 security 적용
@ -395,6 +416,7 @@ app.include_router(social_seo_router, prefix="/social") # Social Upload 라우
app.include_router(social_internal_router) # 내부 스케줄러 전용 라우터 app.include_router(social_internal_router) # 내부 스케줄러 전용 라우터
app.include_router(sns_router) # SNS API 라우터 추가 app.include_router(sns_router) # SNS API 라우터 추가
app.include_router(dashboard_router) # Dashboard API 라우터 추가 app.include_router(dashboard_router) # Dashboard API 라우터 추가
app.include_router(sns_oauth_router) # SNS OAuth 라우터 (Facebook)
# DEBUG 모드에서만 테스트 라우터 등록 # DEBUG 모드에서만 테스트 라우터 등록
if prj_settings.DEBUG: if prj_settings.DEBUG:

6
package-lock.json generated Normal file
View File

@ -0,0 +1,6 @@
{
"name": "o2o-castad-backend",
"lockfileVersion": 3,
"requires": true,
"packages": {}
}