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/") @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/") @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/", 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//") @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/", 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/", 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/", 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/", 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/", 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/", 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/", 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))