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=100): # num_result를 100으로 변경 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 = 100 # num_result를 100으로 설정 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시간 이내 뉴스를 최대 100개 출력합니다.") with gr.Column(): with gr.Row(): query = gr.Textbox(label="검색어") country = gr.Dropdown(MAJOR_COUNTRIES, label="국가", value="South Korea") search_button = gr.Button("검색") # 상태 메시지 컴포넌트 추가 status_message = gr.Markdown(visible=False) # 기사 데이터를 저장할 상태 변수 articles_state = gr.State([]) # 초기값을 빈 리스트로 설정 # 최대 100개의 기사에 대한 컴포넌트를 미리 생성합니다. article_components = [] for i in range(100): # 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() 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): # 상태 메시지 표시 status_update = gr.update(value="처리중입니다. 잠시만 기다리세요.", visible=True) 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) # 상태 메시지 숨김 outputs.append(gr.update(visible=False)) 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) search_outputs.append(status_message) # 상태 메시지 출력에 추가 search_button.click( search_and_display, inputs=[query, country, articles_state], outputs=search_outputs, show_progress=False # Gradio의 기본 로딩 표시를 끕니다. ) # 분석 버튼 클릭 이벤트 설정 for idx, comp in enumerate(article_components): def create_analyze_function(index=idx): def analyze_article(articles): # 상태 메시지 표시 status_update = gr.update(value="처리중입니다. 잠시만 기다리세요.", visible=True) if articles and index < len(articles): article = articles[index] summary = summarize_article(article['title'], article['snippet']) # 상태 메시지 숨김 return gr.update(value=summary, visible=True), gr.update(visible=False) else: # 상태 메시지 숨김 return gr.update(value="기사 정보를 찾을 수 없습니다.", visible=True), gr.update(visible=False) return analyze_article # 분석 버튼의 출력에 상태 메시지를 추가합니다. comp['analyze_button'].click( create_analyze_function(), inputs=[articles_state], outputs=[comp['summary_output'], status_message], show_progress=False # Gradio의 기본 로딩 표시를 끕니다. ) iface.launch(auth=("gini", "pick"))