import os # Hugging Face model caches must be set BEFORE importing anything that loads # transformers / sentence-transformers (which happens transitively via main.py). os.environ["TRANSFORMERS_CACHE"] = "/tmp/model_cache" os.environ["HF_HOME"] = "/tmp/model_cache" os.environ["SENTENCE_TRANSFORMERS_HOME"] = "/tmp/model_cache" os.environ["PYTHONUNBUFFERED"] = "1" print("=== STARTUP: Beginning application initialization ===") print(f"=== STARTUP: PORT environment variable: {os.environ.get('PORT')} ===") # These imports are deliberately late — see env setup above. # noqa: E402 from main import app # noqa: E402 from main import ( # noqa: E402 conversation_memory, document_processor, retrieval_engine, groq_fallback, ) from analytics.conversation_analytics import ConversationAnalytics # noqa: E402 # Phase-2 services from claude_engine import claude_engine # noqa: E402 from opportunities_service import opportunities_service # noqa: E402 from sheet_sync import SheetSync # noqa: E402 from nvidia_fallback import nvidia_fallback # noqa: E402 (free NVIDIA NIM fallback) # Aurora data layer (structured product data + admin dashboard). Separate from # the RAG path above; see db.py / admin_routes.py and schema.sql. import db # noqa: E402 from admin_routes import router as admin_router, public_router as data_router # noqa: E402,E501 from email_briefing import router as email_router # noqa: E402 import email_qa # noqa: E402 (chat answers from the student's own inbox) from fastapi import UploadFile, File, Form, HTTPException, Depends, Request # noqa: E402,E501 from fastapi.responses import StreamingResponse # noqa: E402 from pydantic import Field # noqa: E402 from auth import require_user, require_admin, require_admin_or_token # noqa: E402,E501 from main import limiter # noqa: E402 (shared slowapi Limiter, keyed by client IP) from pydantic import BaseModel # noqa: E402 from typing import Any, Dict, List, Optional # noqa: E402 import asyncio # noqa: E402 import json as _json # noqa: E402 import logging as _logging # noqa: E402 _sync_logger = _logging.getLogger("sheet_sync") # --------------------------------------------------------------------------- # Google Sheet → vector store integration # --------------------------------------------------------------------------- # Knowledge entries from the ALU Google Sheet are pulled in via Apps Script # and upserted into the existing Chroma collection so the Claude RAG path # treats them like any other indexed document. sheet_sync = SheetSync(retrieval_engine) # Mount the Aurora-backed admin/data-layer routes. Self-contained in # admin_routes.py with its own Firebase-derived tenancy guard. We mount each # router under BOTH the bare and /api-prefixed base paths because the v0 # dashboard's exact base URL isn't pinned down — this way it works whether it # calls /admin/* or /api/admin/*. # admin_router -> /admin/* and /api/admin/* (console) # data_router -> /* and /api/db/* (student feed; /api/db avoids # colliding with the legacy Tavily /api/opportunities) app.include_router(admin_router, prefix="/admin") app.include_router(admin_router, prefix="/api/admin") app.include_router(data_router) app.include_router(data_router, prefix="/api/db") # Gmail-powered spoken email briefing (see email_briefing.py). All routes are # Firebase-authed except the Google OAuth callback, which authenticates via a # signed state parameter instead. app.include_router(email_router, prefix="/api/email") SHEET_POLL_INTERVAL_SECONDS = int(os.getenv("SHEET_POLL_INTERVAL_SECONDS", "900")) # 15 min SHEET_REFRESH_TOKEN = os.getenv("SHEET_REFRESH_TOKEN", "") @app.on_event("startup") async def _initial_sheet_sync() -> None: if not sheet_sync.enabled: _sync_logger.info("[sheet_sync] disabled (set KNOWLEDGE_JSON_URL to enable)") return try: result = await asyncio.to_thread(sheet_sync.refresh, False) _sync_logger.info("[sheet_sync] startup sync: %s", result.to_dict()) except Exception as e: _sync_logger.warning("[sheet_sync] startup sync failed: %s", e) async def _poll_loop() -> None: while True: await asyncio.sleep(SHEET_POLL_INTERVAL_SECONDS) try: await asyncio.to_thread(sheet_sync.refresh, False) except Exception as e: _sync_logger.warning("[sheet_sync] poll failed: %s", e) asyncio.create_task(_poll_loop()) # --------------------------------------------------------------------------- # Citation helper — shared by streaming and non-streaming chat endpoints. # Promotes `source_url` (set by sheet_sync) ahead of the legacy `source` # string so the frontend gets a real ALU link to render. # --------------------------------------------------------------------------- def _build_sources(context_docs: Optional[List[Any]]) -> List[Dict[str, Any]]: sources: List[Dict[str, Any]] = [] seen: set = set() for doc in (context_docs or [])[:5]: meta = getattr(doc, "metadata", None) or {} url = (meta.get("source_url") or "").strip() legacy_source = (meta.get("source") or "").strip() # Prefer a real URL; fall back to the legacy free-text source string. link = url if url.startswith("http") else ( legacy_source if legacy_source.startswith("http") else "" ) title = (meta.get("title") or "ALU Knowledge Base").strip() dedupe_key = (title, link or legacy_source) if dedupe_key in seen: continue seen.add(dedupe_key) sources.append( { "title": title, "source": link or legacy_source or "ALU Knowledge Base", "url": link, "department": meta.get("department") or "", "doc_id": meta.get("doc_id"), "origin": meta.get("origin") or "uploaded", } ) if len(sources) >= 3: break return sources # --------------------------------------------------------------------------- # Existing routes from earlier app.py (kept verbatim) # --------------------------------------------------------------------------- # NOTE: /api/alu-events is defined in main.py, which registers first and wins # (app.py does `from main import app`). A handler here would be dead code, so # it's intentionally omitted. See the analytics-dashboard note in main.py for # the same first-registered-wins gotcha. @app.get("/api/analytics/dashboard") async def get_analytics_dashboard(admin=Depends(require_admin)): """Get analytics dashboard data (admin only).""" try: if not conversation_memory: raise HTTPException( status_code=503, detail="Analytics temporarily unavailable" ) analytics = ConversationAnalytics(conversation_memory) dashboard_data = analytics.generate_dashboard_data() return dashboard_data except HTTPException: raise except Exception as e: print(f"Error generating analytics dashboard: {e}") raise HTTPException( status_code=500, detail="Could not generate analytics" ) # --------------------------------------------------------------------------- # Phase 2: Document management routes # Frontend uploads PDF/DOCX/TXT here. The existing DocumentProcessor + # RetrievalEngine handle extraction, chunking, embedding, and indexing. # --------------------------------------------------------------------------- @app.post("/api/documents/upload") @limiter.limit("5/minute") async def upload_document( request: Request, file: UploadFile = File(...), title: Optional[str] = Form(None), admin=Depends(require_admin), ): """ Accept a PDF / DOCX / TXT / MD file, extract text, embed and index it into the existing vector store. Returns the new document id. Admin only — uploads write directly into the knowledge base students are answered from. """ if document_processor is None or retrieval_engine is None: raise HTTPException(status_code=503, detail="Document pipeline unavailable") try: doc_id = await document_processor.process_document(file, title=title, source="frontend-upload") retrieval_engine.update_vector_store(doc_id) return {"id": doc_id, "status": "indexed"} except HTTPException: raise except Exception as e: print(f"[FAIL] Document upload error: {e}") raise HTTPException(status_code=500, detail="Upload failed") @app.get("/api/documents") async def list_documents(admin=Depends(require_admin)): """List all indexed documents (metadata only — no content).""" if document_processor is None: return {"documents": []} docs = document_processor.list_documents() # Strip filesystem paths before returning to clients. sanitized: List[Dict[str, Any]] = [] for d in docs: sanitized.append( { "id": d.get("id"), "title": d.get("title"), "filename": d.get("filename"), "content_type": d.get("content_type"), "source": d.get("source"), "length": d.get("length"), "upload_time": d.get("upload_time"), } ) return {"documents": sanitized} @app.delete("/api/documents/{doc_id}") async def delete_document(doc_id: str, admin=Depends(require_admin)): """Delete a document from storage and remove its chunks from the vector store. Admin only.""" if document_processor is None or retrieval_engine is None: raise HTTPException(status_code=503, detail="Document pipeline unavailable") retrieval_engine.remove_document(doc_id) ok = document_processor.delete_document(doc_id) if not ok: raise HTTPException(status_code=404, detail="Document not found") return {"id": doc_id, "status": "deleted"} # --------------------------------------------------------------------------- # Phase 2: Claude-powered chat (RAG over uploaded docs) # This sits ALONGSIDE the existing /api/chat. The frontend can opt in. # --------------------------------------------------------------------------- class ClaudeChatRequest(BaseModel): # Bounded to keep a single request from running up unbounded LLM cost. message: str = Field(min_length=1, max_length=4000) history: List[Dict[str, str]] = Field(default_factory=list, max_length=40) options: Optional[Dict[str, Any]] = None @app.post("/api/chat/claude") @limiter.limit("20/minute") async def chat_claude( request: Request, body: ClaudeChatRequest, user=Depends(require_user), ): """ RAG-style chat: retrieve from the vector store, then have Claude write the answer grounded in the retrieved context. Falls back to Groq if Claude is unavailable, and returns an error string only as a last resort. """ # Inbox questions ("what did the registrar send me?") are answered from # the student's own Gmail instead of the knowledge base. Cheap regex gate # first; the module falls through (returns None) for non-mailbox questions. if email_qa.is_email_intent(body.message): email_answer = await asyncio.to_thread( email_qa.answer_from_email, user, body.message, body.history ) if email_answer: return email_answer if retrieval_engine is None: raise HTTPException(status_code=503, detail="Retrieval engine unavailable") try: context_docs = retrieval_engine.retrieve_context(query=body.message, role="student") except Exception as e: print(f"[FAIL] Retrieval error: {e}") context_docs = [] sources = _build_sources(context_docs) # Try Claude first. if claude_engine.enabled: answer = claude_engine.generate_response( query=body.message, context_docs=context_docs, history=body.history, ) if answer: return { "response": answer, "sources": sources, "engine": "claude", "model": claude_engine.model, } # Build the SAME rich context Claude got — all retrieved docs formatted with # titles/urls/departments (not one truncated doc) — shared by every fallback. alu_context = claude_engine._format_context(context_docs or []) # First fallback: NVIDIA NIM (stronger free model than Groq/Llama-70B). if nvidia_fallback and nvidia_fallback.enabled: nv_answer = nvidia_fallback.generate_response( body.message, alu_context, history=body.history ) if nv_answer: return { "response": nv_answer, "sources": sources, "engine": "nvidia", "model": nvidia_fallback.model, } # Second fallback: Groq (final backstop) with the same grounded context. if groq_fallback and groq_fallback.enabled: groq_answer = groq_fallback.generate_response( body.message, alu_context, history=body.history ) if groq_answer: return { "response": groq_answer, "sources": sources, "engine": "groq", "model": "llama-3.3-70b-versatile", } return { "response": ( "I'm sorry — the AI service is unavailable right now. " "Please try again in a moment, or contact ALU Support." ), "sources": sources, "engine": "none", } # --------------------------------------------------------------------------- # Phase 3: Streaming chat — Server-Sent Events # --------------------------------------------------------------------------- @app.post("/api/chat/claude/stream") @limiter.limit("20/minute") async def chat_claude_stream( request: Request, body: ClaudeChatRequest, user=Depends(require_user), ): """ Streaming version of /api/chat/claude. Returns a Server-Sent Events (SSE) response. Each event has shape: data: {"type": "chunk", "text": "..."}\\n\\n data: {"type": "sources", "sources": [...]}\\n\\n data: {"type": "done"}\\n\\n On Claude failure the stream emits a single "error" event and the client should fall back to /api/chat/claude (non-streaming) which in turn falls back to Groq. """ def sse_one(event: Dict[str, Any]) -> str: return f"data: {_json.dumps(event, ensure_ascii=False)}\n\n" # Same inbox interception as the non-streaming endpoint: the answer is # generated up front and delivered as a single chunk over SSE. if email_qa.is_email_intent(body.message): email_answer = await asyncio.to_thread( email_qa.answer_from_email, user, body.message, body.history ) if email_answer: def email_events(): yield sse_one({"type": "sources", "sources": email_answer["sources"]}) yield sse_one({"type": "chunk", "text": email_answer["response"]}) yield sse_one({"type": "done"}) return StreamingResponse( email_events(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", }, ) if not claude_engine.enabled: raise HTTPException(status_code=503, detail="Claude is not configured") if retrieval_engine is None: raise HTTPException(status_code=503, detail="Retrieval engine unavailable") # Retrieval is synchronous and happens BEFORE we start streaming so # we can include sources in the first event the client receives. try: context_docs = retrieval_engine.retrieve_context( query=body.message, role="student" ) except Exception as e: print(f"[FAIL] Retrieval error: {e}") context_docs = [] sources = _build_sources(context_docs) def sse_pack(event: Dict[str, Any]) -> str: return f"data: {_json.dumps(event, ensure_ascii=False)}\n\n" def event_generator(): # Send sources first so the UI can render a 'Reading from...' hint. yield sse_pack({"type": "sources", "sources": sources}) emitted_any = False try: for chunk in claude_engine.stream_response( query=body.message, context_docs=context_docs, history=body.history, ): emitted_any = True yield sse_pack({"type": "chunk", "text": chunk}) except Exception as e: print(f"[FAIL] Stream generator error: {e}") yield sse_pack({"type": "error", "message": "Stream interrupted"}) return if not emitted_any: yield sse_pack( {"type": "error", "message": "Claude returned no content"} ) return yield sse_pack({"type": "done"}) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # disable proxy buffering "Connection": "keep-alive", }, ) # --------------------------------------------------------------------------- # Phase 2: Application essay coach # --------------------------------------------------------------------------- class EssayCoachRequest(BaseModel): essay: str = Field(min_length=1, max_length=12000) # the essay question being answered, optional prompt: Optional[str] = Field(default=None, max_length=2000) @app.post("/api/coach/essay") @limiter.limit("5/minute") async def coach_essay( request: Request, body: EssayCoachRequest, user=Depends(require_user), ): """ Give feedback on an ALU application essay draft. Stateless — does NOT use chat history or RAG. Each request is independent. """ if not claude_engine.enabled: raise HTTPException( status_code=503, detail="Essay coach unavailable — Claude is not configured on the server.", ) if len(body.essay.strip()) < 50: raise HTTPException( status_code=400, detail="Essay is too short — please share at least a paragraph.", ) feedback = claude_engine.coach_essay(body.essay, prompt=body.prompt) if not feedback: raise HTTPException(status_code=502, detail="Coach could not produce feedback.") return {"feedback": feedback, "engine": "claude", "model": claude_engine.model} # --------------------------------------------------------------------------- # Phase 2: Opportunities for students (Tavily-backed) # --------------------------------------------------------------------------- @app.get("/api/opportunities") @limiter.limit("10/minute") async def get_opportunities( request: Request, refresh: bool = False, user=Depends(require_user), ): """ Return a list of student opportunities (scholarships, fellowships, internships, etc.). Tavily-powered; falls back to empty list if the service is disabled — the frontend has its own curated list as a secondary fallback. """ opportunities = opportunities_service.get_opportunities(force_refresh=refresh) return { "opportunities": opportunities, "source": "tavily" if opportunities_service.enabled else "disabled", } @app.get("/api/opportunities/status") async def opportunities_status(): return opportunities_service.get_status() # --------------------------------------------------------------------------- # Google Sheet knowledge-base admin endpoints # --------------------------------------------------------------------------- class SheetRefreshRequest(BaseModel): reason: Optional[str] = None force: bool = False @app.post("/api/admin/refresh-sheet") async def refresh_sheet( request: SheetRefreshRequest, caller=Depends(require_admin_or_token), ): """ Re-ingest the Google Sheet knowledge base into the vector store. Called by the Apps Script onEdit trigger (static X-Refresh-Token) so changes propagate within seconds, and also callable by a logged-in admin (Firebase bearer token). Gated by require_admin_or_token. """ result = await asyncio.to_thread(sheet_sync.refresh, request.force) return result.to_dict() @app.get("/api/admin/sheet-status") async def sheet_status(admin=Depends(require_admin)): """Operational view: counts per department + last sync metadata.""" info = sheet_sync.status() info["poll_interval_seconds"] = SHEET_POLL_INTERVAL_SECONDS # Per-department entry counts so staff can confirm their tab landed. by_department: Dict[str, int] = {} collection = getattr(retrieval_engine, "collection", None) if collection is not None: try: existing = collection.get(where={"origin": "sheet"}) for meta in (existing or {}).get("metadatas") or []: dept = (meta or {}).get("department") or "(unknown)" by_department[dept] = by_department.get(dept, 0) + 1 except Exception as e: info["department_count_error"] = str(e) info["by_department"] = by_department return info # --------------------------------------------------------------------------- # Phase 2: extend /health with the new services # --------------------------------------------------------------------------- @app.get("/api/services/status") async def services_status(): return { "claude": claude_engine.get_status(), "opportunities": opportunities_service.get_status(), "nvidia": nvidia_fallback.get_status() if nvidia_fallback else {"enabled": False}, "groq": groq_fallback.get_status() if groq_fallback else {"enabled": False}, "sheet_sync": sheet_sync.status(), "aurora": db.get_status(), "retrieval_engine": retrieval_engine is not None, "document_processor": document_processor is not None, } # This is needed for Hugging Face Spaces if __name__ == "__main__": import uvicorn port = int(os.environ.get("PORT", 7860)) # Hugging Face uses port 7860 print(f"Starting server on port {port}") uvicorn.run(app, host="0.0.0.0", port=port)