Spaces:
Sleeping
Sleeping
| # ============================================================================== | |
| # MegoFlow / Actora - Social Post Engagement Analyzer | |
| # ============================================================================== | |
| # تطبيق Gradio عربي للتنبؤ بتفاعل منشورات السوشيال وإظهار تحليل Gemini. | |
| # | |
| # التعليمات المختصرة للتشغيل (على Hugging Face Spaces أو محليًا): | |
| # 1) ضع المتغيرات في .env أو في Settings → Variables: | |
| # - GEMINI_API_KEY="YOUR_KEY" | |
| # - USE_MOCK_MODEL="True" # غيّرها إلى "False" لتفعيل الموديل الحقيقي لاحقًا | |
| # - MODEL_REPO="amrtweg/Actora" # افتراضيًا | |
| # - (اختياري) ACTORA_SCALE_FACTOR="10.0" لتكبير مخرجات الموديل الحقيقي إن كانت صغيرة | |
| # | |
| # 2) ارفع ملف assets/logo.svg (لو غير موجود سيظهر لوجو بديل تلقائي). | |
| # | |
| # 3) تأكد من وجود الحزم في requirements.txt (لا تضف/تحذف حزم هنا؛ | |
| # فقط إن وجدت غير مستخدم ضع تعليق TODO داخل requirements وليس هنا). | |
| # | |
| # 4) شغّل التطبيق. | |
| # ============================================================================== | |
| import os | |
| import re | |
| import time | |
| import base64 # (1) إضافة الاستيراد المفقود | |
| import random | |
| import logging | |
| from typing import Dict, Optional | |
| import gradio as gr | |
| import bleach | |
| from dotenv import load_dotenv | |
| import google.generativeai as genai | |
| # مكونات الموديل الحقيقي (تبقى موجودة حتى لو شاغلين Mock افتراضيًا) | |
| import torch | |
| import torch.nn.functional as F | |
| from transformers import AutoTokenizer, AutoModelForSequenceClassification | |
| # ---------------------------- | |
| # الإعدادات والتهيئة العامة | |
| # ---------------------------- | |
| load_dotenv() | |
| GEMINI_API_KEY: Optional[str] = os.getenv("AIzaSyC5k5j7zeDWxJATTzx6zNABXwXQWvLIqw") | |
| USE_MOCK_MODEL: bool = os.getenv("USE_MOCK_MODEL", "True").lower() in ("true", "1", "t", "yes", "y") | |
| MODEL_REPO: str = os.getenv("MODEL_REPO", "amrtweg/Actora") | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s - %(levelname)s - %(message)s" | |
| ) | |
| # ---------------------------- | |
| # تحميل اللوجو كـ Data URL | |
| # ---------------------------- | |
| def load_logo_data_url(path: str = "assets/logo.svg") -> str: | |
| """ | |
| يقرأ ملف SVG ويحوله Base64 Data URL. | |
| لو غير موجود، يولّد SVG بسيط Placeholder. | |
| """ | |
| try: | |
| with open(path, "rb") as f: | |
| encoded = base64.b64encode(f.read()).decode("utf-8") | |
| return f"data:image/svg+xml;base64,{encoded}" | |
| except FileNotFoundError: | |
| logging.warning("لم يتم العثور على الملف assets/logo.svg — سيتم استخدام لوجو بديل.") | |
| placeholder_svg = ( | |
| '<svg xmlns="http://www.w3.org/2000/svg" width="200" height="60">' | |
| '<defs><linearGradient id="g" x1="0" y1="0" x2="1" y2="1">' | |
| '<stop offset="0%" stop-color="#24b5a5"/><stop offset="100%" stop-color="#042A5C"/>' | |
| '</linearGradient></defs>' | |
| '<rect width="200" height="60" fill="#0f172a"/>' | |
| '<text x="16" y="40" font-family="Cairo,Arial" font-size="28" font-weight="700" fill="url(#g)">Mego</text>' | |
| '<text x="102" y="40" font-family="Cairo,Arial" font-size="28" font-weight="400" fill="#e2e8f0">Flow</text>' | |
| '</svg>' | |
| ) | |
| encoded = base64.b64encode(placeholder_svg.encode("utf-8")).decode("utf-8") | |
| return f"data:image/svg+xml;base64,{encoded}" | |
| # (2) توحيد اسم متغير اللوجو | |
| LOGO_DATA_URL = load_logo_data_url() | |
| # ---------------------------- | |
| # المنطق الخاص بالموديلات | |
| # ---------------------------- | |
| class MockPredictor: | |
| """ | |
| الموديل الوهمي الافتراضي لسهولة التجربة سريعًا (لا يعتمد على إنترنت/موديلات كبيرة). | |
| لتفعيل الموديل الحقيقي: غيّر USE_MOCK_MODEL="False" في .env أو Variables. | |
| """ | |
| def __init__(self, model_path: str = "mock", **kwargs): | |
| self.is_ready = True | |
| logging.info("MockPredictor قيد الاستخدام. (لتفعيل الحقيقي: USE_MOCK_MODEL=False)") | |
| def predict(self, text: str) -> Dict[str, int]: | |
| if not text or not text.strip(): | |
| return {} | |
| # أرقام شبه عشوائية لعرض الواجهة فقط | |
| time.sleep(0.2) | |
| results = { | |
| "liked": random.randint(20, 250), | |
| "loved": random.randint(5, 90), | |
| "haha": random.randint(0, 40), | |
| "wow": random.randint(0, 35), | |
| "sad": random.randint(0, 15), | |
| "angry": random.randint(0, 10), | |
| "comments": random.randint(5, 80), | |
| "shares": random.randint(3, 50), | |
| } | |
| # إجمالي التفاعل المتوقع (بدون مضاعفة comments/shares) | |
| results["interactions"] = sum(results[k] for k in ["liked","loved","haha","wow","sad","angry"]) | |
| return results | |
| # ===== الموديل الحقيقي (جاهز للاستبدال بالضبط وقت ما تحب) ===== | |
| # لإلغاء الـ Mock وتفعيل هذا: USE_MOCK_MODEL="False" | |
| def _normalize_label(s: str) -> str: | |
| """تبسيط اسم الليبل للمطابقة: lowercase وبدون مسافات/رموز.""" | |
| return re.sub(r"[^a-z]+", "", s.lower()) | |
| class ActoraPredictor: | |
| """ | |
| Predictor حقيقي يعتمد على موديل HuggingFace SequenceClassification. | |
| - يقرأ id2label/label2id من config | |
| - يحوّل logits لقيم موجبة باستخدام softplus | |
| - يطابق الليبلات المرنة إلى المفاتيح القياسية | |
| - يحسب interactions كمجموع القيم المتاحة | |
| """ | |
| CANONICAL_KEYS = ["liked", "loved", "haha", "wow", "sad", "angry", "comments", "shares"] | |
| def __init__(self, model_path: str, device: Optional[str] = None, max_length: int = 256): | |
| self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") | |
| self.max_length = max_length | |
| self.scale = float(os.getenv("ACTORA_SCALE_FACTOR", "1.0")) | |
| logging.info(f"[ActoraPredictor] تحميل الموديل: {model_path} على {self.device}") | |
| self.tokenizer = AutoTokenizer.from_pretrained(model_path) | |
| self.model = AutoModelForSequenceClassification.from_pretrained(model_path).to(self.device) | |
| self.model.eval() | |
| id2label = getattr(self.model.config, "id2label", {}) or {} | |
| if not id2label and hasattr(self.model.config, "label2id"): | |
| label2id = {k.lower(): int(v) for k, v in self.model.config.label2id.items()} | |
| else: | |
| # تحويل id2label إلى label2id | |
| label2id = {str(v).lower(): int(k) for k, v in id2label.items()} | |
| self.idx_map: Dict[str, int] = {} | |
| flattened = { _normalize_label(lbl): idx for lbl, idx in label2id.items() } | |
| for key in self.CANONICAL_KEYS: | |
| candidates = [key, f"{key}s", f"{key}_count", f"{key}count"] | |
| found = None | |
| for c in candidates: | |
| norm = _normalize_label(c) | |
| for flbl, idx in flattened.items(): | |
| if flbl == norm or flbl.endswith(norm) or norm in flbl: | |
| found = idx | |
| break | |
| if found is not None: | |
| break | |
| if found is not None: | |
| self.idx_map[key] = found | |
| logging.info(f"[ActoraPredictor] Label map: {self.idx_map}") | |
| self.is_ready = True | |
| def predict(self, text: str) -> Dict[str, int]: | |
| if not self.is_ready: | |
| raise RuntimeError("Model not ready.") | |
| if not text or not text.strip(): | |
| return {} | |
| enc = self.tokenizer( | |
| text, return_tensors="pt", truncation=True, max_length=self.max_length, padding=False | |
| ) | |
| enc = {k: v.to(self.device) for k, v in enc.items()} | |
| out = self.model(**enc) | |
| logits = out.logits.squeeze(0) # [num_labels] | |
| positive = F.softplus(logits).cpu().tolist() | |
| results: Dict[str, int] = {} | |
| for key in self.CANONICAL_KEYS: | |
| if key in self.idx_map: | |
| val = positive[self.idx_map[key]] * self.scale | |
| results[key] = int(round(float(val))) | |
| else: | |
| results[key] = 0 | |
| results["interactions"] = int( | |
| sum(results[k] for k in ["liked", "loved", "haha", "wow", "sad", "angry", "comments", "shares"]) | |
| ) | |
| return results | |
| # ===== نهاية الموديل الحقيقي ===== | |
| # ---------------------------- | |
| # عميل Gemini (مع Sanitization) | |
| # ---------------------------- | |
| class GeminiClient: | |
| def __init__(self, api_key: Optional[str]): | |
| self.model = None | |
| if api_key: | |
| try: | |
| genai.configure(api_key=api_key) | |
| # نستخدم موديل سريع وحديث | |
| self.model = genai.GenerativeModel("gemini-1.5-flash-latest") | |
| logging.info("تم تهيئة Gemini بنجاح.") | |
| except Exception as e: | |
| logging.error(f"فشل تهيئة Gemini: {e}") | |
| else: | |
| logging.warning("GEMINI_API_KEY غير موجود. سيتم تعطيل تحليل Gemini.") | |
| def analyze(self, text: str, metrics: Dict[str, int]) -> str: | |
| # لو مفيش موديل (أو في وضع Mock)، نرجّع رسالة بسيطة (مع تعقيم) | |
| if not self.model: | |
| return "ميزة تحليل Gemini غير مفعلة. أضف المفتاح في الإعدادات لتفعيلها." | |
| prompt = f""" | |
| # المهمة | |
| أنت "MegoFlow AI"، مستشار تسويق رقمي وخبير في تحليل النصوص الإعلانية. المطلوب تحليل موجز وقابل للتنفيذ. | |
| # المعطيات | |
| - إجمالي التفاعلات المتوقعة: {metrics.get('interactions', 0)} | |
| - إعجابات: {metrics.get('liked', 0)} | |
| - قلوب: {metrics.get('loved', 0)} | |
| - تعليقات: {metrics.get('comments', 0)} | |
| - مشاركات: {metrics.get('shares', 0)} | |
| # نص المنشور | |
| --- | |
| {text} | |
| --- | |
| # المطلوب (Markdown بالعربية): | |
| 1) **تشخيص الأداء**: جملة واحدة توضح الاستقبال المتوقع. | |
| 2) **مقترحات للنمو**: 3 نقاط محددة جدًا (CTA/إعادة صياغة/زاوية مختلفة) مرتبطة بالنص. | |
| """ | |
| try: | |
| response = self.model.generate_content(prompt) | |
| # (4) إبقاء Sanitization بواسطة bleach | |
| safe = bleach.clean(response.text or "", tags=[], strip=True) | |
| return safe | |
| except Exception as e: | |
| logging.error(f"خطأ أثناء استدعاء Gemini: {e}") | |
| return "حدث خطأ أثناء التواصل مع Gemini. حاول لاحقًا." | |
| gemini = GeminiClient(GEMINI_API_KEY) | |
| # ---------------------------- | |
| # تهيئة الـ Predictor المختار | |
| # ---------------------------- | |
| if USE_MOCK_MODEL: | |
| predictor = MockPredictor() | |
| else: | |
| try: | |
| predictor = ActoraPredictor(MODEL_REPO) | |
| except Exception as e: | |
| logging.error(f"تعذر تحميل الموديل الحقيقي: {e}") | |
| logging.warning("الرجوع مؤقتًا إلى MockPredictor للاستمرار في العمل.") | |
| predictor = MockPredictor() | |
| # ---------------------------- | |
| # واجهة المستخدم (Gradio) | |
| # ---------------------------- | |
| APP_CSS = """ | |
| @import url('https://fonts.googleapis.com/css2?family=Cairo:wght@400;700;900&display=swap'); | |
| body { font-family: 'Cairo', sans-serif; direction: rtl; background: #0f172a; color: #e2e8f0; } | |
| .center { max-width: 1200px; margin: 0 auto; } | |
| .logo { text-align: center; padding: 16px 0 8px; } | |
| .card { background: #1e293b; border: 1px solid #334155; border-radius: 14px; padding: 18px; margin-top: 12px; box-shadow: 0 6px 18px rgba(0,0,0,0.18); } | |
| .card-header { font-size: 18px; font-weight: 800; color: #fff; border-bottom: 1px solid #334155; padding-bottom: 10px; margin-bottom: 12px; } /* (3) توحيد .card-header */ | |
| .metrics-grid { display: grid; grid-template-columns: repeat(auto-fill,minmax(120px,1fr)); gap: 10px; } | |
| .metric-item { background: #0f172a; border: 1px solid #334155; border-radius: 10px; padding: 12px; text-align: center; } | |
| .metric-value { font-size: 22px; font-weight: 900; color: #24b5a5; } | |
| .metric-label { font-size: 12px; color: #94a3b8; margin-top: 6px; } | |
| .footer { text-align: center; color: #94a3b8; margin: 24px 0; font-size: 13px; } | |
| .tabpad { padding-top: 6px; } | |
| """ | |
| def render_metrics(metrics: Dict[str, int]) -> str: | |
| if not metrics: | |
| return "<div class='card'><div class='card-header'>المقاييس المتوقعة</div><p>لا توجد بيانات بعد. اكتب نصًا ثم اضغط تحليل.</p></div>" | |
| # (3) استخدام .card-header بالاسم الموحد | |
| html = "<div class='card'><div class='card-header'>المقاييس المتوقعة</div><div class='metrics-grid'>" | |
| items = [ | |
| ("إجمالي التفاعلات", metrics.get("interactions", 0)), | |
| ("أعجبني", metrics.get("liked", 0)), | |
| ("أحببته", metrics.get("loved", 0)), | |
| ("أضحكني", metrics.get("haha", 0)), | |
| ("أدهشني", metrics.get("wow", 0)), | |
| ("أحزنني", metrics.get("sad", 0)), | |
| ("أغضبني", metrics.get("angry", 0)), | |
| ("تعليقات", metrics.get("comments", 0)), | |
| ("مشاركات", metrics.get("shares", 0)), | |
| ] | |
| for label, value in items: | |
| html += f"<div class='metric-item'><div class='metric-value'>{value:,}</div><div class='metric-label'>{label}</div></div>" | |
| html += "</div></div>" | |
| return html | |
| def handle_single(text: str): | |
| if not text or not text.strip(): | |
| return ( | |
| "<div class='card'><div class='card-header'>المقاييس المتوقعة</div><p>يرجى إدخال نص المنشور أولًا.</p></div>", | |
| "لا يمكن عرض التحليل بدون نص.", | |
| [] | |
| ) | |
| metrics = predictor.predict(text) | |
| analysis = gemini.analyze(text, metrics) | |
| metrics_html = render_metrics(metrics) | |
| # بنسجل في السجل: (نص مختصر, إجمالي, وقت) | |
| entry = [ | |
| (text[:100] + "…") if len(text) > 100 else text, | |
| metrics.get("interactions", 0), | |
| time.strftime("%Y-%m-%d %H:%M:%S"), | |
| ] | |
| return metrics_html, analysis, entry | |
| def handle_ab(a: str, b: str): | |
| if not a.strip() or not b.strip(): | |
| msg = "<div class='card'><div class='card-header'>المقارنة</div><p>أدخل نصّين كاملين للمقارنة.</p></div>" | |
| return msg, msg | |
| ma = predictor.predict(a) | |
| mb = predictor.predict(b) | |
| return render_metrics(ma), render_metrics(mb) | |
| def append_history(history, new_entry): | |
| if new_entry and new_entry[0]: | |
| return [new_entry] + (history or []) | |
| return history or [] | |
| with gr.Blocks(css=APP_CSS, theme=gr.themes.Soft(primary_hue="teal")) as app: | |
| with gr.Column(elem_classes="center"): | |
| # الهيدر (يستخدم LOGO_DATA_URL الموحد) (2) | |
| gr.HTML(f""" | |
| <div class="logo"> | |
| <img src="{LOGO_DATA_URL}" alt="MegoFlow Logo" style="height:56px; filter: drop-shadow(0 0 10px rgba(36,181,165,0.3));" /> | |
| <div style="color:#94a3b8; font-size:14px; margin-top:6px;">حوّل نص منشورك لأرقام ورؤى قابلة للتنفيذ</div> | |
| </div> | |
| """) | |
| history_state = gr.State([]) # [[text, interactions, ts], ...] | |
| with gr.Tabs(): | |
| # (لا نغير اللغة/الترتيب) | |
| with gr.TabItem("تحليل منشور"): | |
| with gr.Row(elem_classes="tabpad"): | |
| with gr.Column(scale=5): | |
| post_box = gr.Textbox(lines=12, label="نص المنشور", placeholder="اكتب أو الصق نص المنشور هنا...") | |
| analyze_btn = gr.Button("حلّل الآن", variant="primary") | |
| with gr.Column(scale=5): | |
| metrics_html = gr.HTML() | |
| analysis_md = gr.Markdown() | |
| with gr.TabItem("مقارنة A/B"): | |
| with gr.Row(elem_classes="tabpad"): | |
| with gr.Column(): | |
| a_box = gr.Textbox(lines=8, label="المنشور (أ)") | |
| a_out = gr.HTML() | |
| with gr.Column(): | |
| b_box = gr.Textbox(lines=8, label="المنشور (ب)") | |
| b_out = gr.HTML() | |
| compare_btn = gr.Button("اختبر وقارن", variant="primary") | |
| with gr.TabItem("سجل"): | |
| history_table = gr.DataFrame( | |
| headers=["النص", "إجمالي التفاعلات", "وقت التحليل"], | |
| datatype=["str", "number", "str"], | |
| interactive=False, | |
| row_count=8 | |
| ) | |
| gr.HTML("<div class='footer'>MegoFlow © 2025</div>") | |
| # الأحداث | |
| analyze_btn.click( | |
| fn=handle_single, | |
| inputs=[post_box], | |
| outputs=[metrics_html, analysis_md, history_state] | |
| ).then( | |
| fn=append_history, | |
| inputs=[history_state, history_state], | |
| outputs=[history_table] | |
| ) | |
| compare_btn.click( | |
| fn=handle_ab, | |
| inputs=[a_box, b_box], | |
| outputs=[a_out, b_out] | |
| ) | |
| if __name__ == "__main__": | |
| app.launch() | |