app.py

import pandas as pd
import gradio as gr
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from sentence_transformers import SentenceTransformer, util
from difflib import get_close_matches
import requests
import io
import torch

# GitHub Excel file URL
GITHUB_EXCEL_URL = "https://raw.githubusercontent.com/3Zzamt0o/HospitalData/main/Hospital%20Data%20(1).xlsx"

def load_hospital_data():
    try:
        # Fetch Excel file from GitHub
        response = requests.get(GITHUB_EXCEL_URL)
        response.raise_for_status()  # Raise an exception for bad status codes
        
        # Read Excel file from the response content
        excel_data = io.BytesIO(response.content)
        df = pd.read_excel(excel_data)
        
        print("Successfully loaded data from GitHub")
        return df
    except Exception as e:
        print(f"Error loading data from GitHub: {e}")
        # Fallback to default data if GitHub fetch fails
        return pd.DataFrame({
            'العيادات': ['عيادة الأسنان', 'عيادة الباطنة', 'عيادة العظام', 'عيادة الأطفال'],
            'الدكتور': ['د. أحمد محمد', 'د. محمد علي', 'د. علي حسن', 'د. سارة أحمد'],
            'الأيام': ['السبت والاثنين', 'الأحد والثلاثاء', 'الاثنين والأربعاء', 'الثلاثاء والخميس'],
            'سعر الكشف': ['100 جنيه', '150 جنيه', '120 جنيه', '130 جنيه']
        })

# Load the hospital data
print("Loading hospital data...")
df = load_hospital_data()

# Q&A data for lab tests and equipment
data = {
    "السؤال": [
        "ما هي مواعيد معمل التحاليل ؟", "ما هي المده المستغرقه لعمل التحاليل ؟",
        "ما هي انواع التحاليل الموجوده و المتوفره ؟", "هل يوجد تحليل صوره دم  ؟",
        "ماهي تحاليل صوره الدم ؟", "ما هي تحاليل السكر  ؟", "ما هي تحاليل وظائف الكبد ؟",
        "ما هي تحاليل وظائف الكلي ؟", "ما هي تحاليل البول و البراز ؟",
        "ما هي تحاليل الدهون  ؟", "ما هي تحاليل الألتهابات و الروماتيزم   ؟",
        "ما هي تحاليل الفيروسات   ؟", "ما هي تحاليل الغده الدرقيه ؟",
        "هل يوجد حشو اسنان ؟", "هل يوجد جهاز سونار  ؟", "هل يوجد اجهزه تنفس اصطناعي  ؟",
        "هل يوجد قياس نظر  ؟", "ما هي الاجهزه المتاحه في عياده الرمد ؟",
        "هل يوجد قياس ضغط ؟", "هل يوجد سونار  ؟", "هل يوجد رسم قلب  ؟"
    ],
    "الجواب": [
        "طول ايام الاسبوع من 9 صباحا الي 9 مسائا", "من 24 ساعه الي 48 ساعه",
        "تحاليل صوره الدم، تحاليل سكر، تحاليل وظائف الكبد، تحاليل وظائف الكلي",
        "نعم يوجد جميع انواع التحاليل بالمستشفي", "CBC، Hb%(Heamoglobin)، Blood Group، Rh",
        "Rbs سكر عشوائي، سكر صائم FBS، PPbs سكر فاطر", "ALT(GPT)، AST(GOT)، Billirubin",
        "Ceatinin الكرياتنين، Urea اليوريا، Urine البول، Uric acid اليوريك اسد",
        "Urine البول، Stool البراز، جرثومه المعده في البراز", "Cholesterol كوليسترول، Triglyceride الدهون الثلاثيه",
        "CRP، ASOT", "HCV Ab، GBS Ag، HIC Ab", "Tsh، T3، t4",
        "نعم يوجد حشو أسنان", "نعم يوجد جهاز سونار", "نعم يوجد أجهزة تنفس إصطناعي",
        "نعم يوجد قياس نظر", "جهاز قاع العين، جهاز قياس النظر، جهاز الكشف من الحساسية والالتهابات",
        "نعم يوجد قياس ضغط", "نعم يوجد سونار", "نعم يوجد رسم قلب"
    ]
}

