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=3, status_forcelist=[ 408, # Request Timeout 429, # Too Many Requests 500, # Internal Server Error 502, # Bad Gateway 503, # Service Unavailable 504 # Gateway Timeout ], allowed_methods=["GET", "POST"], raise_on_status=False ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("https://", adapter) session.mount("http://", adapter) # --------------------------- # إعداد قاعدة البيانات (لم أنغي شيء لكي لا تتغير آلية الإدخال) # إذا رغبت لاحقًا ننقل هذه الإعدادات إلى متغيرات بيئة. # --------------------------- db_config = { "host": "74.208.45.9", "user": "newpanal", "password": "ahmedxonegold@2712", "database": "ahmedxonegold_newxone", "cursorclass": DictCursor } # مهلات وأعدادات قابلة للتعديل EPISODE_TIMEOUT = 260 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}") # --------------------------- # دالة جلب حلقات مصدر واحد مع فحص نوع الرد # --------------------------- 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) res = session.get(url, timeout=timeout) res.raise_for_status() 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, [] # --------------------------- # جلب كل المصادر بشكل متوازي # --------------------------- 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 LIMIT 1", (season_id, episode_order) ) row = cursor.fetchone() return row[0] if row else None def link_exists(cursor, ep_id, url, src_type): """تحقق من وجود الرابط مسبقًا في play/download""" cursor.execute( "SELECT id FROM episode_play_links WHERE episode_id=%s AND url=%s AND type=%s LIMIT 1", (ep_id, url, src_type) ) if cursor.fetchone(): return True cursor.execute( "SELECT id FROM episode_download_links WHERE episode_id=%s AND url=%s AND type=%s LIMIT 1", (ep_id, url, src_type) ) return bool(cursor.fetchone()) 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) # --------------------------- # نقطة الدخول لتحديث مسلسل واحد # --------------------------- def get_episode_id(cursor, season_id, order, name): ep_id = local_episode_ids[season_id].get(order) if ep_id: return ep_id db_row = episode_exists(cursor, season_id, order) if db_row: ep_id = db_row 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() ep_id = ep_res.json().get("episode_id") if not ep_id: return None except Exception as e: logging.error(f"فشل إنشاء الحلقة {name} حل {order}: {e}") return None local_episode_ids[season_id][order] = ep_id return ep_id def add_stream_link(ep_id, src, name, url): payload = STREAM_DEFAULTS.copy() payload.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=payload, timeout=STREAM_TIMEOUT) except Exception as e: logging.error(f"فشل إضافة stream link حل {name} من {src}: {e}") if src == "gdrive": download_payload = DOWNLOAD_DEFAULTS.copy() download_payload.update({ "episode_id": ep_id, "name": f"تحميل {name}", "url": url, "type": server_name_mapping[src] }) try: session.post(DOWNLOAD_LINK_API_URL, json=download_payload, timeout=DOWNLOAD_TIMEOUT) except Exception as e: logging.error(f"فشل إضافة download link حل {name}: {e}") def process_source_episodes(cursor, season_id, series_name, source, episodes, log_data): for ep in episodes: 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) local_episode_ids.setdefault(season_id, {}) existing_sources = log_data.get(series_name, {}).get(order_str, {}).get("sources", []) if source in existing_sources: continue ep_id = get_episode_id(cursor, season_id, order, name) if not ep_id: continue # تحقق من الرابط في DB src_type = server_name_mapping[source] if link_exists(cursor, ep_id, url, src_type): continue log_data.setdefault(series_name, {}).setdefault(order_str, {"name": 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) add_stream_link(ep_id, source, name, url) @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 = next((src for src in sorted(server_order_mapping.keys(), key=lambda s: server_order_mapping[s]) if all_sources_episodes.get(src)), None) reference_data = all_sources_episodes.get(reference_source) if reference_source else None if not reference_data: return jsonify({"error": "لم يتم العثور على حلقات من أي مصدر"}), 404 log_data = load_json_file(SAVE_EPISODES_FILE) global local_episode_ids local_episode_ids = {} try: conn = get_db_connection() cursor = conn.cursor() process_source_episodes(cursor, season_id, series_name, reference_source, reference_data, log_data) for src in available_sources: if src != reference_source: process_source_episodes(cursor, season_id, series_name, src, all_sources_episodes.get(src, []), log_data) 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)