707 lines
32 KiB
Python
707 lines
32 KiB
Python
import os
|
||
import uuid
|
||
import smtplib
|
||
import pytz
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from itsdangerous import URLSafeTimedSerializer
|
||
from werkzeug.security import generate_password_hash, check_password_hash
|
||
from flask import Flask, render_template, jsonify, redirect, url_for, request, session, send_from_directory, make_response
|
||
from icalendar import Calendar, Event as ICalEvent
|
||
from .db import init_db, engine
|
||
from . import config, auth, assistant
|
||
from sqlmodel import Session, select
|
||
from .models import User, Event, GlobalStatus, Project, PortalFile, UserSession, ProjectMember
|
||
from datetime import datetime, timezone, timedelta
|
||
from functools import wraps
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = config.WEB_SECRET
|
||
serializer = URLSafeTimedSerializer(app.secret_key)
|
||
|
||
MSK = pytz.timezone("Europe/Moscow")
|
||
|
||
def to_msk(dt):
|
||
"""UTC → Europe/Moscow для отображения в шаблонах и API."""
|
||
if dt is None: return None
|
||
if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(MSK)
|
||
|
||
@app.template_filter("msk_strftime")
|
||
def _msk_strftime(dt, fmt):
|
||
if dt is None: return ""
|
||
return to_msk(dt).strftime(fmt)
|
||
|
||
UPLOAD_FOLDER = '/var/lib/gitea/portal-files'
|
||
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
||
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024
|
||
|
||
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()
|
||
|
||
@app.before_request
|
||
def validate_session():
|
||
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 not db_sess:
|
||
session.clear()
|
||
return redirect(url_for("login"))
|
||
|
||
def send_email(to_email, subject, body):
|
||
if not config.SMTP_SERVER:
|
||
print("\n" + "="*50, flush=True)
|
||
print(f"📧 EMAIL ДЛЯ: {to_email}", flush=True)
|
||
print(f"Тема: {subject}", flush=True)
|
||
print(body, flush=True)
|
||
print("="*50 + "\n", flush=True)
|
||
return True
|
||
|
||
try:
|
||
msg = MIMEMultipart()
|
||
msg['From'] = config.SMTP_FROM
|
||
msg['To'] = to_email
|
||
msg['Subject'] = subject
|
||
msg.attach(MIMEText(body, 'plain'))
|
||
|
||
server = smtplib.SMTP(config.SMTP_SERVER, config.SMTP_PORT)
|
||
server.starttls()
|
||
server.login(config.SMTP_USER, config.SMTP_PASSWORD)
|
||
server.send_message(msg)
|
||
server.quit()
|
||
return True
|
||
except Exception as e:
|
||
print(f"Ошибка отправки email: {e}", flush=True)
|
||
return False
|
||
|
||
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
|
||
|
||
def login_required(f):
|
||
@wraps(f)
|
||
def decorated_function(*args, **kwargs):
|
||
if not session.get("user_id"):
|
||
return redirect(url_for("login"))
|
||
return f(*args, **kwargs)
|
||
return decorated_function
|
||
|
||
def parse_datetime(dt_str):
|
||
"""Ввод из формы — в Europe/Moscow, сохраняем как UTC."""
|
||
if not dt_str: return None
|
||
clean_str = dt_str.replace('T', ' ').strip()
|
||
if len(clean_str) > 16: clean_str = clean_str[:16]
|
||
naive = datetime.strptime(clean_str, "%Y-%m-%d %H:%M")
|
||
return MSK.localize(naive).astimezone(timezone.utc)
|
||
|
||
@app.route("/healthz")
|
||
def healthz():
|
||
try:
|
||
from sqlalchemy import text
|
||
with engine.connect() as conn:
|
||
conn.execute(text("SELECT 1"))
|
||
return jsonify({"status": "ok", "db": "ok"})
|
||
except Exception as e:
|
||
return jsonify({"status": "error", "db": str(e)}), 503
|
||
|
||
@app.route("/")
|
||
def index():
|
||
return render_template("index.html", username=session.get("username"))
|
||
|
||
# --- АУТЕНТИФИКАЦИЯ ---
|
||
@app.route("/register", methods=["GET", "POST"])
|
||
def register():
|
||
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=False)
|
||
db_session.add(new_user)
|
||
db_session.commit()
|
||
|
||
token = serializer.dumps(email, salt='email-confirm')
|
||
confirm_url = url_for('verify_email', token=token, _external=True).replace("127.0.0.1:8000", "portal.morozovka.space").replace("http://", "https://")
|
||
send_email(email, "Подтверждение почты", f"Привет, {full_name}!\nДля подтверждения перейди по ссылке:\n{confirm_url}\n")
|
||
|
||
return render_template("login.html", msg="Регистрация успешна. Проверьте почту (или логи сервера).")
|
||
return render_template("register.html")
|
||
|
||
@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="Подтвердите почту перед входом")
|
||
|
||
session_token = str(uuid.uuid4())
|
||
db_session.add(UserSession(session_token=session_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"] = session_token
|
||
return redirect(url_for("index"))
|
||
|
||
msg = request.args.get("msg")
|
||
return render_template("login.html", msg=msg)
|
||
|
||
@app.route("/login/gitea")
|
||
def login_gitea():
|
||
gitea_auth_url = (
|
||
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}"
|
||
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 "Ошибка профиля Gitea", 400
|
||
|
||
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()
|
||
bootstrap_role = "admin" if not existing_admin else "student"
|
||
user = User(gitea_id=gitea_user["id"], username=gitea_user["login"], full_name=gitea_user.get("full_name"), email=gitea_user.get("email"), role=bootstrap_role, is_verified=True)
|
||
db_session.add(user)
|
||
db_session.commit()
|
||
|
||
session_token = str(uuid.uuid4())
|
||
db_session.add(UserSession(session_token=session_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"] = session_token
|
||
return redirect(url_for("index"))
|
||
|
||
@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}\n")
|
||
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"))
|
||
return render_template("profile.html", user=user)
|
||
|
||
# --- РАСПИСАНИЕ (ПУБЛИЧНОЕ И АДМИНКА) ---
|
||
@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 - timedelta(days=180))
|
||
.where(Event.start_time <= now + timedelta(days=365))
|
||
.order_by(Event.start_time)
|
||
).all()
|
||
|
||
events_json = [
|
||
{
|
||
"id": e.id,
|
||
"title": e.title,
|
||
"type": e.event_type,
|
||
"start": to_msk(e.start_time).isoformat(),
|
||
"end": to_msk(e.end_time).isoformat(),
|
||
"location": e.location,
|
||
"description": e.description,
|
||
}
|
||
for e in events
|
||
]
|
||
return render_template("schedule.html", status=current_status, events_json=events_json)
|
||
|
||
@app.route("/schedule/export.ics")
|
||
def export_ics():
|
||
with Session(engine) as db_session:
|
||
events = db_session.exec(select(Event).where(Event.is_public == True)).all()
|
||
|
||
cal = Calendar()
|
||
cal.add('prodid', '-//Morozovka Portal//mxm.dk//')
|
||
cal.add('version', '2.0')
|
||
|
||
for ev in events:
|
||
ievent = ICalEvent()
|
||
ievent.add('summary', ev.title)
|
||
ievent.add('dtstart', ev.start_time.astimezone(pytz.utc))
|
||
ievent.add('dtend', ev.end_time.astimezone(pytz.utc))
|
||
if ev.location:
|
||
ievent.add('location', ev.location)
|
||
if ev.description:
|
||
ievent.add('description', ev.description)
|
||
cal.add_component(ievent)
|
||
|
||
response = make_response(cal.to_ical())
|
||
response.headers["Content-Disposition"] = "attachment; filename=morozovka_schedule.ics"
|
||
response.headers["Content-Type"] = "text/calendar; charset=utf-8"
|
||
return response
|
||
|
||
@app.route("/api/assistant", methods=["POST"])
|
||
def api_assistant():
|
||
data = request.get_json()
|
||
if not data or "message" not in data: return jsonify({"error": "Нет сообщения"}), 400
|
||
return jsonify({"reply": assistant.get_response(data["message"])})
|
||
|
||
@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()
|
||
events = db_session.exec(select(Event).order_by(Event.start_time.desc()).limit(50)).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_time = parse_datetime(request.form.get("start_time"))
|
||
end_time = parse_datetime(request.form.get("end_time"))
|
||
with Session(engine) as db_session:
|
||
db_session.add(Event(title=title, event_type=event_type, start_time=start_time, end_time=end_time, location=request.form.get("location"), description=request.form.get("description"), is_public=request.form.get("is_public") == "on"))
|
||
db_session.commit()
|
||
return redirect(url_for("admin_schedule"))
|
||
|
||
@app.route("/admin/schedule/event/<int:event_id>/edit", methods=["GET", "POST"])
|
||
@require_role(["admin", "professor"])
|
||
def edit_event(event_id):
|
||
with Session(engine) as db_session:
|
||
event = db_session.get(Event, event_id)
|
||
if not event: return "Событие не найдено", 404
|
||
|
||
if request.method == "POST":
|
||
event.title = request.form.get("title")
|
||
event.event_type = request.form.get("event_type")
|
||
event.start_time = parse_datetime(request.form.get("start_time"))
|
||
event.end_time = parse_datetime(request.form.get("end_time"))
|
||
event.location = request.form.get("location")
|
||
event.description = request.form.get("description")
|
||
event.is_public = request.form.get("is_public") == "on"
|
||
event.manually_edited = True
|
||
db_session.add(event)
|
||
db_session.commit()
|
||
return redirect(url_for("admin_schedule"))
|
||
|
||
return render_template("admin_event_edit.html", event=event)
|
||
|
||
@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"))
|
||
|
||
# --- НОВЫЕ РОУТЫ: ИМПОРТ ICS ---
|
||
@app.route("/admin/schedule/import", methods=["GET", "POST"])
|
||
@require_role(["admin", "professor"])
|
||
def import_ics():
|
||
if request.method == "POST":
|
||
if 'file' not in request.files:
|
||
return "Файл не загружен", 400
|
||
file = request.files['file']
|
||
if file.filename == '':
|
||
return "Файл не выбран", 400
|
||
|
||
if file and file.filename.endswith('.ics'):
|
||
try:
|
||
cal = Calendar.from_ical(file.read())
|
||
count = 0
|
||
with Session(engine) as db_session:
|
||
for component in cal.walk():
|
||
if component.name == "VEVENT":
|
||
summary = str(component.get('summary', 'Без названия'))
|
||
dtstart_prop = component.get('dtstart')
|
||
dtend_prop = component.get('dtend')
|
||
|
||
if not dtstart_prop or not dtend_prop:
|
||
continue
|
||
|
||
dtstart = dtstart_prop.dt
|
||
dtend = dtend_prop.dt
|
||
|
||
if not isinstance(dtstart, datetime):
|
||
dtstart = datetime.combine(dtstart, datetime.min.time())
|
||
if not isinstance(dtend, datetime):
|
||
dtend = datetime.combine(dtend, datetime.min.time())
|
||
|
||
if dtstart.tzinfo is None:
|
||
dtstart = dtstart.replace(tzinfo=timezone.utc)
|
||
if dtend.tzinfo is None:
|
||
dtend = dtend.replace(tzinfo=timezone.utc)
|
||
|
||
loc = str(component.get('location', ''))
|
||
desc = str(component.get('description', ''))
|
||
|
||
new_event = Event(
|
||
title=summary,
|
||
event_type='lecture',
|
||
start_time=dtstart.astimezone(timezone.utc),
|
||
end_time=dtend.astimezone(timezone.utc),
|
||
location=loc,
|
||
description=desc,
|
||
is_public=True
|
||
)
|
||
db_session.add(new_event)
|
||
count += 1
|
||
db_session.commit()
|
||
return redirect(url_for("admin_schedule"))
|
||
except Exception as e:
|
||
return f"Ошибка при чтении файла .ics: {e}", 400
|
||
return render_template("admin_schedule_import.html")
|
||
|
||
# --- ПРОЕКТЫ ---
|
||
@app.route("/projects")
|
||
def projects_list():
|
||
user_role = session.get("role")
|
||
user_id = session.get("user_id")
|
||
with Session(engine) as db_session:
|
||
projects = db_session.exec(select(Project).order_by(Project.created_at.desc())).all()
|
||
member_project_ids = []
|
||
if user_id:
|
||
memberships = db_session.exec(select(ProjectMember).where(ProjectMember.user_id == user_id)).all()
|
||
member_project_ids = [m.project_id for m in memberships]
|
||
|
||
filtered_projects = []
|
||
for p in projects:
|
||
if p.visibility == "public":
|
||
filtered_projects.append(p)
|
||
elif p.visibility == "internal" and user_id:
|
||
filtered_projects.append(p)
|
||
elif p.visibility == "private" and user_id and (p.owner_id == user_id or p.id in member_project_ids or user_role == "admin"):
|
||
filtered_projects.append(p)
|
||
|
||
return render_template("projects.html", projects=filtered_projects, role=user_role, user_id=user_id, member_ids=member_project_ids)
|
||
|
||
@app.route("/projects/new", methods=["GET", "POST"])
|
||
@require_role(["admin", "professor", "lead"])
|
||
def project_new():
|
||
if request.method == "POST":
|
||
with Session(engine) as db_session:
|
||
new_proj = Project(
|
||
title=request.form.get("title"), description=request.form.get("description"),
|
||
status=request.form.get("status"), visibility=request.form.get("visibility"),
|
||
gitea_url=request.form.get("gitea_url"), owner_id=session.get("user_id")
|
||
)
|
||
db_session.add(new_proj)
|
||
db_session.commit()
|
||
db_session.refresh(new_proj)
|
||
db_session.add(ProjectMember(project_id=new_proj.id, user_id=session.get("user_id"), role="lead"))
|
||
db_session.commit()
|
||
return redirect(url_for("projects_list"))
|
||
return render_template("project_form.html", project=None, is_edit=False)
|
||
|
||
@app.route("/projects/<int:project_id>")
|
||
def project_detail(project_id):
|
||
user_id = session.get("user_id")
|
||
user_role = session.get("role")
|
||
with Session(engine) as db_session:
|
||
proj = db_session.get(Project, project_id)
|
||
if not proj: return "Проект не найден", 404
|
||
|
||
members = db_session.exec(select(ProjectMember, User).join(User, ProjectMember.user_id == User.id).where(ProjectMember.project_id == project_id)).all()
|
||
is_member = any(m[0].user_id == user_id for m in members)
|
||
|
||
if proj.visibility == "private" and not is_member and user_role != "admin":
|
||
return "403: Это приватный проект", 403
|
||
if proj.visibility == "internal" and not user_id:
|
||
return redirect(url_for("login"))
|
||
return render_template("project_detail.html", project=proj, members=members, is_member=is_member, user_id=user_id, role=user_role)
|
||
|
||
@app.route("/projects/<int:project_id>/edit", methods=["GET", "POST"])
|
||
@login_required
|
||
def project_edit(project_id):
|
||
with Session(engine) as db_session:
|
||
proj = db_session.get(Project, project_id)
|
||
if not proj: return "Проект не найден", 404
|
||
|
||
membership = db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id).where(ProjectMember.user_id == session.get("user_id"))).first()
|
||
is_lead = membership and membership.role == "lead"
|
||
|
||
if session.get("role") != "admin" and proj.owner_id != session.get("user_id") and not is_lead:
|
||
return "403: У вас нет прав на редактирование", 403
|
||
|
||
if request.method == "POST":
|
||
proj.title = request.form.get("title")
|
||
proj.description = request.form.get("description")
|
||
proj.status = request.form.get("status")
|
||
proj.visibility = request.form.get("visibility")
|
||
proj.gitea_url = request.form.get("gitea_url")
|
||
db_session.add(proj)
|
||
db_session.commit()
|
||
return redirect(url_for("project_detail", project_id=proj.id))
|
||
return render_template("project_form.html", project=proj, is_edit=True)
|
||
|
||
@app.route("/projects/<int:project_id>/members/add", methods=["POST"])
|
||
@login_required
|
||
def project_add_member(project_id):
|
||
username = request.form.get("username")
|
||
with Session(engine) as db_session:
|
||
proj = db_session.get(Project, project_id)
|
||
if not proj: return "Проект не найден", 404
|
||
|
||
membership = db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id).where(ProjectMember.user_id == session.get("user_id"))).first()
|
||
if session.get("role") != "admin" and proj.owner_id != session.get("user_id") and (not membership or membership.role != "lead"):
|
||
return "403: Нет прав", 403
|
||
|
||
user_to_add = db_session.exec(select(User).where(User.username == username)).first()
|
||
if not user_to_add: return redirect(url_for("project_detail", project_id=project_id))
|
||
|
||
existing = db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id).where(ProjectMember.user_id == user_to_add.id)).first()
|
||
if not existing:
|
||
db_session.add(ProjectMember(project_id=project_id, user_id=user_to_add.id, role="member"))
|
||
db_session.commit()
|
||
return redirect(url_for("project_detail", project_id=project_id))
|
||
|
||
@app.route("/projects/<int:project_id>/members/<int:user_id>/remove", methods=["POST"])
|
||
@login_required
|
||
def project_remove_member(project_id, user_id):
|
||
with Session(engine) as db_session:
|
||
proj = db_session.get(Project, project_id)
|
||
if not proj: return "Проект не найден", 404
|
||
|
||
membership = db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id).where(ProjectMember.user_id == session.get("user_id"))).first()
|
||
if session.get("role") != "admin" and proj.owner_id != session.get("user_id") and (not membership or membership.role != "lead"):
|
||
return "403: Нет прав", 403
|
||
|
||
if user_id == proj.owner_id: return "Нельзя удалить владельца", 400
|
||
|
||
to_remove = db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id).where(ProjectMember.user_id == user_id)).first()
|
||
if to_remove:
|
||
db_session.delete(to_remove)
|
||
db_session.commit()
|
||
return redirect(url_for("project_detail", project_id=project_id))
|
||
|
||
@app.route("/projects/<int:project_id>/delete", methods=["POST"])
|
||
@login_required
|
||
def project_delete(project_id):
|
||
with Session(engine) as db_session:
|
||
proj = db_session.get(Project, project_id)
|
||
if proj and (session.get("role") == "admin" or proj.owner_id == session.get("user_id")):
|
||
for m in db_session.exec(select(ProjectMember).where(ProjectMember.project_id == project_id)).all():
|
||
db_session.delete(m)
|
||
db_session.delete(proj)
|
||
db_session.commit()
|
||
return redirect(url_for("projects_list"))
|
||
|
||
# --- ФАЙЛЫ ---
|
||
@app.route("/files")
|
||
@login_required
|
||
def files_list():
|
||
with Session(engine) as db_session:
|
||
files = db_session.exec(select(PortalFile, User.username).join(User, PortalFile.uploaded_by == User.id).order_by(PortalFile.uploaded_at.desc())).all()
|
||
return render_template("files.html", files=files)
|
||
|
||
@app.route("/files/upload", methods=["POST"])
|
||
@login_required
|
||
def file_upload():
|
||
if 'file' not in request.files: return redirect(url_for('files_list'))
|
||
file = request.files['file']
|
||
if file.filename == '': return redirect(url_for('files_list'))
|
||
if file:
|
||
original_name = file.filename
|
||
ext = os.path.splitext(original_name)[1]
|
||
stored_name = f"{uuid.uuid4().hex}{ext}"
|
||
filepath = os.path.join(app.config['UPLOAD_FOLDER'], stored_name)
|
||
file.save(filepath)
|
||
with Session(engine) as db_session:
|
||
db_session.add(PortalFile(original_name=original_name, stored_name=stored_name, size_bytes=os.path.getsize(filepath), uploaded_by=session.get("user_id")))
|
||
db_session.commit()
|
||
return redirect(url_for('files_list'))
|
||
|
||
@app.route("/files/download/<int:file_id>")
|
||
@login_required
|
||
def file_download(file_id):
|
||
with Session(engine) as db_session:
|
||
db_file = db_session.get(PortalFile, file_id)
|
||
if db_file:
|
||
return send_from_directory(app.config['UPLOAD_FOLDER'], db_file.stored_name, as_attachment=True, download_name=db_file.original_name)
|
||
return "Файл не найден", 404
|
||
|
||
@app.route("/files/<int:file_id>/delete", methods=["POST"])
|
||
@login_required
|
||
def file_delete(file_id):
|
||
with Session(engine) as db_session:
|
||
db_file = db_session.get(PortalFile, file_id)
|
||
if db_file and (session.get("role") == "admin" or db_file.uploaded_by == session.get("user_id")):
|
||
filepath = os.path.join(app.config['UPLOAD_FOLDER'], db_file.stored_name)
|
||
if os.path.exists(filepath): os.remove(filepath)
|
||
db_session.delete(db_file)
|
||
db_session.commit()
|
||
return redirect(url_for("files_list"))
|
||
|
||
# --- АДМИНКА: ПОЛЬЗОВАТЕЛИ ---
|
||
@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("/admin/users/<int:user_id>/delete", methods=["POST"])
|
||
@require_role("admin")
|
||
def delete_user(user_id):
|
||
if user_id == session.get("user_id"): return "Нельзя удалить себя", 400
|
||
with Session(engine) as db_session:
|
||
user = db_session.get(User, user_id)
|
||
if not user:
|
||
return redirect(url_for("admin_users"))
|
||
|
||
owned_projects = db_session.exec(select(Project).where(Project.owner_id == user_id)).all()
|
||
uploaded_files = db_session.exec(select(PortalFile).where(PortalFile.uploaded_by == user_id)).all()
|
||
if owned_projects or uploaded_files:
|
||
return (f"Нельзя удалить {user.username}: у него {len(owned_projects)} проект(ов) "
|
||
f"и {len(uploaded_files)} файл(ов). Сначала передайте их другому пользователю."), 400
|
||
|
||
for pm in db_session.exec(select(ProjectMember).where(ProjectMember.user_id == user_id)).all():
|
||
db_session.delete(pm)
|
||
for s in db_session.exec(select(UserSession).where(UserSession.user_id == user_id)).all():
|
||
db_session.delete(s)
|
||
db_session.delete(user)
|
||
db_session.commit()
|
||
return redirect(url_for("admin_users"))
|