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 from app.models import User, Event, Guest, CheckIn from app.services.qr_service import generate_qr_for_guest from app.services.email_service import send_invitation 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 main_bp = Blueprint("main", __name__) # --------------------------- # 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") if user_id: return User.query.get(user_id) return None def login_user(user): session["user_id"] = user.id session["role"] = user.role session.permanent = True def logout_user(): session.pop("user_id", None) session.pop("role", 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()) # --------------------------- # Auth # --------------------------- @main_bp.route("/login", methods=["GET", "POST"]) def login(): 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) if user.is_staff: return redirect(url_for("main.check_in")) return redirect(url_for("main.admin_events")) else: flash("Неверный логин или пароль", "danger") return render_template("login.html") @main_bp.route("/logout") def logout(): 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 = 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) 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 = 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")) 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 db.session.commit() flash("Мероприятие деактивировано", "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 = Path(current_app.config["LOGO_FOLDER"]) 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 = Path(current_app.config["LOGO_FOLDER"]) / event.logo_filename if old_logo.exists(): old_logo.unlink() event.logo_filename = None 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") @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) 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 = Path(current_app.config["PHOTO_FOLDER"]) 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: z.extractall(str(photo_folder)) uploaded_count = len([name for name in z.namelist() if not name.endswith("/")]) 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") 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() query = Guest.query.filter_by(event_id=event.id) if search: query = query.filter(Guest.full_name.ilike(f"%{search}%")) guests = query.order_by(Guest.full_name).all() 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) 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() 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 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) flash("Приглашение повторно отправлено", "success") except Exception as e: flash(f"Ошибка отправки: {str(e)}", "danger") return redirect(url_for("main.admin_guests")) @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 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) 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) 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) 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 db.session.commit() flash("Пользователь заблокирован", "success") return redirect(url_for("main.admin_users")) # --------------------------- # 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") 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 guests = ( query .filter(Guest.full_name.ilike(f"%{q}%")) .order_by(Guest.full_name) .limit(20) .all() ) 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": url_for("static", filename=f"uploads/photos/{g.photo_filename}") if g.photo_filename else "", "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") 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) 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") 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": url_for("static", filename=f"uploads/photos/{guest.photo_filename}") if guest.photo_filename else "", "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(): 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_events")) @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)