fix2

app.py

@@ -2,17 +2,20 @@ import os
 import io
 import re
 import logging
+import asyncio
 from typing import List, Dict, Any
 
 # Core Web Framework and Async Handlers
 from fastapi import FastAPI, Request
+from fastapi.responses import HTMLResponse
 from uvicorn import run as uvicorn_run
 
-# Slack Libraries
-from slack_bolt import App
-from slack_bolt.adapter.fastapi.async_handler import SlackRequestHandler
+# Slack Libraries (Async versions)
+from slack_bolt import AsyncApp
+from slack_bolt.adapter.fastapi import AsyncSlackRequestHandler
 from slack_sdk.oauth.installation_store.async_installation_store import AsyncInstallationStore
 from slack_sdk.oauth import AuthorizeUrlGenerator
+from slack_bolt.models.installation import Installation
 
 # RAG/ML Libraries
 from sentence_transformers import SentenceTransformer
@@ -41,6 +44,12 @@ SLACK_CLIENT_ID = os.environ.get("SLACK_CLIENT_ID")
 SLACK_CLIENT_SECRET = os.environ.get("SLACK_CLIENT_SECRET")
 HF_TOKEN = os.environ.get("HF_TOKEN")
 
+# Check for required environment variables
+required_vars = [SUPABASE_URL, SUPABASE_KEY, SLACK_CLIENT_ID, SLACK_CLIENT_SECRET, SLACK_SIGNING_SECRET]
+if not all(required_vars):
+    missing = [var for var in required_vars if not var]
+    raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
+
 # Set HF_TOKEN if provided (helps with authentication for Hub access)
 if HF_TOKEN:
     try:
@@ -49,27 +58,16 @@ if HF_TOKEN:
     except Exception as e:
         logger.warning(f"Failed to log into Hugging Face Hub: {e}")
 
-# Check for required environment variables
-if not all([SUPABASE_URL, SUPABASE_KEY, SLACK_CLIENT_ID, SLACK_CLIENT_SECRET, SLACK_SIGNING_SECRET]):
-    logger.error("Missing required environment variables (SUPABASE, SLACK Client/Secret/Signing). Exiting.")
-    # In a real app, you might raise an error here
-    # raise EnvironmentError("Missing required environment variables.")
-
-
 # Initialize Supabase client
 supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
 
-# --- Supabase Installation Store (Updated to synchronous for simplicity, but can be Async if needed) ---
-# NOTE: Bolt's `SlackRequestHandler` is often easier to use with synchronous handlers for FastAPI
-# unless you specifically need async in the middleware.
-
-class SupabaseInstallationStore:
+# --- Supabase Async Installation Store ---
+class SupabaseAsyncInstallationStore(AsyncInstallationStore):
     def __init__(self, supabase_client):
         self.supabase = supabase_client
-        # Import Installation here to avoid circular dependency
-        from slack_bolt.models.installation import Installation
 
-    def save(self, installation):
+    async def save(self, installation: Installation) -> None:
+        loop = asyncio.get_event_loop()
         data = {
             "team_id": installation.team_id,
             "bot_token": installation.bot_token,
@@ -77,43 +75,63 @@ class SupabaseInstallationStore:
             "bot_scopes": ",".join(installation.bot_scopes) if installation.bot_scopes else None,
             "app_id": installation.app_id,
         }
-        self.supabase.table("installations").upsert(data, on_conflict="team_id").execute()
-        logger.info(f"Saved installation for team: {installation.team_id}")
-
-    def fetch_installation(self, team_id):
-        result = self.supabase.table("installations").select("*").eq("team_id", team_id).execute()
-        if not result.data:
-            return None
-        
-        # Reconstruct Installation object
-        from slack_bolt.models.installation import Installation
-        data = result.data[0]
-        return Installation(
-            app_id=data.get("app_id"),
-            enterprise_id=None,
-            team_id=team_id,
-            user_id=None,
-            bot_token=data.get("bot_token"),
-            bot_user_id=data.get("bot_user_id"),
-            bot_scopes=data.get("bot_scopes", "").split(',') if data.get("bot_scopes") else [],
-            incoming_webhook=None,
-        )
+        try:
+            await loop.run_in_executor(
+                None,
+                lambda: self.supabase.table("installations").upsert(data, on_conflict="team_id").execute()
+            )
+            logger.info(f"Saved installation for team: {installation.team_id}")
+        except Exception as e:
+            logger.error(f"Failed to save installation for team {installation.team_id}: {e}")
+            raise
 
