أتمتة إدارة بروكسي التصادم | أصلي، ترجم بواسطة AI
هذا المنشور يوضح برنامجًا نصيًا بلغة بايثون، clash.py، مصممًا لأتمتة إدارة تكوين بروكسي Clash الخاص بك. يتولى كل شيء بدءًا من تنزيل تكوينات البروكسي المحدثة دوريًا وإعادة تشغيل خدمة Clash وحتى اختيار التحويل الذكي إلى أسرع بروكسي متاح ضمن مجموعة محددة. إلى جانب clash.py، فإن وحدة speed.py تسهل اختبار زمن الوصل لكل بروكسي Clash بشكل متزامن، مما يضمن أن اتصالك دائمًا ما يتم توجيهه عبر الخادم الأمثل.
clash.py
import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse
# بافتراض أن speed.py موجود في نفس الدليل أو متاح في PYTHONPATH
from speed import get_top_proxies
# --- التكوين ---
CLASH_CONTROLLER_HOST = "127.0.0.1"
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
# اسم مجموعة البروكسي التي سيتم تعيين أفضل بروكسي فردي إليها.
# تأكد من وجود هذه المجموعة في تكوين Clash الخاص بك.
TARGET_PROXY_GROUP = "🚧Proxy"
def setup_logging():
"""يضبط تسجيل الأحداث الأساسي للنص البرمجي."""
logging.basicConfig(
filename='clash.log',
level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def start_system_proxy(global_proxy_address):
"""يضبط متغيرات بيئة البروكسي على مستوى النظام."""
os.environ["GLOBAL_PROXY"] = global_proxy_address # يتم تعيينه للاتساق إذا كان مطلوبًا في مكان آخر
os.environ["HTTP_PROXY"] = f"http://{global_proxy_address}"
os.environ["HTTPS_PROXY"] = f"http://{global_proxy_address}"
os.environ["http_proxy"] = f"http://{global_proxy_address}"
os.environ["https_proxy"] = f"http://{global_proxy_address}"
# هذه عادةً لا تحتاج إلى تعيين صريح إلى "false" مع الأدوات الحديثة،
# ولكن يتم الاحتفاظ بها لتوافقها مع نية النص البرمجي الأصلي.
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "false"
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "false"
os.environ["ALL_PROXY"] = os.environ["http_proxy"]
logging.info(f"تم تعيين البروكسي على مستوى النظام إلى: {global_proxy_address}")
def stop_system_proxy():
"""يمسح متغيرات بيئة البروكسي على مستوى النظام."""
os.environ["http_proxy"] = ""
os.environ["HTTP_PROXY"] = ""
os.environ["https_proxy"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # العودة إلى الإعداد الافتراضي
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
os.environ["ALL_PROXY"] = ""
logging.info("تم إيقاف البروكسي على مستوى النظام (تم مسح متغيرات البيئة).")
def switch_clash_proxy_group(group_name, proxy_name):
"""
يحول البروكسي النشط في مجموعة بروكسي Clash محددة إلى بروكسي جديد.
"""
encoded_group_name = urllib.parse.quote(group_name)
url = f"{CLASH_API_BASE_URL}/proxies/{encoded_group_name}"
headers = {"Content-Type": "application/json"}
payload = {"name": proxy_name}
try:
response = requests.put(url, headers=headers, data=json.dumps(payload), timeout=5)
response.raise_for_status()
logging.info(f"تم التحويل بنجاح من '{group_name}' إلى '{proxy_name}'.")
return True
except requests.exceptions.ConnectionError:
logging.error(f"خطأ: لا يمكن الاتصال بـ Clash API في {CLASH_API_BASE_URL} لتحويل البروكسي.")
logging.error("تأكد من أن Clash يعمل وأن external-controller مضبوط.")
return False
except requests.exceptions.Timeout:
logging.error(f"خطأ: انتهت مهلة الاتصال بـ Clash API أثناء تحويل البروكسي لـ '{group_name}'.")
return False
except requests.exceptions.RequestException as e:
logging.error(f"حدث خطأ غير متوقع أثناء تحويل البروكسي لـ '{group_name}': {e}")
return False
def main():
"""الوظيفة الرئيسية لإدارة تكوين Clash، إعادة التشغيل، واختيار أفضل بروكسي."""
setup_logging()
parser = argparse.ArgumentParser(description="نص برمجي لإدارة وتكوين Clash.")
parser.add_argument("--minutes", type=int, default=10, help="الدقائق بين التحديثات (الافتراضي: 10)")
parser.add_argument("--iterations", type=int, default=1000, help="عدد التكرارات (الافتراضي: 1000)")
parser.add_argument(
"--config-url",
type=str,
default=os.getenv("CLASH_DOWNLOAD_URL"),
help="رابط تنزيل تكوين Clash منه. الافتراضي هو متغير البيئة CLASH_DOWNLOAD_URL إذا تم تعيينه، وإلا فسيتم استخدام رابط ثابت."
)
args = parser.parse_args()
ITERATIONS = args.iterations
SLEEP_SECONDS = args.minutes * 60
config_download_url = args.config_url
if not config_download_url:
logging.critical("خطأ: لم يتم تقديم رابط تنزيل التكوين. يرجى تعيين متغير البيئة CLASH_DOWNLOAD_URL أو استخدام وسيطة --config-url.")
return # الخروج إذا لم يكن هناك رابط متاح
clash_executable_path = "/home/lzw/clash-linux-386-v1.17.0/clash-linux-386"
clash_config_dir = os.path.expanduser("~/.config/clash")
clash_config_path = os.path.join(clash_config_dir, "config.yaml")
for i in range(1, ITERATIONS + 1):
logging.info(f"--- بدء التكرار {i} من {ITERATIONS} ---")
# الخطوة 1: إيقاف أي إعدادات بروكسي نظامية موجودة
stop_system_proxy()
# الخطوة 2: تنزيل وتحديث تكوين Clash
try:
logging.info(f"جاري تنزيل التكوين الجديد من: {config_download_url}")
subprocess.run(["wget", config_download_url, "-O", "zhs4.yaml"], check=True, capture_output=True)
os.makedirs(clash_config_dir, exist_ok=True)
shutil.move("zhs4.yaml", clash_config_path)
logging.info("تم تحديث تكوين Clash بنجاح!")
except subprocess.CalledProcessError as e:
logging.error(f"فشل في تنزيل أو نقل ملف التكوين: {e.stderr.decode().strip()}")
logging.error("تخطي إلى التكرار التالي.")
time.sleep(10) # الانتظار قليلاً قبل إعادة المحاولة
continue
except Exception as e:
logging.error(f"حدث خطأ غير متوقع أثناء تحديث التكوين: {e}")
logging.error("تخطي إلى التكرار التالي.")
time.sleep(10)
continue
# الخطوة 3: تشغيل Clash في الخلفية
clash_process = None
try:
# من الضروري أن يبدأ Clash مع تفعيل external-controller ويمكن الوصول إليه
# هذا عادةً ما يتم تكوينه داخل ملف config.yaml نفسه.
clash_process = subprocess.Popen([clash_executable_path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"تم تشغيل Clash بمعرف العملية {clash_process.pid}")
# إعطاء Clash لحظة للتهيئة الكاملة وفتح منفذ API الخاص به
time.sleep(5)
except FileNotFoundError:
logging.critical(f"لم يتم العثور على ملف تنفيذ Clash في: {clash_executable_path}")
logging.critical("يرجى التأكد من صحة المسار وأن Clash مثبت.")
return # خطأ حرج، الخروج من النص البرمجي
except Exception as e:
logging.error(f"فشل في تشغيل Clash: {e}")
logging.error("تخطي إلى التكرار التالي.")
if clash_process: clash_process.terminate()
time.sleep(10)
continue
# الخطوة 4: اختبار سرعات البروكسي واختيار الأفضل
best_proxy_name = None
try:
logging.info("جاري اختبار سرعات البروكسي للعثور على الأفضل...")
top_proxies = get_top_proxies(num_results=1) # الحصول على أفضل بروكسي واحد فقط
if top_proxies:
best_proxy_name = top_proxies[0]['name']
logging.info(f"تم تحديد أفضل بروكسي: '{best_proxy_name}' بزمن وصل {top_proxies[0]['latency']}ms")
else:
logging.warning("لا توجد اختبارات بروكسي ناجحة. لا يمكن اختيار أفضل بروكسي لهذا التكرار.")
except Exception as e:
logging.error(f"خطأ أثناء اختبار سرعة البروكسي: {e}")
# الخطوة 5: تحويل مجموعة بروكسي Clash إلى البروكسي الأفضل (إذا تم العثور عليه)
if best_proxy_name:
# قبل تعيين بروكسي النظام، تأكد من أن Clash مضبوط بشكل صحيح.
# تعيين بروكسي النظام للإشارة إلى بروكسي HTTP المحلي لـ Clash.
# عادةً ما يعمل Clash بروكسي HTTP على المنفذ 7890 (أو ما شابه، تحقق من تكوينك).
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # اضبط إذا كان منفذ HTTP لـ Clash مختلفًا
start_system_proxy(clash_local_proxy_address)
if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
logging.error(f"فشل في تحويل مجموعة Clash '{TARGET_PROXY_GROUP}' إلى '{best_proxy_name}'.")
else:
logging.warning("لم يتم العثور على أفضل بروكسي، تخطي تحويل مجموعة البروكسي وإعداد بروكسي النظام لهذا التكرار.")
# الخطوة 6: الانتظار للمدة المحددة
logging.info(f"الانتظار لمدة {SLEEP_SECONDS / 60} دقائق قبل التكرار التالي...")
time.sleep(SLEEP_SECONDS)
# الخطوة 7: إيقاف عملية Clash
if clash_process:
logging.info("جاري إنهاء عملية Clash...")
clash_process.terminate()
try:
clash_process.wait(timeout=10) # إعطاء Clash وقتًا إضافيًا للإغلاق بلطف
logging.info("تم إيقاف Clash بنجاح.")
except subprocess.TimeoutExpired:
logging.warning("لم يتم إنهاء Clash بلطف، جاري قتل العملية.")
clash_process.kill()
clash_process.wait() # التأكد من قتل العملية بالكامل
except Exception as e:
logging.error(f"خطأ أثناء انتظار إيقاف Clash: {e}")
logging.info(f"--- اكتمل التكرار {i} ---")
logging.info(f"تم إكمال {ITERATIONS} تكرارات. انتهى النص البرمجي.")
if __name__ == "__main__":
main()
speed.py
```python import requests import json import urllib.parse import time from concurrent.futures import ThreadPoolExecutor, as_completed import logging # استيراد وحدة تسجيل الأحداث
— التكوين —
CLASH_CONTROLLER_HOST = “127.0.0.1” # استخدام 127.0.0.1 حيث أن المتحكم على نفس الجهاز CLASH_CONTROLLER_PORT = 9090 CLASH_API_BASE_URL = f”http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}” LATENCY_TEST_URL = “https://github.com” # رابط الاختبار المحدث LATENCY_TEST_TIMEOUT_MS = 5000 # ميلي ثانية CONCURRENT_CONNECTIONS = 10 # عدد الاختبارات المتزامنة
قائمة بأسماء مجموعات البروكسي المعروفة لاستبعادها من اختبار السرعة
هذه عادةً ما لا تكون عقدًا فردية ولكن مجموعات سياسات أو بروكسيات خاصة.
EXCLUDE_PROXY_GROUPS = [ “DIRECT”, “REJECT”, “GLOBAL”, # مستبعد بالفعل افتراضيًا في API “🇨🇳国内网站或资源”, “🌵其它规则外”, “🎬Netflix等国外流媒体”, “📦ChatGPT”, “📹Youtube”, “📺爱奇艺等国内流媒体”, “🚧Proxy”, # أضف أي أسماء مجموعات أخرى تريد استبعادها هنا ]
— إعداد تسجيل الأحداث لـ speed.py —
تكوين تسجيل الأحداث لهذا النص البرمجي المحدد
هذا يضمن أنه عند استيراد speed.py واستدعاء وظائفه،
يذهب إخراجه إلى speed.log، منفصل عن clash_manager.log.
logging.basicConfig( filename=’clash.log’, level=logging.INFO, format=’%(asctime)s - %(levelname)s - %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’ )
اختياريًا، إذا كنت تريد أيضًا رؤية الإخراج في وحدة التحكم، أضف StreamHandler:
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter(‘%(asctime)s - %(levelname)s - %(message)s’)
console_handler.setFormatter(formatter)
logging.getLogger().addHandler(console_handler)
— منطق النص البرمجي —
def get_all_proxy_names(): “"”يجلب جميع أسماء البروكسي من Clash API، باستثناء المجموعات المعروفة.””” try: response = requests.get(f”{CLASH_API_BASE_URL}/proxies”, timeout=5) response.raise_for_status() # إثارة استثناء لأخطاء HTTP (4xx أو 5xx) proxies_data = response.json()
all_names = proxies_data.get("proxies", {}).keys()
# تصفية أسماء المجموعات
filtered_names = [name for name in all_names if name not in EXCLUDE_PROXY_GROUPS]
logging.info(f"تم جلب {len(filtered_names)} أسماء بروكسي قابلة للاختبار بنجاح.")
return filtered_names
except requests.exceptions.ConnectionError:
logging.error(f"لا يمكن الاتصال بـ Clash API في {CLASH_API_BASE_URL}. تأكد من أن Clash يعمل.")
return []
except requests.exceptions.Timeout:
logging.error(f"انتهت مهلة الاتصال بـ Clash API بعد 5 ثوانٍ.")
return []
except requests.exceptions.RequestException as e:
logging.error(f"حدث خطأ غير متوقع أثناء جلب أسماء البروكسي: {e}")
return []
def test_proxy_latency(proxy_name): “"”يختبر زمن الوصل لبروكسي واحد باستخدام Clash API. يعيد tuple (اسم البروكسي, زمن الوصل) أو (اسم البروكسي, None) في حالة الفشل. “”” encoded_proxy_name = urllib.parse.quote(proxy_name) url = f”{CLASH_API_BASE_URL}/proxies/{encoded_proxy_name}/delay” params = { “url”: LATENCY_TEST_URL, “timeout”: LATENCY_TEST_TIMEOUT_MS } try: # مهلة requests بالثواني، تحويل الميلي ثانية response = requests.get(url, params=params, timeout=(LATENCY_TEST_TIMEOUT_MS / 1000) + 1) response.raise_for_status() latency_data = response.json() latency = latency_data.get(“delay”) logging.info(f”البروكسي: {proxy_name} - زمن الوصل: {latency}ms”) return proxy_name, latency except requests.exceptions.RequestException as e: logging.warning(f”خطأ في اختبار ‘{proxy_name}’: {e}”) return proxy_name, None
def get_top_proxies(num_results=5): “”” يختبر سرعات بروكسي Clash بشكل متزامن ويعيد أفضل N بروكسي فردي من حيث السرعة.
الإرجاع:
list: قائمة من القواميس، كل منها يحتوي على 'name' و 'latency' لأفضل البروكسيات.
تعيد قائمة فارغة إذا لم يتم العثور على بروكس