Refactor routes

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,
+    }

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
+    ]