131 lines
5.8 KiB
Python
131 lines
5.8 KiB
Python
import os
|
|
import socket
|
|
import threading
|
|
from http.server import HTTPServer, SimpleHTTPRequestHandler
|
|
import logging
|
|
|
|
|
|
class LocalImageServer:
|
|
"""로컬 이미지 파일을 웹에서 접근 가능하도록 하는 HTTP 서버"""
|
|
|
|
def __init__(self, logger, image_dir, port=8000):
|
|
self.logger = logger
|
|
self.image_dir = os.path.abspath(image_dir) # 절대 경로로 변환
|
|
self.original_cwd = os.getcwd() # 원래 작업 디렉토리 저장
|
|
self.port = self.find_available_port(port)
|
|
self.server = None
|
|
self.server_thread = None
|
|
|
|
def find_available_port(self, start_port=8000, max_port=8100):
|
|
"""사용 가능한 포트를 찾습니다"""
|
|
for port in range(start_port, max_port + 1):
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(('localhost', port))
|
|
return port
|
|
except OSError:
|
|
continue
|
|
raise RuntimeError(f"포트 {start_port}-{max_port} 범위에서 사용 가능한 포트를 찾을 수 없습니다.")
|
|
|
|
def start_server(self):
|
|
"""HTTP 서버를 시작합니다"""
|
|
if self.server_thread and self.server_thread.is_alive():
|
|
self.logger.log(f"로컬 이미지 서버가 이미 포트 {self.port}에서 실행 중입니다.", level=logging.DEBUG)
|
|
return
|
|
|
|
# 이미지 디렉토리 존재 확인
|
|
if not os.path.exists(self.image_dir):
|
|
try:
|
|
os.makedirs(self.image_dir, exist_ok=True)
|
|
self.logger.log(f"이미지 디렉토리 생성: {self.image_dir}", level=logging.INFO)
|
|
except Exception as e:
|
|
self.logger.log(f"이미지 디렉토리 생성 실패: {e}", level=logging.ERROR)
|
|
raise
|
|
|
|
try:
|
|
# 작업 디렉토리 변경 없이 CustomHandler에서 직접 경로 처리
|
|
class CustomHandler(SimpleHTTPRequestHandler):
|
|
def __init__(self, *args, image_directory=None, **kwargs):
|
|
self.image_directory = image_directory
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def translate_path(self, path):
|
|
"""요청 경로를 이미지 디렉토리 내의 실제 파일 경로로 변환"""
|
|
# 기본 translate_path 호출하여 상대 경로 얻기
|
|
path = super().translate_path(path)
|
|
# 현재 작업 디렉토리 대신 이미지 디렉토리 사용
|
|
rel_path = os.path.relpath(path, os.getcwd())
|
|
return os.path.join(self.image_directory, rel_path)
|
|
|
|
def log_message(self, format, *args):
|
|
# 로그 출력을 비활성화 (너무 많은 로그 방지)
|
|
pass
|
|
|
|
def end_headers(self):
|
|
# CORS 헤더 추가
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
|
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
|
super().end_headers()
|
|
|
|
# 핸들러에 이미지 디렉토리 전달
|
|
def handler_factory(*args, **kwargs):
|
|
return CustomHandler(*args, image_directory=self.image_dir, **kwargs)
|
|
|
|
self.server = HTTPServer(('localhost', self.port), handler_factory)
|
|
self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.server_thread.start()
|
|
|
|
self.logger.log(f"로컬 이미지 서버가 포트 {self.port}에서 시작되었습니다. (디렉토리: {self.image_dir})", level=logging.INFO)
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"로컬 웹서버 시작 실패: {e}", level=logging.ERROR)
|
|
# 실패 시 상태 정리
|
|
self.server = None
|
|
self.server_thread = None
|
|
raise
|
|
|
|
def stop_server(self):
|
|
"""HTTP 서버를 중지합니다"""
|
|
if self.server:
|
|
try:
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
if self.server_thread and self.server_thread.is_alive():
|
|
self.server_thread.join(timeout=5)
|
|
self.logger.log("로컬 이미지 서버가 중지되었습니다.", level=logging.INFO)
|
|
except Exception as e:
|
|
self.logger.log(f"로컬 웹서버 중지 중 오류 발생: {e}", level=logging.ERROR)
|
|
finally:
|
|
self.server = None
|
|
self.server_thread = None
|
|
|
|
def restart_server(self):
|
|
"""서버를 재시작합니다"""
|
|
self.logger.log("로컬 이미지 서버 재시작 중...", level=logging.INFO)
|
|
self.stop_server()
|
|
# 새로운 포트 찾기
|
|
self.port = self.find_available_port(self.port)
|
|
self.start_server()
|
|
|
|
def get_base_url(self):
|
|
"""서버의 기본 URL을 반환합니다"""
|
|
return f"http://localhost:{self.port}"
|
|
|
|
def is_running(self):
|
|
"""서버가 실행 중인지 확인합니다"""
|
|
return self.server is not None and self.server_thread and self.server_thread.is_alive()
|
|
|
|
def get_file_url(self, filename):
|
|
"""특정 파일의 URL을 반환합니다"""
|
|
if not self.is_running():
|
|
self.logger.log("서버가 실행되지 않았습니다.", level=logging.WARNING)
|
|
return None
|
|
return f"{self.get_base_url()}/{filename}"
|
|
|
|
def __del__(self):
|
|
"""소멸자에서 서버 정리"""
|
|
try:
|
|
self.stop_server()
|
|
except:
|
|
pass |