app.py

import gradio as gr
from huggingface_hub import InferenceClient
import json, re, os, requests
import csv
from io import StringIO

# API Keys from Space settings
hf_token = os.getenv("HF_TOKEN")
google_key = os.getenv("GOOGLE_MAPS_API_KEY")

# Initialize Client
client = InferenceClient("Qwen/Qwen2.5-7B-Instruct", token=hf_token)

SYSTEM_PROMPT = """Analyze the Arabic map query and return ONLY a JSON object with these exact keys:

1. "location": City, area, or "near me" if not specified.
2. "category": The main place type. Use standard Arabic terms:
   - Food: مطعم, مقهى, فرن, مخبز, سوق, بقالة, مطعم شاورما, مطعم برجر
   - Shopping: مول, متجر, سوبرماركت, محل ملابس, صيدلية, محل إلكترونيات, محل أثاث
   - Services: بنك, صراف آلي, محطة وقود, مغسلة, كوافير, ورشة سيارات, مكتب بريد
   - Health: مستشفى, عيادة, صيدلية, طبيب أسنان, مركز صحي, طوارئ
   - Education: مدرسة, جامعة, مكتبة, حضانة, معهد
   - Entertainment: سينما, متحف, حديقة, ملعب, نادي رياضي, شاطئ, حديقة حيوان, منتزه
   - Transportation: مطار, محطة قطار, محطة حافلات, موقف سيارات, تأجير سيارات, وقود
   - Lodging: فندق, شقة فندقية, شاليه, مخيم, بيت ضيافة
   - Religious: مسجد, كنيسة, معبد, جامع
   - Government: ديوان بلدية, محكمة, مركز شرطة, سفارة, مركز حكومي
   - Outdoor: شاطئ, ممر مشاة, بحيرة, غابة, جبل, منتزه طبيعي

3. "sub_type": Specific type within category or null.
4. "features": A list of ALL descriptive tags mentioned.
5. "sort_by": ONE of: rating, price, distance, relevance, open_now.

Examples:
Input: "أرخص وأقرب مستشفى طوارئ في جدة"
Output: {"location": "جدة", "category": "مستشفى", "sub_type": "طوارئ", "features": ["أرخص", "أقرب"], "sort_by": "price"}
Input: "حديقة أطفال نظيفة في الرياض"
Output: {"location": "الرياض", "category": "حديقة", "sub_type": "أطفال", "features": ["نظيفة"], "sort_by": "relevance"}
Input: "أفضل فندق 5 نجوم قريب من المطار في دبي"
Output: {"location": "دبي", "category": "فندق", "sub_type": "5 نجوم", "features": ["أفضل", "قريب من المطار"], "sort_by": "rating"}
Input: "محطة وقود 24 ساعة رخيصة"
Output: {"location": "near me", "category": "محطة وقود", "sub_type": null, "features": ["24 ساعة", "رخيصة"], "sort_by": "price"}
Input: "مقهى هادئ للعمل في الرياض"
Output: {"location": "الرياض", "category": "مقهى", "sub_type": null, "features": ["هادئ", "للعمل"], "sort_by": "relevance"}"""

def build_search_query(parsed_data):
    """بناء جملة بحث محسّنة"""
    parts = []
    
    category = parsed_data.get("category", "")
    if category:
        parts.append(category)
    
    sub_type = parsed_data.get("sub_type")
    if sub_type and sub_type != category:
        parts.append(str(sub_type))
    
    location = parsed_data.get("location", "")
    if location and location != "near me":
        parts.append(f"في {location}")
    
    features = parsed_data.get("features", [])
    search_features = [f for f in features if any(keyword in f for keyword in 
        ["24", "طوارئ", "أطفال", "مجاني", "مفتوح", "فاخر", "اقتصادي", "رخيص"])]
    parts.extend(search_features)
    
    return " ".join(parts)

