app.py

# app.py
# Naver News Search API -> Gradio chat-like UI (HF Spaces friendly)
#
# HF Spaces Secrets 설정:
#   NAVER_CLIENT_ID     = 발급받은 Client ID
#   NAVER_CLIENT_SECRET = 발급받은 Client Secret
#   ALLOWED_IPS         = 123.123.123.123,2406:xxxx:xxxx::1234

import os
import html
import re
from datetime import datetime
from typing import Dict, Any, List, Tuple

import requests
import gradio as gr


NAVER_NEWS_ENDPOINT = "https://openapi.naver.com/v1/search/news.json"


def _get_env(name: str) -> str:
    v = os.getenv(name, "").strip()
    return v


# ✅ 허용 IP 목록 로드
raw_ips = _get_env("ALLOWED_IPS")
ALLOWED_IPS = [ip.strip() for ip in raw_ips.split(",") if ip.strip()]


# ✅ 프록시 환경 고려: 실제 클라이언트 IP 추출
def _get_client_ip(request: gr.Request) -> str:
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()

    xri = request.headers.get("x-real-ip")
    if xri:
        return xri.strip()

    return request.client.host


# ✅ IP 접근 차단
def gatekeeper(request: gr.Request):
    if not ALLOWED_IPS:
        return gr.update(visible=True), ""

    client_ip = _get_client_ip(request)

    if client_ip not in ALLOWED_IPS:
        return (
            gr.update(visible=False),
            f"⛔ 접근이 허용되지 않은 IP입니다. (현재 IP: {client_ip})",
        )

    return gr.update(visible=True), ""


def _strip_tags(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"<[^>]+>", "", text)
    return html.unescape(text).strip()


def _format_pubdate(pub_date: str) -> str:
    if not pub_date:
        return ""
    try:
        dt = datetime.strptime(pub_date, "%a, %d %b %Y %H:%M:%S %z")
        return dt.strftime("%Y-%m-%d %H:%M:%S %z")
    except Exception:
        return pub_date


def naver_news_search(
    query: str,
    display: int = 10,
    sort: str = "sim",
    start: int = 1,
    timeout: int = 10,
) -> Dict[str, Any]:
    client_id = _get_env("NAVER_CLIENT_ID")
    client_secret = _get_env("NAVER_CLIENT_SECRET")

    if not client_id or not client_secret:
        raise RuntimeError(
            "환경변수 NAVER_CLIENT_ID / NAVER_CLIENT_SECRET 이 설정되어 있지 않습니다. "
            "Hugging Face Spaces의 Secrets에 추가해 주세요."
        )

    headers = {
        "X-Naver-Client-Id": client_id,
        "X-Naver-Client-Secret": client_secret,
    }

    params = {
        "query": query,
        "display": int(display),
        "start": int(start),
        "sort": sort,
    }

    r = requests.get(
        NAVER_NEWS_ENDPOINT,
        headers=headers,
        params=params,
        timeout=timeout,
    )

    if r.status_code != 200:
        try:
            detail = r.json()
        except Exception:
            detail = {"text": r.text}
        raise RuntimeError(f"네이버 API 호출 실패 (HTTP {r.status_code}): {detail}")

    return r.json()


def render_results(data: Dict[str, Any], max_items: int = 10) -> str:
    items = data.get("items", [])[:max_items]
    total = data.get("total", None)

    lines: List[str] = []
    if total is not None:
        lines.append(f"- 총 검색결과: {total:,}건")
    lines.append(f"- 반환 개수: {len(items)}건")
    lines.append("")

    for i, it in enumerate(items, start=1):
        title = _strip_tags(it.get("title", ""))
        desc = _strip_tags(it.get("description", ""))
        link = it.get("link", "")
        origin = it.get("originallink", "")
        pub = _format_pubdate(it.get("pubDate", ""))

        lines.append(f"{i}. **{title}**")

        if pub:
            lines.append(f"    - 발행: {pub}")
        if origin:
            lines.append(f"    - 원문: {origin}")
        if link:
            lines.append(f"    - 링크: {link}")
        if desc:
            lines.append(f"    - 요약: {desc}")

        lines.append("")
        lines.append("")

    return "\n".join(lines).strip()


