Compare commits

..

No commits in common. "main" and "refresh" have entirely different histories.

34 changed files with 451 additions and 2384 deletions

View File

@ -73,7 +73,7 @@ async def create_db_tables():
# 모델 import (테이블 메타데이터 등록용) # 모델 import (테이블 메타데이터 등록용)
from app.user.models import User, RefreshToken, SocialAccount # noqa: F401 from app.user.models import User, RefreshToken, SocialAccount # noqa: F401
from app.home.models import Image, Project, MarketingIntel # noqa: F401 from app.home.models import Image, Project # noqa: F401
from app.lyric.models import Lyric # noqa: F401 from app.lyric.models import Lyric # noqa: F401
from app.song.models import Song, SongTimestamp # noqa: F401 from app.song.models import Song, SongTimestamp # noqa: F401
from app.video.models import Video # noqa: F401 from app.video.models import Video # noqa: F401
@ -93,7 +93,6 @@ async def create_db_tables():
Video.__table__, Video.__table__,
SNSUploadTask.__table__, SNSUploadTask.__table__,
SocialUpload.__table__, SocialUpload.__table__,
MarketingIntel.__table__,
] ]
logger.info("Creating database tables...") logger.info("Creating database tables...")
@ -126,9 +125,7 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]:
try: try:
yield session yield session
except Exception as e: except Exception as e:
import traceback
await session.rollback() await session.rollback()
logger.error(traceback.format_exc())
logger.error( logger.error(
f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, " f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms" f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"

View File

@ -11,7 +11,7 @@ from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.database.session import get_session, AsyncSessionLocal from app.database.session import get_session, AsyncSessionLocal
from app.home.models import Image, MarketingIntel from app.home.models import Image
from app.user.dependencies.auth import get_current_user from app.user.dependencies.auth import get_current_user
from app.user.models import User from app.user.models import User
from app.home.schemas.home_schema import ( from app.home.schemas.home_schema import (
@ -153,10 +153,8 @@ def _extract_region_from_address(road_address: str | None) -> str:
}, },
tags=["Crawling"], tags=["Crawling"],
) )
async def crawling( async def crawling(request_body: CrawlingRequest):
request_body: CrawlingRequest, return await _crawling_logic(request_body.url)
session: AsyncSession = Depends(get_session)):
return await _crawling_logic(request_body.url, session)
@router.post( @router.post(
"/autocomplete", "/autocomplete",
@ -189,15 +187,11 @@ async def crawling(
}, },
tags=["Crawling"], tags=["Crawling"],
) )
async def autocomplete_crawling( async def autocomplete_crawling(request_body: AutoCompleteRequest):
request_body: AutoCompleteRequest, url = await _autocomplete_logic(request_body.dict())
session: AsyncSession = Depends(get_session)): return await _crawling_logic(url)
url = await _autocomplete_logic(request_body.model_dump())
return await _crawling_logic(url, session)
async def _crawling_logic( async def _crawling_logic(url:str):
url:str,
session: AsyncSession):
request_start = time.perf_counter() request_start = time.perf_counter()
logger.info("[crawling] ========== START ==========") logger.info("[crawling] ========== START ==========")
logger.info(f"[crawling] URL: {url[:80]}...") logger.info(f"[crawling] URL: {url[:80]}...")
@ -286,16 +280,7 @@ async def _crawling_logic(
step3_3_start = time.perf_counter() step3_3_start = time.perf_counter()
structured_report = await chatgpt_service.generate_structured_output( structured_report = await chatgpt_service.generate_structured_output(
marketing_prompt, input_marketing_data marketing_prompt, input_marketing_data
)
marketing_intelligence = MarketingIntel(
place_id = scraper.place_id,
intel_result = structured_report.model_dump()
) )
session.add(marketing_intelligence)
await session.commit()
await session.refresh(marketing_intelligence)
m_id = marketing_intelligence.id
logger.debug(f"[MarketingPrompt] INSERT placeid {marketing_intelligence.place_id}")
step3_3_elapsed = (time.perf_counter() - step3_3_start) * 1000 step3_3_elapsed = (time.perf_counter() - step3_3_start) * 1000
logger.info( logger.info(
f"[crawling] Step 3-3: GPT API 호출 완료 - ({step3_3_elapsed:.1f}ms)" f"[crawling] Step 3-3: GPT API 호출 완료 - ({step3_3_elapsed:.1f}ms)"
@ -375,7 +360,6 @@ async def _crawling_logic(
"image_count": len(scraper.image_link_list) if scraper.image_link_list else 0, "image_count": len(scraper.image_link_list) if scraper.image_link_list else 0,
"processed_info": processed_info, "processed_info": processed_info,
"marketing_analysis": marketing_analysis, "marketing_analysis": marketing_analysis,
"m_id" : m_id
} }

View File

@ -7,9 +7,9 @@ Home 모듈 SQLAlchemy 모델 정의
""" """
from datetime import datetime from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Any from typing import TYPE_CHECKING, List, Optional
from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, JSON, func from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base from app.database.session import Base
@ -107,12 +107,6 @@ class Project(Base):
comment="상세 지역 정보", comment="상세 지역 정보",
) )
marketing_intelligence: Mapped[Optional[str]] = mapped_column(
Integer,
nullable=True,
comment="마케팅 인텔리전스 결과 정보 저장",
)
language: Mapped[str] = mapped_column( language: Mapped[str] = mapped_column(
String(50), String(50),
nullable=False, nullable=False,
@ -255,66 +249,3 @@ class Image(Base):
return ( return (
f"<Image(id={self.id}, task_id='{task_id_str}', img_name='{img_name_str}')>" f"<Image(id={self.id}, task_id='{task_id_str}', img_name='{img_name_str}')>"
) )
class MarketingIntel(Base):
"""
마케팅 인텔리전스 결과물 테이블
마케팅 분석 결과물 저장합니다.
Attributes:
id: 고유 식별자 (자동 증가)
place_id : 데이터 소스별 식별자
intel_result : 마케팅 분석 결과물 json
created_at: 생성 일시 (자동 설정)
"""
__tablename__ = "marketing"
__table_args__ = (
Index("idx_place_id", "place_id"),
{
"mysql_engine": "InnoDB",
"mysql_charset": "utf8mb4",
"mysql_collate": "utf8mb4_unicode_ci",
}
)
id: Mapped[int] = mapped_column(
Integer,
primary_key=True,
nullable=False,
autoincrement=True,
comment="고유 식별자",
)
place_id: Mapped[str] = mapped_column(
String(36),
nullable=False,
comment="매장 소스별 고유 식별자",
)
intel_result : Mapped[dict[str, Any]] = mapped_column(
JSON,
nullable=False,
comment="마케팅 인텔리전스 결과물",
)
created_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
comment="생성 일시",
)
def __repr__(self) -> str:
task_id_str = (
(self.task_id[:10] + "...") if len(self.task_id) > 10 else self.task_id
)
img_name_str = (
(self.img_name[:10] + "...") if len(self.img_name) > 10 else self.img_name
)
return (
f"<Image(id={self.id}, task_id='{task_id_str}', img_name='{img_name_str}')>"
)

View File

@ -3,6 +3,112 @@ from typing import Literal, Optional
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field
from app.utils.prompts.schemas import MarketingPromptOutput from app.utils.prompts.schemas import MarketingPromptOutput
class AttributeInfo(BaseModel):
"""음악 속성 정보"""
genre: str = Field(..., description="음악 장르")
vocal: str = Field(..., description="보컬 스타일")
tempo: str = Field(..., description="템포")
mood: str = Field(..., description="분위기")
class GenerateRequestImg(BaseModel):
"""이미지 URL 스키마"""
url: str = Field(..., description="이미지 URL")
name: Optional[str] = Field(None, description="이미지명 (없으면 URL에서 추출)")
class GenerateRequestInfo(BaseModel):
"""생성 요청 정보 스키마 (이미지 제외)"""
customer_name: str = Field(..., description="고객명/가게명")
region: str = Field(..., description="지역명")
detail_region_info: Optional[str] = Field(None, description="상세 지역 정보")
attribute: AttributeInfo = Field(..., description="음악 속성 정보")
language: str = Field(
default="Korean",
description="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
)
class GenerateRequest(GenerateRequestInfo):
"""기본 생성 요청 스키마 (이미지 없음, JSON body)
이미지 없이 프로젝트 정보만 전달합니다.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"customer_name": "스테이 머뭄",
"region": "군산",
"detail_region_info": "군산 신흥동 말랭이 마을",
"attribute": {
"genre": "K-Pop",
"vocal": "Raspy",
"tempo": "110 BPM",
"mood": "happy",
},
"language": "Korean",
}
}
)
class GenerateUrlsRequest(GenerateRequestInfo):
"""URL 기반 생성 요청 스키마 (JSON body)
GenerateRequestInfo를 상속받아 이미지 목록을 추가합니다.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"customer_name": "스테이 머뭄",
"region": "군산",
"detail_region_info": "군산 신흥동 말랭이 마을",
"attribute": {
"genre": "K-Pop",
"vocal": "Raspy",
"tempo": "110 BPM",
"mood": "happy",
},
"language": "Korean",
"images": [
{"url": "https://example.com/images/image_001.jpg"},
{"url": "https://example.com/images/image_002.jpg", "name": "외관"},
],
}
}
)
images: list[GenerateRequestImg] = Field(
..., description="이미지 URL 목록", min_length=1
)
class GenerateUploadResponse(BaseModel):
"""파일 업로드 기반 생성 응답 스키마"""
task_id: str = Field(..., description="작업 고유 식별자 (UUID7)")
status: Literal["processing", "completed", "failed"] = Field(
..., description="작업 상태"
)
message: str = Field(..., description="응답 메시지")
uploaded_count: int = Field(..., description="업로드된 이미지 개수")
class GenerateResponse(BaseModel):
"""생성 응답 스키마"""
task_id: str = Field(..., description="작업 고유 식별자 (UUID7)")
status: Literal["processing", "completed", "failed"] = Field(
..., description="작업 상태"
)
message: str = Field(..., description="응답 메시지")
class CrawlingRequest(BaseModel): class CrawlingRequest(BaseModel):
"""크롤링 요청 스키마""" """크롤링 요청 스키마"""
@ -223,8 +329,7 @@ class CrawlingResponse(BaseModel):
"힐링스테이", "힐링스테이",
"스테이머뭄" "스테이머뭄"
] ]
}, }
"m_id" : 1
} }
} }
) )
@ -241,7 +346,6 @@ class CrawlingResponse(BaseModel):
marketing_analysis: Optional[MarketingPromptOutput] = Field( marketing_analysis: Optional[MarketingPromptOutput] = Field(
None, description="마케팅 분석 결과 . 실패 시 null" None, description="마케팅 분석 결과 . 실패 시 null"
) )
m_id : int = Field(..., description="마케팅 분석 결과 ID")
class ErrorResponse(BaseModel): class ErrorResponse(BaseModel):
@ -265,6 +369,29 @@ class ImageUrlItem(BaseModel):
name: Optional[str] = Field(None, description="이미지명 (없으면 URL에서 추출)") name: Optional[str] = Field(None, description="이미지명 (없으면 URL에서 추출)")
class ImageUploadRequest(BaseModel):
"""이미지 업로드 요청 스키마 (JSON body 부분)
URL 이미지 목록을 전달합니다.
바이너리 파일은 multipart/form-data로 별도 전달됩니다.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"images": [
{"url": "https://example.com/images/image_001.jpg"},
{"url": "https://example.com/images/image_002.jpg", "name": "외관"},
]
}
}
)
images: Optional[list[ImageUrlItem]] = Field(
None, description="외부 이미지 URL 목록"
)
class ImageUploadResultItem(BaseModel): class ImageUploadResultItem(BaseModel):
"""업로드된 이미지 결과 아이템""" """업로드된 이미지 결과 아이템"""

View File

@ -30,7 +30,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.database.session import get_session from app.database.session import get_session
from app.home.models import Project, MarketingIntel from app.home.models import Project
from app.user.dependencies.auth import get_current_user from app.user.dependencies.auth import get_current_user
from app.user.models import User from app.user.models import User
from app.lyric.models import Lyric from app.lyric.models import Lyric
@ -48,7 +48,6 @@ from app.utils.pagination import PaginatedResponse, get_paginated
from app.utils.prompts.prompts import lyric_prompt from app.utils.prompts.prompts import lyric_prompt
import traceback as tb import traceback as tb
import json
# 로거 설정 # 로거 설정
logger = get_logger("lyric") logger = get_logger("lyric")
@ -279,19 +278,16 @@ async def generate_lyric(
Full verse flow, immersive mood Full verse flow, immersive mood
""" """
} }
marketing_intel_result = await session.execute(select(MarketingIntel).where(MarketingIntel.id == request_body.m_id))
marketing_intel = marketing_intel_result.scalar_one_or_none()
lyric_input_data = { lyric_input_data = {
"customer_name" : request_body.customer_name, "customer_name" : request_body.customer_name,
"region" : request_body.region, "region" : request_body.region,
"detail_region_info" : request_body.detail_region_info or "", "detail_region_info" : request_body.detail_region_info or "",
"marketing_intelligence_summary" : json.dumps(marketing_intel.intel_result, ensure_ascii = False), "marketing_intelligence_summary" : None, # task_idx 변경 후 marketing intelligence summary DB에 저장하고 사용할 필요가 있음
"language" : request_body.language, "language" : request_body.language,
"promotional_expression_example" : promotional_expressions[request_body.language], "promotional_expression_example" : promotional_expressions[request_body.language],
"timing_rules" : timing_rules["60s"], # 아직은 선택지 하나 "timing_rules" : timing_rules["60s"], # 아직은 선택지 하나
} }
estimated_prompt = lyric_prompt.build_prompt(lyric_input_data)
step1_elapsed = (time.perf_counter() - step1_start) * 1000 step1_elapsed = (time.perf_counter() - step1_start) * 1000
#logger.debug(f"[generate_lyric] Step 1 완료 - 프롬프트 {len(prompt)}자 ({step1_elapsed:.1f}ms)") #logger.debug(f"[generate_lyric] Step 1 완료 - 프롬프트 {len(prompt)}자 ({step1_elapsed:.1f}ms)")
@ -318,7 +314,6 @@ async def generate_lyric(
detail_region_info=request_body.detail_region_info, detail_region_info=request_body.detail_region_info,
language=request_body.language, language=request_body.language,
user_uuid=current_user.user_uuid, user_uuid=current_user.user_uuid,
marketing_intelligence = request_body.m_id
) )
session.add(project) session.add(project)
await session.commit() await session.commit()

View File

@ -41,8 +41,7 @@ class GenerateLyricRequest(BaseModel):
"customer_name": "스테이 머뭄", "customer_name": "스테이 머뭄",
"region": "군산", "region": "군산",
"detail_region_info": "군산 신흥동 말랭이 마을", "detail_region_info": "군산 신흥동 말랭이 마을",
"language": "Korean", "language": "Korean"
"m_id" : 1
} }
""" """
@ -54,7 +53,6 @@ class GenerateLyricRequest(BaseModel):
"region": "군산", "region": "군산",
"detail_region_info": "군산 신흥동 말랭이 마을", "detail_region_info": "군산 신흥동 말랭이 마을",
"language": "Korean", "language": "Korean",
"m_id" : 1
} }
} }
) )
@ -69,7 +67,6 @@ class GenerateLyricRequest(BaseModel):
default="Korean", default="Korean",
description="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)", description="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
) )
m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값")
class GenerateLyricResponse(BaseModel): class GenerateLyricResponse(BaseModel):

View File

@ -4,7 +4,7 @@ SNS API Schemas
Instagram 업로드 관련 Pydantic 스키마를 정의합니다. Instagram 업로드 관련 Pydantic 스키마를 정의합니다.
""" """
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field
@ -98,6 +98,20 @@ class Media(BaseModel):
children: Optional[list["Media"]] = None children: Optional[list["Media"]] = None
class MediaList(BaseModel):
"""미디어 목록 응답"""
data: list[Media] = Field(default_factory=list)
paging: Optional[dict[str, Any]] = None
@property
def next_cursor(self) -> Optional[str]:
"""다음 페이지 커서"""
if self.paging and "cursors" in self.paging:
return self.paging["cursors"].get("after")
return None
class MediaContainer(BaseModel): class MediaContainer(BaseModel):
"""미디어 컨테이너 상태""" """미디어 컨테이너 상태"""

View File

@ -4,5 +4,5 @@ Social API Routers v1
from app.social.api.routers.v1.oauth import router as oauth_router from app.social.api.routers.v1.oauth import router as oauth_router
from app.social.api.routers.v1.upload import router as upload_router from app.social.api.routers.v1.upload import router as upload_router
from app.social.api.routers.v1.seo import router as seo_router
__all__ = ["oauth_router", "upload_router", "seo_router"] __all__ = ["oauth_router", "upload_router"]

View File

@ -1,131 +0,0 @@
import logging, json
from redis.asyncio import Redis
from config import social_oauth_settings, db_settings
from app.social.constants import YOUTUBE_SEO_HASH
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.social.schemas import (
YoutubeDescriptionRequest,
YoutubeDescriptionResponse,
)
from app.database.session import get_session
from app.user.dependencies import get_current_user
from app.user.models import User
from app.home.models import Project, MarketingIntel
from fastapi import APIRouter, BackgroundTasks, Depends, Query
from fastapi import HTTPException, status
from app.utils.prompts.prompts import yt_upload_prompt
from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
redis_seo_client = Redis(
host=db_settings.REDIS_HOST,
port=db_settings.REDIS_PORT,
db=0,
decode_responses=True,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/seo", tags=["Social SEO"])
@router.post(
"/youtube",
response_model=YoutubeDescriptionResponse,
summary="유튜브 SEO descrption 생성",
description="유튜브 업로드 시 사용할 descrption을 SEO 적용하여 생성",
)
async def youtube_seo_description(
request_body: YoutubeDescriptionRequest,
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> YoutubeDescriptionResponse:
# TODO : 나중에 Session Task_id 검증 미들웨어 만들면 추가해주세요.
logger.info(
f"[youtube_seo_description] Try Cache - user: {current_user.user_uuid} / task_id : {request_body.task_id}"
)
cached = await get_yt_seo_in_redis(request_body.task_id)
if cached: # redis hit
return cached
logger.info(
f"[youtube_seo_description] Cache miss - user: {current_user.user_uuid} "
)
updated_seo = await make_youtube_seo_description(request_body.task_id, current_user, session)
await set_yt_seo_in_redis(request_body.task_id, updated_seo)
return updated_seo
async def make_youtube_seo_description(
task_id: str,
current_user: User,
session: AsyncSession,
) -> YoutubeDescriptionResponse:
logger.info(
f"[make_youtube_seo_description] START - user: {current_user.user_uuid} "
)
try:
project_query = await session.execute(
select(Project)
.where(
Project.task_id == task_id,
Project.user_uuid == current_user.user_uuid)
.order_by(Project.created_at.desc())
.limit(1)
)
project = project_query.scalar_one_or_none()
marketing_query = await session.execute(
select(MarketingIntel)
.where(MarketingIntel.id == project.marketing_intelligence)
)
marketing_intelligence = marketing_query.scalar_one_or_none()
hashtags = marketing_intelligence.intel_result["target_keywords"]
yt_seo_input_data = {
"customer_name" : project.store_name,
"detail_region_info" : project.detail_region_info,
"marketing_intelligence_summary" : json.dumps(marketing_intelligence.intel_result, ensure_ascii=False),
"language" : project.language,
"target_keywords" : hashtags
}
chatgpt = ChatgptService()
yt_seo_output = await chatgpt.generate_structured_output(yt_upload_prompt, yt_seo_input_data)
result_dict = {
"title" : yt_seo_output.title,
"description" : yt_seo_output.description,
"keywords": hashtags
}
result = YoutubeDescriptionResponse(**result_dict)
return result
except Exception as e:
logger.error(f"[youtube_seo_description] EXCEPTION - error: {e}")
raise HTTPException(
status_code=500,
detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
)
async def get_yt_seo_in_redis(task_id:str) -> YoutubeDescriptionResponse | None:
field = f"task_id:{task_id}"
yt_seo_info = await redis_seo_client.hget(YOUTUBE_SEO_HASH, field)
if yt_seo_info:
yt_seo = json.loads(yt_seo_info)
else:
return None
return YoutubeDescriptionResponse(**yt_seo)
async def set_yt_seo_in_redis(task_id:str, yt_seo : YoutubeDescriptionResponse) -> None:
field = f"task_id:{task_id}"
yt_seo_info = json.dumps(yt_seo.model_dump(), ensure_ascii=False)
await redis_seo_client.hsetex(YOUTUBE_SEO_HASH, field, yt_seo_info, ex=3600)
return

View File

@ -4,11 +4,10 @@
소셜 미디어 영상 업로드 관련 엔드포인트를 제공합니다. 소셜 미디어 영상 업로드 관련 엔드포인트를 제공합니다.
""" """
import logging, json import logging
from typing import Optional from typing import Optional
from fastapi import APIRouter, BackgroundTasks, Depends, Query from fastapi import APIRouter, BackgroundTasks, Depends, Query
from fastapi import HTTPException, status
from sqlalchemy import select, func from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@ -404,12 +403,16 @@ async def cancel_upload(
upload = result.scalar_one_or_none() upload = result.scalar_one_or_none()
if not upload: if not upload:
from fastapi import HTTPException, status
raise HTTPException( raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, status_code=status.HTTP_404_NOT_FOUND,
detail="업로드 정보를 찾을 수 없습니다.", detail="업로드 정보를 찾을 수 없습니다.",
) )
if upload.status != UploadStatus.PENDING.value: if upload.status != UploadStatus.PENDING.value:
from fastapi import HTTPException, status
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,
detail="대기 중인 업로드만 취소할 수 있습니다.", detail="대기 중인 업로드만 취소할 수 있습니다.",
@ -421,4 +424,4 @@ async def cancel_upload(
return MessageResponse( return MessageResponse(
success=True, success=True,
message="업로드가 취소되었습니다.", message="업로드가 취소되었습니다.",
) )

View File

@ -94,8 +94,6 @@ YOUTUBE_SCOPES = [
"https://www.googleapis.com/auth/userinfo.profile", # 사용자 프로필 "https://www.googleapis.com/auth/userinfo.profile", # 사용자 프로필
] ]
YOUTUBE_SEO_HASH = "SEO_Describtion_YT"
# ============================================================================= # =============================================================================
# Instagram/Facebook OAuth Scopes (추후 구현) # Instagram/Facebook OAuth Scopes (추후 구현)
# ============================================================================= # =============================================================================

View File

@ -276,32 +276,6 @@ class SocialUploadHistoryResponse(BaseModel):
} }
) )
class YoutubeDescriptionRequest(BaseModel):
"""유튜브 SEO Description 제안 (자동완성) Request 모델"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"task_id" : "019c739f-65fc-7d15-8c88-b31be00e588e"
}
}
)
task_id: str = Field(..., description="작업 고유 식별자")
class YoutubeDescriptionResponse(BaseModel):
"""유튜브 SEO Description 제안 (자동완성) Response 모델"""
title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
description : str = Field(..., description="제안된 유튜브 SEO Description")
keywords : list[str] = Field(..., description="해시태그 리스트")
model_config = ConfigDict(
json_schema_extra={
"example": {
"title" : "여기에 더미 타이틀",
"description": "여기에 더미 텍스트",
"keywords": ["여기에", "더미", "해시태그"]
}
}
)
# ============================================================================= # =============================================================================
# 공통 응답 스키마 # 공통 응답 스키마

