| import os |
| import time |
| import json |
| import logging |
| import threading |
| import requests |
| from bs4 import BeautifulSoup |
| from flask import Flask |
|
|
| |
| |
| |
| |
| 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" |
|
|
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') |
|
|
| app = Flask(__name__) |
|
|
| 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: |
| logging.error(f"خطأ في قراءة ملف السجل: {e}") |
| 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: |
| logging.error(f"خطأ في حفظ ملف السجل: {e}") |
|
|
| def send_to_telegram(media_url, is_video, caption): |
| """تحميل المنشور الفعلي (صورة أو فيديو) من خوادم سناب شات وإعادة رفعه كملف مباشر إلى التليجرام""" |
| base_url = f"{TELEGRAM_BASE_URL.rstrip('/')}/bot{TELEGRAM_BOT_TOKEN}/" |
| |
| try: |
| logging.info(f"📥 جاري تنزيل ملف القصة من سناب شات (حجم كامل)...") |
| 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: |
| logging.error(f"❌ فشل في تنزيل محتوى القصة. كود الحالة: {media_response.status_code}") |
| return False |
|
|
| 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("✅ تم إرسال القصة إلى تليجرام بنجاح!") |
| return True |
| else: |
| logging.error(f"⚠️ خطأ من تليجرام: {resp.text}") |
| return False |
| |
| except Exception as e: |
| logging.error(f"حدث خطأ أثناء الإرسال للتليجرام: {e}") |
| return False |
|
|
| def scrape_snapchat_stories(): |
| """استخراج قصص سناب شات العامة من واجهة الويب (تعمل لعام 2026)""" |
| 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: |
| logging.warning("الحساب غير موجود أو غير متاح للعامة.") |
| return [] |
| |
| 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: |
| logging.warning("لم يتم العثور على بيانات القصص في كود الصفحة.") |
| return [] |
|
|
| try: |
| json_content = json.loads(script_data.string) |
| except Exception as e: |
| logging.error(f"خطأ في قراءة نص الـ JSON: {e}") |
| 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: |
| logging.error(f"خطأ غير متوقع أثناء الفحص التكراري للـ JSON: {ex}") |
| |
| return stories |
|
|
| except Exception as e: |
| logging.error(f"خطأ أثناء الاتصال بسناب شات: {e}") |
| return [] |
|
|
| def bot_worker(): |
| """الحلقة التكرارية لعمل البوت في الخلفية""" |
| logging.info("🚀 بدء تشغيل عامل البوت في الخلفية...") |
| |
| |
| time.sleep(5) |
| |
| 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🔗 <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) |
|
|
| |
| |
| |
| @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; }} |
| </style> |
| </head> |
| <body> |
| <h1>🚀 البوت يعمل بنجاح</h1> |
| <p class="status">● ONLINE</p> |
| <p>يتم مراقبة حساب سناب شات: <b>{SNAPCHAT_USER}</b></p> |
| <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()} |
|
|
| 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) |
