o2o-castad-backend/app/social/schemas.py

299 lines
11 KiB
Python

"""
Social Media Schemas
소셜 미디어 연동 관련 Pydantic 스키마를 정의합니다.
"""
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field
from app.social.constants import PrivacyStatus, SocialPlatform, UploadStatus
# =============================================================================
# OAuth 관련 스키마
# =============================================================================
class SocialConnectResponse(BaseModel):
"""소셜 계정 연동 시작 응답"""
auth_url: str = Field(..., description="OAuth 인증 URL")
state: str = Field(..., description="CSRF 방지용 state 토큰")
platform: str = Field(..., description="플랫폼명")
model_config = ConfigDict(
json_schema_extra={
"example": {
"auth_url": "https://accounts.google.com/o/oauth2/v2/auth?...",
"state": "abc123xyz",
"platform": "youtube",
}
}
)
class SocialAccountResponse(BaseModel):
"""연동된 소셜 계정 정보"""
id: int = Field(..., description="소셜 계정 ID")
platform: str = Field(..., description="플랫폼명")
platform_user_id: str = Field(..., description="플랫폼 내 사용자 ID")
platform_username: Optional[str] = Field(None, description="플랫폼 내 사용자명")
display_name: Optional[str] = Field(None, description="표시 이름")
profile_image_url: Optional[str] = Field(None, description="프로필 이미지 URL")
is_active: bool = Field(..., description="활성화 상태")
connected_at: datetime = Field(..., description="연동 일시")
platform_data: Optional[dict[str, Any]] = Field(
None, description="플랫폼별 추가 정보 (채널ID, 구독자 수 등)"
)
model_config = ConfigDict(
from_attributes=True,
json_schema_extra={
"example": {
"id": 1,
"platform": "youtube",
"platform_user_id": "UC1234567890",
"platform_username": "my_channel",
"display_name": "My Channel",
"profile_image_url": "https://...",
"is_active": True,
"connected_at": "2024-01-15T12:00:00",
"platform_data": {
"channel_id": "UC1234567890",
"channel_title": "My Channel",
"subscriber_count": 1000,
},
}
}
)
class SocialAccountListResponse(BaseModel):
"""연동된 소셜 계정 목록 응답"""
accounts: list[SocialAccountResponse] = Field(..., description="연동 계정 목록")
total: int = Field(..., description="총 연동 계정 수")
model_config = ConfigDict(
json_schema_extra={
"example": {
"accounts": [
{
"id": 1,
"platform": "youtube",
"platform_user_id": "UC1234567890",
"platform_username": "my_channel",
"display_name": "My Channel",
"is_active": True,
"connected_at": "2024-01-15T12:00:00",
}
],
"total": 1,
}
}
)
# =============================================================================
# 내부 사용 스키마 (OAuth 토큰 응답)
# =============================================================================
class OAuthTokenResponse(BaseModel):
"""OAuth 토큰 응답 (내부 사용)"""
access_token: str
refresh_token: Optional[str] = None
expires_in: int
token_type: str = "Bearer"
scope: Optional[str] = None
class PlatformUserInfo(BaseModel):
"""플랫폼 사용자 정보 (내부 사용)"""
platform_user_id: str
username: Optional[str] = None
display_name: Optional[str] = None
profile_image_url: Optional[str] = None
platform_data: dict[str, Any] = Field(default_factory=dict)
# =============================================================================
# 업로드 관련 스키마
# =============================================================================
class SocialUploadRequest(BaseModel):
"""소셜 업로드 요청"""
video_id: int = Field(..., description="업로드할 영상 ID")
social_account_id: int = Field(..., description="업로드할 소셜 계정 ID (연동 계정 목록의 id)")
title: str = Field(..., min_length=1, max_length=100, description="영상 제목")
description: Optional[str] = Field(
None, max_length=5000, description="영상 설명"
)
tags: Optional[list[str]] = Field(None, description="태그 목록 (쉼표로 구분된 문자열도 가능)")
privacy_status: PrivacyStatus = Field(
default=PrivacyStatus.PRIVATE, description="공개 상태 (public, unlisted, private)"
)
scheduled_at: Optional[datetime] = Field(
None, description="예약 게시 시간 (없으면 즉시 게시)"
)
platform_options: Optional[dict[str, Any]] = Field(
None, description="플랫폼별 추가 옵션"
)
model_config = ConfigDict(
json_schema_extra={
"example": {
"video_id": 123,
"social_account_id": 1,
"title": "도그앤조이 애견펜션 2026.02.02",
"description": "영상 설명입니다.",
"tags": ["여행", "vlog", "애견펜션"],
"privacy_status": "public",
"scheduled_at": "2026-02-02T15:00:00",
"platform_options": {
"category_id": "22", # YouTube 카테고리
},
}
}
)
class SocialUploadResponse(BaseModel):
"""소셜 업로드 요청 응답"""
success: bool = Field(..., description="요청 성공 여부")
upload_id: int = Field(..., description="업로드 작업 ID")
platform: str = Field(..., description="플랫폼명")
status: str = Field(..., description="업로드 상태")
message: str = Field(..., description="응답 메시지")
model_config = ConfigDict(
json_schema_extra={
"example": {
"success": True,
"upload_id": 456,
"platform": "youtube",
"status": "pending",
"message": "업로드 요청이 접수되었습니다.",
}
}
)
class SocialUploadStatusResponse(BaseModel):
"""업로드 상태 조회 응답"""
upload_id: int = Field(..., description="업로드 작업 ID")
video_id: int = Field(..., description="영상 ID")
social_account_id: int = Field(..., description="소셜 계정 ID")
upload_seq: int = Field(..., description="업로드 순번 (동일 영상+채널 조합 내 순번)")
platform: str = Field(..., description="플랫폼명")
status: UploadStatus = Field(..., description="업로드 상태")
upload_progress: int = Field(..., description="업로드 진행률 (0-100)")
title: str = Field(..., description="영상 제목")
platform_video_id: Optional[str] = Field(None, description="플랫폼 영상 ID")
platform_url: Optional[str] = Field(None, description="플랫폼 영상 URL")
error_message: Optional[str] = Field(None, description="에러 메시지")
retry_count: int = Field(default=0, description="재시도 횟수")
created_at: datetime = Field(..., description="생성 일시")
uploaded_at: Optional[datetime] = Field(None, description="업로드 완료 일시")
model_config = ConfigDict(
from_attributes=True,
json_schema_extra={
"example": {
"upload_id": 456,
"video_id": 123,
"social_account_id": 1,
"upload_seq": 2,
"platform": "youtube",
"status": "completed",
"upload_progress": 100,
"title": "나의 첫 영상",
"platform_video_id": "dQw4w9WgXcQ",
"platform_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"error_message": None,
"retry_count": 0,
"created_at": "2024-01-15T12:00:00",
"uploaded_at": "2024-01-15T12:05:00",
}
}
)
class SocialUploadHistoryItem(BaseModel):
"""업로드 이력 아이템"""
upload_id: int = Field(..., description="업로드 작업 ID")
video_id: int = Field(..., description="영상 ID")
social_account_id: int = Field(..., description="소셜 계정 ID")
upload_seq: int = Field(..., description="업로드 순번 (동일 영상+채널 조합 내 순번)")
platform: str = Field(..., description="플랫폼명")
status: str = Field(..., description="업로드 상태")
title: str = Field(..., description="영상 제목")
platform_url: Optional[str] = Field(None, description="플랫폼 영상 URL")
created_at: datetime = Field(..., description="생성 일시")
uploaded_at: Optional[datetime] = Field(None, description="업로드 완료 일시")
model_config = ConfigDict(from_attributes=True)
class SocialUploadHistoryResponse(BaseModel):
"""업로드 이력 목록 응답"""
items: list[SocialUploadHistoryItem] = Field(..., description="업로드 이력 목록")
total: int = Field(..., description="전체 개수")
page: int = Field(..., description="현재 페이지")
size: int = Field(..., description="페이지 크기")
model_config = ConfigDict(
json_schema_extra={
"example": {
"items": [
{
"upload_id": 456,
"video_id": 123,
"platform": "youtube",
"status": "completed",
"title": "나의 첫 영상",
"platform_url": "https://www.youtube.com/watch?v=xxx",
"created_at": "2024-01-15T12:00:00",
"uploaded_at": "2024-01-15T12:05:00",
}
],
"total": 1,
"page": 1,
"size": 20,
}
}
)
# =============================================================================
# 공통 응답 스키마
# =============================================================================
class MessageResponse(BaseModel):
"""단순 메시지 응답"""
success: bool = Field(..., description="성공 여부")
message: str = Field(..., description="응답 메시지")
model_config = ConfigDict(
json_schema_extra={
"example": {
"success": True,
"message": "작업이 완료되었습니다.",
}
}
)