Upload 2 files

app.py

@@ -0,0 +1,276 @@
+# app.py
+# ============================================
+# Netflix (KR) Recommender + Review Analyzer — Live TMDb with Posters
+# - Uses TMDb API (env var: TMDB_API_KEY), with optional UI override
+# - Gradio app suitable for Hugging Face Spaces (CPU-friendly)
+# ============================================
+
+import os
+import time
+import requests
+import traceback
+from typing import Dict, Any, List, Tuple
+
+import numpy as np
+import gradio as gr
+
+# Optional NLP models
+from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM
+from sentence_transformers import SentenceTransformer
+
+# -----------------------------
+# Config
+# -----------------------------
+TMDB_BASE = "https://api.themoviedb.org/3"
+TMDB_IMG_BASE = "https://image.tmdb.org/t/p/w500"  # w500 is a good balance for gallery
+DEFAULT_REGION = "KR"
+
+# Load lightweight NLP models (CPU)
+def _load_models():
+    # Sentiment (multilingual 1~5 stars)
+    sent = pipeline("sentiment-analysis", model="nlptown/bert-base-multilingual-uncased-sentiment", device=-1)
+    # T5 small for Korean one-liners
+    tok = AutoTokenizer.from_pretrained("google/flan-t5-small")
+    mdl = AutoModelForSeq2SeqLM.from_pretrained("google/flan-t5-small")
+    summer = pipeline("text2text-generation", model=mdl, tokenizer=tok, device=-1)
+    # Embedding model for semantic ranking (multilingual)
+    try:
+        emb = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
+    except Exception:
+        emb = None
+    return sent, summer, emb
+
+_sent, _summer, _emb = _load_models()
+
+
+# -----------------------------
+# TMDb helpers
+# -----------------------------
+def tmdb_get(api_key: str, path: str, params: Dict[str, Any]) -> Dict[str, Any]:
+    """GET with simple retry/backoff"""
+    url = f"{TMDB_BASE}{path}"
+    p = {"api_key": api_key, **params}
+    last_err = None
+    for attempt in range(3):
+        try:
+            r = requests.get(url, params=p, timeout=25)
+            if r.status_code == 200:
+                return r.json()
+            last_err = f"{r.status_code} {r.text[:200]}"
+        except Exception as e:
+            last_err = str(e)
+        time.sleep(0.7 * (attempt + 1))
+    raise RuntimeError(f"TMDb request failed: {last_err}")
+
+def get_provider_id(api_key: str, region: str, provider_name="Netflix") -> int:
+    """Fetch provider list for region; return provider_id for Netflix (fallback 8)."""
+    data = tmdb_get(api_key, "/watch/providers/movie", {"watch_region": region})
+    for item in data.get("results", []):
+        if str(item.get("provider_name","")).lower() == provider_name.lower():
+            return int(item["provider_id"])
+    return 8  # common fallback
+
+def discover_quick(api_key: str, region: str, nfx_id: int, ctype="movie",
+                   sort_by="popularity.desc", page_limit=2) -> List[Dict[str, Any]]:
+    """
+    Use TMDb Discover with Netflix provider filter.
+    """
+    params = {
+        "watch_region": region,
+        "with_watch_providers": nfx_id,
+        "sort_by": sort_by,
+        "include_adult": False,
+        "language": "ko-KR"
+    }
+    rows = []
+    for page in range(1, page_limit+1):
+        data = tmdb_get(api_key, f"/discover/{ctype}", {**params, "page": page})
+        rows.extend([{"type": ctype, **r} for r in data.get("results", [])])
+    return rows
+
+def has_netflix_offer(api_key: str, content_type: str, tmdb_id: int, region: str, nfx_id: int) -> bool:
+    """Check if a specific item is offered on Netflix in the region."""
+    data = tmdb_get(api_key, f"/{content_type}/{tmdb_id}/watch/providers", {})
+    results = data.get("results", {})
+    info = results.get(region, {})
+    provs = info.get("flatrate", []) + info.get("ads", []) + info.get("free", [])
+    return any(int(p.get("provider_id", -1)) == nfx_id for p in provs)
+
+def search_and_filter(api_key: str, query: str, region: str, nfx_id: int,
+                      content_types=("movie","tv"), max_pages_each=2, max_total=60) -> List[Dict[str,Any]]:
+    """
+    1) Search movie/tv by query
+    2) Validate Netflix provider for each
+    """
+    out = []
+    for ctype in content_types:
+        for page in range(1, max_pages_each+1):
+            data = tmdb_get(api_key, f"/search/{ctype}", {
+                "query": query, "page": page, "include_adult": False, "language": "ko-KR"
+            })
+            for item in data.get("results", []):
+                tmdb_id = item["id"]
+                try:
+                    if has_netflix_offer(api_key, ctype, tmdb_id, region, nfx_id):
+                        out.append({"type": ctype, **item})
+                except Exception:
+                    pass
+                if len(out) >= max_total:
+                    break
+            if len(out) >= max_total:
+                break
+    return out
+
+
+# -----------------------------
+# Ranking & formatting
+# -----------------------------
+def _embed_texts(texts: List[str]) -> np.ndarray:
+    if _emb is None or not texts:
+        return np.zeros((len(texts), 384), dtype=np.float32)
+    X = _emb.encode(texts, normalize_embeddings=True, convert_to_numpy=True, show_progress_bar=False)
+    return X
+
+def rank_by_query(items: List[Dict[str, Any]], query: str, topk: int = 10) -> List[Dict[str, Any]]:
+    if not items:
+        return []
+    if not query or not query.strip() or _emb is None:
+        return items[:topk]
+    texts = []
+    for it in items:
+        title = it.get("name") or it.get("title") or ""
+        overview = it.get("overview") or ""
+        texts.append(f"{title}. {overview}")
+    q = _emb.encode([query], normalize_embeddings=True, convert_to_numpy=True)[0].reshape(1, -1)
+    X = _emb.encode(texts, normalize_embeddings=True, convert_to_numpy=True)
+    sims = (q @ X.T)[0]
+    idx = np.argsort(-sims)[:topk]
+    return [items[i] for i in idx]
+
+def build_gallery(items: List[Dict[str, Any]]) -> Tuple[list, list]:
+    """
+    Return (gallery_items, table_rows). Gallery expects list of [image, caption]
+    """
+    gallery = []
+    rows = []
+    for it in items:
+        title = it.get("name") or it.get("title") or ""
+        overview = it.get("overview") or ""
+        date = it.get("first_air_date") or it.get("release_date") or ""
+        vote = it.get("vote_average")
+        ctype = "드라마" if it.get("type") == "tv" else "영화"
+        poster = it.get("poster_path")
+        img = f"{TMDB_IMG_BASE}{poster}" if poster else None
+        cap = f"{title} ({ctype})\n평점: {vote} | 공개: {date}\n{overview[:120]}{'...' if len(overview)>120 else ''}"
+        gallery.append([img, cap])
+        rows.append({"제목": title, "유형": ctype, "공개일": date, "TMDb평점": vote, "개요": overview})
+    return gallery, rows
+
+
+# -----------------------------
+# Business logic (callbacks)
+# -----------------------------
+STAR_MAP = {1:"매우 부정", 2:"부정", 3:"중립", 4:"긍정", 5:"매우 긍정"}
+
+def do_recommend(api_key_ui: str, query: str, region: str, mode: str, topk: int,
+                 sort_by: str, include_movie: bool, include_tv: bool):
+    try:
+        api_key = (api_key_ui or "").strip() or os.environ.get("TMDB_API_KEY", "").strip()
+        if not api_key:
+            return "TMDb API Key를 입력하거나 환경변수 TMDB_API_KEY를 설정하세요.", None, None
+        nfx_id = get_provider_id(api_key, region, "Netflix")
+
+        types = []
+        if include_movie: types.append("movie")
+        if include_tv: types.append("tv")
+        if not types:
+            types = ["movie", "tv"]
+
+        # Fetch
+        if mode == "빠른 추천(Discover)":
+            items = []
+            for t in types:
+                items.extend(discover_quick(api_key, region, nfx_id, ctype=t, sort_by=sort_by, page_limit=2))
+        else:
+            items = search_and_filter(api_key, query or "Netflix", region, nfx_id,
+                                      content_types=tuple(types), max_pages_each=2, max_total=80)
+
+        if not items:
+            return f"조건에 맞는 넷플릭스({region}) 작품을 찾지 못했습니다.", None, None
+
+        ranked = rank_by_query(items, query, topk=topk)
+        gallery, rows = build_gallery(ranked)
+        # One-line pitch for top1
+        t = ranked[0]
+        top_title = (t.get("name") or t.get("title") or "")
+        pitch_prompt = (
+            "Summarize in Korean (1-2 sentences):\n"
+            f"사용자 취향/키워드: {query}\n"
+            f"작품: {top_title} / 개요: {t.get('overview','')}"
+        )
+        pitch = _summer(pitch_prompt, max_new_tokens=80, do_sample=False)[0]["generated_text"]
+        md = f"### ✅ 추천 결과 (Region={region}, Provider=Netflix)\n- Top 1: **{top_title}** — {pitch}"
+        return md, gallery, rows
+    except Exception as e:
+        return f"[오류] {e}\n{traceback.format_exc()}", None, None
+
+def analyze_review(title: str, review: str):
+    try:
+        if not review or not review.strip():
+            return "감상평을 입력해 주세요.", ""
+        res = _sent(review)[0]
+        stars = int(res["label"][0])
+        head = f"예측 별점: {stars} ({STAR_MAP.get(stars,'중립')}) / 확신도: {float(res['score']):.3f}"
+        summ = _summer(
+            f"Summarize in Korean (1 sentence):\n제목: {title}\n감상평: {review}",
+            max_new_tokens=60, do_sample=False
+        )[0]["generated_text"]
+        return head, f"한줄평: {summ}"
+    except Exception as e:
+        return f"[오류] {e}\n{traceback.format_exc()}", ""
+
+
+# -----------------------------
+# Gradio UI
+# -----------------------------
+with gr.Blocks() as demo:
+    gr.Markdown("## 🍿 실시간 넷플릭스(KR) 추천 & 감상평 — TMDb API + 포스터 이미지")
+
+    with gr.Accordion("TMDb API 설정", open=True):
+        api_key = gr.Textbox(label="TMDb API Key (UI 입력은 선택, 기본은 환경변수 TMDB_API_KEY 사용)", type="password")
+        region = gr.Dropdown(choices=["KR","US","JP","GB","DE","FR","ES"], value=DEFAULT_REGION, label="지역(Watch Region)")
+
+    with gr.Tab("추천"):
+        query = gr.Textbox(label="키워드/기분(선택)", placeholder="예) 따뜻한 성장 드라마, 무서운 한국 스릴러", lines=2)
+        with gr.Row():
+            mode = gr.Radio(choices=["빠른 추천(Discover)", "키워드 검색(정확)"], value="빠른 추천(Discover)", label="검색 모드")
+            sort_by = gr.Dropdown(choices=["popularity.desc","vote_average.desc","release_date.desc"], value="popularity.desc", label="정렬(Discover용)")
+            topk = gr.Slider(3, 20, value=9, step=1, label="표시 개수")
+        with gr.Row():
+            include_movie = gr.Checkbox(value=True, label="영화 포함")
+            include_tv = gr.Checkbox(value=True, label="드라마 포함")
+        btn = gr.Button("추천 받기")
+
+        out_md = gr.Markdown()
+        out_gallery = gr.Gallery(label="포스터 갤러리", columns=3, height="auto", allow_preview=True)
+        out_table = gr.Dataframe(interactive=False, wrap=True)
+
+        btn.click(
+            do_recommend,
+            inputs=[api_key, query, region, mode, topk, sort_by, include_movie, include_tv],
+            outputs=[out_md, out_gallery, out_table]
+        )
+
+    with gr.Tab("감상평 분석"):
+        title = gr.Textbox(label="제목(선택)", placeholder="추천 탭에서 복사해 붙여넣기")
+        review = gr.Textbox(label="감상평", lines=5, placeholder="예) 초반은 늘어지지만, 배우 연기가 압권이에요.")
+        b2 = gr.Button("분석")
+        head = gr.Markdown()
+        summ = gr.Markdown()
+        b2.click(analyze_review, inputs=[title, review], outputs=[head, summ])
+
+# Expose demo for Spaces
+app = demo
+
+if __name__ == "__main__":
+    demo.launch(share=True, debug=True)

requirements.txt

@@ -0,0 +1,6 @@
+gradio>=4.40.0
+transformers>=4.43.3
+sentence-transformers>=3.0.1
+torch
+requests
+numpy