7d266208ae
Some checks failed
Deploy Fediversion / deploy (push) Failing after 1s
- Video model with VideoType/VideoPlatform enums - Junction tables: VideoShow, VideoPerformance, VideoSong, VideoMusician - Full API router with CRUD, entity-specific endpoints, link management - Legacy compatibility endpoint for existing youtube_link fields - Building for scale, no shortcuts
563 lines
20 KiB
Python
563 lines
20 KiB
Python
"""
|
|
Video API Router - Modular Video Entity System
|
|
Videos can be linked to multiple entities: shows, performances, songs, musicians
|
|
Building for scale - no shortcuts.
|
|
"""
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, select
|
|
from database import get_session
|
|
from models import (
|
|
Video, VideoShow, VideoPerformance, VideoSong, VideoMusician,
|
|
Show, Performance, Song, Musician, Vertical,
|
|
VideoType, VideoPlatform
|
|
)
|
|
from pydantic import BaseModel
|
|
from datetime import datetime
|
|
import re
|
|
|
|
router = APIRouter(prefix="/videos", tags=["videos"])
|
|
|
|
|
|
# --- Schemas ---
|
|
|
|
class VideoCreate(BaseModel):
|
|
url: str
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
platform: Optional[str] = "youtube"
|
|
video_type: Optional[str] = "single_song"
|
|
duration_seconds: Optional[int] = None
|
|
thumbnail_url: Optional[str] = None
|
|
vertical_id: Optional[int] = None
|
|
# Link IDs (optional on create)
|
|
show_ids: Optional[List[int]] = None
|
|
performance_ids: Optional[List[int]] = None
|
|
song_ids: Optional[List[int]] = None
|
|
musician_ids: Optional[List[int]] = None
|
|
|
|
|
|
class VideoRead(BaseModel):
|
|
id: int
|
|
url: str
|
|
title: Optional[str]
|
|
description: Optional[str]
|
|
platform: str
|
|
video_type: str
|
|
duration_seconds: Optional[int]
|
|
thumbnail_url: Optional[str]
|
|
external_id: Optional[str]
|
|
recorded_date: Optional[datetime]
|
|
published_date: Optional[datetime]
|
|
created_at: datetime
|
|
vertical_id: Optional[int]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class VideoWithRelations(VideoRead):
|
|
"""Video with linked entities"""
|
|
shows: List[dict] = []
|
|
performances: List[dict] = []
|
|
songs: List[dict] = []
|
|
musicians: List[dict] = []
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
def extract_youtube_id(url: str) -> Optional[str]:
|
|
"""Extract YouTube video ID from URL"""
|
|
patterns = [
|
|
r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})',
|
|
]
|
|
for pattern in patterns:
|
|
match = re.search(pattern, url)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
def detect_platform(url: str) -> str:
|
|
"""Detect video platform from URL"""
|
|
url_lower = url.lower()
|
|
if 'youtube.com' in url_lower or 'youtu.be' in url_lower:
|
|
return "youtube"
|
|
elif 'vimeo.com' in url_lower:
|
|
return "vimeo"
|
|
elif 'nugs.net' in url_lower:
|
|
return "nugs"
|
|
elif 'bandcamp.com' in url_lower:
|
|
return "bandcamp"
|
|
elif 'archive.org' in url_lower:
|
|
return "archive"
|
|
return "other"
|
|
|
|
|
|
# --- Endpoints ---
|
|
|
|
@router.get("/", response_model=List[VideoRead])
|
|
def list_videos(
|
|
limit: int = Query(default=50, le=200),
|
|
offset: int = 0,
|
|
platform: Optional[str] = None,
|
|
video_type: Optional[str] = None,
|
|
vertical_id: Optional[int] = None,
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""List all videos with optional filters"""
|
|
query = select(Video).order_by(Video.created_at.desc())
|
|
|
|
if platform:
|
|
query = query.where(Video.platform == platform)
|
|
if video_type:
|
|
query = query.where(Video.video_type == video_type)
|
|
if vertical_id:
|
|
query = query.where(Video.vertical_id == vertical_id)
|
|
|
|
query = query.offset(offset).limit(limit)
|
|
videos = session.exec(query).all()
|
|
|
|
# Convert to response format
|
|
return [
|
|
VideoRead(
|
|
id=v.id,
|
|
url=v.url,
|
|
title=v.title,
|
|
description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds,
|
|
thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id,
|
|
recorded_date=v.recorded_date,
|
|
published_date=v.published_date,
|
|
created_at=v.created_at,
|
|
vertical_id=v.vertical_id,
|
|
)
|
|
for v in videos
|
|
]
|
|
|
|
|
|
@router.get("/stats")
|
|
def get_video_stats(session: Session = Depends(get_session)):
|
|
"""Get video statistics"""
|
|
from sqlmodel import func
|
|
|
|
total_videos = session.exec(select(func.count(Video.id))).one()
|
|
|
|
# Count by platform
|
|
youtube_count = session.exec(
|
|
select(func.count(Video.id)).where(Video.platform == VideoPlatform.YOUTUBE)
|
|
).one()
|
|
|
|
# Count by type
|
|
full_show_count = session.exec(
|
|
select(func.count(Video.id)).where(Video.video_type == VideoType.FULL_SHOW)
|
|
).one()
|
|
|
|
return {
|
|
"total_videos": total_videos,
|
|
"youtube_videos": youtube_count,
|
|
"full_show_videos": full_show_count,
|
|
}
|
|
|
|
|
|
@router.get("/{video_id}", response_model=VideoWithRelations)
|
|
def get_video(video_id: int, session: Session = Depends(get_session)):
|
|
"""Get a single video with all relationships"""
|
|
video = session.get(Video, video_id)
|
|
if not video:
|
|
raise HTTPException(status_code=404, detail="Video not found")
|
|
|
|
# Build response with relations
|
|
return VideoWithRelations(
|
|
id=video.id,
|
|
url=video.url,
|
|
title=video.title,
|
|
description=video.description,
|
|
platform=video.platform.value if hasattr(video.platform, 'value') else str(video.platform),
|
|
video_type=video.video_type.value if hasattr(video.video_type, 'value') else str(video.video_type),
|
|
duration_seconds=video.duration_seconds,
|
|
thumbnail_url=video.thumbnail_url,
|
|
external_id=video.external_id,
|
|
recorded_date=video.recorded_date,
|
|
published_date=video.published_date,
|
|
created_at=video.created_at,
|
|
vertical_id=video.vertical_id,
|
|
shows=[{"id": vs.show.id, "date": vs.show.date.isoformat(), "slug": vs.show.slug} for vs in video.shows if vs.show],
|
|
performances=[{"id": vp.performance.id, "slug": vp.performance.slug} for vp in video.performances if vp.performance],
|
|
songs=[{"id": vs.song.id, "title": vs.song.title, "slug": vs.song.slug} for vs in video.songs if vs.song],
|
|
musicians=[{"id": vm.musician.id, "name": vm.musician.name, "slug": vm.musician.slug} for vm in video.musicians if vm.musician],
|
|
)
|
|
|
|
|
|
@router.post("/", response_model=VideoRead)
|
|
def create_video(video_data: VideoCreate, session: Session = Depends(get_session)):
|
|
"""Create a new video with optional entity links"""
|
|
|
|
# Auto-detect platform
|
|
platform = detect_platform(video_data.url)
|
|
|
|
# Extract external ID for YouTube
|
|
external_id = None
|
|
if platform == "youtube":
|
|
external_id = extract_youtube_id(video_data.url)
|
|
|
|
# Map string to enum
|
|
try:
|
|
platform_enum = VideoPlatform(platform)
|
|
except ValueError:
|
|
platform_enum = VideoPlatform.OTHER
|
|
|
|
try:
|
|
video_type_enum = VideoType(video_data.video_type or "single_song")
|
|
except ValueError:
|
|
video_type_enum = VideoType.OTHER
|
|
|
|
# Create video
|
|
video = Video(
|
|
url=video_data.url,
|
|
title=video_data.title,
|
|
description=video_data.description,
|
|
platform=platform_enum,
|
|
video_type=video_type_enum,
|
|
duration_seconds=video_data.duration_seconds,
|
|
thumbnail_url=video_data.thumbnail_url,
|
|
external_id=external_id,
|
|
vertical_id=video_data.vertical_id,
|
|
)
|
|
session.add(video)
|
|
session.commit()
|
|
session.refresh(video)
|
|
|
|
# Create relationships
|
|
if video_data.show_ids:
|
|
for show_id in video_data.show_ids:
|
|
link = VideoShow(video_id=video.id, show_id=show_id)
|
|
session.add(link)
|
|
|
|
if video_data.performance_ids:
|
|
for perf_id in video_data.performance_ids:
|
|
link = VideoPerformance(video_id=video.id, performance_id=perf_id)
|
|
session.add(link)
|
|
|
|
if video_data.song_ids:
|
|
for song_id in video_data.song_ids:
|
|
link = VideoSong(video_id=video.id, song_id=song_id)
|
|
session.add(link)
|
|
|
|
if video_data.musician_ids:
|
|
for musician_id in video_data.musician_ids:
|
|
link = VideoMusician(video_id=video.id, musician_id=musician_id)
|
|
session.add(link)
|
|
|
|
session.commit()
|
|
|
|
return VideoRead(
|
|
id=video.id,
|
|
url=video.url,
|
|
title=video.title,
|
|
description=video.description,
|
|
platform=video.platform.value if hasattr(video.platform, 'value') else str(video.platform),
|
|
video_type=video.video_type.value if hasattr(video.video_type, 'value') else str(video.video_type),
|
|
duration_seconds=video.duration_seconds,
|
|
thumbnail_url=video.thumbnail_url,
|
|
external_id=video.external_id,
|
|
recorded_date=video.recorded_date,
|
|
published_date=video.published_date,
|
|
created_at=video.created_at,
|
|
vertical_id=video.vertical_id,
|
|
)
|
|
|
|
|
|
# --- Entity-specific video endpoints ---
|
|
|
|
@router.get("/by-show/{show_id}", response_model=List[VideoRead])
|
|
def get_videos_for_show(show_id: int, session: Session = Depends(get_session)):
|
|
"""Get all videos linked to a specific show"""
|
|
query = select(Video).join(VideoShow).where(VideoShow.show_id == show_id)
|
|
videos = session.exec(query).all()
|
|
return [
|
|
VideoRead(
|
|
id=v.id, url=v.url, title=v.title, description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds, thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id, recorded_date=v.recorded_date,
|
|
published_date=v.published_date, created_at=v.created_at, vertical_id=v.vertical_id
|
|
) for v in videos
|
|
]
|
|
|
|
|
|
@router.get("/by-performance/{performance_id}", response_model=List[VideoRead])
|
|
def get_videos_for_performance(performance_id: int, session: Session = Depends(get_session)):
|
|
"""Get all videos linked to a specific performance"""
|
|
query = select(Video).join(VideoPerformance).where(VideoPerformance.performance_id == performance_id)
|
|
videos = session.exec(query).all()
|
|
return [
|
|
VideoRead(
|
|
id=v.id, url=v.url, title=v.title, description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds, thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id, recorded_date=v.recorded_date,
|
|
published_date=v.published_date, created_at=v.created_at, vertical_id=v.vertical_id
|
|
) for v in videos
|
|
]
|
|
|
|
|
|
@router.get("/by-song/{song_id}", response_model=List[VideoRead])
|
|
def get_videos_for_song(song_id: int, session: Session = Depends(get_session)):
|
|
"""Get all videos linked to a specific song"""
|
|
query = select(Video).join(VideoSong).where(VideoSong.song_id == song_id)
|
|
videos = session.exec(query).all()
|
|
return [
|
|
VideoRead(
|
|
id=v.id, url=v.url, title=v.title, description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds, thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id, recorded_date=v.recorded_date,
|
|
published_date=v.published_date, created_at=v.created_at, vertical_id=v.vertical_id
|
|
) for v in videos
|
|
]
|
|
|
|
|
|
@router.get("/by-musician/{musician_id}", response_model=List[VideoRead])
|
|
def get_videos_for_musician(musician_id: int, session: Session = Depends(get_session)):
|
|
"""Get all videos linked to a specific musician"""
|
|
query = select(Video).join(VideoMusician).where(VideoMusician.musician_id == musician_id)
|
|
videos = session.exec(query).all()
|
|
return [
|
|
VideoRead(
|
|
id=v.id, url=v.url, title=v.title, description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds, thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id, recorded_date=v.recorded_date,
|
|
published_date=v.published_date, created_at=v.created_at, vertical_id=v.vertical_id
|
|
) for v in videos
|
|
]
|
|
|
|
|
|
@router.get("/by-band/{vertical_slug}", response_model=List[VideoRead])
|
|
def get_videos_for_band(
|
|
vertical_slug: str,
|
|
limit: int = Query(default=50, le=200),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Get all videos for a band/vertical"""
|
|
vertical = session.exec(select(Vertical).where(Vertical.slug == vertical_slug)).first()
|
|
if not vertical:
|
|
raise HTTPException(status_code=404, detail="Band not found")
|
|
|
|
query = select(Video).where(Video.vertical_id == vertical.id).order_by(Video.created_at.desc()).limit(limit)
|
|
videos = session.exec(query).all()
|
|
return [
|
|
VideoRead(
|
|
id=v.id, url=v.url, title=v.title, description=v.description,
|
|
platform=v.platform.value if hasattr(v.platform, 'value') else str(v.platform),
|
|
video_type=v.video_type.value if hasattr(v.video_type, 'value') else str(v.video_type),
|
|
duration_seconds=v.duration_seconds, thumbnail_url=v.thumbnail_url,
|
|
external_id=v.external_id, recorded_date=v.recorded_date,
|
|
published_date=v.published_date, created_at=v.created_at, vertical_id=v.vertical_id
|
|
) for v in videos
|
|
]
|
|
|
|
|
|
# --- Link management endpoints ---
|
|
|
|
@router.post("/{video_id}/link-show/{show_id}")
|
|
def link_video_to_show(video_id: int, show_id: int, session: Session = Depends(get_session)):
|
|
"""Link a video to a show"""
|
|
video = session.get(Video, video_id)
|
|
show = session.get(Show, show_id)
|
|
if not video or not show:
|
|
raise HTTPException(status_code=404, detail="Video or show not found")
|
|
|
|
existing = session.exec(
|
|
select(VideoShow).where(VideoShow.video_id == video_id, VideoShow.show_id == show_id)
|
|
).first()
|
|
if existing:
|
|
return {"message": "Already linked"}
|
|
|
|
link = VideoShow(video_id=video_id, show_id=show_id)
|
|
session.add(link)
|
|
session.commit()
|
|
return {"message": "Linked successfully"}
|
|
|
|
|
|
@router.post("/{video_id}/link-performance/{performance_id}")
|
|
def link_video_to_performance(
|
|
video_id: int,
|
|
performance_id: int,
|
|
timestamp_start: Optional[int] = Query(default=None),
|
|
timestamp_end: Optional[int] = Query(default=None),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Link a video to a performance with optional timestamps"""
|
|
video = session.get(Video, video_id)
|
|
performance = session.get(Performance, performance_id)
|
|
if not video or not performance:
|
|
raise HTTPException(status_code=404, detail="Video or performance not found")
|
|
|
|
existing = session.exec(
|
|
select(VideoPerformance).where(
|
|
VideoPerformance.video_id == video_id,
|
|
VideoPerformance.performance_id == performance_id
|
|
)
|
|
).first()
|
|
if existing:
|
|
return {"message": "Already linked"}
|
|
|
|
link = VideoPerformance(
|
|
video_id=video_id,
|
|
performance_id=performance_id,
|
|
timestamp_start=timestamp_start,
|
|
timestamp_end=timestamp_end
|
|
)
|
|
session.add(link)
|
|
session.commit()
|
|
return {"message": "Linked successfully"}
|
|
|
|
|
|
@router.post("/{video_id}/link-song/{song_id}")
|
|
def link_video_to_song(video_id: int, song_id: int, session: Session = Depends(get_session)):
|
|
"""Link a video to a song"""
|
|
video = session.get(Video, video_id)
|
|
song = session.get(Song, song_id)
|
|
if not video or not song:
|
|
raise HTTPException(status_code=404, detail="Video or song not found")
|
|
|
|
existing = session.exec(
|
|
select(VideoSong).where(VideoSong.video_id == video_id, VideoSong.song_id == song_id)
|
|
).first()
|
|
if existing:
|
|
return {"message": "Already linked"}
|
|
|
|
link = VideoSong(video_id=video_id, song_id=song_id)
|
|
session.add(link)
|
|
session.commit()
|
|
return {"message": "Linked successfully"}
|
|
|
|
|
|
@router.post("/{video_id}/link-musician/{musician_id}")
|
|
def link_video_to_musician(
|
|
video_id: int,
|
|
musician_id: int,
|
|
role: Optional[str] = Query(default=None),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Link a video to a musician"""
|
|
video = session.get(Video, video_id)
|
|
musician = session.get(Musician, musician_id)
|
|
if not video or not musician:
|
|
raise HTTPException(status_code=404, detail="Video or musician not found")
|
|
|
|
existing = session.exec(
|
|
select(VideoMusician).where(VideoMusician.video_id == video_id, VideoMusician.musician_id == musician_id)
|
|
).first()
|
|
if existing:
|
|
return {"message": "Already linked"}
|
|
|
|
link = VideoMusician(video_id=video_id, musician_id=musician_id, role=role)
|
|
session.add(link)
|
|
session.commit()
|
|
return {"message": "Linked successfully"}
|
|
|
|
|
|
# --- Legacy compatibility: Also return youtube_link from existing Show/Performance fields ---
|
|
|
|
@router.get("/legacy/all")
|
|
def get_legacy_videos(
|
|
limit: int = Query(default=100, le=500),
|
|
offset: int = Query(default=0),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Legacy endpoint: Get shows and performances with youtube_link fields (for backwards compatibility)"""
|
|
from models import Venue
|
|
|
|
# Get performances with youtube_link
|
|
perf_query = (
|
|
select(
|
|
Performance.id,
|
|
Performance.youtube_link,
|
|
Performance.show_id,
|
|
Song.id.label("song_id"),
|
|
Song.title.label("song_title"),
|
|
Song.slug.label("song_slug"),
|
|
Show.date,
|
|
Show.slug.label("show_slug"),
|
|
Venue.name.label("venue_name"),
|
|
Venue.city.label("venue_city"),
|
|
Venue.state.label("venue_state"),
|
|
)
|
|
.join(Song, Performance.song_id == Song.id)
|
|
.join(Show, Performance.show_id == Show.id)
|
|
.outerjoin(Venue, Show.venue_id == Venue.id)
|
|
.where(Performance.youtube_link != None)
|
|
.order_by(Show.date.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
|
|
perf_results = session.exec(perf_query).all()
|
|
|
|
performances = [
|
|
{
|
|
"type": "performance",
|
|
"id": r[0],
|
|
"youtube_link": r[1],
|
|
"show_id": r[2],
|
|
"song_id": r[3],
|
|
"song_title": r[4],
|
|
"date": r[6].isoformat() if r[6] else None,
|
|
"show_slug": r[7],
|
|
"venue_name": r[8],
|
|
"venue_city": r[9],
|
|
"venue_state": r[10],
|
|
}
|
|
for r in perf_results
|
|
]
|
|
|
|
# Get shows with youtube_link
|
|
show_query = (
|
|
select(
|
|
Show.id,
|
|
Show.youtube_link,
|
|
Show.date,
|
|
Show.slug,
|
|
Venue.name.label("venue_name"),
|
|
Venue.city.label("venue_city"),
|
|
Venue.state.label("venue_state"),
|
|
)
|
|
.outerjoin(Venue, Show.venue_id == Venue.id)
|
|
.where(Show.youtube_link != None)
|
|
.order_by(Show.date.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
|
|
show_results = session.exec(show_query).all()
|
|
|
|
shows = [
|
|
{
|
|
"type": "full_show",
|
|
"id": r[0],
|
|
"youtube_link": r[1],
|
|
"date": r[2].isoformat() if r[2] else None,
|
|
"show_slug": r[3],
|
|
"venue_name": r[4],
|
|
"venue_city": r[5],
|
|
"venue_state": r[6],
|
|
}
|
|
for r in show_results
|
|
]
|
|
|
|
return {
|
|
"performances": performances,
|
|
"shows": shows,
|
|
"total_performances": len(performances),
|
|
"total_shows": len(shows)
|
|
}
|