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(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("검색") # 기사 데이터를 저장할 상태 변수 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): # Progress 객체 생성 progress = gr.Progress(track_tqdm=True) progress(0, desc="처리중입니다. 잠시만 기다리세요.") error_message, articles = serphouse_search(query, country) progress(50) # 진행률 50%로 업데이트 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(100) # 진행률 100%로 업데이트 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_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): progress = gr.Progress(track_tqdm=True) progress(0, desc="처리중입니다. 잠시만 기다리세요.") if articles and index < len(articles): article = articles[index] summary = summarize_article(article['title'], article['snippet']) progress(100) # 진행률 100%로 업데이트 return gr.update(value=summary, visible=True) else: progress(100) # 진행률 100%로 업데이트 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"))