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

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


def _strip_tags(text: str) -> str:
    # 네이버 뉴스 API 결과의 title/description은 <b>...</b>가 포함될 수 있어 제거
    # (HTML 태그 제거 + 엔티티 처리)
    if not text:
        return ""
    text = re.sub(r"<[^>]+>", "", text)
    return html.unescape(text).strip()


def _format_pubdate(pub_date: str) -> str:
    # 예: "Mon, 19 Jan 2026 09:30:00 +0900"
    # 파싱 실패 시 원문 반환
    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 = "date",
    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,
    )

    # 네이버 API가 에러를 반환하는 경우, body에 상세가 있을 수 있어 그대로 보여주기 위해 처리
    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}**")

        # 하위 항목 4칸 들여쓰기
        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]]:
    """
    결과 중복 제거:
    - originallink 우선, 없으면 link, 없으면 title+pubDate 해시
    """
    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]:
    """
    사용자 입력 문장을 그대로 query로 사용하여 API 호출
    반환: (사용된 쿼리 목록, 최종 아이템 목록, total)
    """
    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:
    """
    items 리스트를 동일 스타일로 출력
    """
    lines: List[str] = []
    # 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("")

    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)

        total_to_show = total if total > 0 else len(items)

        lines: List[str] = []
        # lines.append(
        #     f"\"{q}\"에 대한 검색 결과는 **{total_to_show:,}건** 이며, 최종 응답 개수는 **{len(items)}건** 입니다."
        # )
        # lines.append("---")
        # lines.append("")
        # lines.append("---")  # ✅ 수평선은 “... 입니다” 다음 줄에만
        lines.append("")
        lines.append(render_results_from_items(items))  # ✅ 결과는 1번만 출력
        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:          
    with gr.Accordion("검색 옵셥 - 뉴스 개수와 정렬 방식을 선택하세요.", open=False):
        with gr.Row():
            display = gr.Slider(
                minimum=1, maximum=100, value=30, step=1, label="뉴스 개수"
            )
            sort = gr.Dropdown(
                choices=[("최신순", "date"), ("정확도순(연관도순)", "sim")],
                value="date",
                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)