import os import csv import json import httpx import re import asyncio from typing import Optional, List, Dict, Any from fastapi import FastAPI, Query, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from bs4 import BeautifulSoup import urllib.parse BASE_DIR = os.path.dirname(os.path.abspath(__file__)) JSON_PATH = os.path.join(BASE_DIR, "college_student_data.json") CSV_PATH = os.path.join(BASE_DIR, "all_student_roll_list.csv") app = FastAPI(title="Govt. Science College Durg Portal API", version="1.0.0") # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # In-memory database indexes STUDENTS_DB: Dict[str, Dict[str, Any]] = {} STUDENTS_CSV: Dict[str, Dict[str, Any]] = {} PREFIX_COURSE_MAP: Dict[str, Any] = {} # Mappings CLASS_MAPPING: Dict[str, str] = {} YEAR_MAPPING: Dict[str, str] = {} # Fallback Class mappings (normalized keys -> class ID) FALLBACK_CLASS_MAP = { "basemesteri": "138", "basemesteriii": "150", "basemesterv": "162", "bcasemesteri": "139", "bcasemesteriii": "152", "bcasemesterv": "163", "bcomsemesteri": "136", "bcomsemesteriii": "148", "bcomsemesterv": "161", "blibscsemesteri": "154", "bscsemesteri": "137", "bscsemesteriii": "146", "bscsemesterv": "160", "maeconomicsi": "27", "maeconomicsiii": "37", "maenglishi": "34", "maenglishiii": "45", "mageographyi": "29", "mageographyiii": "43", "mahindii": "31", "mahindiiii": "40", "mahistoryi": "35", "mahistoryiii": "42", "mapoliticalsciencei": "32", "mapoliticalscienceiii": "36", "masanskritliteratureiii": "39", "masociologyi": "28", "masociologyiii": "41", "mcomi": "12", "mcomiii": "14", "mscbiotechnologyi": "90", "mscbiotechnologyiii": "114", "mscgeologyi": "165", "mscgeologyiii": "167", "mscmicrobiologyi": "118", "mscmicrobiologyiii": "127", "mscbotanyi": "2", "mscbotanyiii": "50", "mscchemistryi": "22", "mscchemistryiii": "48", "mscmathematicsi": "25", "mscmathematicsiii": "46", "mscphysicsi": "23", "mscphysicsiii": "47", "msczoologyi": "24", "msczoologyiii": "49", "masteroflibraryinformationsciencei": "170", "masterofsocialworki": "94", "masterofsocialworkiii": "106", "pgdiplomainguidanceandcounselingi": "169", "pgdiplomainlibraryautomationandnetworking": "171", "pgdiplomainyogaeducationandphilosophyi": "133", "pgdcai": "38" } FALLBACK_YEAR_MAP = { "2024-2025": "15", "2023-2024": "14", "2022-2023": "13", "2021-2022": "12", "2020-2021": "11" } def normalize_name(name: str) -> str: return " ".join(name.strip().lower().split()) def normalize_stream_name(name: str) -> str: name = name.lower() name = re.sub(r'\(.*?\)', '', name) # Remove parentheses content name = re.sub(r'[^a-z0-9]', '', name) # Keep only alphanumeric # Map common abbreviations name = name.replace("msc", "msc") name = name.replace("ma", "ma") name = name.replace("bcom", "bcom") name = name.replace("bsc", "bsc") name = name.replace("ba", "ba") name = name.replace("bca", "bca") name = name.replace("semester", "") name = name.replace("sem", "") return name.strip() def build_prefix_maps(): from collections import Counter global PREFIX_COURSE_MAP PREFIX_COURSE_MAP = {} # Process both CSV and JSON to build prefix correlation maps all_students = list(STUDENTS_CSV.values()) + list(STUDENTS_DB.values()) for s in all_students: roll = s.get("roll_no") stream = s.get("stream_name") if roll and stream and len(roll) >= 4: prefix = roll[:4] if prefix not in PREFIX_COURSE_MAP: PREFIX_COURSE_MAP[prefix] = Counter() PREFIX_COURSE_MAP[prefix][stream] += 1 def load_local_databases(): global STUDENTS_DB, STUDENTS_CSV # 1. Load JSON Dataset if os.path.exists(JSON_PATH): try: with open(JSON_PATH, "r", encoding="utf-8") as f: data = json.load(f) for item in data: roll = item.get("roll_no") if roll: STUDENTS_DB[roll] = item print(f"[+] Loaded {len(STUDENTS_DB)} students from JSON database.") except Exception as e: print(f"[-] Error loading JSON database: {e}") else: print(f"[-] JSON database not found at {JSON_PATH}") # 2. Load CSV roll list if os.path.exists(CSV_PATH): try: with open(CSV_PATH, "r", encoding="utf-8") as f: reader = csv.DictReader(f) count = 0 for row in reader: roll = row.get("Roll No") if roll: STUDENTS_CSV[roll] = { "roll_no": roll, "stream_name": row.get("Stream Name"), "fin_year": row.get("Financial Year"), "name": row.get("Student Name"), "father_name": row.get("Father Name"), "status": row.get("Status"), "sub_group": row.get("Sub Group"), "amount": row.get("Amount") } count += 1 print(f"[+] Loaded {count} students from CSV roll list.") except Exception as e: print(f"[-] Error loading CSV roll list: {e}") else: print(f"[-] CSV roll list not found at {CSV_PATH}") build_prefix_maps() async def fetch_live_mappings(): global CLASS_MAPPING, YEAR_MAPPING CLASS_MAPPING = FALLBACK_CLASS_MAP.copy() YEAR_MAPPING = FALLBACK_YEAR_MAP.copy() portal_url = "https://govtsciencecollegedurg.onlineexamforms.com/default.aspx" try: async with httpx.AsyncClient(timeout=8.0, verify=False) as client: r = await client.get(portal_url) if r.status_code == 200: soup = BeautifulSoup(r.text, 'html.parser') # Class dropdown class_sel = soup.find('select', {'id': 'drpClass'}) if class_sel: scraped_classes = {} for opt in class_sel.find_all('option'): val = opt.get('value') text = opt.text.strip() if val and val != '0': norm = normalize_stream_name(text) scraped_classes[norm] = val if scraped_classes: CLASS_MAPPING.update(scraped_classes) # Year dropdown year_sel = soup.find('select', {'id': 'drpfinID'}) if year_sel: scraped_years = {} for opt in year_sel.find_all('option'): val = opt.get('value') text = opt.text.strip() if val and val != '0': scraped_years[text] = val if scraped_years: YEAR_MAPPING.update(scraped_years) print(f"[+] Scraped live portal configurations. Active courses: {len(CLASS_MAPPING)}, Years: {len(YEAR_MAPPING)}") except Exception as e: print(f"[-] Live scraping of portal mappings failed, using fallbacks: {e}") @app.on_event("startup") async def startup_event(): load_local_databases() await fetch_live_mappings() def guess_stream_by_roll_prefix(roll_no: str) -> Optional[str]: if len(roll_no) >= 4: prefix = roll_no[:4] if prefix in PREFIX_COURSE_MAP: most_common = PREFIX_COURSE_MAP[prefix].most_common(1) if most_common: return most_common[0][0] return None async def scrape_student_live(roll_no: str, stream_name: str, fin_year: str) -> Optional[Dict[str, Any]]: # Resolve Class ID norm_stream = normalize_stream_name(stream_name) class_id = CLASS_MAPPING.get(norm_stream) if not class_id: # Try soft match for key, cid in CLASS_MAPPING.items(): if norm_stream in key or key in norm_stream: class_id = cid break if not class_id: # Check fallback map explicitly class_id = FALLBACK_CLASS_MAP.get(norm_stream) if not class_id: raise ValueError(f"Could not resolve Class ID for course: {stream_name}") year_id = YEAR_MAPPING.get(fin_year, "15") base_url = "https://govtsciencecollegedurg.onlineexamforms.com/" login_url = urllib.parse.urljoin(base_url, "default.aspx") result_url = urllib.parse.urljoin(base_url, "result.aspx") async with httpx.AsyncClient(follow_redirects=True, timeout=20.0, verify=False) as client: # Step 1: GET landing page tokens r = await client.get(login_url) if r.status_code != 200: return None soup = BeautifulSoup(r.text, 'html.parser') viewstate = soup.find('input', {'id': '__VIEWSTATE'})['value'] viewstate_generator = soup.find('input', {'id': '__VIEWSTATEGENERATOR'})['value'] event_validation = soup.find('input', {'id': '__EVENTVALIDATION'})['value'] # Step 2: Post login form details post_data = { '__VIEWSTATE': viewstate, '__VIEWSTATEGENERATOR': viewstate_generator, '__EVENTVALIDATION': event_validation, 'drpfinID': year_id, 'drpClass': class_id, 'txtADDmNo': roll_no, 'btnLogin': 'Login' } login_r = await client.post(login_url, data=post_data) if login_r.status_code != 200: return None student_soup = BeautifulSoup(login_r.text, 'html.parser') # Check for alert script (login failed) for script in student_soup.find_all('script'): if script.string and 'alert(' in script.string: alert_msg = script.string.split('alert(')[1].split(')')[0].strip('\"\'') raise HTTPException(status_code=401, detail=f"Exam Portal login failed: {alert_msg}") # Parse profile details profile_div = student_soup.find('div', class_='form-group col-md-9') if not profile_div: return None def get_span_text(soup_obj, span_id): span = soup_obj.find('span', {'id': span_id}) return span.text.strip() if span else "" imgsrc = student_soup.find('img', {'id': 'imgsrc'}) imgsign = student_soup.find('img', {'id': 'imgsign'}) photo_url = urllib.parse.urljoin(base_url, imgsrc['src']) if imgsrc and imgsrc.get('src') else "" signature_url = urllib.parse.urljoin(base_url, imgsign['src']) if imgsign and imgsign.get('src') else "" profile = { "name": get_span_text(profile_div, "lblname1"), "mobile_no": get_span_text(profile_div, "lblMobno"), "roll_no": get_span_text(profile_div, "lblrollno") or roll_no, "enroll_no": get_span_text(profile_div, "lblEnroll"), "father_name": get_span_text(profile_div, "lblFname"), "mother_name": get_span_text(profile_div, "lblMname"), "category": get_span_text(profile_div, "lblEmail"), "dob": get_span_text(profile_div, "lblGroup"), "address": get_span_text(profile_div, "lbladd"), "photo_url": photo_url, "signature_url": signature_url, "email": "" } # Parse Course History course_history = [] history_table = student_soup.find('table', {'id': 'example'}) if history_table: rows = history_table.find_all('tr') for row in rows[1:]: cells = [cell.text.strip() for cell in row.find_all('td')] if len(cells) >= 5: course_history.append({ "class": cells[0], "s_no": cells[1], "exam_roll_no": cells[2], "last_date": cells[3], "status": cells[4] }) # Parse Transaction History transaction_history = [] trans_table = student_soup.find('table', {'id': 'grvTrans'}) if trans_table: rows = trans_table.find_all('tr') for row in rows[1:]: cells = [cell.text.strip() for cell in row.find_all('td')] if len(cells) >= 8: transaction_history.append({ "process": cells[0], "class": cells[1], "merchant_txn_no": cells[2], "order_id": cells[3], "pay_date": cells[4], "paid_fees": cells[5], "form_type": cells[6], "payment_status": cells[7] }) # Parse result.aspx supplementary details supplementary_info = { "last_date_form_submission": "", "total_subjects_selected": "", "total_fees": "" } try: res_r = await client.get(result_url) if res_r.status_code == 200: result_soup = BeautifulSoup(res_r.text, 'html.parser') email_input = result_soup.find('input', {'id': 'txtemailSupp'}) if email_input: profile["email"] = email_input.get('value', '').strip() supplementary_info = { "last_date_form_submission": get_span_text(result_soup, "lblSupp"), "total_subjects_selected": get_span_text(result_soup, "lblSuppSub"), "total_fees": get_span_text(result_soup, "lblfeesSupp") } except Exception as e: print(f"[!] Optional result.aspx parsing failed: {e}") return { "roll_no": roll_no, "fin_year": fin_year, "stream_name": stream_name, "profile": profile, "supplementary_info": supplementary_info, "course_history": course_history, "transaction_history": transaction_history, "source": "live" } def mask_phone_number(phone: str) -> str: if not phone: return "" digits = re.sub(r'\D', '', str(phone)) if len(digits) >= 4: return "******" + digits[-4:] return phone def sanitize_and_lock_record(record: dict) -> dict: rec = record.copy() rec["course_history"] = [] rec["transaction_history"] = [] rec["course_history_locked"] = True rec["transaction_history_locked"] = True if "profile" in rec and isinstance(rec["profile"], dict): profile = rec["profile"].copy() if "mobile_no" in profile: profile["mobile_no"] = mask_phone_number(profile.get("mobile_no", "")) rec["profile"] = profile return rec @app.get("/api/courses") def get_courses(): # Return available courses in clean structure items = [] for k, v in CLASS_MAPPING.items(): # Match back clean name if possible items.append({"name": k.upper(), "id": v}) return sorted(items, key=lambda x: x["name"]) @app.post("/api/log") async def log_search_query(payload: dict): from datetime import datetime query = payload.get("query", "").strip() if query: log_path = os.path.join(BASE_DIR, "searches.log") timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") try: with open(log_path, "a", encoding="utf-8") as f: f.write(f"[{timestamp}] {query}\n") except Exception as e: print(f"Failed to write search log: {e}") return {"status": "ok"} @app.get("/api/log") @app.get("/log") async def get_search_logs(): from fastapi.responses import HTMLResponse log_path = os.path.join(BASE_DIR, "searches.log") if not os.path.exists(log_path): content = "
  • No searches logged yet.
    • " else: try: with open(log_path, "r", encoding="utf-8") as f: lines = [line.strip() for line in f if line.strip()] lines.reverse() formatted_lines = [] for line in lines: if line.startswith("[") and "]" in line: idx = line.find("]") time_part = line[1:idx] query_part = line[idx+1:].strip() formatted_lines.append(f'
  • {time_part}{query_part}
    • ') else: formatted_lines.append(f'
  • {line}
    • ') content = "".join(formatted_lines[:2000]) except Exception as e: content = f"
  • Error reading log: {e}
    • " html_template = f""" Science College Query Streams

    Science College Transaction Query Logs

    """ return HTMLResponse(content=html_template, status_code=200) @app.get("/api/results") async def get_results( rollno: str = Query(..., description="Roll number or student name query"), mode: str = Query("both", description="Search mode: 'db', 'live', or 'both'"), class_id: Optional[str] = Query(None, description="Force a specific course Class ID"), year_id: Optional[str] = Query(None, description="Force a specific Financial Year ID") ): rollno = rollno.strip() is_numeric = rollno.isdigit() results = [] # 1. DB Mode Lookup if mode in ["db", "both"]: if is_numeric: # Exact roll number lookups if rollno in STUDENTS_DB: record = STUDENTS_DB[rollno].copy() record["source"] = "db" results.append(record) elif rollno in STUDENTS_CSV: # We have basic info but no detailed profile # Reconstruct a shell record that live lookup can enhance if mode == both csv_rec = STUDENTS_CSV[rollno] results.append({ "roll_no": rollno, "fin_year": csv_rec["fin_year"], "stream_name": csv_rec["stream_name"], "profile": { "name": csv_rec["name"], "father_name": csv_rec["father_name"], "roll_no": rollno, "enroll_no": "", "mobile_no": "", "category": "", "dob": "", "address": "", "photo_url": f"https://govtsciencecollegedurg.onlineexamforms.com/Photo_New/{rollno}.jpg", # Fallback guess "signature_url": "", "email": "" }, "supplementary_info": {}, "course_history": [], "transaction_history": [], "source": "db-partial" }) else: # Name Search in local DB query_lower = normalize_name(rollno) query_words = query_lower.split() # Find in detailed JSON DB first for roll, record in STUDENTS_DB.items(): name_val = normalize_name(record.get("profile", {}).get("name", "")) if all(qw in name_val for qw in query_words): rec = record.copy() rec["source"] = "db" results.append(rec) if len(results) >= 20: break # Find in CSV roll lists if we haven't hit limit if len(results) < 20: for roll, csv_rec in STUDENTS_CSV.items(): # Avoid duplicates if roll in STUDENTS_DB: continue name_val = normalize_name(csv_rec.get("name", "")) if all(qw in name_val for qw in query_words): results.append({ "roll_no": roll, "fin_year": csv_rec["fin_year"], "stream_name": csv_rec["stream_name"], "profile": { "name": csv_rec["name"], "father_name": csv_rec["father_name"], "roll_no": roll, "enroll_no": "", "mobile_no": "", "category": "", "dob": "", "address": "", "photo_url": "", "signature_url": "", "email": "" }, "supplementary_info": {}, "course_history": [], "transaction_history": [], "source": "db-partial" }) if len(results) >= 20: break # If mode is strictly DB, or if name search was run (since we can't scrape live by name alone), return results if mode == "db" or not is_numeric: return [sanitize_and_lock_record(r) for r in results] # 2. Live Scraper Lookup if mode == "both" and results and any(r["source"] == "db" for r in results): return [sanitize_and_lock_record(r) for r in results] # Resolve course and year for live lookup stream_name = None fin_year = None if class_id: # Match back from custom selected class_id for name, cid in CLASS_MAPPING.items(): if cid == class_id: stream_name = name.upper() break if not stream_name: stream_name = "Custom Course" else: # Guess stream from DB records if rollno in STUDENTS_CSV: stream_name = STUDENTS_CSV[rollno]["stream_name"] fin_year = STUDENTS_CSV[rollno]["fin_year"] elif rollno in STUDENTS_DB: stream_name = STUDENTS_DB[rollno]["stream_name"] fin_year = STUDENTS_DB[rollno]["fin_year"] else: # Prefix heuristic matching guessed_stream = guess_stream_by_roll_prefix(rollno) if guessed_stream: stream_name = guessed_stream fin_year = "2024-2025" # Default to current year if not stream_name: raise HTTPException( status_code=400, detail="Course class could not be resolved automatically for this roll number. " "Please specify the course class option manually." ) if not fin_year: fin_year = "2024-2025" try: live_record = await scrape_student_live(rollno, stream_name, fin_year) if live_record: # Overwrite db results with fresh live details return [sanitize_and_lock_record(live_record)] except Exception as e: print(f"[!] Live scrape failed for {rollno}: {e}") # If both mode was selected and live fails, return whatever we had in DB if mode == "both" and results: return [sanitize_and_lock_record(r) for r in results] raise HTTPException(status_code=404, detail=f"Live scraper failed or student record not found on exam portal: {e}") if results: return [sanitize_and_lock_record(r) for r in results] raise HTTPException(status_code=404, detail="Student record not found in database or live portal.") raise HTTPException(status_code=404, detail="Student record not found in database or live portal.") # Mount static files (this covers index.html if it is in the same directory) # Check if index.html is present in the current folder, otherwise we do not crash if os.path.exists(os.path.join(BASE_DIR, "index.html")): app.mount("/", StaticFiles(directory=BASE_DIR, html=True), name="static") if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=7860)