Create main.py

main.py

@@ -0,0 +1,147 @@
+import json
+import os
+import chromadb
+import numpy as np
+from fastapi import FastAPI, HTTPException, UploadFile, File, Form
+from pydantic import BaseModel
+from typing import List, Optional
+from huggingface_hub import InferenceClient
+from scipy.spatial.distance import cosine
+
+app = FastAPI(title="Course Recommendation API")
+
+# Initialize Hugging Face Inference Client
+HF_API_TOKEN = os.getenv("HF_API_TOKEN", os.getenv['HF_API_TOKEN'])
+client = InferenceClient(model="sentence-transformers/all-MiniLM-L6-v2", token=HF_API_TOKEN)
+
+# Initialize ChromaDB
+chroma_client = chromadb.PersistentClient(path="./chroma_db")
+collection = chroma_client.get_or_create_collection(name="courses")
+
+def get_embedding(text):
+    response = client.post(json={"inputs": text}, task="feature-extraction")
+    
+    # Handle different response formats
+    if hasattr(response, 'tolist'):
+        return response.tolist()  # Handle if it's already a NumPy array
+    elif isinstance(response, list):
+        if len(response) > 0 and isinstance(response[0], list):
+            return response[0]  # Return first item if response is a list of lists
+        else:
+            return response  # Return as is if it's a flat list
+    else:
+        # Convert from bytes if needed
+        try:
+            if isinstance(response, bytes):
+                import ast
+                return ast.literal_eval(response.decode('utf-8'))
+            else:
+                return response
+        except:
+            raise ValueError(f"Unexpected embedding format: {response}")
+
+class Course(BaseModel):
+    course_id: str
+    course_name: str
+    abstract: str
+
+class CourseResponse(BaseModel):
+    course_id: str
+    name: str
+    similarity: float
+
+@app.post("/add_course")
+async def add_course(course: Course):
+    """Add a single course to the database"""
+    text = f"Course: {course.course_name}, Description: {course.abstract}"
+    
+    try:
+        embedding = get_embedding(text)
+        if not isinstance(embedding, list):
+            if hasattr(embedding, 'tolist'):
+                embedding = embedding.tolist()
+            else:
+                embedding = list(embedding)
+        
+        collection.add(
+            ids=[course.course_id],
+            embeddings=[embedding],
+            metadatas=[{"course_id": course.course_id, "name": course.course_name}]
+        )
+        return {"status": "success", "message": "Course added successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error adding course: {str(e)}")
+
+@app.post("/upload_courses")
+async def upload_courses(file: UploadFile = File(...)):
+    """Upload a JSON file with multiple courses"""
+    try:
+        contents = await file.read()
+        courses = json.loads(contents)
+        
+        for course in courses:
+            text = f"Course: {course['course_name']}, Description: {course['abstract']}"
+            embedding = get_embedding(text)
+            
+            if not isinstance(embedding, list):
+                if hasattr(embedding, 'tolist'):
+                    embedding = embedding.tolist()
+                else:
+                    embedding = list(embedding)
+            
+            collection.add(
+                ids=[str(course["course_id"])],
+                embeddings=[embedding],
+                metadatas=[{"course_id": course["course_id"], "name": course["course_name"]}]
+            )
+        
+        return {"status": "success", "message": f"{len(courses)} courses added successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error processing file: {str(e)}")
+
+@app.get("/search", response_model=List[CourseResponse])
+async def search_courses(query: str, limit: Optional[int] = 3):
+    """Find similar courses based on query text"""
+    try:
+        query_embedding = get_embedding(query)
+        
+        # Ensure query embedding is properly formatted
+        if not isinstance(query_embedding, (list, np.ndarray)):
+            if hasattr(query_embedding, 'tolist'):
+                query_embedding = query_embedding.tolist()
+            else:
+                query_embedding = list(query_embedding)
+        
+        # Retrieve stored embeddings
+        results = collection.get(include=["embeddings", "metadatas"])
+        courses = results["metadatas"]
+        stored_embeddings = results["embeddings"]
+        
+        if not courses:
+            return []
+        
+        # Compute cosine similarities
+        similarities = [1 - cosine(query_embedding, emb) for emb in stored_embeddings]
+        
+        # Get top similar courses
+        top_indices = np.argsort(similarities)[-limit:][::-1]
+        
+        # Format response
+        response = []
+        for i in top_indices:
+            response.append(
+                CourseResponse(
+                    course_id=courses[i]["course_id"],
+                    name=courses[i]["name"],
+                    similarity=float(similarities[i])
+                )
+            )
+        
+        return response
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error searching courses: {str(e)}")
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "ok"}