import re from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.database import get_db from app.core.security import get_current_user from app.models.user import User from app.models.playlist import Playlist from app.models.track import Track from app.services.recommender import build_taste_profile from app.schemas.playlist import PlaylistDetailResponse router = APIRouter(prefix="/import", tags=["import"]) class PasteImportRequest(BaseModel): name: str text: str def parse_song_line(line: str) -> dict | None: line = line.strip() if not line: return None # Strip leading numbering: "1.", "2)", "1 -", "01.", etc. line = re.sub(r"^\d+[\.\)\-\:]\s*", "", line).strip() if not line: return None # Try "Artist - Title" (most common) if " - " in line: parts = line.split(" - ", 1) return {"artist": parts[0].strip(), "title": parts[1].strip()} # Try "Title by Artist" if " by " in line.lower(): idx = line.lower().index(" by ") return {"title": line[:idx].strip(), "artist": line[idx + 4 :].strip()} # Try "Artist: Title" if ": " in line: parts = line.split(": ", 1) return {"artist": parts[0].strip(), "title": parts[1].strip()} # Fallback: treat whole line as title with unknown artist return {"title": line, "artist": "Unknown"} @router.post("/paste", response_model=PlaylistDetailResponse) async def import_pasted_songs( data: PasteImportRequest, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not data.name.strip(): raise HTTPException(status_code=400, detail="Playlist name is required") if not data.text.strip(): raise HTTPException(status_code=400, detail="No songs provided") # Free tier limit if not user.is_pro: result = await db.execute( select(Playlist).where(Playlist.user_id == user.id) ) existing = list(result.scalars().all()) if len(existing) >= settings.FREE_MAX_PLAYLISTS: raise HTTPException( status_code=403, detail="Free tier limited to 1 playlist. Upgrade to Pro for unlimited.", ) # Parse lines lines = data.text.strip().splitlines() parsed = [parse_song_line(line) for line in lines] parsed = [p for p in parsed if p is not None] if not parsed: raise HTTPException(status_code=400, detail="Could not parse any songs from the text") # Create playlist playlist = Playlist( user_id=user.id, name=data.name.strip(), platform_source="manual", track_count=len(parsed), ) db.add(playlist) await db.flush() # Create tracks tracks = [] for p in parsed: track = Track( playlist_id=playlist.id, title=p["title"], artist=p["artist"], ) db.add(track) tracks.append(track) await db.flush() # Build taste profile playlist.taste_profile = build_taste_profile(tracks) playlist.tracks = tracks return playlist