app.py

import gradio as gr
import requests
import json
import os
from datetime import datetime, timedelta
from huggingface_hub import InferenceClient  # LLM 사용을 위해 필요

# 환경 변수에서 API 키 가져오기
API_KEY = os.getenv("SERPHOUSE_API_KEY")  # 본인의 SerpHouse API 키를 환경 변수로 설정하세요.
HF_TOKEN = os.getenv("HF_TOKEN")  # Hugging Face API 토큰을 환경 변수로 설정하세요.

MAJOR_COUNTRIES = [
    "United States", "United Kingdom", "Canada", "Australia", "Germany",
    "France", "Japan", "South Korea", "China", "India",
    "Brazil", "Mexico", "Russia", "Italy", "Spain",
    "Netherlands", "Sweden", "Switzerland", "Norway", "Denmark",
    "Finland", "Belgium", "Austria", "New Zealand", "Ireland",
    "Singapore", "Hong Kong", "Israel", "United Arab Emirates", "Saudi Arabia",
    "South Africa", "Turkey", "Egypt", "Poland", "Czech Republic",
    "Hungary", "Greece", "Portugal", "Argentina", "Chile",
    "Colombia", "Peru", "Venezuela", "Thailand", "Malaysia",
    "Indonesia", "Philippines", "Vietnam", "Pakistan", "Bangladesh"
]

def search_serphouse(query, country, page=1, num_result=10):
    url = "https://api.serphouse.com/serp/live"

    now = datetime.utcnow()
    yesterday = now - timedelta(days=1)
    date_range = f"{yesterday.strftime('%Y-%m-%d')},{now.strftime('%Y-%m-%d')}"

    payload = {
        "data": {
            "q": query,
            "domain": "google.com",
            "loc": country,  # 국가 이름을 직접 사용합니다.
            "lang": "en",
            "device": "desktop",
            "serp_type": "news",
            "page": str(page),
            "verbatim": "1",
            "num": str(num_result),
            "date_range": date_range
        }
    }

    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "authorization": f"Bearer {API_KEY}"
    }

    try:
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        error_msg = f"Error: {str(e)}"
        if hasattr(response, 'text'):
            error_msg += f"\nResponse content: {response.text}"
        return {"error": error_msg}

def format_results_from_raw(results):
    try:
        if isinstance(results, dict) and "error" in results:
            return "Error: " + results["error"], []

        if not isinstance(results, dict):
            raise ValueError("결과가 사전 형식이 아닙니다.")

        # 'results' 키 내부의 구조 확인
        if 'results' in results:
            results_content = results['results']
            if 'results' in results_content:
                results_content = results_content['results']
                if 'news' in results_content:
                    news_results = results_content['news']
                else:
                    news_results = []
            else:
                news_results = []
        else:
            news_results = []

        if not news_results:
            return "검색 결과가 없습니다.", []

        articles = []

        for idx, result in enumerate(news_results, 1):
            title = result.get("title", "제목 없음")
            link = result.get("url", result.get("link", "#"))
            snippet = result.get("snippet", "내용 없음")
            channel = result.get("channel", result.get("source", "알 수 없음"))
            time = result.get("time", result.get("date", "알 수 없는 시간"))
            image_url = result.get("img", result.get("thumbnail", ""))

            articles.append({
                "index": idx,
                "title": title,
                "link": link,
                "snippet": snippet,
                "channel": channel,
                "time": time,
                "image_url": image_url
            })

        return "", articles

    except Exception as e:
        error_message = f"결과 처리 중 오류 발생: {str(e)}"
        return "Error: " + error_message, []

def serphouse_search(query, country):
    page = 1
    num_result = 10
    results = search_serphouse(query, country, page, num_result)
    error_message, articles = format_results_from_raw(results)
    return error_message, articles

# LLM 설정
hf_client = InferenceClient("CohereForAI/c4ai-command-r-plus-08-2024", token=HF_TOKEN)

def summarize_article(title, snippet):
    try:
        # 기사 제목과 스니펫을 기반으로 요약 생성
        prompt = f"다음 뉴스 제목과 요약을 바탕으로 한국어로 3문장으로 요약하세요:\n제목: {title}\n요약: {snippet}"
        summary = hf_client.text_generation(prompt, max_new_tokens=500)
        return summary
    except Exception as e:
        return f"요약 중 오류 발생: {str(e)}"

