5b236608f8
- Add routers/canon.py with endpoints:
- GET /canon - list all canonical songs with versions
- GET /canon/{slug} - get canon with all band versions
- GET /canon/song/{id}/related - get related versions
- Add link_canon_songs.py auto-linker script
- Finds songs with same title across bands
- Creates SongCanon entries automatically
- Run with --apply to execute
142 lines
4 KiB
Python
142 lines
4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, select
|
|
from typing import List
|
|
from database import get_session
|
|
from models import SongCanon, Song, Vertical
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter(prefix="/canon", tags=["canon"])
|
|
|
|
|
|
class SongVersionRead(BaseModel):
|
|
id: int
|
|
title: str
|
|
slug: str | None
|
|
vertical_id: int
|
|
vertical_name: str
|
|
vertical_slug: str
|
|
|
|
|
|
class SongCanonRead(BaseModel):
|
|
id: int
|
|
title: str
|
|
slug: str
|
|
original_artist: str | None
|
|
notes: str | None
|
|
versions: List[SongVersionRead]
|
|
|
|
|
|
class SongCanonCreate(BaseModel):
|
|
title: str
|
|
slug: str
|
|
original_artist: str | None = None
|
|
notes: str | None = None
|
|
|
|
|
|
@router.get("/", response_model=List[SongCanonRead])
|
|
def list_canon_songs(
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""List all canonical songs with their cross-band versions"""
|
|
canons = session.exec(
|
|
select(SongCanon).offset(offset).limit(limit)
|
|
).all()
|
|
|
|
result = []
|
|
for canon in canons:
|
|
versions = []
|
|
songs = session.exec(
|
|
select(Song).where(Song.canon_id == canon.id)
|
|
).all()
|
|
|
|
for song in songs:
|
|
vertical = session.get(Vertical, song.vertical_id)
|
|
versions.append({
|
|
"id": song.id,
|
|
"title": song.title,
|
|
"slug": song.slug,
|
|
"vertical_id": song.vertical_id,
|
|
"vertical_name": vertical.name if vertical else "Unknown",
|
|
"vertical_slug": vertical.slug if vertical else "unknown"
|
|
})
|
|
|
|
result.append({
|
|
"id": canon.id,
|
|
"title": canon.title,
|
|
"slug": canon.slug,
|
|
"original_artist": canon.original_artist,
|
|
"notes": canon.notes,
|
|
"versions": versions
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
@router.get("/{slug}", response_model=SongCanonRead)
|
|
def get_canon_song(slug: str, session: Session = Depends(get_session)):
|
|
"""Get a canonical song with all its band-specific versions"""
|
|
canon = session.exec(
|
|
select(SongCanon).where(SongCanon.slug == slug)
|
|
).first()
|
|
|
|
if not canon:
|
|
raise HTTPException(status_code=404, detail="Canonical song not found")
|
|
|
|
versions = []
|
|
songs = session.exec(
|
|
select(Song).where(Song.canon_id == canon.id)
|
|
).all()
|
|
|
|
for song in songs:
|
|
vertical = session.get(Vertical, song.vertical_id)
|
|
versions.append({
|
|
"id": song.id,
|
|
"title": song.title,
|
|
"slug": song.slug,
|
|
"vertical_id": song.vertical_id,
|
|
"vertical_name": vertical.name if vertical else "Unknown",
|
|
"vertical_slug": vertical.slug if vertical else "unknown"
|
|
})
|
|
|
|
return {
|
|
"id": canon.id,
|
|
"title": canon.title,
|
|
"slug": canon.slug,
|
|
"original_artist": canon.original_artist,
|
|
"notes": canon.notes,
|
|
"versions": versions
|
|
}
|
|
|
|
|
|
@router.get("/song/{song_id}/related", response_model=List[SongVersionRead])
|
|
def get_related_versions(song_id: int, session: Session = Depends(get_session)):
|
|
"""Get all versions of the same song across bands"""
|
|
song = session.get(Song, song_id)
|
|
if not song:
|
|
raise HTTPException(status_code=404, detail="Song not found")
|
|
|
|
if not song.canon_id:
|
|
return []
|
|
|
|
# Get all songs with same canon_id (excluding this one)
|
|
related = session.exec(
|
|
select(Song)
|
|
.where(Song.canon_id == song.canon_id)
|
|
.where(Song.id != song_id)
|
|
).all()
|
|
|
|
result = []
|
|
for s in related:
|
|
vertical = session.get(Vertical, s.vertical_id)
|
|
result.append({
|
|
"id": s.id,
|
|
"title": s.title,
|
|
"slug": s.slug,
|
|
"vertical_id": s.vertical_id,
|
|
"vertical_name": vertical.name if vertical else "Unknown",
|
|
"vertical_slug": vertical.slug if vertical else "unknown"
|
|
})
|
|
|
|
return result
|