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,date 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 = Registration.all() else: inform_list = Registration.filter(user_id=user_id).all() if event_id: inform_list = inform_list.filter(event_id=event_id) if is_check != None: inform_list = inform_list.filter(reg_confirm=is_check) if registration_id: reg_list_tmp = await inform_list.filter(id=registration_id,is_del=0).all() else: reg_list_tmp = await inform_list.filter(is_del=0).all() reg_list = [] for infor in reg_list_tmp: try : reg_data = { "Registration_id" : infor.id, "event_id" : infor.event_id, "user_id" : infor.user_id, "reg_confirm" : infor.reg_confirm, "create_time" : infor.create_time } except: reg_data = { "msg" : "fail to get data" } 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=''), display_name : str = Form(default=''), birthday : date = Form(default=datetime.now().date()), gender : str = Form(default=''), phone : str = Form(default=''), address : str = Form(default='') ): try : if not user_id : return {"msg": "no access", "code": 200} inform = await User_information.get_or_create( user_id=user_id, defaults={ 'name': name, 'birthday' :birthday, 'gender': gender, 'phone': phone, 'address': address } ) if display_name != '': user = await User.get(id = user_id) user.username = display_name await user.save() return {"msg": "success", "code": 200, "user_inform_id": inform[0].id,"is_exist":not inform[1]} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/update_information") async def update_information( user_id = Depends(check_token), name : str = Form(default=''), display_name : str = Form(default=''), birthday : date = Form(default=datetime.now().date()), gender : str = Form(default=''), phone : str = Form(default=''), address : str = Form(default='') ): try : if not user_id : return {"msg": "no access", "code": 200} infor = await User_information.get(user_id = user_id) user = await User.get(id = user_id) if name != '': infor.name = name if birthday != datetime.now().date(): infor.birthday = birthday if gender != '': infor.gender = gender if phone != '': infor.phone = phone if address != '': infor.address = address if display_name != '': user.username = display_name await infor.save() await user.save() return {"msg": "success", "code": 200, "user_inform_id": infor.id} except Exception as e: return {"msg": str(e), "code": 500} @registration.get("/get_user_information") async def get_user_information( user_id = Depends(check_token) ): try: if not user_id : return {"msg": "no access", "code": 200} try: infor = await User_information.get(user_id=user_id) except: return {"msg": "no user information", "code": 200} user_obj = await User.get(id = user_id) user_inform = { "user_id" : infor.user_id, "display_name" : user_obj.username, "name" : infor.name, "birthday" : infor.birthday, "gender" : infor.gender, "phone" : infor.phone, "address" : infor.address, "email" : user_obj.email } return {"msg":"success","code":200,"user_inform": user_inform} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/input_registration") async def input_registration( user_id = Depends(check_token), 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} try: await User_information.get(user_id=user_id) except: return {"msg": "no user information", "code": 200} new_registration = await Registration.create( event_id = event_id, user_id = user_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("/recover_registration") async def delete_registration( user_id = Depends(check_token), event_id : int = 0 ): try: if not user_id : return {"msg": "please log in", "code": 200} registration_obj = await Registration.get(event_id=event_id,user_id=user_id) registration_obj.is_del = 0 await registration_obj.save() return {"msg": "success", "code": 200} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/delete_registration") async def delete_registration( user_id = Depends(check_token), event_id : int = 0 ): try: if not user_id : return {"msg": "please log in", "code": 200} registration_obj = await Registration.get(event_id=event_id,user_id=user_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}