qa_data = pd.DataFrame(data)

# Medical knowledge base
medical_kb = {
    "الجفاف": "في حالة الشعور بالجفاف: 1- شرب الكثير من السوائل 2- تناول الأطعمة الغنية بالماء 3- تجنب التعرض المباشر للشمس 4- استشارة الطبيب إذا استمرت الأعراض",
    "الصداع": "لعلاج الصداع: 1- الراحة في مكان هادئ 2- شرب الماء بكثرة 3- تناول مسكن خفيف 4- تجنب الضوء القوي والضوضاء",
    "التعب": "للتغلب على التعب: 1- الحصول على قسط كافٍ من النوم 2- تناول غذاء صحي متوازن 3- ممارسة تمارين خفيفة 4- تجنب الإجهاد",
    "الحمى": "عند الإصابة بالحمى: 1- الراحة التامة 2- شرب السوائل بكثرة 3- خفض درجة الحرارة بالكمادات 4- استشارة الطبيب إذا تجاوزت 39 درجة",
    "السعال": "لتخفيف السعال: 1- شرب الماء الدافئ مع العسل 2- استخدام مرطب الهواء 3- تجنب المثيرات 4- الراحة وتناول الأدوية المناسبة",
    "ألم المعدة": "لعلاج ألم المعدة: 1- تجنب الأطعمة الدهنية والحارة 2- تناول وجبات خفيفة 3- شرب مشروبات دافئة كاليانسون 4- استشارة الطبيب إذا استمر الألم",
    "الإسهال": "لعلاج الإسهال: 1- شرب محلول معالجة الجفاف 2- تجنب الأطعمة الدسمة 3- تناول وجبات خفيفة وسهلة الهضم 4- مراجعة الطبيب إذا استمر لأكثر من يومين",
    "الإمساك": "لعلاج الإمساك: 1- تناول الألياف مثل الخضروات والفواكه 2- شرب الماء بكثرة 3- ممارسة النشاط البدني 4- استخدام ملينات طبيعية إذا لزم الأمر",
    "الزكام": "لعلاج الزكام: 1- الراحة 2- شرب السوائل الساخنة 3- استخدام بخاخات الأنف 4- تجنب البرد والتيارات الهوائية",
    "ألم الظهر": "لتخفيف ألم الظهر: 1- تجنب الجلوس لفترات طويلة 2- ممارسة تمارين التمدد 3- استخدام كمادات دافئة 4- مراجعة الطبيب في حال استمرار الألم"
}

# Initialize models and embeddings
print("Loading Arabic model and tokenizer...")
try:
    model_id = "aubmindlab/bert-base-arabertv2"
    tokenizer = AutoTokenizer.from_pretrained(model_id)
    model = AutoModelForSequenceClassification.from_pretrained(model_id)
    print("Arabic model loaded successfully")
except Exception as e:
    print(f"Warning: Could not load Arabic model: {e}")
    tokenizer = None
    model = None

print("Loading sentence transformer...")
try:
    embedder = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
    print("Sentence transformer loaded successfully")
except Exception as e:
    print(f"Error loading sentence transformer: {e}")
    embedder = None

# Prepare passages from hospital data
passages = []
for _, row in df.iterrows():
    if pd.isna(row['الدكتور']) and pd.isna(row['العيادات']):
        continue
    text = f"العيادة: {row['العيادات']}، الدكتور: {row['الدكتور']}، الأيام: {row['الأيام']}، السعر: {row['سعر الكشف']}"
    passages.append(text)

