Files
AutonetSellCar/temp_content_videos_api.py

269 lines
8.9 KiB
Python

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional
from app.core.database import get_db
from app.models.content_video import ContentVideo
from app.schemas.content_video import (
ContentVideoCreate,
ContentVideoUpdate,
ContentVideoResponse,
ContentVideoPublic,
)
from app.api.auth import get_current_admin
router = APIRouter(prefix="/content-videos", tags=["content-videos"])
def get_localized_field(obj, field: str, locale: str) -> str:
"""Get localized field value with fallback to Korean"""
localized = getattr(obj, f"{field}_{locale}", None)
if localized:
return localized
return getattr(obj, f"{field}_ko", "") or ""
# ============ Public Endpoints ============
@router.get("/entity/{entity_type}/{entity_id}", response_model=List[ContentVideoPublic])
def get_videos_for_entity(
entity_type: str,
entity_id: int,
locale: str = Query(default="ko"),
db: Session = Depends(get_db)
):
"""Get all active videos for a specific entity (public)"""
if entity_type not in ["project", "solution", "product"]:
raise HTTPException(status_code=400, detail="Invalid entity type")
videos = db.query(ContentVideo).filter(
ContentVideo.entity_type == entity_type,
ContentVideo.entity_id == entity_id,
ContentVideo.is_active == True
).order_by(ContentVideo.display_order).all()
return [
ContentVideoPublic(
id=v.id,
youtube_id=v.youtube_id,
title=get_localized_field(v, "title", locale),
description=get_localized_field(v, "description", locale) or None,
youtube_url=f"https://www.youtube.com/watch?v={v.youtube_id}",
youtube_embed_url=f"https://www.youtube.com/embed/{v.youtube_id}",
thumbnail_url=f"https://img.youtube.com/vi/{v.youtube_id}/maxresdefault.jpg",
display_order=v.display_order,
)
for v in videos
]
# ============ Admin Endpoints ============
@router.get("/admin/list", response_model=List[ContentVideoResponse])
def admin_list_videos(
entity_type: Optional[str] = None,
entity_id: Optional[int] = None,
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""List all videos (admin)"""
query = db.query(ContentVideo)
if entity_type:
query = query.filter(ContentVideo.entity_type == entity_type)
if entity_id:
query = query.filter(ContentVideo.entity_id == entity_id)
videos = query.order_by(
ContentVideo.entity_type,
ContentVideo.entity_id,
ContentVideo.display_order
).all()
return [
ContentVideoResponse(
id=v.id,
youtube_id=v.youtube_id,
title_ko=v.title_ko,
title_en=v.title_en,
title_ja=v.title_ja,
title_zh=v.title_zh,
description_ko=v.description_ko,
description_en=v.description_en,
description_ja=v.description_ja,
description_zh=v.description_zh,
entity_type=v.entity_type,
entity_id=v.entity_id,
display_order=v.display_order,
is_active=v.is_active,
created_at=v.created_at,
updated_at=v.updated_at,
youtube_url=f"https://www.youtube.com/watch?v={v.youtube_id}",
youtube_embed_url=f"https://www.youtube.com/embed/{v.youtube_id}",
thumbnail_url=f"https://img.youtube.com/vi/{v.youtube_id}/maxresdefault.jpg",
)
for v in videos
]
@router.get("/admin/{video_id}", response_model=ContentVideoResponse)
def admin_get_video(
video_id: int,
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""Get a specific video (admin)"""
video = db.query(ContentVideo).filter(ContentVideo.id == video_id).first()
if not video:
raise HTTPException(status_code=404, detail="Video not found")
return ContentVideoResponse(
id=video.id,
youtube_id=video.youtube_id,
title_ko=video.title_ko,
title_en=video.title_en,
title_ja=video.title_ja,
title_zh=video.title_zh,
description_ko=video.description_ko,
description_en=video.description_en,
description_ja=video.description_ja,
description_zh=video.description_zh,
entity_type=video.entity_type,
entity_id=video.entity_id,
display_order=video.display_order,
is_active=video.is_active,
created_at=video.created_at,
updated_at=video.updated_at,
youtube_url=f"https://www.youtube.com/watch?v={video.youtube_id}",
youtube_embed_url=f"https://www.youtube.com/embed/{video.youtube_id}",
thumbnail_url=f"https://img.youtube.com/vi/{video.youtube_id}/maxresdefault.jpg",
)
@router.post("/admin", response_model=ContentVideoResponse)
def admin_create_video(
data: ContentVideoCreate,
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""Create a new video (admin)"""
video = ContentVideo(
youtube_id=data.youtube_id,
title_ko=data.title_ko,
title_en=data.title_en,
title_ja=data.title_ja,
title_zh=data.title_zh,
description_ko=data.description_ko,
description_en=data.description_en,
description_ja=data.description_ja,
description_zh=data.description_zh,
entity_type=data.entity_type,
entity_id=data.entity_id,
display_order=data.display_order,
is_active=data.is_active,
)
db.add(video)
db.commit()
db.refresh(video)
return ContentVideoResponse(
id=video.id,
youtube_id=video.youtube_id,
title_ko=video.title_ko,
title_en=video.title_en,
title_ja=video.title_ja,
title_zh=video.title_zh,
description_ko=video.description_ko,
description_en=video.description_en,
description_ja=video.description_ja,
description_zh=video.description_zh,
entity_type=video.entity_type,
entity_id=video.entity_id,
display_order=video.display_order,
is_active=video.is_active,
created_at=video.created_at,
updated_at=video.updated_at,
youtube_url=f"https://www.youtube.com/watch?v={video.youtube_id}",
youtube_embed_url=f"https://www.youtube.com/embed/{video.youtube_id}",
thumbnail_url=f"https://img.youtube.com/vi/{video.youtube_id}/maxresdefault.jpg",
)
@router.put("/admin/{video_id}", response_model=ContentVideoResponse)
def admin_update_video(
video_id: int,
data: ContentVideoUpdate,
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""Update a video (admin)"""
video = db.query(ContentVideo).filter(ContentVideo.id == video_id).first()
if not video:
raise HTTPException(status_code=404, detail="Video not found")
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(video, field, value)
db.commit()
db.refresh(video)
return ContentVideoResponse(
id=video.id,
youtube_id=video.youtube_id,
title_ko=video.title_ko,
title_en=video.title_en,
title_ja=video.title_ja,
title_zh=video.title_zh,
description_ko=video.description_ko,
description_en=video.description_en,
description_ja=video.description_ja,
description_zh=video.description_zh,
entity_type=video.entity_type,
entity_id=video.entity_id,
display_order=video.display_order,
is_active=video.is_active,
created_at=video.created_at,
updated_at=video.updated_at,
youtube_url=f"https://www.youtube.com/watch?v={video.youtube_id}",
youtube_embed_url=f"https://www.youtube.com/embed/{video.youtube_id}",
thumbnail_url=f"https://img.youtube.com/vi/{video.youtube_id}/maxresdefault.jpg",
)
@router.delete("/admin/{video_id}")
def admin_delete_video(
video_id: int,
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""Delete a video (admin)"""
video = db.query(ContentVideo).filter(ContentVideo.id == video_id).first()
if not video:
raise HTTPException(status_code=404, detail="Video not found")
db.delete(video)
db.commit()
return {"message": "Video deleted successfully"}
@router.put("/admin/reorder")
def admin_reorder_videos(
entity_type: str,
entity_id: int,
video_ids: List[int],
db: Session = Depends(get_db),
_: str = Depends(get_current_admin)
):
"""Reorder videos for an entity (admin)"""
for index, video_id in enumerate(video_ids):
video = db.query(ContentVideo).filter(
ContentVideo.id == video_id,
ContentVideo.entity_type == entity_type,
ContentVideo.entity_id == entity_id
).first()
if video:
video.display_order = index
db.commit()
return {"message": "Videos reordered successfully"}