app.py

"""MusicMood — analiza nastroju z kamery + rekomendacja muzyczna z YouTube."""

import asyncio
import hmac
import ipaddress
import json
import os
import re
from typing import Optional
from urllib.parse import quote

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI(title="MusicMood")

WEATHER_CODES = {
    0: "bezchmurnie", 1: "przeważnie bezchmurnie", 2: "częściowe zachmurzenie",
    3: "pochmurno", 45: "mgła", 48: "mgła osadzająca szadź",
    51: "lekka mżawka", 53: "mżawka", 55: "gęsta mżawka",
    61: "lekki deszcz", 63: "deszcz", 65: "ulewny deszcz",
    66: "marznący deszcz", 67: "silny marznący deszcz",
    71: "lekki śnieg", 73: "śnieg", 75: "intensywny śnieg", 77: "ziarna śniegu",
    80: "przelotne opady", 81: "przelotny deszcz", 82: "gwałtowne przelotne opady",
    85: "przelotny śnieg", 86: "intensywny przelotny śnieg",
    95: "burza", 96: "burza z gradem", 99: "silna burza z gradem",
}


APP_PASSWORD = os.environ.get("APP_PASSWORD", "heuristica-music-mood")


class AnalyzeRequest(BaseModel):
    password: str
    image_b64: str          # czysty base64 JPEG (bez prefiksu data:)
    local_time: str         # np. "2026-06-11 14:32"
    day_of_week: str        # np. "czwartek"
    timezone: Optional[str] = None
    lat: Optional[float] = None
    lon: Optional[float] = None


def _is_public_ip(ip: str) -> bool:
    try:
        return ipaddress.ip_address(ip).is_global
    except ValueError:
        return False


async def geolocate_by_ip(request: Request) -> Optional[dict]:
    """Fallback: lokalizacja po IP, gdy przeglądarka nie udostępniła geolokalizacji.

    Na HF Space prawdziwe IP klienta jest w nagłówku x-forwarded-for; lokalnie
    klient to 127.0.0.1, więc pytamy o publiczne IP serwera (= sieć użytkownika).
    """
    fwd = request.headers.get("x-forwarded-for", "")
    client_ip = fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else "")
    url = f"https://ipapi.co/{client_ip}/json/" if _is_public_ip(client_ip) else "https://ipapi.co/json/"
    try:
        async with httpx.AsyncClient(timeout=8) as client:
            r = await client.get(url, headers={"User-Agent": "MusicMood/1.0"})
            r.raise_for_status()
            data = r.json()
            if data.get("latitude") is None:
                return None
            return {
                "lat": data["latitude"],
                "lon": data["longitude"],
                "miejscowosc": data.get("city"),
            }
    except Exception:
        return None


YT_ID_RE = re.compile(r'"videoId"\s*:\s*"([\w-]{11})"')

YT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/126.0 Safari/537.36",
    "Accept-Language": "en-US,en;q=0.9",
    "Cookie": "SOCS=CAI; CONSENT=YES+1",
}

# InnerTube — wewnętrzne API JSON, którego używa strona youtube.com;
# znacznie odporniejsze na blokady antybotowe niż scraping HTML
INNERTUBE_CONTEXT = {
    "context": {"client": {"clientName": "WEB", "clientVersion": "2.20240701.00.00"}}
}

# publiczne proxy YouTube — używane, gdy youtube.com jest niedostępny
# (Google ucina połączenia z IP datacenter, m.in. z Hugging Face)
PIPED_INSTANCES = [
    "https://api.piped.private.coffee",
    "https://pipedapi.adminforge.de",
    "https://pipedapi.kavin.rocks",
]

YT_TIMEOUT = httpx.Timeout(8, connect=4)


def _log(msg: str) -> None:
    print(f"[musicmood] {msg}", flush=True)


def _dedupe_ids(text: str, limit: int) -> list[str]:
    seen: list[str] = []
    for vid in YT_ID_RE.findall(text):
        if vid not in seen:
            seen.append(vid)
        if len(seen) >= limit:
            break
    return seen


async def _search_data_api(client: httpx.AsyncClient, query: str, limit: int) -> list[str]:
    """Oficjalne YouTube Data API v3 — wymaga klucza YOUTUBE_API_KEY (opcjonalny).

    Najpewniejsze źródło: googleapis.com działa też z IP datacenter, a filtr
    videoEmbeddable=true od razu zwraca tylko filmy osadzalne.
    """
    key = os.environ.get("YOUTUBE_API_KEY", "")
    if not key:
        return []
    r = await client.get(
        "https://www.googleapis.com/youtube/v3/search",
        params={"part": "id", "q": query, "type": "video",
                "videoEmbeddable": "true", "maxResults": limit, "key": key},
    )
    _log(f"yt-search data-api '{query}': HTTP {r.status_code}")
    r.raise_for_status()
    return [it["id"]["videoId"] for it in r.json().get("items", []) if it.get("id", {}).get("videoId")]


