from fastapi import FastAPI, HTTPException, Header, Depends, Request from fastapi.responses import JSONResponse, HTMLResponse from fastapi.middleware.cors import CORSMiddleware import json import uvicorn import os from typing import List, Dict, Optional # ----------------------------- # CONFIGURATION & CONSTANTS # ----------------------------- DATA_FILE = "ddcet_dataset_raw.json" BLOG_FILE = "blogs.json" ADMIN_KEY_ENV = os.getenv("ADMIN_KEY", "secret123") # Set this in HF Spaces Secrets # ----------------------------- # LOAD DATA # ----------------------------- # Helper to load raw dataset safely def load_raw_data(): if not os.path.exists(DATA_FILE): # Fallback for demo if file missing return {"subjects": []} with open(DATA_FILE, "r", encoding="utf-8") as f: return json.load(f) raw_data = load_raw_data() def load_blogs(): if not os.path.exists(BLOG_FILE): return [] try: with open(BLOG_FILE, "r", encoding="utf-8") as f: return json.load(f) except: return [] def save_blogs(data): with open(BLOG_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=4) blogs_data = load_blogs() # Create quick lookup maps for performance subjects_map: Dict[int, Dict] = {} units_map: Dict[int, Dict] = {} questions_map: Dict[int, List[Dict]] = {} if "subjects" in raw_data: for subj in raw_data["subjects"]: sid = subj["SubjectID"] subjects_map[sid] = {"SubjectID": sid, "SubjectName": subj["SubjectName"]} for unit in subj["units"]: uid = unit["UnitID"] units_map[uid] = {"UnitID": uid, "UnitName": unit.get("UnitName",""), "SubjectID": sid} questions_map[uid] = unit.get("questions", []) # ----------------------------- # AUTHENTICATION # ----------------------------- async def verify_admin(x_admin_key: str = Header(None)): """ Checks for x-admin-key in the header. """ if x_admin_key != ADMIN_KEY_ENV: raise HTTPException(status_code=401, detail="Invalid Admin Key") return x_admin_key # ----------------------------- # FASTAPI INSTANCE # ----------------------------- app = FastAPI(title="DDCET MCQ API Fast", version="1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ----------------------------- # ENDPOINTS # ----------------------------- @app.get("/subjects", response_class=JSONResponse) def get_subjects(): return list(subjects_map.values()) @app.get("/units/{subject_id}", response_class=JSONResponse) def get_units(subject_id: int): units = [u for u in units_map.values() if u["SubjectID"] == subject_id] if not units: raise HTTPException(status_code=404, detail="No units found for this subject") return units @app.get("/questions/{unit_id}", response_class=JSONResponse) def get_questions(unit_id: int): qs = questions_map.get(unit_id) if qs is None: raise HTTPException(status_code=404, detail="No questions found for this unit") return qs @app.get("/question/{mcqid}", response_class=JSONResponse) def get_question(mcqid: int): for qlist in questions_map.values(): for q in qlist: if q["QuestionObject"]["MCQID"] == mcqid: return q raise HTTPException(status_code=404, detail="Question not found") @app.get("/blogs") def get_blogs(): return blogs_data # PROTECTED ENDPOINT @app.post("/blogs", dependencies=[Depends(verify_admin)]) def add_blog(blog: dict): blog["id"] = blog.get("id", len(blogs_data) + 1) # Ensure fields exist for the frontend blog["date"] = blog.get("date", "") blogs_data.insert(0, blog) # Add to top save_blogs(blogs_data) return {"message": "Blog added", "blog": blog} # ----------------------------- # ADMIN PANEL HTML UI # ----------------------------- @app.get("/admin", response_class=HTMLResponse) def admin_panel(): html_content = """ Admin Panel

Admin /// Console

/// New Entry

/// Live Preview

Start typing HTML content to preview...

/// Database Entries

Loading stream...
""" return html_content # ----------------------------- # RUN SERVER # ----------------------------- if __name__ == "__main__": # In HF Spaces, PORT is usually 7860 port = int(os.environ.get("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)