Initial commit: InviteMe event registration app
This commit is contained in:
+436
@@ -0,0 +1,436 @@
|
||||
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 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
|
||||
|
||||
main_bp = Blueprint("main", __name__)
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Helpers
|
||||
# ---------------------------
|
||||
|
||||
def allowed_file(filename, extensions):
|
||||
return "." in filename and filename.rsplit(".", 1)[1].lower() in extensions
|
||||
|
||||
|
||||
def login_user(role):
|
||||
session["role"] = role
|
||||
session.permanent = True
|
||||
|
||||
|
||||
def logout_user():
|
||||
session.pop("role", None)
|
||||
|
||||
|
||||
def require_role(*roles):
|
||||
def decorator(f):
|
||||
def wrapped(*args, **kwargs):
|
||||
if session.get("role") not in roles:
|
||||
flash("Требуется авторизация", "warning")
|
||||
return redirect(url_for("main.login"))
|
||||
return f(*args, **kwargs)
|
||||
wrapped.__name__ = f.__name__
|
||||
return wrapped
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# 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()
|
||||
config = current_app.config
|
||||
|
||||
if username == config["ADMIN_USERNAME"] and password == config["ADMIN_PASSWORD"]:
|
||||
login_user("admin")
|
||||
return redirect(url_for("main.admin_dashboard"))
|
||||
elif username == config.get("STAFF_USERNAME", "staff") and password == config.get("STAFF_PASSWORD", "staff"):
|
||||
login_user("staff")
|
||||
return redirect(url_for("main.check_in"))
|
||||
else:
|
||||
flash("Неверный логин или пароль", "danger")
|
||||
|
||||
return render_template("login.html")
|
||||
|
||||
|
||||
@main_bp.route("/logout")
|
||||
def logout():
|
||||
logout_user()
|
||||
flash("Вы вышли из системы", "info")
|
||||
return redirect(url_for("main.login"))
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Admin Dashboard
|
||||
# ---------------------------
|
||||
|
||||
@main_bp.route("/admin")
|
||||
@require_role("admin")
|
||||
def admin_dashboard():
|
||||
total = Guest.query.filter_by(is_active=True).count()
|
||||
checked_in = db.session.query(CheckIn).count()
|
||||
not_checked_in = total - checked_in
|
||||
attendance_percent = round((checked_in / total) * 100, 1) if total else 0
|
||||
|
||||
recent_check_ins = (
|
||||
CheckIn.query
|
||||
.order_by(CheckIn.checked_in_at.desc())
|
||||
.limit(10)
|
||||
.all()
|
||||
)
|
||||
|
||||
return render_template(
|
||||
"admin/dashboard.html",
|
||||
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")
|
||||
def admin_upload():
|
||||
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"}):
|
||||
filename = secure_filename(excel_file.filename)
|
||||
upload_path = Path(current_app.config["UPLOAD_FOLDER"]) / "temp" / filename
|
||||
upload_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
excel_file.save(str(upload_path))
|
||||
|
||||
try:
|
||||
result = import_guests_from_excel(str(upload_path))
|
||||
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(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")
|
||||
finally:
|
||||
if upload_path.exists():
|
||||
upload_path.unlink()
|
||||
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")
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Guests management
|
||||
# ---------------------------
|
||||
|
||||
@main_bp.route("/admin/guests")
|
||||
@require_role("admin")
|
||||
def admin_guests():
|
||||
search = request.args.get("q", "").strip()
|
||||
query = Guest.query
|
||||
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", guests=guests, search=search)
|
||||
|
||||
|
||||
@main_bp.route("/admin/guests/add", methods=["GET", "POST"])
|
||||
@require_role("admin")
|
||||
def admin_guest_add():
|
||||
if request.method == "POST":
|
||||
guest = Guest(
|
||||
department=request.form.get("department", "").strip(),
|
||||
full_name=request.form.get("full_name", "").strip(),
|
||||
position=request.form.get("position", "").strip(),
|
||||
email=request.form.get("email", "").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", guest=None)
|
||||
|
||||
|
||||
@main_bp.route("/admin/guests/<int:guest_id>/edit", methods=["GET", "POST"])
|
||||
@require_role("admin")
|
||||
def admin_guest_edit(guest_id):
|
||||
guest = Guest.query.get_or_404(guest_id)
|
||||
if request.method == "POST":
|
||||
guest.department = request.form.get("department", "").strip()
|
||||
guest.full_name = request.form.get("full_name", "").strip()
|
||||
guest.position = request.form.get("position", "").strip()
|
||||
guest.email = request.form.get("email", "").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", guest=guest)
|
||||
|
||||
|
||||
@main_bp.route("/admin/guests/<int:guest_id>/delete", methods=["POST"])
|
||||
@require_role("admin")
|
||||
def admin_guest_delete(guest_id):
|
||||
guest = Guest.query.get_or_404(guest_id)
|
||||
guest.is_active = False
|
||||
db.session.commit()
|
||||
flash("Гость заблокирован (QR-код аннулирован)", "success")
|
||||
return redirect(url_for("main.admin_guests"))
|
||||
|
||||
|
||||
@main_bp.route("/admin/guests/<int:guest_id>/resend", methods=["POST"])
|
||||
@require_role("admin")
|
||||
def admin_guest_resend(guest_id):
|
||||
guest = Guest.query.get_or_404(guest_id)
|
||||
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")
|
||||
def admin_send_invitations():
|
||||
guests = Guest.query.filter_by(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")
|
||||
def admin_report():
|
||||
report_path = Path(current_app.config["UPLOAD_FOLDER"]) / "reports"
|
||||
report_path.mkdir(parents=True, exist_ok=True)
|
||||
filename = f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
||||
output_path = report_path / filename
|
||||
|
||||
try:
|
||||
export_report_to_excel(str(output_path))
|
||||
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"))
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Check-in
|
||||
# ---------------------------
|
||||
|
||||
@main_bp.route("/check-in")
|
||||
@require_role("admin", "staff")
|
||||
def check_in():
|
||||
token = request.args.get("token", "").strip()
|
||||
guest = None
|
||||
already_checked_in = False
|
||||
|
||||
if token:
|
||||
guest = Guest.query.filter_by(token=token, is_active=True).first()
|
||||
if guest:
|
||||
already_checked_in = guest.is_checked_in
|
||||
|
||||
return render_template(
|
||||
"checkin/index.html",
|
||||
guest=guest,
|
||||
token=token,
|
||||
already_checked_in=already_checked_in,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# API
|
||||
# ---------------------------
|
||||
|
||||
@main_bp.route("/api/guests/search")
|
||||
@require_role("admin", "staff")
|
||||
def api_guests_search():
|
||||
q = request.args.get("q", "").strip()
|
||||
if len(q) < 2:
|
||||
return jsonify([])
|
||||
|
||||
guests = (
|
||||
Guest.query
|
||||
.filter(Guest.is_active == True)
|
||||
.filter(Guest.full_name.ilike(f"%{q}%"))
|
||||
.order_by(Guest.full_name)
|
||||
.limit(20)
|
||||
.all()
|
||||
)
|
||||
|
||||
return jsonify([
|
||||
{
|
||||
"id": g.id,
|
||||
"full_name": g.full_name,
|
||||
"department": g.department,
|
||||
"position": g.position,
|
||||
"email": g.email,
|
||||
"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,
|
||||
}
|
||||
for g in guests
|
||||
])
|
||||
|
||||
|
||||
@main_bp.route("/api/check-in/<token>", methods=["POST"])
|
||||
@require_role("admin", "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,
|
||||
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.full_name,
|
||||
"department": guest.department,
|
||||
"position": guest.position,
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
@main_bp.route("/api/guest/<token>")
|
||||
@require_role("admin", "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.full_name,
|
||||
"department": guest.department,
|
||||
"position": guest.position,
|
||||
"email": guest.email,
|
||||
"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,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Home
|
||||
# ---------------------------
|
||||
|
||||
@main_bp.route("/")
|
||||
def index():
|
||||
if session.get("role") == "admin":
|
||||
return redirect(url_for("main.admin_dashboard"))
|
||||
elif session.get("role") == "staff":
|
||||
return redirect(url_for("main.check_in"))
|
||||
return redirect(url_for("main.login"))
|
||||
Reference in New Issue
Block a user