# 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 = ("
프로그램 목록이 없습니다.
", 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 = "
" html_output += "

[테스트 모드]

" html_output += "

서버에 연결되지 않아 테스트 데이터가 표시됩니다.

" html_output += "" html_output += "" for p in test_programs: program_options.append((f"{p['name']} ({p['description']})", p['id'])) html_output += f"" html_output += "
이름설명
{p['name']}{p['description']}
" 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 "
서버에 등록된 프로그램이 없습니다.
", gr.update(choices=[]), gr.update(choices=[]) # 프로그램 목록 테이블 생성 html_output = "
" html_output += "" html_output += "" 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"" else: print(f"[WARN] ID가 없는 프로그램 데이터 발견: {program}") html_output += f"" html_output += "
이름설명경로
{program_name}{program_desc}{program_path}
[주의] ID가 없는 프로그램: {program_name}
" html_output += f"
총 {len(program_options)}개 프로그램 (ID 유효 기준)
" 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 = "
오류: 서버 응답 형식이 잘못되었습니다 (JSON 아님).
" return error_html, gr.update(choices=[]), gr.update(choices=[]) else: print(f"[ERROR] 프로그램 목록 조회 실패: HTTP {response.status_code}") error_html = f"
프로그램 목록 조회 실패: HTTP {response.status_code}
{response.text}
" return error_html, gr.update(choices=[]), gr.update(choices=[]) except RequestException as e: print(f"[ERROR] 프로그램 목록 조회 중 네트워크 오류: {e}") error_html = f"
프로그램 목록 조회 중 오류 발생: {str(e)}
" return error_html, gr.update(choices=[]), gr.update(choices=[]) except Exception as e: # 포괄적 예외 처리 print(f"[ERROR] 프로그램 목록 조회 중 예상치 못한 오류: {e}") error_html = f"
프로그램 목록 조회 중 예상치 못한 오류: {str(e)}
" 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"
{error_msg}
" if not program_id: error_msg = "오류: 실행할 프로그램을 선택해주세요." print(f"[WARN] {error_msg}") return f"
{error_msg}
" # 테스트 ID 처리 if program_id.startswith("test"): print(f"[DEBUG] 테스트 프로그램 '{program_id}' 실행 시뮬레이션") return f"
테스트 프로그램 '{program_id}' 실행 완료 (시뮬레이션)
" if not SERVER_URL: error_msg = "오류: 서버에 먼저 연결해야 합니다." print(f"[WARN] {error_msg}") return f"
{error_msg}
" # 실제 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"
실행 성공: {msg}
" except json.JSONDecodeError: print(f"[ERROR] 실행 응답 JSON 파싱 실패") return f"
오류: 실행 응답 형식이 잘못되었습니다 (JSON 아님).
" else: error_text = response.text # 오류 메시지 포함 print(f"[ERROR] 프로그램 실행 실패: HTTP {response.status_code}, 응답: {error_text}") return f"
실행 실패: HTTP {response.status_code}
{error_text}
" except RequestException as e: error_msg = f"실행 중 네트워크 오류 발생: {str(e)}" print(f"[ERROR] {error_msg}") return f"
{error_msg}
" except Exception as e: # 포괄적 예외 처리 error_msg = f"실행 중 예상치 못한 오류 발생: {str(e)}" print(f"[ERROR] {error_msg}") return f"
{error_msg}
" # --- Gradio UI 구성 (최신 버전 스타일) --- with gr.Blocks(title="LocalPCAgent 제어 인터페이스 v4", theme=gr.themes.Soft()) as demo: # 사용자 정의 CSS (선택 사항) gr.HTML(""" """) 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" (네트워크 상태에 따라 최대 {REQUEST_TIMEOUT}초 소요될 수 있습니다)", 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"
'{choice}' 프로그램이 선택되었습니다. 실행 버튼을 클릭하세요.
" 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("==============================================")