Files
TipPy/tipp/views.py
T
2026-09-19 09:04:26 +02:00

799 lines
25 KiB
Python

# -*- coding: utf-8 -*-
"""
Copyright (c) 2015-2016 by Martin Bley (dev@mb-oss.de)
This file is part of TipPy.
TipPy is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
TipPy is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with TipPy. If not, see <http://www.gnu.org/licenses/>.
"""
import os
from os.path import join as pjoin
import logging
import pytz
from PIL import Image as PImage
from django.conf import settings
from django.contrib.admin.views.decorators import staff_member_required
from django.contrib.auth.decorators import login_required
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.core.files.storage import default_storage
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
from django.db.models import Sum
from django.http import HttpResponseRedirect
from django.shortcuts import render, redirect, get_object_or_404
from django.utils import timezone
from django.views.decorators.http import require_http_methods
from tipp.forms import *
from tipp.models import *
from tipp.openliga import *
TIMEZONE_LOCAL = pytz.timezone('Europe/Berlin')
THUMBNAIL_SIZE = (50, 50)
logger = logging.getLogger(__name__)
def about(request):
"""Render about page."""
return render(request, 'about.html')
@login_required
def home(request):
"""Redirect to current matchday."""
ls = get_current_ls()
season = get_current_season()
md = get_current_md(ls)
return redirect(
'matchday',
ls=ls,
season=season,
matchday=str(md).zfill(2)
)
def _create_thumbnail(image_path):
"""Create a 50x50 thumbnail from the original image."""
try:
im = PImage.open(image_path)
im.thumbnail(THUMBNAIL_SIZE, PImage.Resampling.LANCZOS)
im.save(image_path)
except Exception as e:
logger.error(f"Failed to create thumbnail: {str(e)}")
raise
def _delete_avatar(image_path):
"""Delete avatar image file safely."""
if os.path.isfile(image_path):
try:
os.remove(image_path)
except OSError as e:
logger.error(f"Failed to delete avatar: {str(e)}")
@login_required
@require_http_methods(["GET", "POST"])
def profile(request, pk):
"""Display and edit user profile."""
user = get_object_or_404(User, id=pk)
profile, created = UserProfile.objects.get_or_create(user=user)
if request.method == 'POST':
return _handle_profile_post(request, user, profile)
else:
return _handle_profile_get(request, user, profile)
def _handle_profile_post(request, user, profile):
"""Handle POST request for profile update."""
user_form = UserForm(data=request.POST, instance=user)
profile_form = UserProfileForm(data=request.POST, instance=profile)
if user_form.is_valid() and profile_form.is_valid():
user = user_form.save()
profile = profile_form.save(commit=False)
profile.user = user
# Handle avatar upload
if 'avatar' in request.FILES:
profile.avatar = request.FILES['avatar']
profile.save()
image_path = os.path.join(settings.MEDIA_ROOT, profile.avatar.name)
try:
_create_thumbnail(image_path)
except Exception as e:
logger.error(f"Thumbnail creation failed: {str(e)}")
# Handle avatar deletion
elif 'avatar-clear' in request.POST and request.POST['avatar-clear'] == 'on':
if profile.avatar:
old_image_path = os.path.join(settings.MEDIA_ROOT, profile.avatar.name)
_delete_avatar(old_image_path)
profile.avatar = None
profile.save()
return redirect('profile', pk=user.id)
else:
logger.warning(f"Form validation failed for user {user.id}: "
f"User errors: {user_form.errors}, "
f"Profile errors: {profile_form.errors}")
return _render_profile_form(request, user_form, profile_form, profile)
def _handle_profile_get(request, user, profile):
"""Handle GET request for profile display."""
user_form = UserForm(instance=user)
profile_form = UserProfileForm(instance=profile)
return _render_profile_form(request, user_form, profile_form, profile)
def _render_profile_form(request, user_form, profile_form, profile):
"""Render profile form template."""
context = {
'user_form': user_form,
'profile_form': profile_form,
'avatar': profile.avatar.name if profile.avatar else None,
}
return render(request, 'registration/profile.html', context)
@staff_member_required
def staff(request):
"""Display available competitions for current season."""
season = get_current_season()
competitions = Competition.objects.filter(
season=season
).order_by('leagueName')
return render(request, 'staff.html', {'competitions': competitions})
@staff_member_required
def getSeason(request, ls, season):
"""Fetch and import teams and matches for a season from OpenLiga."""
ol = OpenLiga()
teams_data = ol.getTeams(str(season), ls)
season_data = ol.getSeason(str(season), ls)
# Bulk import teams
_import_teams(teams_data[0])
# Bulk import matches
_import_season_matches(season_data[0], ls)
return redirect(
'matchday',
ls=ls,
season=get_current_season(),
matchday=str(get_current_md(ls)).zfill(2)
)
def _import_teams(teams_data):
"""Import teams, update existing ones with icon URL if missing."""
teams_to_create = []
teams_to_update = []
for team_data in teams_data:
team_id = int(team_data['teamID'])
try:
existing_team = Team.objects.get(teamID=team_id)
if existing_team.iconURL is None:
existing_team.iconURL = team_data['teamIconURL']
teams_to_update.append(existing_team)
except Team.DoesNotExist:
teams_to_create.append(
Team(
teamID=team_id,
name=team_data['teamName'],
iconURL=team_data['teamIconURL']
)
)
if teams_to_create:
Team.objects.bulk_create(teams_to_create, ignore_conflicts=True)
if teams_to_update:
Team.objects.bulk_update(teams_to_update, ['iconURL'], batch_size=100)
def _import_season_matches(season_data, ls):
"""Import all matches from season data."""
matches = []
for match_data in season_data:
try:
team1 = Team.objects.get(teamID=int(match_data['idTeam1']))
team2 = Team.objects.get(teamID=int(match_data['idTeam2']))
except Team.DoesNotExist:
logger = logging.getLogger(__name__)
logger.warning(f"Team not found for match {match_data['matchID']}")
continue
matches.append(
Match(
matchID=match_data['matchID'],
matchDateTime=match_data['matchDateTime'],
group=int(match_data['groupID']),
matchday=int(match_data['groupOrderID']),
matchday_name=match_data['groupName'],
idTeam1=team1,
idTeam2=team2,
pointsTeam1=match_data['pointsTeam1'],
pointsTeam2=match_data['pointsTeam2'],
finished=bool(match_data['matchIsFinished']),
season=match_data['leagueSaison'],
leagueShortcut=match_data['leagueShortcut']
)
)
if matches:
Match.objects.bulk_create(matches, ignore_conflicts=True)
@login_required
def update(request, ls, season, cur_md):
"""Update match results and calculate user scores for current matchday."""
ol = OpenLiga()
md_matches = ol.getMatchday(str(season), int(cur_md), ls)
# Update match results
_update_match_results(md_matches[0])
# Calculate scores for all tips
_calculate_matchday_scores(request.user, ls, season, cur_md)
return redirect('matchday', ls=ls, season=season, matchday=cur_md)
def _update_match_results(matches_data):
"""Update match results from OpenLiga data."""
matches_to_update = []
for match_data in matches_data:
points_team1, points_team2 = _extract_match_result(match_data['matchResults'])
matches_to_update.append(
Match(
matchID=match_data['matchID'],
matchDateTime=match_data['matchDateTime'],
pointsTeam1=points_team1,
pointsTeam2=points_team2,
finished=bool(match_data['matchIsFinished']),
)
)
if matches_to_update:
Match.objects.bulk_update(
matches_to_update,
['matchDateTime', 'pointsTeam1', 'pointsTeam2', 'finished'],
batch_size=100
)
def _extract_match_result(match_results):
"""Extract final match result from match results data.
Returns:
tuple: (pointsTeam1, pointsTeam2), defaults to (-1, -1) if not found.
"""
if not match_results or not match_results[0]:
return -1, -1
try:
for result in match_results[0]:
result_name = result['resultName']
# Prefer final result, but accept extra time if available
if result_name in ("Endergebnis", "nach Verlängerung"):
return int(result['pointsTeam1']), int(result['pointsTeam2'])
# Fallback: half-time result if final not found
for result in match_results[0]:
if result['resultName'] == "Halbzeit":
return int(result['pointsTeam1']), int(result['pointsTeam2'])
except (KeyError, ValueError, TypeError) as e:
logger = logging.getLogger(__name__)
logger.warning(f"Error extracting match result: {str(e)}")
return -1, -1
def _calculate_matchday_scores(user, ls, season, cur_md):
"""Calculate scores for all user tips on finished matches."""
try:
mandant = RelUserMandant.objects.filter(user=user).values_list(
'mandant', flat=True
).first()
if not mandant:
logger = logging.getLogger(__name__)
logger.warning(f"No mandant found for user {user.id}")
return
scores = Score.objects.get(client=mandant)
except Score.DoesNotExist:
logger = logging.getLogger(__name__)
logger.warning(f"No score configuration found for mandant {mandant}")
return
finished_matches = Match.objects.filter(
finished=True,
leagueShortcut=ls,
season=season,
matchday=cur_md
).prefetch_related('tipp_set')
tipps_to_update = []
for match in finished_matches:
for tipp in match.tipp_set.all():
score = _calculate_tipp_score(tipp, match, scores)
tipp.score = score
tipps_to_update.append(tipp)
if tipps_to_update:
Tipp.objects.bulk_update(tipps_to_update, ['score'], batch_size=100)
def _calculate_tipp_score(tipp, match, scores):
"""Calculate score for a single tip based on match result.
Returns:
int: Points earned for this tip.
"""
# Exact result
if tipp.pointsTeam1 == match.pointsTeam1 and tipp.pointsTeam2 == match.pointsTeam2:
goals_total = match.pointsTeam1 + match.pointsTeam2
return scores.exact_high if goals_total >= 5 else scores.exact
# Winner and difference
winner_real = _get_winner(match.pointsTeam1, match.pointsTeam2)
winner_tipp = _get_winner(tipp.pointsTeam1, tipp.pointsTeam2)
if winner_tipp == winner_real:
diff_real = match.pointsTeam1 - match.pointsTeam2
diff_tipp = tipp.pointsTeam1 - tipp.pointsTeam2
if diff_tipp == diff_real:
return scores.diff
else:
return scores.tendency
return 0
def _get_winner(points_team1, points_team2):
"""Determine match winner.
Returns:
int: 1 for team1 win, 2 for team2 win, 0 for draw, -1 for invalid.
"""
if points_team1 > points_team2:
return 1
elif points_team2 > points_team1:
return 2
elif points_team1 == points_team2:
return 0
return -1
@login_required
@require_http_methods(["GET", "POST"])
def matchday(request, ls, season, matchday, template_name='md.html'):
"""Display matchday with user tips, match results, and mandant scores."""
# Process form submissions first
_process_matchday_tips(request, ls, season, matchday)
user = request.user
matches = _build_matchday_matches(request, user, ls, season, matchday)
mandant_dict = _build_mandant_scores(user, matches)
posts = _get_mandant_posts(user)
cur_group = matches[0]['md_name'] if matches else "N/A"
return render(request, template_name, {
'cur_md': matchday,
'md_name': cur_group,
'matches': matches,
'ls': ls,
'season': season,
'username': user.username,
'mandant_dict': mandant_dict,
'posts': posts,
'has_refresh': True
})
def _process_matchday_tips(request, ls, season, matchday):
"""Process and save tip form submissions."""
if request.method != 'POST':
return
md_matches = Match.objects.filter(
leagueShortcut=ls,
season=season,
matchday=int(matchday)
)
for match in md_matches:
form = TippForm(request.POST, prefix=str(match.matchID))
if form.is_valid():
data = form.cleaned_data
Tipp.objects.update_or_create(
tipperID=request.user,
matchID=match,
defaults={
'pointsTeam1': data['tippTeam1'],
'pointsTeam2': data['tippTeam2']
}
)
def _build_matchday_matches(request, user, ls, season, matchday):
"""Build match data with forms and results for template."""
md_matches = Match.objects.filter(
leagueShortcut=ls,
season=season,
matchday=int(matchday)
).select_related('idTeam1', 'idTeam2').order_by('matchDateTime')
# Fetch all user tips in one query
user_tipps = {
tipp.matchID_id: tipp
for tipp in Tipp.objects.filter(tipperID=user, matchID__in=md_matches)
}
matches = []
for match in md_matches:
match_started = timezone.now() >= match.matchDateTime
tipp = user_tipps.get(match.id)
# Build form with initial data
initial = {}
if tipp:
initial = {
'tippTeam1': tipp.pointsTeam1,
'tippTeam2': tipp.pointsTeam2
}
form = TippForm(prefix=str(match.matchID), initial=initial)
if match_started:
form.fields['tippTeam1'].widget.attrs['disabled'] = True
form.fields['tippTeam2'].widget.attrs['disabled'] = True
match_result = _format_match_result(match.pointsTeam1, match.pointsTeam2)
item = {
'matchDateTime': match.matchDateTime.astimezone(TIMEZONE_LOCAL).strftime(
'%a %d.%m, %H:%M'
),
'iconTeam1': str(match.idTeam1.icon),
'iconURLTeam1': str(match.idTeam1.iconURL),
'nameTeam1': str(match.idTeam1.name),
'abbrTeam1': str(match.idTeam1.abbr),
'iconTeam2': str(match.idTeam2.icon),
'iconURLTeam2': str(match.idTeam2.iconURL),
'nameTeam2': str(match.idTeam2.name),
'abbrTeam2': str(match.idTeam2.abbr),
'started': match_started,
'finished': match.finished,
'md_name': match.matchday_name,
'matchResult': match_result,
'tippFormTeam1': form['tippTeam1'],
'tippFormTeam2': form['tippTeam2'],
'match_id': match.id,
}
matches.append(item)
return matches
def _format_match_result(points_team1, points_team2):
"""Format match result for display."""
if points_team1 == -1 or points_team2 == -1:
return "-:-"
return f"{points_team1}:{points_team2}"
def _build_mandant_scores(user, matches):
"""Build mandant dict with user scores and tips."""
mandants = RelUserMandant.objects.filter(user=user).values_list(
'mandant', flat=True
)
if not mandants:
return {}
mandant_dict = {}
# Fetch all mandants and related users
mandant_users = {}
for mandant in mandants:
mandant_obj = get_object_or_404(Mandant, id=mandant)
mandant_users[mandant] = RelUserMandant.objects.filter(
mandant=mandant
).select_related('user')
# Get all tips for all matches and users
match_ids = [m['match_id'] for m in matches]
user_ids = []
for mandant in mandant_users:
user_ids.extend([rel.user_id for rel in mandant_users[mandant]])
all_tipps = {
(tipp.tipperID_id, tipp.matchID_id): tipp
for tipp in Tipp.objects.filter(
tipperID_id__in=user_ids,
matchID_id__in=match_ids
)
}
# Build mandant dict
for mandant in mandants:
mandant_obj = Mandant.objects.get(id=mandant)
mandant_dict[mandant_obj.description] = []
for rel in mandant_users[mandant]:
mate_data = {
'mate': rel.user,
'tipps': [],
'sum_score': 0
}
# Get tips for this mate across all matches
for match in matches:
tipp = all_tipps.get((rel.user_id, match['match_id']))
if tipp:
if match['started']:
score = tipp.score if tipp.score is not None else '\u231B'
tip_str = f"{tipp.pointsTeam1}:{tipp.pointsTeam2}({score})"
mate_data['tipps'].append(tip_str)
if match['finished']:
mate_data['sum_score'] += tipp.score or 0
else:
mate_data['tipps'].append('\u2714')
else:
mate_data['tipps'].append('\u2717')
mandant_dict[mandant_obj.description].append(mate_data)
# Sort by score
mandant_dict[mandant_obj.description] = sorted(
mandant_dict[mandant_obj.description],
key=lambda k: k['sum_score'],
reverse=True
)
return mandant_dict
def _get_mandant_posts(user):
"""Get latest published blog posts for user's mandants."""
mandants = RelUserMandant.objects.filter(user=user).values_list(
'mandant', flat=True
)
post_ids = RelPostMandant.objects.filter(
mandant__in=mandants
).values_list('post', flat=True)
posts = []
for post in Post.objects.filter(
id__in=post_ids,
published=True
).select_related('author__userprofile')[:5]:
avatar = None
try:
avatar = post.author.userprofile.avatar.name
except (ObjectDoesNotExist, AttributeError):
pass
posts.append((post, avatar))
return posts
@login_required
def charts(request, ls, season, pos='default',
template_name='charts.html'):
debug = ''
if pos == 'default':
if get_current_md(ls) < 18:
pos = 'hin'
matches = Match.objects.filter(leagueShortcut=ls,
season=season, matchday__lte=17)
else:
pos = 'rueck'
matches = Match.objects.filter(leagueShortcut=ls,
season=season, matchday__gte=18)
elif pos == 'hin':
matches = Match.objects.filter(leagueShortcut=ls,
season=season, matchday__lte=17)
elif pos == 'rueck':
matches = Match.objects.filter(leagueShortcut=ls,
season=season, matchday__gte=18)
else:
matches = Match.objects.filter(leagueShortcut=ls, season=season)
# get mandants of current user
mandants = RelUserMandant.objects.filter(user=request.user). \
values_list('mandant', flat=True)
# main loop
mandant_dict = {}
for m in mandants:
mandant = Mandant.objects.get(id=m)
# get users
users = RelUserMandant.objects.filter(mandant=m). \
values_list('user', flat=True)
user_dict = {}
for userid in users:
# get user object, then fetch matches for user
user = User.objects.get(id=userid)
score = Tipp.objects.filter(matchID__in=matches,
tipperID=user).aggregate(Sum('score'))
if score['score__sum'] is None:
user_dict[user.first_name] = 0
else:
user_dict[user.first_name] = score['score__sum']
mandant_dict[mandant.name] = sorted(user_dict.items(),
key=lambda x: x[1], reverse=True)
return django.shortcuts.render(request, template_name, {
# 'debug': debug,
'mandant_dict': mandant_dict,
'season': season,
'ls': ls,
'pos': pos
})
@login_required
def blogindex(request, page):
# get the blog posts that are published
# and related to mandant
mandants = RelUserMandant.objects.filter(user=request.user). \
values_list('mandant', flat=True)
posts = RelPostMandant.objects.filter(mandant__in=mandants). \
values_list('post', flat=True)
post_list = []
for post in Post.objects.filter(published=True, id__in=posts):
try:
avatar = UserProfile.objects.get(user_id=post.author_id). \
avatar.name
except:
avatar = None
post_list.append((post, avatar))
p = Paginator(post_list, 5)
try:
posts = p.page(page)
except PageNotAnInteger:
posts = p.page(1)
except EmptyPage:
posts = p.page(p.num_pages)
return django.shortcuts.render(request, 'blogindex.html', {
'posts': posts,
'ls': get_current_ls(),
'season': get_current_season(),
'page': page
})
@login_required
def newBlogpost(request):
debug = ''
has_mandant = False
mandants = RelUserMandant.objects.filter(user=request.user). \
values_list('mandant', flat=True)
choices = []
if mandants.count() > 1:
has_mandant = True
for m in Mandant.objects.filter(id__in=mandants):
choices.append((m.name, m.description))
if request.method == 'POST':
form_bp = BlogpostForm(request.POST)
if has_mandant is True:
form_mandant = MandantsForm(request.POST)
form_mandant.fields["mandants"].choices = choices
if form_bp.is_valid():
data = form_bp.cleaned_data
p = Post(author_id=request.user.id, content=data['content'])
p.save()
if has_mandant is True and form_mandant.is_valid():
data = form_mandant.cleaned_data
for m in data['mandants']:
debug += str(m) + " "
m = RelPostMandant(post=p,
mandant=Mandant.objects.get(name=m))
m.save()
else:
m = RelPostMandant(post=p,
mandant=Mandant.objects.get(id=mandants[0]))
m.save()
return django.shortcuts.redirect('/blog/1')
blogpostForm = BlogpostForm()
form_mandant = MandantsForm()
form_mandant.fields["mandants"].choices = choices
cnt_mandants = len(choices)
return django.shortcuts.render(request, 'newblogpost.html', {
'debug': debug,
'form_bp': blogpostForm,
'form_mandants': form_mandant,
'cnt_mandants': cnt_mandants,
'ls': get_current_ls(),
'season': get_current_season()
})
def logout(request):
"""
Log users out and re-direct them to the main page.
"""
logout(request)
return HttpResponseRedirect('/')
def get_current_md(ls):
"""
get current matchday()
returns dict with keys
['groupName']
['groupOrderID']
['groupID']
"""
try:
ol = OpenLiga()
cur_md = ol.getCurrentGroup(ls)
except:
cur_md = {'groupName': u"1. Spieltag", 'groupOrderID': 1}
return (cur_md['groupOrderID'])
def get_current_ls():
return ("bl1")
def get_current_season():
return ("2022")