import os from flask import Blueprint, current_app, flash, redirect, render_template, request, url_for from flask_login import current_user from sqlalchemy import func from app import db from app.auth_utils import admin_required from app.bootstrap import slugify from app.deploy_utils import ( checkout_version, deploy_rebuild, fetch_remote, get_current_version, get_deploy_status, is_deploy_enabled, ) from app.models import AdBanner, Photo, User, UserGroup from app.quota_utils import get_user_folder_count, get_user_photo_count, get_user_storage_used from app.settings_service import get_settings, update_settings_from_form from app.storage_service import delete_photo_file bp = Blueprint("admin", __name__, url_prefix="/admin") @bp.route("/") @admin_required def dashboard(): stats = { "users": User.query.count(), "photos": Photo.query.count(), "admins": User.query.filter_by(is_admin=True).count(), "groups": UserGroup.query.count(), "storage": int( db.session.query(func.coalesce(func.sum(Photo.file_size), 0)).scalar() or 0 ), } recent_users = User.query.order_by(User.created_at.desc()).limit(5).all() recent_photos = Photo.query.order_by(Photo.created_at.desc()).limit(8).all() current_version, _ = get_current_version() return render_template( "admin/dashboard.html", stats=stats, recent_users=recent_users, recent_photos=recent_photos, current_version=current_version, deploy_enabled=is_deploy_enabled(), ) @bp.route("/users") @admin_required def users(): all_users = User.query.order_by(User.created_at.desc()).all() groups = UserGroup.query.order_by(UserGroup.name).all() return render_template("admin/users.html", users=all_users, groups=groups) @bp.route("/users//set-group", methods=["POST"]) @admin_required def set_user_group(user_id): user = User.query.get_or_404(user_id) group_id = request.form.get("group_id", type=int) group = UserGroup.query.get_or_404(group_id) user.group_id = group.id db.session.commit() flash(f"Пользователь {user.username} перемещён в группу «{group.name}»", "success") return redirect(url_for("admin.users")) @bp.route("/users//toggle-admin", methods=["POST"]) @admin_required def toggle_admin(user_id): user = User.query.get_or_404(user_id) if user.id == current_user.id: flash("Нельзя снять права администратора с самого себя", "error") return redirect(url_for("admin.users")) admin_count = User.query.filter_by(is_admin=True).count() if user.is_admin and admin_count <= 1: flash("Нельзя удалить последнего администратора", "error") return redirect(url_for("admin.users")) user.is_admin = not user.is_admin db.session.commit() action = "назначен администратором" if user.is_admin else "лишён прав администратора" flash(f"Пользователь {user.username} {action}", "success") return redirect(url_for("admin.users")) @bp.route("/users//toggle-active", methods=["POST"]) @admin_required def toggle_active(user_id): user = User.query.get_or_404(user_id) if user.id == current_user.id: flash("Нельзя заблокировать самого себя", "error") return redirect(url_for("admin.users")) user.is_active = not user.is_active db.session.commit() action = "разблокирован" if user.is_active else "заблокирован" flash(f"Пользователь {user.username} {action}", "success") return redirect(url_for("admin.users")) @bp.route("/users//delete", methods=["POST"]) @admin_required def delete_user(user_id): user = User.query.get_or_404(user_id) if user.id == current_user.id: flash("Нельзя удалить самого себя", "error") return redirect(url_for("admin.users")) if user.is_admin and User.query.filter_by(is_admin=True).count() <= 1: flash("Нельзя удалить последнего администратора", "error") return redirect(url_for("admin.users")) for photo in user.photos.all(): delete_photo_file(photo.filename, photo.storage_backend) db.session.delete(photo) db.session.delete(user) db.session.commit() flash(f"Пользователь {user.username} удалён", "success") return redirect(url_for("admin.users")) @bp.route("/groups", methods=["GET", "POST"]) @admin_required def groups(): if request.method == "POST": name = request.form.get("name", "").strip() quota_mb = request.form.get("disk_quota_mb", type=int) or 100 max_folders = request.form.get("max_folders", type=int) max_photos = request.form.get("max_photos", type=int) if max_folders is None: max_folders = 10 if max_photos is None: max_photos = 500 if len(name) < 2: flash("Название группы — минимум 2 символа", "error") elif UserGroup.query.filter_by(name=name).first(): flash("Группа с таким названием уже существует", "error") else: slug = slugify(name) base_slug = slug counter = 1 while UserGroup.query.filter_by(slug=slug).first(): slug = f"{base_slug}-{counter}" counter += 1 group = UserGroup( name=name, slug=slug, disk_quota_mb=max(0, quota_mb), max_folders=max(0, max_folders), max_photos=max(0, max_photos), ) db.session.add(group) db.session.commit() flash(f"Группа «{name}» создана", "success") return redirect(url_for("admin.groups")) all_groups = UserGroup.query.order_by(UserGroup.is_default.desc(), UserGroup.name).all() group_stats = [] for group in all_groups: used = sum(get_user_storage_used(u.id) for u in group.users) photos = sum(get_user_photo_count(u.id) for u in group.users) folders = sum(get_user_folder_count(u.id) for u in group.users) group_stats.append({ "group": group, "storage_used": used, "photo_count": photos, "folder_count": folders, }) return render_template("admin/groups.html", group_stats=group_stats) @bp.route("/groups//edit", methods=["POST"]) @admin_required def edit_group(group_id): group = UserGroup.query.get_or_404(group_id) name = request.form.get("name", "").strip() quota_mb = request.form.get("disk_quota_mb", type=int) max_folders = request.form.get("max_folders", type=int) max_photos = request.form.get("max_photos", type=int) if len(name) < 2: flash("Название группы — минимум 2 символа", "error") return redirect(url_for("admin.groups")) other = UserGroup.query.filter(UserGroup.name == name, UserGroup.id != group.id).first() if other: flash("Группа с таким названием уже существует", "error") return redirect(url_for("admin.groups")) group.name = name if quota_mb is not None: group.disk_quota_mb = max(0, quota_mb) if max_folders is not None: group.max_folders = max(0, max_folders) if max_photos is not None: group.max_photos = max(0, max_photos) db.session.commit() flash(f"Группа «{group.name}» обновлена", "success") return redirect(url_for("admin.groups")) @bp.route("/groups//delete", methods=["POST"]) @admin_required def delete_group(group_id): group = UserGroup.query.get_or_404(group_id) if group.is_default: flash("Нельзя удалить группу по умолчанию", "error") return redirect(url_for("admin.groups")) default_group = UserGroup.query.filter_by(is_default=True).first() if not default_group: flash("Не найдена группа по умолчанию", "error") return redirect(url_for("admin.groups")) User.query.filter_by(group_id=group.id).update({"group_id": default_group.id}) db.session.delete(group) db.session.commit() flash(f"Группа удалена, пользователи перенесены в «{default_group.name}»", "success") return redirect(url_for("admin.groups")) @bp.route("/banners", methods=["GET", "POST"]) @admin_required def banners(): if request.method == "POST": title = request.form.get("title", "").strip() image_url = request.form.get("image_url", "").strip() link_url = request.form.get("link_url", "").strip() or None alt_text = request.form.get("alt_text", "").strip() or None position = request.form.get("position", "main").strip() sort_order = request.form.get("sort_order", type=int) or 0 is_active = request.form.get("is_active") == "on" if len(title) < 2: flash("Название баннера — минимум 2 символа", "error") elif not image_url: flash("Укажите URL изображения", "error") elif position not in AdBanner.POSITIONS: flash("Неверная позиция баннера", "error") else: banner = AdBanner( title=title, image_url=image_url, link_url=link_url, alt_text=alt_text or title, position=position, sort_order=sort_order, is_active=is_active, ) db.session.add(banner) db.session.commit() flash(f"Баннер «{title}» добавлен", "success") return redirect(url_for("admin.banners")) all_banners = AdBanner.query.order_by(AdBanner.position, AdBanner.sort_order, AdBanner.id).all() return render_template("admin/banners.html", banners=all_banners, positions=AdBanner.POSITIONS) @bp.route("/banners//edit", methods=["POST"]) @admin_required def edit_banner(banner_id): banner = AdBanner.query.get_or_404(banner_id) title = request.form.get("title", "").strip() image_url = request.form.get("image_url", "").strip() link_url = request.form.get("link_url", "").strip() or None alt_text = request.form.get("alt_text", "").strip() or None position = request.form.get("position", banner.position).strip() sort_order = request.form.get("sort_order", type=int) is_active = request.form.get("is_active") == "on" if len(title) < 2: flash("Название баннера — минимум 2 символа", "error") elif not image_url: flash("Укажите URL изображения", "error") elif position not in AdBanner.POSITIONS: flash("Неверная позиция баннера", "error") else: banner.title = title banner.image_url = image_url banner.link_url = link_url banner.alt_text = alt_text or title banner.position = position if sort_order is not None: banner.sort_order = sort_order banner.is_active = is_active db.session.commit() flash(f"Баннер «{banner.title}» обновлён", "success") return redirect(url_for("admin.banners")) @bp.route("/banners//delete", methods=["POST"]) @admin_required def delete_banner(banner_id): banner = AdBanner.query.get_or_404(banner_id) db.session.delete(banner) db.session.commit() flash("Баннер удалён", "success") return redirect(url_for("admin.banners")) @bp.route("/banners//toggle", methods=["POST"]) @admin_required def toggle_banner(banner_id): banner = AdBanner.query.get_or_404(banner_id) banner.is_active = not banner.is_active db.session.commit() state = "включён" if banner.is_active else "выключен" flash(f"Баннер «{banner.title}» {state}", "success") return redirect(url_for("admin.banners")) @bp.route("/photos") @admin_required def photos(): all_photos = Photo.query.order_by(Photo.created_at.desc()).all() return render_template("admin/photos.html", photos=all_photos) @bp.route("/photos//delete", methods=["POST"]) @admin_required def delete_photo(photo_id): photo = Photo.query.get_or_404(photo_id) delete_photo_file(photo.filename, photo.storage_backend) db.session.delete(photo) db.session.commit() flash("Фото удалено", "success") return redirect(url_for("admin.photos")) @bp.route("/deploy", methods=["GET", "POST"]) @admin_required def deploy(): status = get_deploy_status() if request.method == "POST": action = request.form.get("action") if action == "fetch": ok, msg = fetch_remote() flash(msg if ok else msg, "success" if ok else "error") elif action == "checkout": ref = request.form.get("ref", "").strip() ok, msg = checkout_version(ref) flash(msg, "success" if ok else "error") elif action == "rebuild": ok, msg = deploy_rebuild() flash(msg, "success" if ok else "error") else: flash("Неизвестное действие", "error") return redirect(url_for("admin.deploy")) return render_template("admin/deploy.html", status=status) @bp.route("/settings", methods=["GET", "POST"]) @admin_required def settings(): site_settings = get_settings() if request.method == "POST": action = request.form.get("action", "save") if action == "test_smtp": from app.email_service import send_email ok, msg = send_email( current_user.email, "PhotoHost — тест SMTP", "SMTP настроен корректно.", ) flash(msg, "success" if ok else "error") return redirect(url_for("admin.settings")) update_settings_from_form(request.form) flash("Настройки сохранены", "success") return redirect(url_for("admin.settings")) return render_template("admin/settings.html", settings=site_settings)