292 lines
9.3 KiB
Python
292 lines
9.3 KiB
Python
# 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
|
|
try:
|
|
from telegram import Bot
|
|
except ImportError:
|
|
Bot = None
|
|
|
|
admin_bp = Blueprint("admin", __name__)
|
|
|
|
|
|
# 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/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 = []
|
|
|
|
return render_template("admin_settings.html", bots=bots)
|
|
|
|
|
|
@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)
|