146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
환경변수 DB 업데이트 스크립트
|
|
- 기존 최신 설정을 불러와서
|
|
- 새로운 값으로 업데이트하고
|
|
- 새 스냅샷으로 저장
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
from database import TradeDB
|
|
|
|
def update_env_config():
|
|
"""환경변수 업데이트 (대화형)"""
|
|
db_path = SCRIPT_DIR / "quant_bot.db"
|
|
db = TradeDB(db_path=str(db_path))
|
|
|
|
# 기존 최신 설정 불러오기
|
|
latest = db.get_latest_env()
|
|
if latest and latest.get("snapshot"):
|
|
current = latest["snapshot"]
|
|
print("📋 현재 설정:")
|
|
print(f" (최신 스냅샷 ID: {latest['id']}, 생성일: {latest['created_at']})")
|
|
print()
|
|
else:
|
|
current = {}
|
|
print("⚠️ 기존 설정 없음 - 새로 생성합니다.")
|
|
print()
|
|
|
|
# 업데이트할 값들
|
|
updates = {}
|
|
|
|
print("=" * 60)
|
|
print("환경변수 업데이트")
|
|
print("=" * 60)
|
|
print("(값을 입력하지 않으면 기존 값 유지, 'skip' 입력 시 건너뜀)")
|
|
print()
|
|
|
|
# 한투 API 설정
|
|
print("🔵 [한투 API 설정]")
|
|
val = input(f"KIS_APP_KEY [{current.get('KIS_APP_KEY', '')[:20]}...]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_APP_KEY"] = val
|
|
|
|
val = input(f"KIS_APP_SECRET [{current.get('KIS_APP_SECRET', '')[:20]}...]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_APP_SECRET"] = val
|
|
|
|
val = input(f"KIS_ACCOUNT_NO [{current.get('KIS_ACCOUNT_NO', '')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_ACCOUNT_NO"] = val
|
|
|
|
val = input(f"KIS_ACCOUNT_CODE [{current.get('KIS_ACCOUNT_CODE', '01')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_ACCOUNT_CODE"] = val
|
|
|
|
val = input(f"KIS_MOCK (true/false) [{current.get('KIS_MOCK', 'true')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_MOCK"] = val.lower()
|
|
|
|
print()
|
|
|
|
# Mattermost 설정
|
|
print("💬 [Mattermost 설정]")
|
|
val = input(f"MM_SERVER_URL [{current.get('MM_SERVER_URL', 'https://mattermost.hoonfam.org')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["MM_SERVER_URL"] = val
|
|
|
|
val = input(f"MM_BOT_TOKEN_ [{current.get('MM_BOT_TOKEN_', '')[:20]}...]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["MM_BOT_TOKEN_"] = val
|
|
|
|
val = input(f"MATTERMOST_CHANNEL (기본 채널 alias) [{current.get('MATTERMOST_CHANNEL', 'stock')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["MATTERMOST_CHANNEL"] = val
|
|
|
|
val = input(f"KIS_SHORT_MM_CHANNEL (단타 봇 채널 alias) [{current.get('KIS_SHORT_MM_CHANNEL', '')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_SHORT_MM_CHANNEL"] = val
|
|
|
|
val = input(f"KIS_LONG_MM_CHANNEL (롱타 봇 채널 alias) [{current.get('KIS_LONG_MM_CHANNEL', '')}]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["KIS_LONG_MM_CHANNEL"] = val
|
|
|
|
print()
|
|
|
|
# Gemini API 설정
|
|
print("🤖 [Gemini AI 설정]")
|
|
val = input(f"GEMINI_API_KEY [{current.get('GEMINI_API_KEY', '')[:20]}...]: ").strip()
|
|
if val and val.lower() != 'skip':
|
|
updates["GEMINI_API_KEY"] = val
|
|
|
|
print()
|
|
|
|
# 기존 값과 병합
|
|
new_snapshot = {**current, **updates}
|
|
|
|
# 확인
|
|
print("=" * 60)
|
|
print("업데이트할 값:")
|
|
print("=" * 60)
|
|
for key, value in updates.items():
|
|
display_value = value[:30] + "..." if len(str(value)) > 30 else value
|
|
print(f" {key}: {display_value}")
|
|
|
|
print()
|
|
confirm = input("위 설정을 저장하시겠습니까? (y/n): ").strip().lower()
|
|
|
|
if confirm != 'y':
|
|
print("❌ 취소되었습니다.")
|
|
db.close()
|
|
return
|
|
|
|
# 새 스냅샷 저장
|
|
env_id = db.insert_env_snapshot(new_snapshot)
|
|
if env_id:
|
|
print(f"✅ 환경변수 저장 완료! (새 스냅샷 ID: {env_id})")
|
|
print()
|
|
print("💡 팁:")
|
|
print(" - 봇을 재시작하면 새 설정이 적용됩니다.")
|
|
print(" - mm_config.json 파일도 확인하세요:")
|
|
print(" {")
|
|
print(' "channels": {')
|
|
print(' "stock": "채널_ID",')
|
|
print(' "kis-short": "단타_채널_ID",')
|
|
print(' "kis-long": "롱타_채널_ID"')
|
|
print(" }")
|
|
print(" }")
|
|
else:
|
|
print("❌ 저장 실패!")
|
|
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
update_env_config()
|
|
except KeyboardInterrupt:
|
|
print("\n\n❌ 사용자 취소")
|
|
except Exception as e:
|
|
print(f"\n\n❌ 에러: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|