main.py

# =========================================
# 1. IMPORTS
# =========================================
import asyncio
import os
import json
import uuid
import cloudinary
import cloudinary.uploader
import firebase_admin

from firebase_admin import credentials, firestore
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from gradio_client import Client
from google.cloud.firestore_v1.base_query import FieldFilter
import edge_tts

from typing import Optional, List
from dotenv import load_dotenv
from contextlib import asynccontextmanager

# =========================================
# 2. INITIALIZATIONS & CONFIG
# =========================================
load_dotenv()

if not firebase_admin._apps:
    fb_json = os.getenv("FIREBASE_JSON")

    if fb_json:
        cred_dict = json.loads(fb_json)
        cred = credentials.Certificate(cred_dict)
    else:
        cred = credentials.Certificate("serviceAccountKey.json")

    firebase_admin.initialize_app(cred)

db = firestore.client()

cloudinary.config(
    cloud_name=os.getenv("CLOUD_NAME"),
    api_key=os.getenv("API_KEY"),
    api_secret=os.getenv("API_SECRET"),
    secure=True
)

HF_SPACE = "Fayza38/Question_and_answer_model"
client = None

# =========================================
# 3. CONSTANTS & MODELS
# =========================================
TECH_CATEGORIES = {
    0: "Security",
    1: "BackEnd",
    2: "Networking",
    3: "FrontEnd",
    4: "DataEngineering",
    5: "WebDevelopment",
    6: "FullStack",
    7: "VersionControl",
    8: "SystemDesign",
    9: "MachineLearning",
    10: "LanguagesAndFrameworks",
    11: "DatabaseSystems",
    12: "ArtificialIntelligence",
    13: "SoftwareTesting",
    14: "DistributedSystems",
    15: "DevOps",
    16: "LowLevelSystems",
    17: "DatabaseAndSql",
    18: "GeneralProgramming",
    19: "DataStructures",
    20: "Algorithms"
}

DIFFICULTY_MAP = {
    0: "Easy",
    1: "Intermediate",
    2: "Hard"
}


class GenerateSessionRequest(BaseModel):
    sessionId: str

    # 0 = Behavioral
    # 1 = Technical
    sessionType: int

    # IMPORTANT:
    # difficultyLevel is ONLY used for technical sessions
    difficultyLevel: Optional[int] = None

    # ONLY required for technical sessions
    trackName: Optional[int] = None


class CleanupRequest(BaseModel):
    audioUrls: List[str]


# =========================================
# 4. LIFESPAN MANAGEMENT
# =========================================
@asynccontextmanager
async def lifespan(app: FastAPI):
    global client

    print("Connecting to Hugging Face Model...")

    try:
        loop = asyncio.get_event_loop()

        client = await loop.run_in_executor(
            None,
            lambda: Client(HF_SPACE)
        )

        print("Model Connected Successfully!")

    except Exception as e:
        print(f"Model Connection Failed: {e}")

    yield

    print("Shutting down Intervision Service...")


app = FastAPI(
    title="Intervision AI Question Service",
    lifespan=lifespan
)

# =========================================
# 5. HELPER FUNCTIONS
# =========================================
async def generate_audio(text: str, filename: str):
    try:
        communicate = edge_tts.Communicate(
            text,
            "en-US-GuyNeural",
            rate="-15%"
        )

        await communicate.save(filename)

        upload_result = cloudinary.uploader.upload(
            filename,
            resource_type="video",
            folder="interview_audio"
        )

        if os.path.exists(filename):
            os.remove(filename)

        return upload_result["secure_url"]

    except Exception as e:
        print(f"Audio Generation Error: {e}")

        if os.path.exists(filename):
            os.remove(filename)

        return None


async def safe_generate(prompt: str, retries: int = 5):
    if client is None:
        raise Exception("AI Client is not initialized.")

    for attempt in range(retries):
        try:
            loop = asyncio.get_running_loop()

            result = await loop.run_in_executor(
                None,
                lambda: client.predict(
                    prompt=prompt,
                    api_name="/generate_questions"
                )
            )

            return result

        except Exception as e:
            if attempt == retries - 1:
                raise e

            await asyncio.sleep(5)


