from collections import defaultdict from datetime import datetime from fastapi import FastAPI, HTTPException, Depends from sqlalchemy import and_,func, or_ from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker,Session from uuid import uuid4 # Import UUID for Python default from fastapi.middleware.cors import CORSMiddleware from models import MainTask, QuizQuestion, QuizUmrah, SessionUmrah, UmrahTask, SubTaskFemale,SubTaskMale,UserCreate,User,UserLogin,MiqatSubmission,UserResponse,Umrah from database import get_db from passlib.context import CryptContext import jwt from typing import List, Optional from fastapi.staticfiles import StaticFiles app = FastAPI() SECRET_KEY = "your-secret-key" ALGORITHM = "HS256" app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allow only Ionic app allow_credentials=True, allow_methods=["*"], # Allow all HTTP methods (GET, POST, etc.) allow_headers=["*"], # Allow all headers ) pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # register @app.post("/users/register") def register_user(user: UserCreate, db: Session = Depends(get_db)): existing_user = db.query(User).filter(User.email == user.email).first() if existing_user: raise HTTPException(status_code=400, detail="Email already registered") hashed_password = pwd_context.hash(user.password) new_user = User( id=str(uuid4()), name=user.name, email=user.email, password=hashed_password, date_birth=user.date_birth, gender=user.gender, country=user.country, phone_number=user.phone_number, ) db.add(new_user) db.commit() db.refresh(new_user) return {"message": "User registered successfully"} def create_access_token(data: dict): return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM) @app.post("/users/login", tags=["Authentication"]) def login_user(user: UserLogin, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.email == user.email).first() if not db_user: print(f"User with email {user.email} not found") elif not pwd_context.verify(user.password, db_user.password): print(f"Password incorrect for {user.email}") else: print(f"Login successful for {user.email}") if not db_user or not pwd_context.verify(user.password, db_user.password): raise HTTPException(status_code=401, detail="Invalid email or password") token = create_access_token({"sub": db_user.email, "user_id": db_user.id}) session=checkSession(db_user.id,db) if (session==None): return { "message": "Login successful", "token": token, "user_id": db_user.id, "gender":db_user.gender } else: return { "message": "Login successful", "token": token, "Session_ID":session, "user_id": db_user.id, "user_name": db_user.name, "email": db_user.email, "gender": db_user.gender, "country": db_user.country, "date_birth": db_user.date_birth, "phone_number": db_user.phone_number } @app.get("/check") def checkSession(user_id:str,db:Session=Depends(get_db)): session=db.query(SessionUmrah).filter(SessionUmrah.id_user==user_id,SessionUmrah.status=='incomplete').first() if session: umrah=db.query(Umrah).filter(Umrah.id_session==session.id,Umrah.status=='incomplete').first() if umrah: return {"message":umrah.id,"miqat":umrah.miqat,"date":umrah.umrah_start} else: return {"error":"No Session"} else: return {"error":"No Session"} #video convert def convert_to_embed_url(url: str) -> str: """Convert YouTube Shorts URL to embeddable format.""" if url and "youtube.com/shorts/" in url: return url.replace("youtube.com/shorts/", "youtube.com/embed/") return url @app.get("/list_task") async def list_task(user_id: str, session_id: str, db: Session = Depends(get_db)): try: # Fetch user user = db.query(User).filter(User.id == user_id).first() if not user: return {"message": "User not found"} gender = user.gender.lower() task_list = [] # Choose correct SubTask table SubTaskModel = SubTaskMale if gender == "male" else SubTaskFemale # Fetch main tasks main_tasks = db.query(MainTask).all() for main in main_tasks: # Special case: rename task-009 for female main_task_name = "BERGUNTING" if gender == "female" and main.id_main_task == "task-009" else main.name_task # Fetch subtasks subtasks = db.query(SubTaskModel)\ .filter(SubTaskModel.id_main_task == main.id_main_task)\ .order_by(SubTaskModel.no_task.asc()).all() subtask_list = [] for sub in subtasks: status = db.query(UmrahTask)\ .filter( UmrahTask.umrah_id == session_id, UmrahTask.task_id == sub.id_sub_task ).first() if status: subtask_list.append({ "subtaskid": sub.id_sub_task, "name": sub.name_subtask, "hukum": sub.hukum, "task_image": sub.task_image, "info": sub.info, "video_path": sub.video_path, "no_task": sub.no_task, "status": status.status if status else None }) # ? Only append if there are subtasks if subtask_list: task_list.append({ "id": main.id_main_task, "name": main_task_name, "task": subtask_list }) return { "message": f"You are a {gender.capitalize()}, here are your tasks", "tasks": task_list } except Exception as e: return {"error": str(e)} @app.get("/list_task_tamatUmrah") async def list_task_tamatUmrah(user_id: str,session_id:str, db: Session = Depends(get_db)): try: # Fetch user from the database user = db.query(User).filter(User.id == user_id).first() if not user: return {"message": "User not found"} gender = user.gender # Get user gender task_list = [] # List to store all tasks # Choose the correct SubTask table based on gender SubTaskModel = SubTaskMale if gender == "male" else SubTaskFemale # Fetch main tasks main_tasks = db.query(MainTask).all() for main in main_tasks: main_task_name = "BERGUNTING" if gender == "female" and main.id_main_task == "task-009" else main.name_task # Fetch subtasks related to the main task subtasks = db.query(SubTaskModel).filter(SubTaskModel.id_main_task == main.id_main_task).order_by (SubTaskModel.no_task.asc()).all() subtask_list = [] # List to store subtasks for this main task for sub in subtasks: status=db.query(UmrahTask).filter(UmrahTask.umrah_id==session_id,UmrahTask.task_id==sub.id_sub_task).first() subtask_list.append({ "subtaskid": sub.id_sub_task, "name": sub.name_subtask, "hukum": sub.hukum, "task_image": sub.task_image, "info": sub.info, "video_path": sub.video_path, "no_task": sub.no_task, "status": status.status }) # Append main task with subtasks task_list.append({ "id": main.id_main_task, "name":main_task_name, "task": subtask_list # Grouped subtasks inside each main task }) return { "message": f"You are a {gender.capitalize()}, here are your tasks", "tasks": task_list } except Exception as e: return {"error": str(e)} @app.get("/list_task_2nd") async def list_task(user_id: str,session_id:str, db: Session = Depends(get_db)): try: # Fetch user from the database user = db.query(User).filter(User.id == user_id).first() if not user: return {"message": "User not found"} gender = user.gender # Get user gender task_list = [] # List to store all tasks # Choose the correct SubTask table based on gender SubTaskModel = SubTaskMale if gender == "male" else SubTaskFemale # Fetch main tasks main_tasks = db.query(MainTask).all() for main in main_tasks: # Fetch subtasks related to the main task subtasks = db.query(SubTaskModel).filter(SubTaskModel.id_main_task == main.id_main_task).order_by (SubTaskModel.no_task.asc()).all() subtask_list = [] # List to store subtasks for this main task for sub in subtasks: status=db.query(UmrahTask).filter(UmrahTask.umrah_id==session_id,UmrahTask.task_id==sub.id_sub_task).first() subtask_list.append({ "subtaskid": sub.id_sub_task, "name": sub.name_subtask, "hukum": sub.hukum, "task_image": sub.task_image, "info": sub.info, "video_path": sub.video_path, "no_task": sub.no_task, "status": status.status }) # Append main task with subtasks task_list.append({ "id": main.id_main_task, "name":main.name_task, "task": subtask_list # Grouped subtasks inside each main task }) return { "message": f"You are a {gender.capitalize()}, here are your tasks", "tasks": task_list } except Exception as e: return {"error": str(e)} @app.put("/edit_data") def edit_data( task_id: str, gender: str, subtask_name: Optional[str] = None, hukum: Optional[str] = None, task_image: Optional[str] = None, info: Optional[str] = None, video_path: Optional[str] = None, no_task: Optional[str] = None, db: Session = Depends(get_db) ): if gender == "male": SubTaskModel = SubTaskMale elif gender == "female": SubTaskModel = SubTaskFemale else: raise HTTPException(status_code=400, detail="Invalid gender") task = db.query(SubTaskModel).filter(SubTaskModel.id_sub_task == task_id).first() if not task: raise HTTPException(status_code=404, detail="Task not found") # Update fields if new values are provided if subtask_name: task.name_subtask = subtask_name if hukum: task.hukum = hukum if task_image: task.task_image = task_image if info: task.info = info if video_path: task.video_path = video_path if no_task: task.no_task = no_task db.commit() return {"message": f"Success edit {gender} task"} @app.delete("/delete_data") def delete_data(task_id: str, gender: str, db: Session = Depends(get_db)): if gender == "male": SubTaskModel = SubTaskMale elif gender == "female": SubTaskModel = SubTaskFemale else: raise HTTPException(status_code=400, detail="Invalid gender") task = db.query(SubTaskModel).filter(SubTaskModel.id_sub_task == task_id).first() if not task: raise HTTPException(status_code=404, detail="Task not found") db.delete(task) db.commit() return {"message": f"Success delete {gender} task"} @app.get("/toggle") async def toggle_task( user_id: str, session_id: str, task_id: str, statuss: str, db: Session = Depends(get_db) ): try: # ? Check task exists first status = db.query(UmrahTask).filter( UmrahTask.umrah_id == session_id, UmrahTask.task_id == task_id ).first() if not status: raise HTTPException(status_code=404, detail="Task not found in this session") # ? Update task status status.status = statuss db.commit() # ? Get user and subtask model user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") SubTaskModel = SubTaskMale if user.gender == "male" else SubTaskFemale # ? Special case: task-010 toggles task-011 subtasks TASK_010_ID = ( "c252f231-9694-11f0-96af-020017010d82" if user.gender == "male" else "3ea997bb-9437-11f0-96af-020017010d82" ) if status.task_id == TASK_010_ID: if status.status == "incomplete": # ?? user turning OFF -> remove task-011 subtasks tasks = db.query(SubTaskModel.id_sub_task).filter( SubTaskModel.id_main_task == 'task-011' ).subquery() db.query(UmrahTask).filter( UmrahTask.umrah_id == session_id, UmrahTask.task_id.in_(tasks) ).delete(synchronize_session=False) db.commit() return {"message": "task-011 subtasks removed"} elif status.status == "completed": # ?? user turning ON -> add task-011 subtasks task_list = db.query(SubTaskModel).filter( SubTaskModel.id_main_task == 'task-011' ).all() if not task_list: raise HTTPException(status_code=404, detail="No subtasks found for this gender") session_tasks = [] for task in task_list: session_task = UmrahTask( id=str(uuid4()), umrah_id=session_id, task_id=task.id_sub_task, status="incomplete", calculate='false' ) db.add(session_task) session_tasks.append({ "task_id": task.id_sub_task, "name": task.name_subtask, "hukum": task.hukum, "task_image": task.task_image, "info": task.info, "video_path": task.video_path, "no_task": task.no_task, "status": False }) db.commit() return { "message": "task-011 subtasks added", "subtasks": session_tasks } # ? Default toggle (normal task) return {"message": "Task updated", "task_id": task_id, "status": status.status} except HTTPException as e: raise e except Exception as e: db.rollback() return {"error": str(e)} @app.get("/repeatcheck") async def repeatcheck(userid:str,db:Session=Depends(get_db)): try: user = db.query(User).filter(User.id == userid).first() if not user: raise HTTPException(status_code=404, detail="User not found") session = db.query(SessionUmrah).filter(SessionUmrah.id_user == userid, SessionUmrah.status == "incomplete").first() if not session: return {"message":"No Session FOund"} umrah=db.query(Umrah).filter(Umrah.id_session==session.id,Umrah.status=="incomplete").first() listTask = db.query(UmrahTask).filter(UmrahTask.umrah_id == umrah.id, or_( UmrahTask.task_id == 'c252f231-9694-11f0-96af-020017010d82', UmrahTask.task_id == '3ea997bb-9437-11f0-96af-020017010d82' )).first() if listTask: return { "message": "exist" } else: return{ "message":"not exist" } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) @app.get("/testendpoint") async def tester(userid:str,db:Session=Depends(get_db)): try: user = db.query(User).filter(User.id == userid).first() if not user: raise HTTPException(status_code=404, detail="User not found") session = db.query(SessionUmrah).filter(SessionUmrah.id_user == userid, SessionUmrah.status == "incomplete").first() if not session: raise HTTPException(status_code=404, detail="Session not found") umrah=db.query(Umrah).filter(Umrah.id_session==session.id,Umrah.status=="incomplete").first() listTask = db.query(UmrahTask).filter(UmrahTask.umrah_id == umrah.id, or_( UmrahTask.task_id == '786d603d-eb2d-436a-b804-717da7ff5a24', UmrahTask.task_id == '64a1af39-52c1-4386-91d5-3f3f3fe25efb' )).first() if listTask: return { "message": "exist" } else: return{ "message":"not exist" } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) @app.get("/end_umrah_add_task") async def end_umrah_add_task(user_id: str, db: Session = Depends(get_db)): try: # ? Check if user exists user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") # ? Determine gender-based subtasks SubTaskModel = SubTaskMale if user.gender == "male" else SubTaskFemale task_list = db.query(SubTaskModel).filter(SubTaskModel.id_main_task.in_(['task-010'])).all() if not task_list: raise HTTPException(status_code=404, detail="No subtasks found for this gender") session = db.query(SessionUmrah).filter(SessionUmrah.id_user == user_id, SessionUmrah.status == "incomplete").first() # ? Check if user has an active session (incomplete) sessions = db.query(Umrah).filter(Umrah.id_session==session.id, Umrah.status=='incomplete').first() if not session: # ?? If no active session, create a new one raise HTTPException(status_code=404, detail="No Umrah found for this User") # ? Create a new `Umrah` under the existing session # ? Insert tasks for the new Umrah session_tasks = [] for task in task_list: session_task = UmrahTask( id=str(uuid4()), umrah_id=sessions.id, task_id=task.id_sub_task, status="incomplete", calculate='false' ) db.add(session_task) session_tasks.append({ "task_id": task.id_sub_task, "name": task.name_subtask, "hukum": task.hukum, "task_image": task.task_image, "info": task.info, "video_path": task.video_path, "no_task": task.no_task, "status": False }) db.commit() return { "message": f"You are {user.gender}, a new Umrah has been started under your session.", "session_id": session.id, "umrah_id": sessions.id, "tasks": session_tasks } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) @app.get("/start_umrah") async def start_umrah(user_id: str, miqats: str, db: Session = Depends(get_db)): try: # ? Check if user exists user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") # ? Determine gender-based subtasks SubTaskModel = SubTaskMale if user.gender == "male" else SubTaskFemale task_list = db.query(SubTaskModel).filter(SubTaskModel.id_main_task.notin_(['task-010', 'task-011'])).all() if not task_list: raise HTTPException(status_code=404, detail="No subtasks found for this gender") # ? Check if user has an active session (incomplete) session = db.query(SessionUmrah).filter(SessionUmrah.id_user == user_id, SessionUmrah.status == "incomplete").first() if not session: # ?? If no active session, create a new one session = SessionUmrah( id=str(uuid4()), id_user=user_id, start_session=datetime.utcnow(), status="incomplete" # ? Matches your required status ) db.add(session) db.flush() db.refresh(session) # ? Create a new `Umrah` under the existing session new_umrah = Umrah( id=str(uuid4()), id_session=session.id, umrah_start=datetime.utcnow(), status="incomplete", # ? Default status miqat=miqats ) db.add(new_umrah) db.commit() db.refresh(new_umrah) # ? Insert tasks for the new Umrah session_tasks = [] for task in task_list: session_task = UmrahTask( id=str(uuid4()), umrah_id=new_umrah.id, task_id=task.id_sub_task, calculate='true', status="incomplete" ) db.add(session_task) session_tasks.append({ "task_id": task.id_sub_task, "name": task.name_subtask, "hukum": task.hukum, "task_image": task.task_image, "info": task.info, "video_path": task.video_path, "no_task": task.no_task, "status": False }) db.commit() return { "message": f"You are {user.gender}, a new Umrah has been started under your session.", "session_id": session.id, "umrah_id": new_umrah.id, "tasks": session_tasks } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) @app.get("/repeat_umrah") async def repeatumrah(user_id: str,miqats:str, db: Session = Depends(get_db)): try: user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") SubTaskModel = SubTaskMale if user.gender == "male" else SubTaskFemale task_list = db.query(SubTaskModel).filter(SubTaskModel.id_main_task.notin_(['task-010', 'task-011'])).all() if not task_list: raise HTTPException(status_code=404, detail="No subtasks found for this gender") # ? Find the active (incomplete) session session = db.query(SessionUmrah).filter(SessionUmrah.id_user == user_id, SessionUmrah.status == "incomplete").first() if not session: raise HTTPException(status_code=404, detail="No active Umrah session found") # ? Find the latest incomplete Umrah in this session ongoing_umrah = db.query(Umrah).filter(Umrah.id_session == session.id, Umrah.status == "incomplete").first() if not ongoing_umrah: raise HTTPException(status_code=404, detail="No ongoing Umrah found to complete") # ? Mark this Umrah as complete ongoing_umrah.status = "complete" ongoing_umrah.umrah_finish = datetime.utcnow() db.commit() db.refresh(ongoing_umrah) # ? Check if all Umrah records in this session are complete new_umrah = Umrah( id=str(uuid4()), id_session=session.id, umrah_start=datetime.utcnow(), status="incomplete", # ? Default status miqat=miqats ) db.add(new_umrah) db.commit() db.refresh(new_umrah) # ? Insert tasks for the new Umrah session_tasks = [] for task in task_list: session_task = UmrahTask( id=str(uuid4()), umrah_id=new_umrah.id, task_id=task.id_sub_task, status="incomplete" , calculate='true' ) db.add(session_task) session_tasks.append({ "task_id": task.id_sub_task, "name": task.name_subtask, "hukum": task.hukum, "task_image": task.task_image, "info": task.info, "video_path": task.video_path, "no_task": task.no_task, "status": False }) db.commit() return { "message": f"You are {user.gender}, a new Umrah has been started under your session.", "session_id": session.id, "umrah_id": new_umrah.id, "tasks": session_tasks } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) @app.get("/end_umrah") async def end_umrah(user_id: str, db: Session = Depends(get_db)): try: # ? Find the active (incomplete) session session = db.query(SessionUmrah).filter(SessionUmrah.id_user == user_id, SessionUmrah.status == "incomplete").first() if not session: raise HTTPException(status_code=404, detail="No active Umrah session found") # ? Find the latest incomplete Umrah in this session ongoing_umrah = db.query(Umrah).filter(Umrah.id_session == session.id, Umrah.status == "incomplete").first() if not ongoing_umrah: raise HTTPException(status_code=404, detail="No ongoing Umrah found to complete") # ? Mark this Umrah as complete ongoing_umrah.status = "complete" ongoing_umrah.umrah_finish = datetime.utcnow() db.commit() db.refresh(ongoing_umrah) # ? Check if all Umrah records in this session are complete remaining_umrah = db.query(Umrah).filter(Umrah.id_session == session.id, Umrah.status == "incomplete").count() if remaining_umrah == 0: session.status = "complete" # ? Marks the session as fully completed session.finish_session = datetime.utcnow() db.commit() db.refresh(session) return { "message": "Umrah completed successfully", "umrah_id": ongoing_umrah.id, "session_status": session.status } except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e)) #@app.put("/sub_tasks/{sub_task_id}") #def update_sub_task_status(sub_task_id: str, db: Session = Depends(get_db)): # sub_task = db.query(SubTask).filter(SubTask.id_sub_task == sub_task_id).first() # if not sub_task: # raise HTTPException(status_code=404, detail="Sub-task not found") # # sub_task.status = not sub_task.status # Toggle the status # db.commit() # db.refresh(sub_task) # return sub_task @app.get("/add_data") def ass_data(mainTask_id:str,gender:str,subtask_name:str,hukum:str,task_image:str,info:str,video_path:str,no_task:str,db: Session = Depends(get_db)): if(gender=="male"): try: sub_task = db.query(SubTaskMale).filter(and_( SubTaskMale.name_subtask == subtask_name, SubTaskMale.id_main_task == mainTask_id # Assuming maintask is the id_main_task value )).first() if sub_task: raise HTTPException(status_code=404, detail="Sub-task already found") newsub_task=SubTaskMale( name_subtask=subtask_name, hukum=hukum, task_image=task_image, info=info, video_path=video_path, no_task=no_task, id_main_task=mainTask_id ) db.add(newsub_task) except: raise HTTPException(status_code=404, detail="Sub-task already found") db.commit() return {"message":"Success add {gender} task"} elif(gender=="female"): try: sub_task = db.query(SubTaskFemale).filter(and_( SubTaskFemale.name_subtask == subtask_name, SubTaskFemale.id_main_task == mainTask_id # Assuming maintask is the id_main_task value )).first() if sub_task: raise HTTPException(status_code=404, detail="Sub-task already found") newsub_task=SubTaskFemale( name_subtask=subtask_name, hukum=hukum, task_image=task_image, info=info, video_path=video_path, no_task=no_task, id_main_task=mainTask_id ) db.add(newsub_task) except: raise HTTPException(status_code=404, detail="Sub-task already found") db.commit() return {"message":"Success add {gender} task"} @app.get("/users/{user_id}") def get_user(user_id: str, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return {"id": user.id, "name": user.name, "email": user.email, "gender": user.gender} #@app.post("/submit-miqat/") #async def submit_miqat(data: MiqatSubmission, db: Session = Depends(get_db)): # session = db.query(SessionUmrah).filter(SessionUmrah.id == data.id).first() #if not session: #raise HTTPException(status_code=404, detail="Umrah session not found") #session.miqat = data.miqat # Update miqat field #db.commit() #db.refresh(session) #return {"message": "Miqat updated successfully", "id": session.id, "miqat": session.miqat} @app.get("/user_report") async def user_report(user_id: str, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") task_table = SubTaskMale if user.gender == "male" else SubTaskFemale sessions = db.query(SessionUmrah).filter( SessionUmrah.id_user == user_id, SessionUmrah.status == 'complete' ).all() overall_completed = defaultdict(int) overall_total = defaultdict(int) report_Session = [] report_Umrah=[] start_dates = [] finish_dates = [] for session in sessions: # Get COMPLETED task counts umrahs=db.query(Umrah).filter(Umrah.id_session==session.id,Umrah.status=="complete").all() for umrah in umrahs: completed_counts = ( db.query(task_table.hukum, func.count()) .join(UmrahTask, UmrahTask.task_id == task_table.id_sub_task) .filter( UmrahTask.umrah_id == umrah.id, UmrahTask.status == "completed" ) .group_by(task_table.hukum) .all() ) # Get TOTAL task counts (ALL TASKS IN SESSION) total_counts = ( db.query(task_table.hukum, func.count()) .join(UmrahTask, UmrahTask.task_id == task_table.id_sub_task) .filter( UmrahTask.umrah_id == umrah.id ) .group_by(task_table.hukum) .all() ) for hukum, count in completed_counts: overall_completed[hukum] += count for hukum, count in total_counts: overall_total[hukum] += count report_Umrah.append({ "umrah_id": umrah.id, "tasks_completed": dict(completed_counts), # Convert list of tuples to dict "total_tasks": dict(total_counts) , "start_umrah": umrah.umrah_start, "finish_umrah":umrah.umrah_finish # Add this line }) if umrah.umrah_start: start_dates.append(umrah.umrah_start) if umrah.umrah_finish: finish_dates.append(umrah.umrah_finish) report_Session.append({ "session_id": session.id, "Umrah":report_Umrah, "start_Session":min(start_dates) if start_dates else None, "end_session":max(finish_dates) if finish_dates else None, "overall_summary": { "completed": dict(overall_completed), "total": dict(overall_total) } }) report_Umrah=[] return {"user_id": user_id, "history": report_Session} @app.get("/getoverall") def calculateoverallpercent(session_umrah_id: str, db: Session = Depends(get_db)): total_tasks = db.query(UmrahTask).filter(UmrahTask.umrah_id == session_umrah_id,UmrahTask.calculate=='true').count() completed_tasks = db.query(UmrahTask).filter(UmrahTask.umrah_id == session_umrah_id,UmrahTask.calculate=='true', UmrahTask.status == 'completed').count() return {"total_tasks": total_tasks, "completed_tasks": completed_tasks} @app.get("/raw_tasks/{session_umrah_id}") def get_raw_tasks(session_umrah_id: str, db: Session = Depends(get_db)): hukum_categories = ["Sunat", "Wajib", "Syarat"] raw_tasks = {} for hukum in hukum_categories: # Fetch all tasks for the given hukum category tasks = ( db.query(UmrahTask) .join(SubTaskMale, UmrahTask.task_id == SubTaskMale.id_sub_task) .filter(SubTaskMale.hukum == hukum, UmrahTask.umrah_id == session_umrah_id) .all() ) # Transform tasks into a list of dictionaries raw_tasks[hukum] = [{"id": task.id, "status": task.status} for task in tasks] return raw_tasks @app.get("/profile/{user_id}", response_model=UserResponse) def get_user(user_id: str, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return user # Seed data (run once) @app.post("/api/quiz/seed") def seed_quiz(db: Session = Depends(get_db)): samples = [ QuizUmrah(kategori="Rukun Umrah", pernyataan="Tawaf termasuk dalam rukun umrah.", jawapan="Betul"), QuizUmrah(kategori="Wajib Umrah", pernyataan="Ihram bukan wajib dalam umrah.", jawapan="Salah"), QuizUmrah(kategori="Persiapan Umrah", pernyataan="Sunat mandi ihram sebelum niat.", jawapan="Betul"), ] db.add_all(samples) db.commit() return {"message": "Data seeded."} # GET random questions @app.get("/api/quiz/random", response_model=List[QuizQuestion]) def get_random_questions(count: int = 10, db: Session = Depends(get_db)): return db.query(QuizUmrah).order_by(func.random()).limit(count).all()