Add: API 서버 스크립트 추가

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
서버 2026-03-04 23:27:35 +09:00
parent 281ca11b19
commit baf19fe0b4

248
api-server.py Normal file
View File

@ -0,0 +1,248 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
밍글 스튜디오 API 서버
정적 파일 서빙 없이 API 엔드포인트만 처리
"""
import http.server
import socketserver
import os
import sys
import json
from datetime import datetime
from urllib.parse import urlparse
# 한글 출력을 위한 인코딩 설정
if os.name == 'nt':
import codecs
try:
sys.stdout.reconfigure(encoding='utf-8')
sys.stderr.reconfigure(encoding='utf-8')
except:
sys.stdout = codecs.getwriter('utf-8')(sys.stdout.detach())
sys.stderr = codecs.getwriter('utf-8')(sys.stderr.detach())
# 서버 설정
PORT = 8001
HOST = '127.0.0.1' # 로컬만 허용 (Caddy가 프록시)
# 작업 디렉토리 설정
WORKING_DIR = os.path.dirname(os.path.abspath(__file__))
os.chdir(WORKING_DIR)
class APIRequestHandler(http.server.BaseHTTPRequestHandler):
"""API 전용 HTTP 요청 핸들러"""
BACKGROUNDS_DATA_FILE = 'data/backgrounds.json'
PROPS_DATA_FILE = 'data/props.json'
def end_headers(self):
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, X-API-Key')
super().end_headers()
def do_OPTIONS(self):
"""OPTIONS 요청 처리 (CORS preflight)"""
self.send_response(200)
self.end_headers()
def do_GET(self):
"""GET 요청 처리"""
parsed_path = urlparse(self.path)
if parsed_path.path == '/api/backgrounds':
self.handle_get_backgrounds()
elif parsed_path.path == '/api/props':
self.handle_get_props()
else:
self.send_error_json(404, 'API endpoint not found')
def do_POST(self):
"""POST 요청 처리"""
parsed_path = urlparse(self.path)
if parsed_path.path == '/api/backgrounds':
self.handle_post_backgrounds()
elif parsed_path.path == '/api/props':
self.handle_post_props()
else:
self.send_error_json(404, 'API endpoint not found')
def handle_get_backgrounds(self):
"""배경 데이터 조회 API"""
try:
if os.path.exists(self.BACKGROUNDS_DATA_FILE):
with open(self.BACKGROUNDS_DATA_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
else:
data = {
'lastUpdated': datetime.now().isoformat(),
'backgrounds': []
}
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
except Exception as e:
self.send_error_json(500, f'데이터 조회 실패: {str(e)}')
def handle_post_backgrounds(self):
"""배경 데이터 업데이트 API"""
try:
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self.send_error_json(400, '요청 본문이 비어있습니다')
return
post_data = self.rfile.read(content_length)
try:
data = json.loads(post_data.decode('utf-8'))
except json.JSONDecodeError as e:
self.send_error_json(400, f'잘못된 JSON 형식: {str(e)}')
return
if 'backgrounds' not in data:
self.send_error_json(400, 'backgrounds 필드가 필요합니다')
return
data['lastUpdated'] = datetime.now().isoformat()
data_dir = os.path.dirname(self.BACKGROUNDS_DATA_FILE)
if data_dir and not os.path.exists(data_dir):
os.makedirs(data_dir)
with open(self.BACKGROUNDS_DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
response = {
'success': True,
'message': f'{len(data["backgrounds"])}개의 배경이 저장되었습니다',
'lastUpdated': data['lastUpdated'],
'count': len(data['backgrounds'])
}
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8'))
print(f"[API] 배경 데이터 업데이트: {len(data['backgrounds'])}개 항목")
except Exception as e:
self.send_error_json(500, f'데이터 저장 실패: {str(e)}')
def handle_get_props(self):
"""프랍 데이터 조회 API"""
try:
if os.path.exists(self.PROPS_DATA_FILE):
with open(self.PROPS_DATA_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
else:
data = {
'lastUpdated': datetime.now().isoformat(),
'props': []
}
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
except Exception as e:
self.send_error_json(500, f'데이터 조회 실패: {str(e)}')
def handle_post_props(self):
"""프랍 데이터 업데이트 API"""
try:
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self.send_error_json(400, '요청 본문이 비어있습니다')
return
post_data = self.rfile.read(content_length)
try:
data = json.loads(post_data.decode('utf-8'))
except json.JSONDecodeError as e:
self.send_error_json(400, f'잘못된 JSON 형식: {str(e)}')
return
if 'props' not in data:
self.send_error_json(400, 'props 필드가 필요합니다')
return
data['lastUpdated'] = datetime.now().isoformat()
data_dir = os.path.dirname(self.PROPS_DATA_FILE)
if data_dir and not os.path.exists(data_dir):
os.makedirs(data_dir)
with open(self.PROPS_DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
response = {
'success': True,
'message': f'{len(data["props"])}개의 프랍이 저장되었습니다',
'lastUpdated': data['lastUpdated'],
'count': len(data['props'])
}
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8'))
print(f"[API] 프랍 데이터 업데이트: {len(data['props'])}개 항목")
except Exception as e:
self.send_error_json(500, f'데이터 저장 실패: {str(e)}')
def send_error_json(self, code, message):
"""JSON 형식 에러 응답"""
self.send_response(code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
error_response = {
'success': False,
'error': message
}
self.wfile.write(json.dumps(error_response, ensure_ascii=False).encode('utf-8'))
def log_message(self, format, *args):
"""로그 메시지 포맷팅"""
print(f"[{self.log_date_time_string()}] {format % args}")
def main():
"""메인 서버 실행 함수"""
print(f"Mingle Studio API Server")
print("=" * 40)
print(f"Listening on {HOST}:{PORT}")
print(f"Working directory: {WORKING_DIR}")
print("=" * 40)
print("API Endpoints:")
print(f" GET /api/backgrounds")
print(f" POST /api/backgrounds")
print(f" GET /api/props")
print(f" POST /api/props")
print("=" * 40)
try:
with socketserver.TCPServer((HOST, PORT), APIRequestHandler) as httpd:
httpd.allow_reuse_address = True
httpd.serve_forever()
except KeyboardInterrupt:
print("\nAPI Server stopped.")
except OSError as e:
if e.errno == 10048:
print(f"Error: Port {PORT} is already in use.")
else:
print(f"Server error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()