-    def delete_installation(self, team_id):
-        self.supabase.table("installations").delete().eq("team_id", team_id).execute()
+    async def fetch_installation(self, team_id: str, *, enterprise_id: str | None = None, user_id: str | None = None) -> Installation | None:
+        loop = asyncio.get_event_loop()
+        try:
+            result = await loop.run_in_executor(
+                None,
+                lambda: self.supabase.table("installations").select("*").eq("team_id", team_id).execute()
+            )
+            if not result.data:
+                return None
+            
+            data = result.data[0]
+            return Installation(
+                app_id=data.get("app_id"),
+                enterprise_id=enterprise_id,
+                team_id=team_id,
+                user_id=user_id,
+                bot_token=data.get("bot_token"),
+                bot_user_id=data.get("bot_user_id"),
+                bot_scopes=data.get("bot_scopes", "").split(',') if data.get("bot_scopes") else [],
+                incoming_webhook=None,
+            )
+        except Exception as e:
+            logger.error(f"Failed to fetch installation for team {team_id}: {e}")
+            raise
 
+    async def delete(self, team_id: str, *, enterprise_id: str | None = None, user_id: str | None = None) -> None:
+        loop = asyncio.get_event_loop()
+        try:
+            await loop.run_in_executor(
+                None,
+                lambda: self.supabase.table("installations").delete().eq("team_id", team_id).execute()
+            )
+            logger.info(f"Deleted installation for team: {team_id}")
+        except Exception as e:
+            logger.error(f"Failed to delete installation for team {team_id}: {e}")
+            raise
 
 # Initialize installation store
-installation_store = SupabaseInstallationStore(supabase)
+installation_store = SupabaseAsyncInstallationStore(supabase)
 
