import pandas as pd import src.data_manager as dm from src.openai_client import create_openai_client from src.node_functions import get_nodes_dataframe from datetime import datetime import pytz import gradio as gr try: from gradio import SelectData except ImportError: # Gradio 버전이 낮은 경우 SelectData = None def generate_idea_with_chatgpt( contest_title, contest_theme, contest_description, contest_context="", search_text="", selected_tenants=None, selected_tags=None, ): """ChatGPT를 이용해 아이디어 생성 (필터링된 노드 사용)""" if not contest_title or not contest_theme or not contest_description: return ( "공모전 제목, 주제, 설명을 모두 입력해주세요.", contest_title, # 입력된 값 유지 contest_theme, # 입력된 값 유지 contest_description, # 입력된 값 유지 contest_context, # 입력된 값 유지 ) try: # OpenAI 클라이언트 생성 (.env에서 API 키 자동 로드) client = create_openai_client() # 공모전 정보 구성 contest_info = { "title": contest_title, "theme": contest_theme, "description": contest_description, "context": contest_context, } # 필터링된 노드들 가져오기 filtered_nodes = get_filtered_nodes( search_text, selected_tenants, selected_tags ) # 디버깅 정보 출력 print(f"[ChatGPT 디버그] 필터링된 노드 수: {len(filtered_nodes)}") print( f"[ChatGPT 디버그] 검색어: '{search_text}', 테넌트: {selected_tenants}, 태그: {selected_tags}" ) # 아이디어 생성 generated_idea = client.generate_idea(contest_info, filtered_nodes) if "error" in generated_idea: return ( generated_idea["error"], "", # contest_title 비우기 "", # contest_theme 비우기 "", # contest_description 비우기 "", # contest_context 비우기 ) # 사용된 노드와 필터 정보 추가 generated_idea["used_nodes"] = filtered_nodes generated_idea["used_filters"] = { "search_text": search_text or "", "selected_tenants": selected_tenants or [], "selected_tags": selected_tags or [], "total_nodes_available": len(dm.nodes_data), "filtered_nodes_count": len(filtered_nodes), } # 생성일자 및 고유 ID 추가 (한국 시간) kst = pytz.timezone("Asia/Seoul") current_time = datetime.now(kst) generated_idea["id"] = current_time.strftime("%Y%m%d%H%M%S") + str( len(dm.ideas_data) ) generated_idea["created_at"] = current_time.strftime("%Y-%m-%d %H:%M:%S") generated_idea["created_date"] = current_time.strftime("%Y-%m-%d") generated_idea["created_time"] = current_time.strftime("%H:%M:%S") dm.ideas_data.append(generated_idea) try: dm.save_ideas() except Exception as save_error: return ( f"아이디어 생성은 완료되었지만 저장 중 오류가 발생했습니다: {save_error}", "", # contest_title 비우기 "", # contest_theme 비우기 "", # contest_description 비우기 "", # contest_context 비우기 ) return ( f"아이디어 '{generated_idea['title']}'가 성공적으로 생성되었습니다!", "", # contest_title 비우기 "", # contest_theme 비우기 "", # contest_description 비우기 "", # contest_context 비우기 ) except Exception as e: return ( f"아이디어 생성 중 오류가 발생했습니다: {str(e)}", "", # contest_title 비우기 "", # contest_theme 비우기 "", # contest_description 비우기 "", # contest_context 비우기 ) def get_ideas_dataframe(search_text=""): """생성된 아이디어들을 데이터프레임으로 변환 (최신순 정렬, 검색 지원)""" columns = [ "생성일시", "공모전 제목", "아이디어 제목", "아이디어 개요", "AI 이름", ] if not dm.ideas_data: return pd.DataFrame(columns=columns) # 생성일시 기준으로 내림차순 정렬 (최신 아이디어가 위로) sorted_ideas = sorted( dm.ideas_data, key=lambda x: x.get("created_at", "1900-01-01 00:00:00"), reverse=True, ) df_data = [] for idx, idea in enumerate(sorted_ideas): # 원본 배열에서의 실제 인덱스 찾기 original_index = dm.ideas_data.index(idea) # 공모전 제목 추출 (contest_info 딕셔너리에서) contest_title = "N/A" if idea.get("contest_info") and isinstance(idea.get("contest_info"), dict): contest_title = idea.get("contest_info", {}).get("title", "N/A") # 검색 필터 적용 (공모전 제목 또는 아이디어 제목에서 검색) if search_text: idea_title = idea.get("title", "").lower() contest_title_search = contest_title.lower() search_lower = search_text.lower() # 공모전 제목이나 아이디어 제목 중 하나라도 검색어를 포함하지 않으면 제외 if not (search_lower in idea_title or search_lower in contest_title_search): continue df_data.append( { "생성일시": idea.get("created_at", "N/A"), "공모전 제목": contest_title, "아이디어 제목": idea.get("title", "제목 없음"), "아이디어 개요": idea.get("overview", "개요 없음"), "AI 이름": idea.get("ai_name", "Unknown"), "_original_index": original_index, # 숨겨진 원본 인덱스 } ) # 필터링 결과가 없어도 컬럼명이 유지되도록 빈 DataFrame 반환 if not df_data: return pd.DataFrame(columns=columns) df = pd.DataFrame(df_data) # _original_index 컬럼은 UI에서 보이지 않도록 처리 return df[["생성일시", "공모전 제목", "아이디어 제목", "아이디어 개요", "AI 이름"]] def filter_ideas(search_text): """아이디어 검색 (공모전 제목 또는 아이디어 제목에서 검색)""" df = get_ideas_dataframe(search_text) return gr.update(value=df) def get_idea_details(selection_data): """선택된 아이디어의 상세 정보 반환""" try: # selection_data가 None이거나 비어있는 경우 if selection_data is None: return "아이디어를 선택해주세요.", "", "", "", "", "", "" # Gradio dataframe.select()는 SelectData 객체를 전달함 selected_index = None # SelectData 객체인 경우 (Gradio v4+) if SelectData and isinstance(selection_data, SelectData): if hasattr(selection_data, "index") and isinstance( selection_data.index, (list, tuple) ): selected_index = selection_data.index[0] # 행 인덱스 # hasattr로 index 속성 확인 (일반적인 경우) elif hasattr(selection_data, "index"): if isinstance(selection_data.index, (list, tuple)): selected_index = selection_data.index[0] # 행 인덱스 elif ( hasattr(selection_data.index, "__len__") and len(selection_data.index) > 0 ): selected_index = selection_data.index[0] # 딕셔너리 형태인 경우 elif isinstance(selection_data, dict): if "index" in selection_data: if isinstance(selection_data["index"], (list, tuple)): selected_index = selection_data["index"][0] else: selected_index = selection_data["index"] # 정수 값이 직접 전달된 경우 elif isinstance(selection_data, (int, float)): selected_index = int(selection_data) # 인덱스를 찾지 못한 경우 if selected_index is None: return "아이디어를 선택해주세요.", "", "", "", "", "", "" except Exception as e: return "아이디어 선택 중 오류가 발생했습니다.", "", "", "", "", "", "" if selected_index >= len(dm.ideas_data): return "선택된 아이디어를 찾을 수 없습니다.", "", "", "", "", "", "" idea = dm.ideas_data[selected_index] title = idea.get("title", "제목 없음") problem = idea.get("problem", "문제의식 정보가 없습니다.") solution = idea.get("solution", "솔루션 정보가 없습니다.") implementation = idea.get("implementation", "구현방안 정보가 없습니다.") expected_effect = idea.get("expected_effect", "기대효과 정보가 없습니다.") # 공모전 정보 및 생성일시 contest_info = idea.get("contest_info", {}) created_at = idea.get("created_at", "N/A") contest_details = f"""공모전 제목: {contest_info.get('title', 'N/A')} 주제: {contest_info.get('theme', 'N/A')} 설명: {contest_info.get('description', 'N/A')} 맥락: {contest_info.get('context', 'N/A')}""" return ( title, contest_details, problem, solution, implementation, expected_effect, created_at, ) def get_idea_details_by_index(selected_index): """인덱스를 직접 받아서 아이디어 상세 정보 반환""" try: if selected_index is None or selected_index < 0: return "올바르지 않은 인덱스입니다.", "", "", "", "", "", "" if selected_index >= len(dm.ideas_data): return "선택된 아이디어를 찾을 수 없습니다.", "", "", "", "", "", "" idea = dm.ideas_data[selected_index] title = idea.get("title", "제목 없음") problem = idea.get("problem", "문제의식 정보가 없습니다.") solution = idea.get("solution", "솔루션 정보가 없습니다.") implementation = idea.get("implementation", "구현방안 정보가 없습니다.") expected_effect = idea.get("expected_effect", "기대효과 정보가 없습니다.") # 공모전 정보 및 생성일시 contest_info = idea.get("contest_info", {}) created_at = idea.get("created_at", "N/A") contest_details = f"""공모전 제목: {contest_info.get('title', 'N/A')} 주제: {contest_info.get('theme', 'N/A')} 설명: {contest_info.get('description', 'N/A')} 맥락: {contest_info.get('context', 'N/A')}""" # 사용된 노드 정보 포맷팅 used_nodes = idea.get("used_nodes", []) nodes_info = "" if used_nodes: nodes_list = [] for i, node in enumerate(used_nodes, 1): nodes_list.append( f"{i}. {node.get('title', '제목 없음')} ({node.get('tenant', '미지정')})" ) nodes_info = "\n".join(nodes_list) else: nodes_info = "사용된 노드 정보가 없습니다." # 사용된 필터 정보 포맷팅 used_filters = idea.get("used_filters", {}) filters_info = f"""검색어: {used_filters.get('search_text', '없음')} 선택된 테넌트: {', '.join(used_filters.get('selected_tenants', [])) or '없음'} 선택된 태그: {', '.join(used_filters.get('selected_tags', [])) or '없음'} 전체 노드 수: {used_filters.get('total_nodes_available', 0)} 필터링된 노드 수: {used_filters.get('filtered_nodes_count', 0)}""" # 아이디어 생성 근거 (기존 아이디어는 rationale 필드가 없을 수 있음) rationale = idea.get("rationale", "") if not rationale: rationale = "이 아이디어는 이전 버전에서 생성되어 근거 정보가 없습니다." return ( title, contest_details, problem, solution, implementation, expected_effect, created_at, # 생성일시를 별도 반환 nodes_info, # 사용된 노드 정보 filters_info, # 사용된 필터 정보 rationale, # 아이디어 생성 근거 ) except Exception as e: print(f"[ERROR] get_idea_details_by_index 에러: {e}") return ( "아이디어 정보 로드 중 오류가 발생했습니다.", "", "", "", "", "", "", "", "", "", ) def refresh_ideas(): """아이디어 목록 새로고침""" dm.load_ideas() return get_ideas_dataframe() def clear_ideas(): """모든 아이디어 삭제""" dm.ideas_data = [] dm.save_ideas() return get_ideas_dataframe() def delete_idea(selected_index): """선택된 아이디어 삭제""" if selected_index is None or selected_index >= len(dm.ideas_data): return "삭제할 아이디어를 선택해주세요.", get_ideas_dataframe() deleted_idea = dm.ideas_data.pop(selected_index) dm.save_ideas() return ( f"아이디어 '{deleted_idea.get('title', '제목 없음')}'가 삭제되었습니다.", get_ideas_dataframe(), ) # 추가: Gemini API 연동을 위한 준비 함수 def generate_idea_with_gemini( contest_title, contest_theme, contest_description, contest_context="", search_text="", selected_tenants=None, selected_tags=None, ): """Gemini를 이용해 아이디어 생성 (향후 구현 예정)""" # 필터링된 노드 정보도 로그로 출력 (디버깅용) filtered_nodes = get_filtered_nodes(search_text, selected_tenants, selected_tags) print(f"[Gemini 디버그] 필터링된 노드 수: {len(filtered_nodes)}") return ( "Gemini API 연동은 아직 구현되지 않았습니다.", contest_title, # 입력된 값 유지 (아직 구현되지 않았으므로) contest_theme, # 입력된 값 유지 contest_description, # 입력된 값 유지 contest_context, # 입력된 값 유지 ) def get_filtered_nodes(search_text="", selected_tenants=None, selected_tags=None): """필터링 조건에 맞는 노드들을 반환""" if not dm.nodes_data: return [] filtered_nodes = [] for node in dm.nodes_data: # 텍스트 검색 필터 (노드 이름에서 검색) if search_text and search_text.lower() not in node["title"].lower(): continue # 테넌트 필터 if selected_tenants and node.get("tenant", "미지정") not in selected_tenants: continue # 태그 필터 (선택된 태그 중 하나라도 포함되어야 함) if selected_tags: node_tags = node.get("tags", []) if not any(tag in node_tags for tag in selected_tags): continue filtered_nodes.append(node) return filtered_nodes