| import gradio as gr |
| import sqlite3 |
| import hashlib |
| import html |
| import re |
| import os |
| import json |
| import logging |
| from datetime import datetime, timedelta |
| from huggingface_hub import HfApi, hf_hub_download |
| import shutil |
|
|
| |
| |
| |
|
|
| DB_NAME = "testimonials.db" |
| REPO_ID = os.getenv("SPACE_ID", "your-username/your-space-name") |
| TOKEN = os.getenv("HF_TOKEN") |
|
|
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger(__name__) |
|
|
| |
| |
| |
|
|
| def download_db_from_hub(): |
| """تحميل قاعدة البيانات من Hugging Face Hub""" |
| try: |
| if not TOKEN: |
| return False |
| local_path = hf_hub_download( |
| repo_id=REPO_ID, |
| filename=DB_NAME, |
| token=TOKEN, |
| local_dir="./", |
| local_dir_use_symlinks=False |
| ) |
| if local_path != DB_NAME: |
| shutil.copy2(local_path, DB_NAME) |
| logger.info("تم تحميل قاعدة البيانات من Hub") |
| return True |
| except Exception as e: |
| logger.warning(f"لم يتم العثور على قاعدة بيانات: {e}") |
| return False |
|
|
| def upload_db_to_hub(): |
| """رفع قاعدة البيانات إلى Hugging Face Hub""" |
| try: |
| if not TOKEN or not os.path.exists(DB_NAME): |
| return False |
| api = HfApi() |
| api.upload_file( |
| path_or_fileobj=DB_NAME, |
| path_in_repo=DB_NAME, |
| repo_id=REPO_ID, |
| token=TOKEN, |
| repo_type="space" |
| ) |
| logger.info("تم رفع قاعدة البيانات إلى Hub") |
| return True |
| except Exception as e: |
| logger.error(f"فشل رفع قاعدة البيانات: {e}") |
| return False |
|
|
| |
| |
| |
|
|
| def get_connection(): |
| return sqlite3.connect(DB_NAME, check_same_thread=False) |
|
|
| def init_db(): |
| conn = get_connection() |
| c = conn.cursor() |
| |
| c.execute(""" |
| CREATE TABLE IF NOT EXISTS testimonials ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| name TEXT, |
| role TEXT, |
| content TEXT NOT NULL, |
| sentiment TEXT, |
| rating INTEGER DEFAULT 5, |
| timestamp TEXT NOT NULL, |
| likes INTEGER DEFAULT 0, |
| is_highlighted INTEGER DEFAULT 0, |
| ip_hash TEXT |
| ) |
| """) |
| |
| c.execute(""" |
| CREATE TABLE IF NOT EXISTS likes ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| testimonial_id INTEGER, |
| ip_hash TEXT, |
| created_at TEXT, |
| FOREIGN KEY (testimonial_id) REFERENCES testimonials(id) ON DELETE CASCADE |
| ) |
| """) |
| |
| |
| c.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON testimonials(timestamp)") |
| c.execute("CREATE INDEX IF NOT EXISTS idx_ip_hash ON testimonials(ip_hash)") |
| c.execute("CREATE INDEX IF NOT EXISTS idx_highlighted ON testimonials(is_highlighted)") |
| c.execute("CREATE INDEX IF NOT EXISTS idx_rating ON testimonials(rating)") |
| |
| conn.commit() |
| conn.close() |
| logger.info("تم تهيئة قاعدة البيانات") |
|
|
| |
| |
| |
|
|
| def clean_text(text): |
| if not text: |
| return "" |
| text = text.strip() |
| text = re.sub(r"\s+", " ", text) |
| return text |
|
|
| def contains_bad_words(text): |
| bad_words = ["يلعن", "تبن", "سافل", "كلب", "خنزير", "عاهرة", "قحبة"] |
| pattern = r'\b(' + '|'.join(bad_words) + r')\b' |
| return bool(re.search(pattern, text.lower())) |
|
|
| def escape_safe(text): |
| return html.escape(text) if text else "" |
|
|
| def get_ip_hash(request: gr.Request): |
| try: |
| if request and hasattr(request, 'client') and request.client: |
| return hashlib.sha256(request.client.host.encode()).hexdigest()[:16] |
| except: |
| pass |
| return "unknown" |
|
|
| def detect_sentiment(text): |
| positive = ["رائع", "ممتاز", "جميل", "احترافي", "مفيد", "مذهل", "أفضل", "متميز"] |
| negative = ["سيء", "ضعيف", "فاشل", "مخيب", "فظيع", "تافه"] |
| |
| text_lower = text.lower() |
| pos_score = sum(1 for w in positive if w in text_lower) |
| neg_score = sum(1 for w in negative if w in text_lower) |
| |
| if pos_score > neg_score: |
| return "إيجابي 😊" |
| elif neg_score > pos_score: |
| return "سلبي 😕" |
| return "محايد 😐" |
|
|
| def format_time(timestamp): |
| try: |
| dt = datetime.fromisoformat(timestamp) |
| diff = datetime.now() - dt |
| |
| if diff.days > 7: |
| return dt.strftime("%Y/%m/%d") |
| elif diff.days > 0: |
| return f"منذ {diff.days} يوم" |
| elif diff.seconds >= 3600: |
| return f"منذ {diff.seconds // 3600} ساعة" |
| elif diff.seconds >= 60: |
| return f"منذ {diff.seconds // 60} دقيقة" |
| return "الآن" |
| except: |
| return "غير معروف" |
|
|
| |
| |
| |
|
|
| def get_stats(): |
| """جلب الإحصائيات لعرضها في الأعلى""" |
| conn = get_connection() |
| c = conn.cursor() |
| |
| try: |
| |
| c.execute("SELECT COUNT(*) FROM testimonials") |
| total = c.fetchone()[0] |
| |
| |
| c.execute("SELECT AVG(rating) FROM testimonials") |
| avg = c.fetchone()[0] |
| avg_rating = round(avg, 1) if avg else 0 |
| |
| |
| c.execute(""" |
| SELECT |
| COUNT(CASE WHEN sentiment LIKE '%إيجابي%' THEN 1 END) as positive, |
| COUNT(CASE WHEN sentiment LIKE '%سلبي%' THEN 1 END) as negative, |
| COUNT(CASE WHEN sentiment LIKE '%محايد%' THEN 1 END) as neutral |
| FROM testimonials |
| """) |
| sentiments = c.fetchone() |
| |
| |
| c.execute(""" |
| SELECT name, content, likes |
| FROM testimonials |
| ORDER BY likes DESC |
| LIMIT 1 |
| """) |
| most_liked = c.fetchone() |
| |
| except Exception as e: |
| logger.error(f"خطأ في الإحصائيات: {e}") |
| return {"total": 0, "avg_rating": 0, "positive": 0, "negative": 0, "neutral": 0, "most_liked": None} |
| finally: |
| conn.close() |
| |
| return { |
| "total": total, |
| "avg_rating": avg_rating, |
| "positive": sentiments[0] if sentiments else 0, |
| "negative": sentiments[1] if sentiments else 0, |
| "neutral": sentiments[2] if sentiments else 0, |
| "most_liked": most_liked |
| } |
|
|
| def render_stats(stats): |
| """عرض الإحصائيات بشكل جميل""" |
| most_liked_html = "" |
| if stats['most_liked'] and stats['most_liked'][2] > 0: |
| most_liked_html = f""" |
| <div style=" |
| background:#fef3c7; |
| padding:0.3rem 0.8rem; |
| border-radius:999px; |
| font-size:0.85rem; |
| color:#92400e; |
| "> |
| 🏆 الأكثر إعجاباً: {escape_safe(stats['most_liked'][0])} 👍 {stats['most_liked'][2]} |
| </div> |
| """ |
| |
| return f""" |
| <div style=" |
| display:grid; |
| grid-template-columns:repeat(auto-fit, minmax(140px, 1fr)); |
| gap:0.8rem; |
| background:white; |
| padding:1rem 1.5rem; |
| border-radius:16px; |
| border:1px solid #e2e8f0; |
| margin-bottom:1.5rem; |
| box-shadow:0 2px 8px rgba(0,0,0,0.04); |
| "> |
| <div style="text-align:center;"> |
| <div style="font-size:1.8rem;font-weight:700;color:#1e293b;">{stats['total']}</div> |
| <div style="font-size:0.8rem;color:#64748b;">📝 آراء</div> |
| </div> |
| <div style="text-align:center;"> |
| <div style="font-size:1.8rem;font-weight:700;color:#f59e0b;">{stats['avg_rating']}</div> |
| <div style="font-size:0.8rem;color:#64748b;">⭐ متوسط التقييم</div> |
| </div> |
| <div style="text-align:center;"> |
| <div style="font-size:1.2rem;font-weight:600;color:#22c55e;">{stats['positive']}</div> |
| <div style="font-size:0.8rem;color:#64748b;">😊 إيجابي</div> |
| </div> |
| <div style="text-align:center;"> |
| <div style="font-size:1.2rem;font-weight:600;color:#ef4444;">{stats['negative']}</div> |
| <div style="font-size:0.8rem;color:#64748b;">😕 سلبي</div> |
| </div> |
| <div style="text-align:center;"> |
| <div style="font-size:1.2rem;font-weight:600;color:#94a3b8;">{stats['neutral']}</div> |
| <div style="font-size:0.8rem;color:#64748b;">😐 محايد</div> |
| </div> |
| {most_liked_html} |
| </div> |
| """ |
|
|
| |
| |
| |
|
|
| def add_testimonial(name, role, content, rating, request: gr.Request): |
| content = clean_text(content) |
| |
| if not content: |
| return "❌ الرجاء كتابة رأيك", "", "", update_display() |
| |
| if len(content) > 800: |
| return "❌ الحد الأقصى 800 حرف", "", "", update_display() |
| |
| if contains_bad_words(content): |
| return "❌ يوجد كلمات غير مناسبة", "", "", update_display() |
| |
| ip_hash = get_ip_hash(request) |
| conn = get_connection() |
| c = conn.cursor() |
| |
| |
| one_hour_ago = (datetime.now() - timedelta(hours=1)).isoformat() |
| c.execute("SELECT COUNT(*) FROM testimonials WHERE ip_hash = ? AND timestamp >= ?", |
| (ip_hash, one_hour_ago)) |
| |
| if c.fetchone()[0] >= 3: |
| conn.close() |
| return "❌ وصلت الحد المسموح (3 آراء في الساعة)", "", "", update_display() |
| |
| name = clean_text(name) or "مجهول" |
| role = clean_text(role) or "" |
| sentiment = detect_sentiment(content) |
| |
| c.execute(""" |
| INSERT INTO testimonials (name, role, content, sentiment, rating, timestamp, likes, is_highlighted, ip_hash) |
| VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) |
| """, (name, role, content, sentiment, int(rating), datetime.now().isoformat(), 0, 0, ip_hash)) |
| |
| conn.commit() |
| conn.close() |
| |
| upload_db_to_hub() |
| return "✅ تم نشر رأيك بنجاح!", "", "", update_display() |
|
|
| |
| |
| |
|
|
| def get_testimonials(filter_type="all", sort_by="newest", page=1, per_page=5): |
| conn = get_connection() |
| c = conn.cursor() |
| |
| query = "SELECT id, name, role, content, sentiment, rating, timestamp, likes, is_highlighted FROM testimonials" |
| conditions = [] |
| params = [] |
| |
| if filter_type == "highlight": |
| conditions.append("is_highlighted = 1") |
| elif filter_type == "liked": |
| conditions.append("likes > 0") |
| |
| if conditions: |
| query += " WHERE " + " AND ".join(conditions) |
| |
| if sort_by == "likes": |
| query += " ORDER BY likes DESC, timestamp DESC" |
| else: |
| query += " ORDER BY timestamp DESC" |
| |
| offset = (page - 1) * per_page |
| query += " LIMIT ? OFFSET ?" |
| params.extend([per_page, offset]) |
| |
| c.execute(query, params) |
| rows = c.fetchall() |
| |
| |
| count_query = "SELECT COUNT(*) FROM testimonials" |
| if conditions: |
| count_query += " WHERE " + " AND ".join(conditions) |
| c.execute(count_query) |
| total = c.fetchone()[0] |
| |
| conn.close() |
| return rows, total, (total + per_page - 1) // per_page if total > 0 else 1 |
|
|
| |
| |
| |
|
|
| def render_card(row): |
| (id, name, role, content, sentiment, rating, timestamp, likes, is_highlighted) = row |
| |
| highlight_style = "" |
| if is_highlighted: |
| highlight_style = "border-right:4px solid #f59e0b; background:#fffbeb;" |
| |
| stars = "⭐" * int(rating) |
| role_html = f'<span style="background:#eef2ff;color:#4338ca;padding:0.2rem 0.6rem;border-radius:999px;font-size:0.75rem;">{escape_safe(role)}</span>' if role else "" |
| |
| return f""" |
| <div style=" |
| background:white; |
| border-radius:16px; |
| padding:1rem 1.2rem; |
| margin-bottom:0.8rem; |
| border:1px solid #e2e8f0; |
| box-shadow:0 2px 8px rgba(0,0,0,0.04); |
| {highlight_style} |
| transition: all 0.2s; |
| "> |
| <div style="display:flex;justify-content:space-between;align-items:center;flex-wrap:wrap;gap:0.5rem;"> |
| <div style="display:flex;align-items:center;gap:0.5rem;flex-wrap:wrap;"> |
| <span style="font-weight:700;color:#0f172a;">{escape_safe(name)}</span> |
| {role_html} |
| <span style="color:#64748b;font-size:0.75rem;">{sentiment}</span> |
| </div> |
| <span style="color:#94a3b8;font-size:0.7rem;">{format_time(timestamp)}</span> |
| </div> |
| <div style="margin-top:0.5rem;color:#1e293b;line-height:1.7;font-size:0.95rem;"> |
| {escape_safe(content)} |
| </div> |
| <div style="margin-top:0.5rem;display:flex;justify-content:space-between;align-items:center;"> |
| <span>{stars}</span> |
| <span style="color:#64748b;font-size:0.8rem;">👍 {likes}</span> |
| </div> |
| </div> |
| """ |
|
|
| |
| |
| |
|
|
| def update_display(filter_type="all", sort_by="newest", page=1): |
| rows, total, total_pages = get_testimonials(filter_type, sort_by, page) |
| |
| if not rows: |
| return """ |
| <div style=" |
| background:white; |
| border-radius:16px; |
| padding:2rem; |
| text-align:center; |
| border:1px dashed #cbd5e1; |
| color:#64748b; |
| "> |
| ✨ لا توجد آراء حالياً |
| </div> |
| """ |
| |
| html = "" |
| for row in rows: |
| html += render_card(row) |
| |
| |
| if total_pages > 1: |
| html += f""" |
| <div style="display:flex;justify-content:center;gap:0.5rem;margin-top:0.8rem;"> |
| """ |
| if page > 1: |
| html += f'<button onclick="changePage({page-1})" style="padding:0.4rem 1.2rem;background:#4338ca;color:white;border:none;border-radius:8px;cursor:pointer;font-size:0.9rem;">⬅ السابق</button>' |
| html += f'<span style="padding:0.4rem 1rem;color:#64748b;">{page} / {total_pages}</span>' |
| if page < total_pages: |
| html += f'<button onclick="changePage({page+1})" style="padding:0.4rem 1.2rem;background:#4338ca;color:white;border:none;border-radius:8px;cursor:pointer;font-size:0.9rem;">التالي ➡</button>' |
| html += "</div>" |
| |
| return html |
|
|
| |
| |
| |
|
|
| def export_data(): |
| conn = get_connection() |
| c = conn.cursor() |
| c.execute("SELECT name, role, content, sentiment, rating, timestamp, likes FROM testimonials ORDER BY timestamp DESC") |
| rows = c.fetchall() |
| conn.close() |
| |
| if not rows: |
| return "لا توجد بيانات للتصدير" |
| |
| |
| data = [] |
| for row in rows: |
| data.append({ |
| "الاسم": row[0], |
| "الدور": row[1], |
| "الرأي": row[2], |
| "المشاعر": row[3], |
| "التقييم": row[4], |
| "التاريخ": row[5], |
| "الإعجابات": row[6] |
| }) |
| |
| return json.dumps(data, ensure_ascii=False, indent=2) |
|
|
| |
| |
| |
|
|
| def admin_panel(): |
| with gr.Blocks() as admin: |
| gr.Markdown("### 🔐 لوحة تحكم المشرف") |
| |
| with gr.Row(): |
| admin_key = gr.Textbox(label="مفتاح المشرف", type="password", scale=1) |
| action = gr.Radio(choices=[("حذف", "delete"), ("تمييز", "highlight")], label="الإجراء", scale=1) |
| testimonial_id = gr.Number(label="رقم الرأي", precision=0, minimum=1, scale=1) |
| |
| execute_btn = gr.Button("تنفيذ", variant="primary") |
| result = gr.Markdown() |
| |
| def admin_action(key, action, id): |
| expected = os.getenv("ADMIN_KEY", "admin123") |
| if key != expected: |
| return "❌ مفتاح المشرف غير صحيح" |
| |
| conn = get_connection() |
| c = conn.cursor() |
| |
| if action == "delete": |
| c.execute("DELETE FROM testimonials WHERE id = ?", (int(id),)) |
| result_text = "✅ تم حذف الرأي" |
| else: |
| c.execute(""" |
| UPDATE testimonials |
| SET is_highlighted = CASE WHEN is_highlighted = 1 THEN 0 ELSE 1 END |
| WHERE id = ? |
| """, (int(id),)) |
| result_text = "✅ تم تحديث التمييز" |
| |
| conn.commit() |
| conn.close() |
| upload_db_to_hub() |
| return result_text + " (تم التحديث)" |
| |
| execute_btn.click( |
| fn=admin_action, |
| inputs=[admin_key, action, testimonial_id], |
| outputs=result |
| ) |
| |
| return admin |
|
|
| |
| |
| |
|
|
| def create_interface(): |
| custom_css = """ |
| .gradio-container { |
| max-width: 1200px !important; |
| margin: 0 auto !important; |
| padding: 0.5rem !important; |
| } |
| @media (max-width: 768px) { |
| .grid-container { |
| grid-template-columns: 1fr !important; |
| } |
| } |
| """ |
| |
| with gr.Blocks( |
| title="منصة الآراء", |
| theme=gr.themes.Soft(), |
| css=custom_css, |
| fill_width=True |
| ) as demo: |
| |
| |
| gr.HTML(""" |
| <div style=" |
| text-align:center; |
| padding:1rem 0 0.5rem 0; |
| "> |
| <h1 style=" |
| font-size:2.2rem; |
| font-weight:800; |
| margin:0; |
| background:linear-gradient(135deg, #1e293b, #4338ca); |
| -webkit-background-clip:text; |
| -webkit-text-fill-color:transparent; |
| "> |
| 💬 منصة الآراء |
| </h1> |
| <p style="color:#64748b;margin:0.2rem 0 0 0;font-size:0.9rem;"> |
| شارك رأيك بكل شفافية • جميع الآراء محفوظة بشكل دائم |
| </p> |
| </div> |
| """) |
| |
| |
| stats_display = gr.HTML() |
| |
| |
| with gr.Row(equal_height=False): |
| |
| |
| with gr.Column(scale=1, min_width=280): |
| gr.Markdown("### ✍️ أضف رأيك") |
| |
| name_input = gr.Textbox( |
| label="الاسم", |
| placeholder="اختياري", |
| container=False |
| ) |
| role_input = gr.Textbox( |
| label="الدور", |
| placeholder="مثال: مهندس", |
| container=False |
| ) |
| rating_input = gr.Slider( |
| minimum=1, |
| maximum=5, |
| value=5, |
| step=1, |
| label="⭐ التقييم" |
| ) |
| content_input = gr.Textbox( |
| label="الرأي", |
| placeholder="اكتب رأيك هنا...", |
| lines=4, |
| container=False |
| ) |
| |
| with gr.Row(): |
| submit_btn = gr.Button("📢 نشر", variant="primary", scale=2) |
| clear_btn = gr.Button("🗑️ مسح", scale=1) |
| |
| status_output = gr.Markdown() |
| |
| |
| with gr.Column(scale=2, min_width=400): |
| gr.Markdown("### 💬 آراء المستخدمين") |
| |
| with gr.Row(): |
| filter_dropdown = gr.Dropdown( |
| choices=[ |
| ("الكل", "all"), |
| ("المميزة", "highlight"), |
| ("الأكثر إعجاباً", "liked") |
| ], |
| value="all", |
| label="🔍 فلترة", |
| scale=1 |
| ) |
| sort_dropdown = gr.Dropdown( |
| choices=[ |
| ("الأحدث", "newest"), |
| ("الأكثر إعجاباً", "likes") |
| ], |
| value="newest", |
| label="📊 ترتيب", |
| scale=1 |
| ) |
| |
| refresh_btn = gr.Button("🔄 تحديث", variant="secondary", size="sm") |
| testimonials_display = gr.HTML() |
| |
| |
| with gr.Row(): |
| with gr.Column(scale=1): |
| with gr.Accordion("⚙️ أدوات إضافية", open=False): |
| with gr.Row(): |
| export_btn = gr.Button("📥 تصدير البيانات (JSON)", size="sm") |
| export_output = gr.Textbox(label="", lines=3, visible=False) |
| |
| |
| admin_btn = gr.Button("🔐 لوحة المشرف", size="sm", variant="stop") |
| |
| |
| admin_interface = admin_panel() |
| admin_interface.visible = False |
| |
| def toggle_admin(): |
| return gr.update(visible=not admin_interface.visible) |
| |
| admin_btn.click( |
| fn=toggle_admin, |
| outputs=admin_interface |
| ) |
| |
| |
| |
| |
| def update_stats(): |
| return render_stats(get_stats()) |
| |
| |
| def full_update(filter_type="all", sort_by="newest", page=1): |
| return update_stats(), update_display(filter_type, sort_by, page) |
| |
| |
| submit_btn.click( |
| fn=add_testimonial, |
| inputs=[name_input, role_input, content_input, rating_input], |
| outputs=[status_output, name_input, role_input, testimonials_display] |
| ).then( |
| fn=update_stats, |
| outputs=stats_display |
| ) |
| |
| |
| refresh_btn.click( |
| fn=full_update, |
| inputs=[filter_dropdown, sort_dropdown], |
| outputs=[stats_display, testimonials_display] |
| ) |
| |
| |
| filter_dropdown.change( |
| fn=full_update, |
| inputs=[filter_dropdown, sort_dropdown], |
| outputs=[stats_display, testimonials_display] |
| ) |
| |
| sort_dropdown.change( |
| fn=full_update, |
| inputs=[filter_dropdown, sort_dropdown], |
| outputs=[stats_display, testimonials_display] |
| ) |
| |
| |
| def show_export(): |
| data = export_data() |
| return gr.update(visible=True, value=data) |
| |
| export_btn.click( |
| fn=show_export, |
| outputs=export_output |
| ) |
| |
| |
| clear_btn.click( |
| fn=lambda: ("", "", "", ""), |
| outputs=[name_input, role_input, content_input, status_output] |
| ) |
| |
| |
| demo.load( |
| fn=full_update, |
| inputs=[filter_dropdown, sort_dropdown], |
| outputs=[stats_display, testimonials_display] |
| ) |
| |
| return demo |
|
|
| |
| |
| |
|
|
| if __name__ == "__main__": |
| logger.info("جاري تحميل قاعدة البيانات...") |
| |
| if not download_db_from_hub(): |
| logger.info("إنشاء قاعدة بيانات جديدة...") |
| init_db() |
| upload_db_to_hub() |
| else: |
| init_db() |
| |
| demo = create_interface() |
| demo.launch(server_name="0.0.0.0", server_port=7860) |
