Telegram Channels Checker

import urllib.request
import re
import threading
from queue import Queue

# Dosya yolları
KANALLAR_DOSYASI = 'kanallar.txt'
GEÇERLİ_KANALLAR_DOSYASI = 'geçerli_kanallar.txt'

# Banner
BANNER = print("\033[93mA\033[0m \033[93mL\033[0m \033[91mI\033[0m \033[93mX\033[0m \033[91mA\033[0m \033[91mN\033[0m")
print("\033[93mT\033[0m \033[93mG\033[0m - \033[91mC\033[0m \033[93mH\033[0m \033[91mE\033[0m \033[93mC\033[0m \033[91mK\033[0m \033[93mE\033[0m \033[91mR\033[0m")


# Agent
KULLANICI_AJANI = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36'

# thread sayısı
İŞÇİ_THREAD_SAYISI = 5

kanal_kuyruğu = Queue()

def kanalları_oku():
    with open(KANALLAR_DOSYASI, 'r') as dosya:
        return dosya.read().splitlines()

def geçerli_kanalı_yaz(kanal_bağlantısı):
    with open(GEÇERLİ_KANALLAR_DOSYASI, 'a+') as dosya:
        dosya.write(kanal_bağlantısı + "\n")

def url_al(kanal_bağlantısı):
    try:
        talep = urllib.request.Request(kanal_bağlantısı, headers={'User-Agent': KULLANICI_AJANI})
        with urllib.request.urlopen(talep, timeout=10) as yanıt:
            html = yanıt.read().decode('utf-8')
            return html
    except Exception as e:
        print(f"{kanal_bağlantısı} adresi alınamadı: {e}")
        return None

def kanalı_kontrol_et(kanal_bağlantısı):
    html = url_al(kanal_bağlantısı)
    if html:
        if 'tgme_page_photo_image' in html and 'tgme_page_extra' in html:
            kanal_adı_eşleşmesi = re.search(r'<div class="tgme_page_title">(.*?)</div>', html)
            if kanal_adı_eşleşmesi:
                kanal_adı = kanal_adı_eşleşmesi.group(1).strip()
            else:
                kanal_adı = "Bilinmiyor"
            abone_sayısı_eşleşmesi = re.search(r'<div class="tgme_page_extra">(.*?)</div>', html)
            if abone_sayısı_eşleşmesi:
                abone_sayısı = abone_sayısı_eşleşmesi.group(1).strip()
            else:
                abone_sayısı = "Bilinmiyor"
            print(f"[TG KONTROL] {kanal_bağlantısı}: geçerli: {kanal_adı} ({abone_sayısı})")
            geçerli_kanalı_yaz(kanal_bağlantısı)
        else:
            print(f"[TG KONTROL] {kanal_bağlantısı}: geçersiz")
    else:
        print(f"[TG KONTROL] {kanal_bağlantısı}: hata")

def işçi():
    while True:
        kanal_bağlantısı = kanal_kuyruğu.get()
        if kanal_bağlantısı is None:
            break
        kanalı_kontrol_et(kanal_bağlantısı)
        kanal_kuyruğu.task_done()

def ana():
    for kanal_bağlantısı in kanalları_oku():
        kanal_kuyruğu.put(kanal_bağlantısı)

    # thread'lerini başlaması
    threadler = []
    for _ in range(İŞÇİ_THREAD_SAYISI):
        t = threading.Thread(target=işçi)
        t.start()
        threadler.append(t)

    # Tüm thread'lerinin tamamlanmasını
    kanal_kuyruğu.join()

    # thread'ler durur
    for _ in range(İŞÇİ_THREAD_SAYISI):
        kanal_kuyruğu.put(None)
    for t in threadler:
        t.join()

if __name__ == '__main__':
    print(BANNER)
    ana()