| import os |
| import time |
| import json |
| import logging |
| import threading |
| import requests |
| from bs4 import BeautifulSoup |
| from flask import Flask |
| from functools import wraps |
|
|
| |
| |
| |
| SNAPCHAT_USER = os.environ.get("SNAPCHAT_USER", "a2001_89") |
| TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "8272603814:AAHaqmdzUM1B4kG0Twtx3eKuk3PesPIvkfQ") |
| TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "-1003584586221") |
| TELEGRAM_BASE_URL = os.environ.get("TELEGRAM_BASE_URL", "https://api.telegram.org/") |
| CHECK_INTERVAL = int(os.environ.get("CHECK_INTERVAL", 15)) * 60 |
|
|
| POSTED_STORIES_FILE = "posted_stories.json" |
|
|
| |
| |
| |
| |
| ERROR_ALERT_THRESHOLD = int(os.environ.get("ERROR_ALERT_THRESHOLD", 3)) |
|
|
| |
| ERROR_ALERT_COOLDOWN = int(os.environ.get("ERROR_ALERT_COOLDOWN", 1800)) |
|
|
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') |
| app = Flask(__name__) |
|
|
| |
| |
| |
| class ErrorAlertManager: |
| """ |
| يتتبع الأخطاء المتكررة ويرسل تنبيهاً عبر تليجرام |
| فقط عند تجاوز الحد المسموح به، مع فترة تبريد لمنع الفيضان. |
| """ |
| def __init__(self): |
| |
| self._error_registry = {} |
| self._lock = threading.Lock() |
|
|
| def report(self, error_type: str, message: str, critical: bool = False): |
| """ |
| يُسجّل خطأ. إذا تجاوز العتبة وانتهت فترة التبريد، يُرسل تنبيهاً لتليجرام. |
| - error_type: معرف نصي للخطأ (مثل "snapchat_fetch_error") |
| - message: تفاصيل الخطأ |
| - critical: إذا True، يُرسَل التنبيه فوراً بغض النظر عن العتبة |
| """ |
| with self._lock: |
| now = time.time() |
| entry = self._error_registry.get(error_type, {"count": 0, "last_alert_time": 0}) |
| entry["count"] += 1 |
| self._error_registry[error_type] = entry |
|
|
| should_alert = critical or (entry["count"] >= ERROR_ALERT_THRESHOLD) |
| cooldown_passed = (now - entry["last_alert_time"]) >= ERROR_ALERT_COOLDOWN |
|
|
| if should_alert and cooldown_passed: |
| self._send_alert(error_type, message, entry["count"], critical) |
| entry["count"] = 0 |
| entry["last_alert_time"] = now |
| self._error_registry[error_type] = entry |
|
|
| def reset(self, error_type: str): |
| """يُعيد تعيين عداد خطأ معين عند نجاح العملية""" |
| with self._lock: |
| if error_type in self._error_registry: |
| self._error_registry[error_type]["count"] = 0 |
|
|
| def _send_alert(self, error_type: str, message: str, count: int, critical: bool): |
| """يرسل رسالة تنبيه مباشرة إلى تليجرام""" |
| base_url = f"{TELEGRAM_BASE_URL.rstrip('/')}/bot{TELEGRAM_BOT_TOKEN}/" |
| endpoint = base_url + "sendMessage" |
|
|
| level_icon = "[CRITICAL]" if critical else "[WARNING]" |
| level_label = "خطا حرج" if critical else "تحذير" |
|
|
| text = ( |
| f"<b>{level_icon} تنبيه من بوت سناب شات</b>\n\n" |
| f"<b>نوع المشكلة:</b> <code>{error_type}</code>\n" |
| f"<b>المستوى:</b> {level_label}\n" |
| f"<b>عدد مرات التكرار:</b> {count}\n" |
| f"<b>الوقت:</b> {time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime())}\n\n" |
| f"<b>التفاصيل:</b>\n<pre>{message[:800]}</pre>\n\n" |
| f"<i>يتم مراقبة: {SNAPCHAT_USER}</i>" |
| ) |
|
|
| payload = { |
| "chat_id": TELEGRAM_CHAT_ID, |
| "text": text, |
| "parse_mode": "HTML", |
| "disable_web_page_preview": True |
| } |
|
|
| try: |
| resp = requests.post(endpoint, json=payload, timeout=15) |
| if resp.status_code == 200: |
| logging.info(f"[AlertManager] تم إرسال تنبيه الخطأ: {error_type}") |
| else: |
| logging.warning(f"[AlertManager] فشل إرسال التنبيه: {resp.text}") |
| except Exception as e: |
| logging.error(f"[AlertManager] خطأ في إرسال التنبيه: {e}") |
|
|
| |
| alert_manager = ErrorAlertManager() |
|
|
|
|
| |
| |
| |
| def load_posted_stories(): |
| if os.path.exists(POSTED_STORIES_FILE): |
| try: |
| with open(POSTED_STORIES_FILE, "r", encoding="utf-8") as f: |
| return json.load(f) |
| except Exception as e: |
| msg = f"خطأ في قراءة ملف السجل: {e}" |
| logging.error(msg) |
| |
| alert_manager.report("log_file_read_error", msg) |
| return [] |
|
|
|
|
| def save_posted_stories(posted_list): |
| try: |
| with open(POSTED_STORIES_FILE, "w", encoding="utf-8") as f: |
| json.dump(posted_list[-200:], f, ensure_ascii=False, indent=2) |
| except Exception as e: |
| msg = f"خطأ في حفظ ملف السجل: {e}" |
| logging.error(msg) |
| |
| alert_manager.report("log_file_write_error", msg, critical=True) |
|
|
|
|
| |
| |
| |
| def send_to_telegram(media_url, is_video, caption): |
| """تحميل المنشور الفعلي من سناب شات وإعادة رفعه كملف مباشر إلى التليجرام""" |
| base_url = f"{TELEGRAM_BASE_URL.rstrip('/')}/bot{TELEGRAM_BOT_TOKEN}/" |
|
|
| try: |
| logging.info("جاري تنزيل ملف القصة من سناب شات...") |
| headers = { |
| "User-Agent": ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) " |
| "Chrome/130.0.0.0 Safari/537.36" |
| ) |
| } |
| media_response = requests.get(media_url, headers=headers, timeout=60) |
|
|
| if media_response.status_code != 200: |
| msg = f"فشل في تنزيل محتوى القصة. كود الحالة: {media_response.status_code} | الرابط: {media_url[:80]}" |
| logging.error(msg) |
| |
| alert_manager.report("media_download_error", msg) |
| return False |
|
|
| |
| alert_manager.reset("media_download_error") |
|
|
| content_type = media_response.headers.get('Content-Type', '') |
| if 'video' in content_type.lower(): |
| is_video = True |
|
|
| files = {} |
| data = { |
| 'chat_id': TELEGRAM_CHAT_ID, |
| 'caption': caption, |
| 'parse_mode': 'HTML' |
| } |
|
|
| if is_video: |
| endpoint = base_url + "sendVideo" |
| files['video'] = ('story.mp4', media_response.content, 'video/mp4') |
| logging.info("تم التعرف على الملف كـ (فيديو). جاري رفعه...") |
| else: |
| endpoint = base_url + "sendPhoto" |
| files['photo'] = ('story.jpg', media_response.content, 'image/jpeg') |
| logging.info("تم التعرف على الملف كـ (صورة). جاري الإرسال...") |
|
|
| resp = requests.post(endpoint, data=data, files=files, timeout=120) |
|
|
| if resp.status_code == 200: |
| logging.info("تم إرسال القصة إلى تليجرام بنجاح!") |
| |
| alert_manager.reset("telegram_send_error") |
| return True |
| else: |
| msg = f"خطأ من تليجرام عند الإرسال: {resp.text[:300]}" |
| logging.error(msg) |
| |
| alert_manager.report("telegram_send_error", msg) |
| return False |
|
|
| except requests.exceptions.Timeout: |
| msg = f"انتهت مهلة الاتصال عند محاولة إرسال: {media_url[:80]}" |
| logging.error(msg) |
| alert_manager.report("send_timeout_error", msg) |
| return False |
| except Exception as e: |
| msg = f"خطأ غير متوقع أثناء الإرسال: {e}" |
| logging.error(msg) |
| alert_manager.report("unexpected_send_error", msg, critical=True) |
| return False |
|
|
|
|
| |
| |
| |
| def scrape_snapchat_stories(): |
| """استخراج قصص سناب شات العامة من واجهة الويب""" |
| url = f"https://www.snapchat.com/add/{SNAPCHAT_USER}" |
| headers = { |
| "User-Agent": ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) " |
| "Chrome/130.0.0.0 Safari/537.36" |
| ), |
| "Accept-Language": "en-US,en;q=0.9,ar;q=0.8" |
| } |
|
|
| try: |
| logging.info(f"فحص الحساب: {SNAPCHAT_USER} ...") |
| res = requests.get(url, headers=headers, timeout=20) |
|
|
| if res.status_code == 404: |
| msg = f"الحساب '{SNAPCHAT_USER}' غير موجود أو غير متاح للعامة." |
| logging.warning(msg) |
| |
| alert_manager.report("snapchat_account_not_found", msg, critical=True) |
| return [] |
|
|
| if res.status_code != 200: |
| msg = f"فشل في الاتصال بسناب شات. كود الحالة: {res.status_code}" |
| logging.error(msg) |
| alert_manager.report("snapchat_fetch_error", msg) |
| return [] |
|
|
| |
| alert_manager.reset("snapchat_fetch_error") |
|
|
| soup = BeautifulSoup(res.text, 'html.parser') |
| script_data = soup.find('script', id='__NEXT_DATA__') |
| if not script_data: |
| for tag in soup.find_all('script', {'type': 'application/json'}): |
| if 'mediaUrl' in tag.text or 'snapList' in tag.text or 'publicStories' in tag.text: |
| script_data = tag |
| break |
|
|
| if not script_data: |
| msg = "لم يتم العثور على بيانات القصص في كود الصفحة (قد يكون سناب شات غيّر بنية الصفحة)." |
| logging.warning(msg) |
| alert_manager.report("snapchat_parse_error", msg) |
| return [] |
|
|
| try: |
| json_content = json.loads(script_data.string) |
| except Exception as e: |
| msg = f"خطأ في قراءة نص الـ JSON: {e}" |
| logging.error(msg) |
| alert_manager.report("json_parse_error", msg) |
| return [] |
|
|
| stories = [] |
| seen_urls = set() |
|
|
| def extract_snaps_recursively(obj): |
| if isinstance(obj, dict): |
| media_url = obj.get('mediaUrl') or obj.get('url') |
| if not media_url and isinstance(obj.get('snapUrls'), dict): |
| media_url = obj['snapUrls'].get('mediaUrl') |
| video_url = obj.get('videoUrl') or obj.get('playbackUrl') or obj.get('defaultMediaUrl') |
| if video_url and isinstance(video_url, str) and ('.mp4' in video_url or 'video' in video_url): |
| media_url = video_url |
|
|
| if media_url and isinstance(media_url, str) and media_url.startswith('http'): |
| if 'profile' not in media_url.lower() and 'avatar' not in media_url.lower(): |
| if media_url not in seen_urls: |
| seen_urls.add(media_url) |
| raw_id = obj.get('id') or obj.get('snapId') |
| if isinstance(raw_id, dict) and 'value' in raw_id: |
| snap_id = str(raw_id['value']) |
| elif raw_id: |
| snap_id = str(raw_id) |
| else: |
| snap_id = str(hash(media_url)) |
|
|
| media_type = str(obj.get('mediaType', '')).upper() |
| is_vid = 'VIDEO' in media_type or '.mp4' in media_url or 'video' in media_url |
| if 'thumbnail' in media_url.lower() and not is_vid: |
| pass |
| else: |
| timestamp = ( |
| obj.get('timestampInSec') |
| or obj.get('creationTimestamp') |
| or int(time.time()) |
| ) |
| stories.append({ |
| "id": str(snap_id), |
| "url": media_url, |
| "is_video": is_vid, |
| "time": timestamp |
| }) |
|
|
| for val in obj.values(): |
| extract_snaps_recursively(val) |
| elif isinstance(obj, list): |
| for item in obj: |
| extract_snaps_recursively(item) |
|
|
| try: |
| extract_snaps_recursively(json_content) |
| except Exception as ex: |
| msg = f"خطأ غير متوقع أثناء الفحص التكراري للـ JSON: {ex}" |
| logging.error(msg) |
| alert_manager.report("recursive_parse_error", msg, critical=True) |
|
|
| return stories |
|
|
| except requests.exceptions.Timeout: |
| msg = f"انتهت مهلة الاتصال بسناب شات (timeout) للحساب: {SNAPCHAT_USER}" |
| logging.error(msg) |
| alert_manager.report("snapchat_timeout_error", msg) |
| return [] |
| except Exception as e: |
| msg = f"خطأ غير متوقع أثناء الاتصال بسناب شات: {e}" |
| logging.error(msg) |
| alert_manager.report("snapchat_fetch_error", msg, critical=True) |
| return [] |
|
|
|
|
| |
| |
| |
| def bot_worker(): |
| """الحلقة التكرارية لعمل البوت في الخلفية""" |
| logging.info("بدء تشغيل عامل البوت في الخلفية...") |
| time.sleep(5) |
|
|
| |
| _send_startup_notification() |
|
|
| while True: |
| posted_ids = load_posted_stories() |
| current_stories = scrape_snapchat_stories() |
|
|
| logging.info(f"تم العثور على {len(current_stories)} قصة عامة متاحة.") |
|
|
| new_posted = False |
| for story in current_stories: |
| s_id = story["id"] |
| if s_id not in posted_ids: |
| logging.info(f"قصة جديدة تم رصدها! المعرف: {s_id[:10]}...") |
| caption = ( |
| f"منشور جديد من <b>{SNAPCHAT_USER}</b>\n\n" |
| f"<a href='https://www.snapchat.com/add/{SNAPCHAT_USER}'>مشاهدة الحساب على سناب شات</a>" |
| ) |
| success = send_to_telegram(story["url"], story["is_video"], caption) |
| if success: |
| posted_ids.append(s_id) |
| new_posted = True |
| time.sleep(3) |
|
|
| if new_posted: |
| save_posted_stories(posted_ids) |
|
|
| logging.info(f"الدخول في وضع الانتظار لمدة {CHECK_INTERVAL // 60} دقيقة...") |
| time.sleep(CHECK_INTERVAL) |
|
|
|
|
| def _send_startup_notification(): |
| """[جديد] إرسال رسالة تأكيد تشغيل البوت عند بدء التشغيل""" |
| base_url = f"{TELEGRAM_BASE_URL.rstrip('/')}/bot{TELEGRAM_BOT_TOKEN}/" |
| endpoint = base_url + "sendMessage" |
| text = ( |
| f"<b>[INFO] تم تشغيل البوت بنجاح</b>\n\n" |
| f"يتم الان مراقبة حساب: <b>{SNAPCHAT_USER}</b>\n" |
| f"فترة الفحص: كل <b>{CHECK_INTERVAL // 60}</b> دقيقة\n" |
| f"عتبة تنبيه الخطأ: بعد <b>{ERROR_ALERT_THRESHOLD}</b> تكرارات\n" |
| f"<i>الوقت: {time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime())}</i>" |
| ) |
| try: |
| requests.post(endpoint, json={ |
| "chat_id": TELEGRAM_CHAT_ID, |
| "text": text, |
| "parse_mode": "HTML", |
| "disable_web_page_preview": True |
| }, timeout=15) |
| except Exception as e: |
| logging.warning(f"لم يتم إرسال رسالة بدء التشغيل: {e}") |
|
|
|
|
| |
| |
| |
| @app.route('/') |
| def home(): |
| return f""" |
| <html> |
| <head> |
| <title>Snap2Tele Bot Status</title> |
| <meta charset="utf-8"> |
| <style> |
| body {{ font-family: system-ui; text-align: center; padding: 50px; background: #f8fafc; color: #0f172a; }} |
| .status {{ display: inline-block; padding: 10px 20px; background: #22c55e; color: white; border-radius: 50px; font-weight: bold; }} |
| .info {{ margin-top: 20px; background: #e0f2fe; padding: 15px; border-radius: 10px; display: inline-block; text-align: left; }} |
| </style> |
| </head> |
| <body> |
| <h1>البوت يعمل بنجاح</h1> |
| <p class="status">ONLINE</p> |
| <div class="info"> |
| <p>يتم مراقبة حساب: <b>{SNAPCHAT_USER}</b></p> |
| <p>فترة الفحص: كل <b>{CHECK_INTERVAL // 60}</b> دقيقة</p> |
| <p>عتبة تنبيه الاخطاء: <b>{ERROR_ALERT_THRESHOLD}</b> تكرارات</p> |
| <p>فترة التبريد بين التنبيهات: <b>{ERROR_ALERT_COOLDOWN // 60}</b> دقيقة</p> |
| </div> |
| <br><br> |
| <p>لإبقاء البوت يعمل 24/7 مجاناً، أضف هذا الرابط إلى <b>cron-job.org</b></p> |
| </body> |
| </html> |
| """ |
|
|
|
|
| @app.route('/health') |
| def health(): |
| return { |
| "status": "ok", |
| "user": SNAPCHAT_USER, |
| "timestamp": time.time(), |
| "error_registry": { |
| k: v["count"] |
| for k, v in alert_manager._error_registry.items() |
| } |
| } |
|
|
|
|
| if __name__ == '__main__': |
| worker_thread = threading.Thread(target=bot_worker, daemon=True) |
| worker_thread.start() |
|
|
| port = int(os.environ.get('PORT', 7860)) |
| app.run(host='0.0.0.0', port=port) |
|
|
