Spaces:
Sleeping
Sleeping
| # In api/main.py | |
| from fastapi import FastAPI, HTTPException, Depends | |
| from pydantic import BaseModel | |
| from typing import List, Optional | |
| import sys | |
| import os | |
| # Add parent directory to path to import modules | |
| sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from ai.huggingface import HuggingFaceService | |
| from ai.agents import AgentCoordinator | |
| from data.content import ContentManager | |
| from data.users import UserManager | |
| app = FastAPI( | |
| title="Micro Learning API", | |
| description="API for microlearning content and personalization", | |
| version="0.1.0" | |
| ) | |
| # Initialize services | |
| hf_service = HuggingFaceService() | |
| agent_coordinator = AgentCoordinator(hf_service) | |
| content_manager = ContentManager() | |
| user_manager = UserManager() | |
| # Models | |
| class ContentBase(BaseModel): | |
| title: str | |
| text: str | |
| tags: List[str] | |
| class ContentCreate(ContentBase): | |
| pass | |
| class Content(ContentBase): | |
| id: str | |
| class UserBase(BaseModel): | |
| name: str | |
| email: str | |
| class UserCreate(UserBase): | |
| pass | |
| class User(UserBase): | |
| id: str | |
| progress: dict = {} | |
| class ProgressUpdate(BaseModel): | |
| module_id: str | |
| completion: float # 0.0 to 1.0 | |
| # Routes | |
| async def get_content(content_id: str): | |
| content = content_manager.get_by_id(content_id) | |
| if not content: | |
| raise HTTPException(status_code=404, detail="Content not found") | |
| return content | |
| async def create_content(content: ContentCreate): | |
| content_id = content_manager.save_content(content.dict()) | |
| return content_id | |
| async def get_summary(content_id: str): | |
| content = content_manager.get_by_id(content_id) | |
| if not content: | |
| raise HTTPException(status_code=404, detail="Content not found") | |
| summary = hf_service.summarize_content(content['text']) | |
| return {"summary": summary[0]['summary_text']} | |
| async def ask_question(content_id: str, question: str): | |
| content = content_manager.get_by_id(content_id) | |
| if not content: | |
| raise HTTPException(status_code=404, detail="Content not found") | |
| answer = hf_service.answer_question(question, content['text']) | |
| return {"answer": answer['answer'], "confidence": answer['score']} | |
| async def generate_quiz(content_id: str): | |
| content = content_manager.get_by_id(content_id) | |
| if not content: | |
| raise HTTPException(status_code=404, detail="Content not found") | |
| questions = agent_coordinator.dispatch("quiz", content['text']) | |
| return {"questions": questions} | |
| async def create_user(user: UserCreate): | |
| user_id = user_manager.create_user(user.dict()) | |
| return user_id | |
| async def get_user(user_id: str): | |
| user = user_manager.get_user(user_id) | |
| if not user: | |
| raise HTTPException(status_code=404, detail="User not found") | |
| return user | |
| async def update_progress(user_id: str, update: ProgressUpdate): | |
| user = user_manager.get_user(user_id) | |
| if not user: | |
| raise HTTPException(status_code=404, detail="User not found") | |
| user_manager.update_progress(user_id, update.module_id, update.completion) | |
| return {"status": "updated"} |