پاسخ تمرینات ماژول threading – کلاس Semaphore و BoundedSemaphore

import requests
import os
import threading

# فهرست آدرس‌های فایل‌ها
file_urls = [f"https://dummyimage.com/{i}.png" for i in range(300, 400)]

# ایجاد پوشه downloads اگر وجود نداشته باشد
if not os.path.exists("downloads"):
    os.makedirs("downloads")

# Semaphore برای محدود کردن تعداد تردهای فعال هم‌زمان
max_threads = 5
thread_limit = threading.BoundedSemaphore(max_threads)

def download_file(file_url, cookies, headers):
    # با استفاده از Semaphore برای کنترل تعداد تردها
    with thread_limit:
        file_name = os.path.basename(file_url)
        file_path = os.path.join("downloads", file_name)
        try:
            print(f"Downloading {file_name}...")
            response = requests.get(file_url, cookies=cookies, headers=headers, timeout=10)
            response.raise_for_status()  # اگر خطای HTTP رخ دهد
            with open(file_path, "wb") as f:
                f.write(response.content)
            print(f"{file_name} downloaded successfully.")
        except Exception as e:
            print(f"❌ Error downloading {file_name}: {e}")

# ساخت و مدیریت تردها
threads = []
for file_url in file_urls:
    t = threading.Thread(target=download_file, args=(file_url, {}, {}))
    threads.append(t)
    t.start()

# صبر کردن تا تمام تردها کارشان را تمام کنند
for t in threads:
    t.join()

print("✅ All files downloaded successfully.")

import requests
import os
import threading

# فهرست آدرس‌های فایل‌ها
file_urls = [f"https://dummyimage.com/{i}.png" for i in range(300, 400)]

# ایجاد پوشه downloads اگر وجود نداشته باشد
if not os.path.exists("downloads"):
    os.makedirs("downloads")

# Semaphore برای محدود کردن تعداد تردهای فعال هم‌زمان
max_threads = 5
thread_limit = threading.BoundedSemaphore(max_threads)

def download_file(file_url, cookies, headers):
    # با استفاده از Semaphore برای کنترل تعداد تردها
    with thread_limit:
        file_name = os.path.basename(file_url)
        file_path = os.path.join("downloads", file_name)
        try:
            print(f"Downloading {file_name}...")
            response = requests.get(file_url, cookies=cookies, headers=headers, timeout=10)
            response.raise_for_status()  # اگر خطای HTTP رخ دهد
            with open(file_path, "wb") as f:
                f.write(response.content)
            print(f"{file_name} downloaded successfully.")
        except Exception as e:
            print(f"❌ Error downloading {file_name}: {e}")

# ساخت و مدیریت تردها
threads = []
for file_url in file_urls:
    t = threading.Thread(target=download_file, args=(file_url, {}, {}))
    threads.append(t)
    t.start()

# صبر کردن تا تمام تردها کارشان را تمام کنند
for t in threads:
    t.join()

print("✅ All files downloaded successfully.")