from datetime import timedelta, datetime, timezone from uuid import UUID from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.api import deps from app.models import ( Project, Slot, SlotTemplate, SlotTag, Sms, User, Volunteer, ) from app.schemas.requests import ( ProjectImportGsheetRequest, ProjectRequest, ProjectSMSBatchRequest, ) from app.schemas.responses import ProjectListResponse, ProjectResponse, SMSResponse from app.importData.gsheet import parseGsheet, extract_doc_uid, ParsingError router = APIRouter() @router.get("/projects", response_model=list[ProjectListResponse]) async def list_project( current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Get project_list""" results = session.execute(select(Project)) return results.scalars().all() @router.get("/public-projects", response_model=list[ProjectListResponse]) async def list_public_project( session: Session = Depends(deps.get_session), ): """Get the list of public projects""" results = session.execute(select(Project).where(Project.is_public)) return results.scalars().all() @router.post("/project", response_model=ProjectResponse) async def create_project( new_project: ProjectRequest, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Create a new project""" project = Project(**new_project.model_dump()) session.add(project) try: session.commit() except IntegrityError: raise HTTPException(400, "Project name already exist") session.refresh(project) return project @router.get("/public-project/{project_id}", response_model=ProjectResponse) async def get_public_project( project_id: UUID, session: Session = Depends(deps.get_session), ): """Get a project that is public""" result = session.get(Project, project_id) if (result is None) or not result.is_public: raise HTTPException(status_code=404, detail="Project not found") return result @router.get("/project/{project_id}", response_model=ProjectResponse) async def get_project( project_id: UUID, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Get a project""" project = session.get(Project, project_id) if project is None: raise HTTPException(status_code=404, detail="Project not found") return project @router.post("/project/{project_id}", response_model=ProjectListResponse) async def update_project( project_id: UUID, edit_project: ProjectRequest, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Edit project""" p = session.get(Project, project_id) if p is None: raise HTTPException(status_code=404, detail="Project not found") p.name = edit_project.name p.is_public = edit_project.is_public session.commit() return p @router.post("/project/{project_id}/import-gsheet", response_model=ProjectResponse) async def update_project_from_gsheet( project_id: UUID, gsheet: ProjectImportGsheetRequest, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Edit project name""" p: Project = session.get(Project, project_id) if p is None: raise HTTPException(status_code=404, detail="Project not found") doc_id = extract_doc_uid(gsheet.sheet_url) if gsheet.erase_data: p.slots = [] p.sms = [] p.volunteers = [] p.tags = [] p.templates = [] # Parse the gsheets try: data = parseGsheet(doc_id, gsheet.satursday_date) except ParsingError as exc: raise HTTPException(status_code=422, detail=str(exc)) # Create the volunteer list volunteer_map: dict[str, Volunteer] = {} for _, row in data.contact.iterrows(): volunteer = Volunteer( project_id=project_id, name=row["Prénom"], surname=row["Nom"], email=row["Mail"], phone_number=row["Tél"], automatic_sms=row["SMS"], ) volunteer_map[row.key] = volunteer session.add(volunteer) # Create creneau templates template_map = {} tags_map = {tag.title: tag for tag in p.tags} for _, row in data.creneauData.iterrows(): template = SlotTemplate( project_id=project_id, title=row.title, description=row.description, place=row.lieu, responsible_contact=row.responsable, ) for s_tag in row.tags.split(","): if s_tag != "": if s_tag in tags_map: tag = tags_map[s_tag] else: tag = SlotTag(project_id=project_id, title=s_tag) session.add(tag) tags_map[s_tag] = tag template.tags.append(tag) template_map[template.title] = template session.add(template) df_planning = data.planning # group planning entry per same name and timing date_format = "%Y/%m/%d %H:%M" df_planning["key"] = ( df_planning.nom.str.strip() + "_" + df_planning.start.dt.strftime(date_format) + "-" + df_planning.end.dt.strftime(date_format) ) df_slots = df_planning.groupby("key") # Create slots for key in df_slots.groups.keys(): group = df_slots.get_group(key) volunteers = group.benevole_nom.tolist() slot = Slot( project_id=project_id, title=group.nom.iloc[0], starting_time=group.start.iloc[0], ending_time=group.end.iloc[0], required_volunteers=len(volunteers), ) # Add volunteer to slots for benevole_key in volunteers: if benevole_key in volunteer_map: slot.volunteers.append(volunteer_map[benevole_key]) # add detail information if available template_id = group.template_id.iloc[0] if template_id in template_map: slot.template = template_map[template_id] session.add(slot) session.commit() session.refresh(p) return p @router.post("/project/{project_id}/create-all-sms", response_model=list[SMSResponse]) async def create_sms_batch( project_id: UUID, sms_batch: ProjectSMSBatchRequest, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Create SMS based on a template and the list of slots and volunteer associated to the project The placeholder that can be used in the template are - {titre} slot.title - {description} slot.description - {debut} slot.starting_time - {fin} slot.ending_ting - {respo} slot.responsible_contact - {prenom} volunteer.name - {nom} volunteer.surname """ p = session.get(Project, project_id) if p is None: raise HTTPException(status_code=404, detail="Project not found") # Get all slots slots = session.execute(select(Slot).where(Slot.project_id == project_id)) sms_list = [] now = datetime.now(timezone.utc) for slot in slots.scalars(): # Replace the slot placeholder by their value slot_content = ( sms_batch.template.replace("{titre}", slot.title) .replace("{debut}", slot.starting_time.strftime("%Hh%M")) .replace("{fin}", slot.ending_time.strftime("%Hh%M")) ) if slot.template is not None: slot_content = slot_content.replace("{description}", slot.template.description).replace( "{respo}", slot.template.responsible_contact ) sending_time = slot.starting_time - timedelta(minutes=sms_batch.delta_t) # Skip SMS that should have been send before now if sending_time < now: continue for volunteer in slot.volunteers: if not volunteer.automatic_sms: continue # Create a new SMS customized for each user attache to the slot personalized_content = slot_content.replace("{prenom}", volunteer.name).replace( "{nom}", volunteer.surname ) sms = Sms( project_id=project_id, volunteer_id=volunteer.id, content=personalized_content, phone_number=volunteer.phone_number, sending_time=sending_time, ) sms_list.append(sms) session.add_all(sms_list) session.commit() return sms_list @router.delete("/project/{project_id}") async def delete_project( project_id: UUID, current_user: User = Depends(deps.get_current_user), session: Session = Depends(deps.get_session), ): """Delete project""" session.execute(delete(Project).where(Project.id == project_id)) session.commit()