Initial Ytapp app

.gitignore

@@ -0,0 +1,37 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+env/
+venv/
+ENV/
+.venv
+
+# Environment variables
+.env
+.env.local
+
+# OAuth credentials (sensitive)
+oauth.json
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Logs
+*.log
+
+# Distribution / packaging
+dist/
+build/
+*.egg-info/
+

Dockerfile

@@ -0,0 +1,37 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV TF_CPP_MIN_LOG_LEVEL=2
+
+RUN apt-get update && apt-get install -y \
+    libgl1 \
+    libglib2.0-0 \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+RUN useradd -m -u 1000 user
+USER user
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+RUN mkdir -p /home/user/.deepface/weights && chmod -R 777 /home/user/.deepface
+
+RUN python - << 'PY'
+import numpy as np
+from deepface import DeepFace
+try:
+    DeepFace.analyze(np.zeros((224, 224, 3), dtype=np.uint8), actions=['emotion'], enforce_detection=False)
+except:
+    pass
+PY
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]

System_overview.md

@@ -0,0 +1,117 @@
+## System overview – Ytapp YouTube Music Mood-based Recommender
+
+Independent FastAPI service for mood-based YouTube Music recommendations using emotionAI's DeepFace + Gemini API approach.
+
+## Features
+
+- Text emotion analysis (Gemini API) → YouTube Music song recommendations
+- Face emotion analysis (DeepFace fast / Gemini API accurate) → YouTube Music song recommendations
+- Search songs, artists, and get song details
+- Standalone service (independent of main VibeCheck app)
+
+## Setup
+
+1. **Install dependencies:**
+   ```bash
+   pip install -r requirements.txt --break-system-packages
+   ```
+
+2. **Configure Gemini API Key (Optional but recommended):**
+   - Create a `.env` file in the Ytapp directory
+   - Add: `GEMINI_API_KEY=your_api_key_here`
+   - Without Gemini API key, face analysis falls back to DeepFace only
+   - Text analysis requires Gemini API key
+
+3. **Optional: OAuth Authentication**
+   - For authenticated requests (library management, playlists, etc.), set up OAuth:
+   - Get Client ID and Secret from [YouTube Data API](https://developers.google.com/youtube/v3)
+   - Select OAuth client ID → TVs and Limited Input devices
+   - Run: `ytmusicapi oauth`
+   - Follow instructions to create `oauth.json`
+   - Pass credentials to `YTMusic()` if needed
+
+## Running Locally
+
+```bash
+python app.py
+```
+
+The service will run on `http://localhost:7860`
+
+## Docker
+
+```bash
+docker build -t ytapp .
+docker run -p 7860:7860 -e GEMINI_API_KEY=your_key_here ytapp
+```
+
+## API Endpoints
+
+### Health Check
+- `GET /health` - Service health status
+
+### Search
+- `GET /search?query=...&limit=20` - Search for songs
+- `GET /song/{video_id}` - Get song details
+- `GET /artists/search?query=...&limit=10` - Search for artists
+- `GET /artists/{artist_id}/songs?limit=50` - Get artist songs
+
+### Mood-based Recommendations
+- `POST /mood/text` - Get song recommendation from text mood (uses Gemini API)
+  ```json
+  {"text": "I feel happy today!"}
+  ```
+- `POST /mood/face` - Get song recommendation from face image (tries Gemini, falls back to DeepFace)
+  ```
+  Form data: file (image/jpeg)
+  ```
+- `POST /mood/face/live` - Fast face analysis using DeepFace only (for live video feeds)
+  ```
+  Form data: file (image/jpeg)
+  ```
+
+## Response Format
+
+Mood endpoints return:
+```json
+{
+  "mood_label": "joy",
+  "mood_score": 0.95,
+  "video_id": "dQw4w9WgXcQ",
+  "title": "Song Title",
+  "artists": ["Artist Name"],
+  "album": "Album Name",
+  "duration": "3:45",
+  "image_url": "https://...",
+  "external_url": "https://music.youtube.com/watch?v=..."
+}
+```
+
+## Emotion Detection Methods
+
+### Face Analysis
+1. **Primary (if Gemini API key available)**: Uses Gemini 1.5 Flash for accurate emotion detection
+2. **Fallback**: Uses DeepFace for fast offline emotion detection
+3. **Live mode**: Uses DeepFace only for real-time video feeds
+
+### Text Analysis
+- Uses Gemini API to analyze emotional tone of text
+- Requires `GEMINI_API_KEY` to be set
+
+## Emotion Mapping
+
+- **Joy/Happy** → Happy, upbeat, dance music
+- **Sad** → Sad songs, ballads, emotional music
+- **Anger** → Rock, metal, intense music
+- **Fear** → Calm, ambient, meditation music
+- **Surprise** → Experimental, indie, alternative music
+- **Neutral** → Chill, background, easy listening
+- **Disgust** → Alternative rock, indie music
+
+## Notes
+
+- DeepFace models are preloaded during Docker build for faster startup
+- Gemini API provides more accurate emotion detection but requires API key
+- DeepFace works offline and is faster for live video feeds
+- No authentication required for basic searches and mood recommendations
+- OAuth optional for library management and playlist creation

__init__.py

@@ -0,0 +1,18 @@
+from .ytmusic_client import (
+    YouTubeMusicError,
+    search_songs,
+    get_song_info,
+    search_artists,
+    get_artist_songs,
+    recommend_song_for_emotion,
+)
+
+__all__ = [
+    "YouTubeMusicError",
+    "search_songs",
+    "get_song_info",
+    "search_artists",
+    "get_artist_songs",
+    "recommend_song_for_emotion",
+]
+

app.py

@@ -0,0 +1,345 @@
+import os
+import base64
+import json
+import traceback
+
+import numpy as np
+import cv2
+import requests
+from deepface import DeepFace
+from dotenv import load_dotenv
+from fastapi import FastAPI, File, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from ytmusic_client import (
+    YouTubeMusicError,
+    search_songs,
+    get_song_info,
+    search_artists,
+    get_artist_songs,
+    recommend_song_for_emotion,
+)
+
+
+os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
+
+load_dotenv()
+
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+if GEMINI_API_KEY:
+    GEMINI_API_KEY = GEMINI_API_KEY.strip().replace('"', '').replace("'", "")
+
+if not GEMINI_API_KEY or GEMINI_API_KEY == "YOUR_API_KEY_HERE":
+    print("⚠️ WARNING: GEMINI_API_KEY not found or using placeholder.")
+else:
+    masked_key = f"{GEMINI_API_KEY[:4]}...{GEMINI_API_KEY[-4:]}"
+    print(f"✅ API Key detected: {masked_key} (Length: {len(GEMINI_API_KEY)})")
+
+YTMUSIC_OAUTH_FILE = os.getenv("YTMUSIC_OAUTH_FILE", "oauth.json")
+YTMUSIC_CLIENT_ID = os.getenv("YTMUSIC_CLIENT_ID")
+YTMUSIC_CLIENT_SECRET = os.getenv("YTMUSIC_CLIENT_SECRET")
+
+if os.path.exists(YTMUSIC_OAUTH_FILE):
+    import json
+    with open(YTMUSIC_OAUTH_FILE, 'r') as f:
+        oauth_data = json.load(f)
+        if "oauth_credentials" in oauth_data:
+            print(f"✅ YouTube Music OAuth file found with credentials: {YTMUSIC_OAUTH_FILE}")
+        else:
+            print(f"ℹ️  YouTube Music OAuth file found but incomplete: {YTMUSIC_OAUTH_FILE}")
+else:
+    print(f"ℹ️  YouTube Music OAuth file not found: {YTMUSIC_OAUTH_FILE}")
+    print("   Run: ytmusicapi oauth to set up authentication (optional)")
+
+
+app = FastAPI(title="Ytapp – YouTube Music Mood-based Recommender")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+class TextMoodRequest(BaseModel):
+    text: str
+
+
+class RecommendationResponse(BaseModel):
+    mood_label: str
+    mood_score: float
+    video_id: str | None
+    title: str | None
+    artists: list[str] | None
+    album: str | None
+    duration: str | None
+    image_url: str | None
+    external_url: str | None
+
+
+def _analyze_face_deepface(image_bytes: bytes) -> tuple[str, float]:
+    npimg = np.frombuffer(image_bytes, np.uint8)
+    img = cv2.imdecode(npimg, cv2.IMREAD_COLOR)
+    
+    result = DeepFace.analyze(img, actions=['emotion'], enforce_detection=False)
+    res = result[0] if isinstance(result, list) else result
+    
+    emotions_dict = {key: float(value) for key, value in res['emotion'].items() if key != 'disgust'}
+    total = sum(emotions_dict.values())
+    if total > 0:
+        emotions_dict = {key: (value / total) * 100 for key, value in emotions_dict.items()}
+    
+    dominant = max(emotions_dict, key=emotions_dict.get)
+    score = emotions_dict[dominant] / 100.0
+    
+    emotion_map = {
+        "happy": "joy",
+        "sad": "sadness",
+        "angry": "anger",
+        "fear": "fear",
+        "surprise": "surprise",
+        "neutral": "neutral",
+    }
+    
+    return emotion_map.get(dominant, "neutral"), score
+
+
+def _analyze_face_gemini(image_bytes: bytes) -> tuple[str, float]:
+    if not GEMINI_API_KEY or GEMINI_API_KEY == "YOUR_API_KEY_HERE":
+        raise ValueError("GEMINI_API_KEY not configured")
+    
+    base64_image = base64.b64encode(image_bytes).decode('utf-8')
+    
+    prompt = """
+    You are an emotion detection AI. Analyze the facial expression in this image.
+    DO NOT use 'disgust'.
+    
+    Return ONLY a valid JSON object with this exact structure:
+    {
+      "dominant_emotion": "happy|sad|angry|neutral|fear|surprise",
+      "confidence": 0.0-1.0
+    }
+    """
+    
+    url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={GEMINI_API_KEY}"
+    payload = {
+        "contents": [{"parts": [{"text": prompt}, {"inline_data": {"mime_type": "image/jpeg", "data": base64_image}}]}],
+        "generationConfig": {"response_mime_type": "application/json"}
+    }
+    
+    raw_response = requests.post(url, json=payload)
+    response = raw_response.json()
+    
+    if 'error' in response:
+        raise ValueError(f"Gemini API Error: {response['error']['message']}")
+    
+    if 'candidates' not in response:
+        raise ValueError("Image blocked by AI Safety Filters")
+    
+    result = json.loads(response['candidates'][0]['content']['parts'][0]['text'])
+    
+    emotion_map = {
+        "happy": "joy",
+        "sad": "sadness",
+        "angry": "anger",
+        "fear": "fear",
+        "surprise": "surprise",
+        "neutral": "neutral",
+    }
+    
+    dominant = result.get("dominant_emotion", "neutral").lower()
+    confidence = float(result.get("confidence", 0.5))
+    
+    return emotion_map.get(dominant, "neutral"), confidence
+
+
+def _analyze_text_gemini(text: str) -> tuple[str, float]:
+    if not GEMINI_API_KEY or GEMINI_API_KEY == "YOUR_API_KEY_HERE":
+        raise ValueError("GEMINI_API_KEY not configured")
+    
+    prompt = f"""
+    Analyze the emotional tone of this text: "{text}"
+    
+    Return ONLY a valid JSON object with this exact structure:
+    {{
+      "dominant_emotion": "joy|sadness|anger|neutral|fear|surprise",
+      "confidence": 0.0-1.0
+    }}
+    """
+    
+    url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={GEMINI_API_KEY}"
+    payload = {
+        "contents": [{"parts": [{"text": prompt}]}],
+        "generationConfig": {"response_mime_type": "application/json"}
+    }
+    
+    raw_response = requests.post(url, json=payload)
+    response = raw_response.json()
+    
+    if 'error' in response:
+        raise ValueError(f"Gemini API Error: {response['error']['message']}")
+    
+    if 'candidates' not in response:
+        raise ValueError("Text analysis blocked")
+    
+    result = json.loads(response['candidates'][0]['content']['parts'][0]['text'])
+    
+    dominant = result.get("dominant_emotion", "neutral").lower()
+    confidence = float(result.get("confidence", 0.5))
+    
+    return dominant, confidence
+
+
+@app.get("/health")
+def health() -> dict:
+    return {"status": "ok"}
+
+
+@app.get("/search")
+def search_songs_endpoint(query: str, limit: int = 20) -> dict:
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        songs = search_songs(query, limit=limit, oauth_file=oauth_file)
+        return {"query": query, "limit": limit, "songs": songs}
+    except YouTubeMusicError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(exc)}") from exc
+
+
+@app.get("/song/{video_id}")
+def get_song_endpoint(video_id: str) -> dict:
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        song = get_song_info(video_id, oauth_file=oauth_file)
+        return song
+    except YouTubeMusicError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(exc)}") from exc
+
+
+@app.get("/artists/search")
+def search_artists_endpoint(query: str, limit: int = 10) -> dict:
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        artists = search_artists(query, limit=limit, oauth_file=oauth_file)
+        return {"query": query, "limit": limit, "artists": artists}
+    except YouTubeMusicError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(exc)}") from exc
+
+
+@app.get("/artists/{artist_id}/songs")
+def artist_songs_endpoint(artist_id: str, limit: int = 50) -> dict:
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        songs = get_artist_songs(artist_id, limit=limit, oauth_file=oauth_file)
+        return {"artist_id": artist_id, "limit": limit, "songs": songs}
+    except YouTubeMusicError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(exc)}") from exc
+
+
+@app.post("/mood/text", response_model=RecommendationResponse)
+def mood_from_text(body: TextMoodRequest) -> RecommendationResponse:
+    if not body.text.strip():
+        raise HTTPException(status_code=400, detail="Text cannot be empty")
+
+    try:
+        label, score = _analyze_text_gemini(body.text)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Emotion analysis failed: {str(e)}")
+
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        song = recommend_song_for_emotion(label, source="text", oauth_file=oauth_file)
+    except Exception:
+        song = {}
+
+    return RecommendationResponse(
+        mood_label=label,
+        mood_score=score,
+        video_id=song.get("video_id"),
+        title=song.get("title"),
+        artists=song.get("artists", []),
+        album=song.get("album"),
+        duration=song.get("duration"),
+        image_url=song.get("image_url"),
+        external_url=song.get("external_url"),
+    )
+
+
+@app.post("/mood/face", response_model=RecommendationResponse)
+async def mood_from_face(file: UploadFile = File(...)) -> RecommendationResponse:
+    contents = await file.read()
+    
+    try:
+        label, score = _analyze_face_gemini(contents)
+    except Exception:
+        try:
+            label, score = _analyze_face_deepface(contents)
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Face analysis failed: {str(e)}")
+
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        song = recommend_song_for_emotion(label, source="face", oauth_file=oauth_file)
+    except Exception:
+        song = {}
+
+    return RecommendationResponse(
+        mood_label=label,
+        mood_score=score,
+        video_id=song.get("video_id"),
+        title=song.get("title"),
+        artists=song.get("artists", []),
+        album=song.get("album"),
+        duration=song.get("duration"),
+        image_url=song.get("image_url"),
+        external_url=song.get("external_url"),
+    )
+
+
+@app.post("/mood/face/live", response_model=RecommendationResponse)
+async def mood_from_face_live(file: UploadFile = File(...)) -> RecommendationResponse:
+    contents = await file.read()
+    
+    try:
+        label, score = _analyze_face_deepface(contents)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Live face analysis failed: {str(e)}")
+
+    try:
+        oauth_file = YTMUSIC_OAUTH_FILE if os.path.exists(YTMUSIC_OAUTH_FILE) else None
+        song = recommend_song_for_emotion(label, source="face", oauth_file=oauth_file)
+    except Exception:
+        song = {}
+
+    return RecommendationResponse(
+        mood_label=label,
+        mood_score=score,
+        video_id=song.get("video_id"),
+        title=song.get("title"),
+        artists=song.get("artists", []),
+        album=song.get("album"),
+        duration=song.get("duration"),
+        image_url=song.get("image_url"),
+        external_url=song.get("external_url"),
+    )
+
+
+print("⏳ Waking up local AI...")
+try:
+    DeepFace.analyze(np.zeros((224, 224, 3), dtype=np.uint8), actions=['emotion'], enforce_detection=False)
+except:
+    pass
+print("✅ SYSTEM READY!")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -0,0 +1,11 @@
+fastapi
+uvicorn[standard]
+ytmusicapi==1.10.0
+deepface
+numpy<2.0
+opencv-python-headless
+requests
+python-dotenv
+tf-keras
+tensorflow
+python-multipart