def search_google_maps_new_api(parsed_data):
    """استخدام Places API (New) v1"""
    if not google_key:
        return "⚠️ الرجاء إضافة GOOGLE_MAPS_API_KEY في إعدادات الـ Space."
    
    query = build_search_query(parsed_data)
    
    if not query.strip():
        query = f"{parsed_data.get("category", "")} {parsed_data.get("location", "")}"
    
    # Places API (New) - Text Search v1
    url = "https://places.googleapis.com/v1/places:searchText"
    
    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": google_key,
        "X-Goog-FieldMask": "places.displayName,places.formattedAddress,places.rating,places.userRatingCount,places.priceLevel,places.currentOpeningHours,places.types,places.location"
    }
    
    data = {
        "textQuery": query,
        "languageCode": "ar",
        "maxResultCount": 10
    }
    
    # إضافة تفضيل الموقع إذا كان الترتيب حسب المسافة
    sort_by = parsed_data.get("sort_by", "relevance")
    if sort_by == "distance" and parsed_data.get("location") != "near me":
        # يمكن إضافة locationBias لاحقاً
        pass
    
    try:
        response = requests.post(url, headers=headers, json=data, timeout=15)
        result = response.json()
        
        if response.status_code != 200:
            error_msg = result.get("error", {}).get("message", "Unknown error")
            # إذا كان الخطأ بسبب عدم تفعيل API، نحاول الـ Legacy كاحتياط
            if "not enabled" in error_msg.lower() or response.status_code == 403:
                return search_google_maps_legacy(parsed_data, query)
            return f"⚠️ خطأ في API: {error_msg}"
        
        places = result.get("places", [])
        
        if not places:
            return "❌ لم يتم العثور على نتائج. جرب تعديل البحث."
        
        # ترتيب النتائج
        if sort_by == "rating":
            places = sorted(places, key=lambda x: x.get("rating", 0), reverse=True)
        elif sort_by == "price":
            price_order = {"PRICE_LEVEL_UNSPECIFIED": 0, "PRICE_LEVEL_FREE": 1, 
                          "PRICE_LEVEL_INEXPENSIVE": 2, "PRICE_LEVEL_MODERATE": 3,
                          "PRICE_LEVEL_EXPENSIVE": 4, "PRICE_LEVEL_VERY_EXPENSIVE": 5}
            places = sorted(places, key=lambda x: price_order.get(x.get("priceLevel"), 99))
        
        # تنسيق النتائج
        output_list = []
        for i, place in enumerate(places[:6], 1):
            name = place.get("displayName", {}).get("text", "غير معروف")
            rating = place.get("rating", "غير متوفر")
            total_ratings = place.get("userRatingCount", 0)
            address = place.get("formattedAddress", "العنوان غير متوفر")
            
            # مستوى السعر
            price_level = place.get("priceLevel")
            price_map = {
                "PRICE_LEVEL_INEXPENSIVE": "💰",
                "PRICE_LEVEL_MODERATE": "💰💰",
                "PRICE_LEVEL_EXPENSIVE": "💰💰💰",
                "PRICE_LEVEL_VERY_EXPENSIVE": "💰💰💰💰"
            }
            price_str = price_map.get(price_level, "")
            
            # حالة الافتتاح
            hours = place.get("currentOpeningHours", {})
            open_now = hours.get("openNow")
            status = ""
            if open_now is True:
                status = " ✅ مفتوح الآن"
            elif open_now is False:
                status = " ❌ مغلق الآن"
            
            # الأنواع
            types = place.get("types", [])
            place_type = ", ".join([t.replace("_", " ").replace("restaurant", "مطعم").replace("cafe", "مقهى") 
                                   .replace("hospital", "مستشفى").replace("park", "حديقة") 
                                   .replace("store", "متجر").replace("gas_station", "محطة وقود") 
                                   for t in types[:2]])
            
            card = f"""### {i}. 📍 {name}
⭐ **التقييم:** {rating}/5 ({total_ratings} تقييم) {price_str}
🏠 **العنوان:** {address}{status}
🏷️ **النوع:** {place_type}
---"""
            output_list.append(card)
        
        summary = f"🔍 **نتائج البحث عن:** `{query}` | **الترتيب حسب:** {sort_by}\n\n"
        return summary + "\n".join(output_list)
        
    except requests.Timeout:
        return "⏱️ انتهى وقت الانتظار. حاول مرة أخرى."
    except Exception as e:
        return f"❌ خطأ في الاتصال: {str(e)}"