# Create embeddings for passages
if embedder:
    try:
        corpus_embeddings = embedder.encode(passages, convert_to_tensor=True)
        print("Corpus embeddings created successfully")
    except Exception as e:
        print(f"Error creating embeddings: {e}")
        corpus_embeddings = None
else:
    corpus_embeddings = None

def check_lab_or_radiology(question):
    keywords = ['أشعة', 'تحاليل', 'رنين', 'سونار', 'تحليل', 'مختبر', 'معمل']
    if any(word in question for word in keywords):
        try:
            matches = get_close_matches(question, qa_data['السؤال'].tolist(), n=1, cutoff=0.5)
            if matches:
                matched_q = matches[0]
                answer = qa_data[qa_data['السؤال'] == matched_q]['الجواب'].values[0]
                return answer
        except Exception as e:
            print(f"Error in lab/radiology check: {e}")
    return None

def answer_question_from_excel(user_question, k=3):
    # First check if it's a medical symptom question
    for symptom, advice in medical_kb.items():
        if symptom in user_question or any(word in user_question for word in ["اشعر", "اعاني", "الم", "وجع", "مريض"]):
            return f"🩺 **الرد:** {advice}\n\n⚠️ **ملاحظة هامة:** هذه النصائح عامة، يرجى استشارة الطبيب للتشخيص الدقيق."

    # Then check in qa_data for lab tests and equipment
    qa_answer = check_lab_or_radiology(user_question)
    if qa_answer:
        return f"💡 **الرد:** {qa_answer}"

    # Define clinic keywords with variations
    clinic_mapping = {
        "عيادة الباطنة": ["باطنة", "الباطنة", "باطنه"],
        "عيادة الجلدية": ["جلد", "جلدية", "الجلدية"],
        "عيادة المسالك": ["مسالك", "المسالك", "مسالك بولية"],
        "عيادة النساء": ["نسا", "نساء", "توليد", "ولادة"],
        "عيادة الأنف والأذن": ["انف", "اذن", "حنجرة", "الأنف", "الأذن"],
        "عيادة الرمد": ["رمد", "الرمد", "عيون", "العيون"],
        "عيادة الأسنان": ["اسنان", "الاسنان", "الأسنان", "سنان"],
        "الحضانة": ["حضان", "حضانة", "الحضانة", "حضانات"],
        "المعمل": ["معمل", "المعمل", "تحاليل", "مختبر"],
        "عيادة الأطفال": ["اطفال", "الاطفال", "الأطفال"],
        "عيادة الجهاز الهضمي": ["هضم", "هضمي", "الجهاز الهضمي"],
        "عيادة التخاطب": ["تخاطب", "التخاطب", "نطق"]
    }

    # Check for specific clinic mentions
    target_clinic = None
    for clinic, keywords in clinic_mapping.items():
        if any(keyword in user_question for keyword in keywords):
            target_clinic = clinic
            break

    # If a specific clinic is mentioned, show all information for that clinic
    if target_clinic:
        clinic_info = [p for p in passages if target_clinic in p]
        if clinic_info:
            return f"🏥 **الرد:** \n" + "\n".join(clinic_info)

    # For questions about doctors or clinics
    if any(word in user_question for word in ["عيادة", "عيادات", "دكتور", "دكاترة", "دكتورة", "مواعيد"]):
        # Check if asking about a specific type of clinic
        for clinic_type in ["الأسنان", "الباطنة", "الجلدية", "المسالك", "النساء", "الأنف", "الرمد", "الحضانة", "المعمل", "الأطفال", "الجهاز", "التخاطب"]:
            if clinic_type in user_question:
                clinic_info = [p for p in passages if clinic_type in p]
                if clinic_info:
                    return f"🏥 **الرد:** \n" + "\n".join(clinic_info)
        
        # If no specific clinic type mentioned, return all clinic information
        return f"💡 **الرد:** \n" + "\n".join([p for p in passages if 'عيادة' in p])
    
    # If no specific clinic found, use semantic search if available
    if embedder and corpus_embeddings is not None:
        try:
            question_embedding = embedder.encode(user_question, convert_to_tensor=True)
            hits = util.semantic_search(question_embedding, corpus_embeddings, top_k=k)
            matched_passages = [passages[hit['corpus_id']] for hit in hits[0]]
            
            if matched_passages:
                hospital_info = "\n".join(matched_passages)
                return f"💡 **الرد:** {hospital_info}"
        except Exception as e:
            print(f"Error in semantic search: {e}")
    
    # If no match found in any source, return a helpful message
    return f"💡 **الرد:** عذراً، لا يمكنني العثور على إجابة محددة لسؤالك. يرجى استشارة الطبيب للحصول على المشورة الطبية المناسبة."

