app.py

import gradio as gr
import requests
import json
import os
from datetime import datetime, timedelta
from huggingface_hub import InferenceClient

API_KEY = os.getenv("SERPHOUSE_API_KEY")
hf_client = InferenceClient("CohereForAI/c4ai-command-r-plus-08-2024", token=os.getenv("HF_TOKEN"))

# 국가별 언어 코드 매핑
COUNTRY_LANGUAGES = {
    "South Korea": "ko",
    "Japan": "ja",
    "China": "zh",
    "Russia": "ru",
    "France": "fr",
    "Germany": "de",
    "Spain": "es",
    "Italy": "it",
    "Netherlands": "nl",
    "Portugal": "pt",
    "Thailand": "th",
    "Vietnam": "vi",
    "Indonesia": "id",
    "Malaysia": "ms",
    "Saudi Arabia": "ar",
    "United Arab Emirates": "ar",
    "Egypt": "ar",
    "Morocco": "ar"
}

COUNTRY_LOCATIONS = {
    "United States": "United States",
    "United Kingdom": "United Kingdom",
    "Canada": "Canada",
    "Australia": "Australia",
    "Germany": "Germany",
    "France": "France",
    "Japan": "Japan",
    "South Korea": "South Korea",
    "China": "China",
    "India": "India",
    "Brazil": "Brazil",
    "Mexico": "Mexico",
    "Russia": "Russia",
    "Italy": "Italy",
    "Spain": "Spain",
    "Netherlands": "Netherlands",
    "Singapore": "Singapore",
    "Hong Kong": "Hong Kong",
    "Indonesia": "Indonesia",
    "Malaysia": "Malaysia",
    "Philippines": "Philippines",
    "Thailand": "Thailand",
    "Vietnam": "Vietnam",
    "Belgium": "Belgium",
    "Denmark": "Denmark",
    "Finland": "Finland",
    "Ireland": "Ireland",
    "Norway": "Norway",
    "Poland": "Poland",
    "Sweden": "Sweden",
    "Switzerland": "Switzerland",
    "Austria": "Austria",
    "Czech Republic": "Czech Republic",
    "Greece": "Greece",
    "Hungary": "Hungary",
    "Portugal": "Portugal",
    "Romania": "Romania",
    "Turkey": "Turkey",
    "Israel": "Israel",
    "Saudi Arabia": "Saudi Arabia",
    "United Arab Emirates": "United Arab Emirates",
    "South Africa": "South Africa",
    "Argentina": "Argentina",
    "Chile": "Chile",
    "Colombia": "Colombia",
    "Peru": "Peru",
    "Venezuela": "Venezuela",
    "New Zealand": "New Zealand",
    "Bangladesh": "Bangladesh",
    "Pakistan": "Pakistan",
    "Egypt": "Egypt",
    "Morocco": "Morocco",
    "Nigeria": "Nigeria",
    "Kenya": "Kenya",
    "Ukraine": "Ukraine",
    "Croatia": "Croatia",
    "Slovakia": "Slovakia",
    "Bulgaria": "Bulgaria",
    "Serbia": "Serbia",
    "Estonia": "Estonia",
    "Latvia": "Latvia",
    "Lithuania": "Lithuania",
    "Slovenia": "Slovenia",
    "Luxembourg": "Luxembourg",
    "Malta": "Malta",
    "Cyprus": "Cyprus",
    "Iceland": "Iceland"
}

MAJOR_COUNTRIES = list(COUNTRY_LOCATIONS.keys())

def translate_query(query, country):
    try:
        # 한글 입력이고 South Korea가 선택된 경우 원문 그대로 반환
        if country == "South Korea":
            return query[:255]  # 길이 제한만 적용
            
        if country in COUNTRY_LANGUAGES:
            query = query[:100]
            target_lang = COUNTRY_LANGUAGES[country]
            prompt = f"Translate only this text to {target_lang} language without any explanation. Output only translated text: {query}"
            
            translated = hf_client.text_generation(
                prompt,
                max_new_tokens=50,
                temperature=0.1
            )
            translated = translated.strip()[:255]
            print(f"Original query: {query}")
            print(f"Translated query: {translated}")
            return translated
        return query[:255]
    except Exception as e:
        print(f"Translation error: {str(e)}")
        return query[:255]



def search_serphouse(query, country, page=1, num_result=10):
    url = "https://api.serphouse.com/serp/live"
    
    # 검색어 번역
    translated_query = translate_query(query, country)
    print(f"Original query: {query}")
    print(f"Translated query: {translated_query}")
    
    payload = {
        "data": {
            "q": translated_query,
            "domain": "google.com",
            "loc": COUNTRY_LOCATIONS.get(country, "United States"),
            "lang": COUNTRY_LANGUAGES.get(country, "en"),
            "device": "desktop",
            "serp_type": "news",
            "page": "1",
            "num": "10"  # num으로 변경
        }
    }

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

    try:
        response = requests.post(url, json=payload, headers=headers)
        print("Request payload:", json.dumps(payload, indent=2, ensure_ascii=False))
        print("Response status:", response.status_code)
        
        response.raise_for_status()
        return {"results": response.json(), "translated_query": translated_query}
    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, "translated_query": query}

