395 lines
15 KiB
Python
395 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
밍글 스튜디오 API 서버
|
|
정적 파일 서빙 없이 API 엔드포인트만 처리
|
|
"""
|
|
|
|
import http.server
|
|
import socketserver
|
|
import os
|
|
import sys
|
|
import json
|
|
import urllib.request
|
|
import ssl
|
|
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)
|
|
|
|
# Discord 설정 (.env 파일에서 로드)
|
|
DISCORD_BOT_TOKEN = ''
|
|
DISCORD_INQUIRY_CHANNEL_ID = '' # 문의 상세 (본계정만 열람)
|
|
DISCORD_NOTIFY_CHANNEL_ID = '' # 관리자 알림 (멘션)
|
|
DISCORD_ADMIN_IDS = [] # 멘션할 관리자 User ID 목록
|
|
|
|
def load_env():
|
|
"""Load .env file"""
|
|
global DISCORD_BOT_TOKEN, DISCORD_INQUIRY_CHANNEL_ID, DISCORD_NOTIFY_CHANNEL_ID, DISCORD_ADMIN_IDS
|
|
env_path = os.path.join(WORKING_DIR, '.env')
|
|
if os.path.exists(env_path):
|
|
with open(env_path, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith('#') and '=' in line:
|
|
key, value = line.split('=', 1)
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
if key == 'DISCORD_BOT_TOKEN':
|
|
DISCORD_BOT_TOKEN = value
|
|
elif key == 'DISCORD_INQUIRY_CHANNEL_ID':
|
|
DISCORD_INQUIRY_CHANNEL_ID = value
|
|
elif key == 'DISCORD_NOTIFY_CHANNEL_ID':
|
|
DISCORD_NOTIFY_CHANNEL_ID = value
|
|
elif key == 'DISCORD_ADMIN_IDS':
|
|
DISCORD_ADMIN_IDS = [id.strip() for id in value.split(',') if id.strip()]
|
|
|
|
load_env()
|
|
|
|
def discord_bot_send(channel_id, payload):
|
|
"""Discord Bot API로 메시지 전송"""
|
|
url = f'https://discord.com/api/v10/channels/{channel_id}/messages'
|
|
body = json.dumps(payload).encode('utf-8')
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=body,
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bot {DISCORD_BOT_TOKEN}',
|
|
'User-Agent': 'DiscordBot (https://minglestudio.co.kr, 1.0)'
|
|
},
|
|
method='POST'
|
|
)
|
|
ctx = ssl.create_default_context()
|
|
with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
|
|
if resp.status not in (200, 201):
|
|
raise Exception(f'Discord API failed: {resp.status}')
|
|
return json.loads(resp.read().decode('utf-8'))
|
|
|
|
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()
|
|
elif parsed_path.path == '/api/contact':
|
|
self.handle_post_contact()
|
|
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 handle_post_contact(self):
|
|
"""문의 폼 처리 → Discord Bot API (2채널: 문의상세 + 관리자알림)"""
|
|
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
|
|
|
|
# 필수 필드 검증
|
|
required = ['name', 'email', 'service', 'message']
|
|
for field in required:
|
|
if not data.get(field, '').strip():
|
|
self.send_error_json(400, f'{field} 필드가 필요합니다')
|
|
return
|
|
|
|
# 문의 유형 한글 매핑
|
|
service_map = {
|
|
'studio_rental': '스튜디오 대관',
|
|
'vtuber': 'VTuber 제작',
|
|
'mocap': '모션캡쳐 촬영',
|
|
'music_video': '뮤직비디오 제작',
|
|
'partnership': '제휴/협력',
|
|
'other': '기타'
|
|
}
|
|
service_label = service_map.get(data['service'], data['service'])
|
|
|
|
# Discord 봇 설정 확인
|
|
if not DISCORD_BOT_TOKEN or not DISCORD_INQUIRY_CHANNEL_ID:
|
|
print("[API] Warning: Discord bot not configured")
|
|
self.send_error_json(500, '서버 설정 오류')
|
|
return
|
|
|
|
now = datetime.now()
|
|
|
|
# 1) 문의 상세 채널에 전체 내용 전송
|
|
embed = {
|
|
'title': '📩 새로운 문의가 접수되었습니다',
|
|
'color': 0xff8800,
|
|
'fields': [
|
|
{'name': '👤 이름', 'value': data['name'], 'inline': True},
|
|
{'name': '📧 이메일', 'value': data['email'], 'inline': True},
|
|
{'name': '📱 전화번호', 'value': data.get('phone', '-') or '-', 'inline': True},
|
|
{'name': '📋 문의 유형', 'value': service_label, 'inline': True},
|
|
{'name': '💬 문의 내용', 'value': data['message'][:1024], 'inline': False},
|
|
],
|
|
'footer': {'text': f'⏰ {now.strftime("%Y-%m-%d %H:%M")} | 7일 후 자동 삭제'},
|
|
'timestamp': now.isoformat()
|
|
}
|
|
discord_bot_send(DISCORD_INQUIRY_CHANNEL_ID, {'embeds': [embed]})
|
|
|
|
# 2) 관리자 알림 채널에 멘션 알림 전송
|
|
if DISCORD_NOTIFY_CHANNEL_ID and DISCORD_ADMIN_IDS:
|
|
mentions = ' '.join(f'<@{uid}>' for uid in DISCORD_ADMIN_IDS)
|
|
notify_msg = f'📩 **고객 문의가 접수되었습니다!**\n\n> **{data["name"]}**님 | {service_label}\n\n{mentions}'
|
|
discord_bot_send(DISCORD_NOTIFY_CHANNEL_ID, {
|
|
'content': notify_msg,
|
|
'allowed_mentions': {'users': DISCORD_ADMIN_IDS}
|
|
})
|
|
|
|
# 성공 응답
|
|
response = {
|
|
'success': True,
|
|
'message': '문의가 성공적으로 전송되었습니다.'
|
|
}
|
|
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] 문의 접수: {data['name']} ({service_label})")
|
|
|
|
except urllib.error.URLError as e:
|
|
print(f"[API] Discord API error: {e}")
|
|
self.send_error_json(500, '문의 전송에 실패했습니다. 잠시 후 다시 시도해 주세요.')
|
|
except Exception as e:
|
|
print(f"[API] Contact error: {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(f" POST /api/contact → Discord Bot API")
|
|
if DISCORD_BOT_TOKEN:
|
|
print(f" Discord Bot: configured")
|
|
print(f" Inquiry channel: {DISCORD_INQUIRY_CHANNEL_ID or 'NOT SET'}")
|
|
print(f" Notify channel: {DISCORD_NOTIFY_CHANNEL_ID or 'NOT SET'}")
|
|
print(f" Admin mentions: {len(DISCORD_ADMIN_IDS)} user(s)")
|
|
else:
|
|
print(f" ⚠ Discord Bot: NOT configured (set DISCORD_BOT_TOKEN in .env)")
|
|
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()
|