clubesportsudes/team-tryouts
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
37f70c89e3dcfe08508fb989bc13fb4262dc5e9c
team-tryouts/app/routes/users.py
T
GGThedandClaude Opus 5 37f70c89e3 feat(i18n): traduire les messages flash et de validation 
198 appels flash dans les sept modules de routes, plus les 23 messages de
validation de app/validators.py. Le catalogue compte desormais 631 chaines,
aucune non traduite.

validators.py utilise lazy_gettext : les champs de schema sont construits a
l'import, donc avant qu'une requete existe. Un gettext ordinaire s'y
resoudrait une seule fois, dans la langue active au demarrage.

Un bug introduit par la conversion, puis corrige
  Le convertisseur automatique ne voyait que le premier litteral d'un appel
  flash, ce qui a casse deux chaines concatenees sur plusieurs lignes dans
  users.py -- le resultat n'etait meme pas du Python valide. Ma premiere
  verification ne l'a pas vu : elle enchainait py_compile sur head, or head
  reussit toujours, donc le "OK" s'affichait quoi qu'il arrive. Les deux
  appels sont reecrits et la verification refaite correctement.

Un bug plus interessant, revele par le test de fumee
  La langue choisie ne survivait pas a la connexion. login() et logout()
  appellent tous deux session.clear() -- l'un contre la fixation de session,
  l'autre pour terminer la session -- et le choix de langue partait avec le
  reste. Concretement : quelqu'un qui lisait la page de connexion en anglais
  se retrouvait en francais des qu'il se connectait.

  La langue est une preference d'affichage, pas un etat appartenant au
  compte. Les deux endroits la reportent maintenant explicitement, a cote du
  jeton CSRF. Quatre tests couvrent le cas, dont un qui verifie que corriger
  une cle preservee n'a pas fait tomber l'autre.

Detail de nommage : le convertisseur avait genere %(value)s pour une
expression conditionnelle, ce qui n'aide pas un traducteur. Renomme en
%(player)s.

Les 14 traductions ecrites avec une apostrophe droite sont normalisees en
apostrophe typographique. Sans consequence en HTML, ou ' s'affiche
correctement -- mais les blocs <script> ne decodent pas les entites, et
autant que le catalogue soit homogene.

200 tests.

Co-Authored-By: Claude Opus 5 <[email protected]>
2026-08-07 22:12:27 -04:00

1390 lines
57 KiB
Python
Raw Blame History

