Skip to content
Snippets Groups Projects
Commit 4846979c authored by Jiří Kalvoda's avatar Jiří Kalvoda
Browse files

Přesun vytváření účastníků, registrací do ročníku a do soutěže z mo.imports do mo.users

Tyto funkce budu potřeba požívat i jinak než při importu
parent 6df48776
No related branches found
No related tags found
1 merge request!90Formulářová políčka; Přidání účastníka do soutěže; Editace registrace do ročníku
This commit is part of merge request !90. Comments created here will be created in the context of that merge request.
......@@ -16,6 +16,8 @@ import mo.util
from mo.util import logger
from mo.util_format import format_decimal
reason = "import"
class ImportType(db.MOEnum):
participants = auto()
......@@ -188,26 +190,11 @@ class Import:
return r
def find_or_create_user(self, email: str, krestni: str, prijmeni: str, is_org: bool) -> Optional[db.User]:
sess = db.get_session()
user = sess.query(db.User).filter_by(email=email).one_or_none()
if user:
if user.first_name != krestni or user.last_name != prijmeni:
return self.error(f'Osoba již registrována s odlišným jménem {user.full_name()}')
if (user.is_admin or user.is_org) != is_org:
if is_org:
return self.error('Nelze předefinovat účastníka na organizátora')
else:
return self.error('Nelze předefinovat organizátora na účastníka')
else:
user = db.User(email=email, first_name=krestni, last_name=prijmeni, is_org=is_org)
sess.add(user)
sess.flush() # Aby uživatel dostal user_id
logger.info(f'Import: Založen uživatel user=#{user.user_id} email=<{user.email}>')
mo.util.log(
type=db.LogType.user,
what=user.user_id,
details={'action': 'import', 'new': db.row2dict(user)},
)
try:
user, is_new = mo.users.find_or_create_user(email, krestni, prijmeni, is_org, reason=reason)
except mo.CheckError as e:
return self.error(str(e))
if is_new:
self.cnt_new_users += 1
self.new_user_ids.append(user.user_id)
return user
......@@ -227,53 +214,21 @@ class Import:
return pts
def find_or_create_participant(self, user: db.User, year: int, school_id: int, birth_year: int, grade: str) -> Optional[db.Participant]:
sess = db.get_session()
part = sess.query(db.Participant).get((user.user_id, year))
if part:
if (part.school != school_id
or part.grade != grade
or part.birth_year != birth_year):
return self.error('Účastník již zaregistrován s odlišnou školou/ročníkem/rokem narození')
else:
part = db.Participant(user=user, year=year, school=school_id, birth_year=birth_year, grade=grade)
sess.add(part)
logger.info(f'Import: Založen účastník #{user.user_id}')
mo.util.log(
type=db.LogType.participant,
what=user.user_id,
details={'action': 'import', 'new': db.row2dict(part)},
)
try:
part, is_new = mo.users.find_or_create_participant(user, year, school_id, birth_year, grade, reason=reason)
except mo.CheckError as e:
return self.error(str(e))
if is_new:
self.cnt_new_participants += 1
return part
def find_or_create_participation(self, user: db.User, contest: db.Contest, place: Optional[db.Place]) -> Optional[db.Participation]:
if place is None:
place = contest.place
sess = db.get_session()
pions = (sess.query(db.Participation)
.filter_by(user=user)
.filter(db.Participation.contest.has(db.Contest.round == contest.round))
.all())
if not pions:
pion = db.Participation(user=user, contest=contest, place_id=place.place_id, state=db.PartState.invited)
sess.add(pion)
logger.info(f'Import: Založena účast user=#{user.user_id} contest=#{contest.contest_id} place=#{place.place_id}')
mo.util.log(
type=db.LogType.participant,
what=user.user_id,
details={'action': 'add-to-contest', 'new': db.row2dict(pion)},
)
try:
pion, is_new = mo.users.find_or_create_participation(user, contest, place, reason=reason)
except mo.CheckError as e:
return self.error(str(e))
if is_new:
self.cnt_new_participations += 1
elif len(pions) == 1:
pion = pions[0]
if pion.place != place:
return self.error(f'Již se tohoto kola účastní v {contest.round.get_level().name_locative("jiném", "jiné", "jiném")} ({pion.place.get_code()})')
else:
return self.error('Již se tohoto kola účastní ve vice oblastech, což by nemělo být možné')
return pion
def obtain_contest(self, oblast: Optional[db.Place], allow_none: bool = False):
......
......@@ -5,11 +5,12 @@ import datetime
import email.errors
import email.headerregistry
import re
from typing import Optional
from typing import Optional, Tuple
import mo
import mo.db as db
import mo.util
from mo.util import logger
import mo.tokens
......@@ -48,6 +49,82 @@ def validate_and_find_school(kod: str) -> db.Place:
return place
def find_or_create_user(email: str, krestni: str, prijmeni: str, is_org: bool, reason: str = "undef-reason") -> Tuple[db.User, bool]:
sess = db.get_session()
user = sess.query(db.User).filter_by(email=email).one_or_none()
is_new = user is None
if user is None: # HACK: Podmínku je nutné zapsat znovu místo užití is_new, jinak si s tím mypy neporadí
user = db.User(email=email, first_name=krestni, last_name=prijmeni, is_org=is_org)
sess.add(user)
sess.flush() # Aby uživatel dostal user_id
logger.info(f'{reason.title()}: Založen uživatel user=#{user.user_id} email=<{user.email}>')
mo.util.log(
type=db.LogType.user,
what=user.user_id,
details={'action': 'create-user', 'reason': reason, 'new': db.row2dict(user)},
)
else:
if user.first_name != krestni or user.last_name != prijmeni:
raise mo.CheckError(f'Osoba již registrována s odlišným jménem {user.full_name()}')
if (user.is_admin or user.is_org) != is_org:
if is_org:
raise mo.CheckError('Nelze předefinovat účastníka na organizátora')
else:
raise mo.CheckError('Nelze předefinovat organizátora na účastníka')
return user, is_new
def find_or_create_participant(user: db.User, year: int, school_id: int, birth_year: int, grade: str, reason: str = "undef-reason") -> Tuple[db.Participant, bool]:
sess = db.get_session()
part = sess.query(db.Participant).get((user.user_id, year))
is_new = part is None
if part is None:
part = db.Participant(user=user, year=year, school=school_id, birth_year=birth_year, grade=grade)
sess.add(part)
logger.info(f'{reason.title()}: Založen účastník #{user.user_id}')
mo.util.log(
type=db.LogType.participant,
what=user.user_id,
details={'action': 'create-participant', 'reason': reason, 'new': db.row2dict(part)},
)
else:
if (part.school != school_id
or part.grade != grade
or part.birth_year != birth_year):
raise mo.CheckError('Účastník již zaregistrován s odlišnou školou/ročníkem/rokem narození')
return part, is_new
def find_or_create_participation(user: db.User, contest: db.Contest, place: Optional[db.Place], reason: str = "undef-reason") -> Tuple[db.Participation, bool]:
if place is None:
place = contest.place
sess = db.get_session()
pions = (sess.query(db.Participation)
.filter_by(user=user)
.filter(db.Participation.contest.has(db.Contest.round == contest.round))
.all())
is_new = pions == []
if is_new:
pion = db.Participation(user=user, contest=contest, place_id=place.place_id, state=db.PartState.invited)
sess.add(pion)
logger.info(f'{reason.title()}: Založena účast user=#{user.user_id} contest=#{contest.contest_id} place=#{place.place_id}')
mo.util.log(
type=db.LogType.participant,
what=user.user_id,
details={'action': 'add-to-contest', 'reason': reason, 'new': db.row2dict(pion)},
)
elif len(pions) == 1:
pion = pions[0]
if pion.place != place:
raise mo.CheckError(f'Již se tohoto kola účastní v {contest.round.get_level().name_locative("jiném", "jiné", "jiném")} ({pion.place.get_code()})')
else:
raise mo.CheckError('Již se tohoto kola účastní ve více oblastech, což by nemělo být možné')
return pion, is_new
def normalize_email(addr: str) -> str:
if not re.fullmatch(r'.+@.+', addr):
raise mo.CheckError('V e-mailové adrese chybí zavináč')
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment