diff --git a/app/routers/admin/account.py b/app/routers/admin/account.py index edb566d..3f87e0a 100644 --- a/app/routers/admin/account.py +++ b/app/routers/admin/account.py @@ -1,151 +1,32 @@ import logging -from datetime import datetime, timezone from pathlib import Path from typing import Annotated from uuid import UUID -from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status +from fastapi import APIRouter, BackgroundTasks, HTTPException, Query from fastapi.responses import FileResponse -from sqlalchemy import and_, func, or_ -from sqlalchemy.orm import aliased -from sqlmodel import col, select +from sqlmodel import select +from app.config import EmailConfig from app.core.db import SessionDep -from app.core.orm import eager_load -from app.models.constants import ( - DEFAULT_FILE_EXTENSION, - QuestionLabel, - RankingSort, - SortOrder, -) -from app.models.forms import ( - Forms_Answer, - Forms_AnswerFile, - Forms_Application, - Forms_HackathonApplicant, - Forms_Question, - StatusEnum, -) +from app.models.constants import RankingSort, SortOrder +from app.models.forms import Forms_Application, StatusEnum from app.models.requests import BulkEmailRequest from app.models.user import Account_User, UserPublic -from app.models.judging import JudgingApplicationScore -from app.services.email import send_email, send_rsvp +from app.services.admin_applications import ( + get_application_detail, + get_resume_metadata, + list_applications, + sanitize_filename, + update_application_status as update_status, +) +from app.services.bulk_email import get_bulk_email_recipients, send_batch_email router = APIRouter() logger = logging.getLogger(__name__) - -def _sanitize_filename(filename: str) -> str: - import re - - filename = Path(filename).name - - filename = filename.replace("\x00", "") - - filename = filename.replace("..", "") - filename = filename.replace("./", "") - filename = filename.replace("../", "") - - filename = re.sub(r"[^\w\s\-.]", "", filename) - - filename = filename.lstrip(".") - - max_length = 255 - if len(filename) > max_length: - name_parts = filename.rsplit(".", 1) - if len(name_parts) == 2: - name, ext = name_parts - filename = name[: max_length - len(ext) - 1] + "." + ext - else: - filename = filename[:max_length] - - if not filename or filename.isspace(): - filename = f"file{DEFAULT_FILE_EXTENSION}" - - return filename - - -def send_batch_email( - users_data: list[dict], - template_path: str, - subject: str, - text_body: str, - base_context: dict, -): - from concurrent.futures import ThreadPoolExecutor - - from app.config import EmailConfig - - total = len(users_data) - successful = 0 - failed = 0 - failures = [] - - max_concurrent = EmailConfig.BULK_MAX_CONCURRENT - chunk_size = EmailConfig.BULK_CHUNK_SIZE - - logger.info( - f"Starting bulk email send: {total} recipients, subject='{subject}', " - f"template='{template_path}', concurrency={max_concurrent}, chunk_size={chunk_size}" - ) - - def send_one(user_data: dict) -> tuple[bool, str, dict]: - email = user_data.get("email", "unknown") - try: - email_context = base_context.copy() if base_context else {} - email_context.update(user_data) - - status_code, response = send_email( - template_path, - email, - subject, - text_body, - email_context, - ) - - if status_code == 200: - return (True, email, {}) - return ( - False, - email, - { - "email": email, - "reason": f"Status {status_code}", - "response": response, - }, - ) - except Exception as e: - return (False, email, {"email": email, "reason": str(e)}) - - for i in range(0, total, chunk_size): - chunk = users_data[i : i + chunk_size] - chunk_num = (i // chunk_size) + 1 - total_chunks = (total + chunk_size - 1) // chunk_size - - logger.info( - f"Processing chunk {chunk_num}/{total_chunks} ({len(chunk)} emails)" - ) - - with ThreadPoolExecutor(max_workers=max_concurrent) as executor: - results = list(executor.map(send_one, chunk)) - - for success, email, error_info in results: - if success: - successful += 1 - logger.debug(f"Email sent successfully to {email}") - else: - failed += 1 - failures.append(error_info) - logger.warning( - f"Email send failed to {email}: {error_info.get('reason')}" - ) - - logger.info( - f"Bulk email send complete: {successful}/{total} successful, {failed}/{total} failed" - ) - - if failures: - logger.warning(f"Failed emails summary: {failures[:10]}") +# Kept as a private alias for compatibility with existing imports. +_sanitize_filename = sanitize_filename @router.get("/users", response_model=list[UserPublic]) @@ -164,64 +45,23 @@ def get_applicants( offset: int = 0, limit: Annotated[int, Query(le=100)] = 100, ): - applicants = session.exec( + return session.exec( select(Account_User) .join(Forms_Application, Account_User.uid == Forms_Application.uid) .offset(offset) .limit(limit) ).all() - return applicants @router.get("/applications/{application_id}/resume") -def get_resume( - application_id: UUID, - session: SessionDep, -): - statement = select(Forms_AnswerFile).where( - Forms_AnswerFile.application_id == application_id - ) - resume = session.exec(statement).first() - - if not resume or not resume.file_path: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Resume not found" - ) - - file_path = Path(resume.file_path) - if not file_path.exists() or not file_path.is_file(): - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="File not found on disk" - ) - - safe_filename = _sanitize_filename( - resume.original_filename or f"resume{DEFAULT_FILE_EXTENSION}" - ) - - return FileResponse( - path=str(file_path), - media_type="application/pdf", - filename=safe_filename, - ) +def get_resume(application_id: UUID, session: SessionDep): + path, filename = get_resume_metadata(session, application_id) + return FileResponse(path=str(path), media_type="application/pdf", filename=filename) @router.get("/applications/{application_id}") def get_application(application_id: UUID, session: SessionDep): - statement = select(Forms_Application).where( - Forms_Application.application_id == application_id - ) - application = session.exec(statement).first() - if application is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Application not found" - ) - return { - "application": application, - "form_answers": application.form_answers, - "form_answersfile": application.form_answersfile.original_filename - if application.form_answersfile - else None, - } + return get_application_detail(session, application_id) @router.get("/applications") @@ -237,313 +77,58 @@ def get_all_apps( ranking_sort: RankingSort | None = None, role: StatusEnum | None = None, ): - from datetime import timedelta - - from app.cache import cache - - def fetch_questions(): - questions_statement = select(Forms_Question).where( - col(Forms_Question.label).in_( - [ - QuestionLabel.CURRENT_LEVEL_OF_STUDY.value, - QuestionLabel.GENDER.value, - QuestionLabel.SCHOOL_NAME.value, - ] - ) - ) - questions = session.exec(questions_statement).all() - return {q.label: q for q in questions} - - question_map = cache.get_or_set( - key="admin_filter_questions", - factory_func=fetch_questions, - ttl=timedelta(minutes=10), + return list_applications( + session, + offset=ofs, + limit=limit, + search=search, + level_of_study=level_of_study, + gender=gender, + school=school, + date_sort=date_sort, + ranking_sort=ranking_sort, + application_status=role, ) - level_of_study_question = question_map.get( - QuestionLabel.CURRENT_LEVEL_OF_STUDY.value - ) - gender_question = question_map.get(QuestionLabel.GENDER.value) - school_question = question_map.get(QuestionLabel.SCHOOL_NAME.value) - - level_of_study_data = aliased(Forms_Answer) - gender_data = aliased(Forms_Answer) - school_data = aliased(Forms_Answer) - - statement = ( - select( - Account_User, - Forms_Application, - Forms_HackathonApplicant, - col(level_of_study_data.answer).label("level_of_study_answer"), - col(gender_data.answer).label("gender_answer"), - col(school_data.answer).label("school_answer"), - col(JudgingApplicationScore.mu).label("ranking_mu"), - col(JudgingApplicationScore.sigma_sq).label("ranking_sigma_sq"), - col(JudgingApplicationScore.comparison_count).label( - "ranking_comparison_count" - ), - ) - .where( - Account_User.is_active, - Account_User.application is not None, - ) - .join(Forms_Application, Account_User.uid == Forms_Application.uid) - .join( - Forms_HackathonApplicant, - Forms_Application.application_id == Forms_HackathonApplicant.application_id, - ) - .outerjoin( - JudgingApplicationScore, - JudgingApplicationScore.application_id - == Forms_Application.application_id, - ) - ) - - if level_of_study_question: - statement = statement.outerjoin( - level_of_study_data, - and_( - level_of_study_data.application_id == Forms_Application.application_id, - level_of_study_data.question_id == level_of_study_question.question_id, - ), - ) - - if gender_question: - statement = statement.outerjoin( - gender_data, - and_( - gender_data.application_id == Forms_Application.application_id, - gender_data.question_id == gender_question.question_id, - ), - ) - - if school_question: - statement = statement.outerjoin( - school_data, - and_( - school_data.application_id == Forms_Application.application_id, - school_data.question_id == school_question.question_id, - ), - ) - - if search: - search_pattern = f"%{search}%" - statement = statement.where( - or_( - col(Account_User.first_name).ilike(search_pattern), - col(Account_User.last_name).ilike(search_pattern), - col(Account_User.email).ilike(search_pattern), - (col(Account_User.first_name) + " " + col(Account_User.last_name)).ilike( - search_pattern - ), - ) - ) - - if role: - statement = statement.where(Forms_HackathonApplicant.status == role) - - if level_of_study and level_of_study_question: - statement = statement.where( - func.lower(level_of_study_data.answer) == level_of_study.lower() - ) - - if gender and gender_question: - statement = statement.where(func.lower(gender_data.answer) == gender.lower()) - - if school and school_question: - statement = statement.where( - and_( - col(school_data.answer).isnot(None), - school_data.answer != "", - func.lower(school_data.answer) == school.lower(), - ) - ) - - if ranking_sort: - if ranking_sort == RankingSort.HIGHEST: - statement = statement.order_by( - col(JudgingApplicationScore.mu).desc().nulls_last() - ) - else: - statement = statement.order_by( - col(JudgingApplicationScore.mu).asc().nulls_last() - ) - elif date_sort: - if date_sort == SortOrder.OLDEST: - statement = statement.order_by(col(Forms_Application.updated_at).asc()) - elif date_sort == SortOrder.LATEST: - statement = statement.order_by(col(Forms_Application.updated_at).desc()) - - statement = statement.offset(ofs).limit(limit) - - results = session.exec(statement).all() - - response = [ - { - "first_name": user.first_name, - "last_name": user.last_name, - "email": user.email, - "status": hacker_applicant.status, - "app_id": hacker_applicant.application_id, - "created_at": user_app.created_at, - "updated_at": user_app.updated_at, - "level_of_study": level_study, - "gender": gender_val, - "school": school_val, - "ranking_mu": ranking_mu, - "ranking_sigma_sq": ranking_sigma_sq, - "ranking_comparison_count": ranking_comparison_count or 0, - } - for ( - user, - user_app, - hacker_applicant, - level_study, - gender_val, - school_val, - ranking_mu, - ranking_sigma_sq, - ranking_comparison_count, - ) in results - ] - - return {"application": response, "offset": ofs, "limit": limit} - @router.patch("/applications/{application_id}/status") def update_application_status( application_id: str, request: StatusEnum, session: SessionDep ): - statement = ( - select(Forms_Application, Account_User) - .join(Account_User, Forms_Application.uid == Account_User.uid) - .where(Forms_Application.application_id == application_id) - .options(eager_load(Forms_Application.hackathonapplicant)) - ) - result = session.exec(statement).first() - - if not result: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Application not found" - ) - - application, user = result - hacker_applicant = application.hackathonapplicant - if hacker_applicant is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Applicant status not found" - ) - - try: - hacker_applicant.status = request.value - application.updated_at = datetime.now(timezone.utc) - - session.add(hacker_applicant) - session.add(application) - session.commit() - session.refresh(hacker_applicant) - session.refresh(application) - - if request == StatusEnum.ACCEPTED: - user_full_name = user.full_name - send_rsvp(user.email, user_full_name, application_id) - - except Exception as e: - session.rollback() - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to update application status: {str(e)}", - ) - - return { - "application_id": application_id, - "new_status": request.value, - "updated_at": application.updated_at, - } + return update_status(session, application_id, request) @router.post("/bulk-emails") def send_bulk_email_endpoint( - request: BulkEmailRequest, session: SessionDep, background_tasks: BackgroundTasks + request: BulkEmailRequest, + session: SessionDep, + background_tasks: BackgroundTasks, ): - template_file = Path(request.template_path) - if not template_file.exists() or not template_file.is_file(): - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Template file not found" - ) - - count_statement = ( - select(func.count()) - .select_from(Account_User) - .join(Forms_Application, Account_User.uid == Forms_Application.uid) - .join( - Forms_HackathonApplicant, - Forms_Application.application_id == Forms_HackathonApplicant.application_id, - ) - .where( - Account_User.is_active, - Forms_HackathonApplicant.status == request.status, - ) - ) - - total_recipients = session.exec(count_statement).one() + template = Path(request.template_path) + if not template.exists() or not template.is_file(): + raise HTTPException(status_code=404, detail="Template file not found") - if total_recipients == 0: + total, recipients = get_bulk_email_recipients(session, request) + if total == 0: return { "message": f"No users found with status: {request.status.value}", "total_recipients": 0, "status": "no_recipients", } - - from app.config import EmailConfig - - if total_recipients > EmailConfig.BULK_WARN_THRESHOLD: - logger.warning( - f"Large bulk email operation: {total_recipients} recipients. " - "Consider using a dedicated task queue (Celery/RQ) for production." - ) - - statement = ( - select( - Account_User.first_name, - Account_User.last_name, - Account_User.email, - ) - .join(Forms_Application, Account_User.uid == Forms_Application.uid) - .join( - Forms_HackathonApplicant, - Forms_Application.application_id == Forms_HackathonApplicant.application_id, - ) - .where( - Account_User.is_active, - Forms_HackathonApplicant.status == request.status, - ) - ) - - results = session.exec(statement).all() - - users_data = [ - { - "first_name": row[0], - "last_name": row[1], - "email": row[2], - } - for row in results - ] + if total > EmailConfig.BULK_WARN_THRESHOLD: + logger.warning("Large bulk email operation: %s recipients", total) background_tasks.add_task( send_batch_email, - users_data, + recipients, request.template_path, request.subject, request.text_body, request.context, ) - return { "message": f"Bulk email job queued for status: {request.status.value}", - "total_recipients": total_recipients, + "total_recipients": total, "status": "queued", "note": "Emails are being sent concurrently in the background (chunks of 100, max 10 concurrent)", } diff --git a/app/routers/forms.py b/app/routers/forms.py index 4fd8c6c..b15a323 100644 --- a/app/routers/forms.py +++ b/app/routers/forms.py @@ -1,109 +1,45 @@ -import os -import shutil -import tempfile -from datetime import datetime, timezone -from pathlib import Path +from datetime import timedelta from typing import Annotated -from uuid import uuid4 -from fastapi import APIRouter, Depends, HTTPException, UploadFile, status -from pypdf import PdfReader +from fastapi import APIRouter, Depends, UploadFile, status from sqlmodel import col, select -from app.config import FileUploadConfig +from app.cache import cache from app.core.db import SessionDep -from app.core.orm import eager_load -from app.models.constants import ( - ALLOWED_FILE_EXTENSIONS, - ALLOWED_FILE_TYPES_MESSAGE, - DEFAULT_FILE_EXTENSION, - MAX_ERROR_MESSAGE_LENGTH, - MIN_PDF_PAGES, - PDF_EMBEDDED_FILES_ERROR, - PDF_ENCRYPTED_ERROR, - PDF_JAVASCRIPT_ERROR, - PDF_NO_PAGES_ERROR, - EmailMessage, - EmailSubject, - EmailTemplate, - QuestionLabel, -) from app.models.forms import ( ApplicationResponse, - Forms_Answer, Forms_AnswerUpdate, - Forms_Application, Forms_Form, - Forms_HackathonApplicant, Forms_Question, - StatusEnum, ) from app.models.user import Account_User -from app.services.applications import create_application, is_valid_submission_time +from app.services.applications import is_valid_submission_time from app.services.auth import get_current_user -from app.services.email import send_email, send_rsvp -from app.validators import validate_profile_url +from app.services.form_workflow import ( + get_or_create_application, + save_answers as save_application_answers, + submit_application, +) +from app.services.resume_uploads import upload_resume as store_resume +from app.services.resume_uploads import validate_pdf router = APIRouter() - -def _validate_pdf(filepath: str, filename: str) -> tuple[bool, str]: - - file_ext = Path(filename).suffix.lower() - if file_ext not in ALLOWED_FILE_EXTENSIONS: - return False, ALLOWED_FILE_TYPES_MESSAGE - - try: - reader = PdfReader(filepath) - if reader.is_encrypted: - return False, PDF_ENCRYPTED_ERROR - - if len(reader.pages) < MIN_PDF_PAGES: - return False, PDF_NO_PAGES_ERROR - - def object_contains(forbidden_keys, obj): - if isinstance(obj, dict): - for key, value in obj.items(): - if key in forbidden_keys: - return True - if isinstance(value, (dict, list)): - if object_contains(forbidden_keys, value): - return True - elif isinstance(obj, list): - for item in obj: - if isinstance(item, (dict, list)): - if object_contains(forbidden_keys, item): - return True - return False - - root = reader.trailer.get("/Root") - - javascript_keys = {"/JavaScript", "/JS", "/AA", "/OpenAction"} - if object_contains(javascript_keys, root): - return False, PDF_JAVASCRIPT_ERROR - - embedded_file_keys = {"/EmbeddedFile", "/EmbeddedFiles", "/AF"} - if object_contains(embedded_file_keys, root): - return False, PDF_EMBEDDED_FILES_ERROR - - except Exception as e: - return False, f"Invalid PDF: {str(e)[:MAX_ERROR_MESSAGE_LENGTH]}" - - return True, "" +# Kept as a private alias for compatibility with existing imports. +_validate_pdf = validate_pdf @router.get("/questions") def get_questions(session: SessionDep) -> list[Forms_Question]: - from datetime import timedelta - - from app.cache import cache - - def fetch_questions(): - statement = select(Forms_Question).order_by(col(Forms_Question.question_order)) - return list(session.exec(statement).all()) + def fetch_questions() -> list[Forms_Question]: + return list( + session.exec( + select(Forms_Question).order_by(col(Forms_Question.question_order)) + ).all() + ) return cache.get_or_set( - key="form_questions", factory_func=fetch_questions, ttl=timedelta(minutes=10) + "form_questions", fetch_questions, timedelta(minutes=10) ) @@ -112,37 +48,7 @@ def get_application( current_user: Annotated[Account_User, Depends(get_current_user)], session: SessionDep, ): - if current_user.application is None: - if not is_valid_submission_time(session, current_user): - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Submitting outside submission time", - ) - application = create_application(current_user, session) - else: - statement = ( - select(Forms_Application) - .where(Forms_Application.uid == current_user.uid) - .options( - eager_load(Forms_Application.form_answers), - eager_load(Forms_Application.form_answersfile), - eager_load(Forms_Application.hackathonapplicant), - ) - ) - application = session.exec(statement).first() - - if application is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Application not found" - ) - - return { - "application": application, - "form_answers": application.form_answers, - "form_answersfile": application.form_answersfile.original_filename - if application.form_answersfile - else None, - } + return get_or_create_application(session, current_user) @router.put("/answers") @@ -151,77 +57,7 @@ def save_answers( current_user: Annotated[Account_User, Depends(get_current_user)], session: SessionDep, ): - if not is_valid_submission_time(session, current_user): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Submission is currently closed", - ) - - if current_user.application is None: - current_user.application = create_application(current_user, session) - - statement = ( - select(Forms_Application) - .where(Forms_Application.uid == current_user.uid) - .options(eager_load(Forms_Application.form_answers)) - ) - application = session.exec(statement).first() - if application is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Application not found" - ) - - answer_map = {str(ans.question_id): ans for ans in application.form_answers} - - questions_statement = select(Forms_Question) - questions = session.exec(questions_statement).all() - question_map = {str(q.question_id): q for q in questions} - - bulk_updates = [] - for update in forms_batchupdate: - form_answer = answer_map.get(update.question_id) - if form_answer: - question = question_map.get(update.question_id) - if question: - is_prefilled_field = QuestionLabel.is_prefilled_field(question.label) - has_existing_value = form_answer.answer and form_answer.answer.strip() - is_empty_update = not update.answer or not update.answer.strip() - - if is_prefilled_field and has_existing_value and is_empty_update: - continue - - try: - validate_profile_url(question.label, update.answer) - except ValueError as error: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=str(error), - ) from error - - bulk_updates.append({"id": form_answer.id, "answer": update.answer}) - else: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Invalid question_id: {update.question_id}", - ) - - try: - if bulk_updates: - session.bulk_update_mappings(Forms_Answer, bulk_updates) - - application.updated_at = datetime.now(timezone.utc) - session.add(application) - - session.commit() - session.refresh(application) - except Exception as e: - session.rollback() - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to save answers: {str(e)}", - ) - - return {"message": "Answers saved successfully", "updated_count": len(bulk_updates)} + return save_application_answers(session, current_user, forms_batchupdate) @router.post("/resume") @@ -230,92 +66,7 @@ def upload_resume( current_user: Annotated[Account_User, Depends(get_current_user)], session: SessionDep, ): - if not is_valid_submission_time(session, current_user): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, detail="Submission is closed" - ) - - if not file.filename: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required" - ) - - file_ext = Path(file.filename).suffix.lower() - if file_ext not in ALLOWED_FILE_EXTENSIONS: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail=ALLOWED_FILE_TYPES_MESSAGE - ) - - upload_dir = Path(FileUploadConfig.UPLOAD_DIR) - upload_dir.mkdir(parents=True, exist_ok=True) - - with tempfile.NamedTemporaryFile( - delete=False, dir=upload_dir, suffix=DEFAULT_FILE_EXTENSION - ) as tmp: - temp_path = tmp.name - with open(temp_path, "wb") as out: - bytes_written = 0 - - while chunk := file.file.read(FileUploadConfig.CHUNK_SIZE_BYTES): - bytes_written += len(chunk) - if bytes_written > FileUploadConfig.MAX_FILE_SIZE_BYTES: - os.unlink(temp_path) - raise HTTPException( - status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, - detail="File too large", - ) - out.write(chunk) - - is_valid, error_msg = _validate_pdf(temp_path, file.filename) - if not is_valid: - os.unlink(temp_path) - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=error_msg) - - if current_user.application is None: - current_user.application = create_application(current_user, session) - - old = current_user.application.form_answersfile - if old and old.file_path: - try: - Path(old.file_path).unlink(missing_ok=True) - except Exception: - pass - - final_name = f"{uuid4()}{DEFAULT_FILE_EXTENSION}" - final_path = upload_dir / final_name - shutil.move(temp_path, final_path) - - answer_file = current_user.application.form_answersfile - if not answer_file: - try: - final_path.unlink(missing_ok=True) - except Exception: - pass - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail="Missing resume model" - ) - - try: - answer_file.original_filename = file.filename - answer_file.file_path = str(final_path) - current_user.application.updated_at = datetime.now(timezone.utc) - - session.add(answer_file) - session.add(current_user.application) - session.commit() - session.refresh(answer_file) - except Exception as e: - session.rollback() - try: - final_path.unlink(missing_ok=True) - except Exception: - pass - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to save resume: {str(e)}", - ) - - return answer_file.original_filename + return store_resume(session, current_user, file) @router.post("/submission", status_code=status.HTTP_201_CREATED) @@ -323,121 +74,7 @@ def submit( current_user: Annotated[Account_User, Depends(get_current_user)], session: SessionDep, ): - if not is_valid_submission_time(session, current_user): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Submission is currently closed", - ) - - application = current_user.application - if application is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail="Application not found" - ) - - questions_statement = select(Forms_Question) - all_questions = session.exec(questions_statement).all() - question_map = {str(q.question_id): q for q in all_questions} - question_labels = {question.label for question in all_questions} - superseded_question_labels = { - "Race/Ethnicity": "Race/Ethnicity (Select all that apply)", - } - - for answer in application.form_answers: - selected_question = question_map.get(str(answer.question_id)) - if ( - selected_question - and selected_question.label in superseded_question_labels - and superseded_question_labels[selected_question.label] in question_labels - ): - continue - if selected_question and selected_question.required: - if ( - answer.answer is None - or answer.answer.strip() == "" - or answer.answer == "false" - ): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Required field not answered: {selected_question.label}", - ) - answer_file = application.form_answersfile - if answer_file is None or answer_file.original_filename is None: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail="Resume is required" - ) - - lock_statement = ( - select(Forms_HackathonApplicant) - .where( - Forms_HackathonApplicant.application_id - == application.application_id - ) - .with_for_update() - ) - hacker_applicant = session.exec(lock_statement).first() - - if not hacker_applicant: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Application not found", - ) - - if hacker_applicant.is_already_submitted(): - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, detail="Application already submitted" - ) - - if not hacker_applicant.can_submit_application(): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="User not in valid state to submit", - ) - - is_walk_in_submission = False - current_status = hacker_applicant.status - - if current_status == StatusEnum.APPLYING: - hacker_applicant.status = StatusEnum.APPLIED - elif current_status == StatusEnum.WALK_IN: - hacker_applicant.status = StatusEnum.WALK_IN_SUBMITTED - is_walk_in_submission = True - if not application.is_draft: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail="Application has already been submitted", - ) - else: - application.is_draft = False - application.updated_at = datetime.now(timezone.utc) - - try: - session.add(hacker_applicant) - session.add(application) - session.commit() - session.refresh(hacker_applicant) - session.refresh(application) - except Exception as e: - session.rollback() - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to submit application: {str(e)}", - ) - - if is_walk_in_submission: - application_id = str(application.application_id) - user_full_name = current_user.full_name - send_rsvp(current_user.email, user_full_name, application_id) - else: - send_email( - EmailTemplate.CONFIRMATION, - current_user.email, - EmailSubject.CONFIRMATION, - EmailMessage.CONFIRMATION, - {}, - ) - - return "Success" + return submit_application(session, current_user) @router.get("/submission-time") @@ -447,15 +84,9 @@ def submission_time(session: SessionDep): @router.get("/registration-timerange", response_model=Forms_Form) def get_reg_time_range(session: SessionDep) -> Forms_Form: - from datetime import timedelta - - from app.cache import cache - def fetch_time_range(): return session.exec(select(Forms_Form)).first() return cache.get_or_set( - key="registration_timerange", - factory_func=fetch_time_range, - ttl=timedelta(minutes=5), + "registration_timerange", fetch_time_range, timedelta(minutes=5) ) diff --git a/app/services/admin_applications.py b/app/services/admin_applications.py new file mode 100644 index 0000000..999b4a3 --- /dev/null +++ b/app/services/admin_applications.py @@ -0,0 +1,273 @@ +import re +from datetime import datetime, timedelta, timezone +from pathlib import Path +from uuid import UUID + +from fastapi import HTTPException, status +from sqlalchemy import and_, func, or_ +from sqlalchemy.orm import aliased +from sqlmodel import Session, col, select + +from app.cache import cache +from app.core.orm import eager_load +from app.models.constants import ( + DEFAULT_FILE_EXTENSION, + QuestionLabel, + RankingSort, + SortOrder, +) +from app.models.forms import ( + Forms_Answer, + Forms_AnswerFile, + Forms_Application, + Forms_HackathonApplicant, + Forms_Question, + StatusEnum, +) +from app.models.judging import JudgingApplicationScore +from app.models.user import Account_User +from app.services.email import send_rsvp + + +def sanitize_filename(filename: str) -> str: + filename = Path(filename).name.replace("\x00", "") + filename = filename.replace("..", "").replace("./", "").replace("../", "") + filename = re.sub(r"[^\w\s\-.]", "", filename).lstrip(".") + if len(filename) > 255: + parts = filename.rsplit(".", 1) + filename = ( + parts[0][: 254 - len(parts[1])] + "." + parts[1] + if len(parts) == 2 + else filename[:255] + ) + return filename if filename and not filename.isspace() else f"file{DEFAULT_FILE_EXTENSION}" + + +def get_resume_metadata( + session: Session, application_id: UUID +) -> tuple[Path, str]: + resume = session.exec( + select(Forms_AnswerFile).where( + Forms_AnswerFile.application_id == application_id + ) + ).first() + if not resume or not resume.file_path: + raise HTTPException(status_code=404, detail="Resume not found") + path = Path(resume.file_path) + if not path.exists() or not path.is_file(): + raise HTTPException(status_code=404, detail="File not found on disk") + return path, sanitize_filename( + resume.original_filename or f"resume{DEFAULT_FILE_EXTENSION}" + ) + + +def get_application_detail(session: Session, application_id: UUID) -> dict: + application = session.exec( + select(Forms_Application).where( + Forms_Application.application_id == application_id + ) + ).first() + if application is None: + raise HTTPException(status_code=404, detail="Application not found") + return { + "application": application, + "form_answers": application.form_answers, + "form_answersfile": application.form_answersfile.original_filename + if application.form_answersfile + else None, + } + + +def list_applications( + session: Session, + *, + offset: int, + limit: int, + search: str, + level_of_study: str, + gender: str, + school: str, + date_sort: SortOrder | None, + ranking_sort: RankingSort | None, + application_status: StatusEnum | None, +) -> dict: + def fetch_questions() -> dict[str, Forms_Question]: + questions = session.exec( + select(Forms_Question).where( + col(Forms_Question.label).in_( + [ + QuestionLabel.CURRENT_LEVEL_OF_STUDY.value, + QuestionLabel.GENDER.value, + QuestionLabel.SCHOOL_NAME.value, + ] + ) + ) + ).all() + return {question.label: question for question in questions} + + question_map = cache.get_or_set( + "admin_filter_questions", fetch_questions, timedelta(minutes=10) + ) + level_question = question_map.get(QuestionLabel.CURRENT_LEVEL_OF_STUDY.value) + gender_question = question_map.get(QuestionLabel.GENDER.value) + school_question = question_map.get(QuestionLabel.SCHOOL_NAME.value) + + level_answer = aliased(Forms_Answer) + gender_answer = aliased(Forms_Answer) + school_answer = aliased(Forms_Answer) + statement = ( + select( + Account_User, + Forms_Application, + Forms_HackathonApplicant, + col(level_answer.answer).label("level_of_study_answer"), + col(gender_answer.answer).label("gender_answer"), + col(school_answer.answer).label("school_answer"), + col(JudgingApplicationScore.mu).label("ranking_mu"), + col(JudgingApplicationScore.sigma_sq).label("ranking_sigma_sq"), + col(JudgingApplicationScore.comparison_count).label( + "ranking_comparison_count" + ), + ) + .where(Account_User.is_active, Account_User.application is not None) + .join(Forms_Application, Account_User.uid == Forms_Application.uid) + .join( + Forms_HackathonApplicant, + Forms_Application.application_id + == Forms_HackathonApplicant.application_id, + ) + .outerjoin( + JudgingApplicationScore, + JudgingApplicationScore.application_id + == Forms_Application.application_id, + ) + ) + + for question, answer_model in ( + (level_question, level_answer), + (gender_question, gender_answer), + (school_question, school_answer), + ): + if question: + statement = statement.outerjoin( + answer_model, + and_( + answer_model.application_id == Forms_Application.application_id, + answer_model.question_id == question.question_id, + ), + ) + + if search: + pattern = f"%{search}%" + statement = statement.where( + or_( + col(Account_User.first_name).ilike(pattern), + col(Account_User.last_name).ilike(pattern), + col(Account_User.email).ilike(pattern), + (col(Account_User.first_name) + " " + col(Account_User.last_name)).ilike( + pattern + ), + ) + ) + if application_status: + statement = statement.where( + Forms_HackathonApplicant.status == application_status + ) + if level_of_study and level_question: + statement = statement.where( + func.lower(level_answer.answer) == level_of_study.lower() + ) + if gender and gender_question: + statement = statement.where( + func.lower(gender_answer.answer) == gender.lower() + ) + if school and school_question: + statement = statement.where( + col(school_answer.answer).isnot(None), + school_answer.answer != "", + func.lower(school_answer.answer) == school.lower(), + ) + + if ranking_sort: + ranking_column = col(JudgingApplicationScore.mu) + statement = statement.order_by( + (ranking_column.desc() if ranking_sort == RankingSort.HIGHEST else ranking_column.asc()).nulls_last() + ) + elif date_sort: + date_column = col(Forms_Application.updated_at) + statement = statement.order_by( + date_column.asc() if date_sort == SortOrder.OLDEST else date_column.desc() + ) + + results = session.exec(statement.offset(offset).limit(limit)).all() + applications = [ + { + "first_name": user.first_name, + "last_name": user.last_name, + "email": user.email, + "status": applicant.status, + "app_id": applicant.application_id, + "created_at": application.created_at, + "updated_at": application.updated_at, + "level_of_study": level, + "gender": gender_value, + "school": school_value, + "ranking_mu": ranking_mu, + "ranking_sigma_sq": ranking_sigma_sq, + "ranking_comparison_count": comparison_count or 0, + } + for ( + user, + application, + applicant, + level, + gender_value, + school_value, + ranking_mu, + ranking_sigma_sq, + comparison_count, + ) in results + ] + return {"application": applications, "offset": offset, "limit": limit} + + +def update_application_status( + session: Session, + application_id: str, + new_status: StatusEnum, +) -> dict: + result = session.exec( + select(Forms_Application, Account_User) + .join(Account_User, Forms_Application.uid == Account_User.uid) + .where(Forms_Application.application_id == application_id) + .options(eager_load(Forms_Application.hackathonapplicant)) + ).first() + if not result: + raise HTTPException(status_code=404, detail="Application not found") + application, user = result + applicant = application.hackathonapplicant + if applicant is None: + raise HTTPException(status_code=404, detail="Applicant status not found") + + try: + applicant.status = new_status.value + application.updated_at = datetime.now(timezone.utc) + session.add(applicant) + session.add(application) + session.commit() + session.refresh(applicant) + session.refresh(application) + if new_status == StatusEnum.ACCEPTED: + send_rsvp(user.email, user.full_name, application_id) + except Exception as error: + session.rollback() + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to update application status: {error}", + ) from error + + return { + "application_id": application_id, + "new_status": new_status.value, + "updated_at": application.updated_at, + } diff --git a/app/services/bulk_email.py b/app/services/bulk_email.py new file mode 100644 index 0000000..b3a2ba0 --- /dev/null +++ b/app/services/bulk_email.py @@ -0,0 +1,119 @@ +import logging +from concurrent.futures import ThreadPoolExecutor + +from sqlmodel import Session, func, select + +from app.config import EmailConfig +from app.models.forms import Forms_Application, Forms_HackathonApplicant +from app.models.requests import BulkEmailRequest +from app.models.user import Account_User +from app.services.email import send_email + +logger = logging.getLogger(__name__) + + +def send_batch_email( + users_data: list[dict], + template_path: str, + subject: str, + text_body: str, + base_context: dict, +) -> None: + total = len(users_data) + successful = 0 + failures: list[dict] = [] + + logger.info( + "Starting bulk email send: %s recipients, subject='%s', template='%s', " + "concurrency=%s, chunk_size=%s", + total, + subject, + template_path, + EmailConfig.BULK_MAX_CONCURRENT, + EmailConfig.BULK_CHUNK_SIZE, + ) + + def send_one(user_data: dict) -> tuple[bool, str, dict]: + email = user_data.get("email", "unknown") + try: + email_context = base_context.copy() if base_context else {} + email_context.update(user_data) + status_code, response = send_email( + template_path, email, subject, text_body, email_context + ) + if status_code == 200: + return True, email, {} + return False, email, { + "email": email, + "reason": f"Status {status_code}", + "response": response, + } + except Exception as error: + return False, email, {"email": email, "reason": str(error)} + + for index in range(0, total, EmailConfig.BULK_CHUNK_SIZE): + chunk = users_data[index : index + EmailConfig.BULK_CHUNK_SIZE] + with ThreadPoolExecutor( + max_workers=EmailConfig.BULK_MAX_CONCURRENT + ) as executor: + results = list(executor.map(send_one, chunk)) + for success, email, error_info in results: + if success: + successful += 1 + logger.debug("Email sent successfully to %s", email) + else: + failures.append(error_info) + logger.warning("Email send failed to %s: %s", email, error_info) + + logger.info( + "Bulk email send complete: %s/%s successful, %s/%s failed", + successful, + total, + len(failures), + total, + ) + + +def get_bulk_email_recipients( + session: Session, request: BulkEmailRequest +) -> tuple[int, list[dict]]: + base_query = ( + select(Account_User) + .join(Forms_Application, Account_User.uid == Forms_Application.uid) + .join( + Forms_HackathonApplicant, + Forms_Application.application_id + == Forms_HackathonApplicant.application_id, + ) + .where( + Account_User.is_active, + Forms_HackathonApplicant.status == request.status, + ) + ) + total = session.exec( + select(func.count()).select_from(base_query.subquery()) + ).one() + if total == 0: + return 0, [] + + rows = session.exec( + select( + Account_User.first_name, + Account_User.last_name, + Account_User.email, + ) + .join(Forms_Application, Account_User.uid == Forms_Application.uid) + .join( + Forms_HackathonApplicant, + Forms_Application.application_id + == Forms_HackathonApplicant.application_id, + ) + .where( + Account_User.is_active, + Forms_HackathonApplicant.status == request.status, + ) + ).all() + return total, [ + {"first_name": row[0], "last_name": row[1], "email": row[2]} + for row in rows + ] diff --git a/app/services/form_workflow.py b/app/services/form_workflow.py new file mode 100644 index 0000000..4f16b3b --- /dev/null +++ b/app/services/form_workflow.py @@ -0,0 +1,191 @@ +from datetime import datetime, timezone + +from fastapi import HTTPException, status +from sqlmodel import Session, select + +from app.core.orm import eager_load +from app.models.constants import EmailMessage, EmailSubject, EmailTemplate, QuestionLabel +from app.models.forms import ( + Forms_Answer, + Forms_AnswerUpdate, + Forms_Application, + Forms_HackathonApplicant, + Forms_Question, + StatusEnum, +) +from app.models.user import Account_User +from app.services.applications import create_application, is_valid_submission_time +from app.services.email import send_email, send_rsvp +from app.validators import validate_profile_url + + +def get_or_create_application(session: Session, user: Account_User) -> dict: + if user.application is None: + if not is_valid_submission_time(session, user): + raise HTTPException(status_code=404, detail="Submitting outside submission time") + application = create_application(user, session) + else: + application = session.exec( + select(Forms_Application) + .where(Forms_Application.uid == user.uid) + .options( + eager_load(Forms_Application.form_answers), + eager_load(Forms_Application.form_answersfile), + eager_load(Forms_Application.hackathonapplicant), + ) + ).first() + if application is None: + raise HTTPException(status_code=404, detail="Application not found") + return { + "application": application, + "form_answers": application.form_answers, + "form_answersfile": application.form_answersfile.original_filename + if application.form_answersfile + else None, + } + + +def save_answers( + session: Session, + user: Account_User, + updates: list[Forms_AnswerUpdate], +) -> dict: + if not is_valid_submission_time(session, user): + raise HTTPException(status_code=403, detail="Submission is currently closed") + if user.application is None: + user.application = create_application(user, session) + + application = session.exec( + select(Forms_Application) + .where(Forms_Application.uid == user.uid) + .options(eager_load(Forms_Application.form_answers)) + ).first() + if application is None: + raise HTTPException(status_code=404, detail="Application not found") + + answers = {str(answer.question_id): answer for answer in application.form_answers} + questions = { + str(question.question_id): question + for question in session.exec(select(Forms_Question)).all() + } + bulk_updates: list[dict] = [] + for update in updates: + answer = answers.get(update.question_id) + if answer is None: + raise HTTPException( + status_code=400, detail=f"Invalid question_id: {update.question_id}" + ) + question = questions.get(update.question_id) + if question: + if ( + QuestionLabel.is_prefilled_field(question.label) + and answer.answer + and answer.answer.strip() + and (not update.answer or not update.answer.strip()) + ): + continue + try: + validate_profile_url(question.label, update.answer) + except ValueError as error: + raise HTTPException(status_code=400, detail=str(error)) from error + bulk_updates.append({"id": answer.id, "answer": update.answer}) + + try: + if bulk_updates: + session.bulk_update_mappings(Forms_Answer, bulk_updates) + application.updated_at = datetime.now(timezone.utc) + session.add(application) + session.commit() + session.refresh(application) + except Exception as error: + session.rollback() + raise HTTPException( + status_code=500, detail=f"Failed to save answers: {error}" + ) from error + return {"message": "Answers saved successfully", "updated_count": len(bulk_updates)} + + +def submit_application(session: Session, user: Account_User) -> str: + if not is_valid_submission_time(session, user): + raise HTTPException(status_code=403, detail="Submission is currently closed") + application = user.application + if application is None: + raise HTTPException(status_code=404, detail="Application not found") + + all_questions = session.exec(select(Forms_Question)).all() + questions = {str(question.question_id): question for question in all_questions} + labels = {question.label for question in all_questions} + superseded_labels = { + "Race/Ethnicity": "Race/Ethnicity (Select all that apply)" + } + for answer in application.form_answers: + question = questions.get(str(answer.question_id)) + if ( + question + and question.label in superseded_labels + and superseded_labels[question.label] in labels + ): + continue + if question and question.required and ( + answer.answer is None + or answer.answer.strip() == "" + or answer.answer == "false" + ): + raise HTTPException( + status_code=400, + detail=f"Required field not answered: {question.label}", + ) + if ( + application.form_answersfile is None + or application.form_answersfile.original_filename is None + ): + raise HTTPException(status_code=400, detail="Resume is required") + + applicant = session.exec( + select(Forms_HackathonApplicant) + .where( + Forms_HackathonApplicant.application_id == application.application_id + ) + .with_for_update() + ).first() + if applicant is None: + raise HTTPException(status_code=404, detail="Application not found") + if applicant.is_already_submitted(): + raise HTTPException(status_code=409, detail="Application already submitted") + if not applicant.can_submit_application(): + raise HTTPException(status_code=403, detail="User not in valid state to submit") + + walk_in = applicant.status == StatusEnum.WALK_IN + if applicant.status == StatusEnum.APPLYING: + applicant.status = StatusEnum.APPLIED + elif walk_in: + applicant.status = StatusEnum.WALK_IN_SUBMITTED + if not application.is_draft: + raise HTTPException(status_code=409, detail="Application has already been submitted") + application.is_draft = False + application.updated_at = datetime.now(timezone.utc) + + try: + session.add(applicant) + session.add(application) + session.commit() + session.refresh(applicant) + session.refresh(application) + except Exception as error: + session.rollback() + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to submit application: {error}", + ) from error + + if walk_in: + send_rsvp(user.email, user.full_name, str(application.application_id)) + else: + send_email( + EmailTemplate.CONFIRMATION, + user.email, + EmailSubject.CONFIRMATION, + EmailMessage.CONFIRMATION, + {}, + ) + return "Success" diff --git a/app/services/resume_uploads.py b/app/services/resume_uploads.py new file mode 100644 index 0000000..2813e5e --- /dev/null +++ b/app/services/resume_uploads.py @@ -0,0 +1,132 @@ +import shutil +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +from fastapi import HTTPException, UploadFile, status +from pypdf import PdfReader +from sqlmodel import Session + +from app.config import FileUploadConfig +from app.models.constants import ( + ALLOWED_FILE_EXTENSIONS, + ALLOWED_FILE_TYPES_MESSAGE, + DEFAULT_FILE_EXTENSION, + MAX_ERROR_MESSAGE_LENGTH, + MIN_PDF_PAGES, + PDF_EMBEDDED_FILES_ERROR, + PDF_ENCRYPTED_ERROR, + PDF_JAVASCRIPT_ERROR, + PDF_NO_PAGES_ERROR, +) +from app.models.user import Account_User +from app.services.applications import create_application, is_valid_submission_time + + +def validate_pdf(filepath: str, filename: str) -> tuple[bool, str]: + if Path(filename).suffix.lower() not in ALLOWED_FILE_EXTENSIONS: + return False, ALLOWED_FILE_TYPES_MESSAGE + try: + reader = PdfReader(filepath) + if reader.is_encrypted: + return False, PDF_ENCRYPTED_ERROR + if len(reader.pages) < MIN_PDF_PAGES: + return False, PDF_NO_PAGES_ERROR + + def contains(forbidden_keys: set[str], value) -> bool: + if isinstance(value, dict): + return any( + key in forbidden_keys + or (isinstance(child, (dict, list)) and contains(forbidden_keys, child)) + for key, child in value.items() + ) + if isinstance(value, list): + return any( + isinstance(child, (dict, list)) and contains(forbidden_keys, child) + for child in value + ) + return False + + root = reader.trailer.get("/Root") + if contains({"/JavaScript", "/JS", "/AA", "/OpenAction"}, root): + return False, PDF_JAVASCRIPT_ERROR + if contains({"/EmbeddedFile", "/EmbeddedFiles", "/AF"}, root): + return False, PDF_EMBEDDED_FILES_ERROR + except Exception as error: + return False, f"Invalid PDF: {str(error)[:MAX_ERROR_MESSAGE_LENGTH]}" + return True, "" + + +def upload_resume( + session: Session, current_user: Account_User, file: UploadFile +) -> str: + if not is_valid_submission_time(session, current_user): + raise HTTPException(status_code=403, detail="Submission is closed") + if not file.filename: + raise HTTPException(status_code=400, detail="Filename is required") + if Path(file.filename).suffix.lower() not in ALLOWED_FILE_EXTENSIONS: + raise HTTPException(status_code=400, detail=ALLOWED_FILE_TYPES_MESSAGE) + + upload_dir = Path(FileUploadConfig.UPLOAD_DIR) + upload_dir.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + delete=False, dir=upload_dir, suffix=DEFAULT_FILE_EXTENSION + ) as temporary_file: + temporary_path = temporary_file.name + final_path: Path | None = None + + try: + with open(temporary_path, "wb") as output: + bytes_written = 0 + while chunk := file.file.read(FileUploadConfig.CHUNK_SIZE_BYTES): + bytes_written += len(chunk) + if bytes_written > FileUploadConfig.MAX_FILE_SIZE_BYTES: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail="File too large", + ) + output.write(chunk) + + valid, error = validate_pdf(temporary_path, file.filename) + if not valid: + raise HTTPException(status_code=400, detail=error) + + if current_user.application is None: + current_user.application = create_application(current_user, session) + application = current_user.application + old_resume = application.form_answersfile + if old_resume and old_resume.file_path: + try: + Path(old_resume.file_path).unlink(missing_ok=True) + except Exception: + pass + + final_path = upload_dir / f"{uuid4()}{DEFAULT_FILE_EXTENSION}" + shutil.move(temporary_path, final_path) + answer_file = application.form_answersfile + if not answer_file: + final_path.unlink(missing_ok=True) + raise HTTPException(status_code=400, detail="Missing resume model") + + answer_file.original_filename = file.filename + answer_file.file_path = str(final_path) + application.updated_at = datetime.now(timezone.utc) + session.add(answer_file) + session.add(application) + session.commit() + session.refresh(answer_file) + return answer_file.original_filename + except HTTPException: + Path(temporary_path).unlink(missing_ok=True) + if final_path: + final_path.unlink(missing_ok=True) + raise + except Exception as error: + session.rollback() + Path(temporary_path).unlink(missing_ok=True) + if final_path: + final_path.unlink(missing_ok=True) + raise HTTPException( + status_code=500, detail=f"Failed to save resume: {error}" + ) from error diff --git a/tests/unit/test_admin_helpers.py b/tests/unit/test_admin_helpers.py index dbfd3b3..09c352f 100644 --- a/tests/unit/test_admin_helpers.py +++ b/tests/unit/test_admin_helpers.py @@ -1,6 +1,7 @@ from importlib import import_module account = import_module("app.routers.admin.account") +bulk_email = import_module("app.services.bulk_email") def test_filename_sanitization(): @@ -16,8 +17,8 @@ def test_batch_email_success_failure_and_exception(monkeypatch): def fake_send(*_args, **_kwargs): return next(responses) - monkeypatch.setattr(account, "send_email", fake_send) - account.send_batch_email( + monkeypatch.setattr(bulk_email, "send_email", fake_send) + bulk_email.send_batch_email( [{"email": "ok@example.com"}, {"email": "bad@example.com"}], "template", "subject", @@ -28,7 +29,7 @@ def fake_send(*_args, **_kwargs): def exploding(*_args, **_kwargs): raise RuntimeError("provider unavailable") - monkeypatch.setattr(account, "send_email", exploding) - account.send_batch_email( + monkeypatch.setattr(bulk_email, "send_email", exploding) + bulk_email.send_batch_email( [{}], "template", "subject", "body", {"shared": True} ) diff --git a/tests/unit/test_pdf_validation.py b/tests/unit/test_pdf_validation.py index ca1147a..615393a 100644 --- a/tests/unit/test_pdf_validation.py +++ b/tests/unit/test_pdf_validation.py @@ -2,39 +2,39 @@ from importlib import import_module -forms = import_module("app.routers.forms") +resume_uploads = import_module("app.services.resume_uploads") def test_pdf_validation_branches(monkeypatch, tmp_path): path = tmp_path / "resume.pdf" path.write_bytes(b"pdf") - assert forms._validate_pdf(str(path), "resume.txt")[0] is False + assert resume_uploads.validate_pdf(str(path), "resume.txt")[0] is False monkeypatch.setattr( - forms, + resume_uploads, "PdfReader", lambda _path: SimpleNamespace(is_encrypted=True, pages=[1], trailer={}), ) - assert forms._validate_pdf(str(path), "resume.pdf")[0] is False + assert resume_uploads.validate_pdf(str(path), "resume.pdf")[0] is False monkeypatch.setattr( - forms, + resume_uploads, "PdfReader", lambda _path: SimpleNamespace(is_encrypted=False, pages=[], trailer={}), ) - assert forms._validate_pdf(str(path), "resume.pdf")[0] is False + assert resume_uploads.validate_pdf(str(path), "resume.pdf")[0] is False for root in ({"nested": [{"/JS": "bad"}]}, {"nested": {"/EmbeddedFile": "bad"}}): monkeypatch.setattr( - forms, + resume_uploads, "PdfReader", lambda _path, root=root: SimpleNamespace( is_encrypted=False, pages=[1], trailer={"/Root": root} ), ) - assert forms._validate_pdf(str(path), "resume.pdf")[0] is False + assert resume_uploads.validate_pdf(str(path), "resume.pdf")[0] is False - monkeypatch.setattr(forms, "PdfReader", lambda _path: (_ for _ in ()).throw(ValueError("bad"))) - valid, error = forms._validate_pdf(str(path), "resume.pdf") + monkeypatch.setattr(resume_uploads, "PdfReader", lambda _path: (_ for _ in ()).throw(ValueError("bad"))) + valid, error = resume_uploads.validate_pdf(str(path), "resume.pdf") assert valid is False assert error.startswith("Invalid PDF")