from flask import Flask, request, jsonify import requests import pymysql from pymysql.cursors import DictCursor import json import os import time import gc import logging from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Lock import subprocess import sys from datetime import datetime app = Flask(__name__) # --------------------------- # إعداد السجلات # --------------------------- logging.basicConfig( filename='series_importer.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) # --------------------------- # إعداد جلسة requests مع Retry # --------------------------- session = requests.Session() retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("https://", adapter) session.mount("http://", adapter) # --------------------------- # إعداد قاعدة البيانات (لم أنغي شيء لكي لا تتغير آلية الإدخال) # إذا رغبت لاحقًا ننقل هذه الإعدادات إلى متغيرات بيئة. # --------------------------- db_config = { "host": "localhost", "user": "xonegold_tv", "password": "]5[k5eWg#zqT", "database": "xonegold_xone", "cursorclass": DictCursor } # مهلات وأعدادات قابلة للتعديل EPISODE_TIMEOUT = 180 STREAM_TIMEOUT = 30 DOWNLOAD_TIMEOUT = 30 # أقل وقت انتظار افتراضي بين طلبات لمسلسل أو دفعة (قابل للتعديل) RATE_LIMIT_DELAY = 3 # أقصى عدد من الخيوط لجلب المصادر في نفس الوقت (مصادرك 5 لذا نضع 5) MAX_SOURCE_WORKERS = 5 def get_db_connection(): return pymysql.connect(**db_config) # تنظيف الذاكرة بنظامية أقل تواتراً def cleanup_memory(): gc.collect() logging.info("تم تنظيف الذاكرة من الكائنات المؤقتة") # --------------------------- # خريطة السيرفرات والـ APIs كما كانت # --------------------------- server_name_mapping = { "gdrive": "GoogleDrive", "vidguard": "M3u8", "cimafree": "Streamwish", "wecima": "Streamwish", "farfesh": "Streamwish" } server_order_mapping = { "gdrive": 1, "vidguard": 2, "cimafree": 3, "wecima": 4, "farfesh": 5 } BASE_API_URL = 'https://iptv.xonegold.com/add' EPISODE_API_URL = f'{BASE_API_URL}/add_episode' STREAM_LINK_API_URL = f'{BASE_API_URL}/add_stream_link' DOWNLOAD_LINK_API_URL = f'{BASE_API_URL}/add_download_link' APIS = { "gdrive": "https://iptv.xonegold.com/google/go?series_name={name}", "vidguard": "https://iptv.xonegold.com/vidguard/go?series_name={name}", "cimafree": "https://iptv.xonegold.com/cimafree/api/search_series?name={name}", "wecima": "https://iptv.xonegold.com/wecima/api/search_series?name={name}", "farfesh": "https://iptv.xonegold.com/farfesh/api/search_series?name={name}" } EPISODE_DEFAULTS = { 'episoade_image': '', 'episoade_description': '', 'downloadable': 1, 'type': 0, 'status': 1, 'source': '', 'url': '', 'skip_available': 0, 'intro_start': '', 'intro_end': '', 'end_credits_marker': '0', 'drm_uuid': '', 'drm_license_uri': '' } STREAM_DEFAULTS = { 'size': '', 'quality': '1080p', 'status': 1, 'skip_available': 0, 'intro_start': '', 'intro_end': '', 'end_credits_marker': '0', 'link_type': 0, 'drm_uuid': '', 'drm_license_uri': '' } DOWNLOAD_DEFAULTS = { 'size': '', 'quality': '1080p', 'link_order': 1, 'download_type': 'Internal', 'status': 1 } SAVE_EPISODES_FILE = "save_episodes_data.json" SERIES_LIST_FILE = "series_list.json" # تأمين عملية الكتابة على ملف السجل من تعدد الخيوط _save_file_lock = Lock() # --------------------------- # دوال مساعدة للقراءة والكتابة الآمنة لملفات JSON (كتابة ذرّية) # --------------------------- def load_json_file(path): if not os.path.exists(path): return {} try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logging.error(f"فشل قراءة JSON من {path}: {e}") return {} def save_json_file_atomic(path, data): tmp = f"{path}.tmp" try: with _save_file_lock: with open(tmp, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp, path) except Exception as e: logging.error(f"فشل كتابة JSON إلى {path}: {e}") # --------------------------- # دالة جلب حلقات مصدر واحد مع فحص نوع الرد # تُستخدم داخل fetch_all_sources بشكل متوازي # --------------------------- def fetch_episodes_from_source(source, series_info, timeout=EPISODE_TIMEOUT): try: if source in ["cimafree", "wecima", "farfesh"]: custom_url = series_info.get("urls", {}).get(source) if not custom_url: return source, [] url = f"https://iptv.xonegold.com/{source}/api/series?series_url={custom_url}" else: search_name = series_info.get("search_name", series_info["name"]) url = APIS[source].format(name=search_name) # لا نضع sleep هنا — التحكم بمعدل الطلبات يتم مركزياً res = session.get(url, timeout=timeout) res.raise_for_status() # نتحقق آمنًا من JSON try: data = res.json() except ValueError: logging.error(f"استجابة ليست JSON من {source} للمسلسل {series_info.get('name')}") return source, [] episodes = data.get("episodes", []) if source in ["cimafree", "wecima", "farfesh"] else data return source, episodes or [] except Exception as e: logging.error(f"خطأ في جلب الحلقات من {source}: {e}") return source, [] # --------------------------- # جلب كل المصادر بشكل متوازي (مناسب للاستضافة المشتركة لأن عدد المصادر محدود) # ترجع dict: {source: episodes_list} # --------------------------- def fetch_all_sources(series_info): results = {} with ThreadPoolExecutor(max_workers=MAX_SOURCE_WORKERS) as executor: futures = {executor.submit(fetch_episodes_from_source, src, series_info): src for src in APIS.keys()} for fut in as_completed(futures): src = futures[fut] try: source, episodes = fut.result() results[source] = episodes except Exception as e: logging.error(f"استثناء عند تنفيذ fetch من {src}: {e}") results[src] = [] return results # --------------------------- # دوال قواعد البيانات وملفات السجل (لم أغير سلوك الإدخال) # --------------------------- def episode_exists(cursor, season_id, episode_order): cursor.execute( "SELECT id FROM web_series_episoade WHERE season_id=%s AND episoade_order=%s", (season_id, episode_order) ) row = cursor.fetchone() return row[0] if row else None def save_episode_log(series_name, episode_name, episode_order, source): log_data = load_json_file(SAVE_EPISODES_FILE) if series_name not in log_data: log_data[series_name] = {} order_str = str(episode_order) if order_str not in log_data[series_name]: log_data[series_name][order_str] = { "name": episode_name, "sources": [] } if source not in log_data[series_name][order_str]["sources"]: log_data[series_name][order_str]["sources"].append(source) save_json_file_atomic(SAVE_EPISODES_FILE, log_data) # --------------------------- # نقطة الدخول لتحديث مسلسل واحد (تحسينات: جلب مصادر متوازي، تقليل الانتظار الثابت، # كاش لمصادر إضافية لمنع جلب متكرر لكل حلقة) # --------------------------- @app.route('/update_series', methods=['GET']) def update_series(): series_name = request.args.get('name') if not os.path.exists(SERIES_LIST_FILE): return jsonify({"error": f"{SERIES_LIST_FILE} غير موجود"}), 400 series_data = load_json_file(SERIES_LIST_FILE).get("series", []) series_info = next((s for s in series_data if s.get("name") == series_name), None) if not series_info: return jsonify({"error": "المسلسل غير موجود في series_list.json"}), 404 season_id = series_info.get("season_id") if not season_id: return jsonify({"error": f"season_id غير موجود للمسلسل: {series_name}"}), 400 # جلب كل المصادر مرة واحدة all_sources_episodes = fetch_all_sources(series_info) # السيرفرات الفعلية (اللي فيها حلقات) available_sources = [src for src, eps in all_sources_episodes.items() if eps] # اختيار المرجع الرئيسي reference_source = None reference_data = None for src in sorted(server_order_mapping.keys(), key=lambda s: server_order_mapping[s]): eps = all_sources_episodes.get(src) or [] if eps: reference_source = src reference_data = eps break if not reference_data: return jsonify({"error": "لم يتم العثور على حلقات من أي مصدر"}), 404 # تحميل سجل الحلقات log_data = load_json_file(SAVE_EPISODES_FILE) try: conn = get_db_connection() cursor = conn.cursor() # 1️⃣ إضافة الحلقات من المرجع الرئيسي for ep in reference_data: order = ep.get("episode_order") or ep.get("episoade_order") name = ep.get("episode_name") or ep.get("Episoade_Name") url = ep.get("url") if not order or not name or not url: continue order_str = str(order) existing_sources = log_data.get(series_name, {}).get(order_str, {}).get("sources", []) if reference_source in existing_sources: # المرجع موجود للحلقة دي، نكمل بعدين من الباقي continue db_row = episode_exists(cursor, season_id, order) if db_row: ep_id = db_row[0] else: # إنشاء الحلقة episode_payload = EPISODE_DEFAULTS.copy() episode_payload.update({ "season_id": season_id, "Episoade_Name": name, "episoade_order": order }) try: ep_res = session.post(EPISODE_API_URL, json=episode_payload, timeout=EPISODE_TIMEOUT) ep_res.raise_for_status() try: ep_json = ep_res.json() except ValueError: logging.error(f"إجابة add_episode ليست JSON للمسلسل {series_name} حل {order}") continue ep_id = ep_json.get("episode_id") except Exception as e: logging.error(f"فشل إنشاء الحلقة عبر API للمسلسل {series_name} حل {order}: {e}") continue if not ep_id: logging.error(f"لم يتم إرجاع episode_id عند إضافة الحلقة: {series_name} حل {order}") continue # حفظ المرجع save_episode_log(series_name, name, order, reference_source) log_data = load_json_file(SAVE_EPISODES_FILE) # إضافة رابط البث الرئيسي stream_payload = STREAM_DEFAULTS.copy() stream_payload.update({ "episode_id": ep_id, "name": f"سيرفر {server_order_mapping[reference_source]}", "url": url, "type": server_name_mapping[reference_source], "link_order": server_order_mapping[reference_source] }) try: session.post(STREAM_LINK_API_URL, json=stream_payload, timeout=STREAM_TIMEOUT) except Exception as e: logging.error(f"فشل إضافة stream link للمسلسل {series_name} حل {order} من {reference_source}: {e}") # رابط التحميل لو GDrive if reference_source == "gdrive": download_payload = DOWNLOAD_DEFAULTS.copy() download_payload.update({ "episode_id": ep_id, "name": f"تحميل {name}", "url": url, "type": server_name_mapping[reference_source] }) try: session.post(DOWNLOAD_LINK_API_URL, json=download_payload, timeout=DOWNLOAD_TIMEOUT) except Exception as e: logging.error(f"فشل إضافة download link للمسلسل {series_name} حل {order}: {e}") time.sleep(RATE_LIMIT_DELAY) # 2️⃣ إكمال الحلقات الناقصة من باقي السيرفرات log_data = load_json_file(SAVE_EPISODES_FILE) for src in available_sources: if src == reference_source: continue for ep in all_sources_episodes.get(src, []): order = ep.get("episode_order") or ep.get("episoade_order") name = ep.get("episode_name") or ep.get("Episoade_Name") url = ep.get("url") if not order or not name or not url: continue order_str = str(order) existing_sources = log_data.get(series_name, {}).get(order_str, {}).get("sources", []) # لو السيرفر ده مضاف خلاص → تجاهل if src in existing_sources: continue # الحصول على episode_id db_row = episode_exists(cursor, season_id, order) if not db_row: # لو الحلقة مش موجودة أصلاً، نعملها من السيرفر الحالي episode_payload = EPISODE_DEFAULTS.copy() episode_payload.update({ "season_id": season_id, "Episoade_Name": name, "episoade_order": order }) try: ep_res = session.post(EPISODE_API_URL, json=episode_payload, timeout=EPISODE_TIMEOUT) ep_res.raise_for_status() ep_id = ep_res.json().get("episode_id") if not ep_id: continue except Exception as e: logging.error(f"فشل إنشاء الحلقة من {src} للمسلسل {series_name} حل {order}: {e}") continue else: ep_id = db_row[0] # إضافة السيرفر الحالي extra_stream = STREAM_DEFAULTS.copy() extra_stream.update({ "episode_id": ep_id, "name": f"سيرفر {server_order_mapping[src]}", "url": url, "type": server_name_mapping[src], "link_order": server_order_mapping[src] }) try: session.post(STREAM_LINK_API_URL, json=extra_stream, timeout=STREAM_TIMEOUT) except Exception as e: logging.error(f"فشل إضافة extra stream للمسلسل {series_name} حل {order} من {src}: {e}") continue # حفظ في السجل save_episode_log(series_name, name, order, src) log_data = load_json_file(SAVE_EPISODES_FILE) time.sleep(RATE_LIMIT_DELAY) except Exception as e: logging.error(f"حدث خطأ في معالجة المسلسل {series_name}: {e}") return jsonify({"error": f"حدث خطأ في معالجة المسلسل: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() cleanup_memory() return jsonify({"message": f"تم تحديث المسلسل {series_name} بنجاح"}) # --------------------------- # تحديث جماعي (بسيط، مع استخدام update_series بدون إعادة جلب كل شيء) # يحافظ على BATCH_SIZE صغير لتوافق الاستضافة المشتركة # --------------------------- @app.route('/bulk_update', methods=['POST', 'GET']) def bulk_update(): try: if not os.path.exists(SERIES_LIST_FILE): return jsonify({"error": f"{SERIES_LIST_FILE} غير موجود"}), 400 data = load_json_file(SERIES_LIST_FILE) results = [] BATCH_SIZE = 1 # احتفظت بقيمة 1 كما في نسختك الأصلية series_list = data.get("series", []) for i in range(0, len(series_list), BATCH_SIZE): batch = series_list[i:i+BATCH_SIZE] logging.info(f"جاري معالجة الدفعة {i//BATCH_SIZE + 1}") for series_info in batch: try: logging.info(f"بدء معالجة المسلسل: {series_info['name']}") # نستخدم عرض مسار update_series بنفس الشكل with app.test_request_context(f"/update_series?name={series_info['name']}"): response = update_series() if isinstance(response, tuple): response_body = response[0] else: response_body = response results.append({series_info["name"]: json.loads(response_body.get_data(as_text=True))}) # راحة قصيرة بين المسلسلات time.sleep(RATE_LIMIT_DELAY) cleanup_memory() except Exception as e: logging.error(f"خطأ في معالجة المسلسل {series_info['name']}: {e}") results.append({series_info["name"]: {"error": str(e)}}) continue # راحة قصيرة بين الدفعات time.sleep(RATE_LIMIT_DELAY) cleanup_memory() return jsonify({"result": results}), 200 except Exception as e: logging.error(f"حدث خطأ عام في bulk_update: {e}") return jsonify({"error": str(e)}), 500 @app.route('/bulk_update_bg', methods=['GET']) def bulk_update_background(): """ تشغيل عملية التحديث bulk_update في الخلفية بدون الانتظار لانهائها """ try: start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') with open('bulk_update_bg.log', 'a', encoding='utf-8') as f: f.write(f"\n--- بدء عملية جديدة في الخلفية ---\n") f.write(f"وقت البدء: {start_time}\n") # تشغيل نسخة من نفس السكربت مع باراميتر خاص subprocess.Popen( [sys.executable, __file__, 'run_bulk_update'], stdout=open('bulk_update_bg.log', 'a'), stderr=subprocess.STDOUT ) return jsonify({"message": "تم بدء عملية التحديث في الخلفية"}), 200 except Exception as e: return jsonify({"error": str(e)}), 500 if __name__ == '__main__': if len(sys.argv) > 1 and sys.argv[1] == 'run_bulk_update': try: with app.test_request_context("/bulk_update"): bulk_update() finally: end_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') with open('bulk_update_bg.log', 'a', encoding='utf-8') as f: f.write(f"وقت الانتهاء: {end_time}\n") f.write(f"--- نهاية العملية ---\n") else: app.run(debug=True, threaded=True)