def dedup_items(all_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    seen = set()
    out = []
    for it in all_items:
        origin = (it.get("originallink") or "").strip()
        link = (it.get("link") or "").strip()
        title = _strip_tags(it.get("title", ""))
        pub = it.get("pubDate", "")

        key = origin or link or f"{title}|{pub}"
        if not key:
            continue
        if key in seen:
            continue
        seen.add(key)
        out.append(it)
    return out


def aggregate_search(
    sentence: str,
    display: int,
    sort: str,
) -> Tuple[List[str], List[Dict[str, Any]], int]:

    queries = [sentence]

    all_items: List[Dict[str, Any]] = []
    total: int = 0

    data = naver_news_search(query=sentence, display=int(display), sort=sort, start=1)
    total = int(data.get("total", 0) or 0)
    all_items.extend(data.get("items", []))

    merged = dedup_items(all_items)
    final_items = merged[:display]
    return queries, final_items, total


def render_results_from_items(items: List[Dict[str, Any]]) -> str:
    lines: List[str] = []
    lines.append("")

    for i, it in enumerate(items, start=1):
        title = _strip_tags(it.get("title", ""))
        desc = _strip_tags(it.get("description", ""))
        link = it.get("link", "")
        origin = it.get("originallink", "")
        pub = _format_pubdate(it.get("pubDate", ""))

        lines.append(f"{i}. **{title}**")

        if pub:
            lines.append(f"    - 발행: {pub}")
        if origin:
            lines.append(f"    - 원문: {origin}")
        if link:
            lines.append(f"    - 링크: {link}")
        if desc:
            lines.append(f"    - 요약: {desc}")

        lines.append("")

    return "\n".join(lines).strip()


def handle_search(
    user_query: str,
    chat_history: List[Dict[str, str]],
    display: int,
    sort: str,
) -> Tuple[List[Dict[str, str]], str]:
    q = (user_query or "").strip()
    if not q:
        return chat_history, ""

    chat_history = chat_history + [{"role": "user", "content": q}]

    try:
        _, items, total = aggregate_search(sentence=q, display=int(display), sort=sort)

        lines: List[str] = []
        lines.append("")
        lines.append(render_results_from_items(items))
        lines.append("")

        assistant_text = "\n".join(lines).strip()

    except Exception as e:
        assistant_text = f"오류가 발생했습니다.\n\n- {e}"

    chat_history = chat_history + [{"role": "assistant", "content": assistant_text}]
    return chat_history, ""


with gr.Blocks(title="Naver News Search (Chat UI)") as demo:

    warning = gr.Markdown("")
    main_ui = gr.Column(visible=True)

    # ✅ 페이지 로드시 IP 검사
    demo.load(
        fn=gatekeeper,
        inputs=None,
        outputs=[main_ui, warning],
    )

    with main_ui:
        with gr.Accordion("검색 옵셥 - 뉴스 개수와 정렬 방식을 선택하세요.", open=False):
            with gr.Row():
                display = gr.Slider(
                    minimum=1, maximum=100, value=10, step=1, label="뉴스 개수"
                )
                sort = gr.Dropdown(
                    choices=[("최신순", "date"), ("정확도순(연관도순)", "sim")],
                    value="sim",
                    label="정렬 방식",
                )

        chatbot = gr.Chatbot(
            value=[],
            label="NewsChat_v0.1",
            type="messages",
            height=600,
        )

        with gr.Row():
            query_in = gr.Textbox(
                label="검색어를 입력하세요.",
                placeholder="(예: 행안부의 인공지능 사업)",
                scale=8,
            )
            with gr.Column(scale=2):
                search_btn = gr.Button("검색", variant="primary")
                clear_btn = gr.Button("대화 지우기", variant="secondary")

        search_btn.click(
            fn=handle_search,
            inputs=[query_in, chatbot, display, sort],
            outputs=[chatbot, query_in],
            api_name="search",
        )
        query_in.submit(
            fn=handle_search,
            inputs=[query_in, chatbot, display, sort],
            outputs=[chatbot, query_in],
            api_name="search_submit",
        )

        def clear_chat():
            return []

        clear_btn.click(fn=clear_chat, inputs=[], outputs=[chatbot], api_name="clear")


if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)