async def _search_innertube(client: httpx.AsyncClient, query: str, limit: int) -> list[str]:
    r = await client.post(
        "https://www.youtube.com/youtubei/v1/search?prettyPrint=false",
        json={**INNERTUBE_CONTEXT, "query": query, "params": "EgIQAQ=="},  # filtr: tylko filmy
        headers=YT_HEADERS,
    )
    _log(f"yt-search innertube '{query}': HTTP {r.status_code}")
    r.raise_for_status()
    return _dedupe_ids(r.text, limit)


async def _search_piped(client: httpx.AsyncClient, query: str, limit: int) -> list[str]:
    for base in PIPED_INSTANCES:
        try:
            r = await client.get(f"{base}/search", params={"q": query, "filter": "videos"})
            _log(f"yt-search piped {base}: HTTP {r.status_code}")
            if r.status_code != 200:
                continue
            ids: list[str] = []
            for item in r.json().get("items", []):
                u = item.get("url", "")
                if "watch?v=" in u:
                    vid = u.split("watch?v=")[1][:11]
                    if vid not in ids:
                        ids.append(vid)
                if len(ids) >= limit:
                    break
            if ids:
                return ids
        except Exception as e:
            _log(f"yt-search piped {base} error: {e!r}")
    return []


async def _search_html(client: httpx.AsyncClient, query: str, limit: int) -> list[str]:
    url = "https://www.youtube.com/results?search_query=" + quote(query)
    r = await client.get(url, headers=YT_HEADERS)
    _log(f"yt-search html '{query}': HTTP {r.status_code}")
    r.raise_for_status()
    return _dedupe_ids(r.text, limit)


async def youtube_search_ids(query: str, limit: int = 8) -> tuple[list[str], bool]:
    """Szuka filmów kaskadą źródeł. Zwraca (ids, czy_juz_przefiltrowane)."""
    async with httpx.AsyncClient(timeout=YT_TIMEOUT, follow_redirects=True) as client:
        for fn, prefiltered in ((_search_data_api, True), (_search_innertube, False),
                                (_search_piped, False), (_search_html, False)):
            try:
                ids = await fn(client, query, limit)
                if ids:
                    return ids, prefiltered
            except Exception as e:
                _log(f"yt-search {fn.__name__} error: {e!r}")
    return [], False


async def filter_embeddable(ids: list[str], limit: int = 4) -> list[str]:
    """Zostawia tylko filmy, które właściciel pozwala osadzać na innych stronach.

    oEmbed zwraca 200 dla filmów osadzalnych, a 401/403 dla zablokowanych.
    """
    async def check(client: httpx.AsyncClient, vid: str) -> Optional[str]:
        try:
            r = await client.get(
                "https://www.youtube.com/oembed"
                f"?url=https://www.youtube.com/watch?v={vid}&format=json",
                headers=YT_HEADERS,
            )
            return vid if r.status_code == 200 else None
        except Exception as e:
            _log(f"yt-oembed {vid} error: {e!r}")
            return None

    async with httpx.AsyncClient(timeout=YT_TIMEOUT, follow_redirects=True) as client:
        results = await asyncio.gather(*(check(client, v) for v in ids))
    found = [v for v in results if v][:limit]
    _log(f"embeddable: {len(found)}/{len(ids)} -> {found}")
    return found


async def _weather_open_meteo(lat: float, lon: float) -> Optional[dict]:
    url = (
        "https://api.open-meteo.com/v1/forecast"
        f"?latitude={lat}&longitude={lon}"
        "&current=temperature_2m,relative_humidity_2m,surface_pressure,"
        "weather_code,wind_speed_10m,cloud_cover"
    )
    try:
        async with httpx.AsyncClient(timeout=8) as client:
            r = await client.get(url)
            r.raise_for_status()
            cur = r.json().get("current", {})
            return {
                "temperatura_c": cur.get("temperature_2m"),
                "cisnienie_hpa": cur.get("surface_pressure"),
                "wilgotnosc_pct": cur.get("relative_humidity_2m"),
                "wiatr_kmh": cur.get("wind_speed_10m"),
                "zachmurzenie_pct": cur.get("cloud_cover"),
                "opis": WEATHER_CODES.get(cur.get("weather_code"), "nieznana"),
            }
    except Exception:
        return None


async def _weather_wttr(lat: float, lon: float) -> Optional[dict]:
    url = f"https://wttr.in/{lat},{lon}?format=j1&lang=pl"
    try:
        async with httpx.AsyncClient(timeout=12) as client:
            r = await client.get(url, headers={"User-Agent": "MusicMood/1.0"})
            r.raise_for_status()
            cur = r.json()["current_condition"][0]
            desc = (cur.get("lang_pl") or cur.get("weatherDesc") or [{}])[0].get("value", "nieznana")
            return {
                "temperatura_c": float(cur["temp_C"]),
                "cisnienie_hpa": float(cur["pressure"]),
                "wilgotnosc_pct": float(cur["humidity"]),
                "wiatr_kmh": float(cur["windspeedKmph"]),
                "zachmurzenie_pct": float(cur["cloudcover"]),
                "opis": desc,
            }
    except Exception:
        return None


async def fetch_weather(lat: float, lon: float) -> Optional[dict]:
    """Aktualna pogoda bez klucza API: Open-Meteo, a gdy niedostępne — wttr.in."""
    return await _weather_open_meteo(lat, lon) or await _weather_wttr(lat, lon)