ytmusic_client.py

@@ -0,0 +1,299 @@
+from __future__ import annotations
+
+import random
+from typing import Any, Dict, List, Optional
+
+try:
+    from ytmusicapi import YTMusic, OAuthCredentials
+except ImportError:
+    YTMusic = None
+    OAuthCredentials = None
+
+
+class YouTubeMusicError(Exception):
+    def __init__(self, message: str, status_code: int = 500):
+        self.message = message
+        self.status_code = status_code
+        super().__init__(self.message)
+
+
+def _get_ytmusic_client(oauth_file: Optional[str] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None) -> Any:
+    if YTMusic is None:
+        raise YouTubeMusicError(
+            "ytmusicapi is not installed. Install it with: pip install ytmusicapi==1.11.5",
+            status_code=500
+        )
+    
+    try:
+        import os
+        from dotenv import load_dotenv
+        load_dotenv()
+        
+        # Get OAuth credentials from environment or parameters
+        oauth_file = oauth_file or os.getenv("YTMUSIC_OAUTH_FILE", "oauth.json")
+        client_id = client_id or os.getenv("YTMUSIC_CLIENT_ID")
+        client_secret = client_secret or os.getenv("YTMUSIC_CLIENT_SECRET")
+        
+        # If OAuth file exists and has credentials, use it
+        if os.path.exists(oauth_file):
+            import json
+            try:
+                with open(oauth_file, 'r') as f:
+                    oauth_data = json.load(f)
+                    if "oauth_credentials" in oauth_data:
+                        if client_id and client_secret:
+                            # Use OAuth with Client ID/Secret (required as of Nov 2024)
+                            if OAuthCredentials is None:
+                                raise YouTubeMusicError("OAuthCredentials not available. Update ytmusicapi.", status_code=500)
+                            return YTMusic(oauth_file, oauth_credentials=OAuthCredentials(client_id=client_id, client_secret=client_secret))
+                        else:
+                            # Try without credentials (may work for some operations)
+                            return YTMusic(oauth_file)
+            except Exception:
+                pass
+        
+        # Fall back to public access (works for search and recommendations)
+        return YTMusic()
+    except Exception as e:
+        raise YouTubeMusicError(f"Failed to initialize YTMusic client: {str(e)}", status_code=500)
+
+
+def search_songs(query: str, limit: int = 20, oauth_file: Optional[str] = None) -> List[Dict[str, Any]]:
+    try:
+        yt = _get_ytmusic_client(oauth_file)
+        results = yt.search(query, filter="songs", limit=limit)
+        
+        songs = []
+        for item in results:
+            if item.get("resultType") == "song":
+                video_id = item.get("videoId")
+                if not video_id:
+                    continue
+                
+                artists = []
+                if "artists" in item:
+                    for artist in item["artists"]:
+                        if isinstance(artist, dict):
+                            artists.append(artist.get("name", ""))
+                        elif isinstance(artist, str):
+                            artists.append(artist)
+                
+                album = None
+                if "album" in item and isinstance(item["album"], dict):
+                    album = item["album"].get("name")
+                elif isinstance(item.get("album"), str):
+                    album = item["album"]
+                
+                thumbnails = item.get("thumbnails", [])
+                image_url = None
+                if thumbnails:
+                    image_url = thumbnails[-1].get("url") if isinstance(thumbnails[-1], dict) else None
+                
+                songs.append({
+                    "video_id": video_id,
+                    "title": item.get("title", ""),
+                    "artists": artists,
+                    "album": album,
+                    "duration": item.get("duration"),
+                    "image_url": image_url,
+                    "playlist_id": item.get("playlistId"),
+                })
+        
+        return songs[:limit]
+    except YouTubeMusicError:
+        raise
+    except Exception as e:
+        raise YouTubeMusicError(f"Search failed: {str(e)}", status_code=500)
+
+
+def get_song_info(video_id: str, oauth_file: Optional[str] = None) -> Dict[str, Any]:
+    try:
+        yt = _get_ytmusic_client(oauth_file)
+        song = yt.get_song(video_id)
+        
+        if not song:
+            raise YouTubeMusicError(f"Song not found: {video_id}", status_code=404)
+        
+        return song
+    except YouTubeMusicError:
+        raise
+    except Exception as e:
+        raise YouTubeMusicError(f"Failed to get song info: {str(e)}", status_code=500)
+
+
+def search_artists(query: str, limit: int = 10, oauth_file: Optional[str] = None) -> List[Dict[str, Any]]:
+    try:
+        yt = _get_ytmusic_client(oauth_file)
+        results = yt.search(query, filter="artists", limit=limit)
+        
+        artists = []
+        for item in results:
+            if item.get("resultType") == "artist":
+                thumbnails = item.get("thumbnails", [])
+                image_url = None
+                if thumbnails:
+                    image_url = thumbnails[-1].get("url") if isinstance(thumbnails[-1], dict) else None
+                
+                artists.append({
+                    "artist_id": item.get("browseId"),
+                    "name": item.get("artist", ""),
+                    "image_url": image_url,
+                })
+        
+        return artists[:limit]
+    except YouTubeMusicError:
+        raise
+    except Exception as e:
+        raise YouTubeMusicError(f"Artist search failed: {str(e)}", status_code=500)
+
+
+def get_artist_songs(artist_id: str, limit: int = 50, oauth_file: Optional[str] = None) -> List[Dict[str, Any]]:
+    try:
+        yt = _get_ytmusic_client(oauth_file)
+        artist = yt.get_artist(artist_id)
+        
+        if not artist:
+            raise YouTubeMusicError(f"Artist not found: {artist_id}", status_code=404)
+        
+        songs = []
+        if "songs" in artist:
+            for song in artist["songs"][:limit]:
+                video_id = song.get("videoId")
+                if not video_id:
+                    continue
+                
+                artists = []
+                if "artists" in song:
+                    for artist_info in song["artists"]:
+                        if isinstance(artist_info, dict):
+                            artists.append(artist_info.get("name", ""))
+                        elif isinstance(artist_info, str):
+                            artists.append(artist_info)
+                
+                thumbnails = song.get("thumbnails", [])
+                image_url = None
+                if thumbnails:
+                    image_url = thumbnails[-1].get("url") if isinstance(thumbnails[-1], dict) else None
+                
+                songs.append({
+                    "video_id": video_id,
+                    "title": song.get("title", ""),
+                    "artists": artists,
+                    "album": song.get("album", {}).get("name") if isinstance(song.get("album"), dict) else None,
+                    "duration": song.get("duration"),
+                    "image_url": image_url,
+                })
+        
+        return songs
+    except YouTubeMusicError:
+        raise
+    except Exception as e:
+        raise YouTubeMusicError(f"Failed to get artist songs: {str(e)}", status_code=500)
+
+
+def emotion_to_ytmusic_queries(emotion: str) -> List[str]:
+    emotion = emotion.lower()
+    
+    if emotion in {"joy", "happy", "happiness"}:
+        return [
+            "happy music",
+            "upbeat songs",
+            "feel good music",
+            "dance music",
+            "party songs",
+            "energetic music",
+            "positive vibes",
+        ]
+    elif emotion in {"sad", "sadness"}:
+        return [
+            "sad songs",
+            "emotional music",
+            "melancholic songs",
+            "ballads",
+            "heartbreak songs",
+            "calm music",
+            "relaxing music",
+        ]
+    elif emotion in {"anger", "angry"}:
+        return [
+            "angry music",
+            "rock music",
+            "metal songs",
+            "intense music",
+            "aggressive songs",
+            "punk rock",
+            "hard rock",
+        ]
+    elif emotion in {"fear"}:
+        return [
+            "calm music",
+            "ambient music",
+            "peaceful songs",
+            "meditation music",
+            "soothing music",
+            "relaxing instrumental",
+        ]
+    elif emotion in {"surprise"}:
+        return [
+            "surprising music",
+            "unexpected songs",
+            "experimental music",
+            "indie music",
+            "alternative music",
+            "unique songs",
+        ]
+    elif emotion in {"neutral"}:
+        return [
+            "chill music",
+            "background music",
+            "easy listening",
+            "soft music",
+            "ambient playlist",
+        ]
+    elif emotion in {"disgust"}:
+        return [
+            "alternative rock",
+            "indie music",
+            "experimental songs",
+            "unique music",
+        ]
+    
+    return ["music", "songs", "popular music"]
+
+
+def recommend_song_for_emotion(emotion: str, source: str = "text", oauth_file: Optional[str] = None) -> Dict[str, Any]:
+    emotion_norm = (emotion or "").lower()
+    queries = emotion_to_ytmusic_queries(emotion_norm)
+    
+    items: List[Dict[str, Any]] = []
+    for q in queries:
+        try:
+            songs = search_songs(q, limit=10, oauth_file=oauth_file)
+            if songs:
+                items.extend(songs)
+            if len(items) >= 20:
+                break
+        except Exception:
+            continue
+    
+    if not items:
+        try:
+            fallback = search_songs("popular music", limit=5, oauth_file=oauth_file)
+            if fallback:
+                items = fallback
+        except Exception:
+            pass
+    
+    if not items:
+        return {}
+    
+    song = random.choice(items)
+    return {
+        "video_id": song.get("video_id"),
+        "title": song.get("title", ""),
+        "artists": song.get("artists", []),
+        "album": song.get("album"),
+        "duration": song.get("duration"),
+        "image_url": song.get("image_url"),
+        "external_url": f"https://music.youtube.com/watch?v={song.get('video_id')}" if song.get("video_id") else None,
+    }