def search_google_maps_legacy(parsed_data, query=None):
    """الرجوع إلى Legacy API كاحتياط"""
    if not google_key:
        return "⚠️ الرجاء إضافة GOOGLE_MAPS_API_KEY في إعدادات الـ Space."

    if not query:
        query = build_search_query(parsed_data)
    
    url = "https://maps.googleapis.com/maps/api/place/textsearch/json"
    params = {
        "query": query,
        "language": "ar",
        "key": google_key
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        data = response.json()
        
        if data.get("status") != "OK":
            return f"⚠️ خطأ في Legacy API أيضاً: {data.get("status")}. الرجاء تفعيل Places API (New) في Google Cloud Console."
        
        results = data.get("results", [])
        
        if not results:
            return "❌ لم يتم العثور على نتائج."
        
        sort_by = parsed_data.get("sort_by", "relevance")
        if sort_by == "rating":
            results = sorted(results, key=lambda x: x.get("rating", 0), reverse=True)
        
        output_list = []
        for i, place in enumerate(results[:6], 1):
            name = place.get("name", "غير معروف")
            rating = place.get("rating", "غير متوفر")
            address = place.get("formatted_address", "العنوان غير متوفر")
            
            price_level = place.get("price_level")
            price_str = "💰" * price_level if price_level else ""
            
            open_now = place.get("opening_hours", {}).get("open_now")
            status = " ✅ مفتوح" if open_now else " ❌ مغلق" if open_now is False else ""
            
            card = f"""### {i}. 📍 {name}
⭐ **التقييم:** {rating}/5 {price_str}
🏠 **العنوان:** {address}{status}
---"""
            output_list.append(card)
        
        return "\n".join(output_list)
        
    except Exception as e:
        return f"❌ فشل في Legacy API أيضاً: {str(e)}"

def parse_arabic_query(user_query):
    """تحليل الاستعلام باستخدام LLM"""
    try:
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_query}
        ]
        
        response = client.chat_completion(
            messages=messages,
            max_tokens=300,
            temperature=0.1,
            stream=False
        )
        
        if hasattr(response, 'choices') and response.choices:
            raw_content = response.choices[0].message.content
        elif isinstance(response, dict) and 'choices' in response:
            raw_content = response['choices'][0]['message']['content']
        else:
            raw_content = str(response)
        
        match = re.search(r'\{.*\}', raw_content, re.DOTALL)
        if not match:
            return {"error": "لم يتم العثور على JSON صالح", "raw_response": raw_content[:500]}
        
        parsed_data = json.loads(match.group(0))
        return parsed_data
        
    except Exception as e:
        return {"error": f"خطأ في التحليل: {str(e)}"}

def main_process(user_query):
    if not user_query.strip():
        return {"error": "الرجاء إدخال استعلام"}, "لا توجد نتائج"
    
    parsed_data = parse_arabic_query(user_query)
    
    if "error" in parsed_data:
        return parsed_data, f"⚠️ {parsed_data.get('error', 'خطأ غير معروف')}"
    
    map_results = search_google_maps_new_api(parsed_data)
    
    return parsed_data, map_results

