Spaces:
Sleeping
Sleeping
| import os | |
| import io | |
| import json | |
| from typing import Optional | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| import pandas as pd | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import HTMLResponse | |
| from pydantic import BaseModel | |
| from huggingface_hub import HfApi, hf_hub_download | |
| # ── Config ──────────────────────────────────────────────────────────────────── | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| HF_REPO_ID = os.environ.get("HF_REPO_ID") | |
| VACANCY_FILE = "hostel_vacancy.xlsx" | |
| CASTE_FILE = "caste.xlsx" | |
| MERIT_FILE = "Merit Data.xlsx" | |
| MASTERS_FILE = "AI Masters Data 13052026 (1).xlsx" | |
| ALLOCATIONS_FILE = "hostel_allocations.json" | |
| # ── In-memory state ─────────────────────────────────────────────────────────── | |
| vacancy_df: pd.DataFrame = pd.DataFrame() | |
| merit_df: pd.DataFrame = pd.DataFrame() | |
| caste_map: dict[int, str] = {} # CasteCategoryID → name | |
| caste_detail_map: dict[int, dict]= {} # caste_id → {name, category_id, category_name} | |
| hostel_map: dict[int, dict]= {} # HostelID → {name, type, district, division} | |
| district_map: dict[int, str] = {} # Districtcode → Districtname | |
| course_map = {1: "Undergraduate", 2: "Postgraduate"} | |
| gender_map = {1: "Male", 2: "Female"} | |
| hostel_type_map = {1: "Boys", 2: "Girls", 0: "Other"} # 1=Boys, 2=Girls per spec | |
| allocations: list[dict] = [] | |
| # ── HF helpers ──────────────────────────────────────────────────────────────── | |
| hf_api = HfApi() | |
| _repo_file_cache: dict[str, str] = {} # lowercase → actual filename | |
| def _build_file_cache(): | |
| global _repo_file_cache | |
| try: | |
| files = list(hf_api.list_repo_files(repo_id=HF_REPO_ID, repo_type="dataset", token=HF_TOKEN)) | |
| _repo_file_cache = {f.lower(): f for f in files} | |
| print(f"HF repo files: {list(_repo_file_cache.values())}") | |
| except Exception as e: | |
| print(f"Warning: could not list repo files: {e}") | |
| def hf_download(filename: str) -> bytes | None: | |
| # Try exact name first | |
| try: | |
| path = hf_hub_download( | |
| repo_id=HF_REPO_ID, filename=filename, | |
| repo_type="dataset", token=HF_TOKEN, | |
| ) | |
| with open(path, "rb") as f: | |
| return f.read() | |
| except Exception: | |
| pass | |
| # Build cache and try case-insensitive match | |
| if not _repo_file_cache: | |
| _build_file_cache() | |
| actual = _repo_file_cache.get(filename.lower()) | |
| if actual and actual != filename: | |
| print(f"Case-insensitive match: '{filename}' -> '{actual}'") | |
| try: | |
| path = hf_hub_download( | |
| repo_id=HF_REPO_ID, filename=actual, | |
| repo_type="dataset", token=HF_TOKEN, | |
| ) | |
| with open(path, "rb") as f: | |
| return f.read() | |
| except Exception as e: | |
| print(f"Failed to download '{actual}': {e}") | |
| print(f"File not found: '{filename}'. Available: {list(_repo_file_cache.values())}") | |
| return None | |
| def hf_upload_bytes(content: bytes, filename: str): | |
| hf_api.upload_file( | |
| path_or_fileobj=io.BytesIO(content), | |
| path_in_repo=filename, | |
| repo_id=HF_REPO_ID, | |
| repo_type="dataset", | |
| token=HF_TOKEN, | |
| ) | |
| def clean_cols(df: pd.DataFrame) -> pd.DataFrame: | |
| df.columns = [c.lstrip("# ").strip() for c in df.columns] | |
| return df | |
| # ── Loaders ─────────────────────────────────────────────────────────────────── | |
| def load_vacancy() -> pd.DataFrame: | |
| raw = hf_download(VACANCY_FILE) | |
| if raw is None: | |
| raise RuntimeError(f"{VACANCY_FILE} not found in HF repo.") | |
| df = clean_cols(pd.read_excel(io.BytesIO(raw))) | |
| return df | |
| def load_merit() -> pd.DataFrame: | |
| raw = hf_download(MERIT_FILE) | |
| if raw is None: | |
| raise RuntimeError(f"{MERIT_FILE} not found in HF repo.") | |
| df = clean_cols(pd.read_excel(io.BytesIO(raw))) | |
| return df | |
| def load_masters() -> dict: | |
| raw = hf_download(MASTERS_FILE) | |
| if raw is None: | |
| raise RuntimeError(f"{MASTERS_FILE} not found in HF repo.") | |
| sheets = pd.read_excel(io.BytesIO(raw), sheet_name=None) | |
| return {k: clean_cols(v) for k, v in sheets.items()} | |
| def load_caste_file() -> dict[int, str]: | |
| raw = hf_download(CASTE_FILE) | |
| if raw is None: | |
| return {} | |
| df = clean_cols(pd.read_excel(io.BytesIO(raw))) | |
| return dict(zip(df["CasteCategoryID"].astype(int), df["CasteCategoryName"].str.strip())) | |
| def load_allocations_from_hf() -> list[dict]: | |
| raw = hf_download(ALLOCATIONS_FILE) | |
| return json.loads(raw.decode()) if raw else [] | |
| def save_allocations_to_hf(data: list[dict]): | |
| hf_upload_bytes(json.dumps(data, indent=2).encode(), ALLOCATIONS_FILE) | |
| # ── Lifespan ────────────────────────────────────────────────────────────────── | |
| async def lifespan(app: FastAPI): | |
| global vacancy_df, merit_df, caste_map, caste_detail_map | |
| global hostel_map, district_map, allocations | |
| _build_file_cache() # list repo files once for case-insensitive matching | |
| vacancy_df = load_vacancy() | |
| merit_df = load_merit() | |
| masters = load_masters() | |
| caste_map = load_caste_file() | |
| # District map | |
| dist_df = masters.get("District", pd.DataFrame()) | |
| if not dist_df.empty: | |
| district_map = dict(zip( | |
| dist_df["Districtcode"].astype(int), | |
| dist_df["Districtname"].str.strip() | |
| )) | |
| # Category map (from masters sheet, supplement caste.xlsx) | |
| cat_df = masters.get("Category", pd.DataFrame()) | |
| if not cat_df.empty: | |
| for _, row in cat_df.iterrows(): | |
| cid = int(row["CasteCategoryID"]) | |
| if cid not in caste_map: | |
| caste_map[cid] = str(row["CasteCategoryName"]).strip() | |
| # Caste detail map: caste_id → {name, category_id, category_name} | |
| caste_df = masters.get("Caste", pd.DataFrame()) | |
| if not caste_df.empty: | |
| for _, row in caste_df.iterrows(): | |
| cid = int(row["caste_id"]) | |
| catid = int(row["caste_category_id"]) | |
| caste_detail_map[cid] = { | |
| "caste_id": cid, | |
| "caste_name": str(row["caste_name"]).strip(), | |
| "category_id": catid, | |
| "category_name": caste_map.get(catid, f"Category {catid}"), | |
| } | |
| # Build reverse district name→code map for hostel enrichment | |
| name_to_dist_code: dict[str, int] = {} | |
| for code, name in district_map.items(): | |
| name_to_dist_code[name.lower().strip()] = code | |
| # aliases | |
| name_to_dist_code["ahmednagar"] = name_to_dist_code.get("ahilyanagar", 0) | |
| # Hostel map | |
| hostel_df = masters.get("Hostels", pd.DataFrame()) | |
| if not hostel_df.empty: | |
| for _, row in hostel_df.iterrows(): | |
| raw_id = str(row["HostelID"]).strip() | |
| if not raw_id.isdigit(): | |
| continue # skip bad rows (address in ID column etc.) | |
| hid = int(raw_id) | |
| h_type_int = int(row["HostelType"]) if pd.notna(row.get("HostelType")) else 0 | |
| dist_name = str(row.get("District_in_English", "")).strip() | |
| dist_code = name_to_dist_code.get(dist_name.lower().strip(), 0) | |
| hostel_map[hid] = { | |
| "hostel_id": hid, | |
| "name": str(row.get("HostelName", row.get("HostelName_Old", ""))).strip(), | |
| "type": hostel_type_map.get(h_type_int, "Other"), | |
| "type_code": h_type_int, # 1=Boys, 2=Girls | |
| "district": dist_name, | |
| "district_code":dist_code, | |
| "division": str(row.get("Division_in_English", "")).strip(), | |
| } | |
| allocations = load_allocations_from_hf() | |
| print(f"✓ {len(vacancy_df)} vacancy rows | {len(merit_df)} merit students | " | |
| f"{len(hostel_map)} hostels | {len(caste_map)} categories | " | |
| f"{len(caste_detail_map)} castes | {len(allocations)} allocations") | |
| yield | |
| app = FastAPI(title="Hostel Allocation API", lifespan=lifespan) | |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) | |
| BASE_DIR = Path(__file__).parent | |
| def serve_ui(): | |
| p = BASE_DIR / "index.html" | |
| return HTMLResponse(p.read_text() if p.exists() else "<h2>index.html not found</h2>") | |
| # ── Schemas ─────────────────────────────────────────────────────────────────── | |
| class StudentRequest(BaseModel): | |
| application_no: int | |
| hostel_id: int | |
| course_type_id: int | |
| caste_category_vacancy_id: int | |
| class AllocateResponse(BaseModel): | |
| application_no: int | |
| student_name: str | |
| hostel_id: int | |
| hostel_name: str | |
| course_type_id: int | |
| course_type_name: str | |
| caste_category_vacancy_id: int | |
| caste_category_name: str | |
| marks_percentage: float | |
| merit_rank: int | |
| priority: int | |
| allocation_percentage: float | |
| status: str | |
| message: str | |
| # ── Core helpers ────────────────────────────────────────────────────────────── | |
| def get_vacancy_row(hostel_id: int, course_type_id: int, caste_id: int) -> pd.Series | None: | |
| mask = ( | |
| (vacancy_df["HostelId"] == hostel_id) | |
| & (vacancy_df["CourseTypeId"] == course_type_id) | |
| & (vacancy_df["CasteCategoryVacancyId"] == caste_id) | |
| ) | |
| rows = vacancy_df[mask] | |
| return rows.iloc[0] if not rows.empty else None | |
| def get_merit_student(application_no: int) -> pd.Series | None: | |
| rows = merit_df[merit_df["applicationno"] == application_no] | |
| return rows.iloc[0] if not rows.empty else None | |
| def get_slot_allocations(hostel_id: int, course_type_id: int, caste_id: int) -> list[dict]: | |
| return [ | |
| a for a in allocations | |
| if a["hostel_id"] == hostel_id | |
| and a["course_type_id"] == course_type_id | |
| and a["caste_category_vacancy_id"] == caste_id | |
| and a["status"] in ("allocated", "waitlisted") | |
| ] | |
| def recalculate_priorities(slot_list: list[dict]) -> list[dict]: | |
| slot_list.sort(key=lambda x: (x["marks_percentage"], -x["merit_rank"]), reverse=True) | |
| n = len(slot_list) | |
| for rank, s in enumerate(slot_list, start=1): | |
| s["priority"] = rank | |
| s["allocation_percentage"] = round((n - rank + 1) / n * 100, 2) | |
| return slot_list | |
| def update_global_allocations(updated_slot: list[dict]): | |
| if not updated_slot: | |
| return | |
| h, c, k = updated_slot[0]["hostel_id"], updated_slot[0]["course_type_id"], updated_slot[0]["caste_category_vacancy_id"] | |
| global allocations | |
| allocations = [ | |
| a for a in allocations | |
| if not (a["hostel_id"] == h and a["course_type_id"] == c and a["caste_category_vacancy_id"] == k) | |
| ] | |
| allocations.extend(updated_slot) | |
| def enrich_record(record: dict) -> dict: | |
| r = dict(record) | |
| hid = r.get("hostel_id", 0) | |
| h = hostel_map.get(hid, {}) | |
| r["hostel_name"] = h.get("name", f"Hostel {hid}") | |
| r["course_type_name"] = course_map.get(r.get("course_type_id", 0), "Unknown") | |
| r["caste_category_name"]= caste_map.get(r.get("caste_category_vacancy_id", 0), "Unknown") | |
| return r | |
| # ── Routes ──────────────────────────────────────────────────────────────────── | |
| def health(): | |
| return { | |
| "status": "ok", | |
| "vacancy_rows": len(vacancy_df), | |
| "merit_students": len(merit_df), | |
| "hostels": len(hostel_map), | |
| "caste_categories":len(caste_map), | |
| "allocations": len(allocations), | |
| } | |
| def debug_files(): | |
| """List all files found in the HF dataset repo. Use this to diagnose filename issues.""" | |
| _build_file_cache() | |
| return { | |
| "repo_id": HF_REPO_ID, | |
| "files": list(_repo_file_cache.values()), | |
| "expected": [VACANCY_FILE, CASTE_FILE, MERIT_FILE, MASTERS_FILE, ALLOCATIONS_FILE], | |
| "missing": [ | |
| f for f in [VACANCY_FILE, CASTE_FILE, MERIT_FILE, MASTERS_FILE] | |
| if f.lower() not in _repo_file_cache | |
| ] | |
| } | |
| # ── Masters ─────────────────────────────────────────────────────────────────── | |
| def list_castes(): | |
| return [{"CasteCategoryID": k, "CasteCategoryName": v} for k, v in sorted(caste_map.items())] | |
| def list_hostels(gender: Optional[int] = None, district_code: Optional[int] = None): | |
| """gender: 1=Male->Boys hostels, 2=Female->Girls hostels. district_code = student DistrictID.""" | |
| result = list(hostel_map.values()) | |
| if gender in (1, 2): | |
| result = [h for h in result if h.get("type_code") == gender] | |
| if district_code: | |
| result = [h for h in result if h.get("district_code") == district_code] | |
| return sorted(result, key=lambda x: x["hostel_id"]) | |
| def list_districts(): | |
| return [{"code": k, "name": v} for k, v in sorted(district_map.items())] | |
| def list_course_types(): | |
| return [{"id": k, "name": v} for k, v in course_map.items()] | |
| # ── Student lookup ──────────────────────────────────────────────────────────── | |
| def get_student(application_no: int): | |
| s = get_merit_student(application_no) | |
| if s is None: | |
| raise HTTPException(404, f"Application no {application_no} not found in merit data.") | |
| caste_id = int(s["Caste"]) if pd.notna(s["Caste"]) else None | |
| caste_info = caste_detail_map.get(caste_id, {}) if caste_id else {} | |
| cat_id = caste_info.get("category_id") | |
| dist_id = int(s["DistrictID"]) if pd.notna(s["DistrictID"]) else None | |
| hostel_id = int(s["HostelID"]) if pd.notna(s["HostelID"]) else None | |
| gender_int = int(s["gender"]) if pd.notna(s["gender"]) else 0 | |
| # gender 1=Male→Boys hostels(type 1), gender 2=Female→Girls hostels(type 2) | |
| expected_hostel_type = gender_int # 1 or 2 maps directly | |
| return { | |
| "application_no": int(s["applicationno"]), | |
| "user_id": int(s["UserId"]) if pd.notna(s["UserId"]) else None, | |
| "gender": gender_map.get(gender_int, "Unknown"), | |
| "gender_int": gender_int, | |
| "expected_hostel_type": expected_hostel_type, | |
| "course_type_id": int(s["CourseType"]), | |
| "course_type_name": course_map.get(int(s["CourseType"]), "Unknown"), | |
| "caste_id": caste_id, | |
| "caste_name": caste_info.get("caste_name", "Unknown"), | |
| "caste_category_id": cat_id, | |
| "caste_category_name": caste_map.get(cat_id, "Unknown") if cat_id else "Unknown", | |
| "current_percentage": float(s["CurrentPercentage"]) if pd.notna(s["CurrentPercentage"]) else None, | |
| "past_percentage": float(s["PastPercentage"]) if pd.notna(s["PastPercentage"]) else None, | |
| "merit_rank": int(s["MeritRank"]) if pd.notna(s["MeritRank"]) else None, | |
| "general_merit_rank": int(s["GeneralMeritRank"]) if pd.notna(s["GeneralMeritRank"]) else None, | |
| "district_id": dist_id, | |
| "district_name": district_map.get(dist_id, "Unknown") if dist_id else "Unknown", | |
| "district_code": dist_id, | |
| "is_orphan": bool(s["IsOrphan"]), | |
| "is_disabled": bool(s["IsDisable"]) if pd.notna(s["IsDisable"]) else False, | |
| "is_already_allotted": bool(s["isalloted"]) if pd.notna(s["isalloted"]) else False, | |
| "existing_hostel_id": hostel_id, | |
| "existing_hostel_name": hostel_map.get(hostel_id, {}).get("name", "") if hostel_id else None, | |
| "scrutiny_approved": bool(s["IsScrutinyApproved"]) if pd.notna(s["IsScrutinyApproved"]) else False, | |
| } | |
| # ── Vacancy ─────────────────────────────────────────────────────────────────── | |
| def list_vacancy(hostel_id: Optional[int] = None, caste_id: Optional[int] = None, | |
| gender: Optional[int] = None, district_id: Optional[int] = None): | |
| rows = vacancy_df.to_dict(orient="records") | |
| result = [] | |
| for r in rows: | |
| hid = int(r["HostelId"]) | |
| cid = int(r["CasteCategoryVacancyId"]) | |
| dist = int(r["DistrictId"]) | |
| h = hostel_map.get(hid, {}) | |
| if hostel_id and hid != hostel_id: continue | |
| if caste_id and cid != caste_id: continue | |
| if district_id and dist != district_id: continue | |
| if gender in (1, 2) and h.get("type_code", 0) != gender: continue | |
| result.append({ | |
| **r, | |
| "HostelName": h.get("name", f"Hostel {hid}"), | |
| "HostelType": h.get("type", ""), | |
| "HostelTypeInt": h.get("type_code", 0), | |
| "DistrictName": district_map.get(dist, str(dist)), | |
| "CasteCategoryName": caste_map.get(cid, f"Category {cid}"), | |
| }) | |
| return result | |
| def get_vacancy(hostel_id: int, course_type_id: int, caste_id: int): | |
| row = get_vacancy_row(hostel_id, course_type_id, caste_id) | |
| if row is None: | |
| raise HTTPException(404, "Vacancy record not found.") | |
| h = hostel_map.get(hostel_id, {}) | |
| return { | |
| **row.to_dict(), | |
| "HostelName": h.get("name", f"Hostel {hostel_id}"), | |
| "CasteCategoryName": caste_map.get(caste_id, f"Category {caste_id}"), | |
| } | |
| # ── Allocation chances preview ──────────────────────────────────────────────── | |
| def allocation_chances(hostel_id: int, course_type_id: int, caste_id: int, | |
| marks: float, merit_rank: Optional[int] = None, | |
| gender: Optional[int] = None, district_id: Optional[int] = None): | |
| h_info = hostel_map.get(hostel_id, {}) | |
| h_name = h_info.get("name", f"Hostel {hostel_id}") | |
| h_type = h_info.get("type_code", 0) | |
| # Gender check | |
| if gender in (1, 2) and h_type in (1, 2) and h_type != gender: | |
| gender_label = gender_map.get(gender, "Unknown") | |
| return { | |
| "hostel_name": h_name, "caste_category": caste_map.get(caste_id, ""), | |
| "chance_percentage": 0, | |
| "reason": f"Gender mismatch: {gender_label} student cannot apply to a {h_info.get('type','')} hostel.", | |
| "remaining_seats": 0, "total_seats": 0, "applicants_in_slot": 0, | |
| "warning": "gender_mismatch" | |
| } | |
| # District check | |
| if district_id: | |
| vac_rows = vacancy_df[vacancy_df["HostelId"] == hostel_id] | |
| if not vac_rows.empty: | |
| hostel_dist = int(vac_rows.iloc[0]["DistrictId"]) | |
| if district_id != hostel_dist: | |
| return { | |
| "hostel_name": h_name, "caste_category": caste_map.get(caste_id, ""), | |
| "chance_percentage": 0, | |
| "reason": f"District mismatch: your district ({district_map.get(district_id,district_id)}) " | |
| f"vs hostel district ({district_map.get(hostel_dist,hostel_dist)}).", | |
| "remaining_seats": 0, "total_seats": 0, "applicants_in_slot": 0, | |
| "warning": "district_mismatch" | |
| } | |
| row = get_vacancy_row(hostel_id, course_type_id, caste_id) | |
| if row is None: | |
| raise HTTPException(404, "Vacancy record not found.") | |
| total = int(row["TotalSeat"]) | |
| remaining = int(row["RemainingSeat"]) | |
| cat_name = caste_map.get(caste_id, f"Category {caste_id}") | |
| if total == 0: | |
| return {"chance_percentage": 0, "reason": "No seats in this category.", | |
| "remaining_seats": 0, "total_seats": 0} | |
| slot_list = get_slot_allocations(hostel_id, course_type_id, caste_id) | |
| allocated = [a for a in slot_list if a["status"] == "allocated"] | |
| if len(allocated) < remaining: | |
| chance = 100.0 | |
| reason = f"{remaining - len(allocated)} seat(s) still available under '{cat_name}' at {h_name}." | |
| else: | |
| weakest_marks = min(a["marks_percentage"] for a in allocated) if allocated else 0 | |
| if marks > weakest_marks: | |
| chance = round(marks / 100 * 100, 2) | |
| reason = f"You would displace the weakest applicant ({weakest_marks}% marks) under '{cat_name}'." | |
| else: | |
| above = sum(1 for a in allocated if a["marks_percentage"] >= marks) | |
| chance = round(max(0, (remaining - above) / total * 100), 2) | |
| reason = f"All seats full under '{cat_name}'. Waitlisted; rank depends on your merit." | |
| return { | |
| "hostel_name": h_name, | |
| "caste_category": cat_name, | |
| "chance_percentage": chance, | |
| "reason": reason, | |
| "remaining_seats": remaining, | |
| "total_seats": total, | |
| "applicants_in_slot": len(slot_list), | |
| } | |
| # ── Allocate ────────────────────────────────────────────────────────────────── | |
| def allocate_student(req: StudentRequest): | |
| global allocations | |
| # Lookup student from merit data | |
| s = get_merit_student(req.application_no) | |
| if s is None: | |
| raise HTTPException(404, f"Application no {req.application_no} not found in merit data.") | |
| marks = float(s["CurrentPercentage"]) if pd.notna(s["CurrentPercentage"]) else 0.0 | |
| merit_rank = int(s["MeritRank"]) if pd.notna(s["MeritRank"]) else 9999 | |
| caste_id = int(s["Caste"]) if pd.notna(s["Caste"]) else None | |
| caste_info = caste_detail_map.get(caste_id, {}) if caste_id else {} | |
| # Validate caste category matches | |
| if req.caste_category_vacancy_id not in caste_map: | |
| raise HTTPException(400, f"Invalid caste_category_vacancy_id {req.caste_category_vacancy_id}.") | |
| # Gender validation: Male→Boys(1), Female→Girls(2) | |
| gender_int = int(s["gender"]) if pd.notna(s["gender"]) else 0 | |
| h_info = hostel_map.get(req.hostel_id, {}) | |
| h_name = h_info.get("name", f"Hostel {req.hostel_id}") | |
| h_type_int = h_info.get("type_code", 0) | |
| gender_label = gender_map.get(gender_int, "Unknown") | |
| hostel_type_label = h_info.get("type", "Unknown") | |
| if h_type_int in (1, 2) and h_type_int != gender_int: | |
| raise HTTPException(400, | |
| f"Gender mismatch: {gender_label} student cannot apply to a {hostel_type_label} hostel ({h_name}).") | |
| # District: student district must match hostel district (via vacancy DistrictId) | |
| student_dist_id = int(s["DistrictID"]) if pd.notna(s["DistrictID"]) else None | |
| vac_rows = vacancy_df[vacancy_df["HostelId"] == req.hostel_id] | |
| if student_dist_id and not vac_rows.empty: | |
| hostel_dist_id = int(vac_rows.iloc[0]["DistrictId"]) | |
| if student_dist_id != hostel_dist_id: | |
| student_dist_name = district_map.get(student_dist_id, str(student_dist_id)) | |
| hostel_dist_name = district_map.get(hostel_dist_id, str(hostel_dist_id)) | |
| raise HTTPException(400, | |
| f"District mismatch: Student belongs to '{student_dist_name}' district " | |
| f"but hostel '{h_name}' is in '{hostel_dist_name}' district.") | |
| row = get_vacancy_row(req.hostel_id, req.course_type_id, req.caste_category_vacancy_id) | |
| if row is None: | |
| raise HTTPException(404, "No vacancy record for given hostel/course/caste combination.") | |
| total_seats = int(row["TotalSeat"]) | |
| remaining_seats = int(row["RemainingSeat"]) | |
| cat_name = caste_map.get(req.caste_category_vacancy_id, "Unknown") | |
| if total_seats == 0: | |
| raise HTTPException(400, f"Caste '{cat_name}' has 0 total seats in {h_name}.") | |
| if any(a["application_no"] == req.application_no for a in allocations): | |
| raise HTTPException(400, f"Application {req.application_no} already has an allocation entry.") | |
| slot_list = get_slot_allocations(req.hostel_id, req.course_type_id, req.caste_category_vacancy_id) | |
| allocated_count = sum(1 for a in slot_list if a["status"] == "allocated") | |
| new_record = { | |
| "application_no": req.application_no, | |
| "student_name": f"Applicant {req.application_no}", | |
| "hostel_id": req.hostel_id, | |
| "hostel_name": h_name, | |
| "hostel_type_code": h_info.get("type_code", 0), | |
| "hostel_type": h_info.get("type", "Unknown"), | |
| "hostel_district": h_info.get("district", ""), | |
| "hostel_district_code": h_info.get("district_code", 0), | |
| "course_type_id": req.course_type_id, | |
| "course_type_name": course_map.get(req.course_type_id, "Unknown"), | |
| "caste_category_vacancy_id": req.caste_category_vacancy_id, | |
| "caste_category_name": cat_name, | |
| "marks_percentage": marks, | |
| "merit_rank": merit_rank, | |
| "priority": 0, | |
| "allocation_percentage": 0.0, | |
| "status": "", | |
| "message": "", | |
| } | |
| if allocated_count < remaining_seats: | |
| new_record["status"] = "allocated" | |
| slot_list.append(new_record) | |
| slot_list = recalculate_priorities(slot_list) | |
| new_record["message"] = ( | |
| f"Room allocated at {h_name} under '{cat_name}'. " | |
| f"Ranked #{new_record['priority']} of {len(slot_list)} applicant(s). " | |
| f"Marks: {marks}%, Merit Rank: {merit_rank}." | |
| ) | |
| else: | |
| allocated_students = [a for a in slot_list if a["status"] == "allocated"] | |
| weakest = min(allocated_students, key=lambda x: (x["marks_percentage"], -x["merit_rank"])) \ | |
| if allocated_students else None | |
| if weakest and (marks > weakest["marks_percentage"] or | |
| (marks == weakest["marks_percentage"] and merit_rank < weakest["merit_rank"])): | |
| weakest["status"] = "waitlisted" | |
| weakest["message"] = ( | |
| f"Displaced by application {req.application_no} " | |
| f"(marks: {marks}%, rank: {merit_rank}) under '{cat_name}'." | |
| ) | |
| new_record["status"] = "allocated" | |
| slot_list.append(new_record) | |
| slot_list = recalculate_priorities(slot_list) | |
| new_record["message"] = ( | |
| f"Room allocated at {h_name} under '{cat_name}' by displacing " | |
| f"application {weakest['application_no']} " | |
| f"({weakest['marks_percentage']}% marks). Ranked #{new_record['priority']}." | |
| ) | |
| else: | |
| new_record["status"] = "waitlisted" | |
| slot_list.append(new_record) | |
| slot_list = recalculate_priorities(slot_list) | |
| new_record["message"] = ( | |
| f"No seat available at {h_name} under '{cat_name}'. " | |
| f"Waitlisted at #{new_record['priority']} of {len(slot_list)}." | |
| ) | |
| update_global_allocations(slot_list) | |
| save_allocations_to_hf(allocations) | |
| return AllocateResponse(**new_record) | |
| # ── Allocations list ────────────────────────────────────────────────────────── | |
| def list_allocations( | |
| hostel_id: Optional[int] = None, | |
| course_type_id: Optional[int] = None, | |
| caste_id: Optional[int] = None, | |
| status: Optional[str] = None, | |
| ): | |
| result = allocations | |
| if hostel_id: result = [a for a in result if a["hostel_id"] == hostel_id] | |
| if course_type_id: result = [a for a in result if a["course_type_id"] == course_type_id] | |
| if caste_id: result = [a for a in result if a["caste_category_vacancy_id"] == caste_id] | |
| if status: result = [a for a in result if a["status"] == status] | |
| return result | |
| def remove_allocation(application_no: int): | |
| global allocations | |
| record = next((a for a in allocations if a["application_no"] == application_no), None) | |
| if record is None: | |
| raise HTTPException(404, f"Application {application_no} not found.") | |
| h, c, k = record["hostel_id"], record["course_type_id"], record["caste_category_vacancy_id"] | |
| allocations = [a for a in allocations if a["application_no"] != application_no] | |
| slot_list = get_slot_allocations(h, c, k) | |
| if record["status"] == "allocated": | |
| waitlisted = [a for a in slot_list if a["status"] == "waitlisted"] | |
| if waitlisted: | |
| top = max(waitlisted, key=lambda x: (x["marks_percentage"], -x["merit_rank"])) | |
| top["status"] = "allocated" | |
| top["message"] = ( | |
| f"Promoted from waitlist under '{caste_map.get(k, k)}' due to a vacancy." | |
| ) | |
| slot_list = recalculate_priorities(slot_list) | |
| update_global_allocations(slot_list) | |
| save_allocations_to_hf(allocations) | |
| return {"message": f"Application {application_no} removed and slot recalculated."} |