import json 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,Class_name,Schools,Class_date,User_resume 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() IMAGEDIR = "/var/www/ntcri/assets/resume_pic/" IMAGEDIR_short = "assets/resume_pic/" @registration.post("/upload_user_resume_imgs") async def upload_user_resume_imgs(files:Optional[List[UploadFile]] = File(None)): files_url = [] if files : for file in files: contents = await file.read() #save the file with open(f"{IMAGEDIR}{file.filename}", "wb") as f: f.write(contents) files_url.append(f"{IMAGEDIR_short}{file.filename}" ) return files_url 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_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_class") async def get_registration_class( event_id : Optional[int] = None, check_user_id = Depends(check_token) ): #inform_list = await Registration.filter(event_id=event_id,is_del=0,reg_confirm=1).all() #use for 8/25 after check_list1 = await User.get(id = check_user_id).all() check_list2 = await Class_list.get(id = event_id).all() if 2!=check_list1.is_superuser: if check_list2.create_user_id !=check_user_id: return {"msg": "permissions denied", "code": 200,"registrations":[]} inform_list = await Registration.filter(event_id=event_id,is_del=0).all() class_data =[] for infor in inform_list: reg_data = { "event_id" : infor.event_id, "user_id" : infor.user_id, "is_attend" : infor.is_attend, "payment_status":infor.payment_status, "five_digits":infor.five_digits, "reg_confirm":infor.reg_confirm, } user = await User.get(id=infor.user_id) inform = await User_information.get(user_id=infor.user_id) reg_data["real_name"] = inform.name reg_data["phone"] = inform.phone reg_data["email"] = user.email class_data.append(reg_data ) return {"msg": "success", "code": 200,"registrations":class_data} @registration.post("/update_registration_class") async def update_registration_class( event_id : Optional[int] , user_id : Optional[int] , is_attend : int = Form(default=None), payment_status : int = Form(default=None), reg_confirm : int = Form(default=None), five_digits : str = Form(default=None), check_user_id = Depends(check_token), ): #inform_list = await Registration.filter(event_id=event_id,is_del=0,reg_confirm=1,user_id=user_id).all() #use for 8/25 after inform_list = await Registration.filter(event_id=event_id,is_del=0,user_id=user_id).all() check_list1 = await User.get(id = check_user_id).all() check_list2 = await Class_list.get(id = event_id).all() if 2!=check_list1.is_superuser: if check_list2.create_user_id !=check_user_id: return {"msg": "permissions denied", "code": 500} class_data =[] for infor in inform_list: if is_attend is not None: infor.is_attend = is_attend if payment_status is not None: infor.payment_status = payment_status if reg_confirm is not None: infor.reg_confirm = reg_confirm if five_digits is not None: if len(five_digits)==5: infor.five_digits = five_digits else: return {"msg": "five_digits must be five_digits", "code": 500} await infor.save() return {"msg": "success", "code": 200, "user_id": infor.user_id,"event_id": infor.event_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().order_by("-event_id","create_time") else: reg_list_tmp = await inform_list.filter(is_del=0).all().order_by("-event_id","create_time") 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" } try : class_obj = await Class_list.get(id = infor.event_id) class_name_obj = await Class_name.get(id = class_obj.name_id) school_obj = await Schools.get(id = class_name_obj.school_id) reg_data["school_name"] = school_obj.name reg_data["class_name"] = class_name_obj.name reg_data["class_event"] = class_obj.event reg_data["start_time"] = str(class_obj.start_time) reg_data["end_time"] =str(class_obj.end_time) except Exception as e: reg_data["class_data"] = str(e) try: user = await User.get(id=infor.user_id) inform = await User_information.get(user_id=infor.user_id) reg_data["real_name"] = inform.name reg_data["phone"] = inform.phone reg_data["email"] = user.email except Exception as e: reg_data["user_data"] = str(e) 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( name : str = Form(default=''), user_name : str = Form(default=''), birthday : date = Form(default=datetime.now().date()), gender : str = Form(default=''), phone : str = Form(default=''), address : str = Form(default=''), user_id = Depends(check_token), position: str = Form(default='[1]'), ): try : if not user_id : return {"msg": "no access", "code": 200} position_list = json.loads(position) d = {"學員":0,"開課工藝家":0,"其他":0} if 1 in position_list: d["學員"] = 1 if 2 in position_list: d["開課工藝家"] = 1 if 3 in position_list: d["其他"] = 1 print(d) infor,created = await User_information.get_or_create( user_id=user_id, defaults={ 'name': name, 'birthday' :birthday, 'gender': gender, 'phone': phone, 'address': address, 'position': d } ) if not created: 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 user_name != '': global user user = await User.get(id = user_id) user.username = user_name if position != '': infor.position = d await infor.save() await user.save() msg = "Update success" else : msg = "input success" return {"msg": "success", "code": 200, "user_inform_id": infor.id,"is_exist":not created} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/update_information") async def update_information( user_name : str = Form(default=''), name : str = Form(default=''), birthday : date = Form(default=datetime.now().date()), gender : str = Form(default=''), phone : str = Form(default=''), address : str = Form(default=''), user_id = Depends(check_token), position : str = Form(default='[1]'), ): 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) position_list = json.loads(position) d = {"學員":0,"開課工藝家":0,"其他":0} if 1 in position_list: d["學員"] = 1 if 2 in position_list: d["開課工藝家"] = 1 if 3 in position_list: d["其他"] = 1 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 user_name != '': user.username = user_name if position != '': infor.position = d 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), get_all : int = 0, get_detail_information : int = 1 ): try: if not user_id : return {"msg": "no access", "code": 200} try: if get_all == 1: user_list = await User.all() else: user_list = await User.filter(id = user_id).all() except: return {"msg": "user table run fail", "code": 500} user_inform_list = [] for user_obj in user_list: user_inform = { "user_id" : user_obj.id, "user_name" : user_obj.username, "email" : user_obj.email, "is_superuser" :user_obj.is_superuser } if get_detail_information: try : inform = await User_information.get(user_id=user_obj.id) user_inform["name"] = inform.name user_inform["birthday"] = inform.birthday user_inform["gender"] = inform.gender user_inform["phone"] = inform.phone user_inform["address"] = inform.address user_inform["msg"] = "user information exist" user_inform["exist"] = True user_inform["position"] = inform.position except: user_inform["msg"] = "no user information" user_inform["exist"] = False user_inform_list.append(user_inform) print(user_inform_list) return {"msg":"success","code":200,"user_inform": user_inform_list} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/update_superuser") async def update_superuser( change_id : int , is_superuser : int , user_id = Depends(check_token), ): if not user_id : return {"msg": "no exit", "code": 200} user = await User.get(id=user_id) if user.is_superuser != 2: return {"msg": "no access", "code": 200} user_list = await User.filter(id = change_id).all() for infor in user_list: if is_superuser is not None: infor.is_superuser = is_superuser await infor.save() return {"msg": "success", "code": 200} @registration.get("/change_class_reg_number") async def change_class_reg_number( event_id: int = 0, reduce_number : int = 1 ): try: if event_id: try: await Class_list.get(id = event_id) except Exception as e: return {"msg": "no this event", "code": 200} try: class_date = await Class_date.get(class_list_id = event_id) except Exception as e: return {"msg": "no this class' number limit", "code": 200} #if class_date.amount_left ==0 and reduce_number>0: if class_date.amount_left ==0: return {"msg": "class is full", "code": 200,"amount_left":-1} elif class_date.amount_left == class_date.number_limit and reduce_number<0: return {"msg": "class is empty", "code": 200,"amount_left":class_date.amount_left } else: class_date.amount_left = class_date.amount_left-reduce_number await class_date.save() return {"msg": "success", "code": 200,"amount_left":class_date.amount_left} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/input_registration") async def input_registration( event_id : int = Form(default=0), user_id = Depends(check_token) ): try : if not user_id : return {"msg": "please log in", "code": 500} try: await Class_list.get(id = event_id) except Exception as e: return {"msg": "no this event", "code": 500} # 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": 500} try: existing_registration = await Registration.get( event_id=event_id, user_id=user_id ) new_registration = existing_registration is_register = 0 except: existing_registration = None if existing_registration is None: is_register = 1 if is_register: amount_left_obj = await change_class_reg_number(event_id=event_id) msg = amount_left_obj["msg"] else: msg = "already registrate" if msg == 'success': new_registration = await Registration.create( event_id=event_id, user_id=user_id, reg_confirm=0, is_del=0, create_time=datetime.now(), is_attend = 0, payment_status = 0 ) return {"msg": msg, "code": 200,"new_registration_id":new_registration.id,"is_already_exist":not is_register} else: return {"msg": msg, "code": 200,"new_registration_id":None,"is_already_exist":not is_register} #new_registration = await Registration.get_or_create( # event_id = event_id, # user_id = user_id, # defaults = { # "reg_confirm" : 0, # "is_del" : 0 , # "create_time" : datetime.now(), # # } #) #print(new_registration) #if new_registration[1]: #return {"msg": msg, "code": 200,"new_registration_id":new_registration[0].id,"is_already_exist":not new_registration[1]} #return {"msg": msg, "code": 200,"new_registration_id":new_registration.id,"is_already_exist":not is_register} 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) amount_left_obj = await change_class_reg_number(event_id=registration_obj.event_id) msg = amount_left_obj["msg"] if msg == "class is full": return {"msg": msg+" cannot recover registration", "code": 200} registration_obj.is_del = 0 await registration_obj.save() return {"msg": msg, "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), super_ad_input_user_id : int = 0, event_id : int = 0 ): try: if not user_id : return {"msg": "please log in", "code": 200} if super_ad_input_user_id: registration_obj = await Registration.get(event_id=event_id,user_id=super_ad_input_user_id) else: registration_obj = await Registration.get(event_id=event_id,user_id=user_id) registration_obj.is_del = 1 amount_left_obj = await change_class_reg_number(event_id=registration_obj.event_id,reduce_number=-1) msg = amount_left_obj["msg"] await registration_obj.save() return {"msg": msg , "code": 200} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/input_user_resume") async def input_user_resume( user_id = Depends(check_token), imgs = Depends(upload_user_resume_imgs), teacher_name : str = Form(default=''), work_type : str = Form(default=''), experience : str = Form(default=''), expertise : str = Form(default=''), license : str = Form(default=''), media : str = Form(default=''), introduction: str = Form(default='') ): try: if not user_id : return {"msg": "please log in", "code": 200} msg = '' user_resume, created = await User_resume.get_or_create( user_id = user_id, defaults = { "teacher_name": teacher_name, "work_type": work_type, "experience": experience, "expertise": expertise, "license": license, "media": media, "imgs": imgs, "introduction": introduction } ) if not created: if teacher_name.strip() != '' : user_resume.teacher_name = teacher_name if work_type.strip() != '' : user_resume.work_type= work_type if experience.strip() != '' : user_resume.experience= experience if license.strip() != '' : user_resume.license= license if media.strip() != '' : user_resume.media= media if imgs != '[]' : user_resume.imgs= imgs if introduction.strip() != '' : user_resume.introduction= introduction await user_resume.save() msg = "Update success" else : msg = "input success" return {"msg": msg , "code": 200} except Exception as e: return {"msg": str(e), "code": 500} @registration.get("/get_user_resume") async def get_user_resume( user_id = Depends(check_token) ): try: if not user_id : return {"msg": "please log in", "code": 200} user_resume = await User_resume.get(user_id = user_id) data = user_resume.show_data() return {"msg": "success" , "code": 200,"user_resume":data} except Exception as e: return {"msg": str(e), "code": 500} @registration.post("/delete_user") async def get_user_resume( user_id : Optional[int] = None, check_user_id = Depends(check_token) ): try: #inform_list = await Registration.filter(event_id=event_id,is_del=0,reg_confirm=1).all() #use for 8/25 after check_list1 = await User.get(id = check_user_id).all() if 2!=check_list1.is_superuser: if check_list2.create_user_id !=check_user_id: return {"msg": "permissions denied", "code": 200,"registrations":[]} await User.filter(user_id = user_id).delete() return {"msg": "success" , "code": 200} except Exception as e: return {"msg": str(e), "code": 500}