123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207 |
- from fastapi_login import LoginManager
- from fastapi import APIRouter, Form, Depends, HTTPException, File, UploadFile
- from typing import List,Optional,Union
- from fastapi.responses import FileResponse
- from random import randint
- from fastapi.security import OAuth2PasswordRequestForm
- from app.models.models import Registration,User,User_information,Class_list
- from app.api import deps
- from sqlalchemy.orm import Session
- from typing import Any, Dict
- import secrets
- from fastapi_login.exceptions import InvalidCredentialsException
- from fastapi_login import LoginManager
- from datetime import timedelta,datetime
- from jose import jwt
- from emails.template import JinjaTemplate
- from tortoise.queryset import Q
- from fastapi.responses import HTMLResponse
- from app.api.users import manager
- registration = APIRouter()
- async def check_token(access_token: str):
- result = await User.filter(token=access_token).first()
-
- if not result:
- print("no access")
- return None
-
- user_id = result.id
- return user_id
- async def check_if_id_exeit(
- table_name : str,
- id : int
- ):
- exec('check_list = await {}.filter(id=id).all()'.format(table_name))
- if check_list == []:
- return True
- else:
- False
- async def check_permissions(user_id):
- user = await User.get(id=user_id)
- if user.is_superuser:
- return True
- else:
- return False
- # @registration.get("/protected")
- # def protected_route(user_id=Depends(check_token)):
- # if not user_id:
- # return {"message": "no access"}
- # return {'user': user_id}
- @registration.get("/get_registration")
- async def get_registration(
- user_id = Depends(check_token),
- get_all : Optional[int] = None,
- event_id : Optional[int] = None,
- registration_id : Optional[int] = None,
- is_check : Optional[int] = None
- ):
- try :
- if not user_id :
- return {"msg": "please log in", "code": 200}
-
- if get_all:
- inform_list = await User_information.all()
- else:
- inform_list = await User_information.filter(user_id=user_id).all()
- reg_list = []
- for infor in inform_list:
- if registration_id:
- reg_list_tmp = await Registration.filter(id=registration_id,user_inform_id=infor.id,is_del=0).all()
- else:
- reg_list_tmp = await Registration.filter(user_inform_id=infor.id,is_del=0).all()
- for reg_obj in reg_list_tmp:
- reg_data = {
- "Registration_id" : reg_obj.id,
- "event_id" : reg_obj.event_id,
- "user_inform_id" : reg_obj.user_inform_id,
- "reg_confirm" : reg_obj.reg_confirm,
- "create_time" : reg_obj.create_time
- }
- if event_id and reg_data["event_id"] == event_id :
- if is_check != None :
- if reg_obj.reg_confirm == is_check :
- reg_list.append(reg_data)
- else :
- reg_list.append(reg_data)
- elif not event_id:
- if is_check != None :
- if reg_obj.reg_confirm == is_check :
- reg_list.append(reg_data)
- else :
- reg_list.append(reg_data)
-
- return {"msg": "success", "code": 200,"registrations":reg_list}
- except Exception as e:
- return {"msg": str(e), "code": 500}
- @registration.post("/input_information")
- async def input_information(
- user_id = Depends(check_token),
- name : str = Form(default=''),
- gender : str = Form(default=''),
- phone : str = Form(default=''),
- email : str = Form(default=''),
- is_default : int = Form(default=0)
- ):
- try :
- if not user_id :
- return {"msg": "no access", "code": 200}
-
- if is_default:
- inform = await User_information.get_or_create(
- user_id=user_id,
- is_default = is_default,
- defaults={
- 'name': name,
- 'gender': gender,
- 'phone': phone,
- 'email': email
- }
- )
- new_inform = inform[0]
- else :
- new_inform = await User_information.create(
- user_id = user_id,
- name = name ,
- gender = gender,
- phone = phone,
- email = email,
- is_default = is_default
- )
- return {"msg": "success", "code": 200, "user_inform_id": new_inform.id}
- except Exception as e:
- return {"msg": str(e), "code": 500}
- @registration.post("/input_registration")
- async def input_registration(
- user_id = Depends(check_token),
- user_inform_id : int= Form(default=0),
- event_id : int = Form(default=0)
- ):
- try :
- if not user_id :
- return {"msg": "please log in", "code": 200}
- # if check_if_id_exeit(User_information,user_inform_id):
- # return {"msg": "no user information", "code": 200}
-
- # if check_if_id_exeit(Class_list,event_id):
- # return {"msg": "no class list", "code": 200}
-
- new_registration = await Registration.create(
- event_id = event_id,
- user_inform_id = user_inform_id,
- reg_confirm = 0,
- is_del = 0 ,
- create_time = datetime.now()
- )
- return {"msg": "success", "code": 200,"new_registration_id":new_registration.id}
- except Exception as e:
- return {"msg": str(e), "code": 500}
-
- @registration.post("/confirm_reg")
- async def confirm_reg(
- #user_id = Depends(check_token),
- registration_id : int
- ):
- try:
- registration_obj = await Registration.get(id=registration_id)
- registration_obj.reg_confirm = 1
- await registration_obj.save()
- return {"msg": "success", "code": 200,"registration_id":registration_obj.id}
- except Exception as e:
- return {"msg": str(e), "code": 500}
- @registration.post("/delete_registration")
- async def delete_registration(
- registration_id : int
- ):
- try:
- registration_obj = await Registration.get(id=registration_id)
- registration_obj.is_del = 1
- await registration_obj.save()
- return {"msg": "success", "code": 200}
- except Exception as e:
- return {"msg": str(e), "code": 500}
|