Update app.py

app.py

@@ -13,10 +13,7 @@ from passlib.context import CryptContext
 from passlib.exc import UnknownHashError
 from pydantic import BaseModel
 
-# ======================
-# Configuration
-# ======================
-
+# Configure logging
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
@@ -29,50 +26,35 @@ pwd_context = CryptContext(
     deprecated="auto"
 )
 
-# Environment variables (fallback to hardcoded values if not set)
+# Configuration
 BACKEND_URL = os.getenv("BACKEND_URL", "https://rocketfarmstudios-cps-api.hf.space")
 ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "yakdhanali97@gmail.com")
 ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "123456")
 
-# ======================
-# Models
-# ======================
-
-class DoctorCreate(BaseModel):
-    full_name: str
-    email: str
-    license_number: str
-    password: str
-    specialty: str
-
-# ======================
-# Token Management
-# ======================
-
 class TokenManager:
     def __init__(self):
-        self.token: Optional[str] = None
-        self.last_refresh: float = 0
-        self.expires_in: int = 3600  # Default expiry
+        self.token = None
+        self.last_refresh = 0
+        self.expires_in = 3600
         self.lock = asyncio.Lock()
+        self.initialized = False
+        self.available = False
 
     async def _make_login_request(self) -> Optional[str]:
-        """Make the login request with proper error handling"""
-        login_payload = {
-            "username": ADMIN_EMAIL,
-            "password": ADMIN_PASSWORD,
-            "device_token": "admin-console"
-        }
-
         try:
             async with aiohttp.ClientSession() as session:
                 async with session.post(
                     f"{BACKEND_URL}/auth/login",
-                    json=login_payload,
+                    json={
+                        "username": ADMIN_EMAIL,
+                        "password": ADMIN_PASSWORD,
+                        "device_token": "admin-console"
+                    },
                     timeout=aiohttp.ClientTimeout(total=10)
                 ) as response:
                     if response.status == 200:
                         data = await response.json()
+                        self.available = True
                         return data.get("access_token")
                     else:
                         error = await response.text()
@@ -82,9 +64,8 @@ class TokenManager:
             logger.error(f"Login request error: {str(e)}")
             return None
 
-    async def refresh_token(self) -> str:
-        """Obtain a new token with retry logic"""
-        async with self.lock:  # Prevent multiple concurrent refreshes
+    async def refresh_token(self) -> Optional[str]:
+        async with self.lock:
             for attempt in range(3):
                 token = await self._make_login_request()
                 if token:
@@ -93,34 +74,23 @@ class TokenManager:
                     logger.info("Successfully refreshed admin token")
                     return token
                 
-                wait_time = min(5, (attempt + 1) * 2)  # Exponential backoff with max 5s
+                wait_time = min(5, (attempt + 1) * 2)
                 logger.warning(f"Attempt {attempt + 1} failed, retrying in {wait_time}s...")
                 await asyncio.sleep(wait_time)
 
-            raise HTTPException(
-                status_code=500,
-                detail="Failed to obtain admin token after multiple attempts"
-            )
+            self.available = False
+            return None
 
-    async def get_token(self) -> str:
-        """Get current valid token, refreshing if needed"""
-        if not self.token or (time.time() - self.last_refresh) > (self.expires_in - 60):
+    async def get_token(self) -> Optional[str]:
+        if not self.initialized or not self.token or (time.time() - self.last_refresh) > (self.expires_in - 60):
+            self.initialized = True
             return await self.refresh_token()
         return self.token
 
 token_manager = TokenManager()
 
-# ======================
-# FastAPI App
-# ======================
+app = FastAPI(title="Medical Admin Portal")
 
-app = FastAPI(
-    title="Medical Admin Portal",
-    description="Doctor account management system",
-    version="1.0.0"
-)
-
-# CORS Configuration
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -129,27 +99,12 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-# ======================
-# Security Utilities
-# ======================
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    try:
-        return pwd_context.verify(plain_password, hashed_password)
-    except UnknownHashError:
-        logger.error(f"Unrecognized hash format: {hashed_password[:15]}...")
-        return False
-    except Exception as e:
-        logger.error(f"Password verification error: {str(e)}")
-        return False
-
-# ======================
-# API Endpoints
-# ======================
-
 @app.get("/")
 async def root():
-    return {"status": "active", "service": "Medical Admin Portal"}
+    return {
+        "status": "running",
+        "admin_api_available": token_manager.available
+    }
 
 @app.post("/login")
 async def handle_login():
@@ -157,77 +112,80 @@ async def handle_login():
 
 @app.get("/health")
 async def health_check():
