hoon/kis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: enhance Optuna backtesting framework, optimize orderbook filtering, and update database management utilities.

Browse Source
This commit is contained in:
Your Name
2026-08-12 10:19:19 +09:00
parent cb7e5037a0
commit c6bd62a25f
218 changed files with 31613 additions and 759 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
mm_butler.py
View File

@@ -795,6 +795,131 @@ def handler_apply(args: str, db: TradeDB, hub: "CommandHub") -> str:
return "\n".join(lines)
# ==================================================================
# 유튜브 음원 추출 핸들러 (yt-dlp 연동)
# ==================================================================
# 사용법: !유튜브 [유튜브URL] [시작시간] [종료시간]
# 예시: !유튜브 https://youtu.be/XXXX 00:01:00 00:02:30
# 저장경로: /mnt/nc/share_drive/mp3/
_YOUTUBE_OUTPUT_DIR = "/mnt/nc/share_drive/mp3"
_YOUTUBE_USAGE = "사용법: `!유튜브 [유튜브URL] [시작시간] [종료시간]`\n예시: `!유튜브 https://youtu.be/XXXX 00:01:00 00:02:30`"
def handler_youtube(args: str, db: TradeDB, hub: "CommandHub") -> str:
"""!유튜브 [URL] [시작] [종료] — 유튜브 구간 MP3 추출 후 공유드라이브에 저장."""
import time as _time
parts = (args or "").strip().split()
if len(parts) != 3:
return (
f"❌ **입력 오류**: 인자 3개가 필요합니다.\n"
f"{_YOUTUBE_USAGE}"
)
url, start_time, end_time = parts
# 기본 URL 형식 체크 (command injection 방지용 간단 검증)
if not (url.startswith("https://") or url.startswith("http://")):
return "❌ URL은 http:// 또는 https://로 시작해야 합니다."
# 시간 형식 체크 (HH:MM:SS 또는 MM:SS 또는 초 단위 정수)
import re as _re
_time_pattern = _re.compile(r'^(\d{1,2}:)?\d{1,2}:\d{2}(\.\d+)?$')
for t_val, t_name in ((start_time, "시작시간"), (end_time, "종료시간")):
if not _time_pattern.match(t_val):
return f"❌ {t_name} 형식 오류: `{t_val}`\n`HH:MM:SS` 또는 `MM:SS` 형식으로 입력하세요."
# 출력 파일명 (타임스탬프로 중복 방지)
ts = int(_time.time())
safe_start = start_time.replace(":", "")
safe_end = end_time.replace(":", "")
output_filename = f"hymn_{safe_start}_{safe_end}_{ts}.mp3"
# 1. 독립된 로컬 폴더에 먼저 다운로드 (동시 요청 꼬임 방지)
import os as _os
import shutil as _shutil
local_tmp_dir = f"/home/hoon/mp3/tmp_{ts}"
_os.makedirs(local_tmp_dir, exist_ok=True)
# 영상 제목을 반영한 파일명 템플릿
output_template = f"{local_tmp_dir}/%(title)s_[{safe_start}-{safe_end}].%(ext)s"
command = [
"yt-dlp",
"--no-progress", # 제어 문자 출력 방지
"-x", # 오디오만 추출
"--audio-format", "mp3", # MP3 포맷 지정
"--audio-quality", "0", # 최고 음질
"--no-playlist", # 플레이리스트 방지
"--download-sections", f"*{start_time}-{end_time}", # 구간 지정
"-o", output_template, # 임시 로컬 경로로 출력
url, # 타겟 유튜브 URL
]
logger.info("[유튜브추출] 시작: %s / %s~%s → (임시) %s", url, start_time, end_time, local_tmp_dir)
try:
result = subprocess.run(
command, check=True, capture_output=True, text=True, timeout=300
)
logger.info("[유튜브추출] yt-dlp 정상 종료")
except subprocess.TimeoutExpired:
logger.error("[유튜브추출] 타임아웃 (5분 초과)")
_shutil.rmtree(local_tmp_dir, ignore_errors=True)
return "❌ **추출 실패**: 5분 제한 초과."
except subprocess.CalledProcessError as e:
err_msg = (e.stderr or "") + (e.stdout or "")
err_msg = err_msg.replace('\r', '\n')
err_msg = ''.join(c for c in err_msg if c.isprintable() or c in '\n\t')
logger.error("[유튜브추출] yt-dlp 오류: %s", err_msg)
err_snippet = err_msg[-500:].strip() or "알 수 없는 오류"
_shutil.rmtree(local_tmp_dir, ignore_errors=True)
return f"❌ **추출 실패**: yt-dlp 오류.\n```text\n{err_snippet}\n```"
except Exception as e:
logger.error("[유튜브추출] 예외: %s", e)
_shutil.rmtree(local_tmp_dir, ignore_errors=True)
return f"❌ **추출 실패**: 예외 발생 — {e}"
# 실제 다운로드된 파일명 찾기 (yt-dlp가 생성한 최종 파일)
downloaded_files = _os.listdir(local_tmp_dir)
if not downloaded_files:
_shutil.rmtree(local_tmp_dir, ignore_errors=True)
return "❌ **추출 실패**: 파일이 생성되지 않았습니다."
actual_filename = downloaded_files[0]
actual_local_path = f"{local_tmp_dir}/{actual_filename}"
# rclone으로 클라우드 드라이브에 파일 이동
rclone_cmd = [
"rclone", "move", actual_local_path, "nc:/share_drive/mp3/"
]
logger.info("[유튜브추출] rclone move 시작: %s", rclone_cmd)
try:
rclone_res = subprocess.run(rclone_cmd, check=True, capture_output=True, text=True)
logger.info("[유튜브추출] rclone 완료: %s", rclone_res.stdout)
except subprocess.CalledProcessError as e:
err_msg = (e.stderr or "") + (e.stdout or "")
err_msg = err_msg.replace('\r', '\n')
err_msg = ''.join(c for c in err_msg if c.isprintable() or c in '\n\t')
logger.error("[유튜브추출] rclone 이동 실패: %s", err_msg)
err_snippet = err_msg[-500:].strip() or "알 수 없는 오류"
return f"❌ **이동 실패**: rclone 이동 중 오류 발생.\n```text\n{err_snippet}\n```"
except Exception as e:
logger.error("[유튜브추출] rclone 예외: %s", e)
return f"❌ **이동 실패**: rclone 예외 발생 — {e}"
# 이동 완료 후 껍데기 폴더 삭제
_shutil.rmtree(local_tmp_dir, ignore_errors=True)
return (
f"🎵 **유튜브 추출 및 업로드 완료!**\n"
f"- 구간: `{start_time}` ~ `{end_time}`\n"
f"- 파일: `{actual_filename}`\n"
f"- Nextcloud에서 `share_drive/mp3/` 폴더를 확인하세요 🎶"
)
# ==================================================================
# 봇 제어 핸들러 (systemctl + journalctl) — 화이트리스트 방식
# ==================================================================
@@ -1037,6 +1162,8 @@ class CommandHub:
self.register_command("봇시작", handler_bot_start, "메인 봇 시작")
self.register_command("봇상태", handler_bot_status, "메인 봇 상태 + status 출력 25줄")
self.register_command("봇로그", handler_bot_log, "메인 봇 최근 로그 (예: !봇로그 200, 최대 500)")
# ── 유튜브 음원 추출 ───────────────────────────────────────────────
self.register_command("유튜브", handler_youtube, "유튜브 구간 MP3 추출 → /mnt/nc/share_drive/mp3 저장 (예: !유튜브 [URL] [시작] [종료])")
def register_command(self, cmd: str, handler: Callable, description: str = "") -> None:
"""외부에서 동적으로 명령어 추가 가능."""
@@ -1168,7 +1295,8 @@ class CommandHub:
msg_text = (p.get("message") or "").strip()
reply = self._dispatch(msg_text, p.get("id", ""))
if reply:
self._post_reply(reply, root_id=p.get("id"))
r_id = p.get("root_id") or p.get("id")
self._post_reply(reply, root_id=r_id)
logger.info("명령 응답 전송: %s -> %s…", msg_text[:30], reply[:60])
self.db.set_kv(self.KV_LAST_SEEN_TS, str(last_seen_ts))