Merge branch 'main' into 68-genai-enhance-course-generation-api-confirmation-chat-and-crawling

Author: mahdibayouli

genai/.gitignore

@@ -20,3 +20,7 @@ genai-devops25-env/
 
 # Misc
 *.log
+
+# Scheduler cache
+scheduler_cache.json
+src/services/scheduler/scheduler_cache.json

genai/requirements.txt

@@ -9,3 +9,4 @@ langchain
 langchain-openai
 langchain_community
 tiktoken
+pydantic

genai/src/main.py

@@ -25,6 +25,10 @@
 from .utils.error_schema import ErrorResponse
 from .utils.handle_httpx_exception import handle_httpx_exception
 
+from .services.scheduler.schemas import SchedulerStatus, SchedulerControl
+from .services.scheduler import start_scheduler, stop_scheduler, get_scheduler_status
+
+
 
 # --- Configuration ---
 load_dotenv()
@@ -44,27 +48,37 @@
     {"name": "Crawler", "description": "Crawl and clean website content."},
     {"name": "Embedder", "description": "Chunk and embed text to vector DB."},
     {"name": "LLM", "description": "Language Model completions and chat."},
+    {"name": "Scheduler", "description": "Scheduled background jobs and automation."},
 ]
 
 # -------------------------------
 # --- Lifespan Manager (for startup) ---
 # -------------------------------
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    logger.info("Service starting up...")
-    try:
-        # Example: Create client connections here
-        client = get_weaviate_client()
-        logger.info("Testing Weaviate connection...")
-        test_weaviate_connection()
-        logger.info("Ensuring Weaviate schema exists...")
-        ensure_schema_exists(client)
-        logger.info("Everything is set up successfully and ready to go!")
-    except Exception as e:
-        logger.error(f"Error during startup: {e}", exc_info=True)
-        raise RuntimeError("Failed during application startup.") from e
-    yield
-    logger.info("Service shutting down...")
+   logger.info("Service starting up...")
+   try:
+       # Example: Create client connections here
+       client = get_weaviate_client()
+       logger.info("Testing Weaviate connection...")
+       test_weaviate_connection()
+       logger.info("Ensuring Weaviate schema exists...")
+       ensure_schema_exists(client)
+       logger.info("Everything is set up successfully and ready to go!")
+      
+       # Start the blog embedder scheduler
+       logger.info("Starting blog embedder scheduler...")
+       start_scheduler()
+      
+   except Exception as e:
+       logger.error(f"Error during startup: {e}", exc_info=True)
+       raise RuntimeError("Failed during application startup.") from e
+   yield
+   logger.info("Service shutting down...")
+  
+   # Stop the scheduler on shutdown
+   logger.info("Stopping blog embedder scheduler...")
+   stop_scheduler()
 
 # --- App Initialization ---
 app = FastAPI(
@@ -263,6 +277,53 @@ async def generate_course(req: CourseGenerationRequest):
     except Exception as e:
         raise HTTPException(500, str(e)) from e
 
+# -------------------------------
+# --- Scheduler Endpoints -------
+# -------------------------------
+@app.get(f"{API_PREFIX}/scheduler/status", response_model=SchedulerStatus, tags=["Scheduler"])
+async def get_scheduler_status_endpoint():
+   """Get the current status of the blog embedder scheduler"""
+   return SchedulerStatus(**get_scheduler_status())
+
+
+@app.post(f"{API_PREFIX}/scheduler/control", tags=["Scheduler"])
+async def control_scheduler(request: SchedulerControl):
+   """Control the blog embedder scheduler (start/stop)"""
+   if request.action == "start":
+       start_scheduler()
+       return {"message": "Scheduler started successfully"}
+   elif request.action == "stop":
+       stop_scheduler()
+       return {"message": "Scheduler stopped successfully"}
+   else:
+       raise HTTPException(status_code=400, detail="Invalid action. Use 'start' or 'stop'")
+
+
+@app.post(f"{API_PREFIX}/scheduler/run-now", tags=["Scheduler"])
+async def run_scheduler_now():
+   """Manually trigger the blog embedder job immediately"""
+   try:
+       # Import here to avoid circular imports
+       from .services.scheduler.scheduler_service import _scheduler
+      
+       # Run the job in a separate thread to avoid blocking
+       import threading
+       def run_job():
+           import asyncio
+           async def async_job():
+               urls = await _scheduler.fetch_freecodecamp_articles()
+               await _scheduler.embed_articles(urls)
+           asyncio.run(async_job())
+      
+       thread = threading.Thread(target=run_job, daemon=True)
+       thread.start()
+      
+       return {"message": "Blog embedder job triggered successfully"}
+   except Exception as e:
+       logger.error(f"Error triggering scheduler job: {e}")
+       raise HTTPException(status_code=500, detail=f"Failed to trigger job: {str(e)}")
+
+
 
 # -------------------------------
 # --------- MAIN ----------------