-    try:
-        # Verify we can get a token
-        await token_manager.get_token()
-        return {"status": "healthy"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+    return {
+        "status": "healthy",
+        "admin_api": "available" if token_manager.available else "unavailable"
+    }
 
-# ======================
-# Doctor Management
-# ======================
+async def create_doctor_api(doctor_data: Dict[str, Any]) -> Dict[str, Any]:
+    if not token_manager.available:
+        return {
+            "success": False,
+            "error": "Admin API currently unavailable"
+        }
 
-async def create_doctor_api(doctor_data: DoctorCreate) -> Dict[str, Any]:
-    """Core doctor creation logic"""
     token = await token_manager.get_token()
-    
+    if not token:
+        return {
+            "success": False,
+            "error": "Failed to obtain admin token"
+        }
+
     headers = {
         "Authorization": f"Bearer {token}",
         "Content-Type": "application/json"
     }
 
-    async with aiohttp.ClientSession() as session:
-        async with session.post(
-            f"{BACKEND_URL}/auth/admin/doctors",
-            json=doctor_data.dict(),
-            headers=headers,
-            timeout=aiohttp.ClientTimeout(total=15)
-        ) as response:
-            
-            if response.status == 201:
-                return {"success": True, "message": "Doctor created successfully"}
-            
-            error_detail = await response.text()
-            return {
-                "success": False,
-                "error": f"Error {response.status}: {error_detail}"
-            }
-
-# ======================
-# Gradio Interface
-# ======================
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                f"{BACKEND_URL}/auth/admin/doctors",
+                json=doctor_data,
+                headers=headers,
+                timeout=aiohttp.ClientTimeout(total=15)
+            ) as response:
+                if response.status == 201:
+                    return {"success": True, "message": "Doctor created successfully"}
+                
+                error_detail = await response.text()
+                return {
+                    "success": False,
+                    "error": f"Error {response.status}: {error_detail}"
+                }
+    except Exception as e:
+        logger.error(f"Doctor creation API error: {str(e)}")
+        return {
+            "success": False,
+            "error": f"API communication error: {str(e)}"
+        }
 
 def sync_create_doctor(full_name: str, email: str, matricule: str, password: str, specialty: str) -> str:
-    """Wrapper to run async code in Gradio's sync context"""
     try:
-        doctor_data = DoctorCreate(
-            full_name=full_name,
-            email=email,
-            license_number=matricule,
-            password=password,
-            specialty=specialty
-        )
+        doctor_data = {
+            "full_name": full_name,
+            "email": email,
+            "license_number": matricule,
+            "password": password,
+            "specialty": specialty
+        }
         
-        # Run async code in sync context
         result = asyncio.run(create_doctor_api(doctor_data))
         
         if result["success"]:
             return "✅ " + result["message"]
-        else:
-            return "❌ " + result["error"]
+        return "❌ " + result.get("error", "Unknown error")
             
     except Exception as e:
         logger.error(f"Doctor creation failed: {str(e)}")
         return f"❌ System error: {str(e)}"
 
-with gr.Blocks(title="Doctor Management", css=".gradio-container {max-width: 800px}") as admin_ui:
+with gr.Blocks(title="Doctor Management") as admin_ui:
     gr.Markdown("# 👨‍⚕️ Doctor Account Creator")
     
     with gr.Row():
         with gr.Column():
-            full_name = gr.Textbox(label="Full Name", placeholder="Dr. First Last")
-            email = gr.Textbox(label="Email", placeholder="doctor@hospital.org")
+            full_name = gr.Textbox(label="Full Name")
+            email = gr.Textbox(label="Email")
             license_num = gr.Textbox(label="License Number")
             
         with gr.Column():
@@ -238,13 +196,9 @@ with gr.Blocks(title="Doctor Management", css=".gradio-container {max-width: 800
                     "Pediatrics", "Orthopedics", "Dermatology"
                 ]
             )
-            password = gr.Textbox(
-                label="Password",
-                type="password",
-                info="Minimum 8 characters"
-            )
+            password = gr.Textbox(label="Password", type="password")
     
-    submit_btn = gr.Button("Create Account", variant="primary")
+    submit_btn = gr.Button("Create Account")
     output = gr.Textbox(label="Status", interactive=False)
     
     submit_btn.click(
@@ -253,20 +207,10 @@ with gr.Blocks(title="Doctor Management", css=".gradio-container {max-width: 800
         outputs=output
     )
 
-# Mount Gradio interface
 app = gr.mount_gradio_app(app, admin_ui, path="/admin")
 
-# ======================
-# Admin Dashboard
-# ======================
-
 @app.get("/admin-auth")
-async def admin_auth(
-    email: str,
-    password: str,
-    response: Response
-):
-    """Secure admin dashboard entry point"""
+async def admin_auth(email: str, password: str, response: Response):
     if (email != ADMIN_EMAIL or 
         not verify_password(password, pwd_context.hash(ADMIN_PASSWORD))):
         response.status_code = 401
@@ -276,24 +220,29 @@ async def admin_auth(
         """)
     return RedirectResponse(url="/admin")
 
-# ======================
-# Startup Event
-# ======================
-
 @app.on_event("startup")
 async def startup_event():
-    """Verify we can connect to the backend on startup"""
+    """Try to connect to backend but don't fail startup"""
     try:
         logger.info("Testing backend connection...")
         await token_manager.get_token()
-        logger.info("Backend connection successful")
+        if token_manager.available:
+            logger.info("Backend connection successful")
+        else:
+            logger.warning("Backend connection failed - running in degraded mode")
     except Exception as e:
-        logger.error(f"Startup connection test failed: {str(e)}")
-        raise
+        logger.error(f"Startup connection test error: {str(e)}")
+        token_manager.available = False
 
-# ======================
-# Main Application
-# ======================
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    try:
+        return pwd_context.verify(plain_password, hashed_password)
+    except UnknownHashError:
+        logger.error(f"Unrecognized hash format")
+        return False
+    except Exception as e:
+        logger.error(f"Password verification error: {str(e)}")
+        return False
 
 if __name__ == "__main__":
     import uvicorn
@@ -301,6 +250,5 @@ if __name__ == "__main__":
         app,
         host="0.0.0.0",
         port=7860,
-        log_level="info",
-        timeout_keep_alive=60
+        log_level="info"
     )