Hi {name},
Thank you for registering. Please use the following code to verify your email address:
This code will expire in 10 minutes.
''', failure_message="Failed to send verification email", ) except Exception: raise HTTPException(status_code=500, detail="Failed to send verification email") return {"message": "Registration successful. A verification code has been sent to your email."} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.post('/verify-email') async def verify_email(verify_data: VerifyEmailRequest): try: username = verify_data.username code = verify_data.code temp_user = query.get_temp_user_by_username(username) if not temp_user: raise HTTPException(status_code=404, detail="User not found or already verified") if temp_user['verification_code'] != code: raise HTTPException(status_code=400, detail="Invalid verification code") if datetime.utcnow() > temp_user['code_expiration']: raise HTTPException(status_code=400, detail="Verification code has expired") user_data = temp_user.copy() user_data['is_verified'] = True user_data.pop('verification_code', None) user_data.pop('code_expiration', None) user_data.pop('_id', None) query.create_user_from_data(user_data) query.delete_temp_user(username) # Set default language to English query.set_user_language(username, "English") # Set default theme to light (passing false for dark theme) query.set_user_theme(username, False) default_preferences = { 'keywords': True, 'references': True, 'websearch': False, 'personalized_recommendations': True, 'environmental_recommendations': True } query.set_user_preferences(username, default_preferences) return {"message": "Email verification successful"} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.post('/resend-code') async def resend_code(resend_data: ResendCodeRequest): try: username = resend_data.username temp_user = query.get_temp_user_by_username(username) if not temp_user: raise HTTPException(status_code=404, detail="User not found or already verified") verification_code = ''.join(random.choices(string.digits, k=6)) code_expiration = datetime.utcnow() + timedelta(minutes=10) temp_user['verification_code'] = verification_code temp_user['code_expiration'] = code_expiration query.create_or_update_temp_user(username, temp_user['email'], temp_user) try: send_email( to_email=temp_user['email'], subject='Your new verification code', html_content=f'''Hi {temp_user['name']},
You requested a new verification code. Please use the following code to verify your email address:
This code will expire in 10 minutes.
''', failure_message="Failed to send verification email", ) except Exception: raise HTTPException(status_code=500, detail="Failed to send verification email") return {"message": "A new verification code has been sent to your email."} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.post('/checkChatsession') async def check_chatsession(data: ChatSessionCheck, username: str = Depends(get_current_user)): session_id = data.session_id is_chat_exit = query.check_chat_session(session_id) return {"ischatexit": is_chat_exit} @router.get('/check-token') async def check_token(username: str = Depends(get_current_user)): try: return {'valid': True, 'user': username} except Exception as e: raise HTTPException(status_code=401, detail=str(e)) @router.post('/forgot-password') async def forgot_password(data: ForgotPasswordRequest): try: email = data.email user = query.get_user_by_identifier(email) if not user: raise HTTPException(status_code=404, detail="Email not found") reset_token = ''.join(random.choices(string.ascii_letters + string.digits, k=32)) expiration = datetime.utcnow() + timedelta(hours=1) query.store_reset_token(email, reset_token, expiration) reset_link = f"http://localhost:3000/reset-password?token={reset_token}" try: send_email( to_email=email, subject='Reset Your Password', html_content=f'''Hi,
You requested to reset your password. Click the link below to reset it:
This link will expire in 1 hour.
If you didn't request this, please ignore this email.
''', failure_message="Failed to send password reset email", ) except Exception: raise HTTPException(status_code=500, detail="Failed to send password reset email") return {"message": "Password reset instructions sent to email"} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.post('/reset-password') async def reset_password(data: ResetPasswordRequest): try: token = data.token new_password = data.password if not token or not new_password: raise HTTPException(status_code=400, detail="Token and new password are required") reset_info = query.verify_reset_token(token) if not reset_info: raise HTTPException(status_code=400, detail="Invalid or expired reset token") hashed_password = generate_password_hash(new_password) query.update_password(reset_info['email'], hashed_password) return {"message": "Password successfully reset"} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\chat.py ```python # app/routers/chat.py import logging import os import json from datetime import datetime from bson import ObjectId from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Header from fastapi.responses import JSONResponse, FileResponse from pydantic import BaseModel from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user, get_optional_user from app.services.skincare_scheduler import SkinCareScheduler from app.services.wheel import EnvironmentalConditions from app.services.RAG_evaluation import RAGEvaluation router = APIRouter() query = DatabaseQuery() class ChatSessionTitleUpdate(BaseModel): title: str @router.get('/image/{filename}') async def serve_image(filename: str): try: # Use an absolute path or environment variable to ensure consistency upload_dir = os.path.abspath('uploads') file_path = os.path.join(upload_dir, filename) # Add logging to debug print(f"Attempting to serve file from: {file_path}") if not os.path.exists(file_path): print(f"File not found: {file_path}") raise FileNotFoundError() return FileResponse(file_path) except FileNotFoundError: raise HTTPException(status_code=404, detail="Image not found") @router.post('/chat-sessions', status_code=201) async def create_chat_session(username: str = Depends(get_current_user)): try: session_id = str(ObjectId()) chat_session = { "user_id": username, "session_id": session_id, "created_at": datetime.utcnow(), "last_accessed": datetime.utcnow(), "title": "New Chat" } query.create_chat_session(chat_session) return {"message": "Chat session created", "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/chat-sessions') async def get_user_chat_sessions(username: str = Depends(get_current_user)): try: sessions = query.get_user_chat_sessions(username) return sessions except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete('/chat-sessions/{session_id}') async def delete_chat_session(session_id: str, username: str = Depends(get_current_user)): try: result = query.delete_chat_session(session_id, username) if result["session_deleted"]: return { "message": "Chat session and associated chats deleted successfully", "chats_deleted": result["chats_deleted"] } raise HTTPException(status_code=404, detail="Chat session not found or unauthorized") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.put('/chat-sessions/{session_id}/title') async def update_chat_title( session_id: str, title_data: ChatSessionTitleUpdate, username: str = Depends(get_current_user) ): try: new_title = title_data.title if not query.verify_session(session_id, username): raise HTTPException(status_code=404, detail="Chat session not found or unauthorized") if query.update_chat_session_title(session_id, new_title): return { 'message': 'Chat session title updated successfully', 'session_id': session_id, 'new_title': new_title } raise HTTPException(status_code=500, detail="Failed to update chat session title") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete('/chat-sessions/all') async def delete_all_sessions_and_chats(username: str = Depends(get_current_user)): try: result = query.delete_all_user_sessions_and_chats(username) return { "message": "Successfully deleted all chat sessions and chats", "deleted_chats": result["deleted_chats"], "deleted_sessions": result["deleted_sessions"] } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/chats/session/{session_id}') async def get_session_chats(session_id: str, username: str = Depends(get_current_user)): try: chats = query.get_session_chats(session_id, username) return chats except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/export-chat/{session_id}') async def export_chat(session_id: str, username: str = Depends(get_current_user)): try: if not query.verify_session(session_id, username): raise HTTPException(status_code=404, detail="Chat session not found or unauthorized") chats = query.get_session_chats(session_id, username) formatted_chats = [] for chat in chats: formatted_chat = { 'query': chat.get('query', ''), 'response': chat.get('response', ''), 'references': chat.get('references', []), 'page_no': chat.get('page_no', ''), 'date': chat.get('timestamp', ''), 'chat_id': chat.get('chat_id', '') } formatted_chats.append(formatted_chat) export_data = { 'session_id': session_id, 'export_date': datetime.utcnow().isoformat(), 'chats': formatted_chats } return export_data except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/export-all-chats') async def export_all_chats(username: str = Depends(get_current_user)): try: all_chats = query.get_all_user_chats(username) formatted_sessions = [] for session in all_chats: formatted_chats = [] for chat in session['chats']: formatted_chat = { 'query': chat.get('query', ''), 'response': chat.get('response', ''), 'references': chat.get('references', []), 'page_no': chat.get('page_no', ''), 'timestamp': chat.get('timestamp', ''), 'chat_id': chat.get('chat_id', '') } formatted_chats.append(formatted_chat) formatted_session = { 'session_id': session['session_id'], 'title': session['title'], 'created_at': session['created_at'], 'last_accessed': session['last_accessed'], 'chats': formatted_chats } formatted_sessions.append(formatted_session) export_data = { 'user': username, 'export_date': datetime.utcnow().isoformat(), 'sessions': formatted_sessions } return export_data except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post('/report-analysis') async def upload_report( file: UploadFile = File(...), session_id: str = Form(...), authorization: str = Header(None), username: str = Depends(get_current_user) ): try: _ = authorization.split(" ")[1] if not file.filename: return JSONResponse( status_code=400, content={"status": "error", "error": "Empty file provided"} ) file_extension = file.filename.rsplit('.', 1)[1].lower() if '.' in file.filename else '' allowed_extensions = { 'pdf', 'xlsx', 'xls', 'csv', 'jpg', 'jpeg', 'png', 'doc', 'docx', 'ppt', 'pptx', 'txt', 'html' } if file_extension not in allowed_extensions: return JSONResponse( status_code=400, content={ "status": "error", "error": f"Unsupported file type. Allowed types: {', '.join(sorted(allowed_extensions))}", } ) default_upload_root = os.path.abspath( os.path.join(os.path.dirname(__file__), "..", "..", "uploads") ) uploads_root = os.getenv('DERMAI_UPLOAD_DIR', default_upload_root) session_upload_dir = os.path.join(uploads_root, session_id) os.makedirs(session_upload_dir, exist_ok=True) timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S%f') sanitized_name = file.filename.replace(' ', '_') stored_filename = f"{timestamp}_{sanitized_name}" stored_path = os.path.join(session_upload_dir, stored_filename) content = await file.read() with open(stored_path, 'wb') as f: f.write(content) relative_root = os.path.abspath(uploads_root) absolute_path = os.path.abspath(stored_path) relative_path = os.path.relpath(absolute_path, relative_root) return { "status": "success", "message": "File uploaded successfully", "file": { "path": relative_path.replace('\\', '/'), "name": file.filename, "content_type": file.content_type, "size": len(content), "extension": file_extension, } } except Exception as e: logging.error(f"Error in upload_report: {str(e)}") raise HTTPException( status_code=500, detail={ "status": "error", "error": "Internal server error", "details": str(e) } ) @router.get('/skin-care-schedule') async def get_skin_care_schedule( authorization: str = Header(None), username: str = Depends(get_current_user) ): try: token = authorization.split(" ")[1] scheduler = SkinCareScheduler(token, "session_id") schedule = scheduler.createTable() return json.loads(schedule) except Exception as e: logging.error(f"Error generating skin care schedule: {str(e)}") raise HTTPException( status_code=500, detail={"error": "Failed to generate skin care schedule"} ) @router.get('/skin-care-wheel') async def get_skin_care_wheel( authorization: str = Header(...), username: str = Depends(get_current_user) ): try: token = authorization.split(" ")[1] condition = EnvironmentalConditions(session_id=token) condition_data = condition.get_conditon() return condition_data except Exception as e: logging.error(f"Error generating skin care wheel: {str(e)}") raise HTTPException( status_code=500, detail={ "error": "Failed to generate skin care wheel", "message": "An unexpected error occurred" } ) @router.post('/image_disease_search') async def upload_skin_image( image: UploadFile = File(...), session_id: str = Form(...), query: str = Form(""), num_results: int = Form(3), num_images: int = Form(3), authorization: str = Header(...), username: str = Depends(get_current_user) ): try: _ = authorization.split(" ")[1] if not image.filename: return JSONResponse( status_code=400, content={"status": "error", "error": "Empty image file provided"}, ) allowed_extensions = { "jpg", "jpeg", "png", "bmp", "webp", "avif", "avifs", "heic", "heif", } file_extension = ( image.filename.rsplit('.', 1)[1].lower() if '.' in image.filename else '' ) if file_extension not in allowed_extensions: return JSONResponse( status_code=400, content={ "status": "error", "error": ( "Unsupported image type. Allowed types: " f"{', '.join(sorted(allowed_extensions))}" ), }, ) default_upload_root = os.path.abspath( os.path.join(os.path.dirname(__file__), "..", "..", "uploads") ) uploads_root = os.getenv('DERMAI_UPLOAD_DIR', default_upload_root) session_upload_dir = os.path.join(uploads_root, session_id) os.makedirs(session_upload_dir, exist_ok=True) timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S%f') sanitized_name = image.filename.replace(' ', '_') stored_filename = f"{timestamp}_{sanitized_name}" stored_path = os.path.join(session_upload_dir, stored_filename) content = await image.read() with open(stored_path, 'wb') as f: f.write(content) absolute_path = os.path.abspath(stored_path) relative_root = os.path.abspath(uploads_root) relative_path = os.path.relpath(absolute_path, relative_root) return { "status": "success", "message": "Image uploaded successfully", "file": { "path": relative_path.replace('\\', '/'), "name": image.filename, "content_type": image.content_type, "size": len(content), "extension": file_extension, "original_query": query, }, } except Exception as e: logging.error(f"Error in upload_skin_image: {str(e)}") raise HTTPException( status_code=500, detail={ "status": "error", "error": "Internal server error", "details": str(e), }, ) @router.post('/get_rag_evaluation') async def rag_evaluation( page: int = Form(3), page_size: int = Form(3), authorization: str = Header(...), username: str = Depends(get_current_user) ): try: token = authorization.split(" ")[1] evaluator = RAGEvaluation( token=token, page=page, page_size=page_size ) report = evaluator.generate_evaluation_report() return {"response": report} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\chat_session.py ```python # app/routers/chat_session.py from datetime import datetime from bson import ObjectId from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user router = APIRouter() query = DatabaseQuery() class ChatSessionTitleUpdate(BaseModel): title: str @router.post('/chat-sessions', status_code=201) async def create_chat_session(username: str = Depends(get_current_user)): try: session_id = str(ObjectId()) chat_session = { "user_id": username, "session_id": session_id, "created_at": datetime.utcnow(), "last_accessed": datetime.utcnow(), "title": "New Chat" } query.create_chat_session(chat_session) return {"message": "Chat session created", "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/chat-sessions') async def get_user_chat_sessions(username: str = Depends(get_current_user)): try: sessions = query.get_user_chat_sessions(username) return sessions except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete('/chat-sessions/{session_id}') async def delete_chat_session(session_id: str, username: str = Depends(get_current_user)): try: result = query.delete_chat_session(session_id, username) if result["session_deleted"]: return { "message": "Chat session and associated chats deleted successfully", "chats_deleted": result["chats_deleted"] } raise HTTPException(status_code=404, detail="Chat session not found or unauthorized") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.put('/chat-sessions/{session_id}/title') async def update_chat_title( session_id: str, title_data: ChatSessionTitleUpdate, username: str = Depends(get_current_user) ): try: new_title = title_data.title if not query.verify_session(session_id, username): raise HTTPException(status_code=404, detail="Chat session not found or unauthorized") if query.update_chat_session_title(session_id, new_title): return { 'message': 'Chat session title updated successfully', 'session_id': session_id, 'new_title': new_title } raise HTTPException(status_code=500, detail="Failed to update chat session title") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete('/chat-sessions/all') async def delete_all_sessions_and_chats(username: str = Depends(get_current_user)): try: result = query.delete_all_user_sessions_and_chats(username) return { "message": "Successfully deleted all chat sessions and chats", "deleted_chats": result["deleted_chats"], "deleted_sessions": result["deleted_sessions"] } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/chats/session/{session_id}') async def get_session_chats(session_id: str, username: str = Depends(get_current_user)): try: chats = query.get_session_chats(session_id, username) return chats except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\language.py ```python from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from app.middleware.auth import get_current_user from app.database.database_query import DatabaseQuery router = APIRouter() query = DatabaseQuery() class LanguageSettings(BaseModel): language: str @router.post('/language', status_code=201) async def set_language( language_data: LanguageSettings, username: str = Depends(get_current_user) ): try: language = language_data.language if not language: raise HTTPException(status_code=400, detail="Language is required") result = query.set_user_language(username, language) return { "message": "Language set successfully", "language": result["language"] } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/language') async def get_language(username: str = Depends(get_current_user)): try: language = query.get_user_language(username) if language is None: raise HTTPException(status_code=404, detail="Language not set") return { "message": "Language retrieved successfully", "language": language } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete('/language') async def delete_language(username: str = Depends(get_current_user)): try: result = query.delete_user_language(username) if not result: raise HTTPException(status_code=404, detail="Language not found or already deleted") return { "message": "Language deleted successfully" } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\location.py ```python # app/routers/location.py from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user router = APIRouter() query = DatabaseQuery() class LocationData(BaseModel): location: str @router.post('/location', status_code=201) async def add_location(location_data: LocationData, username: str = Depends(get_current_user)): try: location = location_data.location if not location: raise HTTPException(status_code=400, detail="Location is required") query.add_or_update_location(username, location) return {'message': 'Location added/updated successfully'} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/location') async def get_location(username: str = Depends(get_current_user)): try: location_data = query.get_location(username) if not location_data: raise HTTPException(status_code=404, detail="No location found for this user") return {'location': location_data['location']} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\preferences.py ```python # app/routers/preferences.py from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from typing import Dict, Any from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user router = APIRouter() query = DatabaseQuery() class ThemeSettings(BaseModel): theme: bool @router.get('/preferences') async def get_preferences(username: str = Depends(get_current_user)): try: user_preferences = query.get_user_preferences(username) return {'preferences': user_preferences} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post('/preferences') async def set_preferences(preferences: Dict[str, Any], username: str = Depends(get_current_user)): try: preferences_result = query.set_user_preferences(username, preferences) return { 'message': 'Preferences updated successfully', 'preferences': preferences_result } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/theme') async def get_theme(username: str = Depends(get_current_user)): try: user_theme = query.get_user_theme(username) return {'theme': user_theme} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post('/theme') async def set_theme(theme_data: ThemeSettings, username: str = Depends(get_current_user)): try: theme = theme_data.theme theme_data = query.set_user_theme(username, theme) return { 'message': 'Theme updated successfully', 'theme': theme_data['theme'] } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\profile.py ```python from fastapi import APIRouter, Depends, HTTPException, Body from pydantic import BaseModel, EmailStr, validator from typing import Optional from werkzeug.security import generate_password_hash from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user router = APIRouter() query = DatabaseQuery() class ProfileUpdateRequest(BaseModel): email: Optional[EmailStr] = None password: Optional[str] = None name: Optional[str] = None age: Optional[int] = None @validator('password') def password_length(cls, v): if v is not None and len(v) < 6: raise ValueError('Password must be at least 6 characters') return v @validator('age') def age_range(cls, v): if v is not None and (v < 13 or v > 120): raise ValueError('Age must be between 13 and 120') return v @router.get('/profile') async def get_profile(username: str = Depends(get_current_user)): try: user = query.get_user_profile(username) if not user: raise HTTPException(status_code=404, detail="User not found") return { 'username': user['username'], 'email': user['email'], 'name': user['name'], 'age': user['age'], 'created_at': user['created_at'] } except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.put('/profile') async def update_profile( update_data: ProfileUpdateRequest = Body(...), username: str = Depends(get_current_user) ): try: update_fields = {} if update_data.email: if not query.is_valid_email(update_data.email): raise HTTPException(status_code=400, detail="Invalid email format") update_fields['email'] = update_data.email if update_data.password: update_fields['password'] = generate_password_hash(update_data.password) if update_data.name: update_fields['name'] = update_data.name if update_data.age is not None: update_fields['age'] = update_data.age if update_fields: if query.update_user_profile(username, update_fields): return {"message": "Profile updated successfully"} return {"message": "No changes made"} except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.delete('/profile') async def delete_account(username: str = Depends(get_current_user)): try: if query.delete_user_account(username): return {"message": "Account deleted successfully"} raise HTTPException(status_code=404, detail="User not found") except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.delete('/delete-account-permanently') async def delete_account_permanently(username: str = Depends(get_current_user)): try: result = query.delete_account_permanently(username) if result['success']: return { 'message': 'Account and all associated data deleted successfully', 'details': result['deleted_data'] } else: raise HTTPException(status_code=500, detail="Failed to delete account") except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) ``` ### app\routers\questionnaire.py ```python from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from typing import Dict, Any from app.database.database_query import DatabaseQuery from app.middleware.auth import get_current_user router = APIRouter() query = DatabaseQuery() class QuestionnaireSubmission(BaseModel): answers: Dict[str, Any] @router.post('/questionnaires', status_code=201) async def submit_questionnaire( submission: QuestionnaireSubmission, username: str = Depends(get_current_user) ): try: if not submission.answers: raise HTTPException(status_code=400, detail="Answers are required") questionnaire_id = query.submit_questionnaire(username, submission.answers) return { 'message': 'Questionnaire submitted successfully', 'questionnaire_id': questionnaire_id } except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.get('/questionnaires') async def get_questionnaire(username: str = Depends(get_current_user)): try: questionnaire = query.get_latest_questionnaire(username) if not questionnaire: return {'message': 'No questionnaire found', 'data': None} return {'message': 'Success', 'data': questionnaire} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.put('/questionnaires/{questionnaire_id}') async def update_questionnaire( questionnaire_id: str, submission: QuestionnaireSubmission, username: str = Depends(get_current_user) ): try: if not submission.answers: raise HTTPException(status_code=400, detail="Answers are required") if query.update_questionnaire(questionnaire_id, username, submission.answers): return {'message': 'Questionnaire updated successfully'} raise HTTPException( status_code=404, detail='Questionnaire not found or unauthorized' ) except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.delete('/questionnaires/{questionnaire_id}') async def delete_questionnaire( questionnaire_id: str, username: str = Depends(get_current_user) ): try: if query.delete_questionnaire(questionnaire_id, username): return {'message': 'Questionnaire deleted successfully'} raise HTTPException( status_code=404, detail='Questionnaire not found or unauthorized' ) except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException(status_code=500, detail=str(e)) @router.get('/check-answers') async def check_answers(username: str = Depends(get_current_user)): try: answered_count = query.count_answered_questions(username) return {'has_at_least_two_answers': answered_count >= 2} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get('/check-questionnaire') async def check_questionnaire_submission(username: str = Depends(get_current_user)): try: questionnaire = query.get_latest_questionnaire(username) has_questionnaire = questionnaire is not None return {'has_questionnaire': has_questionnaire} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` ### app\services\__init__.py ```python # app/services/__init__.py from app.services.image_classification_vit import SkinDiseaseClassifier from app.services.llm_model import Model from app.services.chathistory import ChatSession from app.services.environmental_condition import EnvironmentalData from app.services.prompts import * from app.services.RAG_evaluation import RAGEvaluation from app.services.skincare_scheduler import SkinCareScheduler from app.services.vector_database_search import VectorDatabaseSearch from app.services.websearch import WebSearch from app.services.wheel import EnvironmentalConditions __all__ = [ "AISkinDetector", "SkinDiseaseClassifier", "Model", "ChatSession", "EnvironmentalData", "RAGEvaluation", "SkinCareScheduler", "VectorDatabaseSearch", "WebSearch", "EnvironmentalConditions", ] ``` ### app\services\agentic_prompt.py ```python from typing import Dict def _append_personalization(prompt: str, user_data: Dict) -> str: personalized_tool = user_data.get('personalized_tool_name') if user_data.get('has_personalized_data') and personalized_tool: prompt += ( "\n\n## Personalized Data Access:\n" f"Call `{personalized_tool}` exactly once after search tools to retrieve patient context. " "IMPORTANT: Analyze how the personalized data relates to the user's current query. " "Don't just list the data - explain how their specific conditions, medications, or history " "impacts the topic they're asking about. Make connections between their profile and the condition." ) else: prompt += ( "\n\nNo personalization data is available; omit the `## Personalization Recommendation` section." ) environmental_tool = user_data.get('environmental_tool_name') if user_data.get('has_environmental_data') and environmental_tool: prompt += ( "\n\n## Environmental Data Access:\n" f"Call `{environmental_tool}` exactly once when relevant. " "CRITICAL: Don't just list environmental statistics. Instead, analyze how the specific " "environmental factors (UV index, humidity, pollution, temperature) affect the skin condition " "the user is asking about. Explain the mechanisms and provide actionable advice based on their location." ) else: prompt += ( "\n\nEnvironmental conditions are unavailable; omit the `## Environmental Condition` section." ) if user_data.get('language', 'english').lower() != 'english': prompt += ( f"\n\nRespond in {user_data.get('language')} while keeping the JSON structure intact." ) return prompt def _append_document_guidance(prompt: str, user_data: Dict) -> str: document_info = user_data.get('document_info') or {} tool_name = user_data.get('document_tool_name') document_path = document_info.get('path') if document_info and tool_name and document_path: document_name = document_info.get('name') or 'Uploaded document' prompt += ( "\n\n## Uploaded Document\n" f"The user has provided `{document_name}` located at `{document_path}`. " f"Call `{tool_name}` exactly once to convert it to Markdown. " "Extract medically relevant findings and incorporate them into your response. " "DO NOT include the document path or name in the references array - only web/vector search results belong there." ) return prompt def _append_image_guidance(prompt: str, user_data: Dict) -> str: image_info = user_data.get('image_info') or {} tool_name = user_data.get('image_tool_name') image_path = image_info.get('path') if image_info and tool_name and image_path: image_name = image_info.get('name') or 'Uploaded image' prompt += ( "\n\n## Uploaded Image\n" f"The user shared `{image_name}` located at `{image_path}`. " f"Call `{tool_name}` exactly once to analyze the skin photo. " "Incorporate the analysis results into your response but DO NOT add image paths to references array. " "References should only contain search result URLs/sources." ) return prompt def _format_json_guidance(user_data: Dict) -> str: references_instruction = ( "Populate `references` ONLY with URLs/sources from get_web_search or get_vector_search results. " "NEVER include uploaded document paths, image paths, or tool names in references. Please use same citation which you will includein response if you get similar citation from web search and vector datat with same book and page number then consider itas singlecitationand usesmae ciation in reponse and in citation/refrences.section" if user_data.get('include_references', True) else "Set `references` to an empty array." ) keywords_instruction = ( "Provide 4-6 concise medical keywords related to the skin condition discussed." if user_data.get('include_keywords', True) else "Set `keywords` to an empty array." ) images_instruction = ( "Include up to 3 image URLs from get_image_search when they help visualize the condition. and If needed image url more than one tool calling then do it together not one by one" if user_data.get('include_images', True) else "Set `images` to an empty array." ) has_personalization = user_data.get('has_personalized_data') has_environmental = user_data.get('has_environmental_data') personalization_instruction = ( "In `## Personalization Recommendation`: Analyze how the user's specific medical history, " "current medications, allergies, and conditions relate to their query. Don't just list their data - " "explain the interactions and provide tailored advice based on their profile." if has_personalization else "Omit the `## Personalization Recommendation` section." ) environmental_instruction = ( "In `## Environmental Condition`: Explain how current environmental factors in their location " "specifically impact the skin condition they're asking about. Provide actionable advice for " "protection and management based on UV levels, humidity, pollution, and temperature." if has_environmental else "Omit the `## Environmental Condition` section." ) return ( "\nYour final response MUST be valid JSON Strictly follow below structure:\n" "{\n" " \"response\": \"Start with `## Response from References` containing evidence-based information with citations [1], [2]. " "Add `## Personalization Recommendation` analyzing how their profile affects this condition. " "Add `## Environmental Condition` explaining environmental impact on their skin.\",\n" " \"references\": [\"ONLY search result URLs/sources, NO file paths\"],\n" " \"keywords\": [\"keyword1\", \"keyword2\", ...],\n" " \"images\": [\"image_urls_from_search\"]\n" "}\n\n" f"{references_instruction}\n" f"{keywords_instruction}\n" f"{images_instruction}\n" f"{personalization_instruction}\n" f"{environmental_instruction}\n\n" "CRITICAL RULES:\n" "- Citations [1], [2] should ONLY reference search results, not uploaded files\n" "- Personalization and Environmental sections must analyze impact, not list data\n" "- Always call get_image_search for visual conditions unless non-medical query\n" ) def get_web_search_prompt(user_data: Dict) -> str: prompt_lines = [ "You are Dr. DermAI, an evidence-based dermatology consultant. Do not hallucination , if you donot get context and not have valid answerjust say I do not know.", "", "## QUERY ASSESSMENT:", "First determine if the query is medical/dermatological. If not, politely decline.", "Please invoke each tool one at a time, waiting for the response before invoking the next. If you need to use the same tool multiple times—for example, for different topics—then call that tool multiple times just vector serach adn web search tool, but still one at a time.", "if there multiple topics then call web search multiple time for each topic" "if the query of user required more than one web search then do it but given answer by combining and use same structure do not strictly give 2 separte json do not make any silly mistake." "## TOOL EXECUTION ORDER FOR MEDICAL QUERIES:", ] step_num = 1 if user_data.get('image_info') and user_data.get('image_tool_name'): prompt_lines.append( f"{step_num}. Call `{user_data.get('image_tool_name')}` FIRST if skin image uploaded" ) step_num += 1 prompt_lines.extend([ f"{step_num}. Call `get_web_search` with the user's query to gather current medical knowledge", f"{step_num + 1}. Call `get_image_search` with relevant terms (e.g., 'eczema rash', 'psoriasis patches') " f"to provide visual references UNLESS the condition is internal or non-visual", ]) step_num += 2 if user_data.get('document_info') and user_data.get('document_tool_name'): prompt_lines.append( f"{step_num}. Call `{user_data.get('document_tool_name')}` to analyze uploaded documents" ) step_num += 1 if user_data.get('has_personalized_data'): prompt_lines.append( f"{step_num}. Call `{user_data.get('personalized_tool_name')}` to get user's medical profile" ) step_num += 1 if user_data.get('has_environmental_data'): prompt_lines.append( f"{step_num}. Call `{user_data.get('environmental_tool_name')}` to get environmental factors" ) step_num += 1 prompt_lines.extend([ f"{step_num}. Synthesize all information into structured JSON response", "", "## IMAGE SEARCH GUIDELINES:", "ALWAYS use get_image_search for:", "- Rashes, lesions, spots, patches, bumps", "- Skin discoloration, texture changes", "- Hair/nail conditions with visible symptoms", "- Any condition where visual reference aids understanding", "", "SKIP get_image_search only for:", "- Non-medical queries", "- Internal conditions without skin manifestations", "- General health questions without visual components", "", "## IMPORTANT NOTES:", "- Maximum 1 call per tool type", "- Web search should focus on recent, evidence-based sources", "- Image search should use specific descriptive terms", "- Personalization must analyze relevance to query, not list profile data", "- Environmental section must explain impact on the specific condition", ]) prompt = "\n".join(prompt_lines) recent_history = user_data.get('recent_history') if recent_history: prompt += ( "\n\n## Recent Conversation:\n" f"{recent_history}\n" ) prompt = _append_document_guidance(prompt, user_data) prompt = _append_image_guidance(prompt, user_data) prompt += _format_json_guidance(user_data) prompt = _append_personalization(prompt, user_data) return prompt.strip() def get_vector_search_prompt(user_data: Dict) -> str: prompt_lines = [ "You are Dr. DermAI with access to a curated dermatology knowledge base. Do not hallucination , if you donot get context and not have valid answerjust say I do not know.", "", "## QUERY ASSESSMENT:", "Determine if medical/dermatological. If not, politely decline.", "Please invoke each tool one at a time, waiting for the response before invoking the next. If you need to use the same tool multiple times—for example, for different topics—then call that tool multiple times just vector serach adn web search tool, but still one at a time.", "if there multiple topics then call web search multiple time for each topic", "if the query of user required more than one vector query search then do it but given answer by combining and use same structure do not strictly give 2 separte json do not make any silly mistake." "## TOOL EXECUTION ORDER FOR MEDICAL QUERIES:", ] step_num = 1 if user_data.get('image_info') and user_data.get('image_tool_name'): prompt_lines.append( f"{step_num}. Call `{user_data.get('image_tool_name')}` FIRST if skin image uploaded" ) step_num += 1 prompt_lines.extend([ f"{step_num}. Call `get_vector_search` to retrieve authoritative medical passages", f"{step_num + 1}. Call `get_image_search` with condition-specific terms " f"(e.g., 'melanoma ABCDE', 'atopic dermatitis flexural') for visual aids", ]) step_num += 2 if user_data.get('document_info') and user_data.get('document_tool_name'): prompt_lines.append( f"{step_num}. Call `{user_data.get('document_tool_name')}` for document analysis" ) step_num += 1 if user_data.get('has_personalized_data'): prompt_lines.append( f"{step_num}. Call `{user_data.get('personalized_tool_name')}` for user profile" ) step_num += 1 if user_data.get('has_environmental_data'): prompt_lines.append( f"{step_num}. Call `{user_data.get('environmental_tool_name')}` for environmental data" ) step_num += 1 prompt_lines.extend([ f"{step_num}. Create comprehensive JSON response", "", "## IMAGE SEARCH REQUIREMENTS:", "MANDATORY for these conditions (use specific medical terms):", "- Inflammatory: psoriasis, eczema, dermatitis, rosacea", "- Infectious: fungal infections, bacterial infections, viral exanthems", "- Neoplastic: melanoma, basal cell, squamous cell, keratosis", "- Pigmentary: melasma, vitiligo, post-inflammatory changes", "- Hair/Nail: alopecia patterns, onychomycosis, nail psoriasis", "", "Use descriptive search terms like:", "- 'plaque psoriasis elbows knees'", "- 'atopic dermatitis infant face'", "- 'tinea corporis ring rash'", "", "## PERSONALIZATION ANALYSIS:", "When using personalized data:", "- Identify drug interactions with current medications", "- Consider contraindications based on health conditions", "- Adjust recommendations for age/pregnancy/allergies", "- Explain how their specific profile affects treatment options", "", "## ENVIRONMENTAL IMPACT ANALYSIS:", "When using environmental data:", "- High UV: Explain photosensitivity risks, sun protection needs", "- Low humidity: Impact on barrier function, moisturization needs", "- High pollution: Oxidative stress, cleansing requirements", "- Temperature extremes: Vasodilation/constriction effects", "", "Never just list the data - explain the mechanisms and provide targeted advice.", ]) prompt = "\n".join(prompt_lines) recent_history = user_data.get('recent_history') if recent_history: prompt += ( "\n\n## Recent Conversation:\n" f"{recent_history}\n" ) prompt = _append_document_guidance(prompt, user_data) prompt = _append_image_guidance(prompt, user_data) prompt += _format_json_guidance(user_data) prompt = _append_personalization(prompt, user_data) return prompt.strip() ``` ### app\services\chathistory.py ```python from app.database.database_query import DatabaseQuery import os import jwt from dotenv import load_dotenv from typing import Optional, Dict, List from bson import ObjectId from datetime import datetime load_dotenv() jwt_secret_key = os.getenv('JWT_SECRET_KEY') query = DatabaseQuery() class ChatSession: def __init__(self, token: str , session_id: str): self.token = token self.session_id = session_id self.chats = [] self.identity = self._decode_token(token) self.query = query def _decode_token(self, token: str) -> str: try: decoded_token = jwt.decode(token, jwt_secret_key, algorithms=["HS256"]) identity = decoded_token['sub'] return identity except jwt.ExpiredSignatureError: raise ValueError("The token has expired.") except jwt.InvalidTokenError: raise ValueError("Invalid token.") except Exception as e: raise ValueError(f"Failed to decode token: {e}") def get_user_preferences(self) -> dict: current_user = self.identity preferences = self.query.get_user_preferences(current_user) if preferences is not None: return preferences raise ValueError("Failed to fetch user preferences.") def get_personalized_recommendation(self) -> Optional[str]: current_user = self.identity response = self.query.get_latest_questionnaire(current_user) if not response: return None answers = response.get('answers', {}) if not answers: return None def format_answer(answer): if answer is None: return None if isinstance(answer, str): stripped_answer = answer.strip().lower() if stripped_answer in ['none', ''] or len(stripped_answer) < 3: return None return stripped_answer if isinstance(answer, list): filtered_answer = [item for item in answer if "Other" not in item and item.strip().lower() not in ['none', ''] and len(item.strip()) >= 3] return ", ".join(filtered_answer) if filtered_answer else None return answer questions = { "skinType": "How would you describe your skin type?", "currentConditions": "Do you currently have any skin conditions?", "autoImmuneConditions": "Do you have a history of autoimmune or hormonal conditions?", "allergies": "Do you have any known allergies to skincare ingredients?", "medications": "Are you currently taking any medications that might affect your skin?", "hormonal": "Do you experience hormonal changes that affect your skin?", "diet": "Have you noticed any foods that trigger skin reactions?", "diabetes": "Do you have diabetes?", "outdoorTime": "How much time do you spend outdoors during the day?", "sleep": "How many hours of sleep do you get on average?", "familyHistory": "Do you have a family history of skin conditions?", "products": "What skincare products are you currently using?" } valid_answers = {key: format_answer(answers.get(key)) for key in questions if format_answer(answers.get(key)) is not None} if not valid_answers: return None formatted_response = [] for key, answer in valid_answers.items(): question = questions.get(key) formatted_response.append(f"question: {question}\nUser answer: {answer}") profile = self.get_profile() name = profile.get('name', 'Unknown') age = profile.get('age', 'Unknown') return f"user name: {name}\nuser age: {age}\n\n" + "\n\n".join(formatted_response) def create_new_session(self, title: str = None) -> bool: current_user = self.identity session_id = str(ObjectId()) chat_session = { "user_id": current_user, "session_id": session_id, "created_at": datetime.utcnow(), "last_accessed": datetime.utcnow(), "title": title if title else "New Chat" } try: self.query.create_chat_session(chat_session) self.session_id = session_id return True except Exception as e: raise Exception(f"Failed to create session: {str(e)}") def verify_session_exists(self, session_id: str) -> bool: current_user = self.identity return self.query.verify_session(session_id, current_user) def validate_session(self, session_id: Optional[str] = None, title: str = None) -> bool: if not session_id or not session_id.strip(): return self.create_new_session(title=title) if self.verify_session_exists(session_id): self.session_id = session_id return self.load_chat_history() return self.create_new_session(title=title) def load_session(self, session_id: str) -> bool: return self.validate_session(session_id) def load_chat_history(self) -> bool: if not self.session_id: raise ValueError("No session ID provided.") current_user = self.identity try: self.chats = self.query.get_session_chats(self.session_id, current_user) return True except Exception as e: raise Exception(f"Failed to load chat history: {str(e)}") def get_chat_history(self) -> List[Dict]: return self.chats def format_history(self) -> str: formatted_chats = [] for chat in self.chats: query = chat.get('query', '').strip() response = chat.get('response', '').strip() if query and response: formatted_chats.append(f"User: {query}") formatted_chats.append(f"dermatologist Dr DermAI: {response}") return "\n".join(formatted_chats) if formatted_chats else "" def save_chat(self, chat_data: Dict) -> bool: if not self.session_id: raise ValueError("No active session to save chat") current_user = self.identity data = { "user_id": current_user, "session_id": self.session_id, "query": chat_data.get("query", "").strip(), "response": chat_data.get("response", "").strip(), "references": chat_data.get("references", []), "page_no": chat_data.get("page_no", []), "keywords": chat_data.get("keywords", []), "images": chat_data.get("images", []), "context": chat_data.get("context", ""), "timestamp": datetime.utcnow(), "chat_id": str(ObjectId()) } try: if self.query.create_chat(data): self.query.update_last_accessed_time(self.session_id) self.chats.append(data) return True return False except Exception as e: raise Exception(f"Failed to save chat: {str(e)}") def get_name_and_age(self): current_user = self.identity try: user_profile = self.query.get_user_profile(current_user) return user_profile except Exception as e: raise Exception(f"Failed to get user name and age: {str(e)}") def get_profile(self): current_user = self.identity try: user = query.get_user_profile(current_user) if not user: return {'error': 'User not found'} return { 'username': user['username'], 'email': user['email'], 'name': user['name'], 'age': user['age'], 'created_at': user['created_at'] } except Exception as e: return {'error': str(e)} def update_title(self , sessionId , new_title): query.update_chat_session_title(sessionId, new_title) def get_city(self) -> Optional[str]: current_user = self.identity try: location_data = self.query.get_location(current_user) if location_data and 'location' in location_data: return location_data['location'] return None except Exception as e: raise Exception(f"Failed to get user city: {str(e)}") def get_language(self) -> Optional[str]: current_user = self.identity try: language = self.query.get_user_language(current_user) if not language : return "english" else: return language return None except Exception as e: raise Exception(f"Failed to get user city: {str(e)}") def get_language(self) -> Optional[str]: current_user = self.identity try: language = self.query.get_user_language(current_user) if not language : return "english" else: return language return None except Exception as e: raise Exception(f"Failed to get user city: {str(e)}") def get_today_schedule(self): data = self.query.get_today_schedule(user_id=self.identity) if not data: return "" return data def save_schedule(self, schedule_data): return self.query.save_schedule(user_id=self.identity, schedule_data=schedule_data) def get_last_seven_days_schedules(self): data = self.query.get_last_seven_days_schedules(user_id=self.identity) if not data: return "" return data def save_details(self, session_id, context, query, response, rag_start_time, rag_end_time): data = self.query.save_rag_interaction( user_id="admin", session_id=session_id, context=context, query=query, response=response, rag_start_time=rag_start_time, rag_end_time=rag_end_time ) return data def get_save_details(self, page: int, page_size: int) -> dict: data = self.query.get_rag_interactions( user_id="admin", page=page, page_size=page_size ) return data def log_user_image_upload(self): """Log an image upload for the current user""" try: return self.query.log_image_upload(self.identity) except Exception as e: raise ValueError(f"Failed to log image upload: {e}") def get_user_daily_uploads(self): """Get number of images uploaded by current user in the last 24 hours""" try: return self.query.get_user_daily_uploads(self.identity) except Exception as e: raise ValueError(f"Failed to get user daily uploads: {e}") def get_user_last_upload_time(self): """Get the timestamp of current user's most recent image upload""" try: return self.query.get_user_last_upload_time(self.identity) except Exception as e: raise ValueError(f"Failed to get user's last upload time: {e}") ``` ### app\services\environmental_condition.py ```python import requests from bs4 import BeautifulSoup class EnvironmentalData: def __init__(self, city): self.city = city self.aqi_url = f"https://api.waqi.info/feed/{city}/?token=466cde4d55e7c5d6cc658ad9c391214b593f46b9" self.uv_url = f"https://www.weatheronline.co.uk/Pakistan/{city}/UVindex.html" def fetch_aqi_data(self): try: response = requests.get(self.aqi_url) data = response.json() if data["status"] == "ok": return { "Temperature": data["data"]["iaqi"].get("t", {}).get("v", "N/A"), "Humidity": data["data"]["iaqi"].get("h", {}).get("v", "N/A"), "Wind Speed": data["data"]["iaqi"].get("w", {}).get("v", "N/A"), "Pressure": data["data"]["iaqi"].get("p", {}).get("v", "N/A"), "AQI": data["data"].get("aqi", "N/A"), "Dominant Pollutant": data["data"].get("dominentpol", "N/A"), } return self.get_default_aqi_data() except: return self.get_default_aqi_data() def get_default_aqi_data(self): return { "Temperature": "N/A", "Humidity": "N/A", "Wind Speed": "N/A", "Pressure": "N/A", "AQI": "N/A", "Dominant Pollutant": "N/A" } def fetch_uv_data(self): try: response = requests.get(self.uv_url) soup = BeautifulSoup(response.text, 'html.parser') gr1_elements = soup.find_all(class_='gr1') if gr1_elements: tr_elements = gr1_elements[0].find_all('tr') if len(tr_elements) > 1: second_tr = tr_elements[1] td_elements = second_tr.find_all('td') if len(td_elements) > 1: return int(td_elements[1].text.strip()) return "N/A" except: return "N/A" def get_environmental_data(self): aqi_data = self.fetch_aqi_data() uv_index = self.fetch_uv_data() environmental_data = { "Temperature": f"{aqi_data['Temperature']} °C" if aqi_data['Temperature'] != "N/A" else "N/A", "Humidity": f"{aqi_data['Humidity']} %" if aqi_data['Humidity'] != "N/A" else "N/A", "Wind Speed": f"{aqi_data['Wind Speed']} m/s" if aqi_data['Wind Speed'] != "N/A" else "N/A", "Pressure": f"{aqi_data['Pressure']} hPa" if aqi_data['Pressure'] != "N/A" else "N/A", "Air Quality Index": aqi_data['AQI'], "Dominant Pollutant": aqi_data["Dominant Pollutant"], "UV_Index": uv_index } return environmental_data ``` ### app\services\google_agent_service.py ```python import asyncio import json import logging import os import re from datetime import datetime, timezone from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Tuple from google.adk.agents import Agent from google.adk.agents.run_config import RunConfig, StreamingMode from google.adk.runners import InMemoryRunner from google.adk.tools import FunctionTool from google.genai import types from app.services.agentic_prompt import ( get_vector_search_prompt, get_web_search_prompt, ) from app.services.chathistory import ChatSession from app.services.environmental_condition import EnvironmentalData from app.services.tools import ( analyze_skin_image, convert_document_to_markdown, get_image_search, get_vector_search, get_web_search, ) logger = logging.getLogger(__name__) GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") DEFAULT_MODEL_NAME = os.getenv("GEMINI_MODEL", "gemini-2.0-flash-exp") PERSONALIZED_TOOL_NAME = "get_personalized_context" ENVIRONMENT_TOOL_NAME = "get_environmental_context" DOCUMENT_CONVERSION_TOOL_NAME = "convert_uploaded_document" IMAGE_ANALYSIS_TOOL_NAME = "analyze_skin_image" if not os.getenv("GOOGLE_API_KEY") and GOOGLE_API_KEY: os.environ["GOOGLE_API_KEY"] = GOOGLE_API_KEY if os.name == "nt": try: asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) except Exception: pass class GoogleAgentService: """Chat orchestrator that streams responses from a Google ADK agent.""" def __init__( self, token: str, session_id: Optional[str] = None, document: Optional[Dict[str, Any]] = None, image: Optional[Dict[str, Any]] = None, ) -> None: self.token = token self.session_id = session_id self.chat_session = ChatSession(token, session_id) self.user_preferences = self._load_user_preferences() self.language = self.chat_session.get_language() or "english" self.user_profile = self._load_user_profile() self.user_city = self.chat_session.get_city() self.environment_data = self._load_environmental_data() self.document = document self.image = image async def process_message_async( self, query: str ) -> AsyncGenerator[Dict[str, Any], None]: if not GOOGLE_API_KEY: error = "Google API key is not configured." logger.error(error) yield {"type": "error", "content": error} return try: session_id = self._ensure_valid_session(query) agent_mode = "web" if self.user_preferences.get("websearch") else "vector" user_data = self._prepare_user_data() agent = self._build_agent(agent_mode, user_data) runner = InMemoryRunner(agent=agent) await runner.session_service.create_session( app_name=runner.app_name, user_id=self.chat_session.identity, session_id=session_id, ) user_message = types.Content( role="user", parts=[types.Part(text=query)], ) run_config = RunConfig(streaming_mode=StreamingMode.SSE) tool_calls: List[Dict[str, Any]] = [] tool_call_map: Dict[str, Dict[str, Any]] = {} collected_images: List[str] = [] collected_references: List[str] = [] streamed_text = "" final_text = "" pending_token_buffer = "" def emit_word_chunks(delta: str, *, final: bool = False) -> List[str]: nonlocal pending_token_buffer pending_token_buffer += delta chunks: List[str] = [] while pending_token_buffer: match = re.search(r'\s', pending_token_buffer) if not match: break idx = match.end() token = pending_token_buffer[:idx] pending_token_buffer = pending_token_buffer[idx:] if token: chunks.append(token) if final and pending_token_buffer: chunks.append(pending_token_buffer) pending_token_buffer = "" return chunks async for event in runner.run_async( user_id=self.chat_session.identity, session_id=session_id, new_message=user_message, run_config=run_config, ): if event.error_message: logger.error("Agent error: %s", event.error_message) yield {"type": "error", "content": event.error_message} return for function_call in event.get_function_calls(): call_entry = { "id": function_call.id, "tool_name": function_call.name, "arguments": function_call.args or {}, } tool_call_map[function_call.id] = call_entry tool_calls.append(call_entry) yield { "type": "tool_call", "id": function_call.id, "tool_name": function_call.name, "arguments": function_call.args or {}, } for function_response in event.get_function_responses(): response_payload = function_response.response or {} call_entry = tool_call_map.get(function_response.id) if call_entry is not None: call_entry["result"] = response_payload if isinstance(response_payload, dict): if function_response.name == "get_image_search": collected_images.extend(response_payload.get("images", [])) if response_payload.get("references"): collected_references.extend(response_payload["references"]) yield { "type": "tool_result", "id": function_response.id, "tool_name": function_response.name, "result": response_payload, } text_segment = self._extract_text(event) if not text_segment: continue if event.partial: streamed_text += text_segment for token in emit_word_chunks(text_segment): yield {"type": "chunk", "content": token} else: final_text = text_segment if streamed_text and text_segment.startswith(streamed_text): delta = text_segment[len(streamed_text) :] else: delta = text_segment if text_segment: streamed_text = text_segment for token in emit_word_chunks(delta, final=True): yield {"type": "chunk", "content": token} for leftover in emit_word_chunks("", final=True): if leftover: yield {"type": "chunk", "content": leftover} parsed_response = self._parse_agent_response(final_text or streamed_text) response_text, keywords, response_images, response_refs = parsed_response merged_images = self._dedupe_list(collected_images + response_images) merged_references = self._dedupe_list(collected_references + response_refs) context_chunks: List[str] = [] if self.document and self.document.get("path"): context_chunks.append(f"document:{self.document.get('path')}") if self.image and self.image.get("path"): context_chunks.append(f"image:{self.image.get('path')}") context_payload = " ".join(context_chunks) chat_payload = { "query": query, "response": response_text, "references": merged_references, "keywords": keywords, "images": merged_images, "context": context_payload, "timestamp": datetime.now(timezone.utc).isoformat(), "session_id": session_id, "tool_calls": tool_calls, } saved = self.chat_session.save_chat(chat_payload) yield { "type": "completed", "saved": saved, "session_id": session_id, "response": response_text, "keywords": keywords, "references": merged_references, "images": merged_images, "tool_calls": tool_calls, } except Exception as exc: logger.error("Agent streaming failure: %s", exc, exc_info=True) yield {"type": "error", "content": f"Generation failed: {exc}"} def _build_agent(self, mode: str, user_data: Dict[str, Any]) -> Agent: prompt = ( get_web_search_prompt(user_data) if mode == "web" else get_vector_search_prompt(user_data) ) search_tool = get_web_search if mode == "web" else get_vector_search tools: List[FunctionTool] = [] if self.image and self.image.get("path"): tools.append(FunctionTool(self._create_image_tool())) tools.extend( [ FunctionTool(search_tool), FunctionTool(get_image_search), ] ) if self.document and self.document.get("path"): tools.append(FunctionTool(self._create_document_tool())) if user_data.get("has_personalized_data"): personalized_tool = self._create_personalized_data_tool( user_data.get("personalized_data", "") ) tools.append(FunctionTool(personalized_tool)) if user_data.get("has_environmental_data"): environmental_tool = self._create_environmental_data_tool( user_data.get("environmental_payload") or {} ) tools.append(FunctionTool(environmental_tool)) agent = Agent( name="DermAI", model=DEFAULT_MODEL_NAME, instruction=prompt, tools=tools, ) return agent def _create_document_tool(self) -> Callable[..., Dict[str, Any]]: document_record = self.document or {} allowed_path = (document_record.get("path") or "").strip() def run_document_conversion( file_path: Optional[str] = None, file_extension: Optional[str] = None, ) -> Dict[str, Any]: target_path = (file_path or allowed_path or "").replace("\\", "/").strip() if not target_path: return { "status": "error", "error_message": "file_path is required to convert a document.", } allowed_normalized = allowed_path.replace("\\", "/").strip() if allowed_normalized and target_path != allowed_normalized: return { "status": "error", "error_message": "The provided file_path does not match the uploaded document for this session.", } result = convert_document_to_markdown( file_path=target_path, file_extension=file_extension or document_record.get("extension"), ) if result.get("status") == "success": text_content = result.get("text_content") or "" result["preview"] = text_content[:1000] result["character_count"] = len(text_content) if allowed_path and result.get("source_path"): # Normalize to relative path for transparency result["source_path"] = allowed_path return result run_document_conversion.__name__ = DOCUMENT_CONVERSION_TOOL_NAME run_document_conversion.__doc__ = ( "Convert the user's uploaded dermatology document into Markdown text. " "Provide the `file_path` exactly as supplied in the conversation context." ) return run_document_conversion def _create_image_tool(self) -> Callable[..., Dict[str, Any]]: image_record = self.image or {} allowed_path = (image_record.get("path") or "").strip() def run_image_analysis( file_path: Optional[str] = None, language: Optional[str] = None, ) -> Dict[str, Any]: target_path = (file_path or allowed_path or "").replace("\\", "/").strip() if not target_path: return { "status": "error", "error_message": "file_path is required to analyse the image.", } allowed_normalized = allowed_path.replace("\\", "/").strip() if allowed_normalized and target_path != allowed_normalized: return { "status": "error", "error_message": "The provided file_path does not match the uploaded image for this session.", } result = analyze_skin_image( file_path=target_path, language=language or self.language, ) if result.get("status") == "success" and allowed_path: result["image_path"] = allowed_path return result run_image_analysis.__name__ = IMAGE_ANALYSIS_TOOL_NAME run_image_analysis.__doc__ = ( "Analyse the user's uploaded skin image. Provide the `file_path` exactly as supplied " "in the conversation context to run the classifier." ) return run_image_analysis def _create_personalized_data_tool(self, data: str) -> Callable[[], Dict[str, Any]]: sanitized = (data or "").strip() def personalized_tool() -> Dict[str, Any]: return { "status": "success", "generated_at": datetime.now(timezone.utc).isoformat(), "personalized_data": sanitized, } personalized_tool.__name__ = PERSONALIZED_TOOL_NAME personalized_tool.__doc__ = ( "Return questionnaire-derived personalization details for the current user." ) return personalized_tool def _create_environmental_data_tool( self, data: Dict[str, Any] ) -> Callable[[], Dict[str, Any]]: snapshot = dict(data) if isinstance(data, dict) else {} city = self.user_city def environmental_tool() -> Dict[str, Any]: return { "status": "success", "city": city, "retrieved_at": datetime.now(timezone.utc).isoformat(), "environmental_data": snapshot, } environmental_tool.__name__ = ENVIRONMENT_TOOL_NAME environmental_tool.__doc__ = ( "Return the cached environmental conditions for the user's location." ) return environmental_tool def _load_user_preferences(self) -> Dict[str, Any]: try: return self.chat_session.get_user_preferences() except Exception as exc: logger.warning("Failed to load user preferences: %s", exc) return { "websearch": False, "keywords": True, "references": True, "personalized_recommendations": False, "environmental_recommendations": False, } def _load_user_profile(self) -> Dict[str, Any]: try: profile = self.chat_session.get_name_and_age() or {} return { "name": profile.get("name", "Patient"), "age": profile.get("age", "Unknown"), } except Exception as exc: logger.warning("Failed to load profile: %s", exc) return {"name": "Patient", "age": "Unknown"} def _load_environmental_data(self) -> Optional[Dict[str, Any]]: try: if ( self.user_preferences.get("environmental_recommendations") and self.user_city ): data = EnvironmentalData(self.user_city).get_environmental_data() if data: return data except Exception as exc: logger.warning("Failed to load environmental data: %s", exc) return None def _load_personalized_data(self) -> str: try: if self.user_preferences.get("personalized_recommendations"): data = self.chat_session.get_personalized_recommendation() return data or "" except Exception as exc: logger.warning("Failed to load personalized data: %s", exc) return "" def _prepare_user_data(self) -> Dict[str, Any]: personalized_data = self._load_personalized_data() environmental_payload = ( dict(self.environment_data) if isinstance(self.environment_data, dict) else {} ) has_personalized_data = bool(personalized_data) has_environmental_data = bool(environmental_payload) document_info = None if self.document and self.document.get("path"): document_info = { "path": self.document.get("path"), "name": self.document.get("name") or "Uploaded document", "type": self.document.get("type"), "extension": self.document.get("extension"), } image_info = None if self.image and self.image.get("path"): image_info = { "path": self.image.get("path"), "name": self.image.get("name") or "Uploaded image", "type": self.image.get("type"), "extension": self.image.get("extension"), "prompt": self.image.get("prompt"), } return { "name": self.user_profile.get("name"), "age": self.user_profile.get("age"), "language": self.language, "personalized_recommendations": self.user_preferences.get( "personalized_recommendations" ), "environmental_recommendations": self.user_preferences.get( "environmental_recommendations" ), "personalized_data": personalized_data, "environmental_data": json.dumps(environmental_payload) if has_environmental_data else "", "has_personalized_data": has_personalized_data, "has_environmental_data": has_environmental_data, "personalized_tool_name": PERSONALIZED_TOOL_NAME if has_personalized_data else None, "environmental_tool_name": ENVIRONMENT_TOOL_NAME if has_environmental_data else None, "environmental_payload": environmental_payload, "include_keywords": self.user_preferences.get("keywords", True), "include_references": self.user_preferences.get("references", True), "include_images": True, "recent_history": self._get_recent_history(), "document_info": document_info, "document_tool_name": DOCUMENT_CONVERSION_TOOL_NAME if document_info else None, "image_info": image_info, "image_tool_name": IMAGE_ANALYSIS_TOOL_NAME if image_info else None, } def _get_recent_history(self, limit: int = 10) -> str: try: if not self.session_id: return "" self.chat_session.load_chat_history() history_items = self.chat_session.get_chat_history() or [] if not history_items: return "" recent = history_items[-limit:] formatted = [] for entry in recent: user_q = entry.get("query") or "" bot_a = entry.get("response") or "" if user_q: formatted.append(f"User: {user_q}") if bot_a: formatted.append(f"Dr DermAI: {bot_a}") return "\n".join(formatted[-limit * 2:]) except Exception as exc: logger.warning("Failed to load recent history: %s", exc) return "" def _ensure_valid_session(self, title: Optional[str] = None) -> str: if not self.session_id or not self.session_id.strip(): self.chat_session.create_new_session(title=title) self.session_id = self.chat_session.session_id else: try: if not self.chat_session.validate_session(self.session_id, title=title): self.chat_session.create_new_session(title=title) self.session_id = self.chat_session.session_id except Exception: self.chat_session.create_new_session(title=title) self.session_id = self.chat_session.session_id return self.session_id @staticmethod def _extract_text(event) -> str: if not event.content or not event.content.parts: return "" parts: List[str] = [] for part in event.content.parts: if part.text: parts.append(part.text) return "".join(parts) @staticmethod def _strip_code_fence(text: str) -> str: stripped = text.strip() if stripped.startswith("```") and stripped.endswith("```"): body = stripped.strip("`") if body.lower().startswith("json"): body = body[4:] stripped = body return stripped.strip() def _parse_agent_response(self, text: str) -> Tuple[str, List[str], List[str], List[str]]: cleaned = self._strip_code_fence(text) if not cleaned: return "", [], [], [] try: payload = json.loads(cleaned) except json.JSONDecodeError: logger.warning("Unable to parse agent JSON response; returning raw text.") return cleaned, [], [], [] if not isinstance(payload, dict): return cleaned, [], [], [] response_text = payload.get("response") or cleaned raw_keywords = payload.get("keywords", []) if isinstance(raw_keywords, list): keywords = [str(item).strip() for item in raw_keywords if str(item).strip()] elif raw_keywords: keywords = [str(raw_keywords).strip()] else: keywords = [] raw_images = payload.get("images", []) if isinstance(raw_images, list): images = [str(item).strip() for item in raw_images if str(item).strip()] elif raw_images: images = [str(raw_images).strip()] else: images = [] raw_refs = payload.get("references", []) if isinstance(raw_refs, list): references = [str(item).strip() for item in raw_refs if str(item).strip()] elif raw_refs: references = [str(raw_refs).strip()] else: references = [] return response_text, keywords, images, references @staticmethod def _dedupe_list(items: List[str]) -> List[str]: seen = set() deduped: List[str] = [] for item in items: if not item: continue if item in seen: continue seen.add(item) deduped.append(item) return deduped ``` ### app\services\image_classification_vit.py ```python import torch from PIL import Image import torch.nn.functional as F from torchvision import transforms from transformers import AutoModelForImageClassification, AutoConfig import requests from io import BytesIO import os from huggingface_hub import hf_hub_download from dotenv import load_dotenv load_dotenv() HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_TOKEN") # Set environment variables for better network handling os.environ['HF_HUB_DOWNLOAD_TIMEOUT'] = '300' # Increase timeout to 5 minutes os.environ['TRANSFORMERS_OFFLINE'] = '0' # Ensure online mode class SkinDiseaseClassifier: CLASS_NAMES = [ "Acne", "Basal Cell Carcinoma", "Benign Keratosis-like Lesions", "Chickenpox", "Eczema", "Healthy Skin", "Measles", "Melanocytic Nevi", "Melanoma", "Monkeypox", "Psoriasis Lichen Planus and related diseases", "Seborrheic Keratoses and other Benign Tumors", "Tinea Ringworm Candidiasis and other Fungal Infections", "Vitiligo", "Warts Molluscum and other Viral Infections" ] def __init__(self, repo_id="muhammadnoman76/skin-disease-classifier", cache_dir=None): self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.repo_id = repo_id self.model = self.load_trained_model() self.transform = self.get_inference_transform() def load_trained_model(self): model_path= hf_hub_download(repo_id=self.repo_id, filename="healthy.pth", token=HUGGINGFACE_TOKEN) checkpoint = torch.load(model_path, map_location=self.device, weights_only=True) classifier_weight = checkpoint['model_state_dict']['classifier.3.weight'] num_classes = classifier_weight.size(0) config = AutoConfig.from_pretrained("google/vit-base-patch16-224-in21k", num_labels=num_classes) model = AutoModelForImageClassification.from_pretrained( "google/vit-base-patch16-224-in21k", config=config, ignore_mismatched_sizes=True ) in_features = model.classifier.in_features model.classifier = torch.nn.Sequential( torch.nn.Linear(in_features, 512), torch.nn.ReLU(), torch.nn.Dropout(0.3), torch.nn.Linear(512, num_classes) ) model.load_state_dict(checkpoint['model_state_dict']) model = model.to(self.device) if self.device.type == 'cuda': model = model.half() model.eval() return model def get_inference_transform(self): return transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), ]) def load_image(self, image_input): try: if isinstance(image_input, Image.Image): image = image_input elif isinstance(image_input, str): if image_input.startswith(('http://', 'https://')): response = requests.get(image_input) image = Image.open(BytesIO(response.content)) else: if not os.path.exists(image_input): raise FileNotFoundError(f"Image file not found: {image_input}") image = Image.open(image_input) elif hasattr(image_input, 'read'): image = Image.open(image_input) else: raise ValueError("Unsupported image input type") return image.convert('RGB') except Exception as e: raise Exception(f"Error loading image: {str(e)}") def predict(self, image_input, confidence_threshold=0.3): try: image = self.load_image(image_input) image_tensor = self.transform(image).unsqueeze(0) if self.device.type == 'cuda': image_tensor = image_tensor.half() image_tensor = image_tensor.to(self.device) with torch.inference_mode(): outputs = self.model(pixel_values=image_tensor).logits probabilities = F.softmax(outputs, dim=1) confidence, predicted = torch.max(probabilities, 1) confidence = confidence.item() predicted_class_idx = predicted.item() confidence_percentage = round(confidence * 100, 2) predicted_class_name = self.CLASS_NAMES[predicted_class_idx] return predicted_class_name, confidence_percentage except Exception as e: raise Exception(f"Error during prediction: {str(e)}") ``` ### app\services\llm_model.py ```python import json from dotenv import load_dotenv import os import re from g4f.client import Client load_dotenv() class Model: def __init__(self): self.gemini_api_key = os.getenv("GEMINI_API_KEY") self.gemini_model = os.getenv("GEMINI_MODEL") # Removed genai client initialization since it's not properly imported def fall_back_llm(self, prompt): """Fallback method using gpt-4o-mini when Gemini fails""" try: response = Client().chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], web_search=False ) return response.choices[0].message.content except Exception as e: return f"Both primary and fallback models failed. Error: {str(e)}" def send_message_openrouter(self, prompt): # Since genai client is not available, use fallback model directly return self.fall_back_llm(prompt) def llm(self, prompt, query): # Since genai client is not available, use fallback model directly combined_content = f"{prompt}\n\n{query}" return self.fall_back_llm(combined_content) def llm_image(self, text, image): # Image processing with LLM is not available without genai client print(f"Error in llm_image: genai client not available") return f"Error: Image processing not available - genai client not configured" def clean_json_response(self, response_text): """Clean the model's response to extract valid JSON.""" start = response_text.find('[') end = response_text.rfind(']') + 1 if start != -1 and end != -1: json_str = re.sub(r",\s*]", "]", response_text[start:end]) return json_str return response_text def skinScheduler(self, prompt, max_retries=3): """Generate a skincare schedule with retries and cleaning.""" # Since genai client is not available, use fallback model directly try: fallback_response = self.fall_back_llm(prompt) cleaned_fallback = self.clean_json_response(fallback_response) return json.loads(cleaned_fallback) except json.JSONDecodeError: return {"error": "Failed to produce valid JSON"} except Exception as e: return {"error": f"Model failed: {str(e)}"} ``` ### app\services\prompts.py ```python SKIN_CARE_SCHEDULER = """As a skincare expert, generate a daily schedule based on: - User's skin profile: {personalized_condition} - Current environmental conditions: {environmental_values} - Historical routines: {historical_data} Create EXACTLY 5 entries in this JSON format: [ {{ "time": "6:00 AM - 8:00 AM", "recommendation": "Cleanse with [Product Name]", "icon": "💧", "category": "morning" }}, {{ "time": "8:00 AM - 10:00 AM", "recommendation": "Apply [Sunscreen Name] SPF 50", "icon": "☀️", "category": "morning" }}, {{ "time": "12:00 PM - 2:00 PM", "recommendation": "Reapply sunscreen", "icon": "🌤️", "category": "afternoon" }}, {{ "time": "6:00 PM - 8:00 PM", "recommendation": "Evening cleansing routine", "icon": "🌙", "category": "evening" }}, {{ "time": "9:00 PM - 11:00 PM", "recommendation": "Night serum application", "icon": "✨", "category": "night" }} ] Important rules: 1. Use only double quotes 2. Maintain category order: morning, morning, afternoon, evening, night 3. Include specific product names from historical data when available 4. Never add comments or text outside the JSON array 5. Time ranges must follow "HH:MM AM/PM - HH:MM AM/PM" format 6. Use appropriate emojis for each activity """ DEFAULT_SCHEDULE = [ { "time": "6:00 AM - 8:00 AM", "recommendation": "Cleanse with a gentle cleanser", "icon": "💧", "category": "Dummy" }, { "time": "8:00 AM - 10:00 AM", "recommendation": "Apply sunscreen SPF 30+", "icon": "☀️", "category": "morning" }, { "time": "12:00 PM - 2:00 PM", "recommendation": "Reapply sunscreen if needed", "icon": "🌤️", "category": "afternoon" }, { "time": "6:00 PM - 8:00 PM", "recommendation": "Evening cleansing routine", "icon": "🌙", "category": "evening" }, { "time": "9:00 PM - 11:00 PM", "recommendation": "Apply night cream or serum", "icon": "✨", "category": "night" } ] ADVICE_REPORT_SUGGESTION = """ ## Based on your Image Analysis: We have identified the presence of {diseases_name} with a confidence level of {diseases_detection_confidence}. {response} """ URDU_ADVICE_REPORT_SUGGESTION = """ ## آپ کی تصویر کے تجزیے کی بنیاد پر: ہم نے {diseases_detection_confidence} کی اعتماد کی سطح کے ساتھ {diseases_name} کی موجودگی کی شناخت کی ہے۔ {response} """ SKIN_NON_SKIN_PROMPT = """ You are an expert at analyzing whether an image shows human skin or not. Your task is to determine if the given image should be processed by a skin disease model. Examine the image carefully and provide a clear two-word response: answer