def process_batch_queries(file_obj, progress=gr.Progress()):
    if file_obj is None:
        return None, "❌ الرجاء تحميل ملف للاستعلامات الجماعية."

    queries = []
    try:
        with open(file_obj.name, 'r', encoding='utf-8') as f:
            for line in f:
                query = line.strip()
                if query:
                    queries.append(query)
    except Exception as e:
        return None, f"❌ خطأ في قراءة الملف: {str(e)}"

    if not queries:
        return None, "❌ الملف فارغ أو لا يحتوي على استعلامات صالحة."

    results = []
    for i, query in enumerate(progress.tqdm(queries, desc="معالجة الاستعلامات")):
        parsed_data, map_results = main_process(query)
        
        # Flatten parsed_data for CSV, handle errors
        parsed_json_str = json.dumps(parsed_data, ensure_ascii=False) if "error" not in parsed_data else parsed_data.get("error", "")
        
        # Extract top result name if available, otherwise indicate no results
        top_result_name = ""
        if "نتائج البحث عن" in map_results and "📍" in map_results:
            match = re.search(r'### \d+\. 📍 ([^\n]+)', map_results)
            if match:
                top_result_name = match.group(1).strip()
        elif "لم يتم العثور على نتائج" in map_results:
            top_result_name = "لا توجد نتائج"
        elif "خطأ" in map_results:
            top_result_name = f"خطأ في API: {map_results}"

        results.append({
            "Original Query": query,
            "Parsed JSON": parsed_json_str,
            "Top Map Result": top_result_name,
            "Full Map Results": map_results.replace('\n', '  ') # Replace newlines for CSV readability
        })

    # Write results to a CSV in memory
    output_buffer = StringIO()
    fieldnames = ["Original Query", "Parsed JSON", "Top Map Result", "Full Map Results"]
    writer = csv.DictWriter(output_buffer, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(results)
    
    # Save to a temporary file for Gradio to handle
    output_filepath = "./batch_results.csv"
    with open(output_filepath, 'w', encoding='utf-8') as f:
        f.write(output_buffer.getvalue())

    return output_filepath, "✅ تم الانتهاء من معالجة الدفعة. يمكنك تحميل النتائج."

# ==================== واجهة المستخدم ====================

with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🗺️ Arabic map parser
    
    """)
    
    with gr.Tab("استعلام فردي"):
        with gr.Row():
            input_text = gr.Textbox(
                label="📝 ماذا تبحث عنه؟",
                placeholder="مثال: أرخص مستشفى طوارئ في جدة، أو حديقة أطفال نظيفة في الرياض، أو فندق 5 نجوم في دبي",
                lines=2
            )
        
        btn = gr.Button("🔍 ابحث الآن", variant="primary")
        
        with gr.Row():
            with gr.Column(scale=1):
                output_json = gr.JSON(label="⚙️ تحليل الذكاء الاصطناعي")
            with gr.Column(scale=2):
                output_map = gr.Markdown(label="🗺️ نتائج Google Maps")
        
        gr.Markdown("### 💡 أمثلة للبحث:")
        examples = [
            "أرخص وأقرب مستشفى طوارئ في جدة",
            "حديقة أطفال نظيفة في الرياض",
            "أفضل فندق 5 نجوم في دبي",
            "مقهى هادئ للعمل في الخبر",
            "محطة وقود 24 ساعة رخيصة",
            "صيدلية قريبة مفتوحة الآن"
        ]
        
        with gr.Row():
            for ex in examples[:3]:
                btn_ex = gr.Button(ex, size="sm")
                btn_ex.click(lambda x=ex: x, None, input_text)
        
        with gr.Row():
            for ex in examples[3:]:
                btn_ex = gr.Button(ex, size="sm")
                btn_ex.click(lambda x=ex: x, None, input_text)
        
        btn.click(fn=main_process, inputs=input_text, outputs=[output_json, output_map])

    with gr.Tab("معالجة دفعة"):
        gr.Markdown("""
        ### ⬆️ تحميل ملف الاستعلامات
        قم بتحميل ملف نصي (.txt) أو CSV يحتوي على استعلام واحد في كل سطر.
        """)
        file_input = gr.File(label="ملف الاستعلامات (TXT/CSV)", file_count="single", type="filepath")
        batch_btn = gr.Button("🚀 ابدأ معالجة الدفعة", variant="primary")
        batch_output_file = gr.File(label="ملف نتائج الدفعة (CSV)")
        batch_status_message = gr.Markdown(label="الحالة")

        batch_btn.click(
            fn=process_batch_queries,
            inputs=file_input,
            outputs=[batch_output_file, batch_status_message]
        )

demo.launch()