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")
HF_TOKEN = os.getenv("HF_TOKEN")

# 국가 이름과 위치 ID를 매핑
COUNTRY_CODE_MAPPING = {
    "United States": "2840",
    "South Korea": "2458",
    # 다른 국가들의 위치 ID를 추가하세요
}

MAJOR_COUNTRIES = list(COUNTRY_CODE_MAPPING.keys())

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')}"

    # 국가 이름을 위치 ID로 변환
    loc_id = COUNTRY_CODE_MAPPING.get(country, "2840")  # 기본값은 미국

    payload = {
        "data": {
            "q": query,
            "domain": "google.com",
            "loc": loc_id,  # 위치 ID 사용
            "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(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("검색")

        # 상태 메시지 컴포넌트 제거 (Progress를 사용하므로 불필요)
        # status_message = gr.Markdown(visible=False)

        # 기사 데이터를 저장할 상태 변수
        articles_state = gr.State([])  # 초기값을 빈 리스트로 설정

        # 최대 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,
                    'index': i,
                })

    def search_and_display(query, country, articles_state):
        with gr.Progress() as progress:
            progress(0, desc="처리중입니다. 잠시만 기다리세요.")

            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(visible=False),  # summary_output
                    ])
                articles_state = []
            else:
                outputs.append(gr.update(value="", visible=False))
                for idx, comp in enumerate(article_components):
                    if idx < len(articles):
                        article = articles[idx]
                        # 이미지 처리 수정
                        image_url = article['image_url']
                        if image_url and not image_url.startswith('data:image'):
                            image_update = gr.update(value=image_url, visible=True)
                        else:
                            image_update = gr.update(value=None, visible=False)

                        outputs.extend([
                            gr.update(visible=True),  # group
                            gr.update(value=f"### [{article['title']}]({article['link']})"),  # title
                            image_update,  # image
                            gr.update(value=f"**요약:** {article['snippet']}"),  # snippet
                            gr.update(value=f"**출처:** {article['channel']} | **시간:** {article['time']}"),  # info
                            gr.update(visible=False),  # summary_output
                        ])
                    else:
                        outputs.extend([
                            gr.update(visible=False),  # group
                            gr.update(),  # title
                            gr.update(),  # image
                            gr.update(),  # snippet
                            gr.update(),  # info
                            gr.update(visible=False),  # summary_output
                        ])
                articles_state = articles  # articles_state 업데이트
            outputs.append(articles_state)
            # 상태 메시지 숨김 (Progress는 자동으로 사라집니다)
        return outputs

    # search_button 클릭 시 업데이트될 출력 컴포넌트 목록 생성
    search_outputs = []
    error_output = gr.Markdown(visible=False)
    search_outputs.append(error_output)
    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['summary_output'])
    search_outputs.append(articles_state)
    # status_message 컴포넌트 제거

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

    # 분석 버튼 클릭 이벤트 설정
    for idx, comp in enumerate(article_components):
        def create_analyze_function(index=idx):
            def analyze_article(articles):
                with gr.Progress() as progress:
                    progress(0, desc="처리중입니다. 잠시만 기다리세요.")
                    if articles and index < len(articles):
                        article = articles[index]
                        summary = summarize_article(article['title'], article['snippet'])
                        return gr.update(value=summary, visible=True)
                    else:
                        return gr.update(value="기사 정보를 찾을 수 없습니다.", visible=True)
            return analyze_article

        comp['analyze_button'].click(
            create_analyze_function(),
            inputs=[articles_state],
            outputs=comp['summary_output']
        )

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