"""User management routes for profiles, disponibilities, and contracts.
Uses polymorphic isinstance checks instead of role-string comparisons.
"""
import os
import uuid
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify, send_file
from flask_login import login_required, current_user
from flask_babel import gettext as _
from app.extensions import db, hash_password
from app.models import (
Admin, Manager, Coach, Player, Scout,
User, USER_TYPES, ESPORT_GAMES,
PlayerDisponibility, UserGamertag, GAME_PLATFORMS,
Contract, OrgTeam, CoachAvailability,
TeamNote, PersonalNote, OneOnOneRequest,
Evaluation, Match, Team, TeamMember,
MatchParticipant, Tryout, TryoutRegistration, TeamPlayer,
)
from werkzeug.utils import secure_filename
from datetime import datetime, timedelta
from marshmallow import ValidationError
from app.validators import (
CreateUserSchema, EditUserSchema, EditProfileSchema,
UploadContractSchema,
)
from app.logging_config import log_auth_event
from app.permissions import coach_can_access_player
import requests
ALLOWED_CONTRACT_EXTENSIONS = {'pdf'}
ALLOWED_SIGNED_EXTENSIONS = {'pdf'}
users_bp = Blueprint('users', __name__, url_prefix='/users')
# ---------------------------------------------------------------------------
# Form helpers shared by the schema-validated routes
# ---------------------------------------------------------------------------
def _flash_validation_errors(err):
"""Surface marshmallow errors the same way auth.py already does."""
for field, messages in err.messages.items():
for msg in messages:
flash(_('%(field)s: %(msg)s', field=field, msg=msg), 'danger')
def _form_payload(*, checkboxes=(), list_fields=('games',), optional_blank=('password',)):
"""Turn the multi-valued request form into a plain dict for marshmallow.
request.form.to_dict() keeps only the first value of a repeated key, so
list fields have to be re-read with getlist(). Unchecked HTML checkboxes
are simply absent from the submission, which is not the same as a schema
default, so they are injected explicitly. Blank optional fields are
dropped rather than sent as '' — an empty password means "leave the
current one alone", not "set the password to the empty string".
"""
payload = request.form.to_dict()
for name in list_fields:
payload[name] = request.form.getlist(name)
for name in checkboxes:
payload[name] = name in request.form
for name in optional_blank:
if not payload.get(name):
payload.pop(name, None)
return payload
# ---------------------------------------------------------------------------
# Gamertag helper (shared)
# ---------------------------------------------------------------------------
def update_user_gamertags(user, selected_games):
"""Update gamertags for a user based on form input."""
existing_gamertags = {gt.game: gt for gt in user.gamertags}
for game in selected_games:
gamertag = request.form.get(f'gamertag_{game}', '').strip()
platform = request.form.get(f'platform_{game}', '').strip() if GAME_PLATFORMS.get(game) else None
existing = existing_gamertags.get(game)
if gamertag:
if existing:
existing.gamertag = gamertag
existing.platform = platform
else:
gt = UserGamertag(user_id=user.id, game=game, gamertag=gamertag, platform=platform)
db.session.add(gt)
elif existing:
db.session.delete(existing)
for game in existing_gamertags:
if game not in selected_games:
db.session.delete(existing_gamertags[game])
# ---------------------------------------------------------------------------
# USER_TYPE → Model mapping for create_user
# ---------------------------------------------------------------------------
_USER_CLASS_MAP = {
'admin': Admin,
'manager': Manager,
'coach': Coach,
'player': Player,
'scout': Scout,
}
# ===========================================================================
# ROUTES
# ===========================================================================
@users_bp.route('')
@login_required
def list_users():
"""List all users for management (Admin only)."""
if not isinstance(current_user, Admin):
flash(_('Only the president can manage users.'), 'danger')
return redirect(url_for('main.dashboard'))
users = User.query.order_by(User.role, User.username).all()
return render_template('pages/users.html', users=users, roles=USER_TYPES)
@users_bp.route('/<int:user_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_user(user_id):
"""Edit an existing user (Admin only)."""
if not isinstance(current_user, Admin):
flash(_('Only the president can edit users.'), 'danger')
return redirect(url_for('main.dashboard'))
user = User.query.get_or_404(user_id)
if request.method == 'POST':
# Captured up front, on purpose. A role change below calls
# db.session.remove(), which detaches current_user from the session:
# any later attribute access on it raises DetachedInstanceError.
actor_name, actor_id = current_user.username, current_user.id
def _rerender():
return render_template('pages/edit_user.html', user=user, roles=USER_TYPES,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags={gt.game: {'gamertag': gt.gamertag,
'platform': gt.platform}
for gt in user.gamertags})
try:
validated = EditUserSchema().load(
_form_payload(checkboxes=('is_active_account',))
)
except ValidationError as err:
_flash_validation_errors(err)
return _rerender()
full_name = validated['full_name']
email = validated['email']
phone = validated.get('phone')
role = validated['role']
is_active = validated['is_active_account']
selected_games = validated.get('games', [])
discord_username = validated.get('discord_username')
discord_user_id = validated.get('discord_user_id')
league_os_profile = validated.get('league_os_profile')
# Previously absent: the column is unique, so assigning a taken
# address surfaced as an IntegrityError, i.e. a 500.
clash = User.query.filter(User.email == email, User.id != user.id).first()
if clash:
flash(_('Email already in use by another account.'), 'danger')
return _rerender()
if user.role != role:
# Two ways to lock everyone out of administration, neither of
# which any interface can undo afterwards.
if user.id == actor_id:
flash(_('You cannot change your own role. Ask another '
'president to do it.'), 'danger')
return _rerender()
if user.role == 'admin':
remaining_admins = User.query.filter(
User.role == 'admin',
User.is_active_account.is_(True),
User.id != user.id,
).count()
if remaining_admins == 0:
flash(_('This is the last active president. Promote '
'another account before changing this one.'), 'danger')
return _rerender()
# Change role via raw SQL to avoid polymorphic identity corruption.
# Must discard the entire session because the polymorphic discriminator
# change invalidates the identity map for this instance and anything
# that references it via relationships.
if user.role != role:
log_auth_event('account.role_changed',
actor=actor_name, actor_id=actor_id,
target=user.username, target_id=user.id,
previous_role=user.role, new_role=role)
user_id_local = user.id
db.session.execute(
db.text("UPDATE users SET role = :role WHERE id = :id"),
{"role": role, "id": user_id_local}
)
db.session.commit()
db.session.remove() # discard stale session entirely
user = User.query.get(user_id_local) # fresh session, correct class
user.full_name = full_name
user.email = email
user.phone = phone
user.is_active_account = is_active
user.games = ','.join(selected_games) if selected_games else None
user.discord_username = discord_username or None
user.discord_user_id = discord_user_id or None
user.league_os_profile = league_os_profile or None
update_user_gamertags(user, selected_games)
# Blank means "keep the current password"; anything else has already
# been checked against the policy by the schema.
password = validated.get('password')
if password:
user.password_hash = hash_password(password)
log_auth_event('account.password_reset_by_admin',
actor=actor_name, actor_id=actor_id,
target=user.username, target_id=user.id)
db.session.commit()
log_auth_event('account.updated',
actor=actor_name, actor_id=actor_id,
target=user.username, target_id=user.id, active=is_active)
flash(_('User %(username)s updated successfully!', username=user.username), 'success')
return redirect(url_for('users.list_users'))
user_gamertags = {gt.game: {'gamertag': gt.gamertag, 'platform': gt.platform}
for gt in user.gamertags}
return render_template('pages/edit_user.html', user=user, roles=USER_TYPES,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags=user_gamertags)
@users_bp.route('/<int:user_id>/delete', methods=['POST'])
@login_required
def delete_user(user_id):
"""Delete a user (Admin only)."""
if not isinstance(current_user, Admin):
flash(_('Only the president can delete users.'), 'danger')
return redirect(url_for('main.dashboard'))
if current_user.id == user_id:
flash(_('You cannot delete your own account.'), 'danger')
return redirect(url_for('users.list_users'))
user = User.query.get_or_404(user_id)
Evaluation.query.filter(
db.or_(Evaluation.evaluator_id == user_id, Evaluation.player_id == user_id),
).delete(synchronize_session=False)
PlayerDisponibility.query.filter_by(player_id=user_id).delete()
CoachAvailability.query.filter_by(coach_id=user_id).delete()
PersonalNote.query.filter(
db.or_(PersonalNote.player_id == user_id, PersonalNote.coach_id == user_id),
).delete(synchronize_session=False)
TeamNote.query.filter_by(coach_id=user_id).delete()
OneOnOneRequest.query.filter(
db.or_(OneOnOneRequest.player_id == user_id, OneOnOneRequest.coach_id == user_id),
).delete(synchronize_session=False)
UserGamertag.query.filter_by(user_id=user_id).delete()
Contract.query.filter_by(player_id=user_id).delete()
TryoutRegistration.query.filter_by(player_id=user_id).delete()
TeamPlayer.query.filter_by(player_id=user_id).delete()
TeamMember.query.filter_by(player_id=user_id).delete()
MatchParticipant.query.filter_by(player_id=user_id).delete()
OrgTeam.query.filter_by(coach_id=user_id).update({'coach_id': None})
OrgTeam.query.filter_by(manager_id=user_id).update({'manager_id': None})
Tryout.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
Match.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
Team.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
OrgTeam.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
Contract.query.filter_by(uploaded_by_id=user_id).update({'uploaded_by_id': current_user.id})
deleted_username, deleted_role = user.username, user.role
db.session.delete(user)
db.session.commit()
log_auth_event('account.deleted',
actor=current_user.username, actor_id=current_user.id,
target=deleted_username, target_id=user_id, role=deleted_role)
flash(_('User %(deleted_username)s has been removed.', deleted_username=deleted_username), 'success')
return redirect(url_for('users.list_users'))
@users_bp.route('/create', methods=['GET', 'POST'])
@login_required
def create_user():
"""Create a new user (Admin only). Uses the correct polymorphic subclass."""
if not isinstance(current_user, Admin):
flash(_('Only the president can create users.'), 'danger')
return redirect(url_for('main.dashboard'))
if request.method == 'POST':
try:
validated = CreateUserSchema().load(request.form)
except ValidationError as err:
_flash_validation_errors(err)
return render_template('pages/create_user.html', roles=USER_TYPES)
username = validated['username']
email = validated['email']
password = validated['password']
full_name = validated['full_name']
phone = validated.get('phone')
# The schema constrains role with OneOf(USER_TYPES), so the former
# manual membership check is now redundant.
role = validated['role']
if User.query.filter_by(username=username).first():
flash(_('Username already exists.'), 'danger')
return render_template('pages/create_user.html', roles=USER_TYPES)
if User.query.filter_by(email=email).first():
flash(_('Email already registered.'), 'danger')
return render_template('pages/create_user.html', roles=USER_TYPES)
hashed_password = hash_password(password)
user_cls = _USER_CLASS_MAP.get(role, Player)
user = user_cls(
username=username, password_hash=hashed_password,
role=role, full_name=full_name,
email=email, phone=phone,
)
db.session.add(user)
db.session.commit()
log_auth_event('account.created_by_admin',
actor=current_user.username, actor_id=current_user.id,
target=user.username, target_id=user.id, role=role)
flash(_('User %(full_name)s created as %(role)s!', full_name=full_name, role=role), 'success')
return redirect(url_for('users.list_users'))
return render_template('pages/create_user.html', roles=USER_TYPES)
@users_bp.route('/<int:user_id>/view')
@login_required
def view_user(user_id):
"""View a public profile for any user."""
user = User.query.get_or_404(user_id)
return render_template('pages/view_user.html', profile_user=user)
@users_bp.route('/profile')
@login_required
def profile():
"""View the current user's profile."""
contracts = None
if isinstance(current_user, Player):
contracts = Contract.query.filter_by(
player_id=current_user.id,
).order_by(Contract.uploaded_at.desc()).all()
existing_availability = None
if isinstance(current_user, Coach):
existing_availability = CoachAvailability.query.filter_by(
coach_id=current_user.id,
).all()
return render_template('pages/profile.html', user=current_user, contracts=contracts,
existing_availability=existing_availability)
@users_bp.route('/profile/edit', methods=['GET', 'POST'])
@login_required
def edit_profile():
"""Edit the current user's profile."""
if request.method == 'POST':
try:
validated = EditProfileSchema().load(_form_payload())
except ValidationError as err:
_flash_validation_errors(err)
return render_template('pages/edit_profile.html', user=current_user,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags=current_user.get_gamertags())
username = validated['username']
full_name = validated['full_name']
email = validated['email']
phone = validated.get('phone')
selected_games = validated.get('games', [])
discord_username = validated.get('discord_username')
discord_user_id = validated.get('discord_user_id')
league_os_profile = validated.get('league_os_profile')
if username != current_user.username and User.query.filter_by(username=username).first():
flash(_('Username already taken.'), 'danger')
return render_template('pages/edit_profile.html', user=current_user,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags=current_user.get_gamertags())
if email != current_user.email and User.query.filter_by(email=email).first():
flash(_('Email already in use.'), 'danger')
return render_template('pages/edit_profile.html', user=current_user,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags=current_user.get_gamertags())
current_user.username = username
current_user.full_name = full_name
current_user.email = email
current_user.phone = phone
current_user.games = ','.join(selected_games) if selected_games else None
current_user.discord_username = discord_username or None
current_user.discord_user_id = discord_user_id or None
current_user.league_os_profile = league_os_profile or None
update_user_gamertags(current_user, selected_games)
# Blank means "keep the current password"; anything else has already
# been checked against the policy by the schema.
password = validated.get('password')
if password:
current_user.password_hash = hash_password(password)
log_auth_event('account.password_changed',
username=current_user.username, user_id=current_user.id)
db.session.commit()
flash(_('Profile updated successfully!'), 'success')
return redirect(url_for('users.profile'))
return render_template('pages/edit_profile.html', user=current_user,
esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS,
user_gamertags=current_user.get_gamertags())
# ---------------------------------------------------------------------------
# Disponibilities
# ---------------------------------------------------------------------------
DAY_NAMES = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
def add_30_minutes(t):
return (datetime.combine(datetime.today(), t) + timedelta(minutes=30)).time()
@users_bp.route('/disponibilities')
@login_required
def get_disponibilities():
"""API endpoint to get all player disponibilities for scheduling."""
if not current_user.can_manage_teams() and not current_user.can_schedule_matches():
return jsonify({'error': 'Unauthorized'}), 403
players = User.query.filter_by(role='player', is_active_account=True).order_by(User.username).all()
result = {}
for player in players:
disponibilities = list(player.disponibilities)
result[player.id] = {
'username': player.username,
'disponibilities': [
{
'id': d.id, 'day_of_week': d.day_of_week,
'day_name': DAY_NAMES[d.day_of_week],
'start_time': d.start_time.strftime('%H:%M'),
'end_time': d.end_time.strftime('%H:%M'),
}
for d in disponibilities
],
}
return jsonify(result)
@users_bp.route('/disponibilities/my')
@login_required
def get_my_disponibilities():
"""API endpoint for players to get their own disponibilities."""
disponibilities = PlayerDisponibility.query.filter_by(player_id=current_user.id).all()
result = {}
for d in disponibilities:
day = d.day_of_week
if day not in result:
result[day] = []
result[day].append({
'id': d.id, 'day_of_week': d.day_of_week,
'day_name': DAY_NAMES[d.day_of_week],
'start_time': d.start_time.strftime('%H:%M'),
'end_time': d.end_time.strftime('%H:%M'),
})
return jsonify(result)
@users_bp.route('/disponibilities/add', methods=['POST'])
@login_required
def add_disponibility():
"""Add a disponibility block for the current player."""
day_of_week = request.form.get('day_of_week', type=int)
start_time_str = request.form.get('start_time')
if day_of_week is None or day_of_week < 0 or day_of_week > 6:
return jsonify({'error': 'Invalid day of week'}), 400
try:
start_time = datetime.strptime(start_time_str, '%H:%M').time()
except (ValueError, TypeError):
return jsonify({'error': 'Invalid time format'}), 400
end_time = add_30_minutes(start_time)
disponibility = PlayerDisponibility(
player_id=current_user.id, day_of_week=day_of_week,
start_time=start_time, end_time=end_time,
)
db.session.add(disponibility)
db.session.commit()
return jsonify({
'id': disponibility.id, 'day_of_week': disponibility.day_of_week,
'day_name': DAY_NAMES[disponibility.day_of_week],
'start_time': disponibility.start_time.strftime('%H:%M'),
'end_time': disponibility.end_time.strftime('%H:%M'),
})
@users_bp.route('/disponibilities/add_bulk', methods=['POST'])
@login_required
def add_disponibilities_bulk():
"""Add multiple disponibility blocks at once."""
data = request.get_json()
slots = data.get('slots', [])
created = []
for slot in slots:
day_of_week = slot.get('day_of_week')
start_time_str = slot.get('start_time')
if day_of_week is None or day_of_week < 0 or day_of_week > 6:
continue
try:
start_time = datetime.strptime(start_time_str, '%H:%M').time()
except (ValueError, TypeError):
continue
end_time = add_30_minutes(start_time)
existing = PlayerDisponibility.query.filter_by(
player_id=current_user.id, day_of_week=day_of_week, start_time=start_time,
).first()
if not existing:
disponibility = PlayerDisponibility(
player_id=current_user.id, day_of_week=day_of_week,
start_time=start_time, end_time=end_time,
)
db.session.add(disponibility)
db.session.flush()
created.append({
'id': disponibility.id, 'day_of_week': disponibility.day_of_week,
'day_name': DAY_NAMES[disponibility.day_of_week],
'start_time': disponibility.start_time.strftime('%H:%M'),
})
db.session.commit()
return jsonify({'success': True, 'created': created})
@users_bp.route('/disponibilities/clear', methods=['POST'])
@login_required
def clear_disponibilities():
"""Clear all disponibilities for the current player."""
PlayerDisponibility.query.filter_by(player_id=current_user.id).delete()
db.session.commit()
return jsonify({'success': True})
@users_bp.route('/disponibilities/<int:disponibility_id>/delete', methods=['POST'])
@login_required
def delete_disponibility(disponibility_id):
"""Delete a disponibility block."""
disponibility = PlayerDisponibility.query.get_or_404(disponibility_id)
if disponibility.player_id != current_user.id:
return jsonify({'error': 'Unauthorized'}), 403
db.session.delete(disponibility)
db.session.commit()
return jsonify({'success': True})
# ---------------------------------------------------------------------------
# Contracts
# ---------------------------------------------------------------------------
def can_manage_player_contract(user, player_id):
"""Check if a user can upload contracts for a specific player."""
if isinstance(user, Admin):
return True
if isinstance(user, Manager):
return True
if isinstance(user, Coach):
org_team = OrgTeam.query.filter_by(coach_id=user.id).first()
if org_team:
tp = TeamPlayer.query.filter_by(player_id=player_id, org_team_id=org_team.id).first()
if tp:
return True
return False
@users_bp.route('/contracts')
@login_required
def list_contracts():
"""View contracts for the current user or players they manage."""
contracts = None
players = None
if isinstance(current_user, Player):
contracts = Contract.query.filter_by(
player_id=current_user.id,
).order_by(Contract.uploaded_at.desc()).all()
elif isinstance(current_user, (Admin, Manager, Coach)):
players = []
if isinstance(current_user, Coach):
org_team = OrgTeam.query.filter_by(coach_id=current_user.id).first()
if org_team:
player_ids = [tp.player_id for tp in TeamPlayer.query.filter_by(org_team_id=org_team.id).all()]
players = User.query.filter(User.id.in_(player_ids)).all() if player_ids else []
else:
players = User.query.filter_by(role='player').all()
if players:
player_ids = [p.id for p in players]
contracts = Contract.query.filter(
Contract.player_id.in_(player_ids),
).order_by(Contract.uploaded_at.desc()).all()
return render_template('pages/contracts.html', contracts=contracts, players=players
if isinstance(current_user, (Admin, Manager, Coach)) else None)
@users_bp.route('/contracts/upload', methods=['GET', 'POST'])
@login_required
def upload_contract():
"""Upload a contract for a player."""
if not isinstance(current_user, (Admin, Manager, Coach)):
flash(_('Only presidents, managers, and coaches can upload contracts.'), 'danger')
return redirect(url_for('users.list_contracts'))
if isinstance(current_user, Coach):
org_team = OrgTeam.query.filter_by(coach_id=current_user.id).first()
if org_team:
player_ids = [tp.player_id for tp in TeamPlayer.query.filter_by(org_team_id=org_team.id).all()]
players = User.query.filter(User.id.in_(player_ids)).all() if player_ids else []
else:
players = []
else:
players = User.query.filter_by(role='player').all()
if request.method == 'POST':
contract_schema = UploadContractSchema()
try:
validated = contract_schema.load(request.form)
except ValidationError as err:
for field, messages in err.messages.items():
for msg in messages:
flash(_('%(field)s: %(msg)s', field=field, msg=msg), 'danger')
return render_template('pages/upload_contract.html', players=players)
player_id = validated['player_id']
notes = validated.get('notes')
if not can_manage_player_contract(current_user, player_id):
flash(_('You do not have permission to upload a contract for this player.'), 'danger')
return redirect(url_for('users.upload_contract'))
if 'contract_file' not in request.files:
flash(_('No file selected.'), 'danger')
return redirect(url_for('users.upload_contract'))
file = request.files['contract_file']
if file.filename == '':
flash(_('No file selected.'), 'danger')
return redirect(url_for('users.upload_contract'))
if not file.filename.lower().endswith('.pdf'):
flash(_('Only PDF files are allowed for contracts.'), 'danger')
return redirect(url_for('users.upload_contract'))
upload_dir = os.path.join(os.getcwd(), 'documents', 'contrats signés')
os.makedirs(upload_dir, exist_ok=True)
player = User.query.get_or_404(player_id)
player_teams = player.get_org_teams()
team = player_teams[0] if player_teams else None
if team:
team_folder = os.path.join(upload_dir, secure_filename(team.name))
os.makedirs(team_folder, exist_ok=True)
final_dir = team_folder
else:
final_dir = upload_dir
original_filename = secure_filename(file.filename)
file_uuid = str(uuid.uuid4())
stored_filename = f"{file_uuid}.pdf"
file_path = os.path.join(final_dir, stored_filename)
file.save(file_path)
contract = Contract(
player_id=player_id, team_id=team.id if team else None,
uploaded_by_id=current_user.id,
original_filename=original_filename,
stored_filename=stored_filename,
file_path=file_path,
notes=notes if notes else None,
)
db.session.add(contract)
db.session.commit()
flash(_('Contract uploaded successfully for %(username)s!', username=player.username), 'success')
return redirect(url_for('users.list_contracts'))
return render_template('pages/upload_contract.html', players=players)
@users_bp.route('/contracts/<int:contract_id>/upload_signed', methods=['POST'])
@login_required
def upload_signed_contract(contract_id):
"""Upload a signed contract (player only)."""
contract = Contract.query.get_or_404(contract_id)
if not contract.can_upload_signed(current_user):
flash(_('Only the player can upload their signed contract.'), 'danger')
return redirect(url_for('users.list_contracts'))
if 'signed_file' not in request.files:
flash(_('No file selected.'), 'danger')
return redirect(url_for('users.list_contracts'))
file = request.files['signed_file']
if file.filename == '':
flash(_('No file selected.'), 'danger')
return redirect(url_for('users.list_contracts'))
signed_filename = f"signed_{contract.stored_filename}"
file.save(contract.file_path.replace(contract.stored_filename, signed_filename))
contract.signed_filename = signed_filename
contract.signed_file_path = contract.file_path.replace(contract.stored_filename, signed_filename)
contract.status = 'signed'
contract.signed_at = datetime.utcnow()
db.session.commit()
flash(_('Signed contract uploaded successfully!'), 'success')
return redirect(url_for('users.list_contracts'))
@users_bp.route('/contracts/<int:contract_id>/download')
@login_required
def download_contract(contract_id):
"""Download a contract file."""
contract = Contract.query.get_or_404(contract_id)
if not contract.can_view(current_user):
flash(_('You do not have permission to download this contract.'), 'danger')
return redirect(url_for('users.list_contracts'))
return send_file(contract.file_path, as_attachment=True, download_name=contract.original_filename)
@users_bp.route('/contracts/<int:contract_id>/download_signed')
@login_required
def download_signed_contract(contract_id):
"""Download a signed contract file."""
contract = Contract.query.get_or_404(contract_id)
if not contract.can_view(current_user):
flash(_('You do not have permission to download this contract.'), 'danger')
return redirect(url_for('users.list_contracts'))
if not contract.signed_file_path:
flash(_('No signed contract available.'), 'danger')
return redirect(url_for('users.list_contracts'))
return send_file(contract.signed_file_path, as_attachment=True, download_name=contract.signed_filename)
# ---------------------------------------------------------------------------
# One on One
# ---------------------------------------------------------------------------
DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL', '')
def send_discord_notification(player_name, points, date_str, start_time_str, end_time_str,
team_name, coach_name, coach_discord, coach_discord_id, request_id=None):
"""Send a Discord notification for a One on One request."""
import logging
logger = logging.getLogger(__name__)
if coach_discord_id:
try:
from app.discord_bot import send_one_on_one_dm
send_one_on_one_dm(
coach_name=coach_name, coach_discord_id=coach_discord_id,
player_name=player_name, team_name=team_name,
date_str=date_str, start_time=start_time_str,
end_time=end_time_str, points=points, request_id=request_id,
)
except Exception as e:
logger.warning(f"Failed to send Discord DM: {e}")
if DISCORD_WEBHOOK_URL:
try:
from app.discord_bot import send_one_on_one_dm
if DISCORD_WEBHOOK_URL.isdigit() and not coach_discord_id:
send_one_on_one_dm(
coach_name=coach_name, coach_discord_id=DISCORD_WEBHOOK_URL,
player_name=player_name, team_name=team_name,
date_str=date_str, start_time=start_time_str,
end_time=end_time_str, points=points,
)
elif not DISCORD_WEBHOOK_URL.isdigit():
embed = {
"embeds": [{
"title": "One on One Request", "color": 3447003,
"fields": [
{"name": "Player", "value": player_name, "inline": True},
{"name": "Team", "value": team_name or "Unknown Team", "inline": True},
{"name": "Date", "value": date_str, "inline": True},
{"name": "Time", "value": f"{start_time_str} - {end_time_str}", "inline": True},
{"name": "Discussion Points", "value": points or "No specific points provided", "inline": False},
],
"footer": {
"text": f"Coach: {coach_name}"
+ (f" (Discord: {coach_discord})" if coach_discord else ""),
},
}],
}
requests.post(DISCORD_WEBHOOK_URL, json=embed, timeout=5)
except Exception as e:
logger.warning(f"Failed to send Discord notification: {e}")
@users_bp.route('/one-on-one', methods=['GET', 'POST'])
@login_required
def one_on_one():
"""One on One request page for players."""
if not isinstance(current_user, Player):
flash(_('Only players can request One on One sessions.'), 'danger')
return redirect(url_for('main.dashboard'))
org_teams = current_user.get_org_teams()
org_team = org_teams[0] if org_teams else None
coach = User.query.get(org_team.coach_id) if org_team and org_team.coach_id else None
if not coach:
flash(_('You do not have a coach assigned to your team.'), 'info')
team_notes = []
if org_team:
team_notes = TeamNote.query.filter_by(org_team_id=org_team.id).order_by(TeamNote.created_at.desc()).all()
personal_notes = PersonalNote.query.filter_by(player_id=current_user.id).order_by(PersonalNote.created_at.desc()).all()
coach_availability = []
if coach:
availabilities = CoachAvailability.query.filter_by(coach_id=coach.id).all()
coach_availability = [
{
'day_of_week': av.day_of_week,
'start_time': av.start_time.strftime('%H:%M'),
'end_time': av.end_time.strftime('%H:%M'),
}
for av in availabilities
]
if request.method == 'POST':
date_str = request.form.get('date')
start_time_str = request.form.get('start_time')
end_time_str = request.form.get('end_time')
points = request.form.get('points', '').strip()
if not coach:
flash(_('Cannot request One on One - no coach assigned.'), 'danger')
return redirect(url_for('users.one_on_one'))
try:
date_obj = datetime.strptime(date_str, '%Y-%m-%d').date()
start_time = datetime.strptime(start_time_str, '%H:%M').time()
end_time = datetime.strptime(end_time_str, '%H:%M').time()
except (ValueError, TypeError):
flash(_('Invalid date or time format.'), 'danger')
return redirect(url_for('users.one_on_one'))
check_date = datetime.strptime(date_str, '%Y-%m-%d')
day_of_week = check_date.weekday()
is_available = any(
av['day_of_week'] == day_of_week
and av['start_time'] <= start_time_str
and av['end_time'] >= end_time_str
for av in coach_availability
)
if not is_available:
flash(_("The requested time is not within the coach's availability."), 'danger')
return redirect(url_for('users.one_on_one'))
request_obj = OneOnOneRequest(
player_id=current_user.id, coach_id=coach.id,
org_team_id=org_team.id if org_team else None,
date=date_obj, start_time=start_time, end_time=end_time,
points=points if points else None,
)
db.session.add(request_obj)
db.session.commit()
send_discord_notification(
player_name=current_user.full_name,
points=points, date_str=date_str,
start_time_str=start_time_str, end_time_str=end_time_str,
team_name=org_team.name if org_team else 'Unknown Team',
coach_name=coach.full_name,
coach_discord=coach.discord_username or '',
coach_discord_id=coach.discord_user_id or '',
request_id=request_obj.id,
)
flash(_('Your One on One request has been submitted!'), 'success')
return redirect(url_for('users.one_on_one'))
# Build list of upcoming dates that have coach availability
from datetime import date as date_cls, timedelta as td
today = date_cls.today()
available_days = {av['day_of_week'] for av in coach_availability}
dates = []
for i in range(14): # Next 14 days
d = today + td(days=i)
if d.weekday() in available_days:
dates.append({
'value': d.strftime('%Y-%m-%d'),
'day_of_week': d.weekday(),
'display': d.strftime('%B %d, %Y (%A)'),
})
# Player's own One on One request history
my_requests = OneOnOneRequest.query.filter_by(
player_id=current_user.id
).order_by(OneOnOneRequest.created_at.desc()).all()
return render_template('pages/one_on_one.html',
org_team=org_team, coach=coach,
team_notes=team_notes, personal_notes=personal_notes,
coach_availability=coach_availability,
dates=dates,
my_requests=my_requests)
@users_bp.route('/one-on-one/<int:request_id>/accept', methods=['POST'])
@login_required
def accept_one_on_one(request_id):
"""Coach accepts a One on One request."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can accept One on One requests.'), 'danger')
return redirect(url_for('main.dashboard'))
request_obj = OneOnOneRequest.query.get_or_404(request_id)
if request_obj.coach_id != current_user.id:
flash(_('This request is not for you.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
if request_obj.status != 'pending':
flash(_('This request has already been processed.'), 'info')
return redirect(url_for('users.notes_dashboard'))
player = request_obj.player
request_obj.status = 'approved'
request_obj.responded_at = datetime.utcnow()
db.session.commit()
# Notify player via Discord (same message as if approved through Discord reactions)
if player and player.discord_user_id:
from app.discord_bot import send_one_on_one_response
send_one_on_one_response(
player_discord_id=player.discord_user_id,
player_full_name=player.full_name,
coach_full_name=current_user.full_name,
date_str=request_obj.date.strftime('%A, %B %d, %Y'),
start_time=request_obj.start_time.strftime('%I:%M %p') if request_obj.start_time else 'TBD',
end_time=request_obj.end_time.strftime('%I:%M %p') if request_obj.end_time else 'TBD',
points=request_obj.points or 'No specific points provided',
approved=True,
)
flash(_('One on One request from %(player)s has been approved!', player=player.username if player else 'Unknown'), 'success')
return redirect(url_for('users.notes_dashboard'))
@users_bp.route('/one-on-one/<int:request_id>/reject', methods=['POST'])
@login_required
def reject_one_on_one(request_id):
"""Coach rejects a One on One request."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can reject One on One requests.'), 'danger')
return redirect(url_for('main.dashboard'))
request_obj = OneOnOneRequest.query.get_or_404(request_id)
if request_obj.coach_id != current_user.id:
flash(_('This request is not for you.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
if request_obj.status != 'pending':
flash(_('This request has already been processed.'), 'info')
return redirect(url_for('users.notes_dashboard'))
rejection_reason = request.form.get('rejection_reason', '').strip()
player = request_obj.player
request_obj.status = 'rejected'
request_obj.responded_at = datetime.utcnow()
if rejection_reason:
request_obj.coach_rejection_message = rejection_reason
db.session.commit()
# Notify player via Discord (same message as if rejected through Discord reactions)
if player and player.discord_user_id:
from app.discord_bot import send_one_on_one_response
send_one_on_one_response(
player_discord_id=player.discord_user_id,
player_full_name=player.full_name,
coach_full_name=current_user.full_name,
date_str=request_obj.date.strftime('%A, %B %d, %Y'),
start_time=request_obj.start_time.strftime('%I:%M %p') if request_obj.start_time else 'TBD',
end_time=request_obj.end_time.strftime('%I:%M %p') if request_obj.end_time else 'TBD',
points=request_obj.points or 'No specific points provided',
approved=False,
refusal_note=rejection_reason or None,
)
flash(_('One on One request from %(player)s has been rejected.', player=player.username if player else 'Unknown'), 'info')
return redirect(url_for('users.notes_dashboard'))
# ---------------------------------------------------------------------------
# My Notes (Player)
# ---------------------------------------------------------------------------
@users_bp.route('/my-notes')
@login_required
def my_notes():
"""View personal and team notes for the current player."""
if not isinstance(current_user, Player):
flash(_('This page is for players only.'), 'info')
return redirect(url_for('main.dashboard'))
org_teams = current_user.get_org_teams()
org_team = org_teams[0] if org_teams else None
personal_notes = PersonalNote.query.filter_by(
player_id=current_user.id,
).order_by(PersonalNote.created_at.desc()).all()
team_notes = []
if org_team:
team_notes = TeamNote.query.filter_by(
org_team_id=org_team.id,
).order_by(TeamNote.created_at.desc()).all()
return render_template('pages/player_personal_notes.html',
org_team=org_team,
personal_notes=personal_notes,
team_notes=team_notes)
# ---------------------------------------------------------------------------
# Coach Availability
# ---------------------------------------------------------------------------
@users_bp.route('/coach-availability', methods=['GET', 'POST'])
@login_required
def manage_coach_availability():
"""Manage coach availability for One on One sessions."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can manage availability.'), 'danger')
return redirect(url_for('main.dashboard'))
if request.method == 'POST':
data = request.get_json()
slots = data.get('slots', []) if data else []
# Clear existing availability
CoachAvailability.query.filter_by(coach_id=current_user.id).delete()
# Add new slots
for slot in slots:
day_of_week = slot.get('day_of_week')
start_time_str = slot.get('start_time')
if day_of_week is None or day_of_week < 0 or day_of_week > 6:
continue
try:
start_time = datetime.strptime(start_time_str, '%H:%M').time()
end_time = (datetime.combine(datetime.today(), start_time) + timedelta(minutes=30)).time()
except (ValueError, TypeError):
continue
availability = CoachAvailability(
coach_id=current_user.id,
day_of_week=day_of_week,
start_time=start_time,
end_time=end_time,
)
db.session.add(availability)
db.session.commit()
return jsonify({'success': True})
existing_availability = CoachAvailability.query.filter_by(
coach_id=current_user.id,
).all()
return render_template('pages/coach_availability.html',
existing_availability=existing_availability)
@users_bp.route('/coach-availability/clear', methods=['POST'])
@login_required
def clear_coach_availability():
"""Clear all coach availability slots."""
if not isinstance(current_user, Coach):
return jsonify({'error': 'Unauthorized'}), 403
CoachAvailability.query.filter_by(coach_id=current_user.id).delete()
db.session.commit()
return jsonify({'success': True})
# ---------------------------------------------------------------------------
# Notes Dashboard (Coach)
# ---------------------------------------------------------------------------
@users_bp.route('/notes-dashboard')
@login_required
def notes_dashboard():
"""Notes and One on One dashboard for coaches."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can access the notes dashboard.'), 'danger')
return redirect(url_for('main.dashboard'))
org_team = OrgTeam.query.filter_by(coach_id=current_user.id).first()
players = []
if org_team:
player_ids = [tp.player_id for tp in TeamPlayer.query.filter_by(org_team_id=org_team.id).all()]
players = User.query.filter(User.id.in_(player_ids)).all() if player_ids else []
team_notes = []
personal_notes = []
latest_team_note = None
if org_team:
team_notes = TeamNote.query.filter_by(
org_team_id=org_team.id,
).order_by(TeamNote.created_at.desc()).all()
latest_team_note = team_notes[0] if team_notes else None
personal_notes = PersonalNote.query.filter_by(
coach_id=current_user.id,
).order_by(PersonalNote.created_at.desc()).all()
# One on One requests from team players
one_on_one_requests = []
if org_team and players:
player_ids_list = [p.id for p in players]
one_on_one_requests = OneOnOneRequest.query.filter(
OneOnOneRequest.player_id.in_(player_ids_list)
).order_by(OneOnOneRequest.created_at.desc()).all()
# For context selectors in the form
matches = Match.query.filter(
db.or_(Match.created_by == current_user.id, Match.status == 'scheduled'),
).order_by(Match.date.desc()).limit(20).all()
tryouts = Tryout.query.filter_by(
created_by=current_user.id,
).order_by(Tryout.date.desc()).limit(20).all()
teams = OrgTeam.query.order_by(OrgTeam.name).all()
return render_template('pages/notes.html',
org_team=org_team,
players=players,
team_notes=team_notes,
latest_team_note=latest_team_note,
personal_notes=personal_notes,
one_on_one_requests=one_on_one_requests,
matches=matches,
tryouts=tryouts,
teams=teams)
# ---------------------------------------------------------------------------
# Manage Team Notes (POST)
# ---------------------------------------------------------------------------
@users_bp.route('/team-notes/manage', methods=['POST'])
@login_required
def manage_team_notes():
"""Create or update team notes for the coach's org team."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can manage team notes.'), 'danger')
return redirect(url_for('main.dashboard'))
org_team = OrgTeam.query.filter_by(coach_id=current_user.id).first()
if not org_team:
flash(_('You are not assigned to a team.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
content = request.form.get('content', '').strip()
if content:
note = TeamNote(
org_team_id=org_team.id,
coach_id=current_user.id,
content=content,
)
db.session.add(note)
db.session.commit()
flash(_('Team notes saved successfully!'), 'success')
return redirect(url_for('users.notes_dashboard'))
# ---------------------------------------------------------------------------
# Manage Personal Notes (POST, simple form)
# ---------------------------------------------------------------------------
@users_bp.route('/personal-notes/manage', methods=['POST'])
@login_required
def manage_personal_notes():
"""Create a personal note for a player (coach only, simple form)."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can manage personal notes.'), 'danger')
return redirect(url_for('main.dashboard'))
player_id = request.form.get('player_id', type=int)
content = request.form.get('content', '').strip()
if not player_id or not content:
flash(_('Player and content are required.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
player = User.query.get_or_404(player_id)
if not isinstance(player, Player):
flash(_('Can only add notes for players.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
if not coach_can_access_player(current_user, player_id):
flash(_('You can only write notes about players you work with.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
note = PersonalNote(
player_id=player_id,
coach_id=current_user.id,
content=content,
)
db.session.add(note)
db.session.commit()
flash(_('Note added for %(username)s.', username=player.username), 'success')
return redirect(url_for('users.notes_dashboard'))
# ---------------------------------------------------------------------------
# Add Personal Note (POST, full form with context)
# ---------------------------------------------------------------------------
@users_bp.route('/personal-notes/add', methods=['POST'])
@login_required
def add_personal_note():
"""Create a personal note for a player with optional context (coach only)."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can add personal notes.'), 'danger')
return redirect(url_for('main.dashboard'))
player_id = request.form.get('player_id', type=int)
content = request.form.get('content', '').strip()
match_id = request.form.get('match_id', type=int)
tryout_id = request.form.get('tryout_id', type=int)
team_id_str = request.form.get('team_id')
if not player_id or not content:
flash(_('Player and content are required.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
player = User.query.get_or_404(player_id)
if not isinstance(player, Player):
flash(_('Can only add notes for players.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
if not coach_can_access_player(current_user, player_id):
flash(_('You can only write notes about players you work with.'), 'danger')
return redirect(url_for('users.notes_dashboard'))
note = PersonalNote(
player_id=player_id,
coach_id=current_user.id,
content=content,
match_id=match_id if match_id else None,
tryout_id=tryout_id if tryout_id else None,
team_id=int(team_id_str) if team_id_str and team_id_str.isdigit() else None,
)
db.session.add(note)
db.session.commit()
flash(_('Note added for %(username)s.', username=player.username), 'success')
return redirect(url_for('users.notes_dashboard'))
# ---------------------------------------------------------------------------
# Add Note from Tryout context (GET + POST)
# ---------------------------------------------------------------------------
@users_bp.route('/personal-notes/tryout/<int:tryout_id>', methods=['GET', 'POST'])
@login_required
def add_note_from_tryout(tryout_id):
"""Add a personal note for a player in the context of a tryout."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can add personal notes.'), 'danger')
return redirect(url_for('main.dashboard'))
tryout = Tryout.query.get_or_404(tryout_id)
preselected_player_id = request.args.get('player_id', type=int)
# Get registrations as players for the select list
registrations = TryoutRegistration.query.filter_by(tryout_id=tryout_id).all()
players = [r.player for r in registrations if r.player]
if request.method == 'POST':
player_id = request.form.get('player_id', type=int)
content = request.form.get('content', '').strip()
if not player_id or not content:
flash(_('Player and content are required.'), 'danger')
return redirect(url_for('users.add_note_from_tryout', tryout_id=tryout_id))
if not coach_can_access_player(current_user, player_id):
flash(_('You can only write notes about players you work with.'), 'danger')
return redirect(url_for('users.add_note_from_tryout', tryout_id=tryout_id))
note = PersonalNote(
player_id=player_id,
coach_id=current_user.id,
content=content,
tryout_id=tryout_id,
)
db.session.add(note)
db.session.commit()
flash(_('Note added successfully.'), 'success')
return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
return render_template('pages/add_note.html',
context_type='tryout',
tryout=tryout,
players=players,
preselected_player_id=preselected_player_id,
team_notes=[])
# ---------------------------------------------------------------------------
# Add Note from Match context (GET + POST)
# ---------------------------------------------------------------------------
@users_bp.route('/personal-notes/match/<int:match_id>', methods=['GET', 'POST'])
@login_required
def add_note_from_match(match_id):
"""Add a personal note for a player in the context of a match."""
if not isinstance(current_user, Coach):
flash(_('Only coaches can add personal notes.'), 'danger')
return redirect(url_for('main.dashboard'))
match_obj = Match.query.get_or_404(match_id)
# Get participants as players for the select list
participants = MatchParticipant.query.filter_by(match_id=match_id).all()
players = [p.player for p in participants if p.player]
preselected_player_id = request.args.get('player_id', type=int)
if request.method == 'POST':
player_id = request.form.get('player_id', type=int)
content = request.form.get('content', '').strip()
if not player_id or not content:
flash(_('Player and content are required.'), 'danger')
return redirect(url_for('users.add_note_from_match', match_id=match_id))
if not coach_can_access_player(current_user, player_id):
flash(_('You can only write notes about players you work with.'), 'danger')
return redirect(url_for('users.add_note_from_match', match_id=match_id))
note = PersonalNote(
player_id=player_id,
coach_id=current_user.id,
content=content,
match_id=match_id,
)
db.session.add(note)
db.session.commit()
flash(_('Note added successfully.'), 'success')
return redirect(url_for('tryouts.view_tryout', tryout_id=match_obj.tryout_id))
return render_template('pages/add_note.html',
context_type='match',
tryout=match_obj,
match=match_obj,
players=players,
preselected_player_id=preselected_player_id,
team_notes=[])
Reference in New Issue View Git Blame Copy Permalink