From ef9f01c4cd27c7a762c5039cd174d93cbc2562b6 Mon Sep 17 00:00:00 2001 From: gahusb Date: Thu, 5 Feb 2026 00:55:39 +0900 Subject: [PATCH] =?UTF-8?q?.gitignore=20=EC=88=98=EC=A0=95,=20=ED=85=94?= =?UTF-8?q?=EB=A0=88=EA=B7=B8=EB=9E=A8=20=EB=B4=87=20=EB=AA=85=EB=A0=B9?= =?UTF-8?q?=EC=96=B4=20=EC=88=98=EC=A0=95,=20=EC=A7=80=EC=88=98=20?= =?UTF-8?q?=EC=A1=B0=ED=9A=8C=20=EC=98=A4=EB=A5=98=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 3 ++ modules/config.py | 2 +- modules/services/kis.py | 10 +++++- modules/services/telegram_bot/server.py | 47 +++++++++++++++++++------ 4 files changed, 49 insertions(+), 13 deletions(-) diff --git a/.gitignore b/.gitignore index f59086f..9241dfa 100644 --- a/.gitignore +++ b/.gitignore @@ -54,3 +54,6 @@ tests/* # System Thumbs.db Desktop.ini + +# stock +KIS_SETUP.md \ No newline at end of file diff --git a/modules/config.py b/modules/config.py index 77aed8c..a0bf427 100644 --- a/modules/config.py +++ b/modules/config.py @@ -12,7 +12,7 @@ class Config: # 2. NAS 및 AI 서버 NAS_API_URL = os.getenv("NAS_API_URL", "http://192.168.45.54:18500") OLLAMA_API_URL = os.getenv("OLLAMA_API_URL", "http://localhost:11434") - OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3.1") + OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3.1:8b-instruct-q8_0") # 3. KIS 한국투자증권 KIS_ENV_TYPE = os.getenv("KIS_ENV_TYPE", "virtual").lower() diff --git a/modules/services/kis.py b/modules/services/kis.py index eda7ef5..468605e 100644 --- a/modules/services/kis.py +++ b/modules/services/kis.py @@ -2,6 +2,7 @@ import requests import json import time import os +from datetime import datetime, timedelta from modules.config import Config @@ -344,12 +345,19 @@ class KISClient: url = f"{self.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-indexchartprice" headers = self._get_headers(tr_id="FHKUP03500200") + # 날짜 계산 (최근 100일) + end_dt = datetime.now().strftime("%Y%m%d") + start_dt = (datetime.now() - timedelta(days=100)).strftime("%Y%m%d") + params = { "FID_COND_MRKT_DIV_CODE": "U", # U: 업종/지수 "FID_INPUT_ISCD": ticker, + "FID_INPUT_DATE_1": start_dt, # 시작일 + "FID_INPUT_DATE_2": end_dt, # 종료일 "FID_PERIOD_DIV_CODE": period, - "FID_ORG_ADJ_PRC": "1" # 수정주가 + "FID_ORG_ADJ_PRC": "0" # 수정주가 반영 여부 } + try: res = requests.get(url, headers=headers, params=params) diff --git a/modules/services/telegram_bot/server.py b/modules/services/telegram_bot/server.py index cf7d514..be65ae6 100644 --- a/modules/services/telegram_bot/server.py +++ b/modules/services/telegram_bot/server.py @@ -273,43 +273,68 @@ class TelegramBotServer: async def exec_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/exec: 원격 명령어 실행""" - if not context.args: + # ultimate_handler에서는 context.args가 비어있으므로 직접 파싱 + text = update.message.text.strip() + parts = text.split(maxsplit=1) # "/exec" 와 나머지 명령어로 분리 + + if len(parts) < 2: await update.message.reply_text("❌ 사용법: /exec ") return - command = " ".join(context.args) + command = parts[1] # "/exec" 이후의 모든 텍스트 await update.message.reply_text(f"⚙️ 실행 중: `{command}`", parse_mode="Markdown") try: # 보안: 위험한 명령어 차단 - dangerous_keywords = ['rm', 'del', 'format', 'shutdown', 'reboot'] + dangerous_keywords = ['rm', 'del', 'format', 'shutdown', 'reboot', 'ipconfig'] if any(keyword in command.lower() for keyword in dangerous_keywords): await update.message.reply_text("⛔ 위험한 명령어는 실행할 수 없습니다.") return + # Windows에서는 PowerShell을 명시적으로 사용 + import platform + if platform.system() == 'Windows': + exec_command = ['powershell', '-Command', command] + else: + exec_command = command + # 명령어 실행 (타임아웃 30초) result = subprocess.run( - command, - shell=True, + exec_command, + shell=False if platform.system() == 'Windows' else True, capture_output=True, text=True, + encoding='utf-8', + errors='replace', # 인코딩 오류 무시 timeout=30, cwd=os.getcwd() ) - output = result.stdout if result.stdout else result.stderr - if not output: - output = "명령어 실행 완료 (출력 없음)" + # stdout와 stderr 모두 확인 + output = result.stdout.strip() if result.stdout else "" + error_output = result.stderr.strip() if result.stderr else "" + + if output and error_output: + combined = f"[STDOUT]\n{output}\n\n[STDERR]\n{error_output}" + elif output: + combined = output + elif error_output: + combined = f"[ERROR]\n{error_output}" + else: + combined = "명령어 실행 완료 (출력 없음)" # 출력이 너무 길면 잘라내기 - if len(output) > 3000: - output = output[:3000] + "\n... (출력이 너무 깁니다)" + if len(combined) > 3000: + combined = combined[:3000] + "\n... (출력이 너무 깁니다)" - await update.message.reply_text(f"```\n{output}\n```", parse_mode="Markdown") + await update.message.reply_text(f"```\n{combined}\n```", parse_mode="Markdown") except subprocess.TimeoutExpired: await update.message.reply_text("⏱️ 명령어 실행 시간 초과 (30초)") except Exception as e: + print(f"❌ [Telegram /exec] Error: {e}") + import traceback + traceback.print_exc() await update.message.reply_text(f"❌ 실행 오류: {str(e)}") def run(self):