SYSTEM_PROMPT = """\
Jesteś ekspertem-muzykoterapeutą i analitykiem obrazu. Twoje zadanie:

1. Przeanalizuj zdjęcie twarzy osoby: określ prawdopodobną płeć, przybliżony
   wiek, nastrój oraz emocje, które prawdopodobnie wynikają z mimiki, postawy
   i wyglądu na zdjęciu (zmęczenie, radość, napięcie, spokój itp.).
2. Uwzględnij podany kontekst: dzień tygodnia, lokalny czas i pogodę
   (ciśnienie atmosferyczne wpływa na samopoczucie, pora dnia na energię,
   pogoda na nastrój).
3. Na tej podstawie zarekomenduj muzykę KOMPLEMENTARNĄ do nastroju osoby —
   czyli taką, która najlepiej współgra z jej aktualnym stanem emocjonalnym
   i kontekstem (nie przypadkową): styl/gatunek, tempo (BPM i opisowo),
   rytmikę, charakter brzmienia.
4. Zaproponuj konkretną frazę wyszukiwania na YouTube, która znajdzie
   pełny utwór muzyczny pasujący do rekomendacji (np. tytuł i wykonawca
   znanego utworu albo precyzyjna fraza gatunkowa). Fraza po angielsku,
   chyba że pasuje muzyka polska.

Odpowiedz WYŁĄCZNIE poprawnym JSON-em o strukturze:
{
  "osoba": {
    "plec": "...",
    "wiek": "np. 30-40 lat",
    "nastroj": "...",
    "emocje": ["...", "..."]
  },
  "rekomendacja": {
    "styl": "...",
    "tempo": "np. ~90 BPM, umiarkowane",
    "rytmika": "...",
    "charakter": "...",
    "uzasadnienie": "1-2 zdania po polsku"
  },
  "youtube_query": "..."
}

Wszystkie opisy po polsku. Jeśli na zdjęciu nie widać twarzy, ustaw pole
"osoba" na null i dobierz muzykę wyłącznie na podstawie kontekstu, a w polu
"uzasadnienie" zaznacz, że twarz nie była widoczna.
"""


@app.post("/api/analyze")
async def analyze(req: AnalyzeRequest, request: Request):
    if not hmac.compare_digest(req.password.strip(), APP_PASSWORD):
        raise HTTPException(401, "Nieprawidłowe hasło dostępu.")

    openai_key = os.environ.get("OPENAI_API_KEY", "")
    if not openai_key:
        raise HTTPException(500, "Brak klucza OPENAI_API_KEY w konfiguracji serwera (HF Secrets).")

    # lokalizacja: najpierw geolokalizacja z przeglądarki, potem fallback po IP
    lat, lon, city = req.lat, req.lon, None
    if lat is None or lon is None:
        ip_geo = await geolocate_by_ip(request)
        if ip_geo:
            lat, lon, city = ip_geo["lat"], ip_geo["lon"], ip_geo.get("miejscowosc")

    weather = None
    if lat is not None and lon is not None:
        weather = await fetch_weather(lat, lon)
        if weather and city:
            weather["miejscowosc"] = city

    context = {
        "dzien_tygodnia": req.day_of_week,
        "czas_lokalny": req.local_time,
        "strefa_czasowa": req.timezone or "nieznana",
        "pogoda": weather or "brak danych o lokalizacji",
    }

    client = OpenAI(api_key=openai_key)
    try:
        resp = client.chat.completions.create(
            model="gpt-4o",
            max_tokens=900,
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": "Kontekst:\n" + json.dumps(context, ensure_ascii=False, indent=2),
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{req.image_b64}",
                                "detail": "low",
                            },
                        },
                    ],
                },
            ],
        )
    except Exception as e:
        msg = str(e)
        if "401" in msg or "invalid_api_key" in msg:
            raise HTTPException(401, "Nieprawidłowy klucz OpenAI API.")
        raise HTTPException(502, f"Błąd OpenAI API: {msg}")

    try:
        result = json.loads(resp.choices[0].message.content)
    except (json.JSONDecodeError, IndexError):
        raise HTTPException(502, "Model zwrócił niepoprawną odpowiedź — spróbuj ponownie.")

    result["kontekst"] = context

    # wyszukaj konkretne filmy — iframe z listType=search jest zawodny
    query = result.get("youtube_query") or ""
    candidates, prefiltered = await youtube_search_ids(query) if query else ([], False)
    if candidates and not prefiltered:
        embeddable = await filter_embeddable(candidates)
        # gdy filtr nic nie przepuści (np. oEmbed niedostępny), zwróć surowe wyniki —
        # player i tak sam pominie nieosadzalne filmy
        result["video_ids"] = embeddable or candidates[:5]
    else:
        result["video_ids"] = candidates
    _log(f"query='{query}' kandydatow={len(candidates)} zwracam={result['video_ids']}")
    return result


@app.get("/")
async def index():
    return FileResponse("static/index.html")


app.mount("/static", StaticFiles(directory="static"), name="static")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=7860)