Spaces:
Sleeping
Sleeping
| # requirements.txt 에 다음 내용을 추가하거나 업데이트해야 합니다: | |
| # gradio>=4.0 | |
| # requests | |
| import gradio as gr | |
| import requests | |
| from requests.exceptions import RequestException | |
| import json | |
| import time | |
| # --- 전역 변수 및 세션 설정 --- | |
| SERVER_URL = "" | |
| REQUEST_TIMEOUT = 15 | |
| session = requests.Session() | |
| adapter = requests.adapters.HTTPAdapter( | |
| pool_connections=5, pool_maxsize=10, | |
| max_retries=requests.adapters.Retry( | |
| total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], | |
| allowed_methods=["GET", "POST"], connect=3, read=3, redirect=3 | |
| ) | |
| ) | |
| session.mount('http://', adapter) | |
| session.mount('https://', adapter) | |
| # --- 백엔드 통신 함수 --- | |
| def connect_server(ngrok_url): | |
| """입력된 Ngrok URL로 서버 연결 시도""" | |
| global SERVER_URL | |
| print(f"[DEBUG] 서버 연결 시도: {ngrok_url}") | |
| try: | |
| if not ngrok_url: | |
| return "오류: 서버 URL을 입력해주세요." | |
| # URL 끝 슬래시 제거 | |
| if ngrok_url.endswith('/'): | |
| ngrok_url = ngrok_url[:-1] | |
| # '/health' 또는 '/api/status' 엔드포인트로 상태 확인 | |
| health_url = f"{ngrok_url}/health" | |
| status_url = f"{ngrok_url}/api/status" | |
| test_url = "" | |
| response = None | |
| try: | |
| test_url = health_url | |
| print(f"[DEBUG] {test_url} 확인 중...") | |
| response = session.get(test_url, timeout=5) # 상태 확인은 짧은 타임아웃 | |
| if response.status_code != 200: | |
| raise RequestException("Health check failed or returned non-200") | |
| except RequestException as e_health: | |
| print(f"[DEBUG] {test_url} 실패 ({e_health}), {status_url} 로 재시도...") | |
| try: | |
| test_url = status_url | |
| response = session.get(test_url, timeout=5) # 상태 확인은 짧은 타임아웃 | |
| if response.status_code != 200: | |
| raise RequestException("Status check failed or returned non-200") | |
| except RequestException as e_status: | |
| print(f"[ERROR] 서버 연결 최종 실패: {e_status}") | |
| return f"서버 연결 실패: {test_url} 확인 중 오류 발생 ({e_status})" | |
| # 연결 성공 시 | |
| SERVER_URL = ngrok_url | |
| print(f"[INFO] 서버 연결 성공: {SERVER_URL}") | |
| return f"서버 연결 성공! ({test_url} 응답코드: {response.status_code})" | |
| except Exception as e: # 포괄적인 예외 처리 | |
| print(f"[ERROR] 서버 연결 중 예외 발생: {e}") | |
| return f"서버 연결 중 예상치 못한 오류 발생: {str(e)}" | |
| def check_status(): | |
| """서버의 /api/status 엔드포인트 상태 확인""" | |
| global SERVER_URL | |
| if not SERVER_URL: | |
| return gr.update(value="오류: 서버에 먼저 연결하세요.") | |
| print(f"[DEBUG] 서버 상태 확인 요청: {SERVER_URL}/api/status") | |
| try: | |
| response = session.get(f"{SERVER_URL}/api/status", timeout=REQUEST_TIMEOUT) | |
| if response.status_code == 200: | |
| try: | |
| status_data = json.dumps(response.json(), indent=2, ensure_ascii=False) | |
| print(f"[INFO] 상태 확인 성공") | |
| return gr.update(value=status_data) | |
| except json.JSONDecodeError: | |
| print(f"[ERROR] 상태 응답 JSON 파싱 실패") | |
| return gr.update(value="오류: 서버 응답 형식이 잘못되었습니다 (JSON 아님).") | |
| else: | |
| print(f"[ERROR] 상태 확인 실패: HTTP {response.status_code}") | |
| return gr.update(value=f"상태 확인 실패: HTTP {response.status_code}\n{response.text}") | |
| except RequestException as e: | |
| print(f"[ERROR] 상태 확인 중 네트워크 오류: {e}") | |
| return gr.update(value=f"상태 확인 중 오류 발생: {str(e)}") | |
| def echo_test(message): | |
| """서버의 /api/send (echo 액션) 테스트""" | |
| global SERVER_URL | |
| if not SERVER_URL: | |
| return gr.update(value="오류: 서버에 먼저 연결하세요.") | |
| if not message: | |
| return gr.update(value="오류: 에코 테스트 메시지를 입력하세요.") | |
| print(f"[DEBUG] 에코 테스트 요청: {SERVER_URL}/api/send, 메시지: {message}") | |
| try: | |
| payload = {"action": "echo", "data": {"message": message}, "timestamp": int(time.time() * 1000)} | |
| response = session.post(f"{SERVER_URL}/api/send", json=payload, headers={"Content-Type": "application/json"}, timeout=REQUEST_TIMEOUT) | |
| if response.status_code == 200: | |
| try: | |
| echo_data = json.dumps(response.json(), indent=2, ensure_ascii=False) | |
| print(f"[INFO] 에코 테스트 성공") | |
| return gr.update(value=echo_data) | |
| except json.JSONDecodeError: | |
| print(f"[ERROR] 에코 응답 JSON 파싱 실패") | |
| return gr.update(value="오류: 서버 응답 형식이 잘못되었습니다 (JSON 아님).") | |
| else: | |
| print(f"[ERROR] 에코 테스트 실패: HTTP {response.status_code}") | |
| return gr.update(value=f"에코 테스트 실패: HTTP {response.status_code}\n{response.text}") | |
| except RequestException as e: | |
| print(f"[ERROR] 에코 테스트 중 네트워크 오류: {e}") | |
| return gr.update(value=f"에코 테스트 중 오류 발생: {str(e)}") | |
| def get_programs(): | |
| """서버의 /api/programs 에서 프로그램 목록 조회""" | |
| global SERVER_URL | |
| print(f"[DEBUG] 프로그램 목록 조회 함수 시작") | |
| program_options = [] | |
| default_return = ("<div style='color: gray;'>프로그램 목록이 없습니다.</div>", gr.update(choices=[]), gr.update(choices=[])) | |
| # 테스트 모드 (서버 URL 없을 때) | |
| if not SERVER_URL: | |
| print(f"[DEBUG] 테스트 모드: 프로그램 목록 생성") | |
| test_programs = [ | |
| {"id": "test1", "name": "테스트 메모장", "description": "텍스트 편집기", "path": "C:\\Windows\\notepad.exe"}, | |
| {"id": "test2", "name": "테스트 계산기", "description": "기본 계산기", "path": "C:\\Windows\\System32\\calc.exe"} | |
| ] | |
| html_output = "<div style='padding: 10px; background-color: #fffaeb; border: 1px solid #ffecb3; border-radius: 5px;'>" | |
| html_output += "<h4 style='margin-top:0; color: #856404;'>[테스트 모드]</h4>" | |
| html_output += "<p style='font-size: 0.9em;'>서버에 연결되지 않아 테스트 데이터가 표시됩니다.</p>" | |
| html_output += "<table style='width: 100%; border-collapse: collapse; margin-top: 10px;'><thead><tr style='background-color: #fff3cd;'>" | |
| html_output += "<th style='padding: 8px; text-align: left;'>이름</th><th style='padding: 8px; text-align: left;'>설명</th></tr></thead><tbody>" | |
| for p in test_programs: | |
| program_options.append((f"{p['name']} ({p['description']})", p['id'])) | |
| html_output += f"<tr style='border-bottom: 1px solid #eee;'><td style='padding: 8px;'>{p['name']}</td><td style='padding: 8px;'>{p['description']}</td></tr>" | |
| html_output += "</tbody></table></div>" | |
| print(f"[DEBUG] 테스트 옵션: {program_options}") | |
| return html_output, gr.update(choices=program_options), gr.update(choices=program_options) | |
| # 실제 API 호출 (서버 URL 있을 때) | |
| try: | |
| api_url = f"{SERVER_URL}/api/programs" | |
| print(f"[DEBUG] API 호출: {api_url}") | |
| headers = {"User-Agent": "GradioClient/1.0", "Accept": "application/json"} | |
| response = session.get(api_url, headers=headers, timeout=REQUEST_TIMEOUT) | |
| print(f"[DEBUG] API 응답 코드: {response.status_code}") | |
| if response.status_code == 200: | |
| try: | |
| result_json = response.json() | |
| programs = result_json.get("programs", []) | |
| if not programs: # 프로그램 목록이 비어있을 경우 | |
| print("[INFO] 서버에서 반환된 프로그램 목록이 비어있습니다.") | |
| return "<div style='color: gray; padding: 10px;'>서버에 등록된 프로그램이 없습니다.</div>", gr.update(choices=[]), gr.update(choices=[]) | |
| # 프로그램 목록 테이블 생성 | |
| html_output = "<div style='max-height: 300px; overflow-y: auto; border: 1px solid #eee; padding: 10px; border-radius: 5px;'>" | |
| html_output += "<table style='width: 100%; border-collapse: collapse;'><thead><tr style='background-color: #f8f9fa;'>" | |
| html_output += "<th style='padding: 8px; text-align: left;'>이름</th><th style='padding: 8px; text-align: left;'>설명</th><th style='padding: 8px; text-align: left;'>경로</th></tr></thead><tbody>" | |
| for program in programs: | |
| program_id = program.get("id", "") | |
| program_name = program.get("name", "N/A") | |
| program_desc = program.get("description", "") | |
| program_path = program.get("path", "") | |
| # ID가 유효한 경우에만 옵션 추가 | |
| if program_id: | |
| program_options.append((f"{program_name} ({program_desc})", program_id)) | |
| html_output += f"<tr style='border-bottom: 1px solid #eee;'><td style='padding: 8px;'>{program_name}</td><td style='padding: 8px;'>{program_desc}</td><td style='padding: 8px;'>{program_path}</td></tr>" | |
| else: | |
| print(f"[WARN] ID가 없는 프로그램 데이터 발견: {program}") | |
| html_output += f"<tr style='border-bottom: 1px solid #eee; background-color: #fff3cd;'><td style='padding: 8px; color: orange;' colspan='3'>[주의] ID가 없는 프로그램: {program_name}</td></tr>" | |
| html_output += "</tbody></table></div>" | |
| html_output += f"<div style='margin-top: 10px; font-size: 0.9em; color: #666;'>총 {len(program_options)}개 프로그램 (ID 유효 기준)</div>" | |
| print(f"[INFO] 프로그램 목록 조회 성공, 옵션 개수: {len(program_options)}") | |
| return html_output, gr.update(choices=program_options), gr.update(choices=program_options) | |
| except json.JSONDecodeError: | |
| print(f"[ERROR] 프로그램 목록 응답 JSON 파싱 실패") | |
| error_html = "<div style='color: red;'>오류: 서버 응답 형식이 잘못되었습니다 (JSON 아님).</div>" | |
| return error_html, gr.update(choices=[]), gr.update(choices=[]) | |
| else: | |
| print(f"[ERROR] 프로그램 목록 조회 실패: HTTP {response.status_code}") | |
| error_html = f"<div style='color: red;'>프로그램 목록 조회 실패: HTTP {response.status_code}<br>{response.text}</div>" | |
| return error_html, gr.update(choices=[]), gr.update(choices=[]) | |
| except RequestException as e: | |
| print(f"[ERROR] 프로그램 목록 조회 중 네트워크 오류: {e}") | |
| error_html = f"<div style='color: red;'>프로그램 목록 조회 중 오류 발생: {str(e)}</div>" | |
| return error_html, gr.update(choices=[]), gr.update(choices=[]) | |
| except Exception as e: # 포괄적 예외 처리 | |
| print(f"[ERROR] 프로그램 목록 조회 중 예상치 못한 오류: {e}") | |
| error_html = f"<div style='color: red;'>프로그램 목록 조회 중 예상치 못한 오류: {str(e)}</div>" | |
| return error_html, gr.update(choices=[]), gr.update(choices=[]) | |
| def execute_program(program_id): | |
| """선택된 프로그램 ID를 사용하여 서버에 실행 요청""" | |
| global SERVER_URL | |
| # 입력 값 타입 로깅 및 검증 | |
| print(f"[DEBUG] 프로그램 실행 시도: ID='{program_id}', Type={type(program_id)}") | |
| if not isinstance(program_id, str): | |
| error_msg = f"오류: 프로그램 ID 형식이 잘못되었습니다 (받은 타입: {type(program_id)}). 프로그램을 다시 선택해주세요." | |
| print(f"[ERROR] {error_msg}") | |
| return f"<div style='color: red;' class='error-box'>{error_msg}</div>" | |
| if not program_id: | |
| error_msg = "오류: 실행할 프로그램을 선택해주세요." | |
| print(f"[WARN] {error_msg}") | |
| return f"<div style='color: orange;' class='warning-box'>{error_msg}</div>" | |
| # 테스트 ID 처리 | |
| if program_id.startswith("test"): | |
| print(f"[DEBUG] 테스트 프로그램 '{program_id}' 실행 시뮬레이션") | |
| return f"<div style='color: green;' class='success-box'>테스트 프로그램 '{program_id}' 실행 완료 (시뮬레이션)</div>" | |
| if not SERVER_URL: | |
| error_msg = "오류: 서버에 먼저 연결해야 합니다." | |
| print(f"[WARN] {error_msg}") | |
| return f"<div style='color: orange;' class='warning-box'>{error_msg}</div>" | |
| # 실제 API 호출 | |
| try: | |
| api_url = f"{SERVER_URL}/api/programs/{program_id}/execute" | |
| print(f"[DEBUG] 서버 실행 요청: {api_url}") | |
| headers = {"User-Agent": "GradioClient/1.0", "Accept": "application/json", "Content-Type": "application/json"} | |
| # POST 요청의 본문에 ID 포함 (서버 API 구현에 따라 필요 없을 수도 있음) | |
| payload = {"id": program_id} | |
| response = session.post(api_url, headers=headers, json=payload, timeout=REQUEST_TIMEOUT) | |
| print(f"[DEBUG] 실행 API 응답 코드: {response.status_code}") | |
| if response.status_code == 200: | |
| try: | |
| result = response.json() | |
| msg = result.get('message', '프로그램이 성공적으로 실행되었습니다.') | |
| print(f"[INFO] 프로그램 실행 성공: {msg}") | |
| return f"<div style='color: green;' class='success-box'>실행 성공: {msg}</div>" | |
| except json.JSONDecodeError: | |
| print(f"[ERROR] 실행 응답 JSON 파싱 실패") | |
| return f"<div style='color: red;' class='error-box'>오류: 실행 응답 형식이 잘못되었습니다 (JSON 아님).</div>" | |
| else: | |
| error_text = response.text # 오류 메시지 포함 | |
| print(f"[ERROR] 프로그램 실행 실패: HTTP {response.status_code}, 응답: {error_text}") | |
| return f"<div style='color: red;' class='error-box'>실행 실패: HTTP {response.status_code}<br><pre>{error_text}</pre></div>" | |
| except RequestException as e: | |
| error_msg = f"실행 중 네트워크 오류 발생: {str(e)}" | |
| print(f"[ERROR] {error_msg}") | |
| return f"<div style='color: red;' class='error-box'>{error_msg}</div>" | |
| except Exception as e: # 포괄적 예외 처리 | |
| error_msg = f"실행 중 예상치 못한 오류 발생: {str(e)}" | |
| print(f"[ERROR] {error_msg}") | |
| return f"<div style='color: red;' class='error-box'>{error_msg}</div>" | |
| # --- Gradio UI 구성 (최신 버전 스타일) --- | |
| with gr.Blocks(title="LocalPCAgent 제어 인터페이스 v4", theme=gr.themes.Soft()) as demo: | |
| # 사용자 정의 CSS (선택 사항) | |
| gr.HTML(""" | |
| <style> | |
| .error-box { padding: 10px; border: 1px solid #dc3545; border-radius: 5px; background-color: #f8d7da; } | |
| .success-box { padding: 10px; border: 1px solid #28a745; border-radius: 5px; background-color: #d4edda; } | |
| .warning-box { padding: 10px; border: 1px solid #ffc107; border-radius: 5px; background-color: #fff3cd; } | |
| table { font-size: 0.9em; } | |
| th, td { padding: 6px 8px; border-bottom: 1px solid #eee; } | |
| th { background-color: #f8f9fa; font-weight: 500; } | |
| tbody tr:hover { background-color: #f1f1f1; } | |
| </style> | |
| """) | |
| gr.Markdown("# LocalPCAgent 제어 인터페이스") | |
| gr.Markdown("원격 PC의 상태 확인 및 프로그램 실행을 위한 웹 UI입니다.") | |
| # 섹션 1: 서버 연결 | |
| with gr.Accordion("1. 서버 연결 설정", open=True): # Accordion 사용 | |
| with gr.Row(): | |
| ngrok_url_input = gr.Textbox( | |
| label="Ngrok URL 입력", | |
| placeholder="https://xxxx-xx-xxx-xxx.ngrok-free.app 형식", | |
| scale=3 # 너비 비율 | |
| ) | |
| connect_btn = gr.Button("서버 연결", variant="secondary", scale=1) | |
| connection_status = gr.Textbox(label="연결 상태", interactive=False) | |
| # 섹션 2: 기능 탭 | |
| with gr.Tabs(): | |
| # 탭 1: 기본 기능 | |
| with gr.Tab("기본 기능"): | |
| gr.Markdown("서버의 기본 상태를 확인하고 통신 테스트를 수행합니다.") | |
| with gr.Row(equal_height=True): # Row 안의 요소들 높이 맞춤 | |
| with gr.Column(scale=1): | |
| gr.Markdown("#### 서버 상태") | |
| check_status_btn = gr.Button("상태 확인", variant="secondary") | |
| gr.Markdown("#### 에코 테스트") | |
| echo_message = gr.Textbox(label="보낼 메시지", value="Hello from Gradio!", lines=1) | |
| echo_btn = gr.Button("에코 테스트 전송", variant="secondary") | |
| with gr.Column(scale=2): | |
| status_result = gr.Textbox(label="상태 결과", interactive=False, lines=5, max_lines=10) | |
| echo_result = gr.Textbox(label="에코 결과", interactive=False, lines=5, max_lines=10) | |
| # 탭 2: 프로그램 실행 | |
| with gr.Tab("프로그램 실행"): | |
| gr.Markdown("서버에 등록된 프로그램을 조회하고 실행합니다.") | |
| with gr.Column(): | |
| gr.Markdown("#### 단계 1: 프로그램 목록 조회") | |
| with gr.Row(variant='compact'): # 컴포넌트 간 간격 좁힘 | |
| get_programs_btn = gr.Button("목록 새로고침", variant="secondary") | |
| gr.Markdown(f"<small> (네트워크 상태에 따라 최대 {REQUEST_TIMEOUT}초 소요될 수 있습니다)</small>", visible=True) # 안내 문구 | |
| # HTML 결과 표시 영역 | |
| programs_result = gr.HTML(label="등록된 프로그램 목록") | |
| # 참고용/숨겨진 드롭다운 (UI에는 보이지 않음) | |
| program_dropdown_display = gr.Dropdown(label="내부용", visible=False) | |
| with gr.Column(): | |
| gr.Markdown("#### 단계 2: 프로그램 선택 및 실행") | |
| # 실제 사용자가 선택할 드롭다운 | |
| program_dropdown_exec = gr.Dropdown( | |
| label="실행할 프로그램 선택", | |
| choices=[], # 초기값은 비어있음 | |
| interactive=True, | |
| info="위 '목록 새로고침' 버튼을 눌러 목록을 가져온 후 선택하세요." # Dropdown 아래 안내 문구 | |
| ) | |
| execute_program_btn = gr.Button("선택한 프로그램 실행", variant="primary") # 주요 실행 버튼 강조 | |
| # 실행 결과 표시 영역 | |
| execute_result = gr.HTML(label="실행 결과") | |
| # --- 이벤트 핸들러 연결 --- | |
| connect_btn.click(fn=connect_server, inputs=ngrok_url_input, outputs=connection_status) | |
| check_status_btn.click(fn=check_status, inputs=None, outputs=status_result) | |
| echo_btn.click(fn=echo_test, inputs=echo_message, outputs=echo_result) | |
| # 프로그램 목록 새로고침 버튼 클릭 | |
| get_programs_btn.click( | |
| fn=get_programs, | |
| inputs=None, | |
| # outputs 순서: HTML 결과, 숨겨진 드롭다운, 실행용 드롭다운 | |
| outputs=[programs_result, program_dropdown_display, program_dropdown_exec] | |
| ) | |
| # 실행할 프로그램 드롭다운 선택 변경 시 (선택 사항: 결과 영역에 즉시 피드백) | |
| # program_dropdown_exec.change( | |
| # fn=lambda choice: f"<div style='color: blue; padding: 5px;'>'{choice}' 프로그램이 선택되었습니다. 실행 버튼을 클릭하세요.</div>" if choice else "", | |
| # inputs=program_dropdown_exec, | |
| # outputs=execute_result | |
| # ) | |
| # 프로그램 실행 버튼 클릭 | |
| execute_program_btn.click( | |
| fn=execute_program, | |
| inputs=[program_dropdown_exec], # 선택된 드롭다운 값을 입력으로 사용 | |
| outputs=[execute_result] # 실행 결과를 HTML 영역에 표시 | |
| ) | |
| # --- 앱 실행 --- | |
| if __name__ == "__main__": | |
| print("==============================================") | |
| print(" LocalPCAgent 웹 인터페이스 (Gradio 4.x) 시작 ") | |
| print("==============================================") | |
| print("* 서버 URL을 입력하고 '서버 연결'을 먼저 누르세요.") | |
| print("* 문제가 발생하면 터미널 로그를 확인하세요.") | |
| print("* Hugging Face Spaces 배포 시 requirements.txt 에") | |
| print(" 'gradio>=4.0' 와 'requests' 를 포함해야 합니다.") | |
| print("----------------------------------------------") | |
| # share=True 옵션은 외부 접속 링크 생성 (로컬 테스트 시 불필요하면 제거) | |
| demo.launch(debug=True, show_error=True) | |
| print("==============================================") | |
| print(" 웹 인터페이스 종료됨 ") | |
| print("==============================================") |