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"""
🏆 الأكثر إعجاباً: {escape_safe(stats['most_liked'][0])} 👍 {stats['most_liked'][2]}
"""
return f"""
{stats['total']}
📝 آراء
{stats['avg_rating']}
⭐ متوسط التقييم
{stats['positive']}
😊 إيجابي
{stats['negative']}
😕 سلبي
{stats['neutral']}
😐 محايد
{most_liked_html}
"""
# ==================================================
# إضافة رأي
# ==================================================
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'{escape_safe(role)}' if role else ""
return f"""
{escape_safe(name)}
{role_html}
{sentiment}
{format_time(timestamp)}
{escape_safe(content)}
{stars}👍 {likes}
"""
# ==================================================
# تحديث العرض
# ==================================================
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 """
✨ لا توجد آراء حالياً
"""
html = ""
for row in rows:
html += render_card(row)
# أزرار التنقل
if total_pages > 1:
html += f"""
"""
if page > 1:
html += f''
html += f'{page} / {total_pages}'
if page < total_pages:
html += f''
html += "
"
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 "لا توجد بيانات للتصدير"
# تصدير كـ JSON
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("""
💬 منصة الآراء
شارك رأيك بكل شفافية • جميع الآراء محفوظة بشكل دائم