import io import os import zipfile from datetime import datetime from pathlib import Path from flask import ( Blueprint, render_template, request, redirect, url_for, flash, session, jsonify, send_file, current_app ) from werkzeug.utils import secure_filename from app import db, csrf, limiter from app.models import User, Event, Guest, CheckIn, AuditLog, SmtpSettings from app.services.qr_service import generate_qr_for_guest from app.services.email_service import send_invitation, send_test_email from app.services.excel_service import import_guests_from_excel, export_report_to_excel from app.services.ticket_service import generate_ticket_image, generate_preview_ticket from app.services.audit_service import log_action from app.utils.timezone import utc_to_local, local_to_utc from app.utils.paths import event_photo_folder, event_qr_folder, event_logo_folder, event_tickets_folder, ensure_event_folders main_bp = Blueprint("main", __name__) @main_bp.app_template_filter("localtime") def _localtime_filter(dt, fmt="%d.%m.%Y %H:%M"): if not dt: return "" return utc_to_local(dt).strftime(fmt) @main_bp.app_template_filter("localdate") def _localdate_filter(dt, fmt="%d.%m.%Y"): if not dt: return "" return utc_to_local(dt).strftime(fmt) # --------------------------- # Helpers # --------------------------- def allowed_file(filename, extensions): return "." in filename and filename.rsplit(".", 1)[1].lower() in extensions def current_user(): user_id = session.get("user_id") session_version = session.get("session_version") if user_id: user = User.query.get(user_id) if user and user.is_active and user.session_version == session_version: return user # Сессия устарела или пользователь заблокирован logout_user() return None def login_user(user): session["user_id"] = user.id session["role"] = user.role session["session_version"] = user.session_version session.permanent = True def logout_user(): session.pop("user_id", None) session.pop("role", None) session.pop("session_version", None) session.pop("current_event_id", None) session.pop("checkin_event_id", None) def require_role(*roles): def decorator(f): def wrapped(*args, **kwargs): user = current_user() if not user or not user.is_active or user.role not in roles: flash("Требуется авторизация", "warning") return redirect(url_for("main.login")) return f(*args, **kwargs) wrapped.__name__ = f.__name__ return wrapped return decorator def require_login(f): def wrapped(*args, **kwargs): user = current_user() if not user or not user.is_active: flash("Требуется авторизация", "warning") return redirect(url_for("main.login")) return f(*args, **kwargs) wrapped.__name__ = f.__name__ return wrapped def get_current_event(): event_id = session.get("current_event_id") if event_id: user = current_user() event = Event.query.get(event_id) if event and user and user.can_manage_event(event): return event return None def require_event(f): """Декоратор для маршрутов, требующих выбранного мероприятия (admin/organizer).""" def wrapped(*args, **kwargs): user = current_user() if not user: flash("Требуется авторизация", "warning") return redirect(url_for("main.login")) if not get_current_event(): flash("Сначала выберите мероприятие", "warning") return redirect(url_for("main.admin_events")) return f(*args, **kwargs) wrapped.__name__ = f.__name__ return wrapped def available_events_for_user(user): if user.is_admin: return Event.query.order_by(Event.created_at.desc()) return Event.query.filter_by(created_by_id=user.id).order_by(Event.created_at.desc()) def _guest_matches_search(guest, query): """Регистронезависимый поиск по ФИО, отделу, должности, email и телефону.""" q = query.casefold() fields = [ guest.full_name, guest.last_name, guest.first_name, guest.middle_name, guest.department, guest.position, guest.email, guest.phone, ] return any(f and q in f.casefold() for f in fields) # --------------------------- # Auth # --------------------------- def _is_first_run(): """True, если в базе ещё нет ни одного пользователя.""" return User.query.first() is None @main_bp.route("/setup", methods=["GET", "POST"]) @limiter.limit("10 per minute", methods=["POST"]) def setup(): """Первоначальная настройка: создание первого администратора.""" if not _is_first_run(): return redirect(url_for("main.login")) if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() password_confirm = request.form.get("password_confirm", "").strip() if not username or not password: flash("Логин и пароль обязательны", "danger") return redirect(url_for("main.setup")) if password != password_confirm: flash("Пароли не совпадают", "danger") return redirect(url_for("main.setup")) if User.query.filter_by(username=username).first(): flash("Пользователь с таким логином уже существует", "danger") return redirect(url_for("main.setup")) admin = User(username=username, role="admin", is_active=True) admin.set_password(password) db.session.add(admin) db.session.commit() flash("Первый администратор создан. Войдите в систему.", "success") return redirect(url_for("main.login")) return render_template("setup.html") @main_bp.route("/login", methods=["GET", "POST"]) @limiter.limit("10 per minute", methods=["POST"]) def login(): if _is_first_run(): return redirect(url_for("main.setup")) if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() user = User.query.filter_by(username=username, is_active=True).first() if user and user.check_password(password): login_user(user) log_action("login", target_type="user", target_id=user.id, details="Успешный вход") db.session.commit() if user.is_staff: return redirect(url_for("main.check_in")) return redirect(url_for("main.admin_panel")) else: flash("Неверный логин или пароль", "danger") return render_template("login.html") @main_bp.route("/logout") def logout(): log_action("logout") db.session.commit() logout_user() flash("Вы вышли из системы", "info") return redirect(url_for("main.login")) # --------------------------- # Events # --------------------------- @main_bp.route("/admin/events") @require_role("admin", "organizer") def admin_events(): user = current_user() events = available_events_for_user(user).all() for event in events: event.total_guests = Guest.query.filter_by(event_id=event.id, is_active=True).count() event.checked_in = db.session.query(CheckIn).join(Guest).filter(Guest.event_id == event.id).count() return render_template("admin/events.html", events=events) @main_bp.route("/admin/events/add", methods=["GET", "POST"]) @require_role("admin", "organizer") def admin_event_add(): if request.method == "POST": name = request.form.get("name", "").strip() description = request.form.get("description", "").strip() location = request.form.get("location", "").strip() event_date_str = request.form.get("event_date", "").strip() if not name: flash("Название мероприятия обязательно", "danger") return redirect(url_for("main.admin_event_add")) event_date = None if event_date_str: try: event_date = local_to_utc(datetime.fromisoformat(event_date_str)) except ValueError: flash("Неверный формат даты", "danger") return redirect(url_for("main.admin_event_add")) event = Event( name=name, description=description, location=location, event_date=event_date, created_by_id=current_user().id, ) db.session.add(event) log_action("event_create", target_type="event", target_id=event.id, details=f"Создано мероприятие '{event.name}'") db.session.commit() flash(f"Мероприятие «{event.name}» создано", "success") return redirect(url_for("main.admin_events")) return render_template("admin/event_form.html", event=None) @main_bp.route("/admin/events//edit", methods=["GET", "POST"]) @require_role("admin", "organizer") def admin_event_edit(event_id): event = Event.query.get_or_404(event_id) user = current_user() if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_events")) if request.method == "POST": event.name = request.form.get("name", "").strip() event.description = request.form.get("description", "").strip() event.location = request.form.get("location", "").strip() event_date_str = request.form.get("event_date", "").strip() if not event.name: flash("Название мероприятия обязательно", "danger") return redirect(url_for("main.admin_event_edit", event_id=event.id)) if event_date_str: try: event.event_date = local_to_utc(datetime.fromisoformat(event_date_str)) except ValueError: flash("Неверный формат даты", "danger") return redirect(url_for("main.admin_event_edit", event_id=event.id)) else: event.event_date = None event.is_active = bool(request.form.get("is_active")) log_action("event_update", target_type="event", target_id=event.id, details=f"Обновлено мероприятие '{event.name}'") db.session.commit() flash("Мероприятие обновлено", "success") return redirect(url_for("main.admin_events")) return render_template("admin/event_form.html", event=event) @main_bp.route("/admin/events//delete", methods=["POST"]) @require_role("admin", "organizer") def admin_event_delete(event_id): event = Event.query.get_or_404(event_id) user = current_user() if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_events")) if session.get("current_event_id") == event.id: session.pop("current_event_id", None) event.is_active = False log_action("event_deactivate", target_type="event", target_id=event.id, details=f"Деактивировано мероприятие '{event.name}'") db.session.commit() flash("Мероприятие деактивировано", "success") return redirect(url_for("main.admin_events")) @main_bp.route("/admin/events//delete-hard", methods=["POST"]) @require_role("admin") def admin_event_delete_hard(event_id): """Полное удаление мероприятия вместе со связанными файлами и данными.""" event = Event.query.get_or_404(event_id) reason = request.form.get("reason", "").strip() if not reason: flash("Для удаления необходимо указать причину", "danger") return redirect(url_for("main.admin_events")) if session.get("current_event_id") == event.id: session.pop("current_event_id", None) # Удаляем связанные файлы import shutil def _safe_rmtree(path): try: if path.exists(): shutil.rmtree(str(path)) except Exception as e: current_app.logger.warning(f"Не удалось удалить папку {path}: {e}") # Удаляем папки мероприятия целиком _safe_rmtree(event_photo_folder(event.id)) _safe_rmtree(event_qr_folder(event.id)) _safe_rmtree(event_logo_folder(event.id)) _safe_rmtree(event_tickets_folder(event.id)) event_name = event.name log_action( "event_delete_hard", target_type="event", target_id=event.id, details=f"Удалено мероприятие '{event_name}'. Причина: {reason}", ) db.session.delete(event) db.session.commit() flash(f"Мероприятие «{event_name}» и все связанные данные удалены", "success") return redirect(url_for("main.admin_events")) @main_bp.route("/admin/events//switch", methods=["POST"]) @require_role("admin", "organizer") def admin_event_switch(event_id): event = Event.query.get_or_404(event_id) user = current_user() if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_events")) session["current_event_id"] = event.id flash(f"Выбрано мероприятие: {event.name}", "info") return redirect(url_for("main.admin_dashboard")) @main_bp.route("/admin/events//ticket", methods=["GET", "POST"]) @require_role("admin", "organizer") def admin_event_ticket(event_id): event = Event.query.get_or_404(event_id) user = current_user() if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_events")) if request.method == "POST": event.ticket_title = request.form.get("ticket_title", "").strip() or event.name event.ticket_subtitle = request.form.get("ticket_subtitle", "").strip() event.ticket_body_text = request.form.get("ticket_body_text", "").strip() event.ticket_background_color = request.form.get("ticket_background_color", "#ffffff").strip() event.ticket_text_color = request.form.get("ticket_text_color", "#000000").strip() event.ticket_accent_color = request.form.get("ticket_accent_color", "#0d6efd").strip() event.ticket_show_qr = bool(request.form.get("ticket_show_qr")) event.ticket_show_photo = bool(request.form.get("ticket_show_photo")) event.ticket_layout = request.form.get("ticket_layout", "portrait").strip() # Logo upload logo_file = request.files.get("logo_file") if logo_file and logo_file.filename: if allowed_file(logo_file.filename, {"png", "jpg", "jpeg", "gif", "webp", "bmp", "svg"}): filename = secure_filename(f"event_{event.id}_{logo_file.filename}") logo_folder = event_logo_folder(event.id) logo_folder.mkdir(parents=True, exist_ok=True) logo_path = logo_folder / filename logo_file.save(str(logo_path)) event.logo_filename = filename else: flash("Неверный формат логотипа. Разрешены: png, jpg, jpeg, gif, webp, bmp, svg.", "warning") # Logo deletion if request.form.get("delete_logo") and event.logo_filename: old_logo = event_logo_folder(event.id) / event.logo_filename if old_logo.exists(): old_logo.unlink() event.logo_filename = None log_action("event_ticket_update", target_type="event", target_id=event.id, details=f"Обновлён дизайн билета мероприятия '{event.name}'") db.session.commit() flash("Дизайн билета сохранён", "success") return redirect(url_for("main.admin_event_ticket", event_id=event.id)) return render_template("admin/ticket_constructor.html", event=event) @main_bp.route("/admin/events//ticket/preview") @require_role("admin", "organizer") def admin_event_ticket_preview(event_id): event = Event.query.get_or_404(event_id) user = current_user() if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_events")) try: ticket_path = generate_preview_ticket(event) return send_file(str(ticket_path), mimetype="image/png") except Exception as e: flash(f"Ошибка генерации предпросмотра: {str(e)}", "danger") return redirect(url_for("main.admin_event_ticket", event_id=event.id)) # --------------------------- # Admin Dashboard # --------------------------- @main_bp.route("/admin") @require_role("admin", "organizer", "staff") def admin_index(): return redirect(url_for("main.admin_panel")) @main_bp.route("/admin/dashboard") @require_role("admin", "organizer") @require_event def admin_dashboard(): event = get_current_event() total = Guest.query.filter_by(event_id=event.id, is_active=True).count() checked_in = db.session.query(CheckIn).join(Guest).filter(Guest.event_id == event.id).count() not_checked_in = total - checked_in attendance_percent = round((checked_in / total) * 100, 1) if total else 0 recent_check_ins = ( CheckIn.query .join(Guest) .filter(Guest.event_id == event.id) .order_by(CheckIn.checked_in_at.desc()) .limit(10) .all() ) return render_template( "admin/dashboard.html", event=event, total=total, checked_in=checked_in, not_checked_in=not_checked_in, attendance_percent=attendance_percent, recent_check_ins=recent_check_ins, ) # --------------------------- # Upload Excel / Photos # --------------------------- @main_bp.route("/admin/upload", methods=["GET", "POST"]) @require_role("admin", "organizer") @require_event def admin_upload(): event = get_current_event() if request.method == "POST": # Excel upload excel_file = request.files.get("excel_file") if excel_file and excel_file.filename: if allowed_file(excel_file.filename, {"xlsx", "xls"}): try: excel_stream = io.BytesIO(excel_file.read()) result = import_guests_from_excel(excel_stream, event_id=event.id) flash( f"Импорт завершён: добавлено {result['imported']}, обновлено {result['updated']} из {result['total']}", "success", ) # Generate QR codes for guests without QR base_url = request.form.get("base_url", "").strip() or None guests = Guest.query.filter_by(event_id=event.id).filter(Guest.qr_filename.is_(None)).all() for guest in guests: generate_qr_for_guest(guest, base_url=base_url) log_action( "guests_import", target_type="event", target_id=event.id, details=f"Импорт гостей из Excel: добавлено {result['imported']}, обновлено {result['updated']}", ) db.session.commit() flash(f"Сгенерировано QR-кодов: {len(guests)}", "success") except Exception as e: flash(f"Ошибка импорта Excel: {str(e)}", "danger") else: flash("Неверный формат Excel-файла", "danger") # Photos upload (ZIP or multiple files) photo_files = request.files.getlist("photo_files") zip_file = request.files.get("zip_file") photo_folder = event_photo_folder(event.id) photo_folder.mkdir(parents=True, exist_ok=True) uploaded_count = 0 if zip_file and zip_file.filename: if allowed_file(zip_file.filename, {"zip"}): zip_path = Path(current_app.config["UPLOAD_FOLDER"]) / "temp" / secure_filename(zip_file.filename) zip_path.parent.mkdir(parents=True, exist_ok=True) zip_file.save(str(zip_path)) try: with zipfile.ZipFile(zip_path, "r") as z: extracted = 0 for member in z.infolist(): if member.is_dir(): continue # Защита от path traversal target = (photo_folder / member.filename).resolve() if photo_folder.resolve() not in target.parents and target != photo_folder.resolve(): current_app.logger.warning(f"Пропущен небезопасный путь в ZIP: {member.filename}") continue ext = target.suffix.lower().lstrip(".") if ext not in {"png", "jpg", "jpeg", "gif", "webp", "bmp", "svg"}: continue z.extract(member, str(photo_folder)) extracted += 1 uploaded_count = extracted flash(f"Из ZIP распаковано файлов: {uploaded_count}", "success") except Exception as e: flash(f"Ошибка распаковки ZIP: {str(e)}", "danger") finally: if zip_path.exists(): zip_path.unlink() else: flash("Неверный формат ZIP-архива", "danger") for photo in photo_files: if photo and photo.filename: filename = secure_filename(photo.filename) photo.save(str(photo_folder / filename)) uploaded_count += 1 if uploaded_count: flash(f"Загружено фото: {uploaded_count}", "success") log_action( "photos_upload", target_type="event", target_id=event.id, details=f"Загружено фото гостей: {uploaded_count}", ) db.session.commit() return redirect(url_for("main.admin_upload")) return render_template("admin/upload.html", event=event) # --------------------------- # Guests management # --------------------------- @main_bp.route("/admin/guests") @require_role("admin", "organizer") @require_event def admin_guests(): event = get_current_event() search = request.args.get("q", "").strip() guests = Guest.query.filter_by(event_id=event.id).order_by(Guest.full_name).all() if search: guests = [g for g in guests if _guest_matches_search(g, search)] return render_template("admin/guests.html", event=event, guests=guests, search=search) @main_bp.route("/admin/guests/add", methods=["GET", "POST"]) @require_role("admin", "organizer") @require_event def admin_guest_add(): event = get_current_event() if request.method == "POST": last_name = request.form.get("last_name", "").strip() first_name = request.form.get("first_name", "").strip() middle_name = request.form.get("middle_name", "").strip() full_name = " ".join([p for p in [last_name, first_name, middle_name] if p]) if not full_name: flash("Фамилия и имя обязательны", "danger") return redirect(url_for("main.admin_guest_add")) guest = Guest( event_id=event.id, department=request.form.get("department", "").strip(), last_name=last_name, first_name=first_name, middle_name=middle_name, full_name=full_name, position=request.form.get("position", "").strip(), email=request.form.get("email", "").strip(), phone=request.form.get("phone", "").strip(), photo_filename=request.form.get("photo_filename", "").strip(), ) db.session.add(guest) db.session.commit() base_url = request.form.get("base_url", "").strip() or None generate_qr_for_guest(guest, base_url=base_url) log_action( "guest_create", target_type="guest", target_id=guest.id, details=f"Добавлен гость '{guest.build_full_name()}' ({guest.email})", ) db.session.commit() if request.form.get("send_invitation"): try: send_invitation(guest) flash("Гость добавлен и приглашение отправлено", "success") except Exception as e: flash(f"Гость добавлен, но не удалось отправить письмо: {str(e)}", "warning") else: flash("Гость добавлен", "success") return redirect(url_for("main.admin_guests")) return render_template("admin/guest_form.html", event=event, guest=None) @main_bp.route("/admin/guests//edit", methods=["GET", "POST"]) @require_role("admin", "organizer") @require_event def admin_guest_edit(guest_id): event = get_current_event() guest = Guest.query.filter_by(id=guest_id, event_id=event.id).first_or_404() if request.method == "POST": guest.department = request.form.get("department", "").strip() guest.last_name = request.form.get("last_name", "").strip() guest.first_name = request.form.get("first_name", "").strip() guest.middle_name = request.form.get("middle_name", "").strip() guest.full_name = " ".join([ p for p in [guest.last_name, guest.first_name, guest.middle_name] if p ]) or guest.full_name guest.position = request.form.get("position", "").strip() guest.email = request.form.get("email", "").strip() guest.phone = request.form.get("phone", "").strip() guest.photo_filename = request.form.get("photo_filename", "").strip() log_action( "guest_update", target_type="guest", target_id=guest.id, details=f"Обновлены данные гостя '{guest.build_full_name()}'", ) db.session.commit() flash("Данные гостя обновлены", "success") return redirect(url_for("main.admin_guests")) return render_template("admin/guest_form.html", event=event, guest=guest) @main_bp.route("/admin/guests//delete", methods=["POST"]) @require_role("admin", "organizer") @require_event def admin_guest_delete(guest_id): event = get_current_event() guest = Guest.query.filter_by(id=guest_id, event_id=event.id).first_or_404() guest.is_active = False log_action( "guest_deactivate", target_type="guest", target_id=guest.id, details=f"Гость '{guest.build_full_name()}' заблокирован", ) db.session.commit() flash("Гость заблокирован (QR-код аннулирован)", "success") return redirect(url_for("main.admin_guests")) @main_bp.route("/admin/guests//resend", methods=["POST"]) @require_role("admin", "organizer") @require_event def admin_guest_resend(guest_id): event = get_current_event() guest = Guest.query.filter_by(id=guest_id, event_id=event.id).first_or_404() if not guest.qr_filename: generate_qr_for_guest(guest) db.session.commit() try: send_invitation(guest) log_action( "guest_resend", target_type="guest", target_id=guest.id, details=f"Приглашение повторно отправлено гостю '{guest.build_full_name()}'", ) db.session.commit() flash("Приглашение повторно отправлено", "success") except Exception as e: log_action( "guest_resend", target_type="guest", target_id=guest.id, details=f"Ошибка повторной отправки приглашения гостю '{guest.build_full_name()}': {e}", ) db.session.commit() flash(f"Ошибка отправки: {str(e)}", "danger") return redirect(url_for("main.admin_guests")) @main_bp.route("/admin/guests//ticket") @require_role("admin", "organizer") @require_event def admin_guest_ticket(guest_id): """Генерирует и отдаёт PNG-билет для конкретного гостя.""" event = get_current_event() guest = Guest.query.filter_by(id=guest_id, event_id=event.id).first_or_404() try: ticket_path = generate_ticket_image(guest, event) log_action( "guest_ticket_view", target_type="guest", target_id=guest.id, details=f"Просмотр билета гостя '{guest.build_full_name()}'", ) db.session.commit() return send_file(str(ticket_path), mimetype="image/png") except Exception as e: flash(f"Ошибка генерации билета: {str(e)}", "danger") return redirect(url_for("main.admin_guests")) @main_bp.route("/admin/guests//qr") @require_role("admin", "organizer") @require_event def admin_guest_qr(guest_id): """Отдаёт QR-код гостя без отправки письма.""" event = get_current_event() guest = Guest.query.filter_by(id=guest_id, event_id=event.id).first_or_404() if not guest.qr_filename: generate_qr_for_guest(guest) db.session.commit() qr_path = event_qr_folder(guest.event_id) / guest.qr_filename if not qr_path.exists(): flash("QR-код не найден", "danger") return redirect(url_for("main.admin_guests")) log_action( "guest_qr_view", target_type="guest", target_id=guest.id, details=f"Просмотр QR-кода гостя '{guest.build_full_name()}'", ) db.session.commit() return send_file(str(qr_path), mimetype="image/png") @main_bp.route("/admin/send-invitations", methods=["POST"]) @require_role("admin", "organizer") @require_event def admin_send_invitations(): event = get_current_event() guests = Guest.query.filter_by(event_id=event.id, is_active=True, invitation_sent=False).all() sent = 0 failed = 0 for guest in guests: if not guest.qr_filename: generate_qr_for_guest(guest) try: send_invitation(guest) sent += 1 except Exception: failed += 1 log_action( "event_send_invitations", target_type="event", target_id=event.id, details=f"Рассылка приглашений: отправлено {sent}, ошибок {failed}", ) db.session.commit() flash(f"Отправлено приглашений: {sent}, ошибок: {failed}", "success" if failed == 0 else "warning") return redirect(url_for("main.admin_dashboard")) # --------------------------- # Reports # --------------------------- @main_bp.route("/admin/report") @require_role("admin", "organizer") @require_event def admin_report(): event = get_current_event() report_path = Path(current_app.config["UPLOAD_FOLDER"]) / "reports" report_path.mkdir(parents=True, exist_ok=True) filename = f"report_{event.id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" output_path = report_path / filename try: export_report_to_excel(str(output_path), event_id=event.id, event_name=event.name) log_action( "event_report", target_type="event", target_id=event.id, details=f"Сформирован отчёт по мероприятию '{event.name}'", ) db.session.commit() return send_file(str(output_path), as_attachment=True, download_name=filename) except Exception as e: flash(f"Ошибка формирования отчёта: {str(e)}", "danger") return redirect(url_for("main.admin_dashboard")) # --------------------------- # User management (admin only) # --------------------------- @main_bp.route("/admin/users") @require_role("admin") def admin_users(): users = User.query.order_by(User.created_at.desc()).all() return render_template("admin/users.html", users=users) @main_bp.route("/admin/users/add", methods=["GET", "POST"]) @require_role("admin") def admin_user_add(): if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() role = request.form.get("role", "staff").strip() if not username or not password: flash("Логин и пароль обязательны", "danger") return redirect(url_for("main.admin_user_add")) if role not in ("admin", "organizer", "staff"): flash("Неверная роль", "danger") return redirect(url_for("main.admin_user_add")) if User.query.filter_by(username=username).first(): flash("Пользователь с таким логином уже существует", "danger") return redirect(url_for("main.admin_user_add")) user = User(username=username, role=role, is_active=True) user.set_password(password) db.session.add(user) log_action("user_create", target_type="user", target_id=user.id, details=f"Создан пользователь '{username}' ({role})") db.session.commit() flash(f"Пользователь {username} создан", "success") return redirect(url_for("main.admin_users")) return render_template("admin/user_form.html", user=None) @main_bp.route("/admin/users//edit", methods=["GET", "POST"]) @require_role("admin") def admin_user_edit(user_id): user = User.query.get_or_404(user_id) if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() role = request.form.get("role", user.role).strip() is_active = bool(request.form.get("is_active")) if not username: flash("Логин обязателен", "danger") return redirect(url_for("main.admin_user_edit", user_id=user.id)) if role not in ("admin", "organizer", "staff"): flash("Неверная роль", "danger") return redirect(url_for("main.admin_user_edit", user_id=user.id)) existing = User.query.filter(User.username == username, User.id != user.id).first() if existing: flash("Пользователь с таким логином уже существует", "danger") return redirect(url_for("main.admin_user_edit", user_id=user.id)) user.username = username user.role = role user.is_active = is_active if password: user.set_password(password) log_action("user_update", target_type="user", target_id=user.id, details=f"Обновлён пользователь '{username}' ({role})") db.session.commit() flash("Пользователь обновлён", "success") return redirect(url_for("main.admin_users")) return render_template("admin/user_form.html", user=user) @main_bp.route("/admin/users//delete", methods=["POST"]) @require_role("admin") def admin_user_delete(user_id): user = User.query.get_or_404(user_id) if user.id == current_user().id: flash("Нельзя удалить самого себя", "danger") return redirect(url_for("main.admin_users")) user.is_active = False log_action("user_deactivate", target_type="user", target_id=user.id, details=f"Пользователь '{user.username}' заблокирован") db.session.commit() flash("Пользователь заблокирован", "success") return redirect(url_for("main.admin_users")) @main_bp.route("/admin/logs") @require_role("admin") def admin_logs(): page = request.args.get("page", 1, type=int) per_page = 50 action_filter = request.args.get("action", "").strip() user_filter = request.args.get("user", "").strip() query = AuditLog.query if action_filter: query = query.filter(AuditLog.action == action_filter) if user_filter: query = query.filter(AuditLog.username.ilike(f"%{user_filter}%")) logs = query.order_by(AuditLog.created_at.desc()).paginate( page=page, per_page=per_page, error_out=False ) actions = db.session.query(AuditLog.action).distinct().order_by(AuditLog.action).all() actions = [a[0] for a in actions] return render_template( "admin/logs.html", logs=logs, actions=actions, action_filter=action_filter, user_filter=user_filter, ) # --------------------------- # Admin panel # --------------------------- @main_bp.route("/admin/panel") @require_role("admin", "organizer", "staff") def admin_panel(): """Единая админ-панель со ссылками на все разделы.""" user = current_user() event = get_current_event() if user and (user.is_admin or user.is_organizer) else None cards = [] if user and user.role in ("admin", "organizer"): cards.extend([ { "title": "Мероприятия", "description": "Создание и управление мероприятиями", "url": url_for("main.admin_events"), "icon": "📅", "btn": "Перейти", }, { "title": "Гости", "description": "Список гостей, поиск, редактирование", "url": url_for("main.admin_guests") if event else url_for("main.admin_events"), "icon": "👥", "btn": "Перейти", "disabled": not event, }, { "title": "Загрузка данных", "description": "Импорт гостей из Excel и загрузка фото", "url": url_for("main.admin_upload") if event else url_for("main.admin_events"), "icon": "📤", "btn": "Перейти", "disabled": not event, }, { "title": "Рассылка приглашений", "description": "Настройка почты и массовая отправка", "url": url_for("main.admin_email_send"), "icon": "✉️", "btn": "Перейти", }, { "title": "Конструктор билета", "description": "Дизайн, логотип и предпросмотр билета", "url": url_for("main.admin_event_ticket", event_id=event.id) if event else url_for("main.admin_events"), "icon": "🎫", "btn": "Перейти", "disabled": not event, }, { "title": "Аналитика", "description": "Посещаемость и недавние входы", "url": url_for("main.admin_dashboard") if event else url_for("main.admin_events"), "icon": "📊", "btn": "Перейти", "disabled": not event, }, ]) if user and user.is_admin: cards.extend([ { "title": "Пользователи", "description": "Управление администраторами и стаффом", "url": url_for("main.admin_users"), "icon": "🔐", "btn": "Перейти", }, { "title": "Логи", "description": "История действий пользователей", "url": url_for("main.admin_logs"), "icon": "📝", "btn": "Перейти", }, ]) cards.append({ "title": "Проверка приглашения", "description": "Сканирование QR-кода и отметка входа", "url": url_for("main.check_in"), "icon": "📷", "btn": "Перейти", }) return render_template("admin/panel.html", cards=cards) # --------------------------- # Email / invitations # --------------------------- @main_bp.route("/admin/email/settings", methods=["GET", "POST"]) @require_role("admin") def admin_email_settings(): settings = SmtpSettings.get_settings() if request.method == "POST": mode = request.form.get("mail_mode", "smtp").strip() settings.use_local_outlook = mode == "outlook" settings.use_ews = mode == "ews" if mode == "smtp": settings.mail_server = request.form.get("mail_server", "smtp.yandex.ru").strip() try: settings.mail_port = int(request.form.get("mail_port", "465").strip()) except ValueError: settings.mail_port = 465 settings.mail_use_ssl = bool(request.form.get("mail_use_ssl")) settings.mail_username = request.form.get("mail_username", "").strip() settings.mail_default_sender = request.form.get("mail_default_sender", "").strip() password = request.form.get("mail_password", "").strip() if password: settings.mail_password = password elif mode == "ews": settings.ews_url = request.form.get("ews_url", "").strip() settings.mail_username = request.form.get("mail_username", "").strip() settings.mail_default_sender = request.form.get("mail_default_sender", "").strip() password = request.form.get("mail_password", "").strip() if password: settings.mail_password = password # IMAP: копия отправленных писем (общая для всех режимов отправки) settings.imap_save_copy = bool(request.form.get("imap_save_copy")) settings.imap_server = request.form.get("imap_server", "").strip() try: settings.imap_port = int(request.form.get("imap_port", "993").strip()) except ValueError: settings.imap_port = 993 settings.imap_use_ssl = bool(request.form.get("imap_use_ssl")) settings.imap_username = request.form.get("imap_username", "").strip() settings.imap_folder = request.form.get("imap_folder", "Sent").strip() or "Sent" imap_password = request.form.get("imap_password", "").strip() if imap_password: settings.imap_password = imap_password db.session.commit() SmtpSettings.apply_to_app(current_app) mode_label = {"smtp": "SMTP", "ews": "EWS", "outlook": "Outlook"}.get(mode, mode) log_action( "email_settings_update", target_type="smtp_settings", target_id=settings.id, details=f"Обновлены почтовые настройки ({mode_label})", ) db.session.commit() flash("Настройки почты сохранены", "success") return redirect(url_for("main.admin_email_settings")) return render_template("admin/email_settings.html", settings=settings) @main_bp.route("/admin/email/test", methods=["POST"]) @require_role("admin") def admin_email_test(): test_address = request.form.get("test_address", "").strip() if not test_address: flash("Укажите адрес для тестового письма", "danger") return redirect(url_for("main.admin_email_settings")) try: send_test_email(test_address) log_action( "email_test_send", target_type="smtp_settings", details=f"Тестовое письмо отправлено на {test_address}", ) db.session.commit() flash(f"Тестовое письмо отправлено на {test_address}", "success") except Exception as e: log_action( "email_test_send", target_type="smtp_settings", details=f"Ошибка отправки тестового письма на {test_address}: {e}", ) db.session.commit() flash(f"Ошибка отправки: {str(e)}", "danger") return redirect(url_for("main.admin_email_settings")) @main_bp.route("/admin/email/send", methods=["GET", "POST"]) @require_role("admin", "organizer") def admin_email_send(): user = current_user() events = available_events_for_user(user).all() if request.method == "POST": event_id = request.form.get("event_id", "").strip() if not event_id: flash("Выберите мероприятие", "danger") return redirect(url_for("main.admin_email_send")) event = Event.query.get_or_404(int(event_id)) if not user.can_manage_event(event): flash("Нет доступа к этому мероприятию", "danger") return redirect(url_for("main.admin_email_send")) guests = Guest.query.filter_by( event_id=event.id, is_active=True, invitation_sent=False ).all() if not guests: flash("Нет неотправленных приглашений для выбранного мероприятия", "info") return redirect(url_for("main.admin_email_send")) sent = 0 failed = 0 for guest in guests: if not guest.qr_filename: generate_qr_for_guest(guest) try: send_invitation(guest) sent += 1 db.session.commit() except Exception: failed += 1 db.session.rollback() log_action( "event_send_invitations", target_type="event", target_id=event.id, details=f"Рассылка приглашений из панели: отправлено {sent}, ошибок {failed}", ) db.session.commit() flash(f"Отправлено приглашений: {sent}, ошибок: {failed}", "success" if failed == 0 else "warning") return redirect(url_for("main.admin_email_send")) event_counts = [] for event in events: unsent = Guest.query.filter_by( event_id=event.id, is_active=True, invitation_sent=False ).count() event_counts.append({"event": event, "unsent": unsent}) return render_template("admin/email_send.html", event_counts=event_counts) # --------------------------- # Check-in # --------------------------- @main_bp.route("/check-in", methods=["GET", "POST"]) @require_role("admin", "organizer", "staff") def check_in(): user = current_user() token = request.args.get("token", "").strip() guest = None already_checked_in = False active_events = Event.query.filter_by(is_active=True).order_by(Event.created_at.desc()).all() if request.method == "POST": event_id = request.form.get("event_id", "").strip() if event_id: try: session["checkin_event_id"] = int(event_id) except ValueError: pass return redirect(url_for("main.check_in", token=token)) checkin_event_id = session.get("checkin_event_id") checkin_event = None if checkin_event_id: checkin_event = Event.query.get(checkin_event_id) if token: guest = Guest.query.filter_by(token=token, is_active=True).first() if guest: already_checked_in = guest.is_checked_in if checkin_event is None: checkin_event = guest.event session["checkin_event_id"] = guest.event_id return render_template( "checkin/index.html", guest=guest, token=token, already_checked_in=already_checked_in, active_events=active_events, checkin_event=checkin_event, ) # --------------------------- # API # --------------------------- @main_bp.route("/api/guests/search") @require_role("admin", "organizer", "staff") @csrf.exempt def api_guests_search(): q = request.args.get("q", "").strip() event_id = request.args.get("event_id") or session.get("checkin_event_id") if len(q) < 2: return jsonify([]) query = Guest.query.filter(Guest.is_active == True) if event_id: try: query = query.filter(Guest.event_id == int(event_id)) except ValueError: pass # SQL-фильтр (регистронезависимый для латиницы) + fallback: # если SQLite не нашёл по кириллице, берём всех гостей мероприятия # и фильтруем в Python через casefold(). sql_like = f"%{q}%" candidates = ( query .filter( (Guest.full_name.ilike(sql_like)) | (Guest.department.ilike(sql_like)) | (Guest.email.ilike(sql_like)) ) .order_by(Guest.full_name) .limit(100) .all() ) guests = [g for g in candidates if _guest_matches_search(g, q)] if not guests and event_id: all_event_guests = Guest.query.filter_by( is_active=True, event_id=int(event_id) ).order_by(Guest.full_name).all() guests = [g for g in all_event_guests if _guest_matches_search(g, q)][:20] return jsonify([ { "id": g.id, "full_name": g.build_full_name(), "department": g.department, "position": g.position, "email": g.email, "phone": g.phone, "photo_url": g.photo_url, "token": g.token, "checked_in": g.is_checked_in, "event_id": g.event_id, "event_name": g.event.name if g.event else "", } for g in guests ]) @main_bp.route("/api/check-in/", methods=["POST"]) @require_role("admin", "organizer", "staff") @csrf.exempt def api_check_in(token): guest = Guest.query.filter_by(token=token, is_active=True).first() if not guest: return jsonify({"success": False, "message": "Гость не найден или QR-код аннулирован"}), 404 if guest.is_checked_in: return jsonify({ "success": False, "already_checked_in": True, "checked_in_at": guest.last_check_in.isoformat(), "message": "Гость уже отмечен как вошедший", }), 409 check_in_record = CheckIn( guest_id=guest.id, event_id=guest.event_id, device_info=request.headers.get("User-Agent", "")[:500], ) db.session.add(check_in_record) log_action( "check_in", target_type="guest", target_id=guest.id, details=f"Отмечен вход гостя '{guest.build_full_name()}' на мероприятие '{guest.event.name if guest.event else ''}'", ) db.session.commit() return jsonify({ "success": True, "message": "Вход отмечен", "checked_in_at": check_in_record.checked_in_at.isoformat(), "guest": { "full_name": guest.build_full_name(), "department": guest.department, "position": guest.position, "phone": guest.phone, }, }) @main_bp.route("/api/guest/") @require_role("admin", "organizer", "staff") @csrf.exempt def api_guest_detail(token): guest = Guest.query.filter_by(token=token, is_active=True).first_or_404() return jsonify({ "id": guest.id, "full_name": guest.build_full_name(), "department": guest.department, "position": guest.position, "email": guest.email, "phone": guest.phone, "photo_url": guest.photo_url, "token": guest.token, "checked_in": guest.is_checked_in, "checked_in_at": guest.last_check_in.isoformat() if guest.last_check_in else None, "event_id": guest.event_id, "event_name": guest.event.name if guest.event else "", }) # --------------------------- # Home # --------------------------- @main_bp.route("/") def index(): if _is_first_run(): return redirect(url_for("main.setup")) user = current_user() if not user: return redirect(url_for("main.login")) if user.is_staff: return redirect(url_for("main.check_in")) return redirect(url_for("main.admin_panel")) @main_bp.app_context_processor def inject_globals(): user = current_user() events = [] current_event = None if user and (user.is_admin or user.is_organizer): events = available_events_for_user(user).filter_by(is_active=True).all() event_id = session.get("current_event_id") if event_id: current_event = Event.query.get(event_id) return dict(current_user=user, available_events=events, current_event=current_event)