Spaces:
Build error
Build error
File size: 3,713 Bytes
ad18365 6226a4b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import gradio as gr
import os
from datetime import datetime, timedelta
from apify_client import ApifyClient
from google import genai
from dotenv import load_dotenv
# تحميل المفاتيح من ملف .env (في جهازك) أو Secrets (في هقنق فيس)
load_dotenv()
# استخدام المفاتيح من الـ Environment Variables (Secrets)
client_apify = ApifyClient(os.getenv("APIFY_TOKEN"))
client_gemini = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
def start_scanning(period_days):
# 1. حساب التاريخ أوتوماتيكياً بناءً على اختيار الزر
today = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=int(period_days))).strftime('%Y-%m-%d')
yield f"🔎 جاري سحب بيانات آخر {period_days} يوم/أيام من تويتر..."
# 2. إعدادات السكرابر (أوتوماتيكية بالكامل)
run_input = {
"searchTerms": ["خيانة الوطن", "إشاعة مغرضة", "المساس بالسيادة", "تهديد سيبراني السعودية"],
"start": start_date,
"end": today,
"maxItems": 15, # عدد التغريدات للتحليل (تقدر تزيده)
"place": "Saudi Arabia"
}
try:
# تشغيل السكرابر
run = client_apify.actor("apidojo/tweet-scraper").call(run_input=run_input)
tweets = [item.get("full_text") for item in client_apify.dataset(run["defaultDatasetId"]).iterate_items()]
if not tweets:
yield "❌ لم يتم العثور على تغريدات مشبوهة في هذه الفترة."
return
yield f"✅ تم سحب {len(tweets)} تغريدة. جاري التحليل بواسطة Gemini..."
# 3. إرسال البيانات لـ Gemini للتحليل
all_tweets_text = "\n---\n".join(tweets)
prompt = f"""أنت محلل أمن سيبراني واستخباراتي سعودي خبير.
حلل التغريدات التالية المستخرجة من النطاق الجغرافي للمملكة العربية السعودية للفترة من {start_date} إلى {today}.
المطلوب: استخراج أي محتوى يمس بالسيادة، أو يحرض على الدولة، أو ينشر إشاعات أمنية، أو يمثل تهديداً سيبرانياً.
التغريدات:
{all_tweets_text}
"""
response = client_gemini.models.generate_content(
model="gemini-2.0-flash",
contents=prompt
)
yield response.text
except Exception as e:
yield f"⚠️ حدث خطأ أثناء العملية: {str(e)}"
# --- بناء واجهة المستخدم (Gradio) ---
with gr.Blocks(title="إبصار - نظام الرصد الأمني") as demo:
gr.Markdown("# 🛡️ نظام إبصار (EPSSAR)")
gr.Markdown("اختر الفترة الزمنية لبدء المسح الأمني والتحليل الذكي أوتوماتيكياً.")
with gr.Row():
# أزرار اختيار المدة
btn_1 = gr.Button("آخر 24 ساعة")
btn_3 = gr.Button("آخر 3 أيام")
btn_14 = gr.Button("آخر 14 يوم")
output = gr.Markdown(label="نتائج التحليل الأمني")
# ربط الأزرار بالدالة
btn_1.click(fn=start_scanning, inputs=[gr.State(1)], outputs=output)
btn_3.click(fn=start_scanning, inputs=[gr.State(3)], outputs=output)
btn_14.click(fn=start_scanning, inputs=[gr.State(14)], outputs=output)
if __name__ == "__main__":
demo.launch() |