View File

@ -1,5 +1,8 @@
from typing import Optional from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from fastapi import Request
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@ -104,6 +107,21 @@ class GenerateSongResponse(BaseModel):
error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)") error_message: Optional[str] = Field(None, description="에러 메시지 (실패 시)")
class PollingSongRequest(BaseModel):
"""노래 생성 상태 조회 요청 스키마 (Legacy)
Note:
현재 사용되지 않음. GET /song/status/{song_id} 엔드포인트 사용.
Example Request:
{
"task_id": "abc123..."
}
"""
task_id: str = Field(..., description="Suno 작업 ID")
class SongClipData(BaseModel): class SongClipData(BaseModel):
"""생성된 노래 클립 정보""" """생성된 노래 클립 정보"""
@ -216,3 +234,94 @@ class PollingSongResponse(BaseModel):
song_result_url: Optional[str] = Field( song_result_url: Optional[str] = Field(
None, description="노래 결과 URL (Song 테이블 status가 completed일 때 반환)" None, description="노래 결과 URL (Song 테이블 status가 completed일 때 반환)"
) )
# =============================================================================
# Dataclass Schemas (Legacy)
# =============================================================================
@dataclass
class StoreData:
id: int
created_at: datetime
store_name: str
store_category: str | None = None
store_region: str | None = None
store_address: str | None = None
store_phone_number: str | None = None
store_info: str | None = None
@dataclass
class AttributeData:
id: int
attr_category: str
attr_value: str
created_at: datetime
@dataclass
class SongSampleData:
id: int
ai: str
ai_model: str
sample_song: str
season: str | None = None
num_of_people: int | None = None
people_category: str | None = None
genre: str | None = None
@dataclass
class PromptTemplateData:
id: int
prompt: str
description: str | None = None
@dataclass
class SongFormData:
store_name: str
store_id: str
prompts: str
attributes: Dict[str, str] = field(default_factory=dict)
attributes_str: str = ""
lyrics_ids: List[int] = field(default_factory=list)
llm_model: str = "gpt-5-mini"
@classmethod
async def from_form(cls, request: Request):
"""Request의 form 데이터로부터 dataclass 인스턴스 생성"""
form_data = await request.form()
# 고정 필드명들
fixed_keys = {"store_info_name", "store_id", "llm_model", "prompts"}
# lyrics-{id} 형태의 모든 키를 찾아서 ID 추출
lyrics_ids = []
attributes = {}
for key, value in form_data.items():
if key.startswith("lyrics-"):
lyrics_id = key.split("-")[1]
lyrics_ids.append(int(lyrics_id))
elif key not in fixed_keys:
attributes[key] = value
# attributes를 문자열로 변환
attributes_str = (
"\r\n\r\n".join([f"{key} : {value}" for key, value in attributes.items()])
if attributes
else ""
)
return cls(
store_name=form_data.get("store_info_name", ""),
store_id=form_data.get("store_id", ""),
attributes=attributes,
attributes_str=attributes_str,
lyrics_ids=lyrics_ids,
llm_model=form_data.get("llm_model", "gpt-5-mini"),
prompts=form_data.get("prompts", ""),
)

View File

