# Gerekli kütüphaneler ve modüllerin import edilmesi
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from time import sleep
import time
import logging
import sys
import pyotp
from selenium.common.exceptions import StaleElementReferenceException
from firefox_setup import initialize_driver # Firefox sürücüsünün başlatılması
# Web sayfasındaki öğelerin XPath lokasyonlarını içeren bir sözlük (dizin)
LOCATORS = {
"cookie_accept": "//button[text()='Opsiyonel çerezleri reddet']", # Çerezlerin kabul edilmesi düğmesi
"cookie_text": "//*[text()='Bu tarayıcıda Instagram\'ın çerez kullanımına izin verilsin mi?']", # Çerez metni
"login_username_field": "//input[@name='username']", # Kullanıcı adı giriş alanı
"login_password_field": "//input[@name='password']", # Şifre giriş alanı
"new_dm_btn": "//div[@role= 'button'][.//div//*//*[text()='Yeni mesaj']]", # Yeni mesaj düğmesi
"dm_type_username": "//input[@placeholder='Ara...']", # Kullanıcı adı arama alanı
"dm_select_user": '//span/span/span[text()="{}"]', # Kullanıcı adını seçmek için XPath
"dm_user_not_found": "//span[text()='Hesap bulunamadı.']", # Kullanıcı bulunamadı hatası metni
"dm_start_chat_btn": "//div/div[text()='Sohbet' and @role='button']", # Sohbet başlat düğmesi
"dm_msg_field": "//div[@role='textbox' and @aria-label='Mesaj']", # Mesaj giriş alanı
"dm_notification_present": "//span[text()='Bildirimleri Aç']", # Bildirimleri aç metni
"dm_notification_disable": "//button[text()='Şimdi Değil']", # Bildirimleri kapat düğmesi
"dm_send_button": "//div[@role='button' and text()='Gönder']", # Mesaj gönder düğmesi
"dm_error_present": "//*[contains(text(), 'IGD mesaj gönderme hatası simgesi')]", # Mesaj gönderme hatası simgesi
"2f_screen_present": "//input[@aria-describedby='verificationCodeDescription' and @aria-label='Güvenlik Kodu']", # İki faktörlü kimlik doğrulama ekranı
"2f_entering_error": "//p[@id='twoFactorErrorAlert' and @role='alert']", # İki faktörlü kimlik doğrulama hatası metni
"check_dm_message_sent_to_user": "//div[@role='none']//div[@dir='auto' and @role='none']", # Kullanıcıya gönderilen mesajın doğrulaması
"login_error": "//p[@id='slfErrorAlert']" # Giriş hatası metni
}
# Özel hata sınıfları tanımlamaları
class WaitAndClickException(Exception):
pass
class WaitException(Exception):
pass
# Kullanıcı sınıfı
class User:
def __init__(self, username, password, token=False, debug=False, use_chrome=False, Name="Acc"):
# Kullanıcı bilgileri ve seçenekleri
self.username = username
self.password = password
self.two_factor_token = token
self.name = Name
self.cookies_dict = None
self.is_logged = False
self.debug = debug
self.logger = self.__initialize_log() # Loglama ayarlarının başlatılması
self.driver = None
self.wait = None
self.use_chrome = use_chrome
# Loglama ayarlarını yapılandıran metot
def __initialize_log(self):
logger = logging.getLogger(f"instagram_web {self.name} - {self.username}")
if self.debug:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setLevel(logging.DEBUG)
file_handler = logging.FileHandler(f'log/{self.name} - {self.username}.log')
file_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
logger.addHandler(file_handler)
return logger
# WebDriver'ı başlatan metot
def __initialize_driver(self, use_chrome=False):
try:
self.logger.debug('initialize_driver() çağrıldı, hata ayıklama: %s', self.debug)
if use_chrome:
self.driver, self.wait = initialize_chrome_driver()
else:
self.driver, self.wait = initialize_driver()
self.__accept_cookie(self.driver)
self.logger.debug("Sürücü başlatıldı")
except:
self.logger.exception("__initialize_driver")
# İki faktörlü kimlik doğrulama kodunu üreten metot
def __generate_2factor_code(self, token):
totp = pyotp.TOTP(token)
current_time = time.time()
time_step = 30 # TOTP zaman adımı, genellikle 30 saniye
remaining_time = time_step - (current_time % time_step)
# Kodun geçerli olduğu süre 5 saniyeden azsa, bir sonraki kodu bekleyin
if remaining_time < 4:
time.sleep(remaining_time)
new_code = totp.now()
return new_code
# Çerezleri kabul eden metot
def __accept_cookie(self, driver):
self.logger.debug(f"__accept_cookie() çağrıldı, sürücü: {driver}")
try:
driver.get('https://instagram.com/')
try:
cookie_elem = self.wait.until(EC.presence_of_element_located((By.XPATH, LOCATORS["cookie_text"])))
except:
self.logger.warning("Kabul edilecek çerez yok!")
return True
if cookie_elem:
driver.find_element("xpath", LOCATORS["cookie_accept"]).click()
self.logger.debug("Çerez kabul edildi")
sleep(2)
except:
self.logger.exception("__accept_cookie")
# İki faktörlü kimlik doğrulama işlemini gerçekleştiren metot
def __two_factor(self):
try:
url = self.driver.current_url
self.logger.debug("5 saniye içinde 2f ekranını kontrol ediyor")
sleep(5)
two_factor_token_field = self.wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS['2f_screen_present'])))
two_factor_token_field.send_keys(self.__generate_2factor_code(self.two_factor_token))
two_factor_token_field.send_keys(Keys.RETURN)
if self.__is_element_present(LOCATORS['2f_entering_error'], 3):
self.logger.error("2f_entering_error")
sleep(5)
two_factor_token_field.send_keys(Keys.CONTROL, "a") # Eski kodu sil
two_factor_token_field.send_keys(Keys.DELETE)
two_factor_token_field.send_keys(self.__generate_2factor_code(self.two_factor_token))
two_factor_token_field.send_keys(Keys.RETURN)
if self.__is_element_present(LOCATORS['2f_entering_error'], 3):
self.logger.error("İkinci 2f_entering_error, giriş yapılamıyor")
return "Giriş yapılamıyor"
for i in range(10):
if self.driver.current_url != url:
return True
self.logger.debug("Kod girildikten sonra 5 saniye boyunca URL değişmedi")
sleep(1)
return "Giriş yapılamıyor"
except:
self.logger.exception("two_factor() hatası")
return "Giriş yapılamıyor"
# Kullanıcı girişi işlemini gerçekleştiren metot
def login(self):
self.__initialize_driver(use_chrome=self.use_chrome)
self.logger.debug(f"login() çağrıldı, kullanıcı adı: {self.username}, şifre: {self.password}")
try:
self.driver.get('https://instagram.com/accounts/login')
username_field = self.wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["login_username_field"])))
password_field = self.driver.find_element("xpath", LOCATORS["login_password_field"])
url = self.driver.current_url
username_field.send_keys(self.username)
password_field.send_keys(self.password)
password_field.send_keys(Keys.RETURN)
print("aaaaaaa")
if self.__is_element_present(LOCATORS['login_error'], 5):
self.logger.error(f"Instagram'a giriş yaparken bir sorun oluştu. Lütfen tekrar deneyin.")
print("ddddddd")
return "Instagram'a giriş yaparken bir sorun oluştu. Lütfen tekrar deneyin."
print("aaaadawdawdawaaaa")
# Instagram'ın giriş yapmasını bekleyin
while True:
if self.driver.current_url != url:
self.logger.debug(f"İki faktörlü kimlik doğrulama kodu: {self.two_factor_token}")
if self.two_factor_token:
self.logger.debug("İki faktörlü kimlik doğrulama etkin, giriş yapmaya çalışılıyor")
self.__two_factor()
self.driver.get("https://www.instagram.com/direct/inbox")
self.wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["dm_notification_present"])))
notifications = self.driver.find_element("xpath", LOCATORS["dm_notification_disable"])
notifications.click()
self.logger.info(f"Giriş başarılı yapıldı")
self.is_logged = True
return True
sleep(1)
except:
self.logger.exception("login() hatası")
# Belirtilen kullanıcının önerilerini almak için kullanılan metot
def get_suggestions(self, username):
def get_all_usernames():
usernames = set()
repeated_count = 0
threshold = 5 # Gerektiğinde eşik değerini ayarlayın
while True:
elements = self.wait.until(
EC.presence_of_all_elements_located((By.XPATH, '//a//div//span//div')))
old_len = len(usernames)
for el in elements:
try:
# Öğenin metnini al
text = el.text
# Kullanıcı adı "Doğrulandı" içeriyorsa atla
if 'Doğrulandı' in text:
continue
# Kullanıcı adını küme içine ekle
usernames.add(text)
except StaleElementReferenceException:
continue
if old_len == len(usernames):
repeated_count += 1
if repeated_count >= threshold:
break
else:
repeated_count = 0
# Daha fazla kullanıcı adını yüklemek için aşağı kaydır
self.driver.execute_script("arguments[0].scrollIntoView();", elements[-1])
time.sleep(1) # Gerektiğinde uyku süresini ayarlayın
return usernames
# Kullanıcının önerilerini almak için kullanılan metot
LOCATORS_SUGGESTIONS = {
'page_unavailable': "//span[text()='Üzgünüz, bu sayfa mevcut değil.']", # Sayfa mevcut değil hatası metni
'suggest_button': "//div[@role='button']//div//*[local-name() = 'svg']", # Önerileri göster düğmesi
'see_all_button': "//a[@role='link']//span[@dir='auto' and text()='Tümünü Gör']", # Tümünü gör düğmesi
'similar_acc_presence': "//div[text()='Sizin İçin Öneriliyor']", # Sizin için öneriliyor metni
'err_unable_to_load': "//div[text()='Öneriler yüklenemedi.']", # Öneriler yüklenemedi hatası metni
}
try:
self.logger.debug(f"get_suggestions() çağrıldı, kullanıcı adı: {username}")
self.driver.get(f'https://www.instagram.com/{username}')
try:
self.__wait_and_click(LOCATORS_SUGGESTIONS['suggest_button'])
except WaitAndClickException:
if self.__is_element_present(LOCATORS_SUGGESTIONS["page_unavailable"]):
self.logger.debug("Profil mevcut değil")
return "Profil mevcut değil"
self.logger.exception("AJAJ")
if self.__is_element_present(LOCATORS_SUGGESTIONS['err_unable_to_load'], 3):
return 'Hesap kilitli'
self.__wait_and_click(LOCATORS_SUGGESTIONS['see_all_button'])
self.__wait(LOCATORS_SUGGESTIONS['similar_acc_presence'])
return get_all_usernames()
except WaitAndClickException as e:
self.logger.error(f"get_suggestions() - İşlem durduruldu. Hata: {str(e)}")
return False
except WaitException as e:
self.logger.error(f"get_suggestions() - İşlem durduruldu. Hata: {str(e)}")
return False
# WebDriver'ı kapatmak için kullanılan metot
def __exit_driver(self):
self.driver.quit()
# Çerezleri almak için kullanılan metot
def get_cookies(self, close_after=True):
def transform_cookies(cookies):
headers = {}
cookie_str = ""
for cookie in cookies:
cookie_str += cookie['name'] + "=" + cookie['value'] + "; "
headers["Cookie"] = cookie_str[:-2] # Son noktalı virgül ve boşluğu kaldırın
headers["X-Ig-App-Id"] = "936619743392459" # <---------------------
return headers
try:
self.login()
cookies = self.driver.get_cookies()
cookies_dict = transform_cookies(cookies)
if close_after:
self.__exit_driver()
self.cookies_dict = cookies_dict
return cookies_dict
except:
self.logger.exception("get_cookies")
# Belirtilen XPath ile öğenin varlığını kontrol eden metot
def __is_element_present(self, xpath, time_to_wait=0):
wait = WebDriverWait(self.driver, time_to_wait)
self.logger.debug(f"__is_element_present() çağrıldı, xpath: {xpath} bekleme süresi: {time_to_wait}")
try:
wait.until(EC.presence_of_element_located((By.XPATH, xpath)))
self.logger.debug("__is_element_present() True dönüyor")
return True
except:
self.logger.debug("__is_element_present() False dönüyor")
return False
# Belirtilen XPath ile öğenin tıklanabilir olmasını bekleyen ve tıklayan metot
def __wait_and_click(self, xpath, time=5):
self.logger.debug(f'__wait_and_click() çağrıldı, xpath: {xpath}, süre: {time}')
try:
button = WebDriverWait(self.driver, time).until(EC.element_to_be_clickable((By.XPATH, xpath)))
button.click()
self.logger.debug(f"XPath {xpath} ile öğeye tıklandı")
except Exception as e:
self.logger.debug(f"XPath {xpath} ile öğeye tıklanamadı. Hata: {str(e)}")
raise WaitAndClickException(f"Öğeye tıklanamadığı için işlem durduruldu: {xpath}") from e
# Belirtilen XPath ile öğenin varlığını bekleyen ve öğeyi döndüren metot
def __wait(self, xpath, time=5):
self.logger.debug(f'__wait() çağrıldı, xpath: {xpath}, süre: {time}')
try:
return WebDriverWait(self.driver, time).until(EC.presence_of_element_located((By.XPATH, xpath)))
except Exception as e:
self.logger.debug(f"XPath {xpath} ile öğenin beklenmesinde hata. Hata: {str(e)}")
raise WaitException(f"Öğe beklenemediği için işlem durduruldu: {xpath}") from e
# Belirtilen kullanıcıya mesaj göndermek için kullanılan metot
def send_msg(self, to_username, msg, check_dm_message=False):
def check_dm_message_sent_to_user():
return self.__is_element_present(LOCATORS['check_dm_message_sent_to_user'], 2)
def check_if_freezed():
return self.__is_element_present(LOCATORS["dm_error_present"], 3)
try:
self.logger.debug(f"send_msg() çağrıldı, parametreler: to_username: {to_username}, msg: {msg}")
if not self.is_logged:
self.logger.debug("Hesap oturum açmamış. Giriş yapmaya çalışılıyor")
self.login()
if not self.__is_element_present(LOCATORS["new_dm_btn"], 0):
self.driver.get("https://www.instagram.com/direct/")
wait = WebDriverWait(self.driver, 5)
new_dm_btn = wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["new_dm_btn"]))) # Yeni mesaj düğmesi
try:
new_dm_btn.click()
except:
self.driver.get("https://www.instagram.com/direct/")
new_dm_btn = wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["new_dm_btn"]))) # Yeni mesaj düğmesi
new_dm_btn.click()
search_user_field = wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["dm_type_username"])))
search_user_field.send_keys(to_username)
wait = WebDriverWait(self.driver, 10)
username_path = LOCATORS["dm_select_user"].format(to_username)
try:
username_element = wait.until(EC.presence_of_element_located((By.XPATH, username_path)))
username_element.click()
except:
self.logger.error("Listeden kullanıcı seçilemedi")
return "Hesap bulunamadı."
next_btn = wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["dm_start_chat_btn"])))
last_url = self.driver.current_url
self.driver.execute_script("arguments[0].click();", next_btn)
tries = 0
changed = False
while tries < 50:
sleep(0.5)
if last_url == self.driver.current_url:
tries += 1
continue
else:
changed = True
break
if changed == False:
self.logger.error("Yeni URL'ye erişilemiyor")
return "URL sorunu"
msg_field = wait.until(
EC.presence_of_element_located((By.XPATH, LOCATORS["dm_msg_field"])))
if check_dm_message:
if check_dm_message_sent_to_user():
return 'zaten gönderildi'
action = ActionChains(self.driver)
action.move_to_element(msg_field)
action.click()
action.send_keys(msg)
action.perform()
send_btn = self.driver.find_element("xpath", LOCATORS["dm_send_button"])
send_btn.click()
if check_if_freezed() == True:
self.logger.warn("Hesap dondu")
return "donma"
return "gönderildi"
except Exception as e:
self.logger.error(f"Mesaj gönderme hatası: {str(e)}")
return False
# Mesaj gönderme işlemini kontrol eden ve gerekirse yeniden deneyen metot
def send_message_and_retry(self, to_username, msg):
result = self.send_msg(to_username, msg, check_dm_message=True)
retry = 0
while result == "donma":
self.logger.debug(f"Donma sorunu nedeniyle mesaj yeniden gönderiliyor. Deneme: {retry}")
retry += 1
if retry > 2:
self.logger.error("Mesaj donma sorunu nedeniyle 3 denemede başarısız oldu")
break
self.driver.get("https://www.instagram.com/direct/inbox")
result = self.send_msg(to_username, msg, check_dm_message=True)
if result == 'gönderildi':
self.logger.info(f"Mesaj başarıyla gönderildi: '{msg}'")
return True
elif result == 'zaten gönderildi':
self.logger.info(f"Mesaj zaten bu kullanıcıya daha önce gönderildi: '{msg}'")
return True
else:
self.logger.error(f"Mesaj gönderme işlemi başarısız oldu: '{msg}'")
return False
# Kullanıcı sınıfını temizlemek için kullanılan metot
def cleanup(self):
self.__exit_driver()
self.logger.info("Kullanıcı örneği kapatıldı")
# Kullanıcı nesnesinin oluşturulması ve kullanılması
if __name__ == "__main__":
user = User("kullanici_adi", "sifre", token="iki_faktor_token", debug=True)
user.get_cookies()
suggestions = user.get_suggestions("hedef_kullanici")
if suggestions:
for suggestion in suggestions:
message = "Merhaba, sizi takip etmek istiyorum!"
user.send_message_and_retry(suggestion, message)
user.cleanup() Python Selenium ile Dm Bot ( Ücretsiz )
1
●240
- 11-08-2023, 19:47:19