def answer_question(user_question):
    try:
        answer = check_lab_or_radiology(user_question)
        if answer:
            return f"💡 **الرد:** {answer}"
        return answer_question_from_excel(user_question)
    except Exception as e:
        return f"⚠️ خطأ: {str(e)}"

def ask_question(question):
    """
    Function to handle question answering with error handling
    """
    try:
        if not question or question.strip() == "":
            return "❌ خطأ: السؤال مطلوب"
        
        answer = answer_question(question.strip())
        return answer
    except Exception as e:
        return f"❌ خطأ: {str(e)}"

# Create Gradio interface
with gr.Blocks(
    title="نظام الأسئلة والأجوبة الطبي",
    theme=gr.themes.Soft(),
    css="""
    .gradio-container {
        font-family: 'Arial', sans-serif;
        direction: rtl;
        text-align: right;
    }
    .gr-button {
        background: linear-gradient(45deg, #4CAF50, #45a049);
        color: white;
        font-weight: bold;
    }
    .gr-textbox {
        text-align: right;
        direction: rtl;
    }
    """
) as demo:
    
    gr.Markdown(
        """
        # 🏥 نظام الأسئلة والأجوبة الطبي
        ### اسأل أي سؤال واحصل على إجابة ذكية!
        
        يمكنك السؤال عن:
        - 🏥 مواعيد العيادات والأطباء
        - 🧪 التحاليل الطبية والمعمل
        - 🩺 النصائح الطبية العامة
        - 📋 أسعار الكشف والخدمات
        """,
        elem_id="header"
    )
    
    with gr.Row():
        with gr.Column(scale=4):
            question_input = gr.Textbox(
                label="سؤالك",
                placeholder="اكتب سؤالك هنا...",
                lines=3,
                text_align="right",
                rtl=True
            )
        with gr.Column(scale=1):
            submit_btn = gr.Button(
                "اسأل السؤال", 
                variant="primary",
                size="lg"
            )
    
    answer_output = gr.Textbox(
        label="الإجابة",
        lines=10,
        interactive=False,
        text_align="right",
        rtl=True
    )
    
    # Example questions
    gr.Examples(
        examples=[
            ["ما هي مواعيد عيادة الأسنان؟"],
            ["هل يوجد تحليل صورة دم؟"],
            ["علاج الحمى"],
            ["مين دكاترة الأطفال وميعادهم؟"],
            ["ما هي تحاليل السكر؟"]
        ],
        inputs=question_input,
        label="أمثلة على الأسئلة"
    )
    
    # Set up the interaction
    submit_btn.click(
        fn=ask_question,
        inputs=question_input,
        outputs=answer_output
    )
    
    # Also allow Enter key to submit
    question_input.submit(
        fn=ask_question,
        inputs=question_input,
        outputs=answer_output
    )

if __name__ == "__main__":
    print("Starting Hospital Q&A System...")
    # Launch the Gradio app
    demo.launch(
        server_name="0.0.0.0",  # Equivalent to Flask's host='0.0.0.0'
        server_port=7860,       # Default Gradio port (you can change to 5000 if preferred)
        share=False,            # Set to True if you want a public link
        debug=True,
        show_error=True
    )