-# Initialize Bolt App (Synchronous App with FastAPI Async Handler)
-app = App(
+# Initialize Bolt Async App
+app = AsyncApp(
     client_id=SLACK_CLIENT_ID,
     client_secret=SLACK_CLIENT_SECRET,
     signing_secret=SLACK_SIGNING_SECRET,
     installation_store=installation_store,
-    # NOTE: It's crucial to use the synchronous Bolt App unless all your handlers are async
-    token=SLACK_BOT_TOKEN # Use this for single-workspace setup as a fallback
+    # Removed token for multi-workspace support; add back if single-workspace only
 )
 
 api = FastAPI()
@@ -124,6 +142,8 @@ device = 'cuda' if torch.cuda.is_available() else 'cpu'
 logger.info(f"Using device: {device}")
 
 print("Loading embedding model...")
+embedding_model = None
+qa_pipeline = None
 try:
     embedding_model = SentenceTransformer('all-MiniLM-L6-v2', device=device)
     print("Loading QA model...")
@@ -131,42 +151,51 @@ try:
     print("Models loaded successfully!")
 except Exception as e:
     logger.error(f"Error loading models: {e}")
-    # Consider graceful shutdown or using default models if this fails
-    raise
+    raise RuntimeError("Failed to load required ML models. Check dependencies and hardware.")
 
 # --- Utility Functions ---
 
 def download_slack_file(url: str, token: str) -> bytes:
     """Downloads a file from Slack using the private download URL and bot token."""
-    headers = {"Authorization": f"Bearer {token}"}
-    response = requests.get(url, headers=headers, stream=True)
-    response.raise_for_status()
-    return response.content
+    try:
+        headers = {"Authorization": f"Bearer {token}"}
+        response = requests.get(url, headers=headers, stream=True, timeout=30)
+        response.raise_for_status()
+        return response.content
+    except requests.RequestException as e:
+        logger.error(f"Failed to download Slack file from {url}: {e}")
+        raise
 
 def extract_text_from_pdf(file_content: bytes) -> str:
     """Extracts text from PDF bytes."""
-    pdf_reader = pypdf.PdfReader(io.BytesIO(file_content))
-    text = ""
-    for page in pdf_reader.pages:
-        # Using .extract_text() is standard
-        extracted = page.extract_text()
-        if extracted:
-            text += extracted + "\n"
-    return text
+    try:
+        pdf_reader = pypdf.PdfReader(io.BytesIO(file_content))
+        text = ""
+        for page in pdf_reader.pages:
+            extracted = page.extract_text()
+            if extracted:
+                text += extracted + "\n"
+        return text
+    except Exception as e:
+        logger.error(f"Failed to extract text from PDF: {e}")
+        raise
 
 def extract_text_from_docx(file_content: bytes) -> str:
     """Extracts text from DOCX bytes."""
-    doc = Document(io.BytesIO(file_content))
-    text = ""
-    for paragraph in doc.paragraphs:
-        text += paragraph.text + "\n"
-    return text
+    try:
+        doc = Document(io.BytesIO(file_content))
+        text = ""
+        for paragraph in doc.paragraphs:
+            text += paragraph.text + "\n"
+        return text
+    except Exception as e:
+        logger.error(f"Failed to extract text from DOCX: {e}")
+        raise
 
 def chunk_text(text: str, chunk_size: int = 300) -> List[str]:
     """Chunks text by word count to create manageable RAG chunks."""
     words = text.split()
     chunks = []
-    # Using a simple word-split chunking
     for i in range(0, len(words), chunk_size):
         chunk = " ".join(words[i:i + chunk_size])
         if chunk.strip():
@@ -175,28 +204,35 @@ def chunk_text(text: str, chunk_size: int = 300) -> List[str]:
 
 def embed_text(text: str) -> List[float]:
     """Generates an embedding for a piece of text."""
-    # Use convert_to_tensor=True for better performance, then convert back for Supabase
+    if embedding_model is None:
+        raise RuntimeError("Embedding model not loaded.")
     embedding = embedding_model.encode(text, convert_to_tensor=False)
     return embedding.tolist()
 
 def store_embeddings(chunks: List[str]):
     """Stores text chunks and their embeddings in Supabase."""
+    loop = asyncio.get_event_loop()
     for chunk in chunks:
         embedding = embed_text(chunk)
         try:
-            # Using the upsert method for idempotent inserts (though a unique ID per chunk is better)
-            supabase.table("documents").insert({
-                "content": chunk,
-                "embedding": embedding
-            }).execute()
+            loop.run_in_executor(
+                None,
+                lambda c=chunk, e=embedding: supabase.table("documents").insert({
+                    "content": c,
+                    "embedding": e
+                }).execute()
+            )
         except Exception as e:
             logger.error(f"Failed to insert chunk into Supabase: {str(e)}")
 
 def is_table_empty() -> bool:
     """Checks if the documents table has any records."""
     try:
-        # Use simple count query
-        result = supabase.table("documents").select("id", count="exact").limit(1).execute()
+        loop = asyncio.get_event_loop()
+        result = loop.run_in_executor(
+            None,
+            lambda: supabase.table("documents").select("id", count="exact").limit(1).execute()
+        )
         return result.count == 0
     except Exception as e:
         logger.error(f"Error checking table emptiness: {str(e)}")
@@ -204,13 +240,18 @@ def is_table_empty() -> bool:
 
 def search_documents(query: str, match_count: int = 5) -> List[Dict[str, Any]]:
     """Searches Supabase for documents matching the query using vector similarity."""
+    if embedding_model is None:
+        raise RuntimeError("Embedding model not loaded.")
     query_embedding = embed_text(query)
     try:
-        # Assumes you have a 'match_documents' function in your Supabase database
-        result = supabase.rpc("match_documents", {
-            "query_embedding": query_embedding,
-            "match_count": match_count
-        }).execute()
+        loop = asyncio.get_event_loop()
+        result = loop.run_in_executor(
+            None,
+            lambda: supabase.rpc("match_documents", {
+                "query_embedding": query_embedding,
+                "match_count": match_count
+            }).execute()
+        )
         return result.data
     except Exception as e:
         logger.error(f"Error searching documents for query '{query}': {str(e)}")
@@ -218,10 +259,11 @@ def search_documents(query: str, match_count: int = 5) -> List[Dict[str, Any]]:
 
 def answer_question(question: str, context: str) -> str:
     """Answers a question based on the provided context using the QA pipeline."""
+    if qa_pipeline is None:
+        raise RuntimeError("QA pipeline not loaded.")
     if not context.strip():
         return "No relevant documents found."
     try:
-        # QA models have a context token limit (e.g., 512 for roberta-base); we limit to be safe.
         MAX_CONTEXT_LEN = 4096 
         context_slice = context[:MAX_CONTEXT_LEN]
         
@@ -231,24 +273,23 @@ def answer_question(question: str, context: str) -> str:
         logger.error(f"Error in QA pipeline: {str(e)}")
         return "Error generating answer from context."
 
-# --- Slack Handlers ---
+# --- Slack Handlers (Async) ---
 
 @app.event("file_shared")
-def handle_file_shared(event, say, client):
+async def handle_file_shared(event, say, client):
     """Processes files shared in a channel and adds their content to the RAG knowledge base."""
     file_id = event["file_id"]
     try:
-        file_info = client.files_info(file=file_id)
+        file_info = await client.files_info(file=file_id)
         file_data = file_info["file"]
         
         file_type = file_data.get("mimetype", "")
         file_url = file_data.get("url_private_download")
         
         if not file_url:
-            say("No download URL available for the file.")
+            await say("No download URL available for the file.")
             return
         
-        # Get the bot token dynamically for the workspace
         token = client.token 
         file_content = download_slack_file(file_url, token)
         
@@ -258,23 +299,23 @@ def handle_file_shared(event, say, client):
         elif "wordprocessingml" in file_type or "msword" in file_type:
             text = extract_text_from_docx(file_content)
         else:
-            say("Unsupported file type. Please upload PDF or DOCX files.")
+            await say("Unsupported file type. Please upload PDF or DOCX files.")
             return
         
         if not text.strip():
-            say("No text could be extracted from the file.")
+            await say("No text could be extracted from the file.")
             return
         
         chunks = chunk_text(text)
         store_embeddings(chunks)
         
-        say(f"✅ File processed successfully! Added **{len(chunks)}** chunks to the knowledge base.")
+        await say(f"✅ File processed successfully! Added **{len(chunks)}** chunks to the knowledge base.")
     except Exception as e:
         logger.error(f"Error in file_shared handler for file_id {file_id}: {str(e)}")
-        say(f"❌ Error processing file: {str(e)}")
+        await say(f"❌ Error processing file: {str(e)}")
 
 @app.event("app_mention")
-def handle_mention(event, say, client):
+async def handle_mention(event, say, client):
     """Handles mentions (@bot) for answering questions using RAG."""
     text = event["text"]
     user_query = re.sub(r'<@[A-Z0-9]+>', '', text).strip()
@@ -282,7 +323,7 @@ def handle_mention(event, say, client):
     # Simple check for the presence of a file in the message for combined upload/query
     files = event.get("files", [])
     if files:
-        say("I see a file attached. I'll process it first, and then answer your question.")
+        await say("I see a file attached. I'll process it first, and then answer your question.")
         # NOTE: Handling files in app_mention is complex due to timing/race conditions. 
         # For simplicity, we delegate file processing primarily to file_shared event.
         # This loop is kept as a fallback/redundancy but the file_shared event is more reliable.
@@ -291,19 +332,19 @@ def handle_mention(event, say, client):
         pass # The file processing logic from the original code would go here
     
     if not user_query:
-        say("Please ask me a question after mentioning me!")
+        await say("Please ask me a question after mentioning me!")
         return
     
     try:
         # Check if table is empty
-        if is_table_empty():
-            say("My knowledge base is empty. Please share some PDF or DOCX files first so I can learn!")
+        if await asyncio.to_thread(is_table_empty):
+            await say("My knowledge base is empty. Please share some PDF or DOCX files first so I can learn!")
             return
         
         results = search_documents(user_query, match_count=5)
         
         if not results:
-            say("I couldn't find any relevant information in my knowledge base. Try uploading more documents.")
+            await say("I couldn't find any relevant information in my knowledge base. Try uploading more documents.")
             return
         
         # Combine the content of the top documents into a single context string
@@ -311,21 +352,24 @@ def handle_mention(event, say, client):
         answer = answer_question(user_query, context)
         
         # Format the final response
-        say(f"💡 *Answer:* {answer}\n\n_I found this information by analyzing {len(results)} relevant document chunks._")
+        await say(f"💡 *Answer:* {answer}\n\n_I found this information by analyzing {len(results)} relevant document chunks._")
     except Exception as e:
         logger.error(f"Error in app_mention handler for query '{user_query}': {str(e)}")
-        say(f"❌ Error answering question: {str(e)}")
+        await say(f"❌ Error answering question: {str(e)}")
 
 # --- FastAPI Integration ---
 
-# Initialize the Slack Request Handler with the Bolt App
-# We use the synchronous handler as the App is synchronous
-handler = SlackRequestHandler(app)
+# Initialize the Async Slack Request Handler
+handler = AsyncSlackRequestHandler(app)
 
 @api.post("/slack/events")
 async def slack_events(request: Request):
     """Endpoint for all Slack event subscriptions."""
-    return await handler.handle(request)
+    try:
+        return await handler.handle_async(request)
+    except Exception as e:
+        logger.error(f"Error handling Slack events: {e}")
+        raise
 
 @api.get("/")
 async def root():
@@ -335,40 +379,55 @@ async def root():
 @api.get("/health")
 async def health():
     """Health check endpoint."""
-    # Add a check for Supabase connectivity here for a robust health check
+    db_status = "error"
     try:
-        supabase.table("installations").select("team_id").limit(1).execute()
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(
+            None,
+            lambda: supabase.table("installations").select("team_id").limit(1).execute()
+        )
         db_status = "ok"
-    except Exception:
-        db_status = "error"
-        
-    return {"status": "ok", "database": db_status}
+    except Exception as e:
+        logger.error(f"Database health check failed: {e}")
+    
+    models_status = "ok" if embedding_model and qa_pipeline else "error"
+    return {"status": "ok" if db_status == "ok" and models_status == "ok" else "error", "database": db_status, "models": models_status}
 
 @api.get("/slack/install")
 async def install_url():
     """Generates the Slack installation URL."""
-    generator = AuthorizeUrlGenerator(
-        client_id=SLACK_CLIENT_ID,
-        scopes=["app_mentions:read", "files:read", "chat:write", "im:read", "im:write", "channels:read"]
-    )
-    # NOTE: The redirect_uri should match your Slack App config
-    return {"install_url": generator.generate(state="state")}
+    try:
+        generator = AuthorizeUrlGenerator(
+            client_id=SLACK_CLIENT_ID,
+            scopes=["app_mentions:read", "files:read", "chat:write", "im:read", "im:write", "channels:read"]
+        )
+        # NOTE: The redirect_uri should match your Slack App config (e.g., https://yourspace.hf.space/slack/oauth/callback)
+        url = generator.generate(state="state")
+        return {"install_url": url}
+    except Exception as e:
+        logger.error(f"Error generating install URL: {e}")
+        raise
 
 @api.get("/slack/oauth/callback")
 async def oauth_callback(request: Request):
     """Handles the OAuth callback from Slack to complete installation."""
     try:
-        # Use Bolt's handler to process the OAuth flow
-        # The handler automatically uses the `installation_store` defined in the App object
-        response = await handler.handle(request)
-        # Simple HTML response indicating success
-        return f"<html><body><h1>Installation successful!</h1><p>You can now use the bot in your Slack workspace.</p></body></html>"
+        response = await handler.handle_async(request)
+        # For successful OAuth, return a simple HTML response
+        if response.status_code == 200:
+            return HTMLResponse(
+                content="<html><body><h1>Installation successful!</h1><p>You can now use the bot in your Slack workspace.</p></body></html>",
+                status_code=200
+            )
+        return response
     except Exception as e:
         logger.error(f"OAuth Callback Error: {e}")
-        return f"<html><body><h1>Installation Failed!</h1><p>Error: {str(e)}</p></body></html>"
+        return HTMLResponse(
+            content=f"<html><body><h1>Installation Failed!</h1><p>Error: {str(e)}</p></body></html>",
+            status_code=500
+        )
 
 
 if __name__ == "__main__":
-    # Use uvicorn_run for proper program execution
     port = int(os.environ.get("PORT", 7860))
     uvicorn_run(api, host="0.0.0.0", port=port)