o2o-castad-scheduler 깃 추적 제외 및 패키지 버전 업

- .gitignore에 o2o-castad-scheduler/ 추가
- sqlalchemy 버전 2.0.45 → 2.0.50으로 상향
- uv.lock 갱신

.gitignore

@@ -53,3 +53,6 @@ Dockerfile
 
 zzz/
 credentials/service_account.json
+
+# Scheduler (separate repo)
+o2o-castad-scheduler/

app/archive/api/routers/v1/archive.py

@@ -16,7 +16,9 @@ from app.user.dependencies.auth import get_current_user
 from app.user.models import User
 from app.utils.logger import get_logger
 from app.utils.pagination import PaginatedResponse
-from app.video.models import Video
+from app.comment.models import Comment
+from app.database.like_cache import get_like_counts, mset_like_counts
+from app.video.models import Video, VideoReaction
 from app.video.schemas.video_schema import VideoListItem
 
 logger = get_logger(__name__)
@@ -99,9 +101,22 @@ async def get_videos(
         total_result = await session.execute(count_query)
         total = total_result.scalar() or 0
 
-        # 쿼리 2: Video + Project 데이터 조회 (task_id별 최신 영상만)
+        # 쿼리 2: Video + Project + comment_count 조회 (like_count는 Redis에서)
+        comment_count_subq = (
+            select(func.count(Comment.id))
+            .where(
+                Comment.video_id == Video.id,
+                Comment.is_deleted == False,  # noqa: E712
+            )
+            .correlate(Video)
+            .scalar_subquery()
+        )
         data_query = (
-            select(Video, Project)
+            select(
+                Video,
+                Project,
+                comment_count_subq.label("comment_count"),
+            )
             .join(Project, Video.project_id == Project.id)
             .where(Video.id.in_(select(latest_video_ids.c.latest_id)))
             .order_by(Video.created_at.desc())
@@ -111,6 +126,29 @@ async def get_videos(
         result = await session.execute(data_query)
         rows = result.all()
 
+        # Redis mget으로 like_count 일괄 조회
+        video_ids = [video.id for video, project, _ in rows]
+        like_count_map = await get_like_counts(video_ids)
+
+        # 캐시 미스(None)인 video_id만 DB에서 보정
+        missing_ids = [vid for vid, cnt in like_count_map.items() if cnt is None]
+        if missing_ids:
+            db_counts = (await session.execute(
+                select(VideoReaction.video_id, func.count(VideoReaction.id))
+                .where(VideoReaction.video_id.in_(missing_ids))
+                .group_by(VideoReaction.video_id)
+            )).all()
+            db_found_ids = set()
+            batch = {}
+            for vid, cnt in db_counts:
+                batch[vid] = cnt
+                like_count_map[vid] = cnt
+                db_found_ids.add(vid)
+            await mset_like_counts(batch)
+            for vid in missing_ids:
+                if vid not in db_found_ids:
+                    like_count_map[vid] = 0
+
         # VideoListItem으로 변환
         items = [
             VideoListItem(
@@ -120,8 +158,10 @@ async def get_videos(
                 task_id=video.task_id,
                 result_movie_url=video.result_movie_url,
                 created_at=video.created_at,
+                like_count=like_count_map.get(video.id) or 0,
+                comment_count=comment_count or 0,
             )
-            for video, project in rows
+            for video, project, comment_count in rows
         ]
 
         response = PaginatedResponse.create(

app/comment/api/routers/v1/comment.py

@@ -0,0 +1,176 @@
+"""
+Comment API Router
+
+영상 댓글 관련 엔드포인트를 제공합니다.
+
+엔드포인트 목록:
+    - POST /comment/video/{video_id}: 댓글/대댓글 작성 (로그인 필수)
+    - GET  /comment/video/{video_id}: 댓글 목록 조회 (비로그인 허용)
+    - DELETE /comment/{comment_id}: 본인 댓글 소프트 삭제 (로그인 필수)
+"""
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.comment.schemas.comment_schema import (
+    CommentCreateRequest,
+    CommentCreateResponse,
+    CommentItem,
+    DeleteCommentResponse,
+)
+from app.comment.services.comment import create_comment, delete_comment, list_comments
+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, get_current_user_optional
+from app.user.models import User
+from app.utils.logger import get_logger
+from app.utils.pagination import PaginatedResponse
+
+logger = get_logger("comment")
+
+router = APIRouter(prefix="/comment", tags=["Comment"])
+
+
+@router.post(
+    "/video/{video_id}",
+    summary="댓글/대댓글 작성",
+    description="""
+## 개요
+영상에 댓글 또는 대댓글을 작성합니다. 로그인 필수.
+
+## 경로 파라미터
+- **video_id**: 댓글을 달 영상의 ID
+
+## 요청 본문
+- **content**: 댓글 본문 (1~100자)
+- **parent_id**: 대댓글일 때만 부모 댓글 id (생략 시 최상위 댓글)
+
+## 참고
+- 작성자 정보는 응답에 포함되지 않습니다 (익명 정책).
+- 대댓글에 또 대댓글을 다는 것은 불가합니다 (최대 2-depth).
+    """,
+    response_model=CommentCreateResponse,
+    responses={
+        200: {"description": "댓글 작성 성공"},
+        400: {"description": "잘못된 parent_id (2-depth 초과, 다른 영상의 댓글 등)"},
+        401: {"description": "인증 실패"},
+        404: {"description": "영상을 찾을 수 없음"},
+    },
+)
+async def post_comment(
+    video_id: int,
+    body: CommentCreateRequest,
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> CommentCreateResponse:
+    logger.info(
+        f"[post_comment] START - video_id: {video_id}, user: {current_user.user_uuid}, "
+        f"parent_id: {body.parent_id}"
+    )
+    comment = await create_comment(
+        session=session,
+        video_id=video_id,
+        user_uuid=current_user.user_uuid,
+        nickname=body.nickname,
+        content=body.content,
+        parent_id=body.parent_id,
+    )
+    logger.info(f"[post_comment] SUCCESS - comment_id: {comment.id}")
+    return CommentCreateResponse(
+        id=comment.id,
+        nickname=comment.nickname or "익명",
+        parent_id=comment.parent_id,
+        content=comment.content,
+        created_at=comment.created_at,
+    )
+
+
+@router.get(
+    "/video/{video_id}",
+    summary="댓글 목록 조회",
+    description="""
+## 개요
+영상의 댓글 목록을 페이지네이션하여 반환합니다. 비로그인도 접근 가능.
+
+## 경로 파라미터
+- **video_id**: 댓글을 조회할 영상의 ID
+
+## 쿼리 파라미터
+- **page**: 페이지 번호 (기본값: 1)
+- **page_size**: 페이지당 댓글 수 (기본값: 10, 최대: 100)
+
+## 참고
+- 최상위 댓글만 페이지네이션됩니다. 각 댓글의 대댓글은 전부 포함됩니다.
+- 작성자 정보는 노출되지 않으며, is_mine으로 본인 댓글 여부만 확인 가능합니다.
+- 삭제된 댓글은 content=null로 노출됩니다 (대댓글이 있는 경우).
+    """,
+    response_model=PaginatedResponse[CommentItem],
+    responses={
+        200: {"description": "댓글 목록 조회 성공"},
+        500: {"description": "조회 실패"},
+    },
+)
+async def get_comments(
+    video_id: int,
+    current_user: User | None = Depends(get_current_user_optional),
+    session: AsyncSession = Depends(get_session),
+    pagination: PaginationParams = Depends(get_pagination_params),
+) -> PaginatedResponse[CommentItem]:
+    logger.info(
+        f"[get_comments] START - video_id: {video_id}, "
+        f"page: {pagination.page}, page_size: {pagination.page_size}"
+    )
+    current_user_uuid = current_user.user_uuid if current_user else None
+    result = await list_comments(
+        session=session,
+        video_id=video_id,
+        page=pagination.page,
+        page_size=pagination.page_size,
+        current_user_uuid=current_user_uuid,
+    )
+    logger.info(f"[get_comments] SUCCESS - total: {result.total}, items: {len(result.items)}")
+    return result
+
+
+@router.delete(
+    "/{comment_id}",
+    summary="댓글 소프트 삭제",
+    description="""
+## 개요
+본인이 작성한 댓글을 소프트 삭제합니다. 로그인 필수.
+
+## 경로 파라미터
+- **comment_id**: 삭제할 댓글의 ID
+
+## 참고
+- 본인 댓글만 삭제 가능합니다.
+- 소프트 삭제 방식으로 DB에 데이터는 유지됩니다.
+- 부모 댓글 삭제 시 대댓글은 유지되며, 목록 조회 시 content=null로 표시됩니다.
+    """,
+    response_model=DeleteCommentResponse,
+    responses={
+        200: {"description": "삭제 성공"},
+        401: {"description": "인증 실패"},
+        403: {"description": "삭제 권한 없음"},
+        404: {"description": "댓글을 찾을 수 없음"},
+    },
+)
+async def remove_comment(
+    comment_id: int,
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> DeleteCommentResponse:
+    logger.info(
+        f"[remove_comment] START - comment_id: {comment_id}, user: {current_user.user_uuid}"
+    )
+    await delete_comment(
+        session=session,
+        comment_id=comment_id,
+        current_user_uuid=current_user.user_uuid,
+    )
+    logger.info(f"[remove_comment] SUCCESS - comment_id: {comment_id}")
+    return DeleteCommentResponse(
+        success=True,
+        comment_id=comment_id,
+        message="댓글이 삭제되었습니다.",
+    )

app/comment/models.py

@@ -0,0 +1,81 @@
+from datetime import datetime
+from typing import TYPE_CHECKING, List, Optional
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.database.session import Base
+
+if TYPE_CHECKING:
+    from app.user.models import User
+    from app.video.models import Video
+
+
+class Comment(Base):
+    """
+    영상 댓글 테이블
+
+    2-depth 구조 (최상위 댓글 + 대댓글 1단계).
+    parent_id가 NULL이면 최상위 댓글, 값이 있으면 대댓글.
+    작성자(user_uuid)는 DB에 저장하지만 API 응답에는 미노출 (익명 정책).
+    """
+
+    __tablename__ = "comment"
+    __table_args__ = (
+        Index("idx_comment_video_id", "video_id"),
+        Index("idx_comment_user_uuid", "user_uuid"),
+        Index("idx_comment_parent_id", "parent_id"),
+        Index("idx_comment_is_deleted", "is_deleted"),
+        {
+            "mysql_engine": "InnoDB",
+            "mysql_charset": "utf8mb4",
+            "mysql_collate": "utf8mb4_unicode_ci",
+        },
+    )
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, autoincrement=True, comment="고유 식별자"
+    )
+    video_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("video.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="연결된 Video의 id",
+    )
+    user_uuid: Mapped[str] = mapped_column(
+        ForeignKey("user.user_uuid", ondelete="CASCADE"),
+        nullable=False,
+        comment="작성자 UUID (응답 미노출, 권한 검증용)",
+    )
+    parent_id: Mapped[Optional[int]] = mapped_column(
+        Integer,
+        ForeignKey("comment.id", ondelete="CASCADE"),
+        nullable=True,
+        comment="NULL=최상위 댓글, 값=대댓글의 부모 id",
+    )
+    nickname: Mapped[Optional[str]] = mapped_column(
+        String(50), nullable=True, comment="댓글 작성자 닉네임 (null이면 익명)"
+    )
+    content: Mapped[str] = mapped_column(
+        String(100), nullable=False, comment="댓글 본문 (한글 기준 100자 이내)"
+    )
+    is_deleted: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, comment="소프트 삭제 여부"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        server_default=func.now(),
+        comment="작성 일시",
+    )
+
+    video: Mapped["Video"] = relationship("Video", back_populates="comments")
+    user: Mapped["User"] = relationship("User", back_populates="comments")
+    parent: Mapped[Optional["Comment"]] = relationship(
+        "Comment", remote_side=[id], back_populates="replies"
+    )
+    replies: Mapped[List["Comment"]] = relationship(
+        "Comment",
+        back_populates="parent",
+        cascade="all, delete-orphan",
+    )

app/comment/schemas/comment_schema.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class CommentCreateRequest(BaseModel):
+    nickname: Optional[str] = Field(None, min_length=1, max_length=50, description="작성자 닉네임 (미입력 시 익명)")
+    content: str = Field(..., min_length=1, max_length=100, description="댓글 본문 (한글 기준 100자 이내)")
+    parent_id: Optional[int] = Field(None, description="대댓글일 때만 부모 댓글 id")
+
+
+class ReplyItem(BaseModel):
+    """대댓글 응답"""
+
+    id: int = Field(..., description="댓글 고유 ID")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
+    is_deleted: bool = Field(..., description="삭제 여부")
+    is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
+    created_at: datetime = Field(..., description="작성 일시")
+
+
+class CommentItem(BaseModel):
+    """최상위 댓글 응답 — replies 포함"""
+
+    id: int = Field(..., description="댓글 고유 ID")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
+    is_deleted: bool = Field(..., description="삭제 여부")
+    is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
+    created_at: datetime = Field(..., description="작성 일시")
+    replies: List[ReplyItem] = Field(default_factory=list, description="대댓글 목록")
+
+
+class CommentCreateResponse(BaseModel):
+    id: int = Field(..., description="생성된 댓글 고유 ID")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    parent_id: Optional[int] = Field(None, description="부모 댓글 id (대댓글인 경우)")
+    content: str = Field(..., description="댓글 본문")
+    created_at: datetime = Field(..., description="작성 일시")
+
+
+class DeleteCommentResponse(BaseModel):
+    success: bool = Field(..., description="삭제 성공 여부")
+    comment_id: int = Field(..., description="삭제된 댓글 ID")
+    message: str = Field(..., description="결과 메시지")

app/comment/services/comment.py

@@ -0,0 +1,189 @@
+from collections import defaultdict
+from typing import List, Optional
+
+from fastapi import HTTPException
+from sqlalchemy import exists, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.comment.models import Comment
+from app.comment.schemas.comment_schema import CommentItem, ReplyItem
+from app.utils.pagination import PaginatedResponse
+from app.video.models import Video
+
+
+async def _validate_parent(
+    session: AsyncSession,
+    parent_id: int,
+    video_id: int,
+) -> None:
+    """2-depth 제한 + 동일 video 검증."""
+    result = await session.execute(
+        select(Comment).where(
+            Comment.id == parent_id,
+            Comment.is_deleted == False,  # noqa: E712
+        )
+    )
+    parent = result.scalar_one_or_none()
+
+    if parent is None:
+        raise HTTPException(status_code=400, detail="부모 댓글을 찾을 수 없습니다.")
+    if parent.video_id != video_id:
+        raise HTTPException(status_code=400, detail="다른 영상의 댓글에는 대댓글을 달 수 없습니다.")
+    if parent.parent_id is not None:
+        raise HTTPException(status_code=400, detail="대댓글에는 대댓글을 달 수 없습니다. (최대 2-depth)")
+
+
+def _build_comment_items(
+    parents: list,
+    replies_map: dict,
+    current_user_uuid: Optional[str],
+) -> List[CommentItem]:
+    items = []
+    for c in parents:
+        raw_replies = replies_map.get(c.id, [])
+        replies = [
+            ReplyItem(
+                id=r.id,
+                nickname=r.nickname or "익명",
+                content=None if r.is_deleted else r.content,
+                is_deleted=r.is_deleted,
+                is_mine=(current_user_uuid == r.user_uuid) if current_user_uuid else False,
+                created_at=r.created_at,
+            )
+            for r in raw_replies
+        ]
+        items.append(
+            CommentItem(
+                id=c.id,
+                nickname=c.nickname or "익명",
+                content=None if c.is_deleted else c.content,
+                is_deleted=c.is_deleted,
+                is_mine=(current_user_uuid == c.user_uuid) if current_user_uuid else False,
+                created_at=c.created_at,
+                replies=replies,
+            )
+        )
+    return items
+
+
+async def create_comment(
+    session: AsyncSession,
+    video_id: int,
+    user_uuid: str,
+    nickname: str,
+    content: str,
+    parent_id: Optional[int],
+) -> Comment:
+    # Video 존재 확인
+    video_result = await session.execute(
+        select(Video).where(
+            Video.id == video_id,
+            Video.status == "completed",
+            Video.is_deleted == False,  # noqa: E712
+        )
+    )
+    if video_result.scalar_one_or_none() is None:
+        raise HTTPException(status_code=404, detail="영상을 찾을 수 없습니다.")
+
+    # parent_id 검증
+    if parent_id is not None:
+        await _validate_parent(session, parent_id, video_id)
+
+    comment = Comment(
+        video_id=video_id,
+        user_uuid=user_uuid,
+        nickname=nickname,
+        parent_id=parent_id,
+        content=content,
+    )
+    session.add(comment)
+    await session.commit()
+    await session.refresh(comment)
+    return comment
+
+
+async def list_comments(
+    session: AsyncSession,
+    video_id: int,
+    page: int,
+    page_size: int,
+    current_user_uuid: Optional[str],
+) -> PaginatedResponse[CommentItem]:
+    offset = (page - 1) * page_size
+
+    # 살아있는 자식이 있는지 확인하는 서브쿼리
+    has_live_reply = (
+        exists()
+        .where(
+            Comment.parent_id == Comment.id,
+            Comment.is_deleted == False,  # noqa: E712
+        )
+        .correlate(Comment)
+    )
+
+    # 최상위 댓글 필터: 삭제 안 됐거나 살아있는 대댓글이 있는 것
+    parent_where = [
+        Comment.video_id == video_id,
+        Comment.parent_id.is_(None),
+        (Comment.is_deleted == False) | has_live_reply,  # noqa: E712
+    ]
+
+    from sqlalchemy import func
+
+    count_q = select(func.count(Comment.id)).where(*parent_where)
+    total = (await session.execute(count_q)).scalar() or 0
+
+    parents_q = (
+        select(Comment)
+        .where(*parent_where)
+        .order_by(Comment.created_at.desc())
+        .offset(offset)
+        .limit(page_size)
+    )
+    parents = (await session.execute(parents_q)).scalars().all()
+
+    replies_map: dict = defaultdict(list)
+    if parents:
+        parent_ids = [c.id for c in parents]
+        replies_q = (
+            select(Comment)
+            .where(
+                Comment.parent_id.in_(parent_ids),
+                Comment.is_deleted == False,  # noqa: E712
+            )
+            .order_by(Comment.created_at.asc())
+        )
+        replies = (await session.execute(replies_q)).scalars().all()
+        for r in replies:
+            replies_map[r.parent_id].append(r)
+
+    items = _build_comment_items(list(parents), replies_map, current_user_uuid)
+
+    return PaginatedResponse.create(
+        items=items,
+        total=total,
+        page=page,
+        page_size=page_size,
+    )
+
+
+async def delete_comment(
+    session: AsyncSession,
+    comment_id: int,
+    current_user_uuid: str,
+) -> None:
+    result = await session.execute(
+        select(Comment).where(
+            Comment.id == comment_id,
+            Comment.is_deleted == False,  # noqa: E712
+        )
+    )
+    comment = result.scalar_one_or_none()
+
+    if comment is None:
+        raise HTTPException(status_code=404, detail="댓글을 찾을 수 없습니다.")
+    if comment.user_uuid != current_user_uuid:
+        raise HTTPException(status_code=403, detail="삭제 권한이 없습니다.")
+
+    comment.is_deleted = True
+    await session.commit()

app/core/common.py

@@ -51,6 +51,9 @@ async def lifespan(app: FastAPI):
     await close_shared_client()
     await close_shared_blob_client()
 
+    from app.database.like_cache import close_like_cache
+    await close_like_cache()
+
     # 데이터베이스 엔진 종료
     from app.database.session import dispose_engine
 

app/database/like_cache.py

@@ -0,0 +1,235 @@
+"""
+좋아요 Redis 캐시 클라이언트
+
+Write-Behind 패턴 적용:
+- 토글 시 Redis를 즉시 업데이트하고 dirty SET에 표시
+- 스케줄러가 1분마다 dirty 항목을 MySQL에 bulk write
+
+Key 패턴:
+- video:like:count:{video_id}       INT    — 좋아요 카운트
+- video:like:users:{video_id}       SET    — 좋아요 누른 user_uuid 목록
+- video:reaction:dirty              SET    — DB 동기화 대기 "{video_id}:{user_uuid}"
+- video:reaction:dirty:processing   SET    — 플러시 중 임시 (크래시 복구용)
+
+캐시 미스(Redis 재시작 등) 시 호출부에서 DB 조회 후 backfill_user_set() / set_like_count()로 복구합니다.
+"""
+
+import redis.asyncio as aioredis
+
+from config import db_settings
+
+_client: aioredis.Redis | None = None
+
+# 원자적 토글 Lua 스크립트 — 동시 더블클릭 race condition 방지
+_TOGGLE_LIKE_SCRIPT = """
+local user_key = KEYS[1]
+local count_key = KEYS[2]
+local user_uuid = ARGV[1]
+
+if redis.call('SISMEMBER', user_key, user_uuid) == 1 then
+    redis.call('SREM', user_key, user_uuid)
+    local c = tonumber(redis.call('DECR', count_key))
+    if c < 0 then
+        redis.call('SET', count_key, 0)
+        c = 0
+    end
+    return {0, c}
+else
+    redis.call('SADD', user_key, user_uuid)
+    local c = tonumber(redis.call('INCR', count_key))
+    return {1, c}
+end
+"""
+
+_DIRTY_KEY = "video:reaction:dirty"
+_DIRTY_PROCESSING_KEY = "video:reaction:dirty:processing"
+
+
+def get_like_cache() -> aioredis.Redis:
+    global _client
+    if _client is None:
+        _client = aioredis.Redis(
+            host=db_settings.REDIS_HOST,
+            port=db_settings.REDIS_PORT,
+            db=2,
+            decode_responses=True,
+        )
+    return _client
+
+
+async def close_like_cache() -> None:
+    global _client
+    if _client:
+        await _client.aclose()
+        _client = None
+
+
+# ──────────────────────────────────────────────
+# Key 헬퍼
+# ──────────────────────────────────────────────
+
+def _key(video_id: int) -> str:
+    return f"video:like:count:{video_id}"
+
+
+def _user_key(video_id: int) -> str:
+    return f"video:like:users:{video_id}"
+
+
+# ──────────────────────────────────────────────
+# 카운트 (기존 API 유지)
+# ──────────────────────────────────────────────
+
+async def get_like_count(video_id: int) -> int | None:
+    """Redis에서 like_count 조회. 캐시 미스 시 None 반환."""
+    val = await get_like_cache().get(_key(video_id))
+    if val is None:
+        return None
+    return max(int(val), 0)
+
+
+async def get_like_counts(video_ids: list[int]) -> dict[int, int | None]:
+    """여러 영상의 like_count를 한 번에 조회 (mget).
+    캐시 미스인 video_id는 None으로 반환."""
+    if not video_ids:
+        return {}
+    keys = [_key(vid) for vid in video_ids]
+    values = await get_like_cache().mget(*keys)
+    return {
+        vid: max(int(v), 0) if v is not None else None
+        for vid, v in zip(video_ids, values)
+    }
+
+
+async def set_like_count(video_id: int, count: int) -> None:
+    """like_count를 Redis에 저장 (음수 방지)."""
+    await get_like_cache().set(_key(video_id), max(count, 0))
+
+
+async def mset_like_counts(counts: dict[int, int]) -> None:
+    """여러 영상의 like_count를 한 번에 저장 (mset)."""
+    if not counts:
+        return
+    await get_like_cache().mset({_key(vid): max(cnt, 0) for vid, cnt in counts.items()})
+
+
+async def incr_like_count(video_id: int) -> int:
+    """like_count를 1 증가 후 반환."""
+    return max(int(await get_like_cache().incr(_key(video_id))), 0)
+
+
+async def decr_like_count(video_id: int) -> int:
+    """like_count를 1 감소 후 반환 (음수 방지)."""
+    count = int(await get_like_cache().decr(_key(video_id)))
+    if count < 0:
+        await get_like_cache().set(_key(video_id), 0)
+        return 0
+    return count
+
+
+# ──────────────────────────────────────────────
+# 유저 SET (is_liked_by_me source of truth)
+# ──────────────────────────────────────────────
+
+async def toggle_like_atomic(video_id: int, user_uuid: str) -> tuple[bool, int]:
+    """Lua 스크립트로 원자적 좋아요 토글.
+
+    Returns:
+        (is_liked, new_count) 튜플
+    """
+    result = await get_like_cache().eval(
+        _TOGGLE_LIKE_SCRIPT,
+        2,
+        _user_key(video_id),
+        _key(video_id),
+        user_uuid,
+    )
+    return bool(result[0]), int(result[1])
+
+
+async def is_user_liked(video_id: int, user_uuid: str) -> bool | None:
+    """Redis user-set에서 좋아요 여부 조회.
+
+    Returns:
+        True/False: 조회 성공
+        None: user-set 키가 없음 (cold-start backfill 필요 신호)
+    """
+    client = get_like_cache()
+    key = _user_key(video_id)
+    if not await client.exists(key):
+        return None
+    return bool(await client.sismember(key, user_uuid))
+
+
+async def is_user_set_exists(video_id: int) -> bool:
+    """Redis user-set 키 존재 여부 확인."""
+    return bool(await get_like_cache().exists(_user_key(video_id)))
+
+
+async def bulk_is_user_liked(
+    video_ids: list[int], user_uuid: str
+) -> dict[int, bool | None]:
+    """여러 영상의 is_liked 여부를 한 번에 조회 (pipeline).
+
+    Returns:
+        {video_id: True/False} — user-set 키가 없는 영상은 None
+    """
+    if not video_ids:
+        return {}
+    client = get_like_cache()
+    async with client.pipeline(transaction=False) as pipe:
+        for vid in video_ids:
+            pipe.exists(_user_key(vid))
+            pipe.sismember(_user_key(vid), user_uuid)
+        responses = await pipe.execute()
+
+    return {
+        vid: (bool(responses[i * 2 + 1]) if responses[i * 2] else None)
+        for i, vid in enumerate(video_ids)
+    }
+
+
+async def backfill_user_set(video_id: int, user_uuids: list[str]) -> None:
+    """DB에서 가져온 유저 목록을 Redis SET에 일괄 적재."""
+    if user_uuids:
+        await get_like_cache().sadd(_user_key(video_id), *user_uuids)
+
+
+# ──────────────────────────────────────────────
+# Dirty SET (Write-Behind 큐)
+# ──────────────────────────────────────────────
+
+async def mark_dirty(video_id: int, user_uuid: str) -> None:
+    """DB 동기화 대기 목록에 추가."""
+    await get_like_cache().sadd(_DIRTY_KEY, f"{video_id}:{user_uuid}")
+
+
+async def drain_dirty() -> list[tuple[int, str]]:
+    """dirty SET을 processing으로 RENAME 후 전체 반환.
+
+    이전 실행 중 크래시로 남은 processing 항목은 먼저 병합하여 유실 방지.
+    """
+    client = get_like_cache()
+
+    # 이전 크래시 잔여 항목 병합
+    if await client.exists(_DIRTY_PROCESSING_KEY):
+        await client.sunionstore(_DIRTY_KEY, _DIRTY_KEY, _DIRTY_PROCESSING_KEY)
+        await client.delete(_DIRTY_PROCESSING_KEY)
+
+    if not await client.exists(_DIRTY_KEY):
+        return []
+
+    # RENAME으로 플러시 중 새로 들어오는 토글과 분리
+    await client.rename(_DIRTY_KEY, _DIRTY_PROCESSING_KEY)
+    members = await client.smembers(_DIRTY_PROCESSING_KEY)
+
+    result = []
+    for member in members:
+        vid_str, user_uuid = member.split(":", 1)
+        result.append((int(vid_str), user_uuid))
+    return result
+
+
+async def commit_dirty_processing() -> None:
+    """DB 반영 완료 후 processing SET 삭제."""
+    await get_like_cache().delete(_DIRTY_PROCESSING_KEY)

app/home/api/routers/v1/home.py

@@ -42,31 +42,41 @@ from config import MEDIA_ROOT
 # 로거 설정
 logger = get_logger("home")
 
-# 전국 시 이름 목록 (roadAddress에서 region 추출용)
+# 전국 시/군 이름 목록 (roadAddress에서 region 추출용)
 # fmt: off
 KOREAN_CITIES = [
     # 특별시/광역시
     "서울시", "부산시", "대구시", "인천시", "광주시", "대전시", "울산시", "세종시",
     # 경기도
     "수원시", "성남시", "고양시", "용인시", "부천시", "안산시", "안양시", "남양주시",
-    "화성시", "평택시", "의정부시", "시흥시", "파주시", "광명시", "김포시", "군포시",
-    "광주시", "이천시", "양주시", "오산시", "구리시", "안성시", "포천시", "의왕시",
-    "하남시", "여주시", "동두천시", "과천시",
-    # 강원도
+    "화성시", "평택시", "의정부시", "시흥시", "파주시", "김포시", "광주시", "광명시",
+    "군포시", "하남시", "오산시", "이천시", "안성시", "구리시", "양주시", "포천시",
+    "여주시", "동두천시", "과천시", "가평군", "양평군", "연천군",
+    # 강원특별자치도
     "춘천시", "원주시", "강릉시", "동해시", "태백시", "속초시", "삼척시",
+    "홍천군", "횡성군", "영월군", "평창군", "정선군", "철원군", "화천군",
+    "양구군", "인제군", "고성군", "양양군",
     # 충청북도
     "청주시", "충주시", "제천시",
+    "보은군", "옥천군", "영동군", "증평군", "진천군", "괴산군", "음성군", "단양군",
     # 충청남도
     "천안시", "공주시", "보령시", "아산시", "서산시", "논산시", "계룡시", "당진시",
-    # 전라북도
+    "금산군", "부여군", "서천군", "청양군", "홍성군", "예산군", "태안군",
+    # 전북특별자치도
     "전주시", "군산시", "익산시", "정읍시", "남원시", "김제시",
+    "완주군", "진안군", "무주군", "장수군", "임실군", "순창군", "고창군", "부안군",
     # 전라남도
     "목포시", "여수시", "순천시", "나주시", "광양시",
+    "담양군", "곡성군", "구례군", "고흥군", "보성군", "화순군", "장흥군", "강진군",
+    "해남군", "영암군", "무안군", "함평군", "영광군", "장성군", "완도군", "진도군", "신안군",
     # 경상북도
     "포항시", "경주시", "김천시", "안동시", "구미시", "영주시", "영천시", "상주시", "문경시", "경산시",
+    "의성군", "청송군", "영양군", "영덕군", "청도군", "고령군", "성주군", "칠곡군",
+    "예천군", "봉화군", "울진군", "울릉군",
     # 경상남도
     "창원시", "진주시", "통영시", "사천시", "김해시", "밀양시", "거제시", "양산시",
-    # 제주도
+    "의령군", "함안군", "창녕군", "고성군", "남해군", "하동군", "산청군", "함양군", "거창군", "합천군",
+    # 제주특별자치도
     "제주시", "서귀포시",
 ]
 # fmt: on
@@ -116,13 +126,39 @@ async def search_accommodation(
     )
 
 
+METRO_CITY_MAP = {
+    "서울": "서울시", "부산": "부산시", "대구": "대구시",
+    "인천": "인천시", "광주": "광주시", "대전": "대전시",
+    "울산": "울산시", "세종": "세종시",
+}
+
+
 def _extract_region_from_address(road_address: str | None) -> str:
-    """roadAddress에서 시 이름 추출"""
+    """roadAddress에서 시/군 이름 추출
+
+    매칭 우선순위:
+    1. KOREAN_CITIES 직접 매칭 (시/군 접미사 포함)
+    2. KOREAN_CITIES 접미사 생략 매칭
+    3. 주소 두 번째 토큰이 시/군으로 끝나는 경우 (예: "전북 군산시 ...")
+    4. 주소 두 번째 토큰이 구/동인 경우 → 첫 번째 토큰으로 광역시 매핑 (예: "서울 강남구 ...")
+    """
     if not road_address:
         return ""
+
     for city in KOREAN_CITIES:
         if city in road_address:
             return city
+        if city[:-1] in road_address:
+            return city
+
+    tokens = road_address.split()
+    if len(tokens) >= 2:
+        second = tokens[1]
+        if second.endswith("시") or second.endswith("군"):
+            return second
+        if second.endswith("구") or second.endswith("동"):
+            return METRO_CITY_MAP.get(tokens[0], "")
+
     return ""
 
 
@@ -254,7 +290,7 @@ async def _crawling_logic(
     marketing_analysis = None
 
     if scraper.base_info:
-        road_address = scraper.base_info.get("roadAddress", "")
+        road_address = scraper.base_info.get("roadAddress", "") or scraper.base_info.get("address", "")
         customer_name = scraper.base_info.get("name", "")
         region = _extract_region_from_address(road_address)
 

app/lyric/api/routers/v1/lyric.py

@@ -40,6 +40,7 @@ from app.lyric.schemas.lyric import (
     LyricDetailResponse,
     LyricListItem,
     LyricStatusResponse,
+    SubtitleStatusResponse,
 )
 from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background
 from app.utils.prompts.chatgpt_prompt import ChatgptService
@@ -515,6 +516,86 @@ async def list_lyrics(
     )
 
 
+@router.get(
+    "/subtitle/status/{task_id}",
+    summary="자막 생성 상태 조회",
+    description="""
+자막(subtitle) 생성 완료 여부를 조회합니다.
+
+## 인증
+**Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다.
+
+## 경로 파라미터
+- **task_id**: 프로젝트 task_id (필수)
+
+## 상태 값
+- **pending**: 자막 생성 진행 중 — 잠시 후 재요청
+- **completed**: 자막 생성 완료 — `/video/generate/{task_id}` 호출 가능
+
+## 사용 예시 (cURL)
+```bash
+curl -X GET "http://localhost:8000/lyric/subtitle/status/019123ab-cdef-7890-abcd-ef1234567890" \\
+  -H "Authorization: Bearer {access_token}"
+```
+
+## 참고
+- 자막은 `/lyric/generate` 호출 시 백그라운드에서 자동 생성됩니다.
+- 클라이언트는 `completed` 상태 확인 후 `/video/generate`를 호출해야 합니다.
+    """,
+    response_model=SubtitleStatusResponse,
+    responses={
+        200: {"description": "상태 조회 성공"},
+        401: {"description": "인증 실패 (토큰 없음/만료)"},
+        404: {"description": "해당 task_id에 해당하는 프로젝트를 찾을 수 없음"},
+    },
+)
+async def get_subtitle_status(
+    task_id: str,
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> SubtitleStatusResponse:
+    """task_id로 자막 생성 상태를 조회합니다."""
+    logger.info(f"[get_subtitle_status] START - task_id: {task_id}")
+
+    project_result = await session.execute(
+        select(Project)
+        .where(Project.task_id == task_id)
+        .order_by(Project.created_at.desc())
+        .limit(1)
+    )
+    project = project_result.scalar_one_or_none()
+    if not project:
+        raise HTTPException(
+            status_code=404,
+            detail=f"task_id '{task_id}'에 해당하는 프로젝트를 찾을 수 없습니다.",
+        )
+
+    marketing_result = await session.execute(
+        select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
+    )
+    intel = marketing_result.scalar_one_or_none()
+    if not intel:
+        raise HTTPException(
+            status_code=404,
+            detail=f"task_id '{task_id}'에 해당하는 마케팅 인텔리전스를 찾을 수 없습니다.",
+        )
+
+    if intel.subtitle:
+        logger.info(f"[get_subtitle_status] completed - task_id: {task_id}")
+        return SubtitleStatusResponse(
+            task_id=task_id,
+            status="completed",
+            message="자막 생성이 완료되었습니다.",
+        )
+
+    logger.info(f"[get_subtitle_status] pending - task_id: {task_id}")
+    return SubtitleStatusResponse(
+        task_id=task_id,
+        status="pending",
+        message="자막 생성이 진행 중입니다. 잠시 후 다시 확인해주세요.",
+    )
+
+
 @router.get(
     "/{task_id}",
     summary="가사 상세 조회",

app/lyric/schemas/lyric.py

@@ -202,6 +202,55 @@ class LyricDetailResponse(BaseModel):
     created_at: Optional[datetime] = Field(None, description="생성 일시")
 
 
+class SubtitleStatusResponse(BaseModel):
+    """자막 생성 상태 조회 응답 스키마
+
+    Usage:
+        GET /subtitle/status/{task_id}
+        클라이언트가 subtitle 완료 여부를 polling할 때 사용합니다.
+
+    Status Values:
+        - pending: 자막 생성 진행 중 (재시도 필요)
+        - completed: 자막 생성 완료 (/video/generate 호출 가능)
+        - failed: 자막 생성 실패 (/lyric/generate 재호출 필요)
+    """
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "summary": "생성 중",
+                    "value": {
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "status": "pending",
+                        "message": "자막 생성이 진행 중입니다. 잠시 후 다시 확인해주세요.",
+                    },
+                },
+                {
+                    "summary": "완료",
+                    "value": {
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "status": "completed",
+                        "message": "자막 생성이 완료되었습니다.",
+                    },
+                },
+                {
+                    "summary": "실패",
+                    "value": {
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "status": "failed",
+                        "message": "자막 생성에 실패했습니다. 다시 시도해주세요.",
+                    },
+                },
+            ]
+        }
+    )
+
+    task_id: str = Field(..., description="작업 고유 식별자")
+    status: Literal["pending", "completed", "failed"] = Field(..., description="자막 생성 상태")
+    message: str = Field(..., description="상태 메시지")
+
+
 class LyricListItem(BaseModel):
     """가사 목록 아이템 스키마
 

app/lyric/worker/lyric_task.py

@@ -158,9 +158,13 @@ async def generate_lyric_background(
 
 async def generate_subtitle_background(
     orientation: str,
-    task_id: str
+    task_id: str,
+    max_retries: int = 3,
 ) -> None:
-    logger.info(f"[generate_subtitle_background] task_id: {task_id}, {orientation}")
+    logger.info(f"[generate_subtitle_background] START - task_id: {task_id}, orientation: {orientation}")
+
+    for attempt in range(1, max_retries + 1):
+        try:
             creatomate_service = CreatomateService(orientation=orientation)
             template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
             pitchings = creatomate_service.extract_text_format_from_template(template)
@@ -182,7 +186,7 @@ async def generate_subtitle_background(
 
             store_address = project.detail_region_info
             customer_name = project.store_name
-    logger.info(f"[generate_subtitle_background] customer_name: {customer_name}, {store_address}")
+            logger.info(f"[generate_subtitle_background] customer_name: {customer_name}, store_address: {store_address}")
 
             generated_subtitles = await subtitle_generator.generate_subtitle_contents(
                 marketing_intelligence=marketing_intelligence.intel_result,
@@ -192,7 +196,10 @@ async def generate_subtitle_background(
             )
             pitching_output_list = generated_subtitles.pitching_results
 
-    subtitle_modifications = {pitching_output.pitching_tag : pitching_output.pitching_data for pitching_output in pitching_output_list}
+            subtitle_modifications = {
+                pitching_output.pitching_tag: pitching_output.pitching_data
+                for pitching_output in pitching_output_list
+            }
             logger.info(f"[generate_subtitle_background] subtitle_modifications: {subtitle_modifications}")
 
             async with BackgroundSessionLocal() as session:
@@ -202,8 +209,16 @@ async def generate_subtitle_background(
                 marketing_intelligence = marketing_result.scalar_one_or_none()
                 marketing_intelligence.subtitle = subtitle_modifications
                 await session.commit()
-    logger.info(f"[generate_subtitle_background] task_id: {task_id} DONE")
-    
-
 
+            logger.info(f"[generate_subtitle_background] DONE - task_id: {task_id} (attempt {attempt}/{max_retries})")
             return
+
+        except Exception as e:
+            logger.error(
+                f"[generate_subtitle_background] FAILED (attempt {attempt}/{max_retries}) - task_id: {task_id}, error: {e}",
+                exc_info=True,
+            )
+            if attempt < max_retries:
+                logger.info(f"[generate_subtitle_background] 재시도 중... ({attempt + 1}/{max_retries}) - task_id: {task_id}")
+
+    logger.error(f"[generate_subtitle_background] 모든 재시도 실패 - task_id: {task_id}")

app/user/models.py

@@ -16,8 +16,10 @@ from app.database.session import Base
 
 
 if TYPE_CHECKING:
+    from app.comment.models import Comment
     from app.credit.models import CreditChargeRequest, CreditTransaction
     from app.home.models import Project
+    from app.video.models import VideoReaction
 
 
 class User(Base):
@@ -295,6 +297,20 @@ class User(Base):
         lazy="noload",
     )
 
+    comments: Mapped[List["Comment"]] = relationship(
+        "Comment",
+        back_populates="user",
+        cascade="all, delete-orphan",
+        lazy="noload",
+    )
+
+    video_reactions: Mapped[List["VideoReaction"]] = relationship(
+        "VideoReaction",
+        back_populates="user",
+        cascade="all, delete-orphan",
+        lazy="noload",
+    )
+
     def __repr__(self) -> str:
         return (
             f"<User("

app/utils/nvMapPwScraper.py

@@ -1,5 +1,6 @@
 import asyncio
 import re
+from html import unescape
 from difflib import SequenceMatcher
 from playwright.async_api import async_playwright
 from urllib import parse
@@ -99,7 +100,9 @@ patchedGetter.toString();''')
 
     @staticmethod
     def _clean_title(text: str) -> str:
-        return re.sub(r"<.*?>", "", text).strip()
+        text = unescape(text)                    # HTML 엔티티 디코딩 (&amp; → &)
+        text = re.sub(r"<.*?>", "", text)        # HTML 태그 제거
+        return text.strip()
 
     @staticmethod
     def _similarity(a: str, b: str) -> float:

app/utils/prompts/prompts.py

@@ -9,8 +9,7 @@ from functools import lru_cache
 logger = get_logger("prompt")
 
 _SCOPES = [
-    "https://www.googleapis.com/auth/spreadsheets.readonly", 
-    "https://www.googleapis.com/auth/drive.readonly"
+    "https://www.googleapis.com/auth/spreadsheets.readonly"
 ]
 
 class Prompt():

app/utils/subtitles.py

@@ -10,11 +10,20 @@ from app.utils.prompts.chatgpt_prompt import ChatgptService
 from app.utils.prompts.schemas import *
 from app.utils.prompts.prompts import *
 
+logger = get_logger("subtitle")
+
 class SubtitleContentsGenerator():
     def __init__(self):
-        self.chatgpt_service = ChatgptService()
+        self.chatgpt_service = ChatgptService(timeout=60.0)
 
     async def generate_subtitle_contents(self, marketing_intelligence : dict[str, Any], pitching_label_list : list[Any], customer_name : str, detail_region_info : str) -> SubtitlePromptOutput:
+        start = time.perf_counter()
+        logger.info(
+            f"[SubtitleContentsGenerator] START - customer: {customer_name}, "
+            f"pitching_count: {len(pitching_label_list)}, "
+            f"labels: {pitching_label_list}"
+        )
+
         dynamic_subtitle_prompt = create_dynamic_subtitle_prompt(len(pitching_label_list))
         pitching_label_string = "\n".join(pitching_label_list)
         marketing_intel_string = json.dumps(marketing_intelligence, ensure_ascii=False)
@@ -24,7 +33,17 @@ class SubtitleContentsGenerator():
             "customer_name" : customer_name,
             "detail_region_info" : detail_region_info,
             }
+
+        logger.info(
+            f"[SubtitleContentsGenerator] GPT 호출 시작 - model: {dynamic_subtitle_prompt.prompt_model}"
+        )
         output_data = await self.chatgpt_service.generate_structured_output(dynamic_subtitle_prompt, input_data)
+
+        elapsed = (time.perf_counter() - start) * 1000
+        logger.info(
+            f"[SubtitleContentsGenerator] DONE - 소요시간: {elapsed:.0f}ms, "
+            f"결과: {[r.pitching_tag for r in output_data.pitching_results]}"
+        )
         return output_data
     
 

app/video/api/routers/internal/reactions.py

@@ -0,0 +1,98 @@
+"""
+내부 전용 좋아요 반응 플러시 API
+
+스케줄러가 1분마다 호출하여 Redis dirty SET의 좋아요 토글을 MySQL에 bulk write합니다.
+X-Internal-Secret 헤더로 인증합니다.
+"""
+
+import logging
+
+from fastapi import APIRouter, Depends, Header, HTTPException, status
+from sqlalchemy import delete, insert, tuple_
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.like_cache import (
+    commit_dirty_processing,
+    drain_dirty,
+    is_user_liked,
+)
+from app.database.session import get_session
+from app.video.models import VideoReaction
+from config import internal_settings
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/internal/video", tags=["Internal"])
+
+
+@router.post(
+    "/reactions/flush",
+    summary="[내부] 좋아요 반응 DB 플러시",
+    description="스케줄러 서버에서 1분마다 호출하는 내부 전용 엔드포인트입니다. "
+                "Redis dirty SET의 항목을 MySQL video_reaction 테이블에 bulk write합니다.",
+)
+async def flush_reactions(
+    session: AsyncSession = Depends(get_session),
+    x_internal_secret: str = Header(...),
+) -> dict:
+    """Redis dirty SET → MySQL bulk write.
+
+    1. drain_dirty(): dirty SET을 processing으로 RENAME 후 항목 조회
+    2. 각 항목의 현재 Redis 상태(is_liked) 확인
+    3. is_liked=True → INSERT IGNORE, is_liked=False → DELETE
+    4. commit_dirty_processing(): processing SET 삭제
+    """
+    if x_internal_secret != internal_settings.INTERNAL_SECRET_KEY:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Invalid internal secret",
+        )
+
+    pairs = await drain_dirty()
+    if not pairs:
+        logger.info("[REACTION_FLUSH] dirty 항목 없음, 종료")
+        return {"flushed": 0, "adds": 0, "dels": 0}
+
+    logger.info(f"[REACTION_FLUSH] START - dirty 항목 {len(pairs)}건")
+
+    adds: list[dict] = []
+    dels: list[tuple[int, str]] = []
+
+    # Redis 현재 상태 기준으로 add / delete 분류
+    for video_id, user_uuid in pairs:
+        liked = await is_user_liked(video_id, user_uuid)
+        if liked:
+            adds.append({"video_id": video_id, "user_uuid": user_uuid})
+        else:
+            dels.append((video_id, user_uuid))
+
+    try:
+        # Bulk INSERT IGNORE — UniqueConstraint 보장으로 멱등 처리
+        if adds:
+            await session.execute(
+                insert(VideoReaction).prefix_with("IGNORE").values(adds)
+            )
+
+        # Bulk DELETE
+        if dels:
+            await session.execute(
+                delete(VideoReaction).where(
+                    tuple_(
+                        VideoReaction.video_id,
+                        VideoReaction.user_uuid,
+                    ).in_(dels)
+                )
+            )
+
+        await session.commit()
+        await commit_dirty_processing()
+
+        logger.info(
+            f"[REACTION_FLUSH] SUCCESS - adds: {len(adds)}, dels: {len(dels)}"
+        )
+        return {"flushed": len(pairs), "adds": len(adds), "dels": len(dels)}
+
+    except Exception as e:
+        await session.rollback()
+        logger.error(f"[REACTION_FLUSH] EXCEPTION - error: {e}")
+        raise HTTPException(status_code=500, detail=f"플러시 실패: {str(e)}")

app/video/api/routers/v1/video.py

@@ -14,29 +14,48 @@ Video API Router
 """
 
 import json
-import asyncio
-
+from collections import defaultdict
 from typing import Literal
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
-from sqlalchemy import select
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.database.session import get_session
-from app.user.dependencies.auth import get_current_user
+from app.dependencies.pagination import PaginationParams, get_pagination_params
+from app.user.dependencies.auth import get_current_user, get_current_user_optional
 from app.user.models import User
+from app.utils.pagination import PaginatedResponse
 from app.home.models import Image, Project, MarketingIntel, ImageTag
+from app.home.api.routers.v1.home import _extract_region_from_address
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
 from app.utils.creatomate import CreatomateService
 from app.utils.subtitles import SubtitleContentsGenerator
+
+from app.comment.models import Comment
+from app.database.like_cache import (
+    backfill_user_set,
+    bulk_is_user_liked,
+    get_like_count,
+    get_like_counts,
+    is_user_liked,
+    is_user_set_exists,
+    mark_dirty,
+    mset_like_counts,
+    set_like_count,
+    toggle_like_atomic,
+)
 from app.utils.logger import get_logger
-from app.video.models import Video
+from app.video.models import Video, VideoReaction
 from app.video.schemas.video_schema import (
     DownloadVideoResponse,
     GenerateVideoResponse,
+    LikeToggleResponse,
     PollingVideoResponse,
+    VideoDetailResponse,
     VideoRenderData,
+    VideoThumbnailItem,
 )
 from app.video.worker.video_task import download_and_upload_video_to_blob
 from app.video.services.video import get_image_tags_by_task_id
@@ -148,37 +167,9 @@ async def generate_video(
     image_urls: list[str] = []
 
     try:
-        subtitle_done = False
-        count = 0
-        async with AsyncSessionLocal() as session:
-            project_result = await session.execute(
-                select(Project)
-                .where(Project.task_id == task_id)
-                .order_by(Project.created_at.desc())
-                .limit(1)
-            )
-            project = project_result.scalar_one_or_none()
-
-        while not subtitle_done:
-            async with AsyncSessionLocal() as session:
-                logger.info(f"[generate_video] Checking subtitle- task_id: {task_id}, count : {count}")
-                marketing_result = await session.execute(
-                    select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
-                )
-                marketing_intelligence = marketing_result.scalar_one_or_none()
-                subtitle_done = bool(marketing_intelligence.subtitle)
-            if subtitle_done:
-                logger.info(f"[generate_video] Check subtitle done task_id: {task_id}")
-                break
-            await asyncio.sleep(5)
-            if count > 60 :
-                raise Exception("subtitle 결과 생성 실패")
-            count += 1
-            
-
         # 세션을 명시적으로 열고 DB 작업 후 바로 닫음
         async with AsyncSessionLocal() as session:
-            # ===== 순차 쿼리 실행: Project, Lyric, Song, Image =====
+            # ===== 순차 쿼리 실행: Project, MarketingIntel, Lyric, Song, Image =====
             # Note: AsyncSession은 동일 세션에서 병렬 쿼리를 지원하지 않음
 
             # Project 조회
@@ -188,6 +179,44 @@ async def generate_video(
                 .order_by(Project.created_at.desc())
                 .limit(1)
             )
+            project = project_result.scalar_one_or_none()
+            if not project:
+                logger.warning(f"[generate_video] Project NOT FOUND - task_id: {task_id}")
+                raise HTTPException(
+                    status_code=404,
+                    detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
+                )
+            project_id = project.id
+            store_address = project.detail_region_info
+            brand_name = project.store_name
+            region = project.region
+
+            # MarketingIntel 조회
+            marketing_result = await session.execute(
+                select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
+            )
+            marketing_intelligence: MarketingIntel = marketing_result.scalar_one_or_none()
+
+            # subtitle 미완료 시 즉시 반환 — Lyric/Song/Image 쿼리 전에 체크하여 불필요한 조회 방지
+            # 클라이언트가 /lyric/subtitle/status/{task_id} 폴링 후 재시도
+            if not marketing_intelligence.subtitle:
+                logger.info(f"[generate_video] subtitle pending - task_id: {task_id}")
+                return GenerateVideoResponse(
+                    success=False,
+                    status="subtitle_pending",
+                    task_id=task_id,
+                    creatomate_render_id=None,
+                    message="자막 생성이 아직 완료되지 않았습니다. /lyric/subtitle/status/{task_id}로 완료 확인 후 재요청하세요.",
+                    error_message=None,
+                )
+
+            category_definition = marketing_intelligence.intel_result["market_positioning"]["category_definition"]
+            target_keywords = marketing_intelligence.intel_result["target_keywords"]
+
+            brand_concept = ""
+            for sp in marketing_intelligence.intel_result["selling_points"]:
+                if "concept" in sp["english_category"].lower():
+                    brand_concept = sp["description"]
 
             # Lyric 조회
             lyric_result = await session.execute(
@@ -218,34 +247,6 @@ async def generate_video(
                 f"elapsed: {(query_time - request_start) * 1000:.1f}ms"
             )
 
-            # ===== 결과 처리: Project =====
-            project = project_result.scalar_one_or_none()
-            if not project:
-                logger.warning(
-                    f"[generate_video] Project NOT FOUND - task_id: {task_id}"
-                )
-                raise HTTPException(
-                    status_code=404,
-                    detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
-                )
-            project_id = project.id
-            store_address = project.detail_region_info
-            brand_name = project.store_name
-            region = project.region
-
-            marketing_result = await session.execute(
-                select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
-            )
-            marketing_intelligence : MarketingIntel = marketing_result.scalar_one_or_none()
-
-            category_definition= marketing_intelligence.intel_result["market_positioning"]["category_definition"]
-            target_keywords=marketing_intelligence.intel_result["target_keywords"]
-            
-            brand_concept = ""
-            for sp in marketing_intelligence.intel_result["selling_points"]:
-                if "concept" in sp["english_category"].lower():
-                    brand_concept = sp["description"]
-
             # ===== 결과 처리: Lyric =====
             lyric = lyric_result.scalar_one_or_none()
             if not lyric:
@@ -293,10 +294,19 @@ async def generate_video(
                 )
             image_urls = [img.img_url for img in images]
 
+            # SongTimestamp 조회 (외부 API 호출 전 필요한 데이터이므로 1단계에서 수집)
+            song_timestamp_result = await session.execute(
+                select(SongTimestamp).where(
+                    SongTimestamp.suno_audio_id == song.suno_audio_id
+                )
+            )
+            song_timestamp_list = song_timestamp_result.scalars().all()
+
             logger.info(
                 f"[generate_video] Data loaded - task_id: {task_id}, "
                 f"project_id: {project_id}, lyric_id: {lyric_id}, "
-                f"song_id: {song_id}, images: {len(image_urls)}"
+                f"song_id: {song_id}, images: {len(image_urls)}, "
+                f"timestamps: {len(song_timestamp_list)}"
             )
 
             # ===== Video 테이블에 초기 데이터 저장 및 커밋 =====
@@ -406,13 +416,6 @@ async def generate_video(
 
         logger.debug(f"[generate_video] Duration extended - task_id: {task_id}")
 
-        song_timestamp_result = await session.execute(
-            select(SongTimestamp).where(
-                SongTimestamp.suno_audio_id == song.suno_audio_id
-            )
-        )
-        song_timestamp_list = song_timestamp_result.scalars().all()
-
         logger.debug(f"[generate_video] song_timestamp_list count: {len(song_timestamp_list)}")
 
         for i, ts in enumerate(song_timestamp_list):
@@ -674,7 +677,7 @@ async def get_video_status(
         import traceback
 
         logger.error(
-            f"[get_video_status] EXCEPTION - creatomate_render_id: {creatomate_render_id}, error: {e}"
+            f"[get_video_status] EXCEPTION - creatomate_render_id: {creatomate_render_id}, error: {e}\n{traceback.format_exc()}"
         )
         return PollingVideoResponse(
             success=False,
@@ -682,7 +685,7 @@ async def get_video_status(
             message="상태 조회에 실패했습니다.",
             render_data=None,
             raw_response=None,
-            error_message=f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
+            error_message=f"{type(e).__name__}: {e}",
         )
 
 
@@ -792,7 +795,7 @@ async def download_video(
             status="completed",
             message="영상 다운로드가 완료되었습니다.",
             store_name=project.store_name if project else None,
-            region=project.region if project else None,
+            region=project.region or _extract_region_from_address(project.detail_region_info) if project else None,
             task_id=task_id,
             result_movie_url=video.result_movie_url,
             created_at=video.created_at,
@@ -806,3 +809,353 @@ async def download_video(
             message="영상 다운로드 조회에 실패했습니다.",
             error_message=str(e),
         )
+
+
+@router.get(
+    "/all",
+    summary="ADO2 콘텐츠 - 전체 사용자 영상 갤러리",
+    description="""
+## 개요
+모든 사용자가 생성 완료한 영상을 페이지네이션하여 반환합니다.
+
+## 쿼리 파라미터
+- **page**: 페이지 번호 (1부터 시작, 기본값: 1)
+- **page_size**: 페이지당 데이터 수 (기본값: 10, 최대: 100)
+- **sort_by**: 정렬 기준 (created_at: 최신순, like_count: 좋아요순, comment_count: 댓글순, 기본값: created_at)
+- **order**: 정렬 방향 (desc: 내림차순, asc: 오름차순, 기본값: desc)
+- **store_name**: 업체명 검색 (부분 일치, 값이 있을 때만 전송)
+- **region**: 지역명 검색 (부분 일치, 값이 있을 때만 전송)
+    """,
+    response_model=PaginatedResponse[VideoThumbnailItem],
+    responses={
+        200: {"description": "갤러리 조회 성공"},
+        500: {"description": "조회 실패"},
+    },
+)
+async def get_all_videos(
+    current_user: User | None = Depends(get_current_user_optional),
+    session: AsyncSession = Depends(get_session),
+    pagination: PaginationParams = Depends(get_pagination_params),
+    sort_by: str = Query(default="created_at", description="정렬 기준 (created_at, like_count, comment_count)"),
+    order: str = Query(default="desc", description="정렬 방향 (desc, asc)"),
+    store_name: str | None = Query(default=None, description="업체명 검색 (부분 일치)"),
+    region: str | None = Query(default=None, description="지역명 검색 (부분 일치)"),
+) -> PaginatedResponse[VideoThumbnailItem]:
+    """전체 사용자의 완료된 영상 갤러리를 반환합니다."""
+    logger.info(
+        f"[get_all_videos] START - page: {pagination.page}, page_size: {pagination.page_size}, "
+        f"sort_by: {sort_by}, order: {order}, store_name: {store_name}, region: {region}"
+    )
+
+    try:
+        offset = (pagination.page - 1) * pagination.page_size
+
+        where_clauses = [
+            Video.status == "completed",
+            Video.is_deleted == False,  # noqa: E712
+            Project.is_deleted == False,  # noqa: E712
+            Video.result_movie_url.is_not(None),
+        ]
+        if store_name:
+            where_clauses.append(Project.store_name.ilike(f"%{store_name}%"))
+        if region:
+            where_clauses.append(
+                or_(
+                    Project.region.ilike(f"%{region}%"),
+                    Project.detail_region_info.ilike(f"%{region}%"),
+                )
+            )
+
+        count_q = (
+            select(func.count(Video.id))
+            .join(Project, Video.project_id == Project.id)
+            .where(*where_clauses)
+        )
+        total = (await session.execute(count_q)).scalar() or 0
+
+        comment_count_subq = (
+            select(func.count(Comment.id))
+            .where(
+                Comment.video_id == Video.id,
+                Comment.is_deleted == False,  # noqa: E712
+            )
+            .correlate(Video)
+            .scalar_subquery()
+        )
+
+        # like_count 정렬은 Redis 대신 서브쿼리로 처리 (ORDER BY에만 사용)
+        like_count_subq_for_sort = (
+            select(func.count(VideoReaction.id))
+            .where(VideoReaction.video_id == Video.id)
+            .correlate(Video)
+            .scalar_subquery()
+        )
+        sort_col_map = {
+            "like_count": like_count_subq_for_sort,
+            "comment_count": comment_count_subq,
+            "created_at": Video.created_at,
+        }
+        sort_col = sort_col_map.get(sort_by, Video.created_at)
+        order_clause = sort_col.asc() if order == "asc" else sort_col.desc()
+
+        list_q = (
+            select(
+                Video,
+                Project,
+                comment_count_subq.label("comment_count"),
+            )
+            .join(Project, Video.project_id == Project.id)
+            .where(*where_clauses)
+            .order_by(order_clause)
+            .offset(offset)
+            .limit(pagination.page_size)
+        )
+        rows = (await session.execute(list_q)).all()
+
+        video_ids = [v.id for v, p, _ in rows]
+
+        # Redis mget으로 like_count 일괄 조회
+        like_count_map = await get_like_counts(video_ids)
+
+        # 카운트 캐시 미스 보정
+        missing_ids = [vid for vid, cnt in like_count_map.items() if cnt is None]
+        if missing_ids:
+            db_counts = (await session.execute(
+                select(VideoReaction.video_id, func.count(VideoReaction.id))
+                .where(VideoReaction.video_id.in_(missing_ids))
+                .group_by(VideoReaction.video_id)
+            )).all()
+            db_found_ids = set()
+            batch = {}
+            for vid, cnt in db_counts:
+                batch[vid] = cnt
+                like_count_map[vid] = cnt
+                db_found_ids.add(vid)
+            await mset_like_counts(batch)
+            for vid in missing_ids:
+                if vid not in db_found_ids:
+                    like_count_map[vid] = 0
+
+        # is_liked_by_me: Redis user-set 기준, cold-start 시 DB backfill
+        liked_map: dict[int, bool] = {}
+        if current_user:
+            raw_liked = await bulk_is_user_liked(video_ids, current_user.user_uuid)
+
+            # user-set이 없는(None) 영상 중 count > 0인 것만 backfill 필요
+            needs_backfill = [
+                vid for vid, liked in raw_liked.items()
+                if liked is None and like_count_map.get(vid, 0) > 0
+            ]
+            if needs_backfill:
+                reaction_rows = (await session.execute(
+                    select(VideoReaction.video_id, VideoReaction.user_uuid)
+                    .where(VideoReaction.video_id.in_(needs_backfill))
+                )).all()
+                user_map: dict[int, list[str]] = defaultdict(list)
+                for vid, uuid in reaction_rows:
+                    user_map[vid].append(uuid)
+                for vid in needs_backfill:
+                    await backfill_user_set(vid, user_map.get(vid, []))
+
+                # backfill 후 재조회
+                updated = await bulk_is_user_liked(needs_backfill, current_user.user_uuid)
+                raw_liked.update(updated)
+
+            liked_map = {vid: bool(liked) for vid, liked in raw_liked.items()}
+
+        items = [
+            VideoThumbnailItem(
+                video_id=v.id,
+                store_name=p.store_name,
+                result_movie_url=v.result_movie_url,
+                created_at=v.created_at,
+                like_count=like_count_map.get(v.id) or 0,
+                is_liked_by_me=liked_map.get(v.id, False),
+                comment_count=comment_count or 0,
+            )
+            for v, p, comment_count in rows
+        ]
+
+        response = PaginatedResponse.create(
+            items=items,
+            total=total,
+            page=pagination.page,
+            page_size=pagination.page_size,
+        )
+        logger.info(f"[get_all_videos] SUCCESS - total: {total}, items: {len(items)}")
+        return response
+
+    except Exception as e:
+        logger.error(f"[get_all_videos] EXCEPTION - error: {e}")
+        raise HTTPException(status_code=500, detail=f"갤러리 조회에 실패했습니다: {str(e)}")
+
+
+@router.post(
+    "/{video_id}/like",
+    summary="영상 좋아요 토글",
+    description="""
+## 개요
+영상에 좋아요를 토글합니다. 로그인 필수.
+
+- 처음 호출: 좋아요 추가 (is_liked=true)
+- 다시 호출: 좋아요 취소 (is_liked=false)
+    """,
+    response_model=LikeToggleResponse,
+    responses={
+        200: {"description": "토글 성공"},
+        401: {"description": "인증 실패"},
+        404: {"description": "영상을 찾을 수 없음"},
+    },
+)
+async def toggle_like(
+    video_id: int,
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> LikeToggleResponse:
+    """영상 좋아요를 토글합니다.
+
+    Write-Behind 패턴:
+    1. Redis user-set / count를 즉시 원자적으로 업데이트 (Lua script)
+    2. dirty SET에 표시 → 스케줄러가 1분마다 MySQL에 반영
+    DB write가 없으므로 고트래픽에서도 응답 지연 없음.
+    """
+    logger.info(f"[toggle_like] START - video_id: {video_id}, user: {current_user.user_uuid}")
+
+    try:
+        # 영상 존재 확인 (DB read는 유지 — 404 처리 필수)
+        video_result = await session.execute(
+            select(Video).where(
+                Video.id == video_id,
+                Video.status == "completed",
+                Video.is_deleted == False,  # noqa: E712
+            )
+        )
+        if video_result.scalar_one_or_none() is None:
+            raise HTTPException(status_code=404, detail="영상을 찾을 수 없습니다.")
+
+        # Cold-start 보정: Redis에 데이터가 없으면 DB에서 backfill
+        count = await get_like_count(video_id)
+        if count is None:
+            # 카운트와 user-set 모두 없음 → DB에서 전체 복구
+            user_uuids = (await session.execute(
+                select(VideoReaction.user_uuid)
+                .where(VideoReaction.video_id == video_id)
+            )).scalars().all()
+            await backfill_user_set(video_id, list(user_uuids))
+            await set_like_count(video_id, len(user_uuids))
+        elif count > 0:
+            if not await is_user_set_exists(video_id):
+                # 카운트는 있지만 user-set이 증발한 경우 (부분 캐시 미스)
+                user_uuids = (await session.execute(
+                    select(VideoReaction.user_uuid)
+                    .where(VideoReaction.video_id == video_id)
+                )).scalars().all()
+                await backfill_user_set(video_id, list(user_uuids))
+
+        # Lua 스크립트로 원자적 토글 (race condition 방지)
+        is_liked, like_count = await toggle_like_atomic(video_id, current_user.user_uuid)
+
+        # dirty SET에 표시 → 스케줄러가 DB에 반영
+        await mark_dirty(video_id, current_user.user_uuid)
+
+        logger.info(
+            f"[toggle_like] SUCCESS - video_id: {video_id}, "
+            f"is_liked: {is_liked}, count: {like_count}"
+        )
+        return LikeToggleResponse(video_id=video_id, is_liked=is_liked, like_count=like_count)
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"[toggle_like] EXCEPTION - video_id: {video_id}, error: {e}")
+        raise HTTPException(status_code=500, detail=f"좋아요 처리에 실패했습니다: {str(e)}")
+
+
+@router.get(
+    "/{video_id}",
+    summary="단일 영상 상세 조회",
+    description="""
+## 개요
+video_id에 해당하는 완료된 영상의 상세 정보를 반환합니다.
+
+## 경로 파라미터
+- **video_id**: 조회할 영상의 ID (Video.id)
+    """,
+    response_model=VideoDetailResponse,
+    responses={
+        200: {"description": "상세 조회 성공"},
+        404: {"description": "영상을 찾을 수 없음"},
+        500: {"description": "조회 실패"},
+    },
+)
+async def get_video_detail(
+    video_id: int,
+    current_user: User | None = Depends(get_current_user_optional),
+    session: AsyncSession = Depends(get_session),
+) -> VideoDetailResponse:
+    """video_id에 해당하는 완료된 영상 상세 정보를 반환합니다."""
+    logger.info(f"[get_video_detail] START - video_id: {video_id}")
+
+    try:
+        result = await session.execute(
+            select(Video, Project)
+            .join(Project, Video.project_id == Project.id)
+            .where(
+                Video.id == video_id,
+                Video.status == "completed",
+                Video.is_deleted == False,  # noqa: E712
+                Project.is_deleted == False,  # noqa: E712
+            )
+        )
+        row = result.one_or_none()
+
+        if row is None:
+            logger.warning(f"[get_video_detail] NOT FOUND - video_id: {video_id}")
+            raise HTTPException(status_code=404, detail="영상을 찾을 수 없습니다.")
+
+        video, project = row
+
+        # like_count: Redis 조회, 캐시 미스 시 DB backfill
+        like_count = await get_like_count(video_id)
+        if like_count is None:
+            user_uuids = (await session.execute(
+                select(VideoReaction.user_uuid)
+                .where(VideoReaction.video_id == video_id)
+            )).scalars().all()
+            like_count = len(user_uuids)
+            await backfill_user_set(video_id, list(user_uuids))
+            await set_like_count(video_id, like_count)
+
+        # is_liked_by_me: Redis user-set 기준, cold-start 시 DB backfill
+        is_liked_by_me = False
+        if current_user:
+            liked = await is_user_liked(video_id, current_user.user_uuid)
+            if liked is None:
+                # user-set 없음 → count key로 cold-start 여부 판별
+                if like_count > 0:
+                    user_uuids = (await session.execute(
+                        select(VideoReaction.user_uuid)
+                        .where(VideoReaction.video_id == video_id)
+                    )).scalars().all()
+                    await backfill_user_set(video_id, list(user_uuids))
+                    liked = current_user.user_uuid in set(user_uuids)
+                else:
+                    liked = False
+            is_liked_by_me = liked
+
+        logger.info(f"[get_video_detail] SUCCESS - video_id: {video_id}")
+        return VideoDetailResponse(
+            video_id=video.id,
+            result_movie_url=video.result_movie_url,
+            store_name=project.store_name,
+            region=project.region or _extract_region_from_address(project.detail_region_info),
+            created_at=video.created_at,
+            like_count=like_count,
+            is_liked_by_me=is_liked_by_me,
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"[get_video_detail] EXCEPTION - video_id: {video_id}, error: {e}")
+        raise HTTPException(status_code=500, detail=f"영상 조회에 실패했습니다: {str(e)}")

app/video/models.py

@@ -1,15 +1,17 @@
 from datetime import datetime
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, List, Optional
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database.session import Base
 
 if TYPE_CHECKING:
+    from app.comment.models import Comment
     from app.home.models import Project
     from app.lyric.models import Lyric
     from app.song.models import Song
+    from app.user.models import User
 
 
 class Video(Base):
@@ -33,6 +35,8 @@ class Video(Base):
         project: 연결된 Project
         lyric: 연결된 Lyric
         song: 연결된 Song
+        comments: 영상 댓글 목록
+        likes: 영상 좋아요 목록
     """
 
     __tablename__ = "video"
@@ -132,6 +136,20 @@ class Video(Base):
         back_populates="videos",
     )
 
+    comments: Mapped[List["Comment"]] = relationship(
+        "Comment",
+        back_populates="video",
+        cascade="all, delete-orphan",
+        lazy="noload",
+    )
+
+    reactions: Mapped[List["VideoReaction"]] = relationship(
+        "VideoReaction",
+        back_populates="video",
+        cascade="all, delete-orphan",
+        lazy="noload",
+    )
+
     def __repr__(self) -> str:
         def truncate(value: str | None, max_len: int = 10) -> str:
             if value is None:
@@ -145,3 +163,50 @@ class Video(Base):
             f"status='{self.status}'"
             f")>"
         )
+
+
+class VideoReaction(Base):
+    """
+    영상 반응 테이블
+
+    사용자가 영상에 반응(현재는 좋아요)을 남기면 생성, 다시 누르면 삭제(토글).
+    (user_uuid, video_id) 유니크 제약으로 1인 1회 보장.
+    향후 reaction_type 컬럼 추가로 다양한 반응 종류 확장 가능.
+    """
+
+    __tablename__ = "video_reaction"
+    __table_args__ = (
+        UniqueConstraint("user_uuid", "video_id", name="uq_video_reaction_user_video"),
+        Index("idx_video_reaction_video_id", "video_id"),
+        Index("idx_video_reaction_user_uuid", "user_uuid"),
+        {
+            "mysql_engine": "InnoDB",
+            "mysql_charset": "utf8mb4",
+            "mysql_collate": "utf8mb4_unicode_ci",
+        },
+    )
+
+    id: Mapped[int] = mapped_column(
+        Integer, primary_key=True, autoincrement=True, comment="고유 식별자"
+    )
+    video_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("video.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="연결된 Video의 id",
+    )
+    user_uuid: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("user.user_uuid", ondelete="CASCADE"),
+        nullable=False,
+        comment="반응한 사용자 UUID",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        server_default=func.now(),
+        comment="반응 일시",
+    )
+
+    video: Mapped["Video"] = relationship("Video", back_populates="reactions")
+    user: Mapped["User"] = relationship("User", back_populates="video_reactions")

app/video/schemas/video_schema.py

@@ -36,6 +36,7 @@ class GenerateVideoResponse(BaseModel):
     )
 
     success: bool = Field(..., description="요청 성공 여부")
+    status: Optional[str] = Field(None, description="처리 상태 (subtitle_pending: 자막 미완료, completed: 정상 접수)")
     task_id: Optional[str] = Field(None, description="내부 작업 ID (Project task_id)")
     creatomate_render_id: Optional[str] = Field(None, description="Creatomate 렌더 ID")
     message: str = Field(..., description="응답 메시지")
@@ -157,5 +158,51 @@ class VideoListItem(BaseModel):
     task_id: str = Field(..., description="작업 고유 식별자")
     result_movie_url: Optional[str] = Field(None, description="영상 결과 URL")
     created_at: Optional[datetime] = Field(None, description="생성 일시")
+    like_count: int = Field(0, description="좋아요 수")
+    comment_count: int = Field(0, description="댓글 수 (대댓글 포함)")
+
+
+class VideoThumbnailItem(BaseModel):
+    """ADO2 콘텐츠 갤러리용 최소 영상 정보 (썸네일 표시 + 상세 페이지 이동용)
+
+    Usage:
+        GET /video/all 응답의 개별 영상 정보
+    """
+
+    video_id: int = Field(..., description="영상 고유 ID (상세 페이지 라우팅 키)")
+    store_name: str = Field(..., description="업체명")
+    result_movie_url: str = Field(..., description="영상 URL — 프론트에서 <video> 태그 첫 프레임을 썸네일로 사용")
+    created_at: datetime = Field(..., description="생성 일시")
+    like_count: int = Field(..., description="좋아요 수")
+    is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
+    comment_count: int = Field(..., description="댓글 수 (대댓글 포함)")
+
+
+class VideoDetailResponse(BaseModel):
+    """단일 영상 상세 응답
+
+    Usage:
+        GET /video/{video_id}
+    """
+
+    video_id: int = Field(..., description="영상 고유 ID")
+    result_movie_url: str = Field(..., description="영상 URL")
+    store_name: Optional[str] = Field(None, description="업체명")
+    region: Optional[str] = Field(None, description="지역명")
+    created_at: datetime = Field(..., description="생성 일시")
+    like_count: int = Field(..., description="좋아요 수")
+    is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
+
+
+class LikeToggleResponse(BaseModel):
+    """좋아요 토글 응답
+
+    Usage:
+        POST /video/{video_id}/like
+    """
+
+    video_id: int = Field(..., description="영상 고유 ID")
+    is_liked: bool = Field(..., description="토글 후 상태 (true=좋아요 누름, false=취소됨)")
+    like_count: int = Field(..., description="토글 후 전체 좋아요 수")
 
 

docs/database-schema/migration_2026_05_21_add_comment.sql

@@ -0,0 +1,25 @@
+-- ============================================================
+-- Migration: 영상 댓글 테이블 추가
+-- Date: 2026-05-21
+-- Description: 영상 상세 페이지 댓글/대댓글 기능 (2-depth, 소프트 삭제)
+-- ============================================================
+
+CREATE TABLE IF NOT EXISTS comment (
+    id         BIGINT       NOT NULL AUTO_INCREMENT COMMENT '고유 식별자',
+    video_id   INT          NOT NULL                COMMENT '연결된 Video의 id',
+    user_uuid  VARCHAR(36)  NOT NULL                COMMENT '작성자 UUID (user.user_uuid 참조, 응답 미노출)',
+    nickname   VARCHAR(20)          NULL                COMMENT '작성자 닉네임 (null이면 익명으로 표시)',
+    parent_id  BIGINT           NULL                COMMENT 'NULL=최상위 댓글, 값=대댓글의 부모 id',
+    content    VARCHAR(100) NOT NULL                COMMENT '댓글 본문 (한글 기준 100자 이내)',
+    is_deleted BOOLEAN      NOT NULL DEFAULT FALSE  COMMENT '소프트 삭제 여부',
+    created_at DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '작성 일시',
+    PRIMARY KEY (id),
+    CONSTRAINT fk_comment_video  FOREIGN KEY (video_id)  REFERENCES video(id)         ON DELETE CASCADE,
+    CONSTRAINT fk_comment_user   FOREIGN KEY (user_uuid) REFERENCES `user`(user_uuid) ON DELETE CASCADE,
+    CONSTRAINT fk_comment_parent FOREIGN KEY (parent_id) REFERENCES comment(id)       ON DELETE CASCADE,
+    INDEX idx_comment_video_id   (video_id),
+    INDEX idx_comment_user_uuid  (user_uuid),
+    INDEX idx_comment_parent_id  (parent_id),
+    INDEX idx_comment_is_deleted (is_deleted)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+  COMMENT='영상 댓글/대댓글';

docs/database-schema/migration_2026_05_21_add_video_reaction.sql

@@ -0,0 +1,19 @@
+-- ============================================================
+-- Migration: 영상 반응(좋아요) 테이블 추가
+-- Date: 2026-05-21
+-- Description: 사용자 영상별 좋아요 토글 (1인 1회, 확장 가능한 구조)
+-- ============================================================
+
+CREATE TABLE IF NOT EXISTS video_reaction (
+    id         BIGINT      NOT NULL AUTO_INCREMENT COMMENT '고유 식별자',
+    video_id   INT         NOT NULL                COMMENT '연결된 Video의 id',
+    user_uuid  VARCHAR(36) NOT NULL                COMMENT '반응한 사용자 UUID',
+    created_at DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '반응 일시',
+    PRIMARY KEY (id),
+    CONSTRAINT uq_video_reaction_user_video UNIQUE (user_uuid, video_id),
+    CONSTRAINT fk_video_reaction_video FOREIGN KEY (video_id)  REFERENCES video(id)         ON DELETE CASCADE,
+    CONSTRAINT fk_video_reaction_user  FOREIGN KEY (user_uuid) REFERENCES `user`(user_uuid) ON DELETE CASCADE,
+    INDEX idx_video_reaction_video_id  (video_id),
+    INDEX idx_video_reaction_user_uuid (user_uuid)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+  COMMENT='영상 반응 (user_uuid + video_id 유니크)';

main.py

@@ -24,6 +24,8 @@ 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.seo import router as social_seo_router
 from app.social.api.routers.v1.internal import router as social_internal_router
+from app.comment.api.routers.v1.comment import router as comment_router
+from app.video.api.routers.internal.reactions import router as video_internal_router
 from app.credit.api.routers.v1.credit import router as credit_router
 from app.utils.cors import CustomCORSMiddleware
 from config import prj_settings
@@ -174,6 +176,25 @@ tags_metadata = [
 - created_at 기준 내림차순 정렬됩니다.
 - 삭제는 소프트 삭제(is_deleted=True) 방식으로 처리되며, 데이터 복구가 가능합니다.
 - 삭제 대상: Video, SongTimestamp, Song, Lyric, Image, Project
+""",
+    },
+    {
+        "name": "Comment",
+        "description": """영상 댓글 API - 댓글/대댓글 작성·조회·삭제
+
+**인증: 조회는 불필요, 작성/삭제는 필요** - `Authorization: Bearer {access_token}` 헤더
+
+## 주요 기능
+
+- `POST /comment/video/{video_id}` - 댓글/대댓글 작성 (로그인 필수)
+- `GET /comment/video/{video_id}` - 댓글 목록 조회 (비로그인 허용)
+- `DELETE /comment/{comment_id}` - 본인 댓글 소프트 삭제 (로그인 필수)
+
+## 참고
+
+- 최대 2-depth (댓글 + 대댓글). 대댓글에 대댓글은 불가합니다.
+- 작성자 정보는 응답에 포함되지 않습니다 (익명 정책).
+- is_mine 필드로 본인 댓글 여부를 확인할 수 있습니다.
 """,
     },
     {
@@ -390,10 +411,12 @@ app.include_router(lyric_router)
 app.include_router(song_router)
 app.include_router(video_router)
 app.include_router(archive_router)  # Archive API 라우터 추가
+app.include_router(comment_router)  # Comment API 라우터 추가
 app.include_router(social_oauth_router, prefix="/social")  # Social OAuth 라우터 추가
 app.include_router(social_upload_router, prefix="/social")  # Social Upload 라우터 추가
 app.include_router(social_seo_router, prefix="/social")  # Social Upload 라우터 추가
 app.include_router(social_internal_router)  # 내부 스케줄러 전용 라우터
+app.include_router(video_internal_router)   # 내부 좋아요 플러시 라우터
 app.include_router(sns_router)  # SNS API 라우터 추가
 app.include_router(dashboard_router)  # Dashboard API 라우터 추가
 app.include_router(credit_router, prefix="/user")  # Credit API 라우터 추가

pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "aiofiles>=25.1.0",
     "aiohttp>=3.13.2",
     "aiomysql>=0.3.2",
-    "asyncmy>=0.2.10",
+    "asyncmy>=0.2.11",
     "beautifulsoup4>=4.14.3",
     "fastapi-cli>=0.0.16",
     "fastapi[standard]>=0.125.0",
@@ -23,7 +23,7 @@ dependencies = [
     "ruff>=0.14.9",
     "scalar-fastapi>=1.6.1",
     "sqladmin[full]>=0.22.0",
-    "sqlalchemy[asyncio]>=2.0.45",
+    "sqlalchemy[asyncio]>=2.0.50",
     "uuid7>=0.1.0",
 ]
 

uv.lock

@@ -744,7 +744,7 @@ requires-dist = [
     { name = "aiofiles", specifier = ">=25.1.0" },
     { name = "aiohttp", specifier = ">=3.13.2" },
     { name = "aiomysql", specifier = ">=0.3.2" },
-    { name = "asyncmy", specifier = ">=0.2.10" },
+    { name = "asyncmy", specifier = ">=0.2.11" },
     { name = "beautifulsoup4", specifier = ">=4.14.3" },
     { name = "fastapi", extras = ["standard"], specifier = ">=0.125.0" },
     { name = "fastapi-cli", specifier = ">=0.0.16" },
@@ -759,7 +759,7 @@ requires-dist = [
     { name = "ruff", specifier = ">=0.14.9" },
     { name = "scalar-fastapi", specifier = ">=1.6.1" },
     { name = "sqladmin", extras = ["full"], specifier = ">=0.22.0" },
-    { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.45" },
+    { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.50" },
     { name = "uuid7", specifier = ">=0.1.0" },
 ]
 
@@ -1294,26 +1294,22 @@ full = [
 
 [[package]]
 name = "sqlalchemy"
-version = "2.0.46"
+version = "2.0.50"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "greenlet", marker = "platform_machine == 'AMD64' or platform_machine == 'WIN32' or platform_machine == 'aarch64' or platform_machine == 'amd64' or platform_machine == 'ppc64le' or platform_machine == 'win32' or platform_machine == 'x86_64'" },
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/06/aa/9ce0f3e7a9829ead5c8ce549392f33a12c4555a6c0609bb27d882e9c7ddf/sqlalchemy-2.0.46.tar.gz", hash = "sha256:cf36851ee7219c170bb0793dbc3da3e80c582e04a5437bc601bfe8c85c9216d7", size = 9865393, upload-time = "2026-01-21T18:03:45.119Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/57/da/6fbf010c8ebb347679d0d100b22fe9ba5e13fd04046c5df7280d2f0bf706/sqlalchemy-2.0.50.tar.gz", hash = "sha256:af5607d11ef90fd6a5c0549fe0045dce1663d427426bcfb506dcb5346a85a3b9", size = 9907424, upload-time = "2026-05-24T19:20:04.018Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/b3/4b/fa7838fe20bb752810feed60e45625a9a8b0102c0c09971e2d1d95362992/sqlalchemy-2.0.46-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:93a12da97cca70cea10d4b4fc602589c4511f96c1f8f6c11817620c021d21d00", size = 2150268, upload-time = "2026-01-21T19:05:56.621Z" },
-    { url = "https://files.pythonhosted.org/packages/46/c1/b34dccd712e8ea846edf396e00973dda82d598cb93762e55e43e6835eba9/sqlalchemy-2.0.46-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:af865c18752d416798dae13f83f38927c52f085c52e2f32b8ab0fef46fdd02c2", size = 3276511, upload-time = "2026-01-21T18:46:49.022Z" },
-    { url = "https://files.pythonhosted.org/packages/96/48/a04d9c94753e5d5d096c628c82a98c4793b9c08ca0e7155c3eb7d7db9f24/sqlalchemy-2.0.46-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8d679b5f318423eacb61f933a9a0f75535bfca7056daeadbf6bd5bcee6183aee", size = 3292881, upload-time = "2026-01-21T18:40:13.089Z" },
-    { url = "https://files.pythonhosted.org/packages/be/f4/06eda6e91476f90a7d8058f74311cb65a2fb68d988171aced81707189131/sqlalchemy-2.0.46-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:64901e08c33462acc9ec3bad27fc7a5c2b6491665f2aa57564e57a4f5d7c52ad", size = 3224559, upload-time = "2026-01-21T18:46:50.974Z" },
-    { url = "https://files.pythonhosted.org/packages/ab/a2/d2af04095412ca6345ac22b33b89fe8d6f32a481e613ffcb2377d931d8d0/sqlalchemy-2.0.46-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:e8ac45e8f4eaac0f9f8043ea0e224158855c6a4329fd4ee37c45c61e3beb518e", size = 3262728, upload-time = "2026-01-21T18:40:14.883Z" },
-    { url = "https://files.pythonhosted.org/packages/31/48/1980c7caa5978a3b8225b4d230e69a2a6538a3562b8b31cea679b6933c83/sqlalchemy-2.0.46-cp313-cp313-win32.whl", hash = "sha256:8d3b44b3d0ab2f1319d71d9863d76eeb46766f8cf9e921ac293511804d39813f", size = 2111295, upload-time = "2026-01-21T18:42:52.366Z" },
-    { url = "https://files.pythonhosted.org/packages/2d/54/f8d65bbde3d877617c4720f3c9f60e99bb7266df0d5d78b6e25e7c149f35/sqlalchemy-2.0.46-cp313-cp313-win_amd64.whl", hash = "sha256:77f8071d8fbcbb2dd11b7fd40dedd04e8ebe2eb80497916efedba844298065ef", size = 2137076, upload-time = "2026-01-21T18:42:53.924Z" },
-    { url = "https://files.pythonhosted.org/packages/56/ba/9be4f97c7eb2b9d5544f2624adfc2853e796ed51d2bb8aec90bc94b7137e/sqlalchemy-2.0.46-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a1e8cc6cc01da346dc92d9509a63033b9b1bda4fed7a7a7807ed385c7dccdc10", size = 3556533, upload-time = "2026-01-21T18:33:06.636Z" },
-    { url = "https://files.pythonhosted.org/packages/20/a6/b1fc6634564dbb4415b7ed6419cdfeaadefd2c39cdab1e3aa07a5f2474c2/sqlalchemy-2.0.46-cp313-cp313t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:96c7cca1a4babaaf3bfff3e4e606e38578856917e52f0384635a95b226c87764", size = 3523208, upload-time = "2026-01-21T18:45:08.436Z" },
-    { url = "https://files.pythonhosted.org/packages/a1/d8/41e0bdfc0f930ff236f86fccd12962d8fa03713f17ed57332d38af6a3782/sqlalchemy-2.0.46-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:b2a9f9aee38039cf4755891a1e50e1effcc42ea6ba053743f452c372c3152b1b", size = 3464292, upload-time = "2026-01-21T18:33:08.208Z" },
-    { url = "https://files.pythonhosted.org/packages/f0/8b/9dcbec62d95bea85f5ecad9b8d65b78cc30fb0ffceeb3597961f3712549b/sqlalchemy-2.0.46-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:db23b1bf8cfe1f7fda19018e7207b20cdb5168f83c437ff7e95d19e39289c447", size = 3473497, upload-time = "2026-01-21T18:45:10.552Z" },
-    { url = "https://files.pythonhosted.org/packages/fc/a1/9c4efa03300926601c19c18582531b45aededfb961ab3c3585f1e24f120b/sqlalchemy-2.0.46-py3-none-any.whl", hash = "sha256:f9c11766e7e7c0a2767dda5acb006a118640c9fc0a4104214b96269bfb78399e", size = 1937882, upload-time = "2026-01-21T18:22:10.456Z" },
+    { url = "https://files.pythonhosted.org/packages/0b/c4/c42356b527296e9862f67990efce31ef78b4cf69cd3f80873a528a060320/sqlalchemy-2.0.50-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:06a9210bdc5f4298cff0781087e2ff45683922252dacc452846373a58761f093", size = 2156697, upload-time = "2026-05-24T19:27:54.764Z" },
+    { url = "https://files.pythonhosted.org/packages/60/a1/b1a70e3c4365ac7fe9e347f3710f19b562c866fb96d45e3c891588789a7b/sqlalchemy-2.0.50-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8b53784972ade4f8174b9aa661f31a06f8a936d2cfdd602913ff3c6dd40ae873", size = 3284260, upload-time = "2026-05-24T20:09:34.195Z" },
+    { url = "https://files.pythonhosted.org/packages/3f/4a/f3ac3caa19f263d57b0a47f8c91bbf56583dc2d3fc63acfbf644abb24fe0/sqlalchemy-2.0.50-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:31648fa14460537e768a7303b078e4344d208e0d23e06867c1f376a227ed82db", size = 3302280, upload-time = "2026-05-24T20:17:17.825Z" },
+    { url = "https://files.pythonhosted.org/packages/66/55/ccada3e3d62254587819749a0bc69f41173eb48a6e385d10e66d32a9c88e/sqlalchemy-2.0.50-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:03f4323c980ad0e918cc9e5369b015f759f4e534db5bbaf4dc36832c10d05064", size = 3231580, upload-time = "2026-05-24T20:09:36.406Z" },
+    { url = "https://files.pythonhosted.org/packages/05/f6/6809349130a2de0e109e7f00fd7d431da9565b9b2868b32ee684754f672b/sqlalchemy-2.0.50-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:2b9dcc43afef8ac157cd92fce96985d6b8b0cfbd3df4d666f66b4d55a75d202f", size = 3269375, upload-time = "2026-05-24T20:17:20.34Z" },
+    { url = "https://files.pythonhosted.org/packages/48/84/278a811ef4e07be9c89dc5cdd7be833268509a66a68c4897cf585e67428f/sqlalchemy-2.0.50-cp313-cp313-win32.whl", hash = "sha256:60922d6599065ddca2c6f376b9aa2f41a6b85a271725e0909490bbc50b1998a5", size = 2117229, upload-time = "2026-05-24T19:50:08.215Z" },
+    { url = "https://files.pythonhosted.org/packages/f6/1c/067cc6187ed32d2ec222fe6d2643acc1659a6d0659f8a7cbc5ad3ae83280/sqlalchemy-2.0.50-cp313-cp313-win_amd64.whl", hash = "sha256:287086e67275a212c4582d166a6fb03a65ccc5551d80866270ce0dd9f34eccd3", size = 2143126, upload-time = "2026-05-24T19:50:09.691Z" },
+    { url = "https://files.pythonhosted.org/packages/d0/10/f7220e9b784d295d241c86ed99aeb537f92afcd469a64861f2717e9bb077/sqlalchemy-2.0.50-py3-none-any.whl", hash = "sha256:92064363517a3ff8212b5a93b8c62876579d8dfd1ca5b561335f30152d884fa9", size = 1943861, upload-time = "2026-05-24T19:59:01.119Z" },
 ]
 
 [package.optional-dependencies]