css = """
footer {
    visibility: hidden;
}
"""

# Gradio 인터페이스 구성
with gr.Blocks(theme="Nymbo/Nymbo_Theme", css=css, title="NewsAI 서비스") as iface:
    gr.Markdown("검색어를 입력하고 원하는 국가를 선택하면, 검색어와 일치하는 24시간 이내 뉴스를 최대 10개 출력합니다.")

    with gr.Column():
        with gr.Row():
            query = gr.Textbox(label="검색어")
            country = gr.Dropdown(MAJOR_COUNTRIES, label="국가", value="South Korea")
            search_button = gr.Button("검색")

        # 최대 10개의 기사에 대한 컴포넌트를 미리 생성합니다.
        article_components = []
        for i in range(10):
            with gr.Group(visible=False) as article_group:
                title = gr.Markdown()
                image = gr.Image(width=200, height=150)
                snippet = gr.Markdown()
                info = gr.Markdown()
                analyze_button = gr.Button("분석")
                summary_output = gr.Markdown(visible=False)

                article_components.append({
                    'group': article_group,
                    'title': title,
                    'image': image,
                    'snippet': snippet,
                    'info': info,
                    'analyze_button': analyze_button,
                    'summary_output': summary_output,
                })

    def search_and_display(query, country):
        error_message, articles = serphouse_search(query, country)
        outputs = []
        if error_message:
            outputs.append(gr.update(value=error_message, visible=True))
            # 나머지 컴포넌트 숨기기
            for comp in article_components:
                outputs.extend([
                    gr.update(visible=False),  # group
                    gr.update(),  # title
                    gr.update(),  # image
                    gr.update(),  # snippet
                    gr.update(),  # info
                    gr.update(),  # analyze_button
                    gr.update(visible=False),  # summary_output
                ])
            return outputs
        else:
            # 기사 컴포넌트 업데이트
            for idx, comp in enumerate(article_components):
                if idx < len(articles):
                    article = articles[idx]
                    comp['group'].visible = True
                    comp['title'].value = f"### [{article['title']}]({article['link']})"
                    if article['image_url'] and not article['image_url'].startswith("data:image"):
                        comp['image'].value = article['image_url']
                        comp['image'].visible = True
                    else:
                        comp['image'].visible = False
                    comp['snippet'].value = f"**요약:** {article['snippet']}"
                    comp['info'].value = f"**출처:** {article['channel']} | **시간:** {article['time']}"
                    comp['summary_output'].visible = False  # 초기에는 요약 숨김

                    # 분석 버튼 클릭 이벤트 정의
                    def create_analyze_function(article_title, article_snippet):
                        def analyze_article():
                            summary = summarize_article(article_title, article_snippet)
                            return gr.update(value=summary, visible=True)
                        return analyze_article

                    comp['analyze_button'].click(
                        create_analyze_function(article['title'], article['snippet']),
                        inputs=[],
                        outputs=comp['summary_output']
                    )

                    outputs.extend([
                        gr.update(visible=True),  # group
                        gr.update(),  # title
                        gr.update(),  # image
                        gr.update(),  # snippet
                        gr.update(),  # info
                        gr.update(),  # analyze_button
                        gr.update(visible=False),  # summary_output
                    ])
                else:
                    # 남은 컴포넌트 숨기기
                    comp['group'].visible = False
                    outputs.extend([
                        gr.update(visible=False),  # group
                        gr.update(),  # title
                        gr.update(),  # image
                        gr.update(),  # snippet
                        gr.update(),  # info
                        gr.update(),  # analyze_button
                        gr.update(visible=False),  # summary_output
                    ])
            return outputs

    # search_button 클릭 시 업데이트될 출력 컴포넌트 목록 생성
    search_outputs = []
    search_outputs.append(gr.Markdown(visible=False))  # 오류 메시지 출력용
    for comp in article_components:
        search_outputs.append(comp['group'])
        search_outputs.append(comp['title'])
        search_outputs.append(comp['image'])
        search_outputs.append(comp['snippet'])
        search_outputs.append(comp['info'])
        search_outputs.append(comp['analyze_button'])
        search_outputs.append(comp['summary_output'])

    search_button.click(
        search_and_display,
        inputs=[query, country],
        outputs=search_outputs
    )

iface.launch(auth=("gini", "pick"))