def format_results_from_raw(response_data):
    if "error" in response_data:
        return "Error: " + response_data["error"], []

    try:
        results = response_data["results"]
        translated_query = response_data["translated_query"]
        
        news_results = results.get('results', {}).get('results', {}).get('news', [])
        if not news_results:
            return "검색 결과가 없습니다.", []

        articles = []
        for idx, result in enumerate(news_results, 1):
            articles.append({
                "index": idx,
                "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", "")),
                "translated_query": translated_query
            })
        return "", articles
    except Exception as e:
        return f"결과 처리 중 오류 발생: {str(e)}", []

def serphouse_search(query, country):
    response_data = search_serphouse(query, country)
    return format_results_from_raw(response_data)

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

with gr.Blocks(theme="Nymbo/Nymbo_Theme", css=css, title="NewsAI 서비스") as iface:
    gr.Markdown("검색어를 입력하고 원하는 국가를 선택하면, 해당 국가의 언어로 번역된 검색어로 뉴스를 검색합니다.")

    with gr.Column():
        with gr.Row():
            query = gr.Textbox(label="검색어")
            country = gr.Dropdown(MAJOR_COUNTRIES, label="국가", value="South Korea")
            
        # 번역된 검색어 표시 컴포넌트
        translated_display = gr.Markdown(visible=True)
            
        search_button = gr.Button("검색", variant="primary")

        progress = gr.Progress()
        status_message = gr.Markdown(visible=False)
        articles_state = gr.State([])

        article_components = []
        for i in range(100):
            with gr.Group(visible=False) as article_group:
                title = gr.Markdown()
                image = gr.Image(width=200, height=150)
                snippet = gr.Markdown()
                info = gr.Markdown()

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

    def search_and_display(query, country, articles_state, progress=gr.Progress()):
        progress(0, desc="검색 시작...")
    
    # 번역 처리 (South Korea 선택 시 번역하지 않음)
        translated_query = translate_query(query, country)
        if country == "South Korea":
            translated_display_text = f"검색어: {query}"
        elif translated_query != query:
            translated_display_text = f"원본 검색어: {query}\n번역된 검색어: {translated_query}"
        else:
            translated_display_text = f"검색어: {query}"
        
        progress(0.2, desc="검색 중...")


        response_data = search_serphouse(query, country)
        error_message, articles = format_results_from_raw(response_data)
        
        outputs = [gr.update(value=translated_display_text, visible=True)]
        
        if error_message:
            outputs.append(gr.update(value=error_message, visible=True))
            for comp in article_components:
                outputs.extend([
                    gr.update(visible=False), gr.update(), gr.update(),
                    gr.update(), gr.update()
                ])
            articles_state = []
        else:
            outputs.append(gr.update(value="", visible=False))
            total_articles = len(articles)
            for idx, comp in enumerate(article_components):
                progress((idx + 1) / total_articles, desc=f"결과 표시 중... {idx + 1}/{total_articles}")
                if idx < len(articles):
                    article = articles[idx]
                    image_url = article['image_url']
                    image_update = gr.update(value=image_url, visible=True) if image_url and not image_url.startswith('data:image') else gr.update(value=None, visible=False)
                    
                    outputs.extend([
                        gr.update(visible=True),
                        gr.update(value=f"### [{article['title']}]({article['link']})"),
                        image_update,
                        gr.update(value=f"**요약:** {article['snippet']}"),
                        gr.update(value=f"**출처:** {article['channel']} | **시간:** {article['time']}")
                    ])
                else:
                    outputs.extend([
                        gr.update(visible=False), gr.update(), gr.update(),
                        gr.update(), gr.update()
                    ])
            articles_state = articles

        progress(1.0, desc="완료!")
        outputs.append(articles_state)
        outputs.append(gr.update(visible=False))
        return outputs

    # 검색 버튼 클릭 시 업데이트될 출력 컴포넌트 목록
    search_outputs = [translated_display, gr.Markdown(visible=False)]  # 번역 결과 표시 컴포넌트
    for comp in article_components:
        search_outputs.extend([
            comp['group'],
            comp['title'],
            comp['image'],
            comp['snippet'],
            comp['info']
        ])
    search_outputs.extend([articles_state, status_message])

    # 검색 버튼 클릭 이벤트 설정
    search_button.click(
        fn=search_and_display,
        inputs=[query, country, articles_state],
        outputs=search_outputs,
        show_progress=True
    )

iface.launch()