288 lines
14 KiB
Python
288 lines
14 KiB
Python
"""Роуты аутентификации: регистрация, логин, OAuth, пароль, сессии, профиль, подписки на расписание."""
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from flask import render_template, request, redirect, url_for, session
|
||
from werkzeug.security import generate_password_hash, check_password_hash
|
||
from sqlmodel import Session, select
|
||
from ..db import engine
|
||
from .. import config, auth
|
||
from ..models import User, UserSession, EventSubscriber, UserOnboarding, Notification
|
||
from ..helpers import (
|
||
serializer, send_email, login_required, get_user_subscription,
|
||
)
|
||
|
||
|
||
def _init_user_onboarding(db_session, user_id):
|
||
"""Создать запись онбординга + приветственное уведомление при регистрации."""
|
||
db_session.add(UserOnboarding(user_id=user_id))
|
||
db_session.add(Notification(
|
||
user_id=user_id,
|
||
title="👋 Добро пожаловать на платформу!",
|
||
body="Пройди короткий тур по разделам — это займёт 2 минуты.",
|
||
link_url="/onboarding",
|
||
kind="info",
|
||
))
|
||
db_session.commit()
|
||
|
||
|
||
def register(app):
|
||
|
||
@app.route("/register", methods=["GET", "POST"])
|
||
def register_view():
|
||
if request.method == "POST":
|
||
username = request.form.get("username")
|
||
email = request.form.get("email")
|
||
password = request.form.get("password")
|
||
full_name = request.form.get("full_name")
|
||
organization = request.form.get("organization")
|
||
with Session(engine) as db_session:
|
||
if db_session.exec(select(User).where(User.username == username)).first():
|
||
return render_template("register.html", error="Логин уже занят")
|
||
if db_session.exec(select(User).where(User.email == email)).first():
|
||
return render_template("register.html", error="Email уже используется")
|
||
new_user = User(username=username, email=email,
|
||
password_hash=generate_password_hash(password),
|
||
full_name=full_name, organization=organization,
|
||
role="student", is_verified=True)
|
||
db_session.add(new_user)
|
||
db_session.commit()
|
||
db_session.refresh(new_user)
|
||
_init_user_onboarding(db_session, new_user.id)
|
||
return render_template("login.html", msg="Регистрация успешна.")
|
||
return render_template("register.html")
|
||
|
||
app.view_functions["register"] = register_view
|
||
|
||
@app.route("/verify_email/<token>")
|
||
def verify_email(token):
|
||
try:
|
||
email = serializer.loads(token, salt='email-confirm', max_age=86400)
|
||
except Exception:
|
||
return "Ссылка недействительна.", 400
|
||
with Session(engine) as db_session:
|
||
user = db_session.exec(select(User).where(User.email == email)).first()
|
||
if user:
|
||
user.is_verified = True
|
||
db_session.add(user)
|
||
db_session.commit()
|
||
return redirect(url_for("login", msg="Почта подтверждена!"))
|
||
return "Пользователь не найден.", 404
|
||
|
||
@app.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if request.method == "POST":
|
||
username = request.form.get("username")
|
||
password = request.form.get("password")
|
||
with Session(engine) as db_session:
|
||
user = db_session.exec(select(User).where(User.username == username)).first()
|
||
if not user or not user.password_hash or not check_password_hash(user.password_hash, password):
|
||
return render_template("login.html", error="Неверный логин или пароль")
|
||
if not user.is_verified:
|
||
return render_template("login.html", error="Подтвердите почту")
|
||
if not user.is_active:
|
||
return render_template("login.html", error="Аккаунт заблокирован")
|
||
token = str(uuid.uuid4())
|
||
db_session.add(UserSession(session_token=token, user_id=user.id,
|
||
ip_address=request.remote_addr,
|
||
user_agent=request.user_agent.string))
|
||
db_session.commit()
|
||
session["user_id"] = user.id
|
||
session["username"] = user.username
|
||
session["role"] = user.role
|
||
session["session_id"] = token
|
||
return redirect(url_for("dashboard"))
|
||
return render_template("login.html", msg=request.args.get("msg"))
|
||
|
||
@app.route("/login/gitea")
|
||
def login_gitea():
|
||
return redirect(f"{config.GITEA_PUBLIC_URL}/login/oauth/authorize"
|
||
f"?client_id={config.GITEA_CLIENT_ID}&response_type=code"
|
||
f"&redirect_uri={config.GITEA_REDIRECT_URI}&scope=read:user")
|
||
|
||
@app.route("/auth/callback")
|
||
def auth_callback():
|
||
code = request.args.get("code")
|
||
if not code:
|
||
return "Ошибка авторизации Gitea", 400
|
||
token = auth.get_gitea_token(code)
|
||
if not token:
|
||
return "Ошибка токена Gitea", 400
|
||
gitea_user = auth.get_gitea_user(token)
|
||
if not gitea_user:
|
||
return "Ошибка профиля Gitea", 400
|
||
created = False
|
||
with Session(engine) as db_session:
|
||
user = db_session.exec(select(User).where(User.gitea_id == gitea_user["id"])).first()
|
||
if not user:
|
||
user = db_session.exec(select(User).where(User.email == gitea_user.get("email"))).first()
|
||
if user:
|
||
user.gitea_id = gitea_user["id"]
|
||
else:
|
||
existing_admin = db_session.exec(select(User).where(User.role == "admin")).first()
|
||
user = User(gitea_id=gitea_user["id"], username=gitea_user["login"],
|
||
full_name=gitea_user.get("full_name"), email=gitea_user.get("email"),
|
||
role=("admin" if not existing_admin else "student"), is_verified=True)
|
||
db_session.add(user)
|
||
created = True
|
||
db_session.commit()
|
||
if created:
|
||
db_session.refresh(user)
|
||
_init_user_onboarding(db_session, user.id)
|
||
token2 = str(uuid.uuid4())
|
||
db_session.add(UserSession(session_token=token2, user_id=user.id,
|
||
ip_address=request.remote_addr,
|
||
user_agent=request.user_agent.string))
|
||
db_session.commit()
|
||
session["user_id"] = user.id
|
||
session["username"] = user.username
|
||
session["role"] = user.role
|
||
session["session_id"] = token2
|
||
return redirect(url_for("onboarding_page") if created else url_for("dashboard"))
|
||
|
||
@app.route("/logout")
|
||
def logout():
|
||
if "session_id" in session:
|
||
with Session(engine) as db_session:
|
||
db_sess = db_session.exec(select(UserSession)
|
||
.where(UserSession.session_token == session["session_id"])).first()
|
||
if db_sess:
|
||
db_session.delete(db_sess)
|
||
db_session.commit()
|
||
session.clear()
|
||
return redirect(url_for("index"))
|
||
|
||
@app.route("/forgot_password", methods=["GET", "POST"])
|
||
def forgot_password():
|
||
if request.method == "POST":
|
||
email = request.form.get("email")
|
||
with Session(engine) as db_session:
|
||
user = db_session.exec(select(User).where(User.email == email)).first()
|
||
if user:
|
||
token = serializer.dumps(email, salt='password-reset')
|
||
reset_url = (url_for('reset_password', token=token, _external=True)
|
||
.replace("127.0.0.1:8000", "portal.morozovka.space")
|
||
.replace("http://", "https://"))
|
||
send_email(email, "Сброс пароля", f"Ссылка:\n{reset_url}")
|
||
return render_template("login.html", msg="Если email есть, инструкция отправлена.")
|
||
return render_template("forgot_password.html")
|
||
|
||
@app.route("/reset_password/<token>", methods=["GET", "POST"])
|
||
def reset_password(token):
|
||
try:
|
||
email = serializer.loads(token, salt='password-reset', max_age=3600)
|
||
except Exception:
|
||
return "Ссылка недействительна.", 400
|
||
if request.method == "POST":
|
||
with Session(engine) as db_session:
|
||
user = db_session.exec(select(User).where(User.email == email)).first()
|
||
if user:
|
||
user.password_hash = generate_password_hash(request.form.get("password"))
|
||
db_session.add(user)
|
||
db_session.commit()
|
||
return redirect(url_for("login", msg="Пароль изменён!"))
|
||
return render_template("reset_password.html", token=token)
|
||
|
||
@app.route("/sessions")
|
||
@login_required
|
||
def sessions_list():
|
||
with Session(engine) as db_session:
|
||
sessions = db_session.exec(
|
||
select(UserSession).where(UserSession.user_id == session.get("user_id"))
|
||
.order_by(UserSession.created_at.desc())
|
||
).all()
|
||
return render_template("sessions.html", sessions=sessions,
|
||
current_token=session.get("session_id"))
|
||
|
||
@app.route("/sessions/logout_all", methods=["POST"])
|
||
@login_required
|
||
def logout_all():
|
||
with Session(engine) as db_session:
|
||
for s in db_session.exec(select(UserSession)
|
||
.where(UserSession.user_id == session.get("user_id"))).all():
|
||
db_session.delete(s)
|
||
db_session.commit()
|
||
session.clear()
|
||
return redirect(url_for("login", msg="Вы вышли со всех устройств."))
|
||
|
||
@app.route("/profile")
|
||
@login_required
|
||
def profile():
|
||
with Session(engine) as db_session:
|
||
user = db_session.get(User, session.get("user_id"))
|
||
subscription = get_user_subscription(db_session, user)
|
||
onboarding = db_session.exec(
|
||
select(UserOnboarding).where(UserOnboarding.user_id == user.id)
|
||
).first()
|
||
return render_template("profile.html", user=user, subscription=subscription,
|
||
onboarding=onboarding)
|
||
|
||
@app.route("/subscribe/toggle", methods=["POST"])
|
||
@login_required
|
||
def subscribe_toggle():
|
||
action = request.form.get("action")
|
||
notify_status = request.form.get("notify_status") == "on"
|
||
notify_events = request.form.get("notify_events") == "on"
|
||
next_url = request.form.get("next") or url_for("profile")
|
||
if not next_url.startswith("/"):
|
||
next_url = url_for("profile")
|
||
with Session(engine) as db_session:
|
||
user = db_session.get(User, session.get("user_id"))
|
||
if not user or not user.email:
|
||
return redirect(next_url)
|
||
existing = get_user_subscription(db_session, user)
|
||
if action == "unsubscribe":
|
||
if existing:
|
||
db_session.delete(existing)
|
||
db_session.commit()
|
||
return redirect(next_url)
|
||
if existing:
|
||
existing.email = user.email
|
||
existing.user_id = user.id
|
||
existing.notify_status = notify_status
|
||
existing.notify_events = notify_events
|
||
existing.is_confirmed = True
|
||
existing.confirm_token = None
|
||
db_session.add(existing)
|
||
else:
|
||
db_session.add(EventSubscriber(
|
||
email=user.email, user_id=user.id,
|
||
unsubscribe_token=str(uuid.uuid4()),
|
||
is_confirmed=True, confirm_token=None,
|
||
notify_status=notify_status, notify_events=notify_events))
|
||
db_session.commit()
|
||
return redirect(next_url)
|
||
|
||
@app.route("/subscribe/confirm/<token>")
|
||
def subscribe_confirm(token):
|
||
with Session(engine) as db_session:
|
||
sub = db_session.exec(select(EventSubscriber)
|
||
.where(EventSubscriber.confirm_token == token)).first()
|
||
if not sub:
|
||
return render_template("subscribe_status.html", title="Ссылка недействительна",
|
||
message="Возможно, устарела.", ok=False)
|
||
sub.is_confirmed = True
|
||
sub.confirm_token = None
|
||
db_session.add(sub)
|
||
db_session.commit()
|
||
return render_template("subscribe_status.html", title="Подписка подтверждена!",
|
||
message=f"Уведомления на {sub.email}.", ok=True,
|
||
unsubscribe_token=sub.unsubscribe_token)
|
||
|
||
@app.route("/unsubscribe/<token>", methods=["GET", "POST"])
|
||
def unsubscribe(token):
|
||
with Session(engine) as db_session:
|
||
sub = db_session.exec(select(EventSubscriber)
|
||
.where(EventSubscriber.unsubscribe_token == token)).first()
|
||
if not sub:
|
||
return render_template("subscribe_status.html", title="Ссылка недействительна",
|
||
message="Такой подписки нет.", ok=False)
|
||
if request.method == "POST":
|
||
db_session.delete(sub)
|
||
db_session.commit()
|
||
return render_template("subscribe_status.html", title="Вы отписаны",
|
||
message="Больше не будет присылать письма.", ok=True)
|
||
return render_template("subscribe_status.html", title="Отписаться?",
|
||
message=f"Вы перестанете получать письма на {sub.email}.",
|
||
ok=True, show_confirm_form=True,
|
||
confirm_action=url_for("unsubscribe", token=token))
|