231 lines
8.1 KiB
Python
231 lines
8.1 KiB
Python
from flask import Flask, render_template, jsonify, redirect, url_for, request, session
|
|
from .db import init_db, engine
|
|
from . import config, auth
|
|
from sqlmodel import Session, select
|
|
from .models import User, Event, GlobalStatus
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = config.WEB_SECRET
|
|
|
|
# Инициализация БД и дефолтного статуса
|
|
with app.app_context():
|
|
init_db()
|
|
with Session(engine) as db_session:
|
|
status = db_session.exec(select(GlobalStatus)).first()
|
|
if not status:
|
|
db_session.add(GlobalStatus(status="free", message="Вне расписания"))
|
|
db_session.commit()
|
|
|
|
# --- Декоратор проверки ролей ---
|
|
def require_role(allowed_roles):
|
|
if isinstance(allowed_roles, str):
|
|
allowed_roles = [allowed_roles]
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if session.get("role") not in allowed_roles:
|
|
return "403: Доступ запрещен. Недостаточно прав.", 403
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
return decorator
|
|
|
|
@app.route("/healthz")
|
|
def healthz():
|
|
return jsonify({"status": "ok", "service": "morozovka-portal"})
|
|
|
|
@app.route("/")
|
|
def index():
|
|
current_user = session.get("username")
|
|
return render_template("index.html", username=current_user)
|
|
|
|
# --- АВТОРИЗАЦИЯ ---
|
|
@app.route("/login")
|
|
def login():
|
|
gitea_auth_url = (
|
|
f"{config.GITEA_PUBLIC_URL}/login/oauth/authorize"
|
|
f"?client_id={config.GITEA_CLIENT_ID}&response_type=code"
|
|
f"&redirect_uri=https://portal.morozovka.space/auth/callback"
|
|
f"&scope=read:user"
|
|
)
|
|
return redirect(gitea_auth_url)
|
|
|
|
@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 "Ошибка: не удалось получить профиль пользователя.", 400
|
|
|
|
with Session(engine) as db_session:
|
|
user = db_session.exec(select(User).where(User.gitea_id == gitea_user["id"])).first()
|
|
if not user:
|
|
new_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 gitea_user["login"] == "git_admin" else "student"
|
|
)
|
|
db_session.add(new_user)
|
|
db_session.commit()
|
|
user = new_user
|
|
else:
|
|
user.username = gitea_user["login"]
|
|
user.full_name = gitea_user.get("full_name")
|
|
user.email = gitea_user.get("email")
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
|
|
session["user_id"] = user.id
|
|
session["username"] = user.username
|
|
session["role"] = user.role
|
|
|
|
return redirect(url_for("profile"))
|
|
|
|
@app.route("/profile")
|
|
def profile():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return redirect(url_for("login"))
|
|
|
|
with Session(engine) as db_session:
|
|
user = db_session.get(User, user_id)
|
|
if not user:
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
return render_template("profile.html", user=user)
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("index"))
|
|
|
|
# --- АДМИНКА: ПОЛЬЗОВАТЕЛИ ---
|
|
@app.route("/admin/users")
|
|
@require_role("admin")
|
|
def admin_users():
|
|
with Session(engine) as db_session:
|
|
users = db_session.exec(select(User).order_by(User.username)).all()
|
|
return render_template("admin_users.html", users=users)
|
|
|
|
@app.route("/admin/users/<int:user_id>/role", methods=["POST"])
|
|
@require_role("admin")
|
|
def change_user_role(user_id):
|
|
new_role = request.form.get("role")
|
|
valid_roles = ["admin", "professor", "lead", "member", "student", "guest"]
|
|
|
|
if new_role not in valid_roles:
|
|
return "Ошибка: неверная роль", 400
|
|
|
|
with Session(engine) as db_session:
|
|
user = db_session.get(User, user_id)
|
|
if user:
|
|
if user.id == session.get("user_id") and new_role != "admin":
|
|
return "Ошибка: нельзя лишить самого себя прав администратора", 400
|
|
|
|
user.role = new_role
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
|
|
return redirect(url_for("admin_users"))
|
|
|
|
# --- РАСПИСАНИЕ: ПУБЛИЧНАЯ ЧАСТЬ ---
|
|
@app.route("/schedule")
|
|
def public_schedule():
|
|
with Session(engine) as db_session:
|
|
current_status = db_session.exec(select(GlobalStatus)).first()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
events = db_session.exec(
|
|
select(Event)
|
|
.where(Event.is_public == True)
|
|
.where(Event.start_time >= now)
|
|
.order_by(Event.start_time)
|
|
.limit(10)
|
|
).all()
|
|
|
|
return render_template("schedule.html", status=current_status, events=events)
|
|
|
|
# --- РАСПИСАНИЕ: АДМИН-ПАНЕЛЬ ---
|
|
@app.route("/admin/schedule")
|
|
@require_role(["admin", "professor"])
|
|
def admin_schedule():
|
|
with Session(engine) as db_session:
|
|
current_status = db_session.exec(select(GlobalStatus)).first()
|
|
now = datetime.now(timezone.utc)
|
|
events = db_session.exec(
|
|
select(Event)
|
|
.where(Event.start_time >= now)
|
|
.order_by(Event.start_time)
|
|
).all()
|
|
return render_template("admin_schedule.html", status=current_status, events=events)
|
|
|
|
@app.route("/admin/schedule/status", methods=["POST"])
|
|
@require_role(["admin", "professor"])
|
|
def update_status():
|
|
status_val = request.form.get("status")
|
|
delay = int(request.form.get("delay_minutes") or 0)
|
|
message = request.form.get("message")
|
|
|
|
with Session(engine) as db_session:
|
|
current_status = db_session.exec(select(GlobalStatus)).first()
|
|
if current_status:
|
|
current_status.status = status_val
|
|
current_status.delay_minutes = delay
|
|
current_status.message = message
|
|
current_status.updated_at = datetime.now(timezone.utc)
|
|
db_session.add(current_status)
|
|
db_session.commit()
|
|
|
|
return redirect(url_for("admin_schedule"))
|
|
|
|
@app.route("/admin/schedule/event/add", methods=["POST"])
|
|
@require_role(["admin", "professor"])
|
|
def add_event():
|
|
title = request.form.get("title")
|
|
event_type = request.form.get("event_type")
|
|
start_str = request.form.get("start_time")
|
|
end_str = request.form.get("end_time")
|
|
location = request.form.get("location")
|
|
description = request.form.get("description")
|
|
is_public = request.form.get("is_public") == "on"
|
|
|
|
# Парсим дату из input type="datetime-local"
|
|
start_time = datetime.strptime(start_str, "%Y-%m-%dT%H:%M").replace(tzinfo=timezone.utc)
|
|
end_time = datetime.strptime(end_str, "%Y-%m-%dT%H:%M").replace(tzinfo=timezone.utc)
|
|
|
|
with Session(engine) as db_session:
|
|
new_event = Event(
|
|
title=title,
|
|
event_type=event_type,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
location=location,
|
|
description=description,
|
|
is_public=is_public
|
|
)
|
|
db_session.add(new_event)
|
|
db_session.commit()
|
|
|
|
return redirect(url_for("admin_schedule"))
|
|
|
|
@app.route("/admin/schedule/event/<int:event_id>/delete", methods=["POST"])
|
|
@require_role(["admin", "professor"])
|
|
def delete_event(event_id):
|
|
with Session(engine) as db_session:
|
|
event = db_session.get(Event, event_id)
|
|
if event:
|
|
db_session.delete(event)
|
|
db_session.commit()
|
|
return redirect(url_for("admin_schedule"))
|