""" GeoMedical Helper Client - Asset Management System 400대 에이전트 PC 관리를 위한 자동 업데이트 클라이언트 === APP.EXE VERSION INFORMATION === APP_VERSION = "2.1.1" APP_BUILD_DATE = "2025-08-04" APP_UPDATE_LOG = [ "2.1.1 (2025-08-04): 버전 관리 시스템 개선, 코드 내 버전 관리로 변경", "2.1.0 (2025-08-04): 원격 app.exe 업데이트, 무한 루프 방지, 초기 실행 문제 해결", "2.0.0 (2025-08-03): 관리 대시보드 통합, 원격 업데이트 시스템 구축" ] """ # ============================================================================= # APP.EXE 버전 정보 - 여기서 직접 관리 # ============================================================================= APP_VERSION = "2.1.1" APP_BUILD_DATE = "2025-08-04" APP_DESCRIPTION = "GeoMedical Asset Management Client with Remote Update System" import requests import json import os import sys import importlib.util import time import threading import hashlib import traceback import subprocess import base64 import io from datetime import datetime from urllib.parse import urlparse # 선택적 라이브러리들 - 설치되지 않아도 기본 기능은 작동 try: from PIL import Image, ImageTk except ImportError: print("PIL not installed. Image features will be limited.") Image = None ImageTk = None try: import keyboard except ImportError: print("keyboard not installed. Hotkey features will be limited.") keyboard = None try: import mouse except ImportError: print("mouse not installed. Mouse control features will be limited.") mouse = None try: import pyautogui except ImportError: print("pyautogui not installed. Automation features will be limited.") pyautogui = None try: import win32gui import win32con import win32api except ImportError: print("pywin32 not installed. Windows-specific features will be limited.") win32gui = None try: import tkinter as tk import tkinter.messagebox as messagebox from tkinter import ttk except ImportError: print("tkinter not available. GUI features will be limited.") tk = None messagebox = None ttk = None # 화면 해상도 정보를 위한 모듈 try: from screeninfo import get_monitors except ImportError: print("screeninfo not installed. Using default screen size.") get_monitors = None # 시스템 트레이 아이콘을 위한 모듈 try: import pystray from pystray import MenuItem as item except ImportError: print("pystray not installed. System tray features will be limited.") pystray = None item = None class GeoMedicalClient: """ 메인 클라이언트 클래스 서버와 통신하여 코드를 다운로드하고 실행합니다. """ def __init__(self): # 서버 설정 self.server_url = "https://file.geomedical.kr/file/GEOmedical/update/help" self.api_url = "http://192.168.0.240:8800" # API 서버 # 파일 경로 설정 self.local_version_file = "local_version.json" # 현재 버전 정보 self.local_code_file = "main.py" # 메인 코드 self.app_data_file = "app_data.json" # 애플리케이션 데이터 # 디렉토리 설정 self.downloads_dir = "downloads" # 다운로드 파일 저장 self.cache_dir = "cache" # 이미지 캐시 self.modules_dir = "modules" # 추가 모듈 self.plugins_dir = "plugins" # 플러그인 # 필요한 디렉토리 생성 for dir_path in [self.downloads_dir, self.cache_dir, self.modules_dir, self.plugins_dir]: os.makedirs(dir_path, exist_ok=True) # 런타임 변수들 self.current_module = None # 현재 로드된 모듈 self.gui_root = None # tkinter 루트 윈도우 self.overlay_windows = [] # 오버레이 윈도우 목록 self.background_threads = [] # 백그라운드 스레드 목록 self.hotkeys = {} # 등록된 핫키 목록 self.active_popup = None # 현재 활성화된 관리 요청 팝업 self.processed_requests = set() # 처리된 요청 ID 목록 # 초기화 작업 self.check_dependencies() # 필요한 라이브러리 확인 self.load_plugins() # 플러그인 로드 self.sync_config() # 서버 설정 동기화 # 관리 요청 확인을 위한 별도 스레드 시작 self.management_check_running = False # 시스템 트레이 아이콘 self.tray_icon = None self.is_running = True # 업데이트 상태 추적 (무한 업데이트 방지) self._update_in_progress = False self._last_update_request_id = None self._last_app_update_time = 0 # 시작 시 업데이트 상태 초기화 (새로 시작된 프로세스) self.reset_update_state() # ===== 업데이트 상태 관리 ===== def reset_update_state(self): """업데이트 상태 초기화 (새 프로세스 시작 시)""" print("[INIT] Resetting update state for new process") self._update_in_progress = False self._last_update_request_id = None # 마지막 업데이트 시간은 유지 (쿨다운 효과 지속) print(f"[INIT] Update state reset complete") # ===== 버전 관리 ===== def get_local_version(self): """app.exe 버전 정보를 반환합니다. (코드 내 관리)""" return { 'version': APP_VERSION, 'build_date': APP_BUILD_DATE, 'description': APP_DESCRIPTION, 'hash': self.calculate_hash(APP_VERSION + APP_BUILD_DATE) } def get_main_version(self): """main.py 파일에서 직접 MAIN_VERSION 변수를 파싱하여 버전 정보를 반환합니다.""" try: if os.path.exists(self.local_code_file): with open(self.local_code_file, 'r', encoding='utf-8') as f: content = f.read() # MAIN_VERSION 변수 파싱 import re version_match = re.search(r'MAIN_VERSION\s*=\s*["\']([^"\']+)["\']', content) build_date_match = re.search(r'MAIN_BUILD_DATE\s*=\s*["\']([^"\']+)["\']', content) if version_match: version = version_match.group(1) build_date = build_date_match.group(1) if build_date_match else "Unknown" return { 'version': version, 'build_date': build_date, 'hash': self.calculate_hash(version + build_date) } except Exception as e: print(f"[DEBUG] Error parsing main.py version: {e}") # 기본값 반환 return {'version': '1.0.0', 'build_date': 'Unknown', 'hash': ''} def save_main_version(self, version_data): """main.py 버전 정보를 로컬에 저장합니다.""" with open(self.local_version_file, 'w') as f: json.dump(version_data, f, indent=2) # ===== 데이터 저장/로드 ===== def save_data(self, key, value): """ app_data.json에 데이터를 저장합니다. key: 저장할 데이터의 키 value: 저장할 값 (JSON 직렬화 가능해야 함) """ data = {} if os.path.exists(self.app_data_file): with open(self.app_data_file, 'r', encoding='utf-8') as f: data = json.load(f) data[key] = value with open(self.app_data_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) def load_data(self, key, default=None): """ app_data.json에서 데이터를 불러옵니다. key: 불러올 데이터의 키 default: 키가 없을 때 반환할 기본값 """ if os.path.exists(self.app_data_file): with open(self.app_data_file, 'r', encoding='utf-8') as f: data = json.load(f) return data.get(key, default) return default def get_all_data(self): """저장된 모든 데이터를 반환합니다.""" if os.path.exists(self.app_data_file): with open(self.app_data_file, 'r', encoding='utf-8') as f: return json.load(f) return {} def delete_data(self, key): """특정 키의 데이터를 삭제합니다.""" if os.path.exists(self.app_data_file): with open(self.app_data_file, 'r', encoding='utf-8') as f: data = json.load(f) if key in data: del data[key] with open(self.app_data_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) # ===== 파일 다운로드 ===== def download_file(self, url, filename=None, progress_callback=None): """ 파일을 다운로드합니다. url: 다운로드할 파일의 URL filename: 저장할 파일명 (None이면 URL에서 추출) progress_callback: 진행상황 콜백 함수 (downloaded, total) """ try: if not filename: filename = os.path.basename(urlparse(url).path) if not filename: filename = 'download_' + str(int(time.time())) filepath = os.path.join(self.downloads_dir, filename) response = requests.get(url, stream=True) total_size = int(response.headers.get('content-length', 0)) downloaded = 0 with open(filepath, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) downloaded += len(chunk) if progress_callback and total_size > 0: progress_callback(downloaded, total_size) return filepath except Exception as e: print(f"Download error: {e}") return None def download_image(self, url, cache=True): """ 이미지를 다운로드하고 캐싱합니다. url: 이미지 URL cache: 캐시 사용 여부 """ try: # URL을 해시하여 캐시 파일명 생성 cache_filename = hashlib.md5(url.encode()).hexdigest() + '.jpg' cache_path = os.path.join(self.cache_dir, cache_filename) # 캐시 확인 if cache and os.path.exists(cache_path): return cache_path # 다운로드 response = requests.get(url, timeout=10) if response.status_code == 200: with open(cache_path, 'wb') as f: f.write(response.content) return cache_path except Exception as e: print(f"Image download error: {e}") return None # ===== 화면 캡처 ===== def capture_screen(self, region=None): """ 화면을 캡처합니다. region: 캡처할 영역 (x, y, width, height) 튜플 """ if pyautogui: try: if region: screenshot = pyautogui.screenshot(region=region) else: screenshot = pyautogui.screenshot() return screenshot except Exception as e: print(f"Screenshot error: {e}") return None # ===== 핫키 관리 ===== def register_hotkey(self, key_combination, callback): """ 글로벌 핫키를 등록합니다. key_combination: 키 조합 (예: 'ctrl+space') callback: 핫키가 눌렸을 때 실행할 함수 """ if keyboard: try: keyboard.add_hotkey(key_combination, callback) self.hotkeys[key_combination] = callback return True except Exception as e: print(f"Hotkey registration error: {e}") return False def unregister_hotkey(self, key_combination): """등록된 핫키를 해제합니다.""" if keyboard and key_combination in self.hotkeys: try: keyboard.remove_hotkey(key_combination) del self.hotkeys[key_combination] return True except Exception as e: print(f"Hotkey removal error: {e}") return False # ===== 오버레이 윈도우 ===== def create_overlay_window(self): """ 투명한 전체화면 오버레이 윈도우를 생성합니다. 파이 메뉴 등에 사용할 수 있습니다. """ import tkinter as tk overlay = tk.Toplevel() overlay.attributes('-alpha', 0.7) # 70% 투명도 overlay.attributes('-topmost', True) # 항상 위 overlay.overrideredirect(True) # 테두리 없음 # 전체 화면 크기 overlay.geometry(f"{overlay.winfo_screenwidth()}x{overlay.winfo_screenheight()}+0+0") # 투명 배경 overlay.configure(bg='black') # 클릭 통과 설정 (Windows) if win32gui: hwnd = overlay.winfo_id() styles = win32gui.GetWindowLong(hwnd, win32con.GWL_EXSTYLE) styles = styles | win32con.WS_EX_LAYERED | win32con.WS_EX_TRANSPARENT win32gui.SetWindowLong(hwnd, win32con.GWL_EXSTYLE, styles) self.overlay_windows.append(overlay) return overlay # ===== 백그라운드 작업 ===== def run_in_background(self, func, *args, **kwargs): """ 함수를 백그라운드 스레드에서 실행합니다. func: 실행할 함수 args, kwargs: 함수에 전달할 인자 """ thread = threading.Thread(target=func, args=args, kwargs=kwargs, daemon=True) thread.start() self.background_threads.append(thread) return thread # ===== 외부 서비스 연동 ===== def send_kakaowork_message(self, webhook_url, message): """ 카카오워크 웹훅으로 메시지를 전송합니다. webhook_url: 카카오워크 웹훅 URL message: 전송할 메시지 """ try: headers = {'Content-Type': 'application/json'} data = {'text': message} response = requests.post(webhook_url, json=data, headers=headers) return response.status_code == 200 except Exception as e: print(f"KakaoWork message error: {e}") return False # ===== 모듈 관리 ===== def download_additional_module(self, module_name): """서버에서 추가 모듈을 다운로드합니다.""" try: module_url = f"{self.server_url}/modules/{module_name}.py" response = requests.get(module_url, timeout=10) if response.status_code == 200: module_path = os.path.join(self.modules_dir, f"{module_name}.py") with open(module_path, 'w', encoding='utf-8') as f: f.write(response.text) return module_path except Exception as e: print(f"Module download error: {e}") return None def load_additional_module(self, module_name): """추가 모듈을 로드합니다.""" module_path = os.path.join(self.modules_dir, f"{module_name}.py") if not os.path.exists(module_path): module_path = self.download_additional_module(module_name) if module_path and os.path.exists(module_path): try: spec = importlib.util.spec_from_file_location(module_name, module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module except Exception as e: print(f"Module load error: {e}") return None # ===== 업데이트 관리 ===== def check_update(self): """서버에서 최신 버전 정보를 확인합니다. (400대 대응 재시도 로직)""" for attempt in range(3): # 최대 3회 재시도 try: response = requests.get(f"{self.server_url}/version.json", timeout=15, # 타임아웃 증가 headers={'Connection': 'close'}) # 연결 재사용 방지 if response.status_code == 200: return response.json() except Exception as e: if attempt < 2: # 마지막 시도가 아니면 time.sleep(2 ** attempt) # 지수 백오프 continue print(f"Version check error after {attempt + 1} attempts: {e}") return None def download_code(self, filename='main.py'): """서버에서 코드를 다운로드합니다.""" try: response = requests.get(f"{self.server_url}/{filename}", timeout=10) if response.status_code == 200: return response.text except Exception as e: print(f"Download error: {e}") return None def save_local_code(self, code): """코드를 로컬에 저장합니다.""" with open(self.local_code_file, 'w', encoding='utf-8') as f: f.write(code) def load_local_code(self): """로컬에 저장된 코드를 불러옵니다.""" if os.path.exists(self.local_code_file): with open(self.local_code_file, 'r', encoding='utf-8') as f: return f.read() return None def calculate_hash(self, code): """코드의 SHA256 해시를 계산합니다.""" return hashlib.sha256(code.encode()).hexdigest() def calculate_file_hash(self, filepath): """파일의 SHA256 해시를 계산합니다.""" sha256_hash = hashlib.sha256() try: with open(filepath, "rb") as f: for byte_block in iter(lambda: f.read(4096), b""): sha256_hash.update(byte_block) return sha256_hash.hexdigest() except: return "unknown" # ===== 유틸리티 함수 ===== def check_dependencies(self): """필요한 라이브러리를 확인하고 자동 설치를 시도합니다.""" required = { 'pillow': 'PIL', 'keyboard': 'keyboard', 'pyautogui': 'pyautogui', 'pywin32': 'win32gui' } missing = [] for package, import_name in required.items(): try: __import__(import_name) except ImportError: missing.append(package) if missing and self.load_data('auto_install_deps', False): print(f"Installing missing libraries: {missing}") try: subprocess.check_call([sys.executable, '-m', 'pip', 'install'] + missing) except: print("Failed to install dependencies automatically") def load_plugins(self): """plugins 디렉토리의 모든 플러그인을 로드합니다.""" if os.path.exists(self.plugins_dir): for filename in os.listdir(self.plugins_dir): if filename.endswith('.py') and not filename.startswith('_'): plugin_name = filename[:-3] try: self.load_additional_module(plugin_name) print(f"Loaded plugin: {plugin_name}") except Exception as e: print(f"Failed to load plugin {plugin_name}: {e}") def sync_config(self): """서버의 설정 파일과 동기화합니다.""" try: response = requests.get(f"{self.server_url}/config.json", timeout=5) if response.status_code == 200: config = response.json() self.save_data('server_config', config) return config except: pass return self.load_data('server_config', {}) def report_error(self, error, context=""): """ 에러를 로컬에 기록하고 선택적으로 서버에 전송합니다. error: 발생한 에러 context: 에러가 발생한 상황 설명 """ try: error_data = { 'timestamp': datetime.now().isoformat(), 'version': self.get_local_version()['version'], 'error': str(error), 'traceback': traceback.format_exc(), 'context': context } # 로컬에 저장 error_log = self.load_data('error_log', []) error_log.append(error_data) self.save_data('error_log', error_log[-100:]) # 최근 100개만 유지 # 서버로 전송 (설정에 따라) if self.load_data('send_error_reports', False): try: requests.post(f"{self.server_url}/error_report", json=error_data, timeout=5) except: pass except: pass def send_log(self, level, log_type, message, details=None): """ 서버로 로그를 전송합니다. level: 'ERROR', 'WARNING', 'INFO', 'DEBUG' log_type: 'SYSTEM', 'NETWORK', 'UPDATE', 'GUI', 'CUSTOM' message: 로그 메시지 details: 추가 상세 정보 (선택적) """ try: # 호스트명 가져오기 import os hostname = os.environ.get('COMPUTERNAME', 'Unknown') # 로그 데이터 구성 log_data = { 'hostname': hostname, 'level': level, 'type': log_type, 'message': message, 'details': details } # 서버로 전송 (5초 타임아웃) response = requests.post( f"{self.api_url}/api/agent/log", json=log_data, timeout=5, headers={'Content-Type': 'application/json'} ) if response.status_code == 200: return True else: print(f"Log send failed: {response.status_code}") return False except Exception as e: # 로그 전송 실패는 조용히 처리 (무한 루프 방지) print(f"Log send error: {e}") return False # ===== 모듈 로드 ===== def load_module(self, code): """ 다운로드한 코드를 모듈로 로드하고 필요한 함수들을 주입합니다. main.py에서 사용할 수 있는 모든 기능이 여기서 정의됩니다. """ spec = importlib.util.spec_from_loader('dynamic_main', loader=None) module = importlib.util.module_from_spec(spec) # ===== 기본 유틸리티 함수 ===== module.get_root = lambda: self.gui_root module.set_root = lambda root: setattr(self, 'gui_root', root) module.get_app_path = lambda: os.path.dirname(os.path.abspath(__file__)) module.get_version = lambda: '1.2.0' # main.py 버전 반환 module.get_downloads_dir = lambda: self.downloads_dir module.get_cache_dir = lambda: self.cache_dir # ===== 데이터 관리 함수 ===== module.save_data = self.save_data module.load_data = self.load_data module.get_all_data = self.get_all_data module.delete_data = self.delete_data # ===== 파일 다운로드 함수 ===== module.download_file = self.download_file module.download_image = self.download_image # ===== 화면 캡처 함수 ===== module.capture_screen = self.capture_screen # ===== 핫키 관리 함수 ===== module.register_hotkey = self.register_hotkey module.unregister_hotkey = self.unregister_hotkey # ===== UI 관련 함수 ===== module.create_overlay_window = self.create_overlay_window # ===== 백그라운드 작업 함수 ===== module.run_in_background = self.run_in_background # ===== 외부 서비스 연동 함수 ===== module.send_kakaowork_message = self.send_kakaowork_message # ===== 모듈 관리 함수 ===== module.load_additional_module = self.load_additional_module # ===== 유틸리티 함수 ===== module.report_error = self.report_error module.sync_config = self.sync_config module.send_log = self.send_log # ===== 표준 라이브러리 ===== module.json = json module.os = os module.time = time module.datetime = datetime module.threading = threading module.requests = requests module.base64 = base64 module.io = io module.subprocess = subprocess module.hashlib = hashlib module.math = __import__('math') module.random = __import__('random') module.re = __import__('re') # ===== 선택적 라이브러리 ===== if Image: module.Image = Image module.ImageTk = ImageTk if keyboard: module.keyboard = keyboard if mouse: module.mouse = mouse if pyautogui: module.pyautogui = pyautogui if win32gui: module.win32gui = win32gui module.win32con = win32con module.win32api = win32api # 표준 라이브러리 모듈들 주입 import sqlite3 import logging import webbrowser import math import re import platform import ctypes module.sqlite3 = sqlite3 module.logging = logging module.webbrowser = webbrowser module.math = math module.re = re module.platform = platform module.ctypes = ctypes # 코드 실행 시 오류 위치 표시 try: exec(code, module.__dict__) except SyntaxError as e: # 구문 오류 발생 시 주변 코드 표시 lines = code.split('\n') error_line = e.lineno - 1 if e.lineno else 0 print(f"\n{'='*60}") print(f"Syntax Error at line {e.lineno}") print(f"Error: {e.msg}") # 오류 위치 주변 10줄 표시 start = max(0, error_line - 10) end = min(len(lines), error_line + 11) print(f"\nCode around line {e.lineno}:") print("-" * 60) for i in range(start, end): line_num = i + 1 line_content = lines[i] if i < len(lines) else "" if i == error_line: print(f">>> {line_num:5d} | {line_content}") if e.offset: print(f" | {' ' * (e.offset - 1)}^") else: print(f" {line_num:5d} | {line_content}") print("=" * 60) # 파일에 문제가 있는 코드 저장 (디버깅용) debug_file = os.path.join(self.cache_dir, "error_code.py") with open(debug_file, 'w', encoding='utf-8') as f: f.write(code) print(f"Error code saved to: {debug_file}") raise except Exception as e: print(f"Runtime error in module: {type(e).__name__}: {e}") traceback.print_exc() raise return module def create_initial_mainpy(self): """초기 main.py 버전을 생성합니다. (사용 안함 - 원래 동작은 백그라운드 실행)""" # 이 함수는 사용하지 않습니다. 원래 app.exe는 main.py 없이 백그라운드에서 실행되어야 합니다. return None def create_exec_globals(self): """main.py 실행을 위한 완전한 글로벌 네임스페이스를 생성합니다.""" exec_globals = { '__name__': '__main__', # Basic Utility Functions 'get_root': lambda: getattr(self, 'gui_root', None), 'set_root': lambda root: setattr(self, 'gui_root', root), 'get_app_path': lambda: os.path.dirname(os.path.abspath(__file__)), 'get_version': lambda: '1.2.0', # main.py의 버전 반환 (cached_main.py에서 직접 관리) 'get_downloads_dir': lambda: getattr(self, 'downloads_dir', 'downloads'), 'get_cache_dir': lambda: getattr(self, 'cache_dir', 'cache'), # Data Management Functions 'save_data': getattr(self, 'save_data', lambda k, v: None), 'load_data': getattr(self, 'load_data', lambda k, d=None: d), 'get_all_data': getattr(self, 'get_all_data', lambda: {}), 'delete_data': getattr(self, 'delete_data', lambda k: None), # File Operations - with safe defaults 'download_file': getattr(self, 'download_file', lambda url, filename, callback=None: None), 'download_image': getattr(self, 'download_image', lambda url, cache=True: None), # System Functions - with safe defaults 'capture_screen': getattr(self, 'capture_screen', lambda region=None: None), 'register_hotkey': getattr(self, 'register_hotkey', lambda combo, callback: None), 'unregister_hotkey': getattr(self, 'unregister_hotkey', lambda combo: None), 'create_overlay_window': getattr(self, 'create_overlay_window', lambda: None), 'run_in_background': getattr(self, 'run_in_background', lambda func, *args, **kwargs: None), # External Service Functions - with safe defaults 'send_kakaowork_message': getattr(self, 'send_kakaowork_message', lambda url, msg: None), 'load_additional_module': getattr(self, 'load_additional_module', lambda name: None), # Utility Functions 'report_error': getattr(self, 'report_error', lambda error, context='': print(f"Error: {error}")), 'sync_config': getattr(self, 'sync_config', lambda: None), 'send_log': getattr(self, 'send_log', lambda level, log_type, message, details=None: None), # Standard Libraries 'json': json, 'os': os, 'time': time, 'datetime': datetime, 'threading': threading, 'requests': requests, 'base64': base64, 'io': io, 'subprocess': subprocess, 'hashlib': hashlib, 'math': __import__('math'), 'random': __import__('random'), 're': __import__('re'), 'sqlite3': __import__('sqlite3'), 'logging': __import__('logging'), 'webbrowser': __import__('webbrowser'), 'platform': __import__('platform'), 'ctypes': __import__('ctypes'), } # Add optional libraries if available try: if Image: exec_globals['Image'] = Image if ImageTk: exec_globals['ImageTk'] = ImageTk except: pass try: if keyboard: exec_globals['keyboard'] = keyboard except: pass try: if mouse: exec_globals['mouse'] = mouse except: pass try: if pyautogui: exec_globals['pyautogui'] = pyautogui except: pass try: if win32gui: exec_globals['win32gui'] = win32gui exec_globals['win32con'] = win32con exec_globals['win32api'] = win32api except: pass return exec_globals def create_offline_code(self): """서버에 연결할 수 없을 때 사용할 기본 코드입니다.""" return ''' import tkinter as tk from datetime import datetime def main(): root = tk.Tk() set_root(root) root.title("GEOmedical Helper - Offline") root.geometry("400x300") tk.Label(root, text="GEOmedical Support Tool", font=("Arial", 16)).pack(pady=20) tk.Label(root, text="Running in offline mode", fg="red").pack(pady=10) tk.Label(root, text=f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}").pack(pady=10) tk.Button(root, text="Close", command=root.quit).pack(pady=20) root.mainloop() ''' def run_standalone_mode(self): """main.py 없이 app.exe 자체 기능만 실행합니다.""" print("=== GeoMedical Client - Standalone Mode ===") print("App.exe is running without main.py") # main.py가 있는지 확인하고 실행 main_file = "main.py" if os.path.exists(main_file): print("Found main.py, loading for mouse wheel functionality...") try: with open(main_file, 'r', encoding='utf-8') as f: code = f.read() if code: print("Executing main.py with mouse wheel support...") # main.py에서 필요한 모든 함수들을 글로벌 네임스페이스에 제공 exec_globals = self.create_exec_globals() # exec를 별도 스레드에서 실행하여 메인 스레드 블로킹 방지 import threading import time def execute_main(): try: exec(code, exec_globals) except Exception as e: print(f"Error in main.py execution thread: {e}") import traceback traceback.print_exc() exec_thread = threading.Thread(target=execute_main, daemon=True) exec_thread.start() print("Started main.py execution in background thread") # 짧은 대기 후 스레드가 시작되었는지 확인 time.sleep(0.5) if exec_thread.is_alive(): print("Main.py is running with mouse wheel support!") return True else: print("Warning: Main.py thread finished quickly") else: print("Failed to load main.py content") except Exception as e: print(f"Error executing main.py: {e}") import traceback traceback.print_exc() # main.py가 없으면 서버에서 다운로드 시도 print("No main.py found, attempting to download from server...") try: code = self.download_code() if code: print("Successfully downloaded main.py from server") # main.py로 저장 with open(main_file, 'w', encoding='utf-8') as f: f.write(code) print("Saved as main.py") # 파일 저장 확인을 위한 잠시 대기 import time time.sleep(0.1) # 파일이 실제로 저장되었는지 확인 if os.path.exists(main_file) and os.path.getsize(main_file) > 0: print(f"Verified main.py file exists ({os.path.getsize(main_file)} bytes)") # 다운로드한 코드 실행 print("Executing downloaded main.py with mouse wheel support...") # main.py에서 필요한 모든 함수들을 글로벌 네임스페이스에 제공 exec_globals = self.create_exec_globals() # exec를 별도 스레드에서 실행하여 메인 스레드 블로킹 방지 import threading def execute_main(): try: exec(code, exec_globals) except Exception as e: print(f"Error in main.py execution thread: {e}") import traceback traceback.print_exc() exec_thread = threading.Thread(target=execute_main, daemon=True) exec_thread.start() print("Started main.py execution in background thread") # 짧은 대기 후 스레드가 시작되었는지 확인 time.sleep(0.5) if exec_thread.is_alive(): print("Main.py is running with mouse wheel support!") return True else: print("Warning: Main.py thread finished quickly") else: print("Error: main.py file was not properly saved") else: print("Failed to download main.py from server") except Exception as e: print(f"Error downloading main.py: {e}") import traceback traceback.print_exc() print("No main.py available and server download failed") print("Available functions:") print("- Asset monitoring and reporting") print("- Management request processing") print("- App.exe and main.py updates via dashboard") print("=" * 50) # 관리 요청 확인 스레드 시작 self.start_management_request_checker() # 에셋 모니터링 시작 (이미 백그라운드에서 실행 중) # 백그라운드 모드로 실행 (GUI 없음) self.background_mode() return True # ===== 메인 실행 로직 ===== def update_and_run(self, auto_update_mainpy=True): """main.py 업데이트를 확인하고 최신 코드를 실행합니다.""" print(f"Connecting to: {self.server_url}") print("NOTE: App.exe updates are handled separately via management requests") main_version = self.get_main_version() # main.py 버전 정보 code = None # main.py 자동 업데이트 비활성화 옵션 if not auto_update_mainpy: print("INFO: main.py auto-update is disabled") code = self.load_local_code() if not code: print("No local main.py found, running in standalone mode...") print("Will attempt to download main.py from server for mouse wheel functionality") return self.run_standalone_mode() else: remote_version = self.check_update() if remote_version: print(f"Server main.py version: {remote_version['version']}") print(f"Local main.py version: {main_version['version']}") # main.py만 자동 업데이트 (app.exe는 관리 요청을 통해서만) if remote_version['version'] > main_version['version']: print("Downloading main.py update...") code = self.download_code() if code: code_hash = self.calculate_hash(code) remote_version['hash'] = code_hash self.save_local_code(code) self.save_main_version(remote_version) print(f"Updated main.py to version {remote_version['version']}") else: print("main.py is up to date") else: print("Server not reachable, using local main.py version") # 로컬 코드 사용 if not code: code = self.load_local_code() if not code: print("No local code found, using offline mode...") code = self.create_offline_code() self.save_local_code(code) try: self.current_module = self.load_module(code) return True except Exception as e: print(f"Module load error: {e}") traceback.print_exc() self.report_error(e, "Module loading") return False def restart_gui(self): """GUI를 재시작합니다. 업데이트 후 호출됩니다.""" if self.gui_root: print("Restarting GUI...") try: self.gui_root.quit() self.gui_root.destroy() except: pass self.gui_root = None # 오버레이 윈도우 정리 for overlay in self.overlay_windows: try: overlay.destroy() except: pass self.overlay_windows.clear() # 코드를 다시 로드하고 실행 code = self.load_local_code() if code: try: self.current_module = self.load_module(code) if hasattr(self.current_module, 'main'): self.current_module.main() except Exception as e: print(f"GUI restart error: {e}") traceback.print_exc() self.report_error(e, "GUI restart") def check_for_management_requests(self): """서버에서 소프트웨어 관리 요청을 확인합니다.""" try: hostname = os.environ.get('COMPUTERNAME', 'Unknown') print(f"[DEBUG] Checking management requests for {hostname}") management_url = f"{self.api_url}/api/software/management-requests/{hostname}" print(f"[DEBUG] Requesting management tasks from: {management_url}") response = requests.get(management_url, timeout=10) print(f"[DEBUG] Management request response: {response.status_code}") if response.status_code == 200: try: data = response.json() print(f"[DEBUG] Response data: {data}") except json.JSONDecodeError as e: print(f"[ERROR] Failed to parse JSON response: {e}") print(f"[ERROR] Response content: {response.text[:200]}") # 처음 200자만 출력 return if data.get('success') and data.get('requests'): print(f"[DEBUG] Found {len(data['requests'])} requests") for req in data['requests']: print(f"[DEBUG] Processing request ID: {req['id']}, Software: {req['software_name']}") # app.exe 업데이트 요청 처리 if req.get('action_type') == 'app_update': print(f"[DEBUG] App update request detected - ID: {req['id']}") # 중복 요청 체크 (강화된 중복 방지) if (req['id'] not in self.processed_requests and req['id'] != self._last_update_request_id and not self._update_in_progress): print(f"[DEBUG] Processing new app update request") self.processed_requests.add(req['id']) self._last_update_request_id = req['id'] self._update_in_progress = True # 즉시 요청 완료로 표시 (무한 업데이트 방지) self.report_management_completion(req['id'], 'completed') print(f"[DEBUG] App update request {req['id']} marked as completed immediately") # 업데이트 처리 시작 success = self.handle_app_update_request(req['id']) # 업데이트 실패시에만 플래그 해제 (성공시는 프로세스 종료됨) if not success: self._update_in_progress = False print(f"[DEBUG] App update failed, flags reset") else: print(f"[DEBUG] Duplicate app update request ignored - ID: {req['id']}") print(f" - Already processed: {req['id'] in self.processed_requests}") print(f" - Same as last: {req['id'] == self._last_update_request_id}") print(f" - Update in progress: {self._update_in_progress}") # 중복 요청을 완료로 처리하여 대시보드에서 제거 self.report_management_completion(req['id'], 'completed') # main.py 업데이트 요청 처리 elif req.get('action_type') == 'mainpy_update': print(f"[DEBUG] Main.py update request detected") if req['id'] not in self.processed_requests: self.processed_requests.add(req['id']) success = self.handle_mainpy_update_request() self.report_management_completion(req['id'], 'completed' if success else 'failed') # 업데이트 성공 시 상세 정보 로그 및 버전 정보 업데이트 if success: exe_path = os.path.abspath(sys.argv[0]) exe_size = os.path.getsize(exe_path) / 1024 / 1024 exe_hash = self.calculate_file_hash(exe_path)[:16] details = f"Request ID: {req['id']}, Size: {exe_size:.2f}MB, Hash: {exe_hash}" self.send_log('INFO', 'APP_UPDATE', 'App update completed successfully', details) # 업데이트된 app.exe의 버전 정보 저장 (무한 업데이트 방지) # app.exe 업데이트 완료 - 버전은 코드에서 관리됨 print(f"[APP_UPDATE] App.exe updated to version {APP_VERSION}") print(f"[APP_UPDATE] Update completed successfully") else: self.send_log('ERROR', 'APP_UPDATE', 'App update failed', f"Request ID: {req['id']}") continue # 일반 소프트웨어 관리 요청 # 이미 처리된 요청이거나 팝업이 활성화된 상태면 건너뛰기 if req['id'] not in self.processed_requests: if self.active_popup is None: print(f"[DEBUG] Creating popup for request: {req['id']}") self.handle_management_request(req) # 로그 전송 self.send_log('INFO', 'SYSTEM', f"Management request received: {req['software_name']}", f"Request ID: {req['id']}, Action: {req['action_type']}") else: print(f"[DEBUG] Skipping request {req['id']} - popup already active") else: print(f"[DEBUG] Skipping request {req['id']} - already processed") else: print(f"[DEBUG] No pending requests found") else: print(f"[WARNING] Management request check failed: HTTP {response.status_code}") self.send_log('WARNING', 'SYSTEM', f"Management request check failed: HTTP {response.status_code}") except Exception as e: print(f"Management request check error: {e}") self.send_log('ERROR', 'SYSTEM', f"Management request check error: {str(e)}") def start_management_request_checker(self): """관리 요청 확인을 위한 별도 스레드 시작""" if self.management_check_running: return self.management_check_running = True management_thread = threading.Thread(target=self._management_request_loop, daemon=True) management_thread.start() print("[INFO] Management request checker started (30초 간격)") # 로그 대시보드에는 표시하지 않음 (내부 시스템 로그) def _management_request_loop(self): """관리 요청 확인 루프 (30초마다)""" while self.management_check_running: try: # 관리 요청은 GUI 상태와 관계없이 항상 확인 print("[DEBUG] Checking for management requests...") self.check_for_management_requests() except Exception as e: print(f"[ERROR] Management request loop error: {e}") self.send_log('ERROR', 'SYSTEM', f"Management request loop error: {str(e)}") # 30초 대기 time.sleep(10) # 30초에서 10초로 단축 def create_popup_notification(self, title, message, action_type, request_id, software_name): """카카오스타일 팝업 알림 생성""" if not tk: print(f"GUI not available for notification: {title} - {message}") return # 이미 활성화된 팝업이 있으면 무시 if self.active_popup is not None: try: if self.active_popup.winfo_exists(): print("Popup already active, skipping new popup") return except: # 팝업이 더 이상 존재하지 않으면 None으로 설정 self.active_popup = None try: # 화면 해상도 가져오기 if get_monitors: screen = get_monitors()[0] screen_width = screen.width screen_height = screen.height else: # 기본값 사용 screen_width = 1920 screen_height = 1080 # 팝업 윈도우 생성 popup = tk.Toplevel() popup.overrideredirect(True) popup.attributes("-topmost", True) popup.configure(bg="#ffffff") # 현재 활성화된 팝업으로 설정 self.active_popup = popup # 그림자 효과를 위한 프레임 shadow_frame = tk.Frame(popup, bg="#cccccc") shadow_frame.place(x=2, y=2, relwidth=1, relheight=1) # 메인 프레임 main_frame = tk.Frame(popup, bg="#ffffff", relief="solid", bd=1) main_frame.place(x=0, y=0, relwidth=1, relheight=1) # 크기 설정 (높이를 늘려서 텍스트가 잘리지 않도록) width = 380 height = 200 # 위치 설정 (오른쪽 하단) x = screen_width - width - 20 y = screen_height - height - 60 # 작업표시줄 높이 고려 popup.geometry(f"{width}x{height}+{x}+{y}") # 제목 라벨 title_label = tk.Label( main_frame, text=title, font=("맑은 고딕", 12, "bold"), bg="#ffffff", fg="#333333", anchor="w" ) title_label.pack(padx=15, pady=(15, 5), fill='x') # 메시지 컨테이너 프레임 (충분한 높이 확보) message_container = tk.Frame(main_frame, bg="#ffffff") message_container.pack(padx=15, pady=5, fill='both', expand=True) # 메시지 라벨 (높이를 충분히 확보) message_label = tk.Label( message_container, text=message, font=("맑은 고딕", 10), bg="#ffffff", fg="#666666", anchor="nw", # 북서쪽 정렬로 상단부터 표시 justify="left", wraplength=340, # 텍스트 래핑 너비 height=6 # 명시적 높이 설정 (6줄) ) message_label.pack(fill='both', expand=True) def on_close(): self.processed_requests.add(request_id) # 처리된 요청에 추가 self.active_popup = None # 활성 팝업 해제 popup.destroy() # 메시지만 표시하므로 자동으로 알림으로 처리 if action_type == 'deletion_request': self.report_management_completion(request_id, 'completed') elif action_type == 'force_removal': self.report_management_completion(request_id, 'completed') # 팝업이 닫힐 때 active_popup을 None으로 설정 def on_destroy(): self.active_popup = None popup.protocol("WM_DELETE_WINDOW", on_destroy) # 팝업 영역 클릭으로 닫기 (버튼 제외) popup.bind("", lambda e: on_close()) main_frame.bind("", lambda e: on_close()) shadow_frame.bind("", lambda e: on_close()) message_container.bind("", lambda e: on_close()) message_label.bind("", lambda e: on_close()) title_label.bind("", lambda e: on_close()) # 애니메이션 효과 (아래에서 위로 슬라이드) start_y = screen_height for step in range(0, height + 60, 8): current_y = start_y - step popup.geometry(f"{width}x{height}+{x}+{current_y}") popup.update() time.sleep(0.01) except Exception as e: print(f"Popup creation error: {e}") self.active_popup = None def show_result_notification(self, title, message, color): """결과 알림 팝업""" if not tk: return try: # 화면 해상도 가져오기 if get_monitors: screen = get_monitors()[0] screen_width = screen.width screen_height = screen.height else: screen_width = 1920 screen_height = 1080 result_popup = tk.Toplevel() result_popup.overrideredirect(True) result_popup.attributes("-topmost", True) result_popup.configure(bg=color) width = 300 height = 80 x = screen_width - width - 20 y = screen_height - height - 60 result_popup.geometry(f"{width}x{height}+{x}+{y}") # 제목 tk.Label( result_popup, text=title, font=("맑은 고딕", 11, "bold"), bg=color, fg="white" ).pack(pady=(10, 2)) # 메시지 tk.Label( result_popup, text=message, font=("맑은 고딕", 9), bg=color, fg="white", wraplength=280 ).pack(pady=(0, 10)) # 클릭으로 닫기 def close_result(): result_popup.destroy() result_popup.bind("", lambda e: close_result()) # 3초 후 자동 닫기 result_popup.after(3000, close_result) # 애니메이션 start_y = screen_height for step in range(0, height + 60, 10): current_y = start_y - step result_popup.geometry(f"{width}x{height}+{x}+{current_y}") result_popup.update() time.sleep(0.01) except Exception as e: print(f"Result notification error: {e}") def handle_management_request(self, request): """관리 요청을 처리합니다.""" if not tk: print(f"GUI not available for request: {request}") return try: action_type = request['action_type'] software_name = request['software_name'] request_id = request['id'] if action_type == 'deletion_request': # 삭제 요청 알림 메시지 title = "🗑️ 소프트웨어 삭제 알림" message = f"관리자가 다음 소프트웨어의 삭제를 요청했습니다:\n\n📦 {software_name}\n\n이 알림을 확인했습니다." elif action_type == 'force_removal': # 강제 제거 알림 title = "⚠️ 소프트웨어 강제 제거 알림" message = f"관리자에 의해 다음 소프트웨어가 강제 제거됩니다:\n\n📦 {software_name}\n\n이 알림을 확인했습니다." # 팝업 알림 생성 self.create_popup_notification(title, message, action_type, request_id, software_name) except Exception as e: print(f"Management request handling error: {e}") self.report_management_completion(request.get('id'), 'failed') def report_management_completion(self, request_id, status): """관리 요청 완료를 서버에 보고""" try: response = requests.post( f"{self.api_url}/api/software/management-requests/{request_id}/complete", json={'result': status}, timeout=5 ) if response.status_code == 200: print(f"Management request {request_id} completed with status: {status}") except Exception as e: print(f"Failed to report management completion: {e}") def check_updates_periodically(self): """주기적으로 업데이트를 확인합니다.""" while True: time.sleep(3600) # 1시간(3600초)마다 확인 try: # GUI 상태를 안전하게 확인 gui_exists = False try: if self.gui_root and hasattr(self.gui_root, 'winfo_exists'): gui_exists = self.gui_root.winfo_exists() except: gui_exists = False if gui_exists: print(f"\n[자동 업데이트 확인 중...] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") remote_version = self.check_update() local_version = self.get_local_version() if remote_version and remote_version['version'] > local_version['version']: print(f"\n[Update Available] {local_version['version']} -> {remote_version['version']}") code = self.download_code() if code: self.save_local_code(code) self.save_main_version(remote_version) print("[Update Complete] Hot-reload successful") # 새 코드로 모듈 로드 new_module = self.load_module(code) if hasattr(new_module, 'on_update'): new_module.on_update() # GUI 재시작을 안전하게 스케줄링 try: if self.gui_root and hasattr(self.gui_root, 'after'): self.gui_root.after(100, self.restart_gui) except Exception as gui_e: print(f"[GUI Restart] Error: {gui_e}") else: # GUI가 없으면 업데이트 체크 스레드 종료 print("[Update Check] GUI not available, stopping update checks") break except Exception as e: if "invalid command name" not in str(e) and "main thread is not in main loop" not in str(e): print(f"[Update Check] Error: {e}") try: self.report_error(e, "Update check") except: pass # 에러 보고도 실패하면 조용히 무시 def cleanup(self): """프로그램 종료 시 정리 작업을 수행합니다.""" # 핫키 해제 for key in list(self.hotkeys.keys()): self.unregister_hotkey(key) # 오버레이 윈도우 닫기 for overlay in self.overlay_windows: try: overlay.destroy() except: pass print("Cleanup completed") # ===== 시스템 트레이 기능 ===== def create_tray_icon(self): """시스템 트레이 아이콘을 생성합니다.""" if not pystray: print("System tray not available (pystray not installed)") return None try: # 아이콘 이미지 생성 (간단한 16x16 이미지) if Image: # 파란색 원형 아이콘 생성 image = Image.new('RGBA', (64, 64), (0, 0, 0, 0)) # 간단한 파란색 원 그리기 from PIL import ImageDraw draw = ImageDraw.Draw(image) draw.ellipse([8, 8, 56, 56], fill=(70, 144, 226, 255), outline=(50, 100, 180, 255), width=2) # 중앙에 'GM' 텍스트 추가 try: # 기본 폰트 사용 draw.text((20, 22), "GM", fill=(255, 255, 255, 255)) except: pass # 폰트 에러 무시 else: # PIL이 없으면 기본 아이콘 사용 image = None # 트레이 메뉴 생성 menu = pystray.Menu( item("버전 정보", self.show_version_info), item("프로그램 종료", self.quit_application) ) # 트레이 아이콘 생성 icon = pystray.Icon( "GeoMedical Helper", image, "GeoMedical Helper Client", menu ) return icon except Exception as e: print(f"Failed to create tray icon: {e}") return None def show_version_info(self, icon=None, item=None): """버전 정보를 표시합니다.""" try: app_version = self.get_local_version() main_version = self.get_main_version() # 동적으로 main.py 버전 정보 가져오기 main_version_str = main_version.get('version', '알 수 없음') main_build_date = main_version.get('build_date', '알 수 없음') message = f"""GeoMedical Asset Management Client 📱 App.exe 버전: {app_version['version']} ({app_version['build_date']}) 📄 Main.py 버전: {main_version_str} ({main_build_date}) 🏗️ 설명: {app_version['description']} 🌐 서버: {self.server_url} 🔐 App Hash: {app_version['hash'][:8]} 🔐 Main Hash: {main_version['hash'][:8]} ✅ 상태: 실행 중""" if tk and messagebox: # GUI가 있으면 메시지박스 표시 root = tk.Tk() root.withdraw() # 메인 윈도우 숨김 messagebox.showinfo("버전 정보", message) root.destroy() else: # GUI가 없으면 콘솔에 출력 print("\n" + "="*50) print(message) print("="*50 + "\n") except Exception as e: print(f"Failed to show version info: {e}") def quit_application(self, icon=None, item=None): """프로그램을 종료합니다.""" print("Shutting down GeoMedical Helper...") try: # 관리 요청 체커 중지 self.management_check_running = False # 트레이 아이콘 중지 if self.tray_icon: self.tray_icon.stop() # GUI 정리 if self.gui_root: try: self.gui_root.quit() self.gui_root.destroy() except: pass # 정리 작업 self.cleanup() # 실행 상태 변경 self.is_running = False # 프로그램 종료 os._exit(0) except Exception as e: print(f"Error during shutdown: {e}") os._exit(1) def refresh_tray_menu(self): """시스템 트레이 메뉴를 새로고침합니다 (주로 main.py 업데이트 후 버전 정보 갱신).""" if not pystray or not self.tray_icon: print("[TRAY_REFRESH] System tray not available") return False try: print("[TRAY_REFRESH] Refreshing system tray menu...") # 방법 1: 메뉴 직접 업데이트 시도 try: new_menu = pystray.Menu( item("버전 정보", self.show_version_info), item("프로그램 종료", self.quit_application) ) self.tray_icon.menu = new_menu print("[TRAY_REFRESH] Direct menu update completed") return True except Exception as e1: print(f"[TRAY_REFRESH] Direct menu update failed: {e1}") # 방법 2: 트레이 아이콘 재시작 (확실한 방법) try: print("[TRAY_REFRESH] Attempting tray icon restart...") # 기존 트레이 아이콘 정지 (백그라운드에서) if self.tray_icon: def stop_tray(): try: self.tray_icon.stop() except: pass # 별도 스레드에서 정지 threading.Thread(target=stop_tray, daemon=True).start() time.sleep(0.5) # 정지 대기 # 새 트레이 아이콘 생성 및 시작 self.tray_icon = self.create_tray_icon() if self.tray_icon: tray_thread = threading.Thread(target=self.tray_icon.run, daemon=True) tray_thread.start() print("[TRAY_REFRESH] Tray icon restarted successfully") return True else: print("[TRAY_REFRESH] Failed to create new tray icon") return False except Exception as e2: print(f"[TRAY_REFRESH] Tray icon restart failed: {e2}") return False except Exception as e: print(f"[TRAY_REFRESH] Error refreshing tray menu: {e}") return False def start_tray_icon(self): """시스템 트레이 아이콘을 시작합니다.""" if not pystray: print("System tray not available - using console interface") self.start_console_interface() return try: self.tray_icon = self.create_tray_icon() if self.tray_icon: print("Starting system tray icon...") # 별도 스레드에서 트레이 아이콘 실행 tray_thread = threading.Thread(target=self.tray_icon.run, daemon=True) tray_thread.start() print("System tray icon started") else: print("Failed to create tray icon") self.start_console_interface() except Exception as e: print(f"Failed to start tray icon: {e}") self.start_console_interface() def start_console_interface(self): """콘솔 기반 인터페이스를 시작합니다 (시스템 트레이 대체).""" print("\n" + "="*60) print("🖥️ GeoMedical Helper - Console Interface") print("="*60) print("시스템 트레이를 사용할 수 없어 콘솔 인터페이스를 제공합니다.") print("\n사용 가능한 명령:") print(" 'version' 또는 'v' - 버전 정보 표시") print(" 'quit' 또는 'q' - 프로그램 종료") print(" 'help' 또는 'h' - 도움말 표시") print("\nCtrl+C를 눌러 언제든지 종료할 수 있습니다.") print("="*60 + "\n") def console_input_loop(): """콘솔 입력을 처리하는 루프""" while self.is_running: try: user_input = input("GeoMedical> ").strip().lower() if user_input in ['quit', 'q', 'exit']: print("프로그램을 종료합니다...") self.quit_application() break elif user_input in ['version', 'v']: self.show_version_info() elif user_input in ['help', 'h']: print("\n사용 가능한 명령:") print(" version, v - 버전 정보 표시") print(" quit, q - 프로그램 종료") print(" help, h - 도움말 표시") print() elif user_input == '': continue else: print(f"알 수 없는 명령: '{user_input}'. 'help'를 입력하세요.") except KeyboardInterrupt: print("\n\nCtrl+C 감지됨. 프로그램을 종료합니다...") self.quit_application() break except EOFError: # 입력 스트림이 닫힌 경우 (예: 파이프라인) break except Exception as e: print(f"입력 처리 오류: {e}") # 별도 스레드에서 콘솔 인터페이스 실행 console_thread = threading.Thread(target=console_input_loop, daemon=True) console_thread.start() print("Console interface started. Type 'help' for commands.") # ===== 앱 업데이트 기능 (관리 서버 요청 전용) ===== def handle_mainpy_update_request(self): """관리 서버로부터의 main.py 업데이트 요청을 처리합니다.""" try: print("[MAINPY_UPDATE] Processing main.py update request from management server...") # 업데이트 전 현재 버전 정보 확인 old_version = self.get_main_version() print(f"[MAINPY_UPDATE] Current version: {old_version['version']}") # 강제로 main.py 업데이트 수행 remote_version = self.check_update() if not remote_version: print("[MAINPY_UPDATE] Could not get server version info") self.send_log('ERROR', 'MAINPY_UPDATE', 'Update failed', 'Could not get server version info') return False print(f"[MAINPY_UPDATE] Downloading main.py version {remote_version['version']}") code = self.download_code() if code: code_hash = self.calculate_hash(code) remote_version['hash'] = code_hash self.save_local_code(code) self.save_main_version(remote_version) # 업데이트 후 새로운 버전 정보 확인 및 캐시 갱신 new_version = self.get_main_version() print(f"[MAINPY_UPDATE] Version updated: {old_version['version']} → {new_version['version']}") # 시스템 트레이 메뉴 갱신 (버전 정보 새로고침) if hasattr(self, 'refresh_tray_menu'): self.refresh_tray_menu() print("[MAINPY_UPDATE] System tray menu refreshed with new version") # 성공 로그 self.send_log('INFO', 'MAINPY_UPDATE', 'Update completed successfully', f"Version: {old_version['version']} → {new_version['version']}") print(f"[MAINPY_UPDATE] Successfully updated to version {new_version['version']}") return True else: print("[MAINPY_UPDATE] Failed to download main.py") self.send_log('ERROR', 'MAINPY_UPDATE', 'Update failed', 'Could not download main.py') return False except Exception as e: print(f"[MAINPY_UPDATE] Error: {e}") self.send_log('ERROR', 'MAINPY_UPDATE', 'Update failed', str(e)) return False def handle_app_update_request(self, request_id=None): """관리 서버로부터의 앱 업데이트 요청을 처리합니다.""" try: print(f"[APP_UPDATE] Processing app update request from management server (ID: {request_id})...") # 최근 업데이트 확인 (무한 업데이트 방지) # app.exe 버전은 코드에서 관리되므로 쿨다운만 체크 last_update = getattr(self, '_last_app_update_time', 0) time_since_update = time.time() - last_update # 쿨다운 시간을 10분으로 증가 (무한 업데이트 강력 방지) if time_since_update < 600: # 10분 이내 업데이트 방지 print(f"[APP_UPDATE] Skipping update - last update was {int(time_since_update)} seconds ago (cooldown: 600s)") self.send_log('INFO', 'APP_UPDATE', 'Update skipped - cooldown active', f'Request ID: {request_id}, Last update: {int(time_since_update)}s ago, Cooldown: 600s') return True # 새 exe 다운로드 시도 print("[APP_UPDATE] Checking for app.exe on server...") response = requests.get(f"{self.server_url}/app.exe", timeout=60) if response.status_code == 404: print("[APP_UPDATE] No app.exe found on server - update skipped") self.send_log('WARNING', 'APP_UPDATE', 'App update skipped', 'No app.exe file found on server') return True # 성공으로 처리 (파일이 없는 것은 정상 상황) elif response.status_code != 200: print(f"[APP_UPDATE] Server error: HTTP {response.status_code} - update cancelled") self.send_log('ERROR', 'APP_UPDATE', f'App update failed', f'Server returned HTTP {response.status_code}') return False # 파일 크기 검증 (최소 1MB 이상이어야 유효한 exe) content_length = len(response.content) if content_length < 1024 * 1024: # 1MB 미만 print(f"[APP_UPDATE] Downloaded file too small ({content_length} bytes) - possibly invalid") self.send_log('ERROR', 'APP_UPDATE', 'App update failed', f'Downloaded file too small: {content_length} bytes') return False print(f"[APP_UPDATE] New app.exe downloaded successfully ({content_length} bytes)") # 임시 파일로 저장 temp_exe = "app_new.exe" try: with open(temp_exe, 'wb') as f: f.write(response.content) # 파일 쓰기 검증 if not os.path.exists(temp_exe): print("[APP_UPDATE] Failed to save temporary file") self.send_log('ERROR', 'APP_UPDATE', 'App update failed', 'Failed to save temporary file') return False temp_size = os.path.getsize(temp_exe) if temp_size != content_length: print(f"[APP_UPDATE] File size mismatch: expected {content_length}, got {temp_size}") os.remove(temp_exe) # 잘못된 파일 삭제 self.send_log('ERROR', 'APP_UPDATE', 'App update failed', 'File size mismatch during save') return False # PE 헤더 검증 (Windows exe 파일 확인) try: with open(temp_exe, 'rb') as f: header = f.read(64) if len(header) < 64 or not header.startswith(b'MZ'): print("[APP_UPDATE] Downloaded file is not a valid Windows executable") os.remove(temp_exe) self.send_log('ERROR', 'APP_UPDATE', 'App update failed', 'Downloaded file is not a valid exe') return False print("[APP_UPDATE] Downloaded file PE header verified") except Exception as e: print(f"[APP_UPDATE] Error verifying downloaded file: {e}") os.remove(temp_exe) self.send_log('ERROR', 'APP_UPDATE', 'App update failed', f'File verification error: {str(e)}') return False print(f"[APP_UPDATE] Temporary file saved and verified: {temp_exe}") print(f"[APP_UPDATE] Current exe: {sys.executable if getattr(sys, 'frozen', False) else 'Not frozen'}") print(f"[APP_UPDATE] Working directory: {os.getcwd()}") # updater.exe 사용 시도 (실패 시 배치 스크립트로 fallback) print(f"[APP_UPDATE] Initiating update process...") # 쿨다운 시간 설정 (무한 업데이트 방지) self._last_app_update_time = time.time() self.use_updater_exe(temp_exe) return True except Exception as e: print(f"[APP_UPDATE] Failed to save temporary file: {e}") # 임시 파일 정리 if os.path.exists(temp_exe): try: os.remove(temp_exe) except: pass self.send_log('ERROR', 'APP_UPDATE', 'App update failed', f'File save error: {str(e)}') return False except requests.Timeout: print("[APP_UPDATE] Download timeout - update cancelled") self.send_log('ERROR', 'APP_UPDATE', 'App update failed', 'Download timeout') return False except requests.ConnectionError: print("[APP_UPDATE] Connection error - update cancelled") self.send_log('ERROR', 'APP_UPDATE', 'App update failed', 'Server connection error') return False except Exception as e: print(f"[APP_UPDATE] Unexpected error: {e}") self.send_log('ERROR', 'APP_UPDATE', 'App update failed', f'Unexpected error: {str(e)}') return False def use_updater_exe(self, new_exe_path): """updater.exe를 사용하여 업데이트 수행""" current_exe = sys.executable if getattr(sys, 'frozen', False) else "app.exe" current_pid = os.getpid() # plugins 폴더의 updater.exe 경로 if getattr(sys, 'frozen', False): # exe로 실행 중인 경우 - PyInstaller가 생성한 _internal 폴더 확인 app_dir = os.path.dirname(sys.executable) # PyInstaller로 번들된 경우 여러 경로 시도 possible_paths = [ os.path.join(app_dir, "plugins", "updater.exe"), os.path.join(app_dir, "_internal", "plugins", "updater.exe"), os.path.join(sys._MEIPASS, "plugins", "updater.exe") if hasattr(sys, '_MEIPASS') else None, ] updater_path = None for path in possible_paths: if path and os.path.exists(path): updater_path = path break if not updater_path: print(f"[APP_UPDATE] Searching for updater.exe in:") for path in possible_paths: if path: print(f" - {path}: {'EXISTS' if os.path.exists(path) else 'NOT FOUND'}") updater_path = possible_paths[0] # 기본값 else: # 스크립트로 실행 중인 경우 app_dir = os.path.dirname(os.path.abspath(__file__)) updater_path = os.path.join(app_dir, "plugins", "updater.exe") # updater.exe 존재 확인 if not os.path.exists(updater_path): print(f"[APP_UPDATE] Error: updater.exe not found at {updater_path}") print("[APP_UPDATE] Cannot proceed without updater.exe") self.send_log('ERROR', 'APP_UPDATE', 'Update failed', f'updater.exe not found at {updater_path}') return False print(f"[APP_UPDATE] Using updater.exe for update") print(f"[APP_UPDATE] Current PID: {current_pid}") print(f"[APP_UPDATE] Updater path: {updater_path}") # updater.exe 실행 명령 구성 (절대 경로 사용) updater_cmd = [ updater_path, "--pid", str(current_pid), "--old-exe", os.path.abspath(current_exe), "--new-exe", os.path.abspath(new_exe_path), "--auto-launch" # 업데이트 후 자동 실행 ] print(f"[APP_UPDATE] Launching updater: {' '.join(updater_cmd)}") try: # updater.exe 실행 (별도 프로세스) subprocess.Popen(updater_cmd, creationflags=subprocess.CREATE_NEW_CONSOLE) print("[APP_UPDATE] Updater launched successfully") print("[APP_UPDATE] Current process will exit in 3 seconds...") # 로그 전송 self.send_log('INFO', 'APP_UPDATE', 'Update initiated with updater.exe', f'PID: {current_pid}, New size: {os.path.getsize(new_exe_path)} bytes') # 1초 후 현재 프로세스 강제 종료 time.sleep(1) print("[APP_UPDATE] Exiting current process for update...") # 모든 스레드 정리 self.is_running = False self.management_check_running = False # GUI 종료 시도 if self.gui_root: try: self.gui_root.quit() except: pass # 시스템 트레이 종료 if self.tray_icon: try: self.tray_icon.stop() except: pass # 강제 종료 os._exit(0) # sys.exit() 대신 os._exit() 사용 except Exception as e: print(f"[APP_UPDATE] Error launching updater: {e}") self.send_log('ERROR', 'APP_UPDATE', 'Update failed', f'Error launching updater: {str(e)}') return False def background_mode(self): """백그라운드 모드에서 실행 (GUI 없이)""" print("\n" + "="*50) print("🔄 GeoMedical Helper - Background Mode") print("="*50) print("프로그램이 백그라운드에서 실행 중입니다.") print("관리 요청과 업데이트를 모니터링하고 있습니다.") print("="*50) try: # 프로그램이 실행 중인 동안 계속 대기 while self.is_running: time.sleep(1) except KeyboardInterrupt: print("\n\nKeyboard interrupt received. Shutting down...") self.quit_application() def run(self): """메인 실행 함수입니다.""" print("=== GeoMedical Client Starting ===") print(f"Build Version: {APP_VERSION}") print(f"Build Date: {APP_BUILD_DATE}") # 실행 파일 정보 exe_path = os.path.abspath(sys.argv[0]) exe_size = os.path.getsize(exe_path) / 1024 / 1024 exe_modified = datetime.fromtimestamp(os.path.getmtime(exe_path)) # 파일 해시 계산 exe_hash = self.calculate_file_hash(exe_path)[:16] # 처음 16자만 표시 print(f"Executable: {exe_path}") print(f"File Size: {exe_size:.2f} MB") print(f"File Hash: {exe_hash}") print(f"Modified: {exe_modified.strftime('%Y-%m-%d %H:%M:%S')}") print(f"Main.py Version: {self.get_main_version()['version']}") print(f"Server: {self.server_url}") print("=" * 50) # 선택적 라이브러리 설치 안내 if not pystray: print("TIP: pip install pystray pillow - for system tray icon") if not Image: print("TIP: pip install pillow - for image processing") if not keyboard: print("TIP: pip install keyboard - for global hotkeys") if not mouse: print("TIP: pip install mouse - for mouse control") if not pyautogui: print("TIP: pip install pyautogui - for automation") if not win32gui: print("TIP: pip install pywin32 - for Windows features") if self.update_and_run(auto_update_mainpy=False): # 자동 업데이트 비활성화 # 시스템 트레이 아이콘 시작 self.start_tray_icon() # 업데이트 확인 스레드 시작 update_thread = threading.Thread(target=self.check_updates_periodically, daemon=True) update_thread.start() # 관리 요청 확인 스레드 시작 self.start_management_request_checker() try: # main 함수 실행 if hasattr(self.current_module, 'main'): self.current_module.main() else: print("Error: No main function found in module") # GUI 모듈이 없으면 단순히 대기 상태로 유지 print("Running in background mode...") self.background_mode() except Exception as e: print(f"Runtime error: {e}") print("Falling back to background mode...") self.background_mode() finally: # 정리 작업 self.cleanup() else: print("Failed to start application") if __name__ == "__main__": """ 프로그램 진입점 python app.py 로 실행합니다. """ try: print("=== DEBUG: Starting GeoMedical Client ===") client = GeoMedicalClient() print("=== DEBUG: Client created successfully ===") client.run() print("=== DEBUG: Client run completed ===") except Exception as e: print(f"=== ERROR: Failed to start client: {e} ===") traceback.print_exc() input("Press Enter to exit...")