Initial project upload

This commit is contained in:
Kim.KANGHEE
2026-05-05 17:14:11 +09:00
commit 122b73d254
282 changed files with 72135 additions and 0 deletions
+44
View File
@@ -0,0 +1,44 @@
from __future__ import annotations
from flask import Flask
from .home import register_home_routes
from .auth import register_auth_routes
from .admin import register_admin_routes
from .main import register_main_routes
from .xml import register_xml_routes
from .utilities import register_util_routes, utils_bp
from .file_view import register_file_view
from .jobs import register_jobs_routes
from .idrac_routes import register_idrac_routes
from .catalog_sync import catalog_bp
from .scp_routes import scp_bp
from .drm_sync import drm_sync_bp
from .server_config_routes import server_config_bp
from .bios_baseline_routes import register_bios_baseline_routes
from .script_manager import script_manager_bp
def register_routes(app: Flask, socketio=None) -> None:
"""블루프린트 일괄 등록. socketio는 main 라우트에서만 사용."""
register_home_routes(app)
register_auth_routes(app)
register_admin_routes(app)
register_main_routes(app, socketio)
register_xml_routes(app)
app.register_blueprint(utils_bp, url_prefix="/utils")
register_file_view(app)
register_jobs_routes(app)
register_idrac_routes(app)
app.register_blueprint(catalog_bp)
app.register_blueprint(scp_bp)
app.register_blueprint(drm_sync_bp)
app.register_blueprint(server_config_bp)
register_bios_baseline_routes(app)
app.register_blueprint(script_manager_bp)
# CSRF 제외 - server_config API 엔드포인트
try:
csrf = app.extensions.get('csrf')
if csrf:
csrf.exempt(server_config_bp)
except Exception as e:
app.logger.warning(f"CSRF exemption failed for server_config_bp: {e}")
+373
View File
@@ -0,0 +1,373 @@
# backend/routes/admin.py
from __future__ import annotations
import logging
from functools import wraps
from typing import Callable
from flask import (
Blueprint,
render_template,
redirect,
url_for,
flash,
abort,
request,
current_app,
)
from flask_login import login_required, current_user
from backend.models.user import User, db
from backend.models.telegram_bot import TelegramBot
from backend.models.system_setting import SystemSetting
try:
from telegram import Bot
except ImportError:
Bot = None
admin_bp = Blueprint("admin", __name__)
def get_system_setting(key: str, default: str = "") -> str:
setting = db.session.get(SystemSetting, key)
return setting.value if setting else default
def set_system_setting(key: str, value: str) -> None:
setting = db.session.get(SystemSetting, key)
if setting:
setting.value = value
else:
db.session.add(SystemSetting(key=key, value=value))
# Blueprint 등록
def register_admin_routes(app):
app.register_blueprint(admin_bp)
# 관리자 권한 데코레이터
def admin_required(view_func: Callable):
@wraps(view_func)
def wrapper(*args, **kwargs):
if not current_user.is_authenticated:
return redirect(url_for("auth.login"))
if not getattr(current_user, "is_admin", False):
flash("관리자 권한이 필요합니다.", "danger")
return redirect(url_for("main.index"))
return view_func(*args, **kwargs)
return wrapper
# 관리자 대시보드
@admin_bp.route("/admin", methods=["GET"])
@login_required
@admin_required
def admin_panel():
users = db.session.query(User).order_by(User.id.asc()).all()
return render_template("admin.html", users=users)
# 사용자 승인
@admin_bp.route("/admin/approve/<int:user_id>", methods=["GET"])
@login_required
@admin_required
def approve_user(user_id: int):
user = db.session.get(User, user_id)
if not user:
abort(404)
user.is_active = True
db.session.commit()
flash("사용자가 승인되었습니다.", "success")
logging.info("✅ 승인된 사용자: %s (id=%s)", user.username, user.id)
return redirect(url_for("admin.admin_panel"))
# 사용자 관리자 승격
@admin_bp.route("/admin/users/<int:user_id>/promote", methods=["POST"])
@login_required
@admin_required
def promote_user(user_id: int):
user = db.session.get(User, user_id)
if not user:
abort(404)
if user.id == current_user.id:
flash("본인 계정은 승격 대상이 아닙니다.", "warning")
return redirect(url_for("admin.admin_panel"))
if user.is_admin:
flash("이미 관리자 권한을 가진 사용자입니다.", "info")
return redirect(url_for("admin.admin_panel"))
user.is_admin = True
db.session.commit()
flash(f"{user.username} 사용자를 관리자로 승격했습니다.", "success")
logging.info(
"ADMIN: promoted user_id=%s to admin by admin_id=%s",
user.id,
current_user.id,
)
return redirect(url_for("admin.admin_panel"))
# 사용자 삭제
@admin_bp.route("/admin/delete/<int:user_id>", methods=["GET"])
@login_required
@admin_required
def delete_user(user_id: int):
user = db.session.get(User, user_id)
if not user:
abort(404)
username = user.username
db.session.delete(user)
db.session.commit()
flash("사용자가 삭제되었습니다.", "success")
logging.info("🗑 삭제된 사용자: %s (id=%s)", username, user_id)
return redirect(url_for("admin.admin_panel"))
# ▼▼▼ 사용자 비밀번호 변경(관리자용) ▼▼▼
@admin_bp.route("/admin/users/<int:user_id>/reset_password", methods=["POST"])
@login_required
@admin_required
def reset_password(user_id: int):
"""
admin.html에서 각 사용자 행 아래 폼으로부터 POST:
- name="new_password"
- name="confirm_password"
CSRF는 템플릿에서 {{ csrf_token() }} 또는 {{ form.hidden_tag() }}로 포함되어야 합니다.
"""
new_pw = (request.form.get("new_password") or "").strip()
confirm = (request.form.get("confirm_password") or "").strip()
# 서버측 검증
if not new_pw or not confirm:
flash("비밀번호와 확인 값을 모두 입력하세요.", "warning")
return redirect(url_for("admin.admin_panel"))
if new_pw != confirm:
flash("비밀번호 확인이 일치하지 않습니다.", "warning")
return redirect(url_for("admin.admin_panel"))
if len(new_pw) < 8:
flash("비밀번호는 최소 8자 이상이어야 합니다.", "warning")
return redirect(url_for("admin.admin_panel"))
user = db.session.get(User, user_id)
if not user:
abort(404)
try:
# passlib(Argon2id) 기반 set_password 사용 (models.user에 구현됨)
user.set_password(new_pw)
db.session.commit()
flash(f"사용자(ID={user.id}) 비밀번호를 변경했습니다.", "success")
current_app.logger.info(
"ADMIN: reset password for user_id=%s by admin_id=%s",
user.id, current_user.id
)
except Exception as e:
db.session.rollback()
current_app.logger.exception("ADMIN: reset password failed: %s", e)
flash("비밀번호 변경 중 오류가 발생했습니다.", "danger")
return redirect(url_for("admin.admin_panel"))
# ▼▼▼ 시스템 설정 (텔레그램 봇 관리) ▼▼▼
@admin_bp.route("/admin/settings", methods=["GET"])
@login_required
@admin_required
def settings():
# 테이블 생성 확인 (임시)
try:
bots = TelegramBot.query.all()
except Exception:
db.create_all()
bots = []
try:
tsr_share_settings = {
"share_url": get_system_setting("tsr_share_url", "//10.10.3.15/share/"),
"share_user": get_system_setting("tsr_share_user", ""),
"share_password": get_system_setting("tsr_share_password", ""),
}
except Exception:
db.create_all()
tsr_share_settings = {
"share_url": "//10.10.3.15/share/",
"share_user": "",
"share_password": "",
}
return render_template("admin_settings.html", bots=bots, tsr_share_settings=tsr_share_settings)
@admin_bp.route("/admin/settings/tsr-share", methods=["POST"])
@login_required
@admin_required
def save_tsr_share_settings():
share_url = (request.form.get("share_url") or "").strip()
share_user = (request.form.get("share_user") or "").strip()
share_password = request.form.get("share_password") or ""
if not share_url:
flash("TSR 저장 Share URL을 입력하세요.", "warning")
return redirect(url_for("admin.settings"))
try:
set_system_setting("tsr_share_url", share_url)
set_system_setting("tsr_share_user", share_user)
set_system_setting("tsr_share_password", share_password)
db.session.commit()
flash("TSR 저장소 설정이 저장되었습니다.", "success")
except Exception as e:
db.session.rollback()
current_app.logger.exception("TSR share settings save failed: %s", e)
flash("TSR 저장소 설정 저장 중 오류가 발생했습니다.", "danger")
return redirect(url_for("admin.settings"))
@admin_bp.route("/admin/settings/bot/add", methods=["POST"])
@login_required
@admin_required
def add_bot():
name = request.form.get("name")
token = request.form.get("token")
chat_id = request.form.get("chat_id")
desc = request.form.get("description")
# 알림 유형 (체크박스 다중 선택)
notify_types = request.form.getlist("notify_types")
notify_str = ",".join(notify_types) if notify_types else ""
if not (name and token and chat_id):
flash("필수 항목(이름, 토큰, Chat ID)을 입력하세요.", "warning")
return redirect(url_for("admin.settings"))
bot = TelegramBot(
name=name,
token=token,
chat_id=chat_id,
description=desc,
is_active=True,
notification_types=notify_str
)
db.session.add(bot)
db.session.commit()
flash("텔레그램 봇이 추가되었습니다.", "success")
return redirect(url_for("admin.settings"))
@admin_bp.route("/admin/settings/bot/edit/<int:bot_id>", methods=["POST"])
@login_required
@admin_required
def edit_bot(bot_id):
bot = db.session.get(TelegramBot, bot_id)
if not bot:
abort(404)
bot.name = request.form.get("name")
bot.token = request.form.get("token")
bot.chat_id = request.form.get("chat_id")
bot.description = request.form.get("description")
bot.is_active = request.form.get("is_active") == "on"
# 알림 유형 업데이트
notify_types = request.form.getlist("notify_types")
bot.notification_types = ",".join(notify_types) if notify_types else ""
db.session.commit()
flash("봇 설정이 수정되었습니다.", "success")
return redirect(url_for("admin.settings"))
@admin_bp.route("/admin/settings/bot/delete/<int:bot_id>", methods=["POST"])
@login_required
@admin_required
def delete_bot(bot_id):
bot = db.session.get(TelegramBot, bot_id)
if bot:
db.session.delete(bot)
db.session.commit()
flash("봇이 삭제되었습니다.", "success")
return redirect(url_for("admin.settings"))
@admin_bp.route("/admin/settings/bot/test/<int:bot_id>", methods=["POST"])
@login_required
@admin_required
def test_bot(bot_id):
if not Bot:
flash("python-telegram-bot 라이브러리가 설치되지 않았습니다.", "danger")
return redirect(url_for("admin.settings"))
bot_obj = db.session.get(TelegramBot, bot_id)
if not bot_obj:
abort(404)
import asyncio
async def _send():
bot = Bot(token=bot_obj.token)
await bot.send_message(chat_id=bot_obj.chat_id, text="🔔 <b>테스트 메시지</b>\n이 메시지가 보이면 설정이 정상입니다.", parse_mode="HTML")
try:
asyncio.run(_send())
flash(f"'{bot_obj.name}' 봇으로 테스트 메시지를 보냈습니다.", "success")
except Exception as e:
flash(f"테스트 실패: {e}", "danger")
return redirect(url_for("admin.settings"))
# ▼▼▼ 시스템 로그 뷰어 ▼▼▼
@admin_bp.route("/admin/logs", methods=["GET"])
@login_required
@admin_required
def view_logs():
import os
import re
from collections import deque
log_folder = current_app.config.get('LOG_FOLDER')
log_file = os.path.join(log_folder, 'app.log') if log_folder else None
# 1. 실제 ANSI 이스케이프 코드 (\x1B로 시작)
ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')
# 2. 텍스트로 찍힌 ANSI 코드 패턴 (예: [36m, [0m 등) - Werkzeug가 이스케이프 된 상태로 로그에 남길 경우 대비
literal_ansi = re.compile(r'\[[0-9;]+m')
# 3. 제어 문자 제거
control_char_re = re.compile(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]')
logs = []
if log_file and os.path.exists(log_file):
try:
with open(log_file, 'r', encoding='utf-8', errors='replace') as f:
raw_lines = deque(f, 1000)
for line in raw_lines:
# A. 실제 ANSI 코드 제거
clean_line = ansi_escape.sub('', line)
# B. 리터럴 ANSI 패턴 제거 (사용자가 [36m 등을 텍스트로 보고 있다면 이것이 원인)
clean_line = literal_ansi.sub('', clean_line)
# C. 제어 문자 제거
clean_line = control_char_re.sub('', clean_line)
# D. 앞뒤 공백 제거
clean_line = clean_line.strip()
# E. 빈 줄 제외
if clean_line:
logs.append(clean_line)
except Exception as e:
logs = [f"Error reading log file: {str(e)}"]
else:
logs = [f"Log file not found at: {log_file}"]
return render_template("admin_logs.html", logs=logs)
+380
View File
@@ -0,0 +1,380 @@
# backend/routes/auth.py
from __future__ import annotations
import logging
import threading
import asyncio
import secrets
from typing import Optional
from urllib.parse import urlparse, urljoin
from datetime import datetime
from flask import (
Blueprint,
render_template,
redirect,
url_for,
flash,
request,
session,
current_app,
)
from flask_login import login_user, logout_user, current_user, login_required
from backend.forms.auth_forms import RegistrationForm, LoginForm
from backend.models.user import User, db
# ── (선택) Telegram: 미설정이면 조용히 패스
try:
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ParseMode
except Exception: # 라이브러리 미설치/미사용 환경
Bot = None
ParseMode = None
InlineKeyboardButton = None
InlineKeyboardMarkup = None
auth_bp = Blueprint("auth", __name__)
# ─────────────────────────────────────────────────────────────
# 유틸
# ─────────────────────────────────────────────────────────────
def _is_safe_url(target: str) -> bool:
"""로그인 후 next 파라미터의 안전성 확인(동일 호스트만 허용)."""
ref = urlparse(request.host_url)
test = urlparse(urljoin(request.host_url, target))
return (test.scheme in ("http", "https")) and (ref.netloc == test.netloc)
def _notify(text: str, category: str = "system") -> None:
"""
텔레그램 알림 전송
- DB(TelegramBot)에 등록된 활성 봇들에게 전송
- category: 'auth', 'activity', 'system'
"""
try:
from backend.models.telegram_bot import TelegramBot
# 앱 컨텍스트 안에서 실행되므로 바로 DB 접근 가능
try:
bots = TelegramBot.query.filter_by(is_active=True).all()
except Exception:
db.create_all()
bots = []
# 1. DB에 봇이 없고, Config에 설정이 있는 경우 -> 자동 마이그레이션
if not bots:
cfg_token = (current_app.config.get("TELEGRAM_BOT_TOKEN") or "").strip()
cfg_chat = (current_app.config.get("TELEGRAM_CHAT_ID") or "").strip()
if cfg_token and cfg_chat:
new_bot = TelegramBot(
name="기본 봇 (Config)",
token=cfg_token,
chat_id=cfg_chat,
is_active=True,
description="config.py 설정에서 자동 가져옴",
notification_types="auth,activity,system"
)
db.session.add(new_bot)
db.session.commit()
bots = [new_bot]
current_app.logger.info("Telegram: Config settings migrated to DB.")
if not bots:
return
# 카테고리 필터링
target_bots = []
for b in bots:
allowed = (b.notification_types or "auth,activity,system").split(",")
if category in allowed:
target_bots.append(b)
if not target_bots:
return
if not (Bot and ParseMode):
current_app.logger.warning("Telegram: Library not installed.")
return
app = current_app._get_current_object()
async def _send_to_bot(bot_obj, msg):
try:
t_bot = Bot(token=bot_obj.token)
await t_bot.send_message(chat_id=bot_obj.chat_id, text=msg, parse_mode=ParseMode.HTML)
except Exception as e:
app.logger.error(f"Telegram fail ({bot_obj.name}): {e}")
async def _send_all():
tasks = [_send_to_bot(b, text) for b in target_bots]
await asyncio.gather(*tasks)
def _run_thread():
try:
asyncio.run(_send_all())
except Exception as e:
app.logger.error(f"Telegram async loop error: {e}")
threading.Thread(target=_run_thread, daemon=True).start()
except Exception as e:
current_app.logger.error(f"Telegram notification error: {e}")
def _notify_with_buttons(text: str, buttons: list, category: str = "auth") -> None:
"""
텔레그램 알림 전송 (인라인 버튼 포함)
- buttons: [(text, callback_data), ...] 형식의 리스트
"""
try:
from backend.models.telegram_bot import TelegramBot
try:
bots = TelegramBot.query.filter_by(is_active=True).all()
except Exception:
db.create_all()
bots = []
if not bots:
cfg_token = (current_app.config.get("TELEGRAM_BOT_TOKEN") or "").strip()
cfg_chat = (current_app.config.get("TELEGRAM_CHAT_ID") or "").strip()
if cfg_token and cfg_chat:
new_bot = TelegramBot(
name="기본 봇 (Config)",
token=cfg_token,
chat_id=cfg_chat,
is_active=True,
description="config.py 설정에서 자동 가져옴",
notification_types="auth,activity,system"
)
db.session.add(new_bot)
db.session.commit()
bots = [new_bot]
if not bots:
return
target_bots = []
for b in bots:
allowed = (b.notification_types or "auth,activity,system").split(",")
if category in allowed:
target_bots.append(b)
if not target_bots:
return
if not (Bot and ParseMode and InlineKeyboardButton and InlineKeyboardMarkup):
current_app.logger.warning("Telegram: Library not installed.")
return
# 인라인 키보드 생성
keyboard = [[InlineKeyboardButton(btn[0], callback_data=btn[1]) for btn in buttons]]
reply_markup = InlineKeyboardMarkup(keyboard)
app = current_app._get_current_object()
async def _send_to_bot(bot_obj, msg, markup):
try:
t_bot = Bot(token=bot_obj.token)
await t_bot.send_message(
chat_id=bot_obj.chat_id,
text=msg,
parse_mode=ParseMode.HTML,
reply_markup=markup
)
except Exception as e:
app.logger.error(f"Telegram fail ({bot_obj.name}): {e}")
async def _send_all():
tasks = [_send_to_bot(b, text, reply_markup) for b in target_bots]
await asyncio.gather(*tasks)
def _run_thread():
try:
asyncio.run(_send_all())
except Exception as e:
app.logger.error(f"Telegram async loop error: {e}")
threading.Thread(target=_run_thread, daemon=True).start()
except Exception as e:
current_app.logger.error(f"Telegram notification with buttons error: {e}")
# ─────────────────────────────────────────────────────────────
# Blueprint 등록 훅
# ─────────────────────────────────────────────────────────────
def register_auth_routes(app):
"""app.py에서 register_routes(app, socketio) 호출 시 사용."""
app.register_blueprint(auth_bp)
@app.before_request
def _global_hooks():
# 1. 세션 갱신 (요청마다 세션 타임아웃 연장)
session.modified = True
# 2. 활동 알림 (로그인된 사용자)
if current_user.is_authenticated:
# 정적 리소스 및 불필요한 경로 제외
if request.endpoint == 'static':
return
# 제외할 확장자/경로 (필요 시 추가)
ignored_exts = ('.css', '.js', '.png', '.jpg', '.jpeg', '.gif', '.ico', '.woff', '.woff2')
if request.path.lower().endswith(ignored_exts):
return
# 활동 내용 구성
msg = (
f"👣 <b>활동 감지</b>\n"
f"👤 <code>{current_user.username}</code>\n"
f"📍 <code>{request.method} {request.path}</code>"
)
_notify(msg, category="activity")
# ─────────────────────────────────────────────────────────────
# 회원가입
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/register", methods=["GET", "POST"])
def register():
if current_user.is_authenticated:
current_app.logger.info("REGISTER: already auth → /index")
return redirect(url_for("main.index"))
form = RegistrationForm()
if form.validate_on_submit():
# 모델 내부에서 email/username 정규화됨(find_by_*)
if User.find_by_email(form.email.data):
flash("이미 등록된 이메일입니다.", "warning")
current_app.logger.info("REGISTER: dup email %s", form.email.data)
return render_template("register.html", form=form)
if User.find_by_username(form.username.data):
flash("이미 사용 중인 사용자명입니다.", "warning")
current_app.logger.info("REGISTER: dup username %s", form.username.data)
return render_template("register.html", form=form)
# 승인 토큰 생성
approval_token = secrets.token_urlsafe(32)
user = User(
username=form.username.data,
email=form.email.data,
is_active=False,
is_approved=False,
approval_token=approval_token
)
user.set_password(form.password.data)
try:
db.session.add(user)
db.session.commit()
except Exception as e:
db.session.rollback()
current_app.logger.error("REGISTER: DB commit failed: %s", e)
flash("회원가입 처리 중 오류가 발생했습니다. (DB Error)", "danger")
return render_template("register.html", form=form)
# 텔레그램 알림 (인라인 버튼 포함)
message = (
f"🆕 <b>신규 가입 요청</b>\n\n"
f" 사용자: <code>{user.username}</code>\n"
f"📧 이메일: <code>{user.email}</code>\n"
f"🕐 신청시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
)
buttons = [
("✅ 승인", f"approve_{approval_token}"),
("❌ 거부", f"reject_{approval_token}")
]
_notify_with_buttons(message, buttons, category="auth")
current_app.logger.info("REGISTER: created id=%s email=%s token=%s", user.id, user.email, approval_token[:10])
flash("회원가입이 완료되었습니다. 관리자의 승인을 기다려주세요.", "success")
return redirect(url_for("auth.login"))
else:
if request.method == "POST":
# 폼 검증 실패 에러를 Flash 메시지로 출력
for field_name, errors in form.errors.items():
for error in errors:
# 필드 객체 가져오기 (라벨 텍스트 확인용)
field = getattr(form, field_name, None)
label = field.label.text if field else field_name
flash(f"{label}: {error}", "warning")
current_app.logger.info("REGISTER: form errors=%s", form.errors)
return render_template("register.html", form=form)
# ─────────────────────────────────────────────────────────────
# 로그인
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/login", methods=["GET", "POST"])
def login():
if current_user.is_authenticated:
current_app.logger.info("LOGIN: already auth → /index")
return redirect(url_for("main.index"))
form = LoginForm()
if form.validate_on_submit():
current_app.logger.info("LOGIN: form ok email=%s", form.email.data)
user: Optional[User] = User.find_by_email(form.email.data)
if not user:
flash("이메일 또는 비밀번호가 올바르지 않습니다.", "danger")
current_app.logger.info("LOGIN: user not found")
return render_template("login.html", form=form)
pass_ok = user.check_password(form.password.data)
current_app.logger.info(
"LOGIN: found id=%s active=%s approved=%s pass_ok=%s",
user.id, user.is_active, user.is_approved, pass_ok
)
if not pass_ok:
flash("이메일 또는 비밀번호가 올바르지 않습니다.", "danger")
return render_template("login.html", form=form)
if not user.is_approved:
flash("계정이 아직 승인되지 않았습니다. 관리자의 승인을 기다려주세요.", "warning")
return render_template("login.html", form=form)
if not user.is_active:
flash("계정이 비활성화되었습니다. 관리자에게 문의하세요.", "warning")
return render_template("login.html", form=form)
# 성공
login_user(user, remember=form.remember.data)
session.permanent = True
_notify(f"🔐 <b>로그인 성공</b>\n👤 <code>{user.username}</code>", category="auth")
current_app.logger.info("LOGIN: SUCCESS → redirect")
nxt = request.args.get("next")
if nxt and _is_safe_url(nxt):
return redirect(nxt)
return redirect(url_for("main.index"))
else:
if request.method == "POST":
current_app.logger.info("LOGIN: form errors=%s", form.errors)
return render_template("login.html", form=form)
# ─────────────────────────────────────────────────────────────
# 로그아웃
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/logout", methods=["GET"])
@login_required
def logout():
if current_user.is_authenticated:
current_app.logger.info("LOGOUT: user=%s", current_user.username)
_notify(f"🚪 <b>로그아웃</b>\n👤 <code>{current_user.username}</code>", category="auth")
logout_user()
flash("정상적으로 로그아웃 되었습니다.", "success")
return redirect(url_for("auth.login"))
+789
View File
@@ -0,0 +1,789 @@
"""
BIOS Baseline 관리 라우트
backend/routes/bios_baseline_routes.py
BIOS Baseline 생성, 조회, 수정, 삭제 및 비교 기능 제공
"""
from flask import Blueprint, render_template, request, jsonify, current_app
from backend.models.user import db
from backend.models.bios_baseline import BiosBaseline
from backend.services.idrac_redfish_client import DellRedfishClient
from datetime import datetime
import json
from sqlalchemy import text
# 비교 제외 대상 키 (서버별 고유값 등)
IGNORED_KEYS = {
'BIOS': {
'ServiceTag', 'AssetTag', 'SerialNumber', 'SystemServiceTag', 'SystemAssetTag', 'SystemSerialNumber'
},
'iDRAC': {
'SystemID', 'DNSDomainName', 'DNSRacName', 'DNSDomainNameFromDHCP',
'IPv4.1.Address', 'IPv4.1.Gateway', 'IPv4.1.Netmask',
'IPv4Static.1.Address', 'IPv4Static.1.Gateway', 'IPv4Static.1.Netmask',
'IPv6.1.Address', 'IPv6.1.Gateway',
'MACAddress', 'PermanentMACAddress'
}
}
def compare_dictionaries(expected, actual, section_name, diff_list, metadata=None):
"""딕셔너리 비교 헬퍼 함수"""
if not expected:
return 0, 0
matched = 0
mismatched = 0
ignored = IGNORED_KEYS.get(section_name, set())
for key, exp_val in expected.items():
if key in ignored:
continue
act_val = actual.get(key)
# 문자열로 변환하여 비교 (타입 불일치 방지)
if str(act_val) == str(exp_val):
matched += 1
# 일치하는 항목도 결과에 포함
diff_item = {
'section': section_name,
'setting_name': key,
'expected': exp_val,
'actual': act_val,
'status': 'match'
}
if metadata and key in metadata:
diff_item['display_name'] = metadata[key]
diff_list.append(diff_item)
else:
mismatched += 1
diff_item = {
'section': section_name,
'setting_name': key,
'expected': exp_val,
'actual': act_val,
'status': 'mismatch'
}
# Display Name 추가
if metadata and key in metadata:
diff_item['display_name'] = metadata[key]
diff_list.append(diff_item)
return matched, mismatched
def compare_raid_config(expected, actual, diff_list):
"""RAID 구성 비교"""
if not expected or 'Controllers' not in expected:
return 0, 0
matched = 0
mismatched = 0
total_checks = 0
# 컨트롤러별 비교
exp_ctrls = {c.get('Name'): c for c in expected.get('Controllers', [])}
act_ctrls = {c.get('Name'): c for c in actual.get('Controllers', [])}
for name, exp_c in exp_ctrls.items():
act_c = act_ctrls.get(name)
if not act_c:
mismatched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Controller [{name}]",
'expected': "Present",
'actual': "Missing",
'status': 'mismatch'
})
continue
matched += 1 # 컨트롤러 존재함
diff_list.append({
'section': 'RAID',
'setting_name': f"Controller [{name}]",
'expected': "Present",
'actual': "Present",
'status': 'match'
})
# 펌웨어 버전 비교
exp_fw = exp_c.get('FirmwareVersion')
act_fw = act_c.get('FirmwareVersion')
if exp_fw and (exp_fw == act_fw):
matched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Controller [{name}] Firmware",
'expected': exp_fw,
'actual': act_fw,
'status': 'match'
})
else:
mismatched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Controller [{name}] Firmware",
'expected': exp_fw,
'actual': act_fw,
'status': 'mismatch'
})
# Volumes (Virtual Disks) 비교
exp_vols = {v.get('Name'): v for v in exp_c.get('Volumes', [])}
act_vols = {v.get('Name'): v for v in act_c.get('Volumes', [])}
# Volume 개수 확인
if len(exp_vols) != len(act_vols):
diff_list.append({
'section': 'RAID',
'setting_name': f"Controller [{name}] Volume Count",
'expected': len(exp_vols),
'actual': len(act_vols),
'status': 'warning'
})
for vname, exp_v in exp_vols.items():
act_v = act_vols.get(vname)
if not act_v:
mismatched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Volume [{name}/{vname}]",
'expected': "Present",
'actual': "Missing",
'status': 'mismatch'
})
continue
# RAID Level 비교
if exp_v.get('RaidType') == act_v.get('RaidType'):
matched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Volume [{name}/{vname}] RAID Level",
'expected': exp_v.get('RaidType'),
'actual': act_v.get('RaidType'),
'status': 'match'
})
else:
mismatched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Volume [{name}/{vname}] RAID Level",
'expected': exp_v.get('RaidType'),
'actual': act_v.get('RaidType'),
'status': 'mismatch'
})
# 용량 비교 (오차 허용 없이)
if exp_v.get('CapacityBytes') == act_v.get('CapacityBytes'):
matched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Volume [{name}/{vname}] Capacity",
'expected': f"{exp_v.get('CapacityBytes', 0) / (1024**3):.2f} GB",
'actual': f"{act_v.get('CapacityBytes', 0) / (1024**3):.2f} GB",
'status': 'match'
})
else:
mismatched += 1
diff_list.append({
'section': 'RAID',
'setting_name': f"Volume [{name}/{vname}] Capacity",
'expected': f"{exp_v.get('CapacityBytes', 0) / (1024**3):.2f} GB",
'actual': f"{act_v.get('CapacityBytes', 0) / (1024**3):.2f} GB",
'status': 'mismatch'
})
return matched, mismatched
bios_baseline_bp = Blueprint('bios_baseline', __name__, url_prefix='/bios-baseline')
# ========================================
# 페이지 라우트
# ========================================
@bios_baseline_bp.route('/')
def index():
"""BIOS Baseline 비교 메인 페이지"""
return render_template('bios_baseline.html')
# ========================================
# Baseline 관리 API
# ========================================
@bios_baseline_bp.route('/api/baselines', methods=['GET'])
def get_baselines():
"""
등록된 baseline 목록 조회
Query params:
- server_model: 특정 모델 필터링
- server_type: 특정 타입 필터링
- include_data: true면 bios_data도 포함
"""
try:
server_model = request.args.get('server_model')
server_type = request.args.get('server_type')
include_data = request.args.get('include_data', 'false').lower() == 'true'
query = BiosBaseline.query.filter_by(is_active=True)
if server_model:
query = query.filter_by(server_model=server_model)
if server_type:
query = query.filter_by(server_type=server_type)
baselines = query.order_by(BiosBaseline.created_at.desc()).all()
return jsonify({
'success': True,
'baselines': [b.to_dict(include_data=include_data) for b in baselines]
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
@bios_baseline_bp.route('/api/baselines/<int:baseline_id>', methods=['GET'])
def get_baseline(baseline_id):
"""특정 baseline 상세 조회 (bios_data 포함)"""
try:
baseline = BiosBaseline.query.get(baseline_id)
if not baseline:
return jsonify({'success': False, 'message': 'Baseline을 찾을 수 없습니다'})
return jsonify({
'success': True,
'baseline': baseline.to_dict(include_data=True)
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
def compare_nic_config(baseline_nic, target_nic, diff_list):
"""NIC 구성 비교 (ID 기준 매칭)"""
matched = 0
mismatched = 0
if not baseline_nic and not target_nic:
return 0, 0
if not baseline_nic:
diff_list.append({'section': 'NIC', 'setting_name': 'NIC Configuration', 'expected': 'Present', 'actual': 'Missing', 'status': 'mismatch'})
return 0, 1
if not target_nic:
diff_list.append({'section': 'NIC', 'setting_name': 'NIC Configuration', 'expected': 'Missing', 'actual': 'Present', 'status': 'mismatch'})
return 0, 1
# Adapters 리스트를 ID 기준 딕셔너리로 변환
def to_dict_by_id(adapters):
return {a.get('Id', 'Unknown'): a for a in adapters.get('Adapters', [])}
base_map = to_dict_by_id(baseline_nic)
target_map = to_dict_by_id(target_nic)
all_ids = set(base_map.keys()) | set(target_map.keys())
for aid in all_ids:
b_adp = base_map.get(aid)
t_adp = target_map.get(aid)
if not b_adp:
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"NIC [{aid}]", 'expected': 'None', 'actual': 'Added', 'status': 'mismatch'})
continue
if not t_adp:
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"NIC [{aid}]", 'expected': 'Present', 'actual': 'Missing', 'status': 'mismatch'})
continue
matched += 1 # Adapter exists in both
diff_list.append({'section': 'NIC', 'setting_name': f"NIC [{aid}]", 'expected': 'Present', 'actual': 'Present', 'status': 'match'})
# 어댑터 속성 비교 (Firmware 등)
for field in ['FirmwareVersion', 'Model']:
bv = b_adp.get(field)
tv = t_adp.get(field)
if bv == tv:
matched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"NIC [{aid}] {field}", 'expected': bv, 'actual': tv, 'status': 'match'})
else:
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"NIC [{aid}] {field}", 'expected': bv, 'actual': tv, 'status': 'mismatch'})
# DeviceFunctions 비교
b_funcs = {f.get('Id'): f for f in b_adp.get('DeviceFunctions', [])}
t_funcs = {f.get('Id'): f for f in t_adp.get('DeviceFunctions', [])}
for fid in set(b_funcs.keys()) | set(t_funcs.keys()):
bf = b_funcs.get(fid)
tf = t_funcs.get(fid)
prefix = f"NIC [{aid}] Func [{fid}]"
if not bf:
# Baseline에 없는데 Target에 있음 (일반적이지 않음, 설정 변경일 수 있음)
# This is an 'added' function, count as mismatch
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': prefix, 'expected': 'None', 'actual': 'Added', 'status': 'mismatch'})
continue
if not tf:
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': prefix, 'expected': 'Present', 'actual': 'Missing', 'status': 'mismatch'})
continue
matched += 1 # Function exists in both
diff_list.append({'section': 'NIC', 'setting_name': prefix, 'expected': 'Present', 'actual': 'Present', 'status': 'match'})
# 속성 비교 (BootMode, iSCSI, Ethernet)
if bf.get('NetBootPMMode') == tf.get('NetBootPMMode'):
matched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"{prefix} BootMode", 'expected': bf.get('NetBootPMMode'), 'actual': tf.get('NetBootPMMode'), 'status': 'match'})
else:
mismatched += 1
diff_list.append({'section': 'NIC', 'setting_name': f"{prefix} BootMode", 'expected': bf.get('NetBootPMMode'), 'actual': tf.get('NetBootPMMode'), 'status': 'mismatch'})
# Add more specific NIC settings comparison here if needed
# For example, comparing specific Ethernet or iSCSI settings
# For now, we'll consider the presence and BootMode.
return matched, mismatched
@bios_baseline_bp.route('/api/baselines/create-from-server', methods=['POST'])
def create_baseline_from_server():
"""
Redfish로 서버에서 BIOS 설정을 가져와 새 baseline 생성
Request: {
"ip_address": "10.10.0.1",
"username": "root",
"password": "calvin",
"name": "R6615_GPU_Standard_2024",
"server_type": "GPU Server",
"description": "표준 GPU 서버 설정",
"created_by": "admin"
}
"""
try:
data = request.json
# 필수 필드 체크
if not all(k in data for k in ['ip_address', 'username', 'password', 'name']):
return jsonify({'success': False, 'message': '필수 필드 누락'})
# 이름 중복 체크
if BiosBaseline.query.filter_by(name=data['name'], is_active=True).first():
return jsonify({'success': False, 'message': '이미 존재하는 Baseline 이름입니다'})
# 1. Redfish로 BIOS 설정 가져오기
client = DellRedfishClient(data['ip_address'], data['username'], data['password'])
if not client.check_connection():
return jsonify({'success': False, 'message': '서버 연결 실패'})
# 2. 서버 정보 및 BIOS 설정 가져오기
system_info = client.get_system_info_detailed()
bios_data = client.get_bios_attributes_all()
# BIOS Registry (Display Name) 가져오기
bios_metadata = client.get_bios_attribute_registry()
idrac_data = client.get_idrac_attributes_all()
# iDRAC Registry (Display Name) 가져오기
idrac_metadata = client.get_idrac_attribute_registry()
raid_data = client.get_storage_configuration()
nic_data = client.get_nic_configuration()
boot_data = client.get_boot_settings()
firmware_data = client.get_firmware_baseline()
if not bios_data:
return jsonify({'success': False, 'message': 'BIOS 설정 가져오기 실패'})
# 3. Baseline 생성
baseline = BiosBaseline(
name=data['name'],
server_model=system_info.get('Model', 'Unknown'),
server_type=data.get('server_type'),
description=data.get('description'),
bios_data=bios_data,
bios_metadata=bios_metadata, # 메타데이터 저장
idrac_data=idrac_data,
idrac_metadata=idrac_metadata, # iDRAC 메타데이터 저장
raid_data=raid_data,
nic_data=nic_data,
boot_data=boot_data,
firmware_data=firmware_data,
source_ip=data['ip_address'],
source_service_tag=system_info.get('ServiceTag'),
bios_version=bios_data.get('BiosVersion'),
total_settings=len(bios_data) + len(idrac_data) + len(raid_data.get('Controllers', [])) + len(nic_data.get('Adapters', [])) + len(boot_data.get('BootSources', [])) + len(firmware_data.get('FirmwareInventory', [])),
created_by=data.get('created_by', 'admin')
)
db.session.add(baseline)
db.session.commit()
return jsonify({
'success': True,
'message': 'Baseline 생성 완료',
'baseline': baseline.to_dict()
})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': f'오류: {str(e)}'})
@bios_baseline_bp.route('/api/baselines/<int:baseline_id>/duplicate', methods=['POST'])
def duplicate_baseline(baseline_id):
"""
기존 baseline을 복사하여 새 baseline 생성
Request: {
"new_name": "R6615_GPU_ClientA",
"description": "고객사 A 맞춤 설정"
}
"""
try:
data = request.json
original = BiosBaseline.query.get(baseline_id)
if not original:
return jsonify({'success': False, 'message': '원본 Baseline을 찾을 수 없습니다'})
# 이름 중복 체크
if BiosBaseline.query.filter_by(name=data['new_name'], is_active=True).first():
return jsonify({'success': False, 'message': '이미 존재하는 Baseline 이름입니다'})
# 복사 생성
new_baseline = BiosBaseline(
name=data['new_name'],
server_model=original.server_model,
server_type=original.server_type,
description=data.get('description', f"{original.name}의 복사본"),
bios_data=original.bios_data.copy() if original.bios_data else {},
bios_metadata=original.bios_metadata.copy() if original.bios_metadata else {}, # 메타데이터 복사
idrac_data=original.idrac_data.copy() if original.idrac_data else {},
idrac_metadata=original.idrac_metadata.copy() if original.idrac_metadata else {}, # iDRAC 메타데이터 복사
raid_data=original.raid_data.copy() if original.raid_data else {}, # JSON 복사
nic_data=original.nic_data.copy() if original.nic_data else {},
boot_data=original.boot_data.copy() if original.boot_data else {},
firmware_data=original.firmware_data.copy() if original.firmware_data else {},
source_ip=original.source_ip,
source_service_tag=original.source_service_tag,
bios_version=original.bios_version,
total_settings=original.total_settings,
copied_from_id=original.id,
created_by=data.get('created_by', 'admin')
)
db.session.add(new_baseline)
db.session.commit()
return jsonify({
'success': True,
'message': 'Baseline 복사 완료',
'baseline': new_baseline.to_dict()
})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': f'오류: {str(e)}'})
@bios_baseline_bp.route('/api/baselines/<int:baseline_id>', methods=['PUT'])
def update_baseline(baseline_id):
"""
Baseline 수정 (메타데이터 및 bios_data)
Request: {
"name": "새 이름",
"description": "새 설명",
"bios_data": { ... } // 선택사항
}
"""
try:
baseline = BiosBaseline.query.get_or_404(baseline_id)
data = request.json
# 이름 변경 시 중복 체크
if 'name' in data and data['name'] != baseline.name:
if BiosBaseline.query.filter_by(name=data['name'], is_active=True).first():
return jsonify({'success': False, 'message': '이미 존재하는 Baseline 이름입니다'})
baseline.name = data['name']
# 메타데이터 업데이트
if 'description' in data:
baseline.description = data['description']
if 'server_type' in data:
baseline.server_type = data['server_type']
# BIOS 데이터 업데이트 (선택사항)
if 'bios_data' in data:
baseline.bios_data = data['bios_data']
if 'bios_metadata' in data:
baseline.bios_metadata = data['bios_metadata']
if 'idrac_data' in data:
baseline.idrac_data = data['idrac_data']
if 'idrac_metadata' in data:
baseline.idrac_metadata = data['idrac_metadata']
if 'raid_data' in data:
baseline.raid_data = data['raid_data']
if 'nic_data' in data:
baseline.nic_data = data['nic_data']
if 'boot_data' in data:
baseline.boot_data = data['boot_data']
if 'firmware_data' in data:
baseline.firmware_data = data['firmware_data']
# 총 설정 수 재계산
total = 0
if baseline.bios_data: total += len(baseline.bios_data)
if baseline.idrac_data: total += len(baseline.idrac_data)
if baseline.raid_data and 'Controllers' in baseline.raid_data: total += len(baseline.raid_data['Controllers'])
if baseline.nic_data and 'Adapters' in baseline.nic_data: total += len(baseline.nic_data['Adapters'])
if baseline.boot_data and 'BootSources' in baseline.boot_data: total += len(baseline.boot_data['BootSources'])
if baseline.firmware_data and 'FirmwareInventory' in baseline.firmware_data: total += len(baseline.firmware_data['FirmwareInventory'])
baseline.total_settings = total
baseline.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({
'success': True,
'message': 'Baseline 수정 완료',
'baseline': baseline.to_dict()
})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': f'오류: {str(e)}'})
@bios_baseline_bp.route('/api/baselines/<int:baseline_id>', methods=['DELETE'])
def delete_baseline(baseline_id):
"""Baseline 삭제 (Soft Delete)"""
try:
baseline = BiosBaseline.query.get(baseline_id)
if not baseline:
return jsonify({'success': False, 'message': 'Baseline을 찾을 수 없습니다'})
baseline.is_active = False
# 이름 충돌 방지를 위해 이름 변경 (Soft Delete 시)
# 예: "MyServer" -> "MyServer_deleted_1678901234"
baseline.name = f"{baseline.name}_deleted_{int(datetime.utcnow().timestamp())}"
db.session.commit()
return jsonify({'success': True, 'message': 'Baseline 삭제 완료'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
# ========================================
# 비교 API
# ========================================
@bios_baseline_bp.route('/api/compare', methods=['POST'])
def compare_with_baseline():
"""
IP 주소로 서버 BIOS 설정을 가져와 baseline과 비교
Request: {
"ip_addresses": ["10.10.0.1", "10.10.0.2"],
"baseline_id": 1,
"username": "root",
"password": "calvin"
}
"""
try:
data = request.json
baseline = BiosBaseline.query.get(data['baseline_id'])
if not baseline:
return jsonify({'success': False, 'message': 'Baseline을 찾을 수 없습니다'})
results = []
# 병렬 처리를 위한 함수 정의
def process_server(ip):
try:
# 1. Redfish로 현재 BIOS 설정 가져오기
client = DellRedfishClient(ip, data['username'], data['password'])
if not client.check_connection():
return {
'ip': ip,
'status': 'connection_failed',
'message': '서버 연결 실패'
}
# 2. Redfish로 현재 설정 가져오기
system_info = client.get_system_info_detailed()
current_bios = client.get_bios_attributes_all()
current_idrac = client.get_idrac_attributes_all()
current_raid = client.get_storage_configuration()
current_nic = client.get_nic_configuration()
current_boot = client.get_boot_settings()
current_firmware = client.get_firmware_baseline()
if not current_bios:
return {
'ip': ip,
'status': 'error',
'message': 'BIOS 설정 가져오기 실패'
}
# 3. Baseline과 비교
differences = []
total_matched = 0
total_mismatched = 0
# 3-1. BIOS 비교
b_matched, b_mismatched = compare_dictionaries(baseline.bios_data, current_bios, 'BIOS', differences, baseline.bios_metadata)
total_matched += b_matched
total_mismatched += b_mismatched
# 3-2. iDRAC 비교
if baseline.idrac_data:
i_matched, i_mismatched = compare_dictionaries(baseline.idrac_data, current_idrac, 'iDRAC', differences, baseline.idrac_metadata)
total_matched += i_matched
total_mismatched += i_mismatched
# 3-3. RAID 비교
if baseline.raid_data:
r_matched, r_mismatched = compare_raid_config(baseline.raid_data, current_raid, differences)
total_matched += r_matched
total_mismatched += r_mismatched
# 3-4. NIC 비교
if baseline.nic_data:
n_matched, n_mismatched = compare_nic_config(baseline.nic_data, current_nic, differences)
total_matched += n_matched
total_mismatched += n_mismatched
# 3-5. Boot 비교
if baseline.boot_data:
boot_matched, boot_mismatched = compare_dictionaries(baseline.boot_data, current_boot, 'Boot', differences)
total_matched += boot_matched
total_mismatched += boot_mismatched
# 3-6. Firmware 비교
if baseline.firmware_data:
fw_matched, fw_mismatched = compare_dictionaries(baseline.firmware_data, current_firmware, 'Firmware', differences)
total_matched += fw_matched
total_mismatched += fw_mismatched
# 4. 결과 저장
match_rate = round((total_matched / (total_matched + total_mismatched)) * 100, 2) if (total_matched + total_mismatched) > 0 else 0
return {
'ip': ip,
'status': 'success',
'server_model': system_info.get('Model'),
'service_tag': system_info.get('ServiceTag'),
'bios_version': current_bios.get('BiosVersion'), # Corrected from system_info.get('BiosVersion')
'total_items': total_matched + total_mismatched,
'matched_items': total_matched,
'mismatched_items': total_mismatched,
'match_rate': match_rate,
'differences': differences
}
except Exception as e:
import traceback
traceback.print_exc() # Keep traceback for debugging in logs
return {
'ip': ip,
'status': 'error',
'message': str(e)
}
import concurrent.futures
results = []
max_workers = min(len(data['ip_addresses']), 20) # 최대 20개 동시 실행
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_ip = {executor.submit(process_server, ip): ip for ip in data['ip_addresses']}
for future in concurrent.futures.as_completed(future_to_ip):
ip = future_to_ip[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append({
'ip': ip,
'status': 'error',
'message': f"예상치 못한 오류: {str(e)}"
})
return jsonify({
'success': True,
'baseline': baseline.to_dict(),
'results': results
})
except Exception as e:
return jsonify({'success': False, 'message': f'오류: {str(e)}'})
# ========================================
# Blueprint 등록 함수
# ========================================
def register_bios_baseline_routes(app):
"""
bios_baseline_routes 등록 및 초기화
"""
app.register_blueprint(bios_baseline_bp)
# CSRF 제외
try:
csrf = app.extensions.get('csrf')
if csrf:
csrf.exempt(bios_baseline_bp)
except Exception as e:
app.logger.warning(f"CSRF exemption failed for bios_baseline_bp: {e}")
# DB 마이그레이션: 필요한 컬럼이 없으면 추가
# SQLite 등에서 ALTER TABLE은 제한적일 수 있으나 여기서는 단순 컬럼 추가 시도
with app.app_context():
inspector = db.inspect(db.engine)
# Ensure table exists before checking columns
db.create_all()
columns = [c['name'] for c in inspector.get_columns('bios_baselines')]
if 'idrac_data' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding idrac_data")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN idrac_data JSON"))
conn.commit()
if 'bios_metadata' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding bios_metadata")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN bios_metadata JSON"))
conn.commit()
if 'idrac_metadata' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding idrac_metadata")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN idrac_metadata JSON"))
conn.commit()
if 'raid_data' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding raid_data")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN raid_data JSON"))
conn.commit()
if 'nic_data' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding nic_data")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN nic_data JSON"))
conn.commit()
if 'boot_data' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding boot_data")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN boot_data JSON"))
conn.commit()
if 'firmware_data' not in columns:
print("[INFO] Migrating BIOS Baseline table: Adding firmware_data")
with db.engine.connect() as conn:
conn.execute(text("ALTER TABLE bios_baselines ADD COLUMN firmware_data JSON"))
conn.commit()
+23
View File
@@ -0,0 +1,23 @@
# backend/routes/catalog_sync.py
from flask import Blueprint, jsonify, request
from backend.services.dell_catalog_sync import sync_dell_catalog
catalog_bp = Blueprint("catalog", __name__, url_prefix="/catalog")
@catalog_bp.route("/sync", methods=["POST"])
def sync_catalog():
"""Dell Catalog 버전 정보를 동기화 (CSRF 보호 유지)"""
try:
data = request.get_json(silent=True) or {}
model = data.get("model", "PowerEdge R750")
sync_dell_catalog(model)
return jsonify({
"success": True,
"message": f"{model} 동기화 완료"
})
except Exception as e:
return jsonify({
"success": False,
"message": f"오류: {str(e)}"
})
+61
View File
@@ -0,0 +1,61 @@
"""
DRM 카탈로그 동기화 라우트
backend/routes/drm_sync.py
"""
from flask import Blueprint, request, jsonify
from backend.services.drm_catalog_sync import sync_from_drm, check_drm_repository
drm_sync_bp = Blueprint('drm_sync', __name__, url_prefix='/drm')
@drm_sync_bp.route('/check', methods=['POST'])
def check_repository():
"""DRM 리포지토리 상태 확인"""
try:
data = request.get_json() or {}
repository_path = data.get('repository_path')
if not repository_path:
return jsonify({
'success': False,
'message': 'repository_path가 필요합니다'
}), 400
info = check_drm_repository(repository_path)
return jsonify({
'success': info.get('exists', False),
'info': info
})
except Exception as e:
return jsonify({
'success': False,
'message': f'오류: {str(e)}'
}), 500
@drm_sync_bp.route('/sync', methods=['POST'])
def sync_repository():
"""DRM 리포지토리에서 펌웨어 동기화"""
try:
data = request.get_json() or {}
repository_path = data.get('repository_path')
model = data.get('model', 'PowerEdge R750')
if not repository_path:
return jsonify({
'success': False,
'message': 'repository_path가 필요합니다'
}), 400
result = sync_from_drm(repository_path, model)
return jsonify(result)
except Exception as e:
return jsonify({
'success': False,
'message': f'오류: {str(e)}'
}), 500
+109
View File
@@ -0,0 +1,109 @@
from __future__ import annotations
import logging
from pathlib import Path
from flask import Blueprint, request, jsonify, Response
from flask_login import login_required
from config import Config
import chardet
file_view_bp = Blueprint("file_view", __name__)
def register_file_view(app):
"""블루프린트 등록"""
app.register_blueprint(file_view_bp)
def _safe_within(base: Path, target: Path) -> bool:
"""
target 이 base 디렉터리 내부인지 검사 (경로 탈출 방지)
"""
try:
target.resolve().relative_to(base.resolve())
return True
except Exception:
return False
def _decode_bytes(raw: bytes) -> str:
"""
파일 바이트 → 문자열 디코딩 (감지 → utf-8 → cp949 순서로 시도)
"""
enc = (chardet.detect(raw).get("encoding") or "utf-8").strip().lower()
for cand in (enc, "utf-8", "cp949"):
try:
return raw.decode(cand)
except Exception:
continue
# 최후의 수단: 손실 허용 디코딩
return raw.decode("utf-8", errors="replace")
@file_view_bp.route("/view_file", methods=["GET"])
@login_required
def view_file():
"""
파일 내용을 읽어 반환.
- /view_file?folder=idrac_info&filename=abc.txt
- /view_file?folder=backup&date=<백업폴더명>&filename=abc.txt
- ?raw=1 을 붙이면 text/plain 으로 원문을 반환 (모달 표시용)
"""
folder = request.args.get("folder", "").strip()
date = request.args.get("date", "").strip()
filename = request.args.get("filename", "").strip()
want_raw = request.args.get("raw")
if not filename:
return jsonify({"error": "파일 이름이 없습니다."}), 400
# 파일명/폴더명은 유니코드 보존. 상위 경로만 제거하여 보안 유지
safe_name = Path(filename).name
if folder == "backup":
base = Path(Config.BACKUP_FOLDER)
safe_date = Path(date).name if date else ""
target = (base / safe_date / safe_name).resolve()
elif folder == "server_info":
base = Path(Config.BACKUP_FOLDER)
safe_date = Path(date).name if date else ""
target = (base / safe_date / safe_name).resolve()
elif folder == "mac":
base = Path(Config.MAC_FOLDER)
target = (base / safe_name).resolve()
elif folder == "guid":
base = Path(Config.GUID_FOLDER)
target = (base / safe_name).resolve()
elif folder == "gpu":
base = Path(Config.GPU_FOLDER)
target = (base / safe_name).resolve()
else:
# Default: Processed files (Staging)
base = Path(Config.IDRAC_INFO_FOLDER)
target = (base / safe_name).resolve()
logging.info(
"file_view: folder=%s date=%s filename=%s | base=%s | target=%s",
folder, date, filename, str(base), str(target)
)
if not _safe_within(base, target) or not target.is_file():
logging.warning("file_view: 파일 없음: %s", str(target))
return jsonify({"error": "파일을 찾을 수 없습니다."}), 404
try:
raw = target.read_bytes()
content = _decode_bytes(raw)
if want_raw:
# 텍스트 원문 반환 (모달에서 fetch().text()로 사용)
return Response(content, mimetype="text/plain; charset=utf-8")
# JSON으로 감싸서 반환 (기존 사용처 호환)
return jsonify({"content": content})
except Exception as e:
logging.error("file_view: 파일 읽기 실패: %s%s", str(target), e)
return jsonify({"error": "파일 열기 중 오류 발생"}), 500
+13
View File
@@ -0,0 +1,13 @@
from __future__ import annotations
from flask import Blueprint, render_template
home_bp = Blueprint("home", __name__, url_prefix="/home")
def register_home_routes(app):
app.register_blueprint(home_bp)
@home_bp.route("/", methods=["GET"])
def home():
return render_template("home.html")
+249
View File
@@ -0,0 +1,249 @@
"""
Dell iDRAC 멀티 서버 관리 라우트
backend/routes/idrac_routes.py
- CSRF 보호 제외 추가
- 펌웨어 관리 기능 제거됨 (Server Config Analysis로 대체)
"""
from flask import Blueprint, request, jsonify, current_app
from backend.services.idrac_redfish_client import DellRedfishClient
from backend.models.idrac_server import IdracServer, db
from datetime import datetime
import threading
# Blueprint 생성
idrac_bp = Blueprint('idrac', __name__, url_prefix='/idrac')
# ========================================
# 페이지 라우트
# ========================================
@idrac_bp.route('/')
def index():
"""iDRAC 서버 관리 메인 페이지"""
from flask import render_template
return render_template('idrac_firmware.html')
# ========================================
# 서버 관리 API (CRUD)
# ========================================
@idrac_bp.route('/api/servers', methods=['GET'])
def get_servers():
"""서버 목록 조회"""
try:
group_filter = request.args.get('group')
query = IdracServer.query.filter_by(is_active=True)
if group_filter and group_filter != 'all':
query = query.filter_by(group_name=group_filter)
servers = query.order_by(IdracServer.name).all()
return jsonify({
'success': True,
'servers': [s.to_dict() for s in servers]
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
@idrac_bp.route('/api/servers', methods=['POST'])
def add_server():
"""서버 추가"""
try:
data = request.json
# 필수 필드 체크
if not all(k in data for k in ['name', 'ip_address', 'password']):
return jsonify({'success': False, 'message': '필수 필드 누락'})
# 활성 서버 중복 체크
if IdracServer.query.filter_by(ip_address=data['ip_address'], is_active=True).first():
return jsonify({'success': False, 'message': '이미 등록된 IP입니다.'})
# Soft delete된 서버가 있는지 확인하고 재활성화
existing_server = IdracServer.query.filter_by(ip_address=data['ip_address'], is_active=False).first()
if existing_server:
# 기존 서버 재활성화 및 정보 업데이트
existing_server.is_active = True
existing_server.name = data['name']
existing_server.username = data.get('username', 'root')
existing_server.password = data['password']
existing_server.group_name = data.get('group_name')
existing_server.model = data.get('model')
existing_server.notes = data.get('notes')
existing_server.status = 'registered'
existing_server.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({'success': True, 'message': '서버 재등록 완료', 'server': existing_server.to_dict()})
# 새 서버 생성
new_server = IdracServer(
name=data['name'],
ip_address=data['ip_address'],
username=data.get('username', 'root'),
password=data['password'],
group_name=data.get('group_name'),
model=data.get('model'),
notes=data.get('notes')
)
db.session.add(new_server)
db.session.commit()
return jsonify({'success': True, 'message': '서버 추가 완료', 'server': new_server.to_dict()})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@idrac_bp.route('/api/servers/<int:server_id>', methods=['DELETE'])
def delete_server(server_id):
"""서버 삭제 (Soft Delete)"""
try:
server = IdracServer.query.get(server_id)
if not server:
return jsonify({'success': False, 'message': '서버를 찾을 수 없습니다'})
# Soft delete: is_active를 False로 설정
# IP 주소는 유지하여 재등록 시 재활성화 가능
server.is_active = False
server.status = 'deleted'
db.session.commit()
return jsonify({'success': True, 'message': '서버 삭제 완료'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@idrac_bp.route('/api/groups', methods=['GET'])
def get_groups():
"""등록된 그룹 목록"""
try:
groups = db.session.query(IdracServer.group_name).filter(
IdracServer.group_name != None,
IdracServer.is_active == True
).distinct().all()
return jsonify({
'success': True,
'groups': [g[0] for g in groups if g[0]]
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
# ========================================
# 연결 테스트 API
# ========================================
@idrac_bp.route('/api/servers/<int:server_id>/test', methods=['POST'])
def test_connection(server_id):
"""단일 서버 연결 테스트"""
try:
server = IdracServer.query.get(server_id)
if not server:
return jsonify({'success': False, 'message': '서버 없음'})
client = DellRedfishClient(server.ip_address, server.username, server.password)
if client.check_connection():
server.status = 'online'
server.last_connected = datetime.utcnow()
db.session.commit()
return jsonify({'success': True, 'message': '연결 성공'})
else:
server.status = 'offline'
db.session.commit()
return jsonify({'success': False, 'message': '연결 실패'})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
@idrac_bp.route('/api/servers/test-multi', methods=['POST'])
def test_connections_multi():
"""다중 서버 연결 테스트"""
try:
data = request.json
server_ids = data.get('server_ids', [])
results = []
for sid in server_ids:
server = IdracServer.query.get(sid)
if not server: continue
try:
client = DellRedfishClient(server.ip_address, server.username, server.password)
if client.check_connection():
server.status = 'online'
server.last_connected = datetime.utcnow()
results.append({'server_id': sid, 'success': True})
else:
server.status = 'offline'
results.append({'server_id': sid, 'success': False, 'message': 'Failed'})
except Exception as e:
server.status = 'offline'
results.append({'server_id': sid, 'success': False, 'message': str(e)})
db.session.commit()
success = sum(1 for r in results if r['success'])
return jsonify({
'success': True,
'summary': {'success': success, 'failed': len(results)-success},
'results': results
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
# ========================================
# 전원 제어 API
# ========================================
@idrac_bp.route('/api/servers/reboot-multi', methods=['POST'])
def reboot_servers_multi():
"""다중 서버 재부팅"""
try:
data = request.json
server_ids = data.get('server_ids', [])
reboot_type = data.get('type', 'GracefulRestart')
results = []
for sid in server_ids:
server = IdracServer.query.get(sid)
if not server: continue
try:
client = DellRedfishClient(server.ip_address, server.username, server.password)
if client.reboot_server(reboot_type):
results.append({'server_id': sid, 'success': True, 'message': 'Reboot initiated'})
else:
results.append({'server_id': sid, 'success': False, 'message': 'Reboot failed'})
except Exception as e:
results.append({'server_id': sid, 'success': False, 'message': str(e)})
success = sum(1 for r in results if r['success'])
return jsonify({
'success': True,
'summary': {'success': success, 'failed': len(results)-success},
'results': results
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
# ========================================
# Blueprint 등록 함수
# ========================================
def register_idrac_routes(app):
"""
idrac_routes 등록 및 초기화
"""
# Blueprint 등록
app.register_blueprint(idrac_bp)
# CSRF 제외 - app에 이미 설정된 csrf 인스턴스 사용
try:
from flask_wtf.csrf import CSRFProtect
# app에 이미 csrf가 설정되어 있으므로, 해당 인스턴스를 가져와서 exempt 설정
csrf = app.extensions.get('csrf')
if csrf:
csrf.exempt(idrac_bp)
except Exception as e:
app.logger.warning(f"CSRF exemption failed for idrac_bp: {e}")
# DB 생성
with app.app_context():
db.create_all()
+279
View File
@@ -0,0 +1,279 @@
"""
Flask Blueprint for iDRAC Job Monitoring (Redfish 버전)
기존 routes/jobs.py 또는 backend/routes/jobs.py를 이 파일로 교체하세요.
"""
import time
import logging
from flask import Blueprint, render_template, jsonify, request
from flask_login import login_required
from backend.services.idrac_jobs import (
scan_all,
parse_ip_list,
load_ip_list,
LRUJobCache,
is_active_status,
is_done_status,
parse_iso_datetime,
iso_now
)
import os
logger = logging.getLogger(__name__)
# Blueprint 생성
jobs_bp = Blueprint("jobs", __name__, url_prefix="/jobs")
# Job 캐시 (전역)
MAX_CACHE_SIZE = int(os.getenv("MAX_CACHE_SIZE", "10000"))
CACHE_GC_INTERVAL = int(os.getenv("CACHE_GC_INTERVAL", "3600"))
JOB_GRACE_MINUTES = int(os.getenv("JOB_GRACE_MINUTES", "60"))
JOB_RECENCY_HOURS = int(os.getenv("JOB_RECENCY_HOURS", "24"))
JOB_CACHE = LRUJobCache(max_size=MAX_CACHE_SIZE)
# ────────────────────────────────────────────────────────────
# Routes
# ────────────────────────────────────────────────────────────
@jobs_bp.route("", methods=["GET"])
@login_required
def jobs_page():
"""메인 페이지"""
return render_template("jobs.html")
@jobs_bp.route("/config", methods=["GET"])
@login_required
def jobs_config():
"""프론트엔드 설정 제공"""
return jsonify({
"ok": True,
"config": {
"grace_minutes": JOB_GRACE_MINUTES,
"recency_hours": JOB_RECENCY_HOURS,
"poll_interval_ms": int(os.getenv("POLL_INTERVAL_MS", "10000")),
}
})
@jobs_bp.route("/iplist", methods=["GET"])
@login_required
def get_ip_list():
"""IP 목록 조회 (파일에서)"""
try:
ips = load_ip_list()
return jsonify({
"ok": True,
"ips": ips,
"count": len(ips)
})
except Exception as e:
logger.exception("Failed to load IP list")
return jsonify({
"ok": False,
"error": str(e)
}), 500
@jobs_bp.route("/scan", methods=["POST"])
@login_required
def scan_jobs():
"""
Job 스캔 및 모니터링
Request Body:
{
"ips": List[str] (optional),
"method": "redfish" (기본값),
"recency_hours": int (기본: 24),
"grace_minutes": int (기본: 60),
"include_tracked_done": bool (기본: True)
}
Response:
{
"ok": True,
"count": int,
"items": [
{
"ip": str,
"ok": bool,
"error": str (if not ok),
"jobs": List[Dict]
}
]
}
"""
data = request.get_json(silent=True) or {}
# IP 목록
ip_input = data.get("ips")
if ip_input:
ips = parse_ip_list("\n".join(ip_input) if isinstance(ip_input, list) else str(ip_input))
else:
ips = load_ip_list()
if not ips:
return jsonify({
"ok": False,
"error": "No IPs provided",
"items": []
}), 400
# 파라미터
method = data.get("method", "redfish") # redfish가 기본값
recency_hours = int(data.get("recency_hours", JOB_RECENCY_HOURS))
grace_minutes = int(data.get("grace_minutes", JOB_GRACE_MINUTES))
include_tracked_done = bool(data.get("include_tracked_done", True))
grace_sec = grace_minutes * 60
cutoff = time.time() - recency_hours * 3600
# 현재 IP 목록과 다른 캐시 항목 제거
JOB_CACHE.clear_for_ips(set(ips))
# 스캔 실행
try:
items = scan_all(ips, method=method)
except Exception as e:
logger.exception("Scan failed")
return jsonify({
"ok": False,
"error": str(e),
"items": []
}), 500
now = time.time()
# 캐시 업데이트
for item in items:
ip = item.get("ip", "")
if not item.get("ok") or not isinstance(item.get("jobs"), list):
continue
for job in item["jobs"]:
status = job.get("Status")
message = job.get("Message")
active_now = is_active_status(status, message)
done_now = is_done_status(status)
# 시작 시간 파싱
start_ts = parse_iso_datetime(job.get("StartTime"))
# 리센시 판정
if not active_now:
if start_ts is None or start_ts < cutoff:
continue
# 캐시 키 생성
key = _make_cache_key(ip, job)
entry = JOB_CACHE.get(key)
if entry is None:
JOB_CACHE.set(key, {
"record": dict(job),
"first_seen_active": (now if active_now else None),
"became_done_at": (now if done_now else None),
"first_seen": now,
"last_seen": now,
"start_ts": start_ts,
})
else:
entry["record"] = dict(job)
entry["last_seen"] = now
if active_now and not entry.get("first_seen_active"):
entry["first_seen_active"] = now
if done_now and not entry.get("became_done_at"):
entry["became_done_at"] = now
elif not done_now:
entry["became_done_at"] = None
if start_ts:
entry["start_ts"] = start_ts
JOB_CACHE.set(key, entry)
# 응답 생성
out_items = []
for item in items:
ip = item.get("ip", "")
shown_jobs = []
# 현재 Active Job
current_active = []
if item.get("ok") and isinstance(item.get("jobs"), list):
for job in item["jobs"]:
if is_active_status(job.get("Status"), job.get("Message")):
key = _make_cache_key(ip, job)
if key in JOB_CACHE.keys():
current_active.append(JOB_CACHE.get(key)["record"])
if current_active:
shown_jobs = current_active
else:
# Active가 없을 때: 추적된 최근 완료 Job 표시
if include_tracked_done:
for key in JOB_CACHE.keys():
if key[0] != ip:
continue
entry = JOB_CACHE.get(key)
if not entry:
continue
start_ok = (entry.get("start_ts") or 0) >= cutoff
done_at = entry.get("became_done_at")
done_ok = bool(done_at and now - done_at <= grace_sec)
still_active = entry.get("became_done_at") is None
if still_active and start_ok:
shown_jobs.append(entry["record"])
elif done_ok and start_ok:
rec = dict(entry["record"])
rec["RecentlyCompleted"] = True
rec["CompletedAt"] = iso_now()
shown_jobs.append(rec)
out_items.append({
"ip": ip,
"ok": item.get("ok"),
"error": item.get("error"),
"jobs": sorted(shown_jobs, key=lambda r: r.get("JID", ""))
})
# 캐시 GC (조건부)
if now - JOB_CACHE.last_gc >= CACHE_GC_INTERVAL:
JOB_CACHE.gc(max_age_seconds=24 * 3600)
return jsonify({
"ok": True,
"count": len(out_items),
"items": out_items
})
def _make_cache_key(ip: str, job: dict):
"""캐시 키 생성"""
jid = (job.get("JID") or "").strip()
if jid:
return (ip, jid)
name = (job.get("Name") or "").strip()
return (ip, f"NOJID::{name}")
# ────────────────────────────────────────────────────────────
# 기존 패턴에 맞는 register 함수 추가
# ────────────────────────────────────────────────────────────
def register_jobs_routes(app):
"""
iDRAC Job 모니터링 라우트 등록
기존 프로젝트 패턴에 맞춘 함수
"""
from flask import Flask
app.register_blueprint(jobs_bp)
logger.info("Jobs routes registered at /jobs")
+675
View File
@@ -0,0 +1,675 @@
from __future__ import annotations
import os
import time
import shutil
import zipfile
import logging
from pathlib import Path
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, session, send_from_directory, send_file
from flask_login import login_required, current_user
from concurrent.futures import ThreadPoolExecutor
from watchdog.observers import Observer
from natsort import natsorted
from backend.services.ip_processor import (
save_ip_addresses,
process_ips_concurrently,
get_progress,
on_complete,
)
from backend.services.watchdog_handler import FileCreatedHandler
from config import Config
main_bp = Blueprint("main", __name__)
executor = ThreadPoolExecutor(max_workers=Config.MAX_WORKERS)
def register_main_routes(app, socketio):
app.register_blueprint(main_bp)
@app.context_processor
def inject_user():
return dict(current_user=current_user)
@app.before_request
def make_session_permanent():
session.permanent = True
if current_user.is_authenticated:
session.modified = True
def _is_safe_backup_folder_name(folder_name: str) -> bool:
if not folder_name or folder_name == "archive":
return False
if ".." in folder_name:
return False
return "/" not in folder_name and "\\" not in folder_name
def _get_repository_folder(repo_type: str) -> tuple[str, Path] | tuple[None, None]:
repositories = {
"mac": ("MAC", Path(Config.MAC_FOLDER)),
"guid": ("GUID", Path(Config.GUID_FOLDER)),
"gpu": ("GPU", Path(Config.GPU_FOLDER)),
}
return repositories.get(repo_type, (None, None))
def _safe_upload_filename(filename: str) -> str:
return Path(filename or "").name.strip()
def _unique_upload_path(target_dir: Path, filename: str) -> Path:
candidate = target_dir / filename
if not candidate.exists():
return candidate
stem = candidate.stem
suffix = candidate.suffix
index = 1
while True:
renamed = target_dir / f"{stem}_{index}{suffix}"
if not renamed.exists():
return renamed
index += 1
def _safe_repo_file_path(source_dir: Path, filename: str) -> Path | None:
safe_name = Path(filename or "").name
if not safe_name:
return None
target = (source_dir / safe_name).resolve()
try:
target.relative_to(source_dir.resolve())
except Exception:
return None
return target if target.is_file() else None
@main_bp.route("/")
@main_bp.route("/index", methods=["GET"])
@login_required
def index():
script_dir = Path(Config.SCRIPT_FOLDER)
xml_dir = Path(Config.XML_FOLDER)
info_dir = Path(Config.IDRAC_INFO_FOLDER)
backup_dir = Path(Config.BACKUP_FOLDER)
# 1. 스크립트 목록 조회 및 카테고리 분류
all_scripts = [f.name for f in script_dir.glob("*") if f.is_file() and f.name != ".env"]
all_scripts = natsorted(all_scripts)
grouped_scripts = {}
for script in all_scripts:
upper = script.upper()
category = "General"
if upper.startswith("GPU"):
category = "GPU"
elif upper.startswith("LOM"):
category = "LOM"
elif upper.startswith("TYPE") or upper.startswith("XE"):
category = "Server Models"
elif "MAC" in upper:
category = "MAC Info"
elif "GUID" in upper:
category = "GUID Info"
elif "SET_" in upper or "CONFIG" in upper:
category = "Configuration"
if category not in grouped_scripts:
grouped_scripts[category] = []
grouped_scripts[category].append(script)
# 카테고리 정렬 (General은 마지막에)
sorted_categories = sorted(grouped_scripts.keys())
if "General" in sorted_categories:
sorted_categories.remove("General")
sorted_categories.append("General")
grouped_scripts_sorted = {k: grouped_scripts[k] for k in sorted_categories}
# 2. XML 파일 목록
xml_files = [f.name for f in xml_dir.glob("*.xml")]
# 3. 페이지네이션 및 파일 목록
page = int(request.args.get("page", 1))
info_files = [f.name for f in info_dir.glob("*") if f.is_file()]
info_files = natsorted(info_files)
start = (page - 1) * Config.FILES_PER_PAGE
end = start + Config.FILES_PER_PAGE
files_to_display = [{"name": Path(f).stem, "file": f} for f in info_files[start:end]]
total_pages = (len(info_files) + Config.FILES_PER_PAGE - 1) // Config.FILES_PER_PAGE
start_page = ((page - 1) // 10) * 10 + 1
end_page = min(start_page + 9, total_pages)
# 4. 백업 폴더 목록
backup_dirs = [d for d in backup_dir.iterdir() if d.is_dir() and d.name != "archive"]
backup_dirs.sort(key=lambda p: p.stat().st_mtime, reverse=True)
backup_folder_names = [d.name for d in backup_dirs]
backup_page = int(request.args.get("backup_page", 1))
start_b = (backup_page - 1) * Config.BACKUP_FILES_PER_PAGE
end_b = start_b + Config.BACKUP_FILES_PER_PAGE
backup_slice = backup_dirs[start_b:end_b]
total_backup_pages = (len(backup_dirs) + Config.BACKUP_FILES_PER_PAGE - 1) // Config.BACKUP_FILES_PER_PAGE
backup_files = {}
for d in backup_slice:
files = [f.name for f in d.iterdir() if f.is_file()]
backup_files[d.name] = {"files": files, "count": len(files)}
return render_template(
"index.html",
files_to_display=files_to_display,
page=page,
total_pages=total_pages,
start_page=start_page,
end_page=end_page,
backup_files={}, # [DISABLED] Use server_info_files instead
total_backup_pages=total_backup_pages,
backup_page=backup_page,
server_info_files=backup_files, # Server Info (Paginated)
backup_folder_names=backup_folder_names,
scripts=all_scripts, # 기존 리스트 호환
grouped_scripts=grouped_scripts_sorted, # 카테고리별 분류
xml_files=xml_files,
# [NEW] Repositories
mac_files=_get_file_info_list(Config.MAC_FOLDER),
guid_files=_get_file_info_list(Config.GUID_FOLDER),
gpu_files=_get_file_info_list(Config.GPU_FOLDER),
)
def _get_backup_info(folder_path):
"""
폴더 내의 하위 폴더들을 순회하며 파일 목록 정보를 반환 (백업/아카이브 용)
"""
path = Path(folder_path)
if not path.exists():
return {}
dirs = [d for d in path.iterdir() if d.is_dir() and d.name != "archive"]
dirs.sort(key=lambda p: p.stat().st_mtime, reverse=True)
result = {}
for d in dirs:
files = [f.name for f in d.iterdir() if f.is_file()]
result[d.name] = {"files": files, "count": len(files)}
return result
def _get_file_info_list(folder_path_str):
"""
폴더 내 파일들의 상세 정보(이름, 크기, 수정일)를 리스트로 반환
"""
path = Path(folder_path_str)
if not path.exists():
return []
files = []
try:
for f in path.glob("*"):
if f.is_file():
stat = f.stat()
files.append({
"name": f.name,
"size": stat.st_size,
"mtime": stat.st_mtime, # timestamp for sorting
"date": time.strftime('%Y-%m-%d %H:%M', time.localtime(stat.st_mtime))
})
except Exception as e:
logging.error(f"Error reading folder {folder_path_str}: {e}")
return []
# 기본 정렬: 최신순
return sorted(files, key=lambda x: x['mtime'], reverse=True)
@main_bp.route("/process_ips", methods=["POST"])
@login_required
def process_ips():
ips = request.form.get("ips")
job_type = request.form.get("job_type")
# 변수 초기화
selected_script = ""
profile_name = None
xml_file_path = None
slot_priority = request.form.get("slot_priority") # [NEW] 슬롯 우선순위
# 작업 유형에 따른 스크립트 및 프로파일 설정
if job_type in ["mac", "server_info", "guid", "gpu"]:
profile_name = request.form.get("profile")
if job_type in ["mac", "guid"] and not profile_name:
return jsonify({"error": "프로파일을 선택하세요."}), 400
if job_type == "mac":
selected_script = "unified/collect_mac.py"
elif job_type == "server_info":
selected_script = "unified/collect_server_info.py"
elif job_type == "guid":
selected_script = "unified/collect_guid.py"
elif job_type == "gpu":
selected_script = "unified/collect_gpu.py"
profile_name = None # GPU는 프로파일 불필요
else:
# Legacy 모드 (기존 스크립트 선택)
selected_script = request.form.get("script")
selected_xml_file = request.form.get("xmlFile")
if not selected_script:
return jsonify({"error": "스크립트를 선택하세요."}), 400
if selected_script == "02-set_config.py" and selected_xml_file:
xml_path = Path(Config.XML_FOLDER) / selected_xml_file
if not xml_path.exists():
return jsonify({"error": "선택한 XML 파일이 존재하지 않습니다."}), 400
xml_file_path = str(xml_path)
if not ips:
return jsonify({"error": "IP 주소를 입력하세요."}), 400
job_id = str(time.time())
session["job_id"] = job_id
ip_files = save_ip_addresses(ips, Config.UPLOAD_FOLDER)
total_files = len(ip_files)
handler = FileCreatedHandler(job_id, total_files)
observer = Observer()
observer.schedule(handler, Config.IDRAC_INFO_FOLDER, recursive=False)
observer.start()
future = executor.submit(
process_ips_concurrently, ip_files, job_id, observer, selected_script, xml_file_path, profile_name, slot_priority
)
future.add_done_callback(lambda x: on_complete(job_id))
logging.info(f"[AJAX] 작업 시작: {job_id}, type: {job_type}, script: {selected_script}, profile: {profile_name}")
return jsonify({"job_id": job_id})
@main_bp.route("/progress_status/<job_id>")
@login_required
def progress_status(job_id: str):
return jsonify({"progress": get_progress(job_id)})
@main_bp.route("/backup", methods=["POST"])
@login_required
def backup_files():
selected_folder = request.form.get("backup_target_folder", "").strip()
new_folder = request.form.get("backup_prefix", "").strip()
is_new_folder = not selected_folder or selected_folder == "__new__"
folder_name = new_folder if is_new_folder else selected_folder
if not _is_safe_backup_folder_name(folder_name):
flash("사용할 수 없는 백업 폴더명입니다.")
return redirect(url_for("main.index"))
if is_new_folder and not folder_name.startswith("PO"):
flash("새 백업 폴더명은 PO로 시작해야 합니다.")
return redirect(url_for("main.index"))
backup_root = Path(Config.BACKUP_FOLDER).resolve()
backup_path = (backup_root / folder_name).resolve()
if backup_root not in backup_path.parents:
flash("사용할 수 없는 백업 경로입니다.")
return redirect(url_for("main.index"))
backup_path.mkdir(parents=True, exist_ok=True)
info_dir = Path(Config.IDRAC_INFO_FOLDER)
moved_count = 0
skipped_count = 0
for file in info_dir.iterdir():
if file.is_file():
target_file = backup_path / file.name
if target_file.exists():
skipped_count += 1
continue
shutil.move(str(file), str(target_file))
moved_count += 1
if moved_count:
message = f"'{folder_name}' 폴더에 {moved_count}개 파일을 백업했습니다."
if skipped_count:
message += f" 중복 파일 {skipped_count}개는 건너뛰었습니다."
flash(message)
else:
flash("백업할 처리된 파일이 없습니다.")
logging.info(f"백업 완료: {folder_name}, moved={moved_count}, skipped={skipped_count}")
return redirect(url_for("main.index"))
@main_bp.route("/download/<filename>")
@login_required
def download_file(filename: str):
# send_from_directory는 내부적으로 안전 검사를 수행
return send_from_directory(Config.IDRAC_INFO_FOLDER, filename, as_attachment=True)
@main_bp.route("/delete/<filename>", methods=["POST"])
@login_required
def delete_file(filename: str):
file_path = Path(Config.IDRAC_INFO_FOLDER) / filename
if file_path.exists():
try:
file_path.unlink()
flash(f"'{filename}' 파일이 삭제되었습니다.", "success")
logging.info(f"파일 삭제됨: {filename}")
except Exception as e:
logging.error(f"파일 삭제 오류: {e}")
flash("파일 삭제 중 오류가 발생했습니다.", "danger")
else:
flash("파일이 존재하지 않습니다.", "warning")
return redirect(url_for("main.index"))
@main_bp.route("/download_zip", methods=["POST"])
@login_required
def download_zip():
zip_filename = request.form.get("zip_filename", "export").strip()
zip_path = Path(Config.TEMP_ZIP_FOLDER) / f"{zip_filename}.zip"
# 기본 대상은 스테이징 폴더 (IDRAC_INFO_FOLDER)
target_dir = Path(Config.IDRAC_INFO_FOLDER)
# 입력된 이름이 백업 폴더에 존재하는지 확인
if zip_filename:
possible_backup_path = Path(Config.BACKUP_FOLDER) / zip_filename
if possible_backup_path.exists() and possible_backup_path.is_dir():
target_dir = possible_backup_path
logging.info(f"백업 폴더 ZIP 다운로드 요청: {zip_filename}")
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zipf:
for file in target_dir.glob("*"):
if file.is_file():
zipf.write(file, arcname=file.name)
try:
response = send_file(str(zip_path), as_attachment=True)
return response
finally:
# 응답 후 임시 ZIP 삭제
try:
if zip_path.exists():
zip_path.unlink()
except Exception as e:
logging.warning(f"임시 ZIP 삭제 실패: {e}")
@main_bp.route("/download_backup/<date>/<filename>")
@login_required
def download_backup_file(date: str, filename: str):
backup_path = Path(Config.BACKUP_FOLDER) / date
return send_from_directory(str(backup_path), filename, as_attachment=True)
@main_bp.route("/move_backup_files", methods=["POST"])
@login_required
def move_backup_files():
data = request.get_json()
filename = data.get("filename")
source_folder = data.get("source_folder")
target_folder = data.get("target_folder")
if not all([filename, source_folder, target_folder]):
return jsonify({"success": False, "message": "필수 파라미터가 누락되었습니다."}), 400
base_backup = Path(Config.BACKUP_FOLDER)
src_path = base_backup / source_folder / filename
dst_path = base_backup / target_folder / filename
# 경로 보안 검사 및 파일 존재 확인
try:
if not src_path.exists():
return jsonify({"success": False, "message": "원본 파일이 존재하지 않습니다."}), 404
# 상위 경로 탈출 방지 확인 (간단 검증)
if ".." in source_folder or ".." in target_folder:
return jsonify({"success": False, "message": "잘못된 경로입니다."}), 400
if not (base_backup / target_folder).exists():
return jsonify({"success": False, "message": "대상 폴더가 존재하지 않습니다."}), 404
shutil.move(str(src_path), str(dst_path))
logging.info(f"파일 이동 성공: {filename} from {source_folder} to {target_folder}")
return jsonify({"success": True, "message": "파일이 이동되었습니다."})
except Exception as e:
logging.error(f"파일 이동 실패: {e}")
return jsonify({"success": False, "message": f"이동 중 오류 발생: {str(e)}"}), 500
@main_bp.route("/delete_backup_folder/<folder_name>", methods=["POST"])
@login_required
def delete_backup_folder(folder_name: str):
folder_path = Path(Config.BACKUP_FOLDER) / folder_name
# 기본 폴더 보호
if folder_name == "archive":
flash("이 폴더는 삭제할 수 없습니다.", "warning")
return redirect(url_for("main.index"))
if folder_path.exists() and folder_path.is_dir():
try:
shutil.rmtree(folder_path)
flash(f"백업 폴더 '{folder_name}'가 삭제되었습니다.", "success")
logging.info(f"백업 폴더 삭제됨: {folder_name}")
except Exception as e:
logging.error(f"백업 폴더 삭제 오류: {e}")
flash("폴더 삭제 중 오류가 발생했습니다.", "danger")
else:
flash("폴더가 존재하지 않습니다.", "warning")
return redirect(url_for("main.index"))
@main_bp.route("/archive_backup_folder/<folder_name>", methods=["POST"])
@login_required
def archive_backup_folder(folder_name: str):
src_path = Path(Config.BACKUP_FOLDER) / folder_name
archive_root = Path(Config.ARCHIVE_FOLDER)
archive_root.mkdir(parents=True, exist_ok=True)
dst_path = archive_root / folder_name
if src_path.exists() and src_path.is_dir():
try:
# 이름 중복 시 처리 (timestamp 추가)
if dst_path.exists():
timestamp = int(time.time())
dst_path = archive_root / f"{folder_name}_{timestamp}"
shutil.move(str(src_path), str(dst_path))
flash(f"'{folder_name}' 폴더가 아카이브(archive)로 이동되었습니다.", "success")
logging.info(f"백업 아카이빙 성공: {folder_name} -> {dst_path}")
except Exception as e:
logging.error(f"백업 아카이빙 오류: {e}")
flash("폴더 이동 중 오류가 발생했습니다.", "danger")
else:
flash("폴더가 존재하지 않습니다.", "warning")
return redirect(url_for("main.index"))
@main_bp.route("/archive_repository/<repo_type>", methods=["POST"])
@login_required
def archive_repository(repo_type: str):
label, source_dir = _get_repository_folder(repo_type)
if not label or not source_dir:
flash("알 수 없는 저장소입니다.", "warning")
return redirect(url_for("main.index"))
files = [f for f in source_dir.iterdir() if f.is_file()] if source_dir.exists() else []
if not files:
flash(f"{label} 저장소에 아카이브할 파일이 없습니다.", "warning")
return redirect(url_for("main.index"))
timestamp = time.strftime("%Y%m%d_%H%M%S")
archive_dir = Path(Config.ARCHIVE_FOLDER) / "repository" / repo_type / timestamp
archive_dir.mkdir(parents=True, exist_ok=True)
moved_count = 0
try:
for file in files:
shutil.move(str(file), str(archive_dir / file.name))
moved_count += 1
flash(f"{label} 파일 {moved_count}개를 아카이브로 이동했습니다.", "success")
logging.info(f"{label} 저장소 아카이빙 성공: {moved_count} files -> {archive_dir}")
except Exception as e:
logging.error(f"{label} 저장소 아카이빙 오류: {e}")
flash("저장소 파일을 아카이브로 이동하는 중 오류가 발생했습니다.", "danger")
return redirect(url_for("main.index"))
@main_bp.route("/download_repository_files/<repo_type>", methods=["POST"])
@login_required
def download_repository_files(repo_type: str):
label, source_dir = _get_repository_folder(repo_type)
if not label or not source_dir:
flash("알 수 없는 저장소입니다.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
selected_paths = []
for filename in request.form.getlist("files"):
file_path = _safe_repo_file_path(source_dir, filename)
if file_path:
selected_paths.append(file_path)
if not selected_paths:
flash("다운로드할 파일을 선택하세요.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
if len(selected_paths) == 1:
return send_from_directory(str(source_dir), selected_paths[0].name, as_attachment=True)
timestamp = time.strftime("%Y%m%d_%H%M%S")
zip_path = Path(Config.TEMP_ZIP_FOLDER) / f"{repo_type}_selected_{timestamp}.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zipf:
for file_path in selected_paths:
zipf.write(file_path, arcname=file_path.name)
try:
return send_file(str(zip_path), as_attachment=True, download_name=zip_path.name)
finally:
try:
if zip_path.exists():
zip_path.unlink()
except Exception as e:
logging.warning("선택 파일 ZIP 삭제 실패: %s", e)
@main_bp.route("/delete_repository_file/<repo_type>", methods=["POST"])
@login_required
def delete_repository_file(repo_type: str):
label, source_dir = _get_repository_folder(repo_type)
if not label or not source_dir:
flash("알 수 없는 저장소입니다.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
file_path = _safe_repo_file_path(source_dir, request.form.get("filename", ""))
if not file_path:
flash("삭제할 파일을 찾을 수 없습니다.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
try:
file_path.unlink()
flash(f"{label} 파일 '{file_path.name}'을 삭제했습니다.", "success")
except Exception as e:
logging.error("%s 저장소 파일 삭제 오류: %s", label, e)
flash("파일 삭제 중 오류가 발생했습니다.", "danger")
return redirect(url_for("main.index", workspace=repo_type))
@main_bp.route("/delete_repository_files/<repo_type>", methods=["POST"])
@login_required
def delete_repository_files(repo_type: str):
label, source_dir = _get_repository_folder(repo_type)
if not label or not source_dir:
flash("알 수 없는 저장소입니다.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
deleted_count = 0
skipped_count = 0
for filename in request.form.getlist("files"):
file_path = _safe_repo_file_path(source_dir, filename)
if not file_path:
skipped_count += 1
continue
try:
file_path.unlink()
deleted_count += 1
except Exception as e:
skipped_count += 1
logging.error("%s 저장소 선택 파일 삭제 오류: %s -> %s", label, filename, e)
if deleted_count:
message = f"{label} 파일 {deleted_count}개를 삭제했습니다."
if skipped_count:
message += f" {skipped_count}개는 건너뛰었습니다."
flash(message, "success")
else:
flash("삭제할 파일을 선택하세요.", "warning")
return redirect(url_for("main.index", workspace=repo_type))
@main_bp.route("/upload_workspace_files/<target_type>", methods=["POST"])
@login_required
def upload_workspace_files(target_type: str):
if target_type == "server_info":
folder_name = request.form.get("folder_name", "").strip()
if not _is_safe_backup_folder_name(folder_name):
flash("업로드할 서버 정보 폴더를 확인할 수 없습니다.", "warning")
return redirect(url_for("main.index"))
base_dir = Path(Config.BACKUP_FOLDER).resolve()
target_dir = (base_dir / folder_name).resolve()
if base_dir not in target_dir.parents or not target_dir.is_dir():
flash("업로드할 서버 정보 폴더가 존재하지 않습니다.", "warning")
return redirect(url_for("main.index"))
label = f"서버 정보 '{folder_name}'"
else:
label, target_dir = _get_repository_folder(target_type)
if not label or not target_dir:
flash("알 수 없는 업로드 대상입니다.", "warning")
return redirect(url_for("main.index"))
target_dir.mkdir(parents=True, exist_ok=True)
duplicate_policy = request.form.get("duplicate_policy", "rename")
should_overwrite = duplicate_policy == "overwrite"
uploaded_files = request.files.getlist("files")
saved_count = 0
skipped_count = 0
for uploaded in uploaded_files:
filename = _safe_upload_filename(uploaded.filename)
if not filename:
skipped_count += 1
continue
destination = target_dir / filename if should_overwrite else _unique_upload_path(target_dir, filename)
try:
uploaded.save(destination)
saved_count += 1
except Exception as e:
skipped_count += 1
logging.error("워크스페이스 파일 업로드 실패: %s -> %s", filename, e)
if saved_count:
message = f"{label}에 파일 {saved_count}개를 업로드했습니다."
if skipped_count:
message += f" {skipped_count}개는 건너뛰었습니다."
flash(message, "success")
else:
flash("업로드할 파일을 선택하지 않았거나 저장할 수 없습니다.", "warning")
logging.info("워크스페이스 업로드 완료: target=%s saved=%s skipped=%s", target_type, saved_count, skipped_count)
return redirect(url_for("main.index", workspace=target_type))
+166
View File
@@ -0,0 +1,166 @@
from flask import Blueprint, render_template, request, jsonify, flash, redirect, url_for
from flask_login import login_required, current_user
import logging
import difflib
from pathlib import Path
from config import Config
from backend.services.redfish_client import RedfishClient
from backend.routes.xml import sanitize_preserve_unicode
scp_bp = Blueprint("scp", __name__)
logger = logging.getLogger(__name__)
@scp_bp.route("/scp/diff", methods=["GET"])
@login_required
def diff_scp():
"""
두 XML 파일의 차이점을 비교하여 보여줍니다.
"""
file1_name = request.args.get("file1")
file2_name = request.args.get("file2")
if not file1_name or not file2_name:
flash("비교할 두 파일을 선택해주세요.", "warning")
return redirect(url_for("xml.xml_management"))
try:
file1_path = Path(Config.XML_FOLDER) / sanitize_preserve_unicode(file1_name)
file2_path = Path(Config.XML_FOLDER) / sanitize_preserve_unicode(file2_name)
if not file1_path.exists() or not file2_path.exists():
flash("파일을 찾을 수 없습니다.", "danger")
return redirect(url_for("xml.xml_management"))
# 파일 내용 읽기 (LF로 통일)
# 파일 내용 읽기 (LF로 통일)
# Monaco Editor에 원본 텍스트를 그대로 전달하기 위해 splitlines() 제거
# 파일 내용 읽기 (LF로 통일)
logger.info(f"Reading file1: {file1_path}")
content1 = file1_path.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n")
logger.info(f"Reading file2: {file2_path}")
content2 = file2_path.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n")
logger.info(f"Content1 length: {len(content1)}, Content2 length: {len(content2)}")
return render_template("scp_diff.html",
file1=file1_name,
file2=file2_name,
content1=content1,
content2=content2)
except Exception as e:
logger.error(f"Diff error: {e}")
flash(f"비교 중 오류가 발생했습니다: {str(e)}", "danger")
return redirect(url_for("xml.xml_management"))
@scp_bp.route("/scp/content/<path:filename>")
@login_required
def get_scp_content(filename):
"""
XML 파일 내용을 반환하는 API (Monaco Editor용)
"""
try:
safe_name = sanitize_preserve_unicode(filename)
path = Path(Config.XML_FOLDER) / safe_name
if not path.exists():
return "File not found", 404
# 텍스트로 읽어서 반환
content = path.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n")
return content, 200, {'Content-Type': 'text/plain; charset=utf-8'}
except Exception as e:
logger.error(f"Content read error: {e}")
return str(e), 500
@scp_bp.route("/scp/export", methods=["POST"])
@login_required
def export_scp():
"""
iDRAC에서 설정을 내보내기 (Export)
네트워크 공유 설정이 필요합니다.
"""
data = request.form
target_ip = data.get("target_ip")
username = data.get("username")
password = data.get("password")
# Share Parameters
share_ip = data.get("share_ip")
share_name = data.get("share_name")
share_user = data.get("share_user")
share_pwd = data.get("share_pwd")
filename = data.get("filename")
if not all([target_ip, username, password, share_ip, share_name, filename]):
flash("필수 정보가 누락되었습니다.", "warning")
return redirect(url_for("xml.xml_management"))
share_params = {
"IPAddress": share_ip,
"ShareName": share_name,
"FileName": filename,
"ShareType": "CIFS", # 기본값 CIFS
"UserName": share_user,
"Password": share_pwd
}
try:
with RedfishClient(target_ip, username, password) as client:
job_id = client.export_system_configuration(share_params)
if job_id:
flash(f"내보내기 작업이 시작되었습니다. Job ID: {job_id}", "success")
else:
flash("작업을 시작했으나 Job ID를 받지 못했습니다.", "warning")
except Exception as e:
logger.error(f"Export failed: {e}")
flash(f"내보내기 실패: {str(e)}", "danger")
return redirect(url_for("xml.xml_management"))
@scp_bp.route("/scp/import", methods=["POST"])
@login_required
def import_scp():
"""
iDRAC로 설정 가져오기 (Import/Deploy)
"""
data = request.form
target_ip = data.get("target_ip")
username = data.get("username")
password = data.get("password")
# Share Parameters
share_ip = data.get("share_ip")
share_name = data.get("share_name")
share_user = data.get("share_user")
share_pwd = data.get("share_pwd")
filename = data.get("filename")
import_mode = data.get("import_mode", "Replace")
if not all([target_ip, username, password, share_ip, share_name, filename]):
flash("필수 정보가 누락되었습니다.", "warning")
return redirect(url_for("xml.xml_management"))
share_params = {
"IPAddress": share_ip,
"ShareName": share_name,
"FileName": filename,
"ShareType": "CIFS",
"UserName": share_user,
"Password": share_pwd
}
try:
with RedfishClient(target_ip, username, password) as client:
job_id = client.import_system_configuration(share_params, import_mode=import_mode)
if job_id:
flash(f"설정 적용(Import) 작업이 시작되었습니다. Job ID: {job_id}", "success")
else:
flash("작업을 시작했으나 Job ID를 받지 못했습니다.", "warning")
except Exception as e:
logger.error(f"Import failed: {e}")
flash(f"설정 적용 실패: {str(e)}", "danger")
return redirect(url_for("xml.xml_management"))
+325
View File
@@ -0,0 +1,325 @@
"""
스크립트 관리 라우트 및 API
"""
from flask import Blueprint, render_template, request, jsonify, send_file
from flask_login import login_required
from pathlib import Path
import yaml
import json
import logging
script_manager_bp = Blueprint("script_manager", __name__)
# 프로파일 디렉토리
PROFILES_DIR = Path(__file__).parent.parent.parent / "data" / "scripts" / "profiles"
UNIFIED_DIR = Path(__file__).parent.parent.parent / "data" / "scripts" / "unified"
@script_manager_bp.route("/script_manager")
@login_required
def index():
"""스크립트 관리 메인 페이지"""
return render_template("script_manager.html")
@script_manager_bp.route("/api/profiles", methods=["GET"])
@login_required
def get_profiles():
"""모든 프로파일 목록 (카드 뷰용)"""
profiles = []
for yaml_file in PROFILES_DIR.glob("*.yaml"):
try:
with open(yaml_file, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
category = yaml_file.stem.replace('_profiles', '') # mac, guid, server_info
for name, config in data.get('profiles', {}).items():
profiles.append({
"name": name,
"description": config.get("description", ""),
"category": category,
"items": _count_items(config, category),
"is_default": name == "default"
})
except Exception as e:
logging.error(f"프로파일 로드 오류 ({yaml_file}): {e}")
return jsonify(profiles)
@script_manager_bp.route("/api/profiles/<category>", methods=["GET"])
@login_required
def get_profiles_by_category(category):
"""특정 카테고리의 프로파일 목록"""
profiles = []
# 파일명 매핑 (server_info -> server_info_profiles.yaml)
# category가 mac, guid, server_info 등으로 넘어옴
yaml_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not yaml_file.exists():
return jsonify({"error": f"프로파일 파일이 없습니다: {category}"}), 404
try:
with open(yaml_file, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
for name, config in data.get('profiles', {}).items():
profiles.append({
"name": name,
"description": config.get("description", ""),
"category": category,
"items": _count_items(config, category),
"is_default": name == "default"
})
except Exception as e:
logging.error(f"프로파일 로드 오류 ({yaml_file}): {e}")
return jsonify({"error": str(e)}), 500
return jsonify(profiles)
@script_manager_bp.route("/api/profiles/<category>/<path:name>", methods=["GET"])
@login_required
def get_profile(category, name):
"""특정 프로파일 상세 정보 (편집용)"""
profile_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not profile_file.exists():
return jsonify({"error": "프로파일 파일 없음"}), 404
try:
with open(profile_file, 'r', encoding='utf-8') as f:
profiles = yaml.safe_load(f)['profiles']
profile = profiles.get(name)
if not profile:
return jsonify({"error": "프로파일 없음"}), 404
# UI 친화적 형식으로 변환
return jsonify({
"name": name,
"description": profile.get("description", ""),
"category": category,
"config": profile
})
except Exception as e:
logging.error(f"프로파일 로드 오류: {e}")
return jsonify({"error": str(e)}), 500
@script_manager_bp.route("/api/profiles/<category>/<path:name>", methods=["PUT"])
@login_required
def update_profile(category, name):
"""프로파일 업데이트"""
profile_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not profile_file.exists():
return jsonify({"error": "프로파일 파일 없음"}), 404
try:
data = request.json
# 기존 프로파일 로드
with open(profile_file, 'r', encoding='utf-8') as f:
profiles_data = yaml.safe_load(f)
# 업데이트
profiles_data['profiles'][name] = data.get('config', {})
# 저장
with open(profile_file, 'w', encoding='utf-8') as f:
yaml.dump(profiles_data, f, allow_unicode=True, default_flow_style=False)
return jsonify({"success": True, "message": "프로파일이 저장되었습니다."})
except Exception as e:
logging.error(f"프로파일 저장 오류: {e}")
return jsonify({"error": str(e)}), 500
@script_manager_bp.route("/api/profiles/<category>", methods=["POST"])
@login_required
def create_profile(category):
"""새 프로파일 생성"""
profile_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not profile_file.exists():
return jsonify({"error": "프로파일 파일 없음"}), 404
try:
data = request.json
name = data.get('name')
if not name:
return jsonify({"error": "프로파일 이름 필요"}), 400
# 기존 프로파일 로드
with open(profile_file, 'r', encoding='utf-8') as f:
profiles_data = yaml.safe_load(f)
# 중복 체크
if name in profiles_data['profiles']:
return jsonify({"error": "이미 존재하는 프로파일"}), 400
# 새 프로파일 추가
profiles_data['profiles'][name] = data.get('config', {})
# 저장
with open(profile_file, 'w', encoding='utf-8') as f:
yaml.dump(profiles_data, f, allow_unicode=True, default_flow_style=False)
return jsonify({"success": True, "message": "프로파일이 생성되었습니다."})
except Exception as e:
logging.error(f"프로파일 생성 오류: {e}")
return jsonify({"error": str(e)}), 500
@script_manager_bp.route("/api/profiles/<category>/<path:name>", methods=["DELETE"])
@login_required
def delete_profile(category, name):
"""프로파일 삭제"""
profile_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not profile_file.exists():
return jsonify({"error": "프로파일 파일 없음"}), 404
try:
# default 프로파일은 삭제 불가
if name == "default":
return jsonify({"error": "기본 프로파일은 삭제할 수 없습니다."}), 400
# 기존 프로파일 로드
with open(profile_file, 'r', encoding='utf-8') as f:
profiles_data = yaml.safe_load(f)
# 삭제
if name in profiles_data['profiles']:
del profiles_data['profiles'][name]
else:
return jsonify({"error": "프로파일 없음"}), 404
# 저장
with open(profile_file, 'w', encoding='utf-8') as f:
yaml.dump(profiles_data, f, allow_unicode=True, default_flow_style=False)
return jsonify({"success": True, "message": "프로파일이 삭제되었습니다."})
except Exception as e:
logging.error(f"프로파일 삭제 오류: {e}")
return jsonify({"error": str(e)}), 500
@script_manager_bp.route("/api/profiles/<category>/export", methods=["GET"])
@login_required
def export_profile(category):
"""프로파일 YAML 파일 다운로드"""
name = request.args.get("name", "")
if not name:
return jsonify({"error": "프로파일 이름 필요"}), 400
profile_file = PROFILES_DIR / f"{category}_profiles.yaml"
if not profile_file.exists():
return jsonify({"error": "프로파일 파일 없음"}), 404
try:
with open(profile_file, 'r', encoding='utf-8') as f:
profiles = yaml.safe_load(f)
# 단일 프로파일만 추출
single_profile = {
"profiles": {
name: profiles['profiles'][name]
}
}
# 임시 파일 생성
temp_dir = Path(__file__).parent.parent.parent / "data" / "temp_zips"
temp_dir.mkdir(parents=True, exist_ok=True)
export_file = temp_dir / f"{name}_profile.yaml"
with open(export_file, 'w', encoding='utf-8') as f:
yaml.dump(single_profile, f, allow_unicode=True)
return send_file(export_file, as_attachment=True, download_name=f"{name}_profile.yaml")
except Exception as e:
logging.error(f"프로파일 내보내기 오류: {e}")
return jsonify({"error": str(e)}), 500
@script_manager_bp.route("/api/test_racadm", methods=["POST"])
@login_required
def test_racadm():
"""Racadm 명령어 테스트"""
try:
import subprocess
data = request.json
ip = data.get("ip")
user = data.get("user", "root")
password = data.get("password")
command = data.get("command")
if not all([ip, user, password, command]):
return jsonify({"success": False, "error": "필수 입력값 누락"}), 400
# 명령어 구성 (racadm -r <IP> -u <USER> -p <PW> <CMD>)
# 보안을 위해 shell=False로 리스트 형태로 전달
full_cmd = ["racadm", "-r", ip, "-u", user, "-p", password] + command.split()
logging.info(f"[Test] Executing racadm on {ip}: {command}")
# 타임아웃 30초
result = subprocess.run(
full_cmd,
capture_output=True,
text=True,
timeout=30,
check=False
)
return jsonify({
"success": True,
"return_code": result.returncode,
"output": result.stdout + (result.stderr if result.stderr else "")
})
except subprocess.TimeoutExpired:
return jsonify({
"success": False,
"error": "명령어 실행 시간 초과 (30초)"
})
except Exception as e:
logging.error(f"[Test] Racadm execution error: {e}")
return jsonify({"success": False, "error": str(e)})
def _count_items(profile: dict, category: str) -> int:
"""프로파일에서 수집할 항목 개수 계산"""
count = 0
if category == "mac":
count += len(profile.get("nic_patterns", []))
if profile.get("infiniband_slots"):
count += len(profile["infiniband_slots"])
if profile.get("include_idrac_mac"):
count += 1
elif category == "guid":
count += len(profile.get("slot_priority", []))
elif category == "gpu":
# GUID slot priority + GPU settings
count += len(profile.get("slot_priority", []))
if profile.get("gpu_settings"):
count += 1
elif category == "server_info":
count += len(profile.get("firmware", []))
count += len(profile.get("bios_settings", []))
count += len(profile.get("idrac_settings", []))
count += len(profile.get("raid_settings", []))
count += len(profile.get("custom_items", []))
return count
def register_script_manager(app):
"""블루프린트 등록"""
app.register_blueprint(script_manager_bp)
+185
View File
@@ -0,0 +1,185 @@
from flask import Blueprint, render_template, request, jsonify
from backend.models.user import db
from backend.models.idrac_server import IdracServer
from backend.models.server_config import ServerConfigSnapshot
from backend.services.idrac_redfish_client import DellRedfishClient
import concurrent.futures
import json
import hashlib
from datetime import datetime
server_config_bp = Blueprint('server_config', __name__, url_prefix='/server-config')
@server_config_bp.route('/')
def index():
"""서버 설정 분석 및 비교 페이지"""
return render_template('server_config.html')
def calculate_checksum(data):
"""JSON 데이터의 SHA256 체크섬 계산"""
json_str = json.dumps(data, sort_keys=True)
return hashlib.sha256(json_str.encode('utf-8')).hexdigest()
@server_config_bp.route('/api/sync', methods=['POST'])
def sync_configs():
"""
선택한 서버들의 설정을 iDRAC에서 가져와 DB에 저장 (동기화)
"""
try:
data = request.json
server_ids = data.get('server_ids', [])
categories = data.get('categories', ['bios', 'idrac', 'system'])
if not server_ids:
return jsonify({'success': False, 'message': '서버를 선택해주세요.'})
results = {'success': [], 'failed': []}
def process_server(server_id):
server = IdracServer.query.get(server_id)
if not server:
return {'id': server_id, 'status': 'failed', 'message': 'Server not found'}
client = DellRedfishClient(server.ip_address, server.username, server.password)
if not client.check_connection():
return {'id': server.id, 'name': server.name, 'status': 'failed', 'message': 'Connection failed'}
sync_results = []
# System Info
if 'system' in categories:
sys_data = client.get_system_info_detailed()
if sys_data:
save_snapshot(server.id, 'system', sys_data)
sync_results.append('system')
# BIOS Attributes
if 'bios' in categories:
bios_data = client.get_bios_attributes_all()
if bios_data:
save_snapshot(server.id, 'bios', bios_data)
sync_results.append('bios')
# iDRAC Attributes
if 'idrac' in categories:
idrac_data = client.get_idrac_attributes_all()
if idrac_data:
save_snapshot(server.id, 'idrac', idrac_data)
sync_results.append('idrac')
return {'id': server.id, 'name': server.name, 'status': 'success', 'synced': sync_results}
def save_snapshot(server_id, config_type, data):
# 기존 스냅샷 확인
snapshot = ServerConfigSnapshot.query.filter_by(
server_id=server_id,
config_type=config_type
).first()
new_hash = calculate_checksum(data)
if snapshot:
# 변경사항이 있는 경우에만 업데이트
if snapshot.hash_value != new_hash:
snapshot.data = data
snapshot.hash_value = new_hash
snapshot.updated_at = datetime.utcnow() # 명시적 업데이트 시간 갱신
else:
snapshot = ServerConfigSnapshot(
server_id=server_id,
config_type=config_type,
data=data,
hash_value=new_hash
)
db.session.add(snapshot)
# 트랜잭션은 개별 commit 혹은 묶어서 commit.
# 여기서는 스레드 안전성을 위해 함수 내에서는 session 조작만 하고
# 실제 commit은 메인 스레드나 별도 처리 필요하지만, Flask-SQLAlchemy는 scoped_session이므로
# 각 요청(스레드)마다 세션이 다를 수 있음.
# 하지만 ThreadPoolExecutor 사용 시 app context 문제가 발생할 수 있음.
# safe way: return data -> main thread saves. OR push app context.
# NOTE: For simplicity in this refactor with ThreadPoolExecutor,
# we will handle DB operations inside the thread but need app context.
# See below for implementation adjustment.
return True
# DB 저장을 위해 app context 필요
from flask import current_app
app = current_app._get_current_object()
def process_with_context(server_id):
with app.app_context():
try:
res = process_server(server_id)
db.session.commit() # Commit per server
return res
except Exception as e:
db.session.rollback()
return {'id': server_id, 'status': 'failed', 'message': str(e)}
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_id = {executor.submit(process_with_context, sid): sid for sid in server_ids}
for future in concurrent.futures.as_completed(future_to_id):
try:
res = future.result()
if res['status'] == 'success':
results['success'].append(res)
else:
results['failed'].append(res)
except Exception as e:
results['failed'].append({'error': str(e)})
return jsonify({
'success': True,
'results': results
})
except Exception as e:
return jsonify({'success': False, 'message': f"Sync Error: {str(e)}"})
@server_config_bp.route('/api/fetch', methods=['POST'])
def fetch_configs():
"""
DB에 저장된 서버 설정 스냅샷 조회
"""
try:
data = request.json
server_ids = data.get('server_ids', [])
categories = data.get('categories', ['bios', 'idrac', 'system'])
if not server_ids:
return jsonify({'success': False, 'message': '서버를 선택해주세요.'})
results = {}
for server_id in server_ids:
server = IdracServer.query.get(server_id)
if not server:
continue
server_data = {'id': server.id, 'name': server.name, 'ip': server.ip_address}
for cat in categories:
snapshot = ServerConfigSnapshot.query.filter_by(
server_id=server_id,
config_type=cat
).first()
if snapshot:
server_data[cat] = snapshot.data
server_data[f'{cat}_updated'] = snapshot.updated_at.isoformat()
else:
server_data[cat] = None
results[server_id] = server_data
return jsonify({
'success': True,
'results': results
})
except Exception as e:
return jsonify({'success': False, 'message': f"Fetch Error: {str(e)}"})
+721
View File
@@ -0,0 +1,721 @@
from __future__ import annotations
import os
import sys
import shutil
import subprocess
import logging
import zipfile
import time
from pathlib import Path
from flask import Blueprint, request, redirect, url_for, flash, jsonify, send_file
from flask_login import login_required
from config import Config
from backend.models.user import db
from backend.models.system_setting import SystemSetting
utils_bp = Blueprint("utils", __name__)
def get_system_setting(key: str, default: str = "") -> str:
try:
setting = db.session.get(SystemSetting, key)
return setting.value if setting else default
except Exception:
return default
def register_util_routes(app):
app.register_blueprint(utils_bp)
@utils_bp.route("/move_mac_files", methods=["POST"])
@login_required
def move_mac_files():
src = Path(Config.IDRAC_INFO_FOLDER)
dst = Path(Config.MAC_FOLDER)
dst.mkdir(parents=True, exist_ok=True)
moved = 0
skipped = 0
missing = 0
errors = []
# JSON 요청 파싱 (overwrite 플래그 확인)
data = request.get_json(silent=True) or {}
overwrite = data.get("overwrite", False)
# 1. 대상 파일 스냅샷 (이동 시도할, 또는 해야할 파일들)
try:
current_files = [f for f in src.iterdir() if f.is_file()]
except Exception as e:
logging.error(f"파일 목록 조회 실패: {e}")
return jsonify({"success": False, "error": str(e)})
# [중복 체크 로직] 덮어쓰기 모드가 아닐 때, 미리 중복 검사
if not overwrite:
duplicates = []
for file in current_files:
target = dst / file.name
if target.exists():
duplicates.append(file.name)
if duplicates:
return jsonify({
"success": False,
"requires_confirmation": True,
"duplicates": duplicates,
"duplicate_count": len(duplicates)
})
else:
logging.warning(f"⚠️ [MAC] 덮어쓰기 모드 활성화됨 - 중복 파일을 덮어씁니다.")
total_target_count = len(current_files)
# 카운터
moved_count = 0 # 내가 직접 옮김 (또는 덮어씀)
verified_count = 0 # 최종적으로 목적지에 있음을 확인 (성공)
lost_count = 0 # 소스에도 없고 목적지에도 없음 (진짜 유실?)
errors = []
for file in current_files:
target = dst / file.name
# [Step 1] 이미 목적지에 있는지 확인
if target.exists():
if overwrite:
# 덮어쓰기 모드: 기존 파일 삭제 후 이동 진행 (또는 바로 move로 덮어쓰기)
# shutil.move는 대상이 존재하면 에러가 날 수 있으므로(버전/OS따라 다름), 안전하게 삭제 시도
try:
# Windows에서는 사용 중인 파일 삭제 시 에러 발생 가능
# shutil.move(src, dst)는 dst가 존재하면 덮어쓰기 시도함 (Python 3.x)
pass
except Exception:
pass
else:
# (중복 체크를 통과했거나 Race Condition으로 생성된 경우) -> 이미 완료된 것으로 간주
verified_count += 1
logging.info(f"⏭️ 파일 이미 존재 (Skipped): {file.name}")
continue
# [Step 2] 소스에 있는지 확인 (Race Condition)
if not file.exists():
if target.exists():
verified_count += 1
continue
else:
lost_count += 1
logging.warning(f"❓ 이동 중 사라짐: {file.name}")
continue
# [Step 3] 이동 시도 (덮어쓰기 포함)
try:
shutil.move(str(file), str(target))
moved_count += 1
verified_count += 1
except shutil.Error as e:
# shutil.move might raise Error if destination exists depending on implementation,
# but standard behavior overwrites if not same file.
# If exact same file, verified.
if target.exists():
verified_count += 1
else:
errors.append(f"{file.name}: {str(e)}")
except FileNotFoundError:
# 옮기려는 찰나에 사라짐 -> 목적지 재확인
if target.exists():
verified_count += 1
logging.info(f"⏭️ 동시 처리됨 (완료): {file.name}")
else:
lost_count += 1
except Exception as e:
# 권한 에러 등 진짜 실패
error_msg = f"{file.name}: {str(e)}"
errors.append(error_msg)
logging.error(f"❌ 이동 에러: {error_msg}")
# 결과 요약
msg = f"{total_target_count}건 중 {verified_count}건 처리 완료"
if moved_count < verified_count:
msg += f" (이동: {moved_count}, 이미 완료: {verified_count - moved_count})"
if lost_count > 0:
msg += f", 확인 불가: {lost_count}"
logging.info(f"✅ MAC 처리 결과: {msg}")
flash(msg, "success" if lost_count == 0 else "warning")
return jsonify({
"success": True,
"total": total_target_count,
"verified": verified_count,
"message": msg,
"errors": errors if errors else None
})
@utils_bp.route("/move_guid_files", methods=["POST"])
@login_required
def move_guid_files():
src = Path(Config.IDRAC_INFO_FOLDER)
dst = Path(Config.GUID_FOLDER)
dst.mkdir(parents=True, exist_ok=True)
moved = 0
skipped = 0
missing = 0
errors = []
# JSON 요청 파싱 (overwrite 플래그 확인)
data = request.get_json(silent=True) or {}
overwrite = data.get("overwrite", False)
try:
files = [f for f in src.iterdir() if f.is_file()]
except Exception:
files = []
# [중복 체크]
if not overwrite:
duplicates = []
for file in files:
target = dst / file.name
if target.exists():
duplicates.append(file.name)
if duplicates:
return jsonify({
"success": False,
"requires_confirmation": True,
"duplicates": duplicates,
"duplicate_count": len(duplicates)
})
else:
logging.warning(f"⚠️ [GUID] 덮어쓰기 모드 활성화됨 - 중복 파일을 덮어씁니다.")
total_target_count = len(files)
verified_count = 0
moved_count = 0
errors = []
lost_count = 0
try:
for file in files:
target = dst / file.name
# 1. 이미 완료되었는지 확인
if target.exists():
if not overwrite:
verified_count += 1
continue
# overwrite=True면 계속 진행하여 덮어쓰기 시도
# 2. 소스 확인
if not file.exists():
if target.exists(): verified_count += 1
else: lost_count += 1
continue
# 3. 이동
try:
shutil.move(str(file), str(target))
moved_count += 1
verified_count += 1
except FileNotFoundError:
if target.exists(): verified_count += 1
else: lost_count += 1
except Exception as e:
errors.append(f"{file.name}: {e}")
# 상세 메시지
msg = f"{total_target_count}건 중 {verified_count}건 처리 완료"
logging.info(f"✅ GUID 처리: {msg}")
flash(msg, "success" if lost_count == 0 else "warning")
return jsonify({
"success": True,
"total": total_target_count,
"verified": verified_count,
"message": msg,
"errors": errors if errors else None
})
except Exception as e:
logging.error(f"❌ GUID 이동 중 오류: {e}")
return jsonify({"success": False, "error": str(e)})
@utils_bp.route("/move_gpu_files", methods=["POST"])
@login_required
def move_gpu_files():
"""
data/idrac_info → data/gpu_serial 로 GPU 시리얼 텍스트 파일 이동
"""
src = Path(Config.IDRAC_INFO_FOLDER) # 예: data/idrac_info
dst = Path(Config.GPU_FOLDER) # 예: data/gpu_serial
dst.mkdir(parents=True, exist_ok=True)
moved = 0
skipped = 0
missing = 0
errors = []
# JSON 요청 파싱
data = request.get_json(silent=True) or {}
overwrite = data.get("overwrite", False)
try:
files = [f for f in src.iterdir() if f.is_file()]
except Exception:
files = []
# [중복 체크]
if not overwrite:
duplicates = []
for file in files:
target = dst / file.name
if target.exists():
duplicates.append(file.name)
if duplicates:
return jsonify({
"success": False,
"requires_confirmation": True,
"duplicates": duplicates,
"duplicate_count": len(duplicates)
})
else:
logging.warning(f"⚠️ [GPU] 덮어쓰기 모드 활성화됨 - 중복 파일을 덮어씁니다.")
total_target_count = len(files)
verified_count = 0
moved_count = 0
errors = []
lost_count = 0
try:
for file in files:
target = dst / file.name
# 1. 존재 확인 (덮어쓰기 아닐 경우)
if target.exists():
if not overwrite:
verified_count += 1
continue
# 2. 소스 확인
if not file.exists():
if target.exists(): verified_count += 1
else: lost_count += 1
continue
# 3. 이동
try:
shutil.move(str(file), str(target))
moved_count += 1
verified_count += 1
except FileNotFoundError:
if target.exists(): verified_count += 1
else: lost_count += 1
except Exception as e:
errors.append(f"{file.name}: {e}")
# 상세 메시지
msg = f"{total_target_count}건 중 {verified_count}건 처리 완료"
logging.info(f"✅ GPU 처리: {msg}")
flash(msg, "success")
return jsonify({
"success": True,
"total": total_target_count,
"verified": verified_count,
"message": msg,
"errors": errors if errors else None
})
except Exception as e:
logging.error(f"❌ GPU 이동 오류: {e}")
return jsonify({"success": False, "error": str(e)})
@utils_bp.route("/update_server_list", methods=["POST"])
@login_required
def update_server_list():
content = request.form.get("server_list_content")
if not content:
flash("내용을 입력하세요.", "warning")
return redirect(url_for("main.index"))
path = Path(Config.SERVER_LIST_FOLDER) / "server_list.txt"
try:
path.write_text(content, encoding="utf-8")
result = subprocess.run(
[sys.executable, str(Path(Config.SERVER_LIST_FOLDER) / "unified_excel_generator.py"), "--mode", "mac"],
capture_output=True,
text=True,
check=True,
cwd=str(Path(Config.SERVER_LIST_FOLDER)),
timeout=300,
)
logging.info(f"서버 리스트 스크립트 실행 결과: {result.stdout}")
flash("서버 리스트가 업데이트되었습니다.", "success")
except subprocess.CalledProcessError as e:
logging.error(f"서버 리스트 스크립트 오류: {e.stderr}")
flash(f"스크립트 실행 실패: {e.stderr}", "danger")
except Exception as e:
logging.error(f"서버 리스트 처리 오류: {e}")
flash(f"서버 리스트 처리 중 오류 발생: {e}", "danger")
return redirect(url_for("main.index"))
@utils_bp.route("/update_guid_list", methods=["POST"])
@login_required
def update_guid_list():
content = request.form.get("server_list_content")
slot_priority = request.form.get("slot_priority", "") # 슬롯 우선순위 받기
if not content:
flash("내용을 입력하세요.", "warning")
return redirect(url_for("main.index"))
path = Path(Config.SERVER_LIST_FOLDER) / "guid_list.txt"
try:
path.write_text(content, encoding="utf-8")
# 슬롯 우선순위를 환경변수로 전달
env = os.environ.copy()
if slot_priority:
env["GUID_SLOT_PRIORITY"] = slot_priority
logging.info(f"GUID 슬롯 우선순위: {slot_priority}")
result = subprocess.run(
[sys.executable, str(Path(Config.SERVER_LIST_FOLDER) / "unified_excel_generator.py"), "--mode", "guid"],
capture_output=True,
text=True,
check=True,
cwd=str(Path(Config.SERVER_LIST_FOLDER)),
timeout=300,
env=env, # 환경변수 전달
)
logging.info(f"GUID 리스트 스크립트 실행 결과: {result.stdout}")
flash("GUID 리스트가 업데이트되었습니다.", "success")
except subprocess.CalledProcessError as e:
logging.error(f"GUID 리스트 스크립트 오류: {e.stderr}")
flash(f"스크립트 실행 실패: {e.stderr}", "danger")
except Exception as e:
logging.error(f"GUID 리스트 처리 오류: {e}")
flash(f"GUID 리스트 처리 중 오류 발생: {e}", "danger")
return redirect(url_for("main.index"))
@utils_bp.route("/update_gpu_list", methods=["POST"])
@login_required
def update_gpu_list():
"""
GPU 시리얼용 리스트(gpu_serial_list.txt)를 갱신하고 Excel을 생성합니다.
- form name="gpu_list_content" 로 내용 전달 (S/T 목록 라인별)
- txt_to_excel.py --preset gpu --list-file <gpu_serial_list.txt>
"""
content = request.form.get("server_list_content")
if not content:
flash("내용을 입력하세요.", "warning")
return redirect(url_for("main.index"))
server_list_dir = Path(Config.SERVER_LIST_FOLDER)
list_path = server_list_dir / "gpu_list.txt"
# txt_to_excel.py는 server_list 폴더에 둔다고 가정 (위치 다르면 경로만 수정)
script_path = server_list_dir / "GPUTOExecl.py"
try:
# 1) gpu_serial_list.txt 저장
list_path.write_text(content, encoding="utf-8")
# 2) 엑셀 생성 실행 (GPU 프리셋)
cmd = [
sys.executable,
str(Path(Config.SERVER_LIST_FOLDER) / "unified_excel_generator.py"),
"--mode", "gpu",
"--list-file", str(list_path),
]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True,
cwd=str(server_list_dir), # data/server_list 기준 실행
timeout=300,
)
logging.info(f"[GPU] 리스트 스크립트 실행 STDOUT:\n{result.stdout}")
if result.stderr:
logging.warning(f"[GPU] 리스트 스크립트 STDERR:\n{result.stderr}")
flash("GPU 리스트가 업데이트되었습니다.", "success")
except subprocess.CalledProcessError as e:
logging.error(f"[GPU] 스크립트 오류: {e.stderr}")
flash(f"스크립트 실행 실패: {e.stderr}", "danger")
except Exception as e:
logging.error(f"[GPU] 처리 오류: {e}")
flash(f"GPU 리스트 처리 중 오류 발생: {e}", "danger")
return redirect(url_for("main.index"))
logging.info(f"엑셀 파일 다운로드: {path}")
return send_file(str(path), as_attachment=True, download_name="mac_info.xlsx")
@utils_bp.route("/scan_network", methods=["POST"])
@login_required
def scan_network():
"""
지정된 IP 범위(Start ~ End)에 대해 Ping 테스트를 수행하고
응답이 있는 IP 목록을 반환합니다.
"""
try:
import ipaddress
import platform
import concurrent.futures
data = request.get_json(force=True, silent=True) or {}
start_ip_str = data.get('start_ip')
end_ip_str = data.get('end_ip')
if not start_ip_str or not end_ip_str:
return jsonify({"success": False, "error": "시작 IP와 종료 IP를 모두 입력해주세요."}), 400
try:
start_ip = ipaddress.IPv4Address(start_ip_str)
end_ip = ipaddress.IPv4Address(end_ip_str)
if start_ip > end_ip:
return jsonify({"success": False, "error": "시작 IP가 종료 IP보다 큽니다."}), 400
# IP 개수 제한 (너무 많은 스캔 방지, 예: C클래스 2개 분량 512개)
if int(end_ip) - int(start_ip) > 512:
return jsonify({"success": False, "error": "스캔 범위가 너무 넓습니다. (최대 512개)"}), 400
except ValueError:
return jsonify({"success": False, "error": "유효하지 않은 IP 주소 형식입니다."}), 400
# Ping 함수 정의
def ping_ip(ip_obj):
ip = str(ip_obj)
param = '-n' if platform.system().lower() == 'windows' else '-c'
timeout_param = '-w' if platform.system().lower() == 'windows' else '-W'
# Windows: -w 200 (ms), Linux: -W 1 (s)
timeout_val = '200' if platform.system().lower() == 'windows' else '1'
command = ['ping', param, '1', timeout_param, timeout_val, ip]
try:
# shell=False로 보안 강화, stdout/stderr 무시
res = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return ip if res.returncode == 0 else None
except Exception:
return None
active_ips = []
# IP 리스트 생성
target_ips = []
temp_ip = start_ip
while temp_ip <= end_ip:
target_ips.append(temp_ip)
temp_ip += 1
# 병렬 처리 (최대 50 쓰레드)
with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
results = executor.map(ping_ip, target_ips)
# 결과 수집 (None 제외)
active_ips = [ip for ip in results if ip is not None]
return jsonify({
"success": True,
"active_ips": active_ips,
"count": len(active_ips),
"message": f"스캔 완료: {len(active_ips)}개의 활성 IP 발견"
})
except Exception as e:
logging.error(f"Scan network fatal error: {e}")
return jsonify({"success": False, "error": f"서버 내부 오류: {str(e)}"}), 500
@utils_bp.route("/api/system/control", methods=["POST"])
@login_required
def system_control():
"""
시스템 제어 명령 실행 (Power ON/OFF, Log Clear 등)
"""
data = request.get_json(silent=True) or {}
action = data.get("action")
ips = data.get("ips", [])
if not action or not ips:
return jsonify({"success": False, "error": "Action 또는 IP 목록이 없습니다."}), 400
# 통합 스크립트 사용
script_name = "unified/system_control.py"
script_path = Path(Config.SCRIPT_FOLDER) / script_name
if not script_path.exists():
return jsonify({"success": False, "error": f"스크립트 파일을 찾을 수 없습니다: {script_name}"}), 500
# 임시 IP 파일 생성
import tempfile
try:
# data/temp/uploads 폴더 사용
temp_dir = Path(Config.UPLOAD_FOLDER)
temp_dir.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(mode='w+', delete=False, dir=temp_dir, suffix=".txt", encoding='utf-8') as tf:
tf.write("\n".join(ips))
temp_ip_file = tf.name
# 스크립트 실행: python unified/system_control.py <action> <ip_file>
cmd = [sys.executable, str(script_path), action, temp_ip_file]
env = os.environ.copy()
if action == "tsr_save":
env["TSR_SHARE_URL"] = get_system_setting("tsr_share_url", "//10.10.3.15/share/")
env["OME_USER"] = get_system_setting("tsr_share_user", env.get("OME_USER", ""))
env["OME_PASS"] = get_system_setting("tsr_share_password", env.get("OME_PASS", ""))
logging.info(f"[SystemControl] Executing {action} with {len(ips)} IPs (Unified Script)...")
# 타임아웃 넉넉하게 (작업에 따라 다름)
timeout = 1200 if action == "tsr_save" else 600 if action == "tsr_collect" else 300
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False, # 에러 발생해도 결과 받음
timeout=timeout,
env=env
)
# 임시 파일 삭제
try:
os.remove(temp_ip_file)
except OSError:
pass
output = "\n".join(
part for part in [result.stdout.strip(), result.stderr.strip()] if part
)
if result.returncode == 0:
logging.info(f"[SystemControl] Success: {output}")
return jsonify({
"success": True,
"message": f"작업({action})이 성공적으로 완료되었습니다.",
"details": output
})
else:
logging.error(f"[SystemControl] Failed: {output}")
return jsonify({
"success": False,
"error": f"스크립트 실행 중 오류가 발생했습니다.\n{output}"
}), 500
except subprocess.TimeoutExpired:
return jsonify({"success": False, "error": "스크립트 실행 시간이 초과되었습니다."}), 504
except Exception as e:
logging.error(f"[SystemControl] Exception: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@utils_bp.route("/api/export_by_service_tag", methods=["POST"])
@login_required
def export_by_service_tag():
"""
서비스태그 목록(텍스트 입력 또는 .txt 파일)을 받아
하위 폴더에서 <서비스태그>.txt 파일을 검색하고 ZIP으로 묶어 다운로드.
"""
tags_text = request.form.get("service_tags", "").strip()
tag_file = request.files.get("tag_file")
if tag_file and tag_file.filename:
try:
tags_text = tag_file.read().decode("utf-8", errors="ignore")
except Exception as e:
return jsonify({"success": False, "error": f"파일 읽기 실패: {e}"}), 400
if not tags_text:
return jsonify({"success": False, "error": "서비스태그를 입력하거나 .txt 파일을 업로드하세요."}), 400
# 서비스태그 파싱 (줄 단위, 중복 제거, 공백 제거)
tags = list(dict.fromkeys(
line.strip() for line in tags_text.splitlines() if line.strip()
))
if not tags:
return jsonify({"success": False, "error": "유효한 서비스태그가 없습니다."}), 400
# 검색 대상 디렉토리 (재귀 검색)
search_dirs = [
Path(Config.BACKUP_FOLDER), # data/system/backup (날짜 하위폴더 포함)
Path(Config.MAC_FOLDER), # data/repository/mac
Path(Config.GUID_FOLDER), # data/repository/guid_file
Path(Config.GPU_FOLDER), # data/repository/gpu_serial
Path(Config.IDRAC_INFO_FOLDER), # data/temp/staging
]
found = [] # (Path, arcname) 튜플 목록
not_found = []
for tag in tags:
target_name = f"{tag}.txt"
tag_found = False
for search_dir in search_dirs:
if not search_dir.exists():
continue
matches = list(search_dir.rglob(target_name))
if matches:
# 검색된 파일만, 파일명 그대로 (flat 구조)
found.append((matches[0], matches[0].name))
tag_found = True
break
if not tag_found:
not_found.append(tag)
if not found:
msg = f"검색된 파일이 없습니다. ({len(not_found)}개 태그)"
return jsonify({"success": False, "error": msg, "not_found": not_found}), 404
# 사용자 지정 파일명 처리
user_zip_name = request.form.get("zip_name", "").strip()
# 위험 문자 제거
import re
safe_name = re.sub(r'[\\/:*?"<>|]', '_', user_zip_name) if user_zip_name else ""
if not safe_name:
safe_name = f"service_tag_export_{time.strftime('%Y%m%d_%H%M%S')}"
zip_name = safe_name if safe_name.endswith(".zip") else safe_name + ".zip"
zip_path = Path(Config.TEMP_ZIP_FOLDER) / zip_name
try:
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zipf:
for file_path, arcname in found:
# writestr 사용 → 폴더 구조 없이 파일명만 ZIP에 저장
with open(file_path, "rb") as f:
zipf.writestr(arcname, f.read())
not_found_header = ",".join(not_found) if not_found else ""
response = send_file(
str(zip_path),
as_attachment=True,
download_name=zip_name,
mimetype="application/zip",
)
response.headers["X-Found-Count"] = str(len(found))
response.headers["X-Not-Found-Count"] = str(len(not_found))
response.headers["X-Not-Found-Tags"] = not_found_header
response.headers["Access-Control-Expose-Headers"] = (
"X-Found-Count, X-Not-Found-Count, X-Not-Found-Tags"
)
return response
except Exception as e:
logging.error(f"[ServiceTagExport] ZIP 생성 오류: {e}")
return jsonify({"success": False, "error": str(e)}), 500
finally:
try:
if zip_path.exists():
zip_path.unlink()
except Exception as e:
logging.warning(f"임시 ZIP 삭제 실패: {e}")
+146
View File
@@ -0,0 +1,146 @@
from __future__ import annotations
import logging
import unicodedata
from pathlib import Path
from flask import Blueprint, render_template, request, redirect, url_for, flash
from flask_login import login_required
from config import Config
xml_bp = Blueprint("xml", __name__)
def register_xml_routes(app):
app.register_blueprint(xml_bp)
def allowed_file(filename: str) -> bool:
return "." in filename and filename.rsplit(".", 1)[1].lower() in Config.ALLOWED_EXTENSIONS
def sanitize_preserve_unicode(filename: str) -> str:
"""
디렉터리 탐색/제어 문자를 차단하면서, 한글/유니코드 파일명은 그대로 보존합니다.
- 경로 요소 제거 (Path(...).name)
- 유니코드 정규화(NFC)로 OS간 차이 최소화
- 널문자/슬래시/역슬래시 차단
"""
name = Path(filename).name
name = unicodedata.normalize("NFC", name)
if not name or any(ch in name for ch in ["\x00", "/", "\\"]):
raise ValueError("잘못된 파일명입니다.")
return name
@xml_bp.route("/xml_management")
@login_required
def xml_management():
xml_dir = Path(Config.XML_FOLDER)
try:
files = sorted([f.name for f in xml_dir.iterdir() if f.is_file()])
except FileNotFoundError:
files = []
flash("XML 폴더가 존재하지 않습니다.", "danger")
return render_template("manage_xml.html", xml_files=files)
@xml_bp.route("/upload_xml", methods=["POST"])
@login_required
def upload_xml():
file = request.files.get("xmlFile")
if not file or not file.filename:
flash("업로드할 파일을 선택하세요.", "warning")
return redirect(url_for("xml.xml_management"))
if not allowed_file(file.filename):
flash("XML 확장자만 업로드할 수 있습니다.", "warning")
return redirect(url_for("xml.xml_management"))
try:
filename = sanitize_preserve_unicode(file.filename) # 한글/유니코드 보존
except ValueError:
flash("파일명이 올바르지 않습니다.", "danger")
return redirect(url_for("xml.xml_management"))
save_path = Path(Config.XML_FOLDER) / filename
try:
save_path.parent.mkdir(parents=True, exist_ok=True)
file.save(str(save_path))
# 텍스트 파일 권장 권한 (Windows에서는 무시될 수 있음)
try:
save_path.chmod(0o644)
except Exception:
pass
logging.info(f"XML 업로드됨: {filename}")
flash("파일이 성공적으로 업로드되었습니다.", "success")
except Exception as e:
logging.error(f"파일 업로드 오류: {e}")
flash("파일 저장 중 오류가 발생했습니다.", "danger")
return redirect(url_for("xml.xml_management"))
@xml_bp.route("/delete_xml/<path:filename>", methods=["POST"])
@login_required
def delete_xml(filename: str):
try:
safe_name = sanitize_preserve_unicode(filename)
except ValueError:
flash("잘못된 파일명입니다.", "danger")
return redirect(url_for("xml.xml_management"))
path = Path(Config.XML_FOLDER) / safe_name
if not path.exists():
flash("해당 파일이 존재하지 않습니다.", "warning")
return redirect(url_for("xml.xml_management"))
try:
path.unlink()
flash(f"{safe_name} 파일이 삭제되었습니다.", "success")
logging.info(f"XML 삭제됨: {safe_name}")
except Exception as e:
logging.error(f"XML 삭제 오류: {e}")
flash("파일 삭제 중 오류 발생", "danger")
return redirect(url_for("xml.xml_management"))
@xml_bp.route("/edit_xml/<path:filename>", methods=["GET", "POST"])
@login_required
def edit_xml(filename: str):
try:
safe_name = sanitize_preserve_unicode(filename)
except ValueError:
flash("잘못된 파일명입니다.", "danger")
return redirect(url_for("xml.xml_management"))
path = Path(Config.XML_FOLDER) / safe_name
if not path.exists():
flash("파일을 찾을 수 없습니다.", "danger")
return redirect(url_for("xml.xml_management"))
if request.method == "POST":
raw = request.form.get("content", "")
# 1) 개행 통일: CRLF/CR → LF
normalized = raw.replace("\r\n", "\n").replace("\r", "\n")
try:
# 2) 항상 LF로 저장 (Windows에서도 강제)
path.write_text(normalized, encoding="utf-8", newline="\n")
logging.info(f"XML 수정됨: {safe_name}")
flash("파일이 성공적으로 수정되었습니다.", "success")
return redirect(url_for("xml.xml_management"))
except Exception as e:
logging.error(f"XML 저장 실패: {e}")
flash("파일 저장 중 오류가 발생했습니다.", "danger")
try:
# 보기/편집 일관성을 위해 읽을 때도 LF로 맞춰서 textarea에 넣음
content = path.read_text(encoding="utf-8").replace("\r\n", "\n").replace("\r", "\n")
except Exception as e:
logging.error(f"XML 열기 실패: {e}")
flash("파일 열기 중 오류가 발생했습니다.", "danger")
content = ""
return render_template("edit_xml.html", filename=safe_name, content=content)