@ -23,6 +23,7 @@ logger = logging.getLogger(__name__)
from app.user.dependencies import get_current_user from app.user.dependencies import get_current_user
from app.user.models import RefreshToken, User from app.user.models import RefreshToken, User
from app.user.schemas.user_schema import ( from app.user.schemas.user_schema import (
AccessTokenResponse,
KakaoCodeRequest, KakaoCodeRequest,
KakaoLoginResponse, KakaoLoginResponse,
LoginResponse, LoginResponse,
@ -254,16 +255,10 @@ async def refresh_token(
유효한 리프레시 토큰을 제출하면 액세스 토큰과 리프레시 토큰을 발급합니다. 유효한 리프레시 토큰을 제출하면 액세스 토큰과 리프레시 토큰을 발급합니다.
사용된 기존 리프레시 토큰은 즉시 폐기(revoke)됩니다. 사용된 기존 리프레시 토큰은 즉시 폐기(revoke)됩니다.
""" """
logger.info(f"[ROUTER] POST /auth/refresh - token: ...{body.refresh_token[-20:]}") return await auth_service.refresh_tokens(
result = await auth_service.refresh_tokens(
refresh_token=body.refresh_token, refresh_token=body.refresh_token,
session=session, session=session,
) )
logger.info(
f"[ROUTER] POST /auth/refresh 완료 - new_access: ...{result.access_token[-20:]}, "
f"new_refresh: ...{result.refresh_token[-20:]}"
)
return result
@router.post( @router.post(
@ -287,16 +282,11 @@ async def logout(
현재 사용 중인 리프레시 토큰을 폐기합니다. 현재 사용 중인 리프레시 토큰을 폐기합니다.
해당 토큰으로는 이상 액세스 토큰을 갱신할 없습니다. 해당 토큰으로는 이상 액세스 토큰을 갱신할 없습니다.
""" """
logger.info(
f"[ROUTER] POST /auth/logout - user_id: {current_user.id}, "
f"user_uuid: {current_user.user_uuid}, token: ...{body.refresh_token[-20:]}"
)
await auth_service.logout( await auth_service.logout(
user_id=current_user.id, user_id=current_user.id,
refresh_token=body.refresh_token, refresh_token=body.refresh_token,
session=session, session=session,
) )
logger.info(f"[ROUTER] POST /auth/logout 완료 - user_id: {current_user.id}")
return Response(status_code=status.HTTP_204_NO_CONTENT) return Response(status_code=status.HTTP_204_NO_CONTENT)
@ -320,15 +310,10 @@ async def logout_all(
사용자의 모든 리프레시 토큰을 폐기합니다. 사용자의 모든 리프레시 토큰을 폐기합니다.
모든 기기에서 재로그인이 필요합니다. 모든 기기에서 재로그인이 필요합니다.
""" """
logger.info(
f"[ROUTER] POST /auth/logout/all - user_id: {current_user.id}, "
f"user_uuid: {current_user.user_uuid}"
)
await auth_service.logout_all( await auth_service.logout_all(
user_id=current_user.id, user_id=current_user.id,
session=session, session=session,
) )
logger.info(f"[ROUTER] POST /auth/logout/all 완료 - user_id: {current_user.id}")
return Response(status_code=status.HTTP_204_NO_CONTENT) return Response(status_code=status.HTTP_204_NO_CONTENT)

View File

@ -4,7 +4,6 @@
FastAPI 라우터에서 사용할 인증 관련 의존성을 정의합니다. FastAPI 라우터에서 사용할 인증 관련 의존성을 정의합니다.
""" """
import logging
from typing import Optional from typing import Optional
from fastapi import Depends from fastapi import Depends
@ -23,8 +22,6 @@ from app.user.services.auth import (
) )
from app.user.services.jwt import decode_token from app.user.services.jwt import decode_token
logger = logging.getLogger(__name__)
security = HTTPBearer(auto_error=False) security = HTTPBearer(auto_error=False)
@ -50,28 +47,18 @@ async def get_current_user(
UserInactiveError: 비활성화된 계정인 경우 UserInactiveError: 비활성화된 계정인 경우
""" """
if credentials is None: if credentials is None:
logger.info("[AUTH-DEP] 토큰 없음 - MissingTokenError")
raise MissingTokenError() raise MissingTokenError()
token = credentials.credentials payload = decode_token(credentials.credentials)
logger.debug(f"[AUTH-DEP] Access Token 검증 시작 - token: ...{token[-20:]}")
payload = decode_token(token)
if payload is None: if payload is None:
logger.warning(f"[AUTH-DEP] Access Token 디코딩 실패 - token: ...{token[-20:]}")
raise InvalidTokenError() raise InvalidTokenError()
# 토큰 타입 확인 # 토큰 타입 확인
if payload.get("type") != "access": if payload.get("type") != "access":
logger.warning(
f"[AUTH-DEP] 토큰 타입 불일치 - expected: access, "
f"got: {payload.get('type')}, sub: {payload.get('sub')}"
)
raise InvalidTokenError("액세스 토큰이 아닙니다.") raise InvalidTokenError("액세스 토큰이 아닙니다.")
user_uuid = payload.get("sub") user_uuid = payload.get("sub")
if user_uuid is None: if user_uuid is None:
logger.warning(f"[AUTH-DEP] 토큰에 sub 클레임 없음 - token: ...{token[-20:]}")
raise InvalidTokenError() raise InvalidTokenError()
# 사용자 조회 # 사용자 조회
@ -84,18 +71,11 @@ async def get_current_user(
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if user is None: if user is None:
logger.warning(f"[AUTH-DEP] 사용자 미존재 - user_uuid: {user_uuid}")
raise UserNotFoundError() raise UserNotFoundError()
if not user.is_active: if not user.is_active:
logger.warning(
f"[AUTH-DEP] 비활성 사용자 접근 - user_uuid: {user_uuid}, user_id: {user.id}"
)
raise UserInactiveError() raise UserInactiveError()
logger.debug(
f"[AUTH-DEP] Access Token 검증 성공 - user_uuid: {user_uuid}, user_id: {user.id}"
)
return user return user
@ -116,24 +96,17 @@ async def get_current_user_optional(
User | None: 로그인한 사용자 또는 None User | None: 로그인한 사용자 또는 None
""" """
if credentials is None: if credentials is None:
logger.debug("[AUTH-DEP] 선택적 인증 - 토큰 없음")
return None return None
token = credentials.credentials payload = decode_token(credentials.credentials)
payload = decode_token(token)
if payload is None: if payload is None:
logger.debug(f"[AUTH-DEP] 선택적 인증 - 디코딩 실패, token: ...{token[-20:]}")
return None return None
if payload.get("type") != "access": if payload.get("type") != "access":
logger.debug(
f"[AUTH-DEP] 선택적 인증 - 타입 불일치 (type={payload.get('type')})"
)
return None return None
user_uuid = payload.get("sub") user_uuid = payload.get("sub")
if user_uuid is None: if user_uuid is None:
logger.debug("[AUTH-DEP] 선택적 인증 - sub 없음")
return None return None
result = await session.execute( result = await session.execute(
@ -145,14 +118,8 @@ async def get_current_user_optional(
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if user is None or not user.is_active: if user is None or not user.is_active:
logger.debug(
f"[AUTH-DEP] 선택적 인증 - 사용자 미존재 또는 비활성, user_uuid: {user_uuid}"
)
return None return None
logger.debug(
f"[AUTH-DEP] 선택적 인증 성공 - user_uuid: {user_uuid}, user_id: {user.id}"
)
return user return user

View File

@ -1,4 +1,5 @@
from app.user.schemas.user_schema import ( from app.user.schemas.user_schema import (
AccessTokenResponse,
KakaoCodeRequest, KakaoCodeRequest,
KakaoLoginResponse, KakaoLoginResponse,
KakaoTokenResponse, KakaoTokenResponse,
@ -11,6 +12,7 @@ from app.user.schemas.user_schema import (
) )
__all__ = [ __all__ = [
"AccessTokenResponse",
"KakaoCodeRequest", "KakaoCodeRequest",
"KakaoLoginResponse", "KakaoLoginResponse",
"KakaoTokenResponse", "KakaoTokenResponse",

View File

@ -64,6 +64,24 @@ class TokenResponse(BaseModel):
} }
class AccessTokenResponse(BaseModel):
"""액세스 토큰 갱신 응답"""
access_token: str = Field(..., description="액세스 토큰")
token_type: str = Field(default="Bearer", description="토큰 타입")
expires_in: int = Field(..., description="액세스 토큰 만료 시간 (초)")
model_config = {
"json_schema_extra": {
"example": {
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzA1MzE1MjAwfQ.new_token",
"token_type": "Bearer",
"expires_in": 3600
}
}
}
class RefreshTokenRequest(BaseModel): class RefreshTokenRequest(BaseModel):
"""토큰 갱신 요청""" """토큰 갱신 요청"""

View File

@ -81,6 +81,7 @@ class AdminRequiredError(AuthException):
from app.user.models import RefreshToken, User from app.user.models import RefreshToken, User
from app.utils.common import generate_uuid from app.utils.common import generate_uuid
from app.user.schemas.user_schema import ( from app.user.schemas.user_schema import (
AccessTokenResponse,
KakaoUserInfo, KakaoUserInfo,
LoginResponse, LoginResponse,
TokenResponse, TokenResponse,
@ -207,91 +208,49 @@ class AuthService:
TokenExpiredError: 토큰이 만료된 경우 TokenExpiredError: 토큰이 만료된 경우
TokenRevokedError: 토큰이 폐기된 경우 TokenRevokedError: 토큰이 폐기된 경우
""" """
logger.info(f"[AUTH] 토큰 갱신 시작 (Rotation) - token: ...{refresh_token[-20:]}") logger.info("[AUTH] 토큰 갱신 시작 (Refresh Token Rotation)")
# 1. 토큰 디코딩 및 검증 # 1. 토큰 디코딩 및 검증
payload = decode_token(refresh_token) payload = decode_token(refresh_token)
if payload is None: if payload is None:
logger.warning(f"[AUTH] 토큰 갱신 실패 [1/8 디코딩] - token: ...{refresh_token[-20:]}")
raise InvalidTokenError() raise InvalidTokenError()
if payload.get("type") != "refresh": if payload.get("type") != "refresh":
logger.warning(
f"[AUTH] 토큰 갱신 실패 [1/8 타입] - type={payload.get('type')}, "
f"sub: {payload.get('sub')}"
)
raise InvalidTokenError("리프레시 토큰이 아닙니다.") raise InvalidTokenError("리프레시 토큰이 아닙니다.")
logger.debug(
f"[AUTH] 토큰 갱신 [1/8] 디코딩 성공 - sub: {payload.get('sub')}, "
f"exp: {payload.get('exp')}"
)
# 2. DB에서 리프레시 토큰 조회 # 2. DB에서 리프레시 토큰 조회
token_hash = get_token_hash(refresh_token) token_hash = get_token_hash(refresh_token)
db_token = await self._get_refresh_token_by_hash(token_hash, session) db_token = await self._get_refresh_token_by_hash(token_hash, session)
if db_token is None: if db_token is None:
logger.warning(
f"[AUTH] 토큰 갱신 실패 [2/8 DB조회] - DB에 없음, "
f"token_hash: {token_hash[:16]}..."
)
raise InvalidTokenError() raise InvalidTokenError()
logger.debug(
f"[AUTH] 토큰 갱신 [2/8] DB 조회 성공 - token_hash: {token_hash[:16]}..., "
f"user_uuid: {db_token.user_uuid}, is_revoked: {db_token.is_revoked}, "
f"expires_at: {db_token.expires_at}"
)
# 3. 토큰 상태 확인 # 3. 토큰 상태 확인
if db_token.is_revoked: if db_token.is_revoked:
logger.warning(
f"[AUTH] 토큰 갱신 실패 [3/8 폐기됨] - 이미 폐기된 토큰 (replay attack 의심), "
f"token_hash: {token_hash[:16]}..., user_uuid: {db_token.user_uuid}, "
f"revoked_at: {db_token.revoked_at}"
)
raise TokenRevokedError() raise TokenRevokedError()
# 4. 만료 확인
if db_token.expires_at < now().replace(tzinfo=None): if db_token.expires_at < now().replace(tzinfo=None):
logger.info(
f"[AUTH] 토큰 갱신 실패 [4/8 만료] - expires_at: {db_token.expires_at}, "
f"user_uuid: {db_token.user_uuid}"
)
raise TokenExpiredError() raise TokenExpiredError()
# 5. 사용자 확인 # 4. 사용자 확인
user_uuid = payload.get("sub") user_uuid = payload.get("sub")
user = await self._get_user_by_uuid(user_uuid, session) user = await self._get_user_by_uuid(user_uuid, session)
if user is None: if user is None:
logger.warning(
f"[AUTH] 토큰 갱신 실패 [5/8 사용자] - 사용자 미존재, user_uuid: {user_uuid}"
)
raise UserNotFoundError() raise UserNotFoundError()
if not user.is_active: if not user.is_active:
logger.warning(
f"[AUTH] 토큰 갱신 실패 [5/8 비활성] - user_uuid: {user_uuid}, "
f"user_id: {user.id}"
)
raise UserInactiveError() raise UserInactiveError()
# 6. 기존 리프레시 토큰 폐기 (ORM 직접 수정 — _revoke_refresh_token_by_hash는 내부 commit이 있어 사용하지 않음) # 5. 기존 리프레시 토큰 폐기 (ORM 직접 수정 — _revoke_refresh_token_by_hash는 내부 commit이 있어 사용하지 않음)
db_token.is_revoked = True db_token.is_revoked = True
db_token.revoked_at = now().replace(tzinfo=None) db_token.revoked_at = now().replace(tzinfo=None)
logger.debug(f"[AUTH] 토큰 갱신 [6/8] 기존 토큰 폐기 - token_hash: {token_hash[:16]}...")
# 7. 새 토큰 발급 # 6. 새 토큰 발급
new_access_token = create_access_token(user.user_uuid) new_access_token = create_access_token(user.user_uuid)
new_refresh_token = create_refresh_token(user.user_uuid) new_refresh_token = create_refresh_token(user.user_uuid)
logger.debug(
f"[AUTH] 토큰 갱신 [7/8] 새 토큰 발급 - new_access: ...{new_access_token[-20:]}, "
f"new_refresh: ...{new_refresh_token[-20:]}"
)
# 8. 새 리프레시 토큰 DB 저장 (_save_refresh_token은 flush만 수행) # 7. 새 리프레시 토큰 DB 저장 (_save_refresh_token은 flush만 수행)
await self._save_refresh_token( await self._save_refresh_token(
user_id=user.id, user_id=user.id,
user_uuid=user.user_uuid, user_uuid=user.user_uuid,
@ -299,14 +258,10 @@ class AuthService:
session=session, session=session,
) )
# 폐기 + 저장을 하나의 트랜잭션으로 커밋 # 8. 폐기 + 저장을 하나의 트랜잭션으로 커밋
await session.commit() await session.commit()
logger.info( logger.info(f"[AUTH] 토큰 갱신 완료 - user_uuid: {user.user_uuid}")
f"[AUTH] 토큰 갱신 완료 [8/8] - user_uuid: {user.user_uuid}, "
f"user_id: {user.id}, old_hash: {token_hash[:16]}..., "
f"new_refresh: ...{new_refresh_token[-20:]}"
)
return TokenResponse( return TokenResponse(
access_token=new_access_token, access_token=new_access_token,
@ -330,12 +285,7 @@ class AuthService:
session: DB 세션 session: DB 세션
""" """
token_hash = get_token_hash(refresh_token) token_hash = get_token_hash(refresh_token)
logger.info(
f"[AUTH] 로그아웃 - user_id: {user_id}, token_hash: {token_hash[:16]}..., "
f"token: ...{refresh_token[-20:]}"
)
await self._revoke_refresh_token_by_hash(token_hash, session) await self._revoke_refresh_token_by_hash(token_hash, session)
logger.info(f"[AUTH] 로그아웃 완료 - user_id: {user_id}")
async def logout_all( async def logout_all(
self, self,
@ -349,9 +299,7 @@ class AuthService:
user_id: 사용자 ID user_id: 사용자 ID
session: DB 세션 session: DB 세션
""" """
logger.info(f"[AUTH] 전체 로그아웃 - user_id: {user_id}")
await self._revoke_all_user_tokens(user_id, session) await self._revoke_all_user_tokens(user_id, session)
logger.info(f"[AUTH] 전체 로그아웃 완료 - user_id: {user_id}")
async def _get_or_create_user( async def _get_or_create_user(
self, self,
@ -481,11 +429,6 @@ class AuthService:
) )
session.add(refresh_token) session.add(refresh_token)
await session.flush() await session.flush()
logger.debug(
f"[AUTH] Refresh Token DB 저장 - user_uuid: {user_uuid}, "
f"token_hash: {token_hash[:16]}..., expires_at: {expires_at}"
)
return refresh_token return refresh_token
async def _get_refresh_token_by_hash( async def _get_refresh_token_by_hash(

View File

@ -5,18 +5,14 @@ Access Token과 Refresh Token의 생성, 검증, 해시 기능을 제공합니
""" """
import hashlib import hashlib
import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Optional from typing import Optional
from jose import JWTError, jwt from jose import JWTError, jwt
from jose.exceptions import ExpiredSignatureError, JWTClaimsError
from app.utils.timezone import now from app.utils.timezone import now
from config import jwt_settings from config import jwt_settings
logger = logging.getLogger(__name__)
def create_access_token(user_uuid: str) -> str: def create_access_token(user_uuid: str) -> str:
""" """
@ -36,16 +32,11 @@ def create_access_token(user_uuid: str) -> str:
"exp": expire, "exp": expire,
"type": "access", "type": "access",
} }
token = jwt.encode( return jwt.encode(
to_encode, to_encode,
jwt_settings.JWT_SECRET, jwt_settings.JWT_SECRET,
algorithm=jwt_settings.JWT_ALGORITHM, algorithm=jwt_settings.JWT_ALGORITHM,
) )
logger.debug(
f"[JWT] Access Token 발급 - user_uuid: {user_uuid}, "
f"expires: {expire}, token: ...{token[-20:]}"
)
return token
def create_refresh_token(user_uuid: str) -> str: def create_refresh_token(user_uuid: str) -> str:
@ -66,16 +57,11 @@ def create_refresh_token(user_uuid: str) -> str:
"exp": expire, "exp": expire,
"type": "refresh", "type": "refresh",
} }
token = jwt.encode( return jwt.encode(
to_encode, to_encode,
jwt_settings.JWT_SECRET, jwt_settings.JWT_SECRET,
algorithm=jwt_settings.JWT_ALGORITHM, algorithm=jwt_settings.JWT_ALGORITHM,
) )
logger.debug(
f"[JWT] Refresh Token 발급 - user_uuid: {user_uuid}, "
f"expires: {expire}, token: ...{token[-20:]}"
)
return token
def decode_token(token: str) -> Optional[dict]: def decode_token(token: str) -> Optional[dict]:
@ -94,25 +80,8 @@ def decode_token(token: str) -> Optional[dict]:
jwt_settings.JWT_SECRET, jwt_settings.JWT_SECRET,
algorithms=[jwt_settings.JWT_ALGORITHM], algorithms=[jwt_settings.JWT_ALGORITHM],
) )
logger.debug(
f"[JWT] 토큰 디코딩 성공 - type: {payload.get('type')}, "
f"sub: {payload.get('sub')}, exp: {payload.get('exp')}, "
f"token: ...{token[-20:]}"
)
return payload return payload
except ExpiredSignatureError: except JWTError:
logger.info(f"[JWT] 토큰 만료 - token: ...{token[-20:]}")
return None
except JWTClaimsError as e:
logger.warning(
f"[JWT] 클레임 검증 실패 - error: {e}, token: ...{token[-20:]}"
)
return None
except JWTError as e:
logger.warning(
f"[JWT] 토큰 디코딩 실패 - error: {type(e).__name__}: {e}, "
f"token: ...{token[-20:]}"
)
return None return None

View File

@ -82,7 +82,7 @@ class ChatgptService:
self, self,
prompt : Prompt, prompt : Prompt,
input_data : dict, input_data : dict,
) -> BaseModel: ) -> str:
prompt_text = prompt.build_prompt(input_data) prompt_text = prompt.build_prompt(input_data)
logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})") logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")

View File

@ -15,8 +15,7 @@ class NvMapPwScraper():
_context = None _context = None
_win_width = 1280 _win_width = 1280
_win_height = 720 _win_height = 720
_max_retry = 3 _max_retry = 60 # place id timeout threshold seconds
_timeout = 60 # place id timeout threshold seconds
# instance var # instance var
page = None page = None
@ -98,7 +97,7 @@ patchedGetter.toString();''')
async def get_place_id_url(self, selected): async def get_place_id_url(self, selected):
count = 0 count = 0
get_place_id_url_start = time.perf_counter() get_place_id_url_start = time.perf_counter()
while (count <= self._max_retry): while (count <= 1):
title = selected['title'].replace("<b>", "").replace("</b>", "") title = selected['title'].replace("<b>", "").replace("</b>", "")
address = selected.get('roadAddress', selected['address']).replace("<b>", "").replace("</b>", "") address = selected.get('roadAddress', selected['address']).replace("<b>", "").replace("</b>", "")
encoded_query = parse.quote(f"{address} {title}") encoded_query = parse.quote(f"{address} {title}")
@ -107,12 +106,9 @@ patchedGetter.toString();''')
wait_first_start = time.perf_counter() wait_first_start = time.perf_counter()
try: try:
await self.goto_url(url, wait_until="networkidle",timeout = self._timeout*1000) await self.goto_url(url, wait_until="networkidle",timeout = self._max_retry/2*1000)
except: except:
if "/place/" in self.page.url: await self.page.reload(wait_until="networkidle", timeout = self._max_retry/2*1000)
return self.page.url
logger.error(f"[ERROR] Can't Finish networkidle")
wait_first_time = (time.perf_counter() - wait_first_start) * 1000 wait_first_time = (time.perf_counter() - wait_first_start) * 1000
@ -127,11 +123,9 @@ patchedGetter.toString();''')
url = self.page.url.replace("?","?isCorrectAnswer=true&") url = self.page.url.replace("?","?isCorrectAnswer=true&")
try: try:
await self.goto_url(url, wait_until="networkidle",timeout = self._timeout*1000) await self.goto_url(url, wait_until="networkidle",timeout = self._max_retry/2*1000)
except: except:
if "/place/" in self.page.url: await self.page.reload(wait_until="networkidle", timeout = self._max_retry/2*1000)
return self.page.url
logger.error(f"[ERROR] Can't Finish networkidle")
wait_forced_correct_time = (time.perf_counter() - wait_forced_correct_start) * 1000 wait_forced_correct_time = (time.perf_counter() - wait_forced_correct_start) * 1000
logger.debug(f"[DEBUG] Try {count+1} : Wait for forced isCorrectAnswer flag : {wait_forced_correct_time}ms") logger.debug(f"[DEBUG] Try {count+1} : Wait for forced isCorrectAnswer flag : {wait_forced_correct_time}ms")

View File

@ -30,7 +30,7 @@ class NvMapScraper:
GRAPHQL_URL: str = "https://pcmap-api.place.naver.com/graphql" GRAPHQL_URL: str = "https://pcmap-api.place.naver.com/graphql"
REQUEST_TIMEOUT = 120 # 초 REQUEST_TIMEOUT = 120 # 초
data_source_identifier = "nv"
OVERVIEW_QUERY: str = """ OVERVIEW_QUERY: str = """
query getAccommodation($id: String!, $deviceType: String) { query getAccommodation($id: String!, $deviceType: String) {
business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) { business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) {
@ -99,8 +99,6 @@ query getAccommodation($id: String!, $deviceType: String) {
data = await self._call_get_accommodation(place_id) data = await self._call_get_accommodation(place_id)
self.rawdata = data self.rawdata = data
fac_data = await self._get_facility_string(place_id) fac_data = await self._get_facility_string(place_id)
# Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것.
self.place_id = self.data_source_identifier + place_id
self.rawdata["facilities"] = fac_data self.rawdata["facilities"] = fac_data
self.image_link_list = [ self.image_link_list = [
nv_image["origin"] nv_image["origin"]

View File

@ -52,14 +52,6 @@ lyric_prompt = Prompt(
prompt_model = prompt_settings.LYRIC_PROMPT_MODEL prompt_model = prompt_settings.LYRIC_PROMPT_MODEL
) )
yt_upload_prompt = Prompt(
prompt_template_path=os.path.join(prompt_settings.PROMPT_FOLDER_ROOT, prompt_settings.YOUTUBE_PROMPT_FILE_NAME),
prompt_input_class = YTUploadPromptInput,
prompt_output_class = YTUploadPromptOutput,
prompt_model = prompt_settings.YOUTUBE_PROMPT_MODEL
)
def reload_all_prompt(): def reload_all_prompt():
marketing_prompt._reload_prompt() marketing_prompt._reload_prompt()
lyric_prompt._reload_prompt() lyric_prompt._reload_prompt()
yt_upload_prompt._reload_prompt()

View File

@ -1,3 +1,2 @@
from .lyric import LyricPromptInput, LyricPromptOutput from .lyric import LyricPromptInput, LyricPromptOutput
from .marketing import MarketingPromptInput, MarketingPromptOutput from .marketing import MarketingPromptInput, MarketingPromptOutput
from .youtube import YTUploadPromptInput, YTUploadPromptOutput

View File

@ -1,16 +0,0 @@
from pydantic import BaseModel, Field
from typing import List, Optional
# Input 정의
class YTUploadPromptInput(BaseModel):
customer_name : str = Field(..., description = "마케팅 대상 사업체 이름")
detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세")
marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서")
language : str= Field(..., description = "영상 언어")
target_keywords: List[str] = Field(..., description="태그 키워드 리스트")
# Output 정의
class YTUploadPromptOutput(BaseModel):
title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
description: str = Field(..., description = "유튜브 영상 설명 - SEO/AEO 최적화")

View File

@ -4,7 +4,6 @@ You are a content marketing expert, brand strategist, and creative songwriter
specializing in Korean pension / accommodation businesses. specializing in Korean pension / accommodation businesses.
You create lyrics strictly based on Brand & Marketing Intelligence analysis You create lyrics strictly based on Brand & Marketing Intelligence analysis
and optimized for viral short-form video content. and optimized for viral short-form video content.
Marketing Intelligence Report is background reference.
[INPUT] [INPUT]
Business Name: {customer_name} Business Name: {customer_name}

View File

@ -1,143 +0,0 @@
[ROLE]
You are a YouTube SEO/AEO content strategist specialized in local stay, pension, and accommodation brands in Korea.
You create search-optimized, emotionally appealing, and action-driving titles and descriptions based on Brand & Marketing Intelligence.
Your goal is to:
Increase search visibility
Improve click-through rate
Reflect the brands positioning
Trigger emotional interest
Encourage booking or inquiry actions through subtle CTA
[INPUT]
Business Name: {customer_name}
Region Details: {detail_region_info}
Brand & Marketing Intelligence Report: {marketing_intelligence_summary}
Target Keywords: {target_keywords}
Output Language: {language}
[INTERNAL ANALYSIS DO NOT OUTPUT]
Analyze the following from the marketing intelligence:
Core brand concept
Main emotional promise
Primary target persona
Top 23 USP signals
Stay context (date, healing, local trip, etc.)
Search intent behind the target keywords
Main booking trigger
Emotional moment that would make the viewer want to stay
Use these to guide:
Title tone
Opening CTA line
Emotional hook in the first sentences
[TITLE GENERATION RULES]
The title must:
Include the business name or region when natural
Always wrap the business name in quotation marks
Example: “스테이 머뭄”
Include 12 high-intent keywords
Reflect emotional positioning
Suggest a desirable stay moment
Sound like a natural YouTube title, not an advertisement
Length rules:
Hard limit: 100 characters
Target range: 4565 characters
Place primary keyword in the first half
Avoid:
ALL CAPS
Excessive symbols
Price or promotion language
Hard-sell expressions
[DESCRIPTION GENERATION RULES]
Character rules:
Maximum length: 1,000 characters
Critical information must appear within the first 150 characters
Language style rules (mandatory):
Use polite Korean honorific style
Replace “있나요?” with “있으신가요?”
Do not start sentences with “이곳은”
Replace “선택이 됩니다” with “추천 드립니다”
Always wrap the business name in quotation marks
Example: “스테이 머뭄”
Avoid vague location words like “근대거리” alone
Use specific phrasing such as:
“군산 근대역사문화거리 일대”
Structure:
Opening CTA (first line)
Must be a question or gentle suggestion
Must use honorific tone
Example:
“조용히 쉴 수 있는 군산숙소를 찾고 있으신가요?”
Core Stay Introduction (within first 150 characters total)
Mention business name with quotation marks
Mention region
Include main keyword
Briefly describe the stay experience
Brand Experience
Core value and emotional promise
Based on marketing intelligence positioning
Key Highlights (34 short lines)
Derived from USP signals
Natural sentences
Focus on booking-trigger moments
Local Context
Mention nearby experiences
Use specific local references
Example:
“군산 근대역사문화거리 일대 산책이나 로컬 카페 투어”
Soft Closing Line
One gentle, non-salesy closing sentence
Must end with a recommendation tone
Example:
“군산에서 조용한 시간을 보내고 싶다면 ‘스테이 머뭄’을 추천 드립니다.”
[SEO & AEO RULES]
Naturally integrate 35 keywords from {target_keywords}
Avoid keyword stuffing
Use conversational, search-like phrasing
Optimize for:
YouTube search
Google video results
AI answer summaries
Keywords should appear in:
Title (12)
First 150 characters of description
Highlight or context sections
[LANGUAGE RULE]
All output must be written entirely in {language}.
No mixed languages.
[OUTPUT FORMAT STRICT]
title:
description:
No explanations.
No headings.
No extra text.

View File

@ -7,6 +7,7 @@ Video API Router
- POST /video/generate/{task_id}: 영상 생성 요청 (task_id로 Project/Lyric/Song 연결) - POST /video/generate/{task_id}: 영상 생성 요청 (task_id로 Project/Lyric/Song 연결)
- GET /video/status/{creatomate_render_id}: Creatomate API 영상 생성 상태 조회 - GET /video/status/{creatomate_render_id}: Creatomate API 영상 생성 상태 조회
- GET /video/download/{task_id}: 영상 다운로드 상태 조회 (DB polling) - GET /video/download/{task_id}: 영상 다운로드 상태 조회 (DB polling)
- GET /video/list: 완료된 영상 목록 조회 (페이지네이션)
사용 예시: 사용 예시:
from app.video.api.routers.v1.video import router from app.video.api.routers.v1.video import router
@ -17,10 +18,11 @@ import json
from typing import Literal from typing import Literal
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
from sqlalchemy import select from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.database.session import get_session from app.database.session import get_session
from app.dependencies.pagination import PaginationParams, get_pagination_params
from app.user.dependencies.auth import get_current_user from app.user.dependencies.auth import get_current_user
from app.user.models import User from app.user.models import User
from app.home.models import Image, Project from app.home.models import Image, Project
@ -28,11 +30,13 @@ from app.lyric.models import Lyric
from app.song.models import Song, SongTimestamp from app.song.models import Song, SongTimestamp
from app.utils.creatomate import CreatomateService from app.utils.creatomate import CreatomateService
from app.utils.logger import get_logger from app.utils.logger import get_logger
from app.utils.pagination import PaginatedResponse
from app.video.models import Video from app.video.models import Video
from app.video.schemas.video_schema import ( from app.video.schemas.video_schema import (
DownloadVideoResponse, DownloadVideoResponse,
GenerateVideoResponse, GenerateVideoResponse,
PollingVideoResponse, PollingVideoResponse,
VideoListItem,
VideoRenderData, VideoRenderData,
) )
from app.video.worker.video_task import download_and_upload_video_to_blob from app.video.worker.video_task import download_and_upload_video_to_blob
@ -734,3 +738,126 @@ async def download_video(
message="영상 다운로드 조회에 실패했습니다.", message="영상 다운로드 조회에 실패했습니다.",
error_message=str(e), error_message=str(e),
) )
@router.get(
"/list",
summary="생성된 영상 목록 조회",
description="""
완료된 영상 목록을 페이지네이션하여 조회합니다.
## 인증
**Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다.
## 쿼리 파라미터
- **page**: 페이지 번호 (1부터 시작, 기본값: 1)
- **page_size**: 페이지당 데이터 (기본값: 10, 최대: 100)
## 반환 정보
- **items**: 영상 목록 (store_name, region, task_id, result_movie_url, created_at)
- **total**: 전체 데이터
- **page**: 현재 페이지
- **page_size**: 페이지당 데이터
- **total_pages**: 전체 페이지
- **has_next**: 다음 페이지 존재 여부
- **has_prev**: 이전 페이지 존재 여부
## 사용 예시 (cURL)
```bash
curl -X GET "http://localhost:8000/video/list?page=1&page_size=10" \\
-H "Authorization: Bearer {access_token}"
```
## 참고
- status가 'completed' 영상만 반환됩니다.
- 동일한 task_id가 있는 경우 가장 최근에 생성된 1개만 반환됩니다.
- created_at 기준 내림차순 정렬됩니다.
""",
response_model=PaginatedResponse[VideoListItem],
responses={
200: {"description": "영상 목록 조회 성공"},
401: {"description": "인증 실패 (토큰 없음/만료)"},
500: {"description": "조회 실패"},
},
)
async def get_videos(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
pagination: PaginationParams = Depends(get_pagination_params),
) -> PaginatedResponse[VideoListItem]:
"""완료된 영상 목록을 페이지네이션하여 반환합니다."""
logger.info(
f"[get_videos] START - page: {pagination.page}, page_size: {pagination.page_size}"
)
try:
offset = (pagination.page - 1) * pagination.page_size
# 서브쿼리: task_id별 최신 Video의 id 조회 (completed 상태만)
subquery = (
select(func.max(Video.id).label("max_id"))
.where(Video.status == "completed")
.group_by(Video.task_id)
.subquery()
)
# 전체 개수 조회 (task_id별 최신 1개만)
count_query = select(func.count()).select_from(subquery)
total_result = await session.execute(count_query)
total = total_result.scalar() or 0
# 데이터 조회 (completed 상태, task_id별 최신 1개만, 최신순)
query = (
select(Video)
.where(Video.id.in_(select(subquery.c.max_id)))
.order_by(Video.created_at.desc())
.offset(offset)
.limit(pagination.page_size)
)
result = await session.execute(query)
videos = result.scalars().all()
# Project 정보 일괄 조회 (N+1 문제 해결)
project_ids = [v.project_id for v in videos if v.project_id]
projects_map: dict = {}
if project_ids:
projects_result = await session.execute(
select(Project).where(Project.id.in_(project_ids))
)
projects_map = {p.id: p for p in projects_result.scalars().all()}
# VideoListItem으로 변환
items = []
for video in videos:
project = projects_map.get(video.project_id)
item = VideoListItem(
video_id=video.id,
store_name=project.store_name if project else None,
region=project.region if project else None,
task_id=video.task_id,
result_movie_url=video.result_movie_url,
created_at=video.created_at,
)
items.append(item)
response = PaginatedResponse.create(
items=items,
total=total,
page=pagination.page,
page_size=pagination.page_size,
)
logger.info(
f"[get_videos] SUCCESS - total: {total}, page: {pagination.page}, "
f"page_size: {pagination.page_size}, items_count: {len(items)}"
)
return response
except Exception as e:
logger.error(f"[get_videos] EXCEPTION - error: {e}")
raise HTTPException(
status_code=500,
detail=f"영상 목록 조회에 실패했습니다: {str(e)}",
)

View File

@ -184,8 +184,6 @@ class PromptSettings(BaseSettings):
LYRIC_PROMPT_FILE_NAME : str = Field(default="lyric_prompt.txt") LYRIC_PROMPT_FILE_NAME : str = Field(default="lyric_prompt.txt")
LYRIC_PROMPT_MODEL : str = Field(default="gpt-5-mini") LYRIC_PROMPT_MODEL : str = Field(default="gpt-5-mini")
YOUTUBE_PROMPT_FILE_NAME : str = Field(default="yt_upload_prompt.txt")
YOUTUBE_PROMPT_MODEL : str = Field(default="gpt-5-mini")
model_config = _base_config model_config = _base_config
@ -600,9 +598,9 @@ class SocialUploadSettings(BaseSettings):
prj_settings = ProjectSettings() prj_settings = ProjectSettings()
cors_settings = CORSSettings()
apikey_settings = APIKeySettings() apikey_settings = APIKeySettings()
db_settings = DatabaseSettings() db_settings = DatabaseSettings()
cors_settings = CORSSettings()
crawler_settings = CrawlerSettings() crawler_settings = CrawlerSettings()
naver_api_settings = NaverAPISettings() naver_api_settings = NaverAPISettings()
azure_blob_settings = AzureBlobSettings() azure_blob_settings = AzureBlobSettings()

View File

@ -1,788 +0,0 @@
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>O2O CastAD Backend - 인프라 아키텍처</title>
<script src="https://cdn.jsdelivr.net/npm/mermaid@11/dist/mermaid.min.js"></script>
<style>
:root {
--bg: #0f1117;
--surface: #1a1d27;
--surface2: #232733;
--border: #2e3345;
--text: #e1e4ed;
--text-dim: #8b90a0;
--accent: #6c8cff;
--accent2: #a78bfa;
--green: #34d399;
--orange: #fb923c;
--red: #f87171;
}
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: 'Pretendard', -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
background: var(--bg);
color: var(--text);
line-height: 1.8;
}
/* 네비게이션 */
nav {
position: fixed;
top: 0;
width: 100%;
background: var(--surface);
border-bottom: 1px solid var(--border);
z-index: 100;
display: flex;
align-items: center;
padding: 0 40px;
height: 60px;
gap: 40px;
}
nav .logo {
font-size: 1.1rem;
font-weight: 700;
color: var(--accent);
white-space: nowrap;
}
nav a {
color: var(--text-dim);
text-decoration: none;
font-size: 0.9rem;
transition: color 0.2s;
}
nav a:hover { color: var(--accent); }
/* 메인 */
main {
max-width: 1400px;
margin: 0 auto;
padding: 100px 32px 80px;
}
.hero {
text-align: center;
margin-bottom: 60px;
}
.hero h1 {
font-size: 2.2rem;
margin-bottom: 8px;
background: linear-gradient(135deg, var(--accent), var(--accent2));
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
.hero p {
color: var(--text-dim);
font-size: 1rem;
}
/* 섹션 */
section {
background: var(--surface);
border: 1px solid var(--border);
border-radius: 12px;
padding: 36px;
margin-bottom: 40px;
}
.section-header {
margin-bottom: 24px;
}
.section-header h2 {
font-size: 1.4rem;
margin-bottom: 4px;
color: var(--accent);
}
.section-header h2 .num {
display: inline-block;
background: var(--accent);
color: #fff;
width: 28px;
height: 28px;
border-radius: 50%;
text-align: center;
line-height: 28px;
font-size: 0.85rem;
margin-right: 8px;
}
.section-header .desc {
color: var(--text-dim);
font-size: 0.95rem;
}
/* 좌우 2컬럼 (테이블/텍스트용) */
.cols {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 28px;
align-items: start;
}
.col { min-width: 0; }
/* 다이어그램 - 항상 풀 와이드, 아래 배치 */
.diagram-box {
background: var(--surface2);
border: 1px solid var(--border);
border-radius: 10px;
padding: 32px;
margin-top: 28px;
overflow-x: auto;
}
.diagram-box .mermaid {
display: flex;
justify-content: center;
}
.diagram-box .mermaid svg {
max-width: 100%;
height: auto;
min-height: 300px;
}
.diagram-label {
text-align: center;
font-size: 0.85rem;
color: var(--text-dim);
margin-top: 12px;
}
/* 서브 타이틀 */
h3 {
font-size: 1rem;
color: var(--text);
margin: 20px 0 10px;
}
h3:first-child { margin-top: 0; }
/* 테이블 */
table {
width: 100%;
border-collapse: collapse;
margin: 12px 0;
font-size: 0.88rem;
}
th, td {
padding: 9px 12px;
text-align: left;
border-bottom: 1px solid var(--border);
}
th {
background: var(--surface2);
color: var(--accent);
font-weight: 600;
font-size: 0.82rem;
text-transform: uppercase;
white-space: nowrap;
}
td { color: var(--text-dim); }
/* 태그 */
.tag {
display: inline-block;
padding: 2px 8px;
border-radius: 12px;
font-size: 0.75rem;
font-weight: 600;
}
.tag-green { background: rgba(52,211,153,0.15); color: var(--green); }
.tag-orange { background: rgba(251,146,60,0.15); color: var(--orange); }
.tag-red { background: rgba(248,113,113,0.15); color: var(--red); }
/* 비용 카드 */
.cost-cards {
display: grid;
grid-template-columns: repeat(3, 1fr);
gap: 16px;
margin: 16px 0 24px;
}
.cost-card {
background: var(--surface2);
border: 1px solid var(--border);
border-radius: 8px;
padding: 20px;
text-align: center;
}
.cost-card .stage {
font-size: 0.85rem;
color: var(--text-dim);
margin-bottom: 4px;
}
.cost-card .amount {
font-size: 1.4rem;
font-weight: 700;
}
.cost-card .krw {
font-size: 0.85rem;
color: var(--text-dim);
margin-top: 2px;
}
.cost-card.s1 { border-top: 3px solid var(--green); }
.cost-card.s1 .amount { color: var(--green); }
.cost-card.s2 { border-top: 3px solid var(--orange); }
.cost-card.s2 .amount { color: var(--orange); }
.cost-card.s3 { border-top: 3px solid var(--red); }
.cost-card.s3 .amount { color: var(--red); }
/* 리스트 */
ul {
padding-left: 18px;
margin: 10px 0;
}
ul li {
color: var(--text-dim);
margin-bottom: 5px;
font-size: 0.92rem;
}
ul li strong { color: var(--text); }
/* 노트 */
.note {
background: rgba(108,140,255,0.08);
border-left: 3px solid var(--accent);
padding: 12px 16px;
border-radius: 0 8px 8px 0;
font-size: 0.88rem;
color: var(--text-dim);
margin-top: 20px;
}
code {
background: var(--surface2);
border: 1px solid var(--border);
padding: 1px 5px;
border-radius: 4px;
font-size: 0.84rem;
color: var(--green);
}
/* PDF 다운로드 버튼 */
.btn-pdf {
margin-left: auto;
padding: 6px 16px;
background: var(--accent);
color: #fff;
border: none;
border-radius: 6px;
font-size: 0.85rem;
font-weight: 600;
cursor: pointer;
transition: background 0.2s;
white-space: nowrap;
}
.btn-pdf:hover { background: #5a7ae6; }
/* 반응형 */
@media (max-width: 1024px) {
main { padding: 80px 20px 60px; }
.cols { grid-template-columns: 1fr; }
.cost-cards { grid-template-columns: 1fr; }
}
@media (max-width: 640px) {
nav { padding: 0 16px; gap: 16px; }
nav a { font-size: 0.8rem; }
section { padding: 20px; }
.hero h1 { font-size: 1.6rem; }
table { font-size: 0.8rem; }
th, td { padding: 6px 8px; }
.diagram-box { padding: 16px; }
}
/* 인쇄 / PDF 저장 — 화면과 동일한 다크 테마 유지 */
@media print {
@page {
size: A3 landscape;
margin: 10mm;
}
* {
-webkit-print-color-adjust: exact !important;
print-color-adjust: exact !important;
}
nav { display: none !important; }
.btn-pdf { display: none !important; }
main {
max-width: 100% !important;
padding: 10px !important;
}
section {
break-inside: auto;
page-break-inside: auto;
}
.diagram-box {
break-inside: auto;
page-break-inside: auto;
}
.hero {
margin-bottom: 20px;
}
section {
margin-bottom: 20px;
padding: 20px;
}
.cols {
grid-template-columns: 1fr 1fr !important;
}
.cost-cards {
grid-template-columns: repeat(3, 1fr) !important;
}
}
</style>
</head>
<body>
<nav>
<div class="logo">O2O CastAD</div>
<a href="#load-balancing">부하 분산</a>
<a href="#architecture">아키텍처</a>
<a href="#cost">비용 산출</a>
<button class="btn-pdf" onclick="downloadPDF()">PDF 다운로드</button>
</nav>
<main>
<div class="hero">
<h1>O2O CastAD Backend</h1>
<p>인프라 아키텍처 및 비용 산출 문서</p>
</div>
<!-- ==================== 1. 부하 분산 ==================== -->
<section id="load-balancing">
<div class="section-header">
<h2><span class="num">1</span>DB 및 서버 부하 분산 방법</h2>
<p class="desc">Nginx 로드밸런싱, 커넥션 풀 관리, 단계별 수평 확장 전략</p>
</div>
<div class="cols">
<div class="col">
<h3>현재 구현 현황 (단일 인스턴스)</h3>
<ul>
<li><strong>API 커넥션 풀</strong>: pool_size=20, max_overflow=20 → 최대 <code>40</code></li>
<li><strong>백그라운드 풀</strong>: pool_size=10, max_overflow=10 → 최대 <code>20</code></li>
<li><strong>인스턴스당 총 DB 연결</strong>: <code>40 + 20 = 60</code></li>
<li><strong>풀 리사이클</strong>: 280초 (MySQL wait_timeout 300초 이전), pre-ping 활성화</li>
</ul>
<h3>단계별 확장 전략</h3>
<table>
<thead>
<tr>
<th>단계</th>
<th>동시접속</th>
<th>App Server</th>
<th>LB</th>
<th>DB ( MySQL Flexible)</th>
</tr>
</thead>
<tbody>
<tr>
<td><span class="tag tag-green">S1</span></td>
<td>~50명</td>
<td>x1</td>
<td>Nginx x1</td>
<td>Burstable B1ms</td>
</tr>
<tr>
<td><span class="tag tag-orange">S2</span></td>
<td>50~200명</td>
<td>x2~4</td>
<td>Nginx</td>
<td>GP D2ds_v4 + Replica x1</td>
</tr>
<tr>
<td><span class="tag tag-red">S3</span></td>
<td>200~1,000명</td>
<td><span style="font-size:0.75rem; line-height:1.4;">API ServerxN<br/>+ Scheduler</span></td>
<td>Nginx</td>
<td>BC D4ds_v4 + Replica x2 + Redis P1</td>
</tr>
</tbody>
</table>
</div>
<div class="col">
<h3>커넥션 풀 수치 계산</h3>
<table>
<thead>
<tr>
<th>항목</th>
<th>Stage 1 (1대)</th>
<th>Stage 2 (4대)</th>
<th>Stage 3 (8대)</th>
</tr>
</thead>
<tbody>
<tr>
<td><strong>Main Pool / 인스턴스</strong></td>
<td>20+20 = 40</td>
<td>10+10 = 20</td>
<td>5+5 = 10</td>
</tr>
<tr>
<td><strong>BG Pool / 인스턴스</strong></td>
<td>10+10 = 20</td>
<td>5+5 = 10</td>
<td>3+3 = 6</td>
</tr>
<tr>
<td><strong>인스턴스당 소계</strong></td>
<td><code>60</code></td>
<td><code>30</code></td>
<td><code>16</code></td>
</tr>
<tr>
<td><strong>Primary 총 연결</strong></td>
<td>60</td>
<td>4 x 30 = <code>120</code></td>
<td>8 x 16 = <code>128</code></td>
</tr>
<tr>
<td><strong>max_connections 권장</strong></td>
<td>100</td>
<td>200</td>
<td>300</td>
</tr>
</tbody>
</table>
<div class="note" style="margin-top: 16px;">
<strong>핵심:</strong>
JWT Stateless 설계로 Nginx 세션 어피니티 불필요 (round-robin / least_conn).
Stage 2부터 Read Replica로 읽기 분산, Redis는 Stage 3에서 캐싱/Rate Limiting 도입.
</div>
</div>
</div>
<!-- 다이어그램: 내용 아래 풀 와이드 -->
<div class="diagram-box">
<pre class="mermaid">
graph TB
subgraph S1["Stage 1: ~50명"]
direction LR
S1N["Nginx<br/>(Reverse Proxy)"] --> S1A["App Server x1"]
S1A --> S1D[" MySQL<br/>Burstable B1ms"]
end
subgraph S2["Stage 2: 50~200명"]
direction LR
S2N["Nginx<br/>(Reverse Proxy)"] --> S2API["APP Server<br/>x 1 ~ 2"]
S2N --> S2WK["Scheduler<br/>Server"]
S2API --> S2P["MySQL BC<br/>Primary<br/>(D4ds_v4)"]
S2API --> S2R1["Read Replica<br/>x1"]
S2WK --> S2P
S2WK --> S2R1
end
subgraph S3["Stage 3: 200~1,000명"]
direction LR
S3N["Nginx<br/>(Reverse Proxy)"] --> S3API["APP Server<br/>x N"]
S3N --> S3WK["Scheduler<br/>Server"]
S3API --> S3P["MySQL BC<br/>Primary<br/>(D4ds_v4)"]
S3API --> S3R1["Read Replica<br/>xN"]
S3API --> S3RD["Redis<br/>Premium P1"]
S3WK --> S3P
S3WK --> S3R1
end
S1 ~~~ S2 ~~~ S3
style S1 fill:#0d3320,stroke:#34d399,stroke-width:2px,color:#e1e4ed
style S2 fill:#3b2506,stroke:#fb923c,stroke-width:2px,color:#e1e4ed
style S3 fill:#3b1010,stroke:#f87171,stroke-width:2px,color:#e1e4ed
</pre>
</div>
</section>
<!-- ==================== 2. 아키텍처 ==================== -->
<section id="architecture">
<div class="section-header">
<h2><span class="num">2</span>전체 아키텍처 다이어그램</h2>
<p class="desc">Nginx + FastAPI App Server 구성과 외부 서비스 연동 구조</p>
</div>
<div class="cols">
<div class="col">
<ul>
<li><strong>로드밸런서</strong>: Nginx (Reverse Proxy, L7 LB, SSL 종단)</li>
<li><strong>App Server</strong>: FastAPI (Python 3.13) — Auth, Home, Lyric, Song, Video, Social, SNS, Archive, Admin, Background Worker</li>
<li><strong>DB</strong>: Database for MySQL Flexible Server — Stage 2+ Read Replica</li>
</ul>
</div>
<div class="col">
<ul>
<li><strong>캐시</strong>: Cache for Redis (Stage 3 도입)</li>
<li><strong>콘텐츠 생성</strong>: 가사(ChatGPT) → 음악(Suno AI) → 영상(Creatomate) → SNS 업로드</li>
<li><strong>외부 연동</strong>: Kakao OAuth, Naver Map/Search API, Blob Storage</li>
</ul>
</div>
</div>
<!-- 다이어그램: 내용 아래 풀 와이드 -->
<div class="diagram-box">
<pre class="mermaid">
graph TB
Client["클라이언트<br/>(Web / App)"]
LB["Nginx<br/>(Reverse Proxy + SSL 종단)"]
subgraph APP["App Server (FastAPI)"]
direction LR
Auth["Auth"] --- Home["Home"] --- Lyric["Lyric"] --- Song["Song"] --- Video["Video"]
Social["Social"] --- SNS["SNS"] --- Archive["Archive"] --- Admin["Admin"] --- BG["BG Worker"]
end
subgraph DB[" MySQL Flexible Server"]
direction LR
Primary["Primary (R/W)"]
Replica["Read Replica"]
end
subgraph AI["AI 콘텐츠 생성 파이프라인"]
direction LR
ChatGPT["ChatGPT<br/>(가사 생성)"]
Suno["Suno AI<br/>(음악 생성)"]
Creatomate["Creatomate<br/>(영상 생성)"]
ChatGPT --> Suno --> Creatomate
end
subgraph EXT["외부 서비스"]
direction LR
Blob[" Blob<br/>Storage"]
Kakao["Kakao<br/>OAuth"]
YT["YouTube /<br/>Instagram"]
Naver["Naver Map /<br/>Search API"]
end
Redis[" Cache for Redis<br/>(Stage 3 도입)"]
Client -->|HTTPS| LB
LB --> APP
APP --> Primary
APP -->|"읽기 전용"| Replica
APP -.->|"Stage 3"| Redis
APP --> AI
APP --> Blob
APP --> Kakao
APP --> YT
APP --> Naver
style Client fill:#1a3a1a,stroke:#34d399,stroke-width:2px,color:#e1e4ed
style LB fill:#1a3a1a,stroke:#34d399,stroke-width:2px,color:#e1e4ed
style APP fill:#1a2744,stroke:#6c8cff,stroke-width:2px,color:#e1e4ed
style DB fill:#2a1f00,stroke:#fb923c,stroke-width:2px,color:#e1e4ed
style AI fill:#2a0f2a,stroke:#a78bfa,stroke-width:2px,color:#e1e4ed
style EXT fill:#0d2a2a,stroke:#34d399,stroke-width:2px,color:#e1e4ed
style Redis fill:#3b1010,stroke:#f87171,stroke-width:1px,color:#e1e4ed
</pre>
</div>
<p class="diagram-label">전체 시스템 아키텍처 구성도</p>
<div class="note">
<strong>콘텐츠 생성 흐름:</strong> 사용자 요청 → Naver 크롤링 → ChatGPT 가사 생성 → Suno AI 음악 생성 → Creatomate 영상 생성 → Blob 저장 → YouTube/Instagram 업로드
</div>
</section>
<!-- ==================== 3. 비용 산출 ==================== -->
<section id="cost">
<div class="section-header">
<h2><span class="num">3</span>예상 리소스 및 비용</h2>
<p class="desc"> 기반 단계별 월 예상 비용 (인프라 + 외부 API)</p>
</div>
<div class="cost-cards">
<div class="cost-card s1">
<div class="stage">Stage 1 · 동시 ~50명</div>
<div class="amount">$170~390</div>
<div class="krw">약 22~51만원/월</div>
</div>
<div class="cost-card s2">
<div class="stage">Stage 2 · 동시 50~200명</div>
<div class="amount">$960~2,160</div>
<div class="krw">약 125~280만원/월</div>
</div>
<div class="cost-card s3">
<div class="stage">Stage 3 · 동시 200~1,000명</div>
<div class="amount">$3,850~8,500</div>
<div class="krw">약 500~1,100만원/월</div>
</div>
</div>
<div class="cols">
<div class="col">
<h3>항목별 비용 상세</h3>
<table>
<thead>
<tr>
<th>항목</th>
<th>Stage 1</th>
<th>Stage 2</th>
<th>Stage 3</th>
</tr>
</thead>
<tbody>
<tr>
<td><strong>App Server</strong></td>
<td>$50~70</td>
<td>$200~400</td>
<td>$600~1,000</td>
</tr>
<tr>
<td><strong>Nginx</strong></td>
<td>-</td>
<td>포함 / VM $15~30</td>
<td>VM $30~60</td>
</tr>
<tr>
<td><strong>MySQL Primary</strong></td>
<td>B1ms $15~25</td>
<td>GP $130~160</td>
<td>BC $350~450</td>
</tr>
<tr>
<td><strong>MySQL Replica</strong></td>
<td>-</td>
<td>GP x1 $130~160</td>
<td>BC x2 $260~360</td>
</tr>
<tr>
<td><strong>Redis</strong></td>
<td>-</td>
<td>-</td>
<td>P1 $225</td>
</tr>
<tr>
<td><strong>스토리지/네트워크</strong></td>
<td>$10~20</td>
<td>$55~100</td>
<td>$160~270</td>
</tr>
<tr>
<td><strong>AI API (합계)</strong></td>
<td>$90~280</td>
<td>$400~1,250</td>
<td>$2,100~5,800</td>
</tr>
</tbody>
</table>
</div>
<div class="col">
<h3>DB 용량 예측 (1년 후)</h3>
<table>
<thead>
<tr>
<th></th>
<th>Stage 1 (500명)</th>
<th>Stage 2 (5,000명)</th>
<th>Stage 3 (50,000명)</th>
</tr>
</thead>
<tbody>
<tr>
<td><strong>DB 용량</strong></td>
<td>~1.2GB</td>
<td>~12GB</td>
<td>~120GB</td>
</tr>
<tr>
<td><strong>Blob 스토리지</strong></td>
<td>~1.1TB</td>
<td>~11TB</td>
<td>~110TB</td>
</tr>
<tr>
<td><strong>MySQL 추천</strong></td>
<td>32GB SSD</td>
<td>128GB SSD</td>
<td>512GB SSD</td>
</tr>
</tbody>
</table>
<div class="note" style="margin-top: 16px;">
<strong>비용 최적화 팁:</strong>
3rd party 의존도 낮춰야함
<br>
Blob Lifecycle Policy (30일 미접근 → Cool 티어),
</div>
</div>
</div>
<!-- 다이어그램: 내용 아래 풀 와이드 -->
<div class="diagram-box">
<pre class="mermaid">
pie title Stage 3 월 비용 구성 비중
"App Server (APP+Scheduler)" : 800
"Nginx" : 45
"MySQL Primary" : 400
"MySQL Replica x2" : 310
"Redis Premium" : 225
"스토리지/네트워크" : 215
"OpenAI API" : 550
"Suno AI" : 1400
"Creatomate" : 2000
</pre>
</div>
<p class="diagram-label">Stage 3 월간 비용 구성 비율 — AI API 비중이 전체의 약 66%</p>
</section>
</main>
<script>
function downloadPDF() {
// Mermaid SVG가 렌더링된 후 인쇄
window.print();
}
mermaid.initialize({
startOnLoad: true,
theme: 'dark',
themeVariables: {
primaryColor: '#1a2744',
primaryTextColor: '#e1e4ed',
primaryBorderColor: '#6c8cff',
lineColor: '#6c8cff',
secondaryColor: '#232733',
tertiaryColor: '#2e3345',
pieSectionTextColor: '#e1e4ed',
pieLegendTextColor: '#e1e4ed',
pieTitleTextColor: '#e1e4ed',
pieStrokeColor: '#2e3345'
}
});
</script>
</body>
</html>

Binary file not shown.

View File

@ -1,551 +0,0 @@
"""
O2O CastAD Backend - 인프라 아키텍처 PPT 생성 스크립트
실행: python3 docs/generate_ppt.py
출력: docs/architecture.pptx
"""
from pptx import Presentation
from pptx.util import Inches, Pt, Emu
from pptx.dml.color import RGBColor
from pptx.enum.text import PP_ALIGN, MSO_ANCHOR
from pptx.enum.shapes import MSO_SHAPE
# ── 색상 팔레트 (HTML 다크 테마 매칭) ──
BG = RGBColor(0x0F, 0x11, 0x17)
SURFACE = RGBColor(0x1A, 0x1D, 0x27)
SURFACE2 = RGBColor(0x23, 0x27, 0x33)
BORDER = RGBColor(0x2E, 0x33, 0x45)
TEXT = RGBColor(0xE1, 0xE4, 0xED)
TEXT_DIM = RGBColor(0x8B, 0x90, 0xA0)
ACCENT = RGBColor(0x6C, 0x8C, 0xFF)
ACCENT2 = RGBColor(0xA7, 0x8B, 0xFA)
GREEN = RGBColor(0x34, 0xD3, 0x99)
ORANGE = RGBColor(0xFB, 0x92, 0x3C)
RED = RGBColor(0xF8, 0x71, 0x71)
WHITE = RGBColor(0xFF, 0xFF, 0xFF)
SLIDE_W = Inches(13.333)
SLIDE_H = Inches(7.5)
def set_slide_bg(slide, color):
bg = slide.background
fill = bg.fill
fill.solid()
fill.fore_color.rgb = color
def add_textbox(slide, left, top, width, height, text, font_size=14,
color=TEXT, bold=False, alignment=PP_ALIGN.LEFT, font_name="맑은 고딕"):
txBox = slide.shapes.add_textbox(left, top, width, height)
tf = txBox.text_frame
tf.word_wrap = True
p = tf.paragraphs[0]
p.text = text
p.font.size = Pt(font_size)
p.font.color.rgb = color
p.font.bold = bold
p.font.name = font_name
p.alignment = alignment
return txBox
def add_bullet_list(slide, left, top, width, height, items, font_size=13):
txBox = slide.shapes.add_textbox(left, top, width, height)
tf = txBox.text_frame
tf.word_wrap = True
for i, item in enumerate(items):
if i == 0:
p = tf.paragraphs[0]
else:
p = tf.add_paragraph()
p.space_after = Pt(4)
p.font.size = Pt(font_size)
p.font.color.rgb = TEXT_DIM
p.font.name = "맑은 고딕"
# bold 부분 처리
if isinstance(item, tuple):
run_bold = p.add_run()
run_bold.text = item[0]
run_bold.font.bold = True
run_bold.font.color.rgb = TEXT
run_bold.font.size = Pt(font_size)
run_bold.font.name = "맑은 고딕"
run_normal = p.add_run()
run_normal.text = item[1]
run_normal.font.color.rgb = TEXT_DIM
run_normal.font.size = Pt(font_size)
run_normal.font.name = "맑은 고딕"
else:
p.text = f"{item}"
return txBox
def add_table(slide, left, top, width, height, headers, rows, col_widths=None):
n_rows = len(rows) + 1
n_cols = len(headers)
table_shape = slide.shapes.add_table(n_rows, n_cols, left, top, width, height)
table = table_shape.table
# 컬럼 폭 설정
if col_widths:
for i, w in enumerate(col_widths):
table.columns[i].width = w
# 헤더 행
for j, h in enumerate(headers):
cell = table.cell(0, j)
cell.text = h
for paragraph in cell.text_frame.paragraphs:
paragraph.font.size = Pt(11)
paragraph.font.bold = True
paragraph.font.color.rgb = ACCENT
paragraph.font.name = "맑은 고딕"
paragraph.alignment = PP_ALIGN.CENTER
cell.fill.solid()
cell.fill.fore_color.rgb = SURFACE2
# 데이터 행
for i, row in enumerate(rows):
for j, val in enumerate(row):
cell = table.cell(i + 1, j)
cell.text = str(val)
for paragraph in cell.text_frame.paragraphs:
paragraph.font.size = Pt(10)
paragraph.font.color.rgb = TEXT_DIM
paragraph.font.name = "맑은 고딕"
paragraph.alignment = PP_ALIGN.CENTER
cell.fill.solid()
cell.fill.fore_color.rgb = SURFACE if i % 2 == 0 else BG
# 테이블 테두리 제거 (깔끔하게)
for i in range(n_rows):
for j in range(n_cols):
cell = table.cell(i, j)
cell.vertical_anchor = MSO_ANCHOR.MIDDLE
for border_name in ['top', 'bottom', 'left', 'right']:
border = getattr(cell, f'border_{border_name}' if hasattr(cell, f'border_{border_name}') else border_name, None)
return table_shape
def add_rounded_rect(slide, left, top, width, height, fill_color, border_color=None, text="",
font_size=12, text_color=TEXT, bold=False):
shape = slide.shapes.add_shape(MSO_SHAPE.ROUNDED_RECTANGLE, left, top, width, height)
shape.fill.solid()
shape.fill.fore_color.rgb = fill_color
if border_color:
shape.line.color.rgb = border_color
shape.line.width = Pt(1.5)
else:
shape.line.fill.background()
if text:
tf = shape.text_frame
tf.word_wrap = True
p = tf.paragraphs[0]
p.text = text
p.font.size = Pt(font_size)
p.font.color.rgb = text_color
p.font.bold = bold
p.font.name = "맑은 고딕"
p.alignment = PP_ALIGN.CENTER
tf.paragraphs[0].space_before = Pt(0)
tf.paragraphs[0].space_after = Pt(0)
return shape
def add_section_number(slide, left, top, number, color=ACCENT):
shape = slide.shapes.add_shape(MSO_SHAPE.OVAL, left, top, Inches(0.4), Inches(0.4))
shape.fill.solid()
shape.fill.fore_color.rgb = color
shape.line.fill.background()
tf = shape.text_frame
p = tf.paragraphs[0]
p.text = str(number)
p.font.size = Pt(14)
p.font.color.rgb = WHITE
p.font.bold = True
p.font.name = "맑은 고딕"
p.alignment = PP_ALIGN.CENTER
return shape
def add_arrow(slide, x1, y1, x2, y2, color=ACCENT):
connector = slide.shapes.add_connector(1, x1, y1, x2, y2) # 1 = straight
connector.line.color.rgb = color
connector.line.width = Pt(1.5)
connector.end_x = x2
connector.end_y = y2
return connector
# ══════════════════════════════════════════════════════════════════
# PPT 생성 시작
# ══════════════════════════════════════════════════════════════════
prs = Presentation()
prs.slide_width = SLIDE_W
prs.slide_height = SLIDE_H
blank_layout = prs.slide_layouts[6] # Blank
# ── Slide 1: 타이틀 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_textbox(slide, Inches(0), Inches(2.2), SLIDE_W, Inches(1),
"O2O CastAD Backend", font_size=44, color=ACCENT, bold=True,
alignment=PP_ALIGN.CENTER)
add_textbox(slide, Inches(0), Inches(3.3), SLIDE_W, Inches(0.6),
"인프라 아키텍처 및 비용 산출 문서", font_size=20, color=TEXT_DIM,
alignment=PP_ALIGN.CENTER)
# 하단 구분선
line = slide.shapes.add_connector(1, Inches(4.5), Inches(4.2), Inches(8.8), Inches(4.2))
line.line.color.rgb = ACCENT
line.line.width = Pt(2)
add_textbox(slide, Inches(0), Inches(4.5), SLIDE_W, Inches(0.5),
"Nginx + FastAPI + MySQL + AI Pipeline", font_size=14, color=TEXT_DIM,
alignment=PP_ALIGN.CENTER)
# ── Slide 2: 부하 분산 - 현재 구현 & 확장 전략 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "1")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"DB 및 서버 부하 분산 방법", font_size=26, color=ACCENT, bold=True)
add_textbox(slide, Inches(1.0), Inches(0.85), Inches(10), Inches(0.3),
"Nginx 로드밸런싱, 커넥션 풀 관리, 단계별 수평 확장 전략", font_size=12, color=TEXT_DIM)
# 좌측: 현재 구현 현황
add_textbox(slide, Inches(0.5), Inches(1.5), Inches(5), Inches(0.4),
"현재 구현 현황 (단일 인스턴스)", font_size=16, color=TEXT, bold=True)
items = [
("API 커넥션 풀: ", "pool_size=20, max_overflow=20 → 최대 40"),
("백그라운드 풀: ", "pool_size=10, max_overflow=10 → 최대 20"),
("인스턴스당 총 DB 연결: ", "40 + 20 = 60"),
("풀 리사이클: ", "280초 (MySQL wait_timeout 300초 이전)"),
("Pre-ping: ", "활성화 (죽은 커넥션 자동 복구)"),
]
add_bullet_list(slide, Inches(0.5), Inches(2.0), Inches(5.5), Inches(2.5), items, font_size=12)
# 우측: 확장 전략 테이블
add_textbox(slide, Inches(6.8), Inches(1.5), Inches(6), Inches(0.4),
"단계별 확장 전략", font_size=16, color=TEXT, bold=True)
headers = ["단계", "동시접속", "App Server", "LB", "DB (MySQL Flexible)"]
rows = [
["S1", "~50명", "x1", "Nginx x1", "Burstable B1ms"],
["S2", "50~200명", "x2~4", "Nginx", "GP D2ds + Replica x1"],
["S3", "200~1,000명", "API xN\n+ Scheduler", "Nginx", "BC D4ds + Replica x2\n+ Redis P1"],
]
add_table(slide, Inches(6.8), Inches(2.0), Inches(6), Inches(2.0), headers, rows)
# 하단: 핵심 노트
note_shape = add_rounded_rect(slide, Inches(0.5), Inches(4.6), Inches(12.3), Inches(0.7),
SURFACE2, ACCENT)
tf = note_shape.text_frame
tf.word_wrap = True
p = tf.paragraphs[0]
p.text = ""
run = p.add_run()
run.text = "핵심: "
run.font.bold = True
run.font.color.rgb = ACCENT
run.font.size = Pt(11)
run.font.name = "맑은 고딕"
run = p.add_run()
run.text = "JWT Stateless 설계로 Nginx 세션 어피니티 불필요 (round-robin / least_conn). Stage 2부터 Read Replica로 읽기 분산, Redis는 Stage 3에서 캐싱/Rate Limiting 도입."
run.font.color.rgb = TEXT_DIM
run.font.size = Pt(11)
run.font.name = "맑은 고딕"
# ── Slide 3: 커넥션 풀 수치 계산 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "1")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"커넥션 풀 수치 계산", font_size=26, color=ACCENT, bold=True)
add_textbox(slide, Inches(1.0), Inches(0.85), Inches(10), Inches(0.3),
"인스턴스 수 증가에 따른 인스턴스당 풀 사이즈 축소 및 총 DB 커넥션 관리", font_size=12, color=TEXT_DIM)
headers = ["항목", "Stage 1 (1대)", "Stage 2 (4대)", "Stage 3 (8대)"]
rows = [
["Main Pool / 인스턴스", "20+20 = 40", "10+10 = 20", "5+5 = 10"],
["BG Pool / 인스턴스", "10+10 = 20", "5+5 = 10", "3+3 = 6"],
["인스턴스당 소계", "60", "30", "16"],
["Primary 총 연결", "60", "4 x 30 = 120", "8 x 16 = 128"],
["max_connections 권장", "100", "200", "300"],
]
add_table(slide, Inches(1.5), Inches(1.6), Inches(10.3), Inches(2.8), headers, rows)
# 시각적 요약 박스
stages = [
("Stage 1", "1대 × 60 = 60", GREEN, Inches(2)),
("Stage 2", "4대 × 30 = 120", ORANGE, Inches(5.5)),
("Stage 3", "8대 × 16 = 128", RED, Inches(9)),
]
for label, val, color, left in stages:
add_rounded_rect(slide, left, Inches(4.8), Inches(2.3), Inches(0.9),
SURFACE2, color, f"{label}\n{val}", font_size=13, text_color=color, bold=True)
# ── Slide 4: 아키텍처 다이어그램 (상세 블록) ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "2")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"전체 아키텍처 다이어그램", font_size=26, color=ACCENT, bold=True)
add_textbox(slide, Inches(1.0), Inches(0.85), Inches(10), Inches(0.3),
"Nginx + FastAPI App Server 구성과 외부 서비스 연동 구조", font_size=12, color=TEXT_DIM)
# 클라이언트
add_rounded_rect(slide, Inches(5.5), Inches(1.4), Inches(2.3), Inches(0.6),
RGBColor(0x1A, 0x3A, 0x1A), GREEN, "클라이언트 (Web / App)",
font_size=11, text_color=TEXT)
# Nginx
add_rounded_rect(slide, Inches(5.5), Inches(2.3), Inches(2.3), Inches(0.6),
RGBColor(0x1A, 0x3A, 0x1A), GREEN, "Nginx\n(Reverse Proxy + SSL)",
font_size=10, text_color=TEXT)
# App Server
app_box = add_rounded_rect(slide, Inches(2.5), Inches(3.2), Inches(8.3), Inches(1.2),
RGBColor(0x1A, 0x27, 0x44), ACCENT, "", font_size=10)
add_textbox(slide, Inches(2.7), Inches(3.15), Inches(3), Inches(0.3),
"App Server (FastAPI)", font_size=11, color=ACCENT, bold=True)
modules = ["Auth", "Home", "Lyric", "Song", "Video", "Social", "SNS", "Archive", "Admin", "BG Worker"]
for i, mod in enumerate(modules):
col = i % 10
x = Inches(2.7 + col * 0.8)
y = Inches(3.55)
add_rounded_rect(slide, x, y, Inches(0.7), Inches(0.5),
SURFACE2, BORDER, mod, font_size=8, text_color=TEXT_DIM)
# DB
add_rounded_rect(slide, Inches(1.0), Inches(4.85), Inches(3.5), Inches(0.7),
RGBColor(0x2A, 0x1F, 0x00), ORANGE, "MySQL Flexible Server\nPrimary (R/W) + Read Replica",
font_size=10, text_color=TEXT)
# Redis
add_rounded_rect(slide, Inches(5.0), Inches(4.85), Inches(2.3), Inches(0.7),
RGBColor(0x3B, 0x10, 0x10), RED, "Cache for Redis\n(Stage 3 도입)",
font_size=10, text_color=TEXT)
# AI Pipeline
add_rounded_rect(slide, Inches(7.8), Inches(4.85), Inches(4.5), Inches(0.7),
RGBColor(0x2A, 0x0F, 0x2A), ACCENT2, "AI Pipeline: ChatGPT → Suno AI → Creatomate",
font_size=10, text_color=TEXT)
# 외부 서비스
ext_items = [("Blob\nStorage", Inches(1.0)), ("Kakao\nOAuth", Inches(3.2)),
("YouTube /\nInstagram", Inches(5.4)), ("Naver Map /\nSearch", Inches(7.6))]
for label, x in ext_items:
add_rounded_rect(slide, x, Inches(6.0), Inches(1.8), Inches(0.7),
RGBColor(0x0D, 0x2A, 0x2A), GREEN, label,
font_size=9, text_color=TEXT_DIM)
# 콘텐츠 생성 흐름 노트
note_shape = add_rounded_rect(slide, Inches(1.0), Inches(6.85), Inches(11.3), Inches(0.45),
SURFACE2, ACCENT)
tf = note_shape.text_frame
tf.word_wrap = True
p = tf.paragraphs[0]
run = p.add_run()
run.text = "콘텐츠 생성 흐름: "
run.font.bold = True
run.font.color.rgb = ACCENT
run.font.size = Pt(10)
run.font.name = "맑은 고딕"
run = p.add_run()
run.text = "사용자 요청 → Naver 크롤링 → ChatGPT 가사 → Suno AI 음악 → Creatomate 영상 → Blob 저장 → YouTube/Instagram 업로드"
run.font.color.rgb = TEXT_DIM
run.font.size = Pt(10)
run.font.name = "맑은 고딕"
# ── Slide 5: Stage별 인프라 스케일링 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "2")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"단계별 인프라 스케일링", font_size=26, color=ACCENT, bold=True)
# Stage 1
stage1_box = add_rounded_rect(slide, Inches(0.5), Inches(1.3), Inches(3.8), Inches(5.2),
RGBColor(0x0D, 0x33, 0x20), GREEN)
add_textbox(slide, Inches(0.7), Inches(1.4), Inches(3.4), Inches(0.3),
"Stage 1: ~50명", font_size=16, color=GREEN, bold=True)
items = [
("Nginx: ", "Reverse Proxy x1"),
("App Server: ", "x1"),
("MySQL: ", "Burstable B1ms"),
("Redis: ", "미사용"),
("월 비용: ", "$170~390"),
]
add_bullet_list(slide, Inches(0.7), Inches(1.9), Inches(3.4), Inches(3), items, font_size=12)
# Stage 2
stage2_box = add_rounded_rect(slide, Inches(4.7), Inches(1.3), Inches(3.8), Inches(5.2),
RGBColor(0x3B, 0x25, 0x06), ORANGE)
add_textbox(slide, Inches(4.9), Inches(1.4), Inches(3.4), Inches(0.3),
"Stage 2: 50~200명", font_size=16, color=ORANGE, bold=True)
items = [
("Nginx: ", "Reverse Proxy (LB)"),
("App Server: ", "x2~4"),
("Scheduler: ", "Server x1"),
("MySQL: ", "GP D2ds + Replica x1"),
("Redis: ", "미사용"),
("월 비용: ", "$960~2,160"),
]
add_bullet_list(slide, Inches(4.9), Inches(1.9), Inches(3.4), Inches(3.5), items, font_size=12)
# Stage 3
stage3_box = add_rounded_rect(slide, Inches(8.9), Inches(1.3), Inches(3.8), Inches(5.2),
RGBColor(0x3B, 0x10, 0x10), RED)
add_textbox(slide, Inches(9.1), Inches(1.4), Inches(3.4), Inches(0.3),
"Stage 3: 200~1,000명", font_size=16, color=RED, bold=True)
items = [
("Nginx: ", "Reverse Proxy (LB)"),
("API Server: ", "x N (Auto Scale)"),
("Scheduler: ", "Server x1"),
("MySQL: ", "BC D4ds + Replica x2"),
("Redis: ", "Premium P1 (캐싱)"),
("월 비용: ", "$3,850~8,500"),
]
add_bullet_list(slide, Inches(9.1), Inches(1.9), Inches(3.4), Inches(3.5), items, font_size=12)
# ── Slide 6: 비용 산출 개요 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "3")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"예상 리소스 및 비용", font_size=26, color=ACCENT, bold=True)
add_textbox(slide, Inches(1.0), Inches(0.85), Inches(10), Inches(0.3),
"단계별 월 예상 비용 (인프라 + 외부 API)", font_size=12, color=TEXT_DIM)
# 비용 카드 3개
cost_data = [
("Stage 1", "동시 ~50명", "$170~390", "약 22~51만원/월", GREEN),
("Stage 2", "동시 50~200명", "$960~2,160", "약 125~280만원/월", ORANGE),
("Stage 3", "동시 200~1,000명", "$3,850~8,500", "약 500~1,100만원/월", RED),
]
for i, (stage, users, usd, krw, color) in enumerate(cost_data):
x = Inches(1.0 + i * 3.9)
card = add_rounded_rect(slide, x, Inches(1.5), Inches(3.5), Inches(1.8), SURFACE2, color)
add_textbox(slide, x, Inches(1.6), Inches(3.5), Inches(0.3),
f"{stage} · {users}", font_size=12, color=TEXT_DIM, alignment=PP_ALIGN.CENTER)
add_textbox(slide, x, Inches(2.0), Inches(3.5), Inches(0.5),
usd, font_size=28, color=color, bold=True, alignment=PP_ALIGN.CENTER)
add_textbox(slide, x, Inches(2.6), Inches(3.5), Inches(0.3),
krw, font_size=12, color=TEXT_DIM, alignment=PP_ALIGN.CENTER)
# 항목별 비용 상세 테이블
add_textbox(slide, Inches(0.5), Inches(3.6), Inches(5), Inches(0.4),
"항목별 비용 상세", font_size=16, color=TEXT, bold=True)
headers = ["항목", "Stage 1", "Stage 2", "Stage 3"]
rows = [
["App Server", "$50~70", "$200~400", "$600~1,000"],
["Nginx", "포함", "포함 / VM $15~30", "VM $30~60"],
["MySQL Primary", "B1ms $15~25", "GP $130~160", "BC $350~450"],
["MySQL Replica", "-", "GP x1 $130~160", "BC x2 $260~360"],
["Redis", "-", "-", "P1 $225"],
["스토리지/네트워크", "$10~20", "$55~100", "$160~270"],
["AI API (합계)", "$90~280", "$400~1,250", "$2,100~5,800"],
]
add_table(slide, Inches(0.5), Inches(4.1), Inches(12.3), Inches(3.0), headers, rows)
# ── Slide 7: 비용 구성 비중 & DB 용량 ──
slide = prs.slides.add_slide(blank_layout)
set_slide_bg(slide, BG)
add_section_number(slide, Inches(0.5), Inches(0.4), "3")
add_textbox(slide, Inches(1.0), Inches(0.35), Inches(8), Inches(0.5),
"Stage 3 비용 구성 & DB 용량", font_size=26, color=ACCENT, bold=True)
# 좌측: Stage 3 비용 구성 (수평 바 차트 시뮬레이션)
add_textbox(slide, Inches(0.5), Inches(1.2), Inches(5), Inches(0.4),
"Stage 3 월 비용 구성 비중", font_size=16, color=TEXT, bold=True)
cost_items = [
("Creatomate", 2000, RED),
("Suno AI", 1400, ORANGE),
("App Server", 800, ACCENT),
("OpenAI API", 550, ACCENT2),
("MySQL Primary", 400, ORANGE),
("MySQL Replica x2", 310, ORANGE),
("Redis Premium", 225, RED),
("스토리지/네트워크", 215, GREEN),
("Nginx", 45, GREEN),
]
total = sum(v for _, v, _ in cost_items)
max_bar_width = 5.0 # inches
y_start = Inches(1.7)
for i, (label, value, color) in enumerate(cost_items):
y = y_start + Emu(int(i * Inches(0.45)))
bar_w = Inches(max_bar_width * value / total)
# 바
bar = slide.shapes.add_shape(MSO_SHAPE.ROUNDED_RECTANGLE, Inches(2.2), y, bar_w, Inches(0.32))
bar.fill.solid()
bar.fill.fore_color.rgb = color
bar.line.fill.background()
# 레이블
add_textbox(slide, Inches(0.5), y, Inches(1.6), Inches(0.32),
label, font_size=9, color=TEXT_DIM, alignment=PP_ALIGN.RIGHT)
# 값
pct = value / total * 100
val_x = Inches(2.3) + bar_w
add_textbox(slide, val_x, y, Inches(1.5), Inches(0.32),
f"${value:,} ({pct:.0f}%)", font_size=9, color=TEXT_DIM)
add_textbox(slide, Inches(0.5), y_start + Emu(int(len(cost_items) * Inches(0.45))), Inches(6), Inches(0.3),
f"AI API 비중: 전체의 약 66% (${(2000+1400+550):,} / ${total:,})",
font_size=11, color=ACCENT, bold=True)
# 우측: DB 용량 예측
add_textbox(slide, Inches(7.5), Inches(1.2), Inches(5), Inches(0.4),
"DB 용량 예측 (1년 후)", font_size=16, color=TEXT, bold=True)
headers = ["", "Stage 1\n(500명)", "Stage 2\n(5,000명)", "Stage 3\n(50,000명)"]
rows = [
["DB 용량", "~1.2GB", "~12GB", "~120GB"],
["Blob 스토리지", "~1.1TB", "~11TB", "~110TB"],
["MySQL 추천", "32GB SSD", "128GB SSD", "512GB SSD"],
]
add_table(slide, Inches(7.5), Inches(1.7), Inches(5.3), Inches(1.8), headers, rows)
# 비용 최적화 팁
add_textbox(slide, Inches(7.5), Inches(3.8), Inches(5), Inches(0.4),
"비용 최적화 팁", font_size=16, color=TEXT, bold=True)
items = [
("3rd party 의존도: ", "낮춰야 함 (AI API가 전체 비용의 66%)"),
("Blob Lifecycle: ", "30일 미접근 미디어 → Cool 티어 자동 이전"),
("App Server: ", "비활성 시간대(야간) 인스턴스 축소"),
("OpenAI Batch API: ", "비실시간 가사생성은 50% 절감 가능"),
("Reserved Instances: ", "1년 예약 시 ~30% 할인"),
]
add_bullet_list(slide, Inches(7.5), Inches(4.3), Inches(5.3), Inches(3), items, font_size=11)
# ── 저장 ──
output_path = "/Users/marineyang/Desktop/work/code/o2o-castad-backend/docs/architecture.pptx"
prs.save(output_path)
print(f"PPT 생성 완료: {output_path}")

View File

@ -21,7 +21,6 @@ from app.sns.api.routers.v1.sns import router as sns_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
from app.social.api.routers.v1.seo import router as social_seo_router
from app.utils.cors import CustomCORSMiddleware from app.utils.cors import CustomCORSMiddleware
from config import prj_settings from config import prj_settings
@ -361,7 +360,6 @@ app.include_router(video_router)
app.include_router(archive_router) # Archive API 라우터 추가 app.include_router(archive_router) # Archive API 라우터 추가
app.include_router(social_oauth_router, prefix="/social") # Social OAuth 라우터 추가 app.include_router(social_oauth_router, prefix="/social") # Social OAuth 라우터 추가
app.include_router(social_upload_router, prefix="/social") # Social Upload 라우터 추가 app.include_router(social_upload_router, prefix="/social") # Social Upload 라우터 추가
app.include_router(social_seo_router, prefix="/social") # Social Upload 라우터 추가
app.include_router(sns_router) # SNS API 라우터 추가 app.include_router(sns_router) # SNS API 라우터 추가
# DEBUG 모드에서만 테스트 라우터 등록 # DEBUG 모드에서만 테스트 라우터 등록

View File

@ -1,422 +0,0 @@
# 서버 JWT 토큰 라이프사이클 로깅 강화 계획
## 1. 토큰 라이프사이클 개요
서버가 직접 발급/관리하는 JWT 토큰의 전체 흐름:
```
[발급] kakao_login() / generate_test_token()
├── Access Token 생성 (sub=user_uuid, type=access, exp=60분)
├── Refresh Token 생성 (sub=user_uuid, type=refresh, exp=7일)
└── Refresh Token DB 저장 (token_hash, user_id, expires_at)
[검증] get_current_user() — 매 요청마다 Access Token 검증
├── Bearer 헤더에서 토큰 추출
├── decode_token() → payload (sub, type, exp)
├── type == "access" 확인
└── user_uuid로 사용자 조회/활성 확인
[갱신] refresh_tokens() — Access Token 만료 시 Refresh Token으로 갱신
├── 기존 Refresh Token 디코딩 → payload
├── token_hash로 DB 조회 → is_revoked / expires_at 확인
├── 기존 Refresh Token 폐기 (is_revoked=True)
├── 새 Access Token + 새 Refresh Token 발급
└── 새 Refresh Token DB 저장
[폐기] logout() / logout_all()
├── 단일: token_hash로 해당 Refresh Token 폐기
└── 전체: user_id로 모든 Refresh Token 폐기
```
---
## 2. 현황 분석 — 로깅 공백 지점
### 발급 단계
| 위치 | 함수 | 현재 로깅 | 부족한 정보 |
|------|------|----------|------------|
| `jwt.py` | `create_access_token()` | 없음 | 발급 대상(user_uuid), 만료시간 |
| `jwt.py` | `create_refresh_token()` | 없음 | 발급 대상(user_uuid), 만료시간 |
| `auth.py` (service) | `_save_refresh_token()` | 없음 | DB 저장 결과(token_hash, expires_at) |
| `auth.py` (service) | `kakao_login()` | `debug`로 토큰 앞 30자 출력 | 충분 (변경 불필요) |
### 검증 단계
| 위치 | 함수 | 현재 로깅 | 부족한 정보 |
|------|------|----------|------------|
| `jwt.py` | `decode_token()` | 없음 | 디코딩 성공 시 payload 내용, 실패 시 원인 |
| `auth.py` (dependency) | `get_current_user()` | 없음 | 검증 각 단계 통과/실패 사유, 토큰 내 정보 |
| `auth.py` (dependency) | `get_current_user_optional()` | 없음 | 위와 동일 |
### 갱신 단계
| 위치 | 함수 | 현재 로깅 | 부족한 정보 |
|------|------|----------|------------|
| `auth.py` (router) | `refresh_token()` | 없음 | 수신 토큰 정보, 갱신 결과 |
| `auth.py` (service) | `refresh_tokens()` | 진입/완료 `info` 1줄씩 | 각 단계 실패 사유, DB 토큰 상태, 신규 토큰 정보 |
### 폐기 단계
| 위치 | 함수 | 현재 로깅 | 부족한 정보 |
|------|------|----------|------------|
| `auth.py` (router) | `logout()`, `logout_all()` | 없음 | 요청 수신, 대상 사용자 |
| `auth.py` (service) | `logout()`, `logout_all()` | 없음 | 폐기 대상, 폐기 결과 |
---
## 3. 수정 대상 파일
| # | 파일 | 수정 내용 |
|---|------|----------|
| 1 | `app/user/services/jwt.py` | 토큰 발급 로그 + `decode_token()` 실패 원인 분류 |
| 2 | `app/user/dependencies/auth.py` | Access Token 검증 과정 로깅 |
| 3 | `app/user/services/auth.py` | `refresh_tokens()`, `_save_refresh_token()`, `logout()`, `logout_all()` 로깅 |
| 4 | `app/user/api/routers/v1/auth.py` | `refresh_token()`, `logout()`, `logout_all()` 라우터 로깅 |
---
## 4. 상세 구현 계획
### 4-1. `jwt.py` — 토큰 발급 로그 + 디코딩 실패 원인 분류
**import 추가:**
```python
import logging
from jose import JWTError, ExpiredSignatureError, JWTClaimsError, jwt
logger = logging.getLogger(__name__)
```
**`create_access_token()` — 발급 로그 추가:**
```python
def create_access_token(user_uuid: str) -> str:
expire = now() + timedelta(minutes=jwt_settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode = {"sub": user_uuid, "exp": expire, "type": "access"}
token = jwt.encode(to_encode, jwt_settings.JWT_SECRET, algorithm=jwt_settings.JWT_ALGORITHM)
logger.debug(f"[JWT] Access Token 발급 - user_uuid: {user_uuid}, expires: {expire}, token: ...{token[-20:]}")
return token
```
**`create_refresh_token()` — 발급 로그 추가:**
```python
def create_refresh_token(user_uuid: str) -> str:
expire = now() + timedelta(days=jwt_settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS)
to_encode = {"sub": user_uuid, "exp": expire, "type": "refresh"}
token = jwt.encode(to_encode, jwt_settings.JWT_SECRET, algorithm=jwt_settings.JWT_ALGORITHM)
logger.debug(f"[JWT] Refresh Token 발급 - user_uuid: {user_uuid}, expires: {expire}, token: ...{token[-20:]}")
return token
```
**`decode_token()` — 성공/실패 분류 로그:**
```python
def decode_token(token: str) -> Optional[dict]:
try:
payload = jwt.decode(token, jwt_settings.JWT_SECRET, algorithms=[jwt_settings.JWT_ALGORITHM])
logger.debug(
f"[JWT] 토큰 디코딩 성공 - type: {payload.get('type')}, "
f"sub: {payload.get('sub')}, exp: {payload.get('exp')}, "
f"token: ...{token[-20:]}"
)
return payload
except ExpiredSignatureError:
logger.info(f"[JWT] 토큰 만료 - token: ...{token[-20:]}")
return None
except JWTClaimsError as e:
logger.warning(f"[JWT] 클레임 검증 실패 - error: {e}, token: ...{token[-20:]}")
return None
except JWTError as e:
logger.warning(f"[JWT] 토큰 디코딩 실패 - error: {type(e).__name__}: {e}, token: ...{token[-20:]}")
return None
```
### 4-2. `dependencies/auth.py` — Access Token 검증 로깅
**import 추가:**
```python
import logging
logger = logging.getLogger(__name__)
```
**`get_current_user()` — 검증 과정 로그:**
```python
async def get_current_user(...) -> User:
if credentials is None:
logger.info("[AUTH-DEP] 토큰 없음 - MissingTokenError")
raise MissingTokenError()
token = credentials.credentials
logger.debug(f"[AUTH-DEP] Access Token 검증 시작 - token: ...{token[-20:]}")
payload = decode_token(token)
if payload is None:
logger.warning(f"[AUTH-DEP] Access Token 디코딩 실패 - token: ...{token[-20:]}")
raise InvalidTokenError()
if payload.get("type") != "access":
logger.warning(f"[AUTH-DEP] 토큰 타입 불일치 - expected: access, got: {payload.get('type')}, sub: {payload.get('sub')}")
raise InvalidTokenError("액세스 토큰이 아닙니다.")
user_uuid = payload.get("sub")
if user_uuid is None:
logger.warning(f"[AUTH-DEP] 토큰에 sub 클레임 없음 - token: ...{token[-20:]}")
raise InvalidTokenError()
# 사용자 조회
result = await session.execute(...)
user = result.scalar_one_or_none()
if user is None:
logger.warning(f"[AUTH-DEP] 사용자 미존재 - user_uuid: {user_uuid}")
raise UserNotFoundError()
if not user.is_active:
logger.warning(f"[AUTH-DEP] 비활성 사용자 접근 - user_uuid: {user_uuid}, user_id: {user.id}")
raise UserInactiveError()
logger.debug(f"[AUTH-DEP] Access Token 검증 성공 - user_uuid: {user_uuid}, user_id: {user.id}")
return user
```
**`get_current_user_optional()` — 동일 패턴, `debug` 레벨:**
```python
async def get_current_user_optional(...) -> Optional[User]:
if credentials is None:
logger.debug("[AUTH-DEP] 선택적 인증 - 토큰 없음")
return None
token = credentials.credentials
payload = decode_token(token)
if payload is None:
logger.debug(f"[AUTH-DEP] 선택적 인증 - 디코딩 실패, token: ...{token[-20:]}")
return None
if payload.get("type") != "access":
logger.debug(f"[AUTH-DEP] 선택적 인증 - 타입 불일치 (type={payload.get('type')})")
return None
user_uuid = payload.get("sub")
if user_uuid is None:
logger.debug("[AUTH-DEP] 선택적 인증 - sub 없음")
return None
result = await session.execute(...)
user = result.scalar_one_or_none()
if user is None or not user.is_active:
logger.debug(f"[AUTH-DEP] 선택적 인증 - 사용자 미존재 또는 비활성, user_uuid: {user_uuid}")
return None
logger.debug(f"[AUTH-DEP] 선택적 인증 성공 - user_uuid: {user_uuid}, user_id: {user.id}")
return user
```
### 4-3. `services/auth.py` — Refresh Token 갱신/폐기 로깅
**`refresh_tokens()` — 전체 흐름 로그:**
```python
async def refresh_tokens(self, refresh_token: str, session: AsyncSession) -> TokenResponse:
logger.info(f"[AUTH] 토큰 갱신 시작 (Rotation) - token: ...{refresh_token[-20:]}")
# 1. 디코딩
payload = decode_token(refresh_token)
if payload is None:
logger.warning(f"[AUTH] 토큰 갱신 실패 [1/8 디코딩] - token: ...{refresh_token[-20:]}")
raise InvalidTokenError()
if payload.get("type") != "refresh":
logger.warning(f"[AUTH] 토큰 갱신 실패 [1/8 타입] - type={payload.get('type')}, sub: {payload.get('sub')}")
raise InvalidTokenError("리프레시 토큰이 아닙니다.")
logger.debug(f"[AUTH] 토큰 갱신 [1/8] 디코딩 성공 - sub: {payload.get('sub')}, exp: {payload.get('exp')}")
# 2. DB 조회
token_hash = get_token_hash(refresh_token)
db_token = await self._get_refresh_token_by_hash(token_hash, session)
if db_token is None:
logger.warning(f"[AUTH] 토큰 갱신 실패 [2/8 DB조회] - DB에 없음, token_hash: {token_hash[:16]}...")
raise InvalidTokenError()
logger.debug(f"[AUTH] 토큰 갱신 [2/8] DB 조회 성공 - token_hash: {token_hash[:16]}..., user_uuid: {db_token.user_uuid}, is_revoked: {db_token.is_revoked}, expires_at: {db_token.expires_at}")
# 3. 폐기 여부
if db_token.is_revoked:
logger.warning(f"[AUTH] 토큰 갱신 실패 [3/8 폐기됨] - 이미 폐기된 토큰 (replay attack 의심), token_hash: {token_hash[:16]}..., user_uuid: {db_token.user_uuid}, revoked_at: {db_token.revoked_at}")
raise TokenRevokedError()
# 4. 만료 확인
if db_token.expires_at < now().replace(tzinfo=None):
logger.info(f"[AUTH] 토큰 갱신 실패 [4/8 만료] - expires_at: {db_token.expires_at}, user_uuid: {db_token.user_uuid}")
raise TokenExpiredError()
# 5. 사용자 확인
user_uuid = payload.get("sub")
user = await self._get_user_by_uuid(user_uuid, session)
if user is None:
logger.warning(f"[AUTH] 토큰 갱신 실패 [5/8 사용자] - 사용자 미존재, user_uuid: {user_uuid}")
raise UserNotFoundError()
if not user.is_active:
logger.warning(f"[AUTH] 토큰 갱신 실패 [5/8 비활성] - user_uuid: {user_uuid}, user_id: {user.id}")
raise UserInactiveError()
# 6. 기존 토큰 폐기
db_token.is_revoked = True
db_token.revoked_at = now().replace(tzinfo=None)
logger.debug(f"[AUTH] 토큰 갱신 [6/8] 기존 토큰 폐기 - token_hash: {token_hash[:16]}...")
# 7. 새 토큰 발급
new_access_token = create_access_token(user.user_uuid)
new_refresh_token = create_refresh_token(user.user_uuid)
logger.debug(f"[AUTH] 토큰 갱신 [7/8] 새 토큰 발급 - new_access: ...{new_access_token[-20:]}, new_refresh: ...{new_refresh_token[-20:]}")
# 8. 새 Refresh Token DB 저장 + 커밋
await self._save_refresh_token(user_id=user.id, user_uuid=user.user_uuid, token=new_refresh_token, session=session)
await session.commit()
logger.info(f"[AUTH] 토큰 갱신 완료 [8/8] - user_uuid: {user.user_uuid}, user_id: {user.id}, old_hash: {token_hash[:16]}..., new_refresh: ...{new_refresh_token[-20:]}")
return TokenResponse(...)
```
**`_save_refresh_token()` — DB 저장 로그:**
```python
async def _save_refresh_token(self, ...) -> RefreshToken:
token_hash = get_token_hash(token)
expires_at = get_refresh_token_expires_at()
refresh_token = RefreshToken(...)
session.add(refresh_token)
await session.flush()
logger.debug(f"[AUTH] Refresh Token DB 저장 - user_uuid: {user_uuid}, token_hash: {token_hash[:16]}..., expires_at: {expires_at}")
return refresh_token
```
**`logout()` — 단일 로그아웃 로그:**
```python
async def logout(self, user_id: int, refresh_token: str, session: AsyncSession) -> None:
token_hash = get_token_hash(refresh_token)
logger.info(f"[AUTH] 로그아웃 - user_id: {user_id}, token_hash: {token_hash[:16]}..., token: ...{refresh_token[-20:]}")
await self._revoke_refresh_token_by_hash(token_hash, session)
logger.info(f"[AUTH] 로그아웃 완료 - user_id: {user_id}")
```
**`logout_all()` — 전체 로그아웃 로그:**
```python
async def logout_all(self, user_id: int, session: AsyncSession) -> None:
logger.info(f"[AUTH] 전체 로그아웃 - user_id: {user_id}")
await self._revoke_all_user_tokens(user_id, session)
logger.info(f"[AUTH] 전체 로그아웃 완료 - user_id: {user_id}")
```
### 4-4. `routers/v1/auth.py` — 라우터 진입/완료 로깅
```python
# POST /auth/refresh
async def refresh_token(body, session) -> TokenResponse:
logger.info(f"[ROUTER] POST /auth/refresh - token: ...{body.refresh_token[-20:]}")
result = await auth_service.refresh_tokens(refresh_token=body.refresh_token, session=session)
logger.info(f"[ROUTER] POST /auth/refresh 완료 - new_access: ...{result.access_token[-20:]}, new_refresh: ...{result.refresh_token[-20:]}")
return result
# POST /auth/logout
async def logout(body, current_user, session) -> Response:
logger.info(f"[ROUTER] POST /auth/logout - user_id: {current_user.id}, user_uuid: {current_user.user_uuid}, token: ...{body.refresh_token[-20:]}")
await auth_service.logout(user_id=current_user.id, refresh_token=body.refresh_token, session=session)
logger.info(f"[ROUTER] POST /auth/logout 완료 - user_id: {current_user.id}")
return Response(status_code=status.HTTP_204_NO_CONTENT)
# POST /auth/logout/all
async def logout_all(current_user, session) -> Response:
logger.info(f"[ROUTER] POST /auth/logout/all - user_id: {current_user.id}, user_uuid: {current_user.user_uuid}")
await auth_service.logout_all(user_id=current_user.id, session=session)
logger.info(f"[ROUTER] POST /auth/logout/all 완료 - user_id: {current_user.id}")
return Response(status_code=status.HTTP_204_NO_CONTENT)
```
---
## 5. 보안 원칙
| 원칙 | 적용 방법 | 이유 |
|------|----------|------|
| 토큰 전체 노출 금지 | 뒷 20자만: `...{token[-20:]}` | 토큰 탈취 시 세션 하이재킹 가능 |
| 해시값 부분 노출 | 앞 16자만: `{hash[:16]}...` | DB 레코드 식별에 충분 |
| user_uuid 전체 허용 | 전체 출력 | 내부 식별자, 토큰이 아님 |
| 페이로드 내용 출력 | `sub`, `type`, `exp` 출력 | 디버깅에 필수, 민감정보 아님 |
| DB 토큰 상태 출력 | `is_revoked`, `expires_at`, `revoked_at` | 토큰 라이프사이클 추적 |
| 로그 레벨 구분 | 하단 표 참조 | 운영 환경에서 불필요한 로그 억제 |
### 로그 레벨 기준
| 레벨 | 사용 기준 | 예시 |
|------|----------|------|
| `debug` | 정상 처리 과정 상세 (운영환경에서 비활성) | 토큰 발급, 디코딩 성공, 검증 통과 |
| `info` | 주요 이벤트 (운영환경에서 활성) | 갱신 시작/완료, 로그아웃, 만료로 인한 실패 |
| `warning` | 비정상/의심 상황 | 디코딩 실패, 폐기된 토큰 사용, 사용자 미존재 |
---
## 6. 구현 순서
| 순서 | 파일 | 이유 |
|------|------|------|
| 1 | `app/user/services/jwt.py` | 최하위 유틸리티. 토큰 발급/디코딩의 기본 로그 |
| 2 | `app/user/dependencies/auth.py` | 모든 인증 API의 공통 진입점 |
| 3 | `app/user/services/auth.py` | 갱신/폐기 비즈니스 로직 |
| 4 | `app/user/api/routers/v1/auth.py` | 라우터 진입/완료 + 응답 토큰 정보 |
---
## 7. 기대 효과 — 시나리오별 로그 출력 예시
### 시나리오 1: 정상 토큰 갱신
```
[ROUTER] POST /auth/refresh - token: ...7d90-aac8-ecf1385c
[AUTH] 토큰 갱신 시작 (Rotation) - token: ...7d90-aac8-ecf1385c
[JWT] 토큰 디코딩 성공 - type: refresh, sub: 019c5452-b1cf-7d90-aac8-ecf1385c9dc4, exp: 1739450400
[AUTH] 토큰 갱신 [1/8] 디코딩 성공 - sub: 019c5452-..., exp: 1739450400
[AUTH] 토큰 갱신 [2/8] DB 조회 성공 - token_hash: a1b2c3d4e5f6g7h8..., is_revoked: False, expires_at: 2026-02-20 11:46:36
[AUTH] 토큰 갱신 [6/8] 기존 토큰 폐기 - token_hash: a1b2c3d4e5f6g7h8...
[JWT] Access Token 발급 - user_uuid: 019c5452-..., expires: 2026-02-13 12:46:36
[JWT] Refresh Token 발급 - user_uuid: 019c5452-..., expires: 2026-02-20 11:46:36
[AUTH] 토큰 갱신 [7/8] 새 토큰 발급 - new_access: ...xNewAccess12345, new_refresh: ...xNewRefresh6789
[AUTH] Refresh Token DB 저장 - user_uuid: 019c5452-..., token_hash: f8e9d0c1b2a3..., expires_at: 2026-02-20 11:46:36
[AUTH] 토큰 갱신 완료 [8/8] - user_uuid: 019c5452-..., user_id: 42, old_hash: a1b2c3d4e5f6g7h8..., new_refresh: ...xNewRefresh6789
[ROUTER] POST /auth/refresh 완료 - new_access: ...xNewAccess12345, new_refresh: ...xNewRefresh6789
```
### 시나리오 2: 만료된 Refresh Token으로 갱신 시도
```
[ROUTER] POST /auth/refresh - token: ...expiredToken12345
[AUTH] 토큰 갱신 시작 (Rotation) - token: ...expiredToken12345
[JWT] 토큰 만료 - token: ...expiredToken12345
[AUTH] 토큰 갱신 실패 [1/8 디코딩] - token: ...expiredToken12345
→ 401 InvalidTokenError 응답
```
### 시나리오 3: 이미 폐기된 Refresh Token 재사용 (Replay Attack)
```
[ROUTER] POST /auth/refresh - token: ...revokedToken98765
[AUTH] 토큰 갱신 시작 (Rotation) - token: ...revokedToken98765
[JWT] 토큰 디코딩 성공 - type: refresh, sub: 019c5452-..., exp: 1739450400
[AUTH] 토큰 갱신 [2/8] DB 조회 성공 - token_hash: c3d4e5f6..., is_revoked: True, expires_at: 2026-02-20
[AUTH] 토큰 갱신 실패 [3/8 폐기됨] - replay attack 의심, token_hash: c3d4e5f6..., user_uuid: 019c5452-..., revoked_at: 2026-02-13 10:30:00
→ 401 TokenRevokedError 응답
```
### 시나리오 4: Access Token 검증 (매 API 요청)
```
[AUTH-DEP] Access Token 검증 시작 - token: ...validAccess12345
[JWT] 토큰 디코딩 성공 - type: access, sub: 019c5452-..., exp: 1739450400
[AUTH-DEP] Access Token 검증 성공 - user_uuid: 019c5452-..., user_id: 42
```
### 시나리오 5: 로그아웃
```
[ROUTER] POST /auth/logout - user_id: 42, user_uuid: 019c5452-..., token: ...refreshToRevoke99
[AUTH] 로그아웃 - user_id: 42, token_hash: d5e6f7g8..., token: ...refreshToRevoke99
[AUTH] 로그아웃 완료 - user_id: 42
[ROUTER] POST /auth/logout 완료 - user_id: 42
```