def parse_question_output(raw_output: str):
    if not raw_output:
        return None, None

    text = (
        raw_output.split("assistant")[-1].strip()
        if "assistant" in raw_output
        else raw_output
    )

    if "Q:" in text and "A:" in text:
        try:
            parts = text.split("A:")

            question = parts[0].replace("Q:", "").strip()

            answer = (
                parts[1]
                .split("<|im_end|>")[0]
                .strip()
            )

            return question, answer

        except Exception:
            return None, None

    return None, None


# =========================================
# 6. REFILL QUESTION POOLS
# =========================================
async def refill_specific_pool(
    track_id: Optional[int],
    difficulty: Optional[int],
    count: int,
    session_type: int
):
    while client is None:
        await asyncio.sleep(5)

    # =====================================
    # Behavioral Questions
    # =====================================
    if session_type == 0:

        track_text = "Behavioral"

        prompt = (
            "Generate ONE unique behavioral interview question. "
            "Format: Q: [Question] A: [Answer]"
        )

    # =====================================
    # Technical Questions
    # =====================================
    else:

        track_text = TECH_CATEGORIES.get(track_id)

        level_text = DIFFICULTY_MAP.get(difficulty)

        prompt = (
            f"Generate ONE unique {track_text} "
            f"question for {level_text} level. "
            f"Format: Q: [Question] A: [Answer]"
        )

    success_count = 0

    while success_count < count:
        try:
            raw_output = await safe_generate(prompt)

            question, answer = parse_question_output(raw_output)

            if question and answer:

                audio_url = await generate_audio(
                    question,
                    f"{uuid.uuid4()}.mp3"
                )

                if audio_url:

                    question_data = {
                        "session_type": session_type,
                        "questionText": question,
                        "questionIdealAnswer": answer,
                        "audio_url": audio_url,
                        "created_at": firestore.SERVER_TIMESTAMP
                    }

                    # =================================
                    # Technical ONLY
                    # =================================
                    if session_type == 1:
                        question_data["track_id"] = track_id
                        question_data["difficulty"] = difficulty

                    db.collection("questions_pool").add(question_data)

                    success_count += 1

                    print(
                        f"Successfully added "
                        f"{track_text} question "
                        f"{success_count}/{count}"
                    )

                    await asyncio.sleep(2)

        except Exception as e:
            print(f"Refill error: {e}")
            await asyncio.sleep(5)

# =========================================
# 7. API ENDPOINTS
# =========================================
@app.post("/generate-session")
async def generate_session(
    request: GenerateSessionRequest,
    background_tasks: BackgroundTasks
):
    session_type = request.sessionType
    track_id = request.trackName

    # =====================================
    # Behavioral Session
    # =====================================
    if session_type == 0:

        query = db.collection("questions_pool").where(
            filter=FieldFilter("session_type", "==", 0)
        )

    # =====================================
    # Technical Session
    # =====================================
    elif session_type == 1:

        if track_id is None:
            raise HTTPException(
                status_code=400,
                detail="trackName is required for technical sessions."
            )

        if request.difficultyLevel is None:
            raise HTTPException(
                status_code=400,
                detail="difficultyLevel is required for technical sessions."
            )

        difficulty = request.difficultyLevel

        query = (
            db.collection("questions_pool")
            .where(filter=FieldFilter("session_type", "==", 1))
            .where(filter=FieldFilter("track_id", "==", track_id))
            .where(filter=FieldFilter("difficulty", "==", difficulty))
        )

    else:
        raise HTTPException(
            status_code=400,
            detail="Invalid sessionType."
        )

    docs = query.limit(10).get()

    final_questions = []

    for index, doc in enumerate(docs, start=1):

        data = doc.to_dict()

        final_questions.append({
            "question_id": index,
            "text": data["questionText"],
            "expected_answer": data["questionIdealAnswer"],
            "audio_url": data.get("audio_url", "")
        })

        # remove used question
        db.collection("questions_pool").document(doc.id).delete()

    # =====================================
    # BACKGROUND REFILL
    # =====================================
    async def check_and_refill_background():

        snap = query.count().get()
        current_count = snap[0][0].value

        if current_count < 50:

            if session_type == 0:

                print(
                    f"Behavioral stock low "
                    f"({current_count}) -> refilling..."
                )

                await refill_specific_pool(
                    track_id=None,
                    difficulty=None,
                    count=50 - current_count,
                    session_type=0
                )

            else:

                print(
                    f"{TECH_CATEGORIES[track_id]} stock low "
                    f"({current_count}) -> refilling..."
                )

                await refill_specific_pool(
                    track_id=track_id,
                    difficulty=difficulty,
                    count=50 - current_count,
                    session_type=1
                )

    background_tasks.add_task(check_and_refill_background)

    if not final_questions:
        raise HTTPException(
            status_code=503,
            detail="The question pool is empty. Please try again in a few minutes."
        )

    return {
        "session_id": request.sessionId,
        "questions": final_questions
    }


# =========================================
# 8. ADMIN PREFILL
# =========================================
@app.get("/admin/prefill-all")
async def prefill_all(background_tasks: BackgroundTasks):

    async def run_sync():

        print("Starting Global Smart Prefill...")

        # =================================
        # Behavioral Questions
        # =================================
        behavioral_query = (
            db.collection("questions_pool")
            .where(filter=FieldFilter("session_type", "==", 0))
        )

        behavioral_snap = behavioral_query.count().get()

        behavioral_count = behavioral_snap[0][0].value

        if behavioral_count < 50:

            needed = 50 - behavioral_count

            print(f"Syncing Behavioral: adding {needed}")

            await refill_specific_pool(
                track_id=None,
                difficulty=None,
                count=needed,
                session_type=0
            )

        # =================================
        # Technical Questions
        # =================================
        for track_id, track_name in TECH_CATEGORIES.items():

            for diff_id, diff_name in DIFFICULTY_MAP.items():

                technical_query = (
                    db.collection("questions_pool")
                    .where(filter=FieldFilter("session_type", "==", 1))
                    .where(filter=FieldFilter("track_id", "==", track_id))
                    .where(filter=FieldFilter("difficulty", "==", diff_id))
                )

                snap = technical_query.count().get()

                current = snap[0][0].value

                if current < 50:

                    needed = 50 - current

                    print(
                        f"Syncing {track_name} "
                        f"({diff_name}): adding {needed}"
                    )

                    await refill_specific_pool(
                        track_id=track_id,
                        difficulty=diff_id,
                        count=needed,
                        session_type=1
                    )

        print("Global Smart Prefill Completed!")

    background_tasks.add_task(run_sync)

    return {
        "message": "Global prefill process started in the background."
    }


# =========================================
# 9. CLEANUP AUDIO
# =========================================
@app.post("/cleanup-audio")
async def cleanup_audio(
    request: CleanupRequest,
    background_tasks: BackgroundTasks
):
    def delete_job(urls):

        for url in urls:
            try:
                public_id = (
                    "interview_audio/"
                    + url.split("/")[-1].split(".")[0]
                )

                cloudinary.uploader.destroy(
                    public_id,
                    resource_type="video"
                )

            except Exception:
                pass

    background_tasks.add_task(
        delete_job,
        request.audioUrls
    )

    return {
        "message": "Cloudinary cleanup process initiated."
    }


# =========================================
# 10. HEALTH CHECK
# =========================================
@app.get("/health")
async def health():
    return {
        "status": "active",
        "ai_model_connected": client is not None
    }


@app.get("/")
async def root():
    return {
        "app": "Intervision AI Engine",
        "status": "Running"
    }


# =========================================
# 11. RUN SERVER
# =========================================
if __name__ == "__main__":

    import uvicorn

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )