169 lines
4.7 KiB
Python
169 lines
4.7 KiB
Python
"""
|
|
Base Social Uploader
|
|
|
|
소셜 미디어 영상 업로더의 추상 기본 클래스입니다.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Optional
|
|
|
|
from app.social.constants import PrivacyStatus, SocialPlatform
|
|
|
|
|
|
@dataclass
|
|
class UploadMetadata:
|
|
"""
|
|
업로드 메타데이터
|
|
|
|
영상 업로드 시 필요한 메타데이터를 정의합니다.
|
|
|
|
Attributes:
|
|
title: 영상 제목
|
|
description: 영상 설명
|
|
tags: 태그 목록
|
|
privacy_status: 공개 상태
|
|
platform_options: 플랫폼별 추가 옵션
|
|
"""
|
|
|
|
title: str
|
|
description: Optional[str] = None
|
|
tags: Optional[list[str]] = None
|
|
privacy_status: PrivacyStatus = PrivacyStatus.PRIVATE
|
|
platform_options: Optional[dict[str, Any]] = None
|
|
|
|
|
|
@dataclass
|
|
class UploadResult:
|
|
"""
|
|
업로드 결과
|
|
|
|
Attributes:
|
|
success: 성공 여부
|
|
platform_video_id: 플랫폼에서 부여한 영상 ID
|
|
platform_url: 플랫폼에서의 영상 URL
|
|
error_message: 에러 메시지 (실패 시)
|
|
platform_response: 플랫폼 원본 응답 (디버깅용)
|
|
"""
|
|
|
|
success: bool
|
|
platform_video_id: Optional[str] = None
|
|
platform_url: Optional[str] = None
|
|
error_message: Optional[str] = None
|
|
platform_response: Optional[dict[str, Any]] = None
|
|
|
|
|
|
class BaseSocialUploader(ABC):
|
|
"""
|
|
소셜 미디어 영상 업로더 추상 기본 클래스
|
|
|
|
모든 플랫폼별 업로더는 이 클래스를 상속받아 구현합니다.
|
|
|
|
Attributes:
|
|
platform: 소셜 플랫폼 종류
|
|
"""
|
|
|
|
platform: SocialPlatform
|
|
|
|
@abstractmethod
|
|
async def upload(
|
|
self,
|
|
video_path: str,
|
|
access_token: str,
|
|
metadata: UploadMetadata,
|
|
progress_callback: Optional[Callable[[int], None]] = None,
|
|
) -> UploadResult:
|
|
"""
|
|
영상 업로드
|
|
|
|
Args:
|
|
video_path: 업로드할 영상 파일 경로 (로컬 또는 URL)
|
|
access_token: OAuth 액세스 토큰
|
|
metadata: 업로드 메타데이터
|
|
progress_callback: 진행률 콜백 함수 (0-100)
|
|
|
|
Returns:
|
|
UploadResult: 업로드 결과
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_upload_status(
|
|
self,
|
|
platform_video_id: str,
|
|
access_token: str,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
업로드 상태 조회
|
|
|
|
플랫폼에서 영상 처리 상태를 조회합니다.
|
|
|
|
Args:
|
|
platform_video_id: 플랫폼 영상 ID
|
|
access_token: OAuth 액세스 토큰
|
|
|
|
Returns:
|
|
dict: 업로드 상태 정보
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def delete_video(
|
|
self,
|
|
platform_video_id: str,
|
|
access_token: str,
|
|
) -> bool:
|
|
"""
|
|
업로드된 영상 삭제
|
|
|
|
Args:
|
|
platform_video_id: 플랫폼 영상 ID
|
|
access_token: OAuth 액세스 토큰
|
|
|
|
Returns:
|
|
bool: 삭제 성공 여부
|
|
"""
|
|
pass
|
|
|
|
def validate_metadata(self, metadata: UploadMetadata) -> None:
|
|
"""
|
|
메타데이터 유효성 검증
|
|
|
|
플랫폼별 제한사항을 확인합니다.
|
|
|
|
Args:
|
|
metadata: 검증할 메타데이터
|
|
|
|
Raises:
|
|
ValueError: 유효하지 않은 메타데이터
|
|
"""
|
|
if not metadata.title or len(metadata.title) == 0:
|
|
raise ValueError("제목은 필수입니다.")
|
|
|
|
if len(metadata.title) > 100:
|
|
raise ValueError("제목은 100자를 초과할 수 없습니다.")
|
|
|
|
if metadata.description and len(metadata.description) > 5000:
|
|
raise ValueError("설명은 5000자를 초과할 수 없습니다.")
|
|
|
|
def get_video_url(self, platform_video_id: str) -> str:
|
|
"""
|
|
플랫폼 영상 URL 생성
|
|
|
|
Args:
|
|
platform_video_id: 플랫폼 영상 ID
|
|
|
|
Returns:
|
|
str: 영상 URL
|
|
"""
|
|
if self.platform == SocialPlatform.YOUTUBE:
|
|
return f"https://www.youtube.com/watch?v={platform_video_id}"
|
|
elif self.platform == SocialPlatform.INSTAGRAM:
|
|
return f"https://www.instagram.com/reel/{platform_video_id}/"
|
|
elif self.platform == SocialPlatform.FACEBOOK:
|
|
return f"https://www.facebook.com/watch/?v={platform_video_id}"
|
|
elif self.platform == SocialPlatform.TIKTOK:
|
|
return f"https://www.tiktok.com/video/{platform_video_id}"
|
|
else:
|
|
return ""
|