Merhabalar arkadaşlar , Daha önce kullandığım pyhon ile yazılmış Google Analytics e tamı tamına yansıyan kodlarımı veriyorum, artık kullanmıyorum requestte gectiğim için.
import asyncio
from playwright.async_api import async_playwright
import random
import time
import logging
import argparse
import os
import sys
import subprocess
import json
from datetime import datetime
from urllib.parse import urlparse
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f"analytics_traffic_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log", encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger()
class ProxyManager:
def __init__(self, proxy_file):
"""
Args:
proxy_file (str): Proxy dosyasının yolu
"""
self.proxy_file = proxy_file
self.proxies = []
self.working_proxies = []
self.load_proxies()
if not self.proxies:
logger.error(f"Proxy bulunamadi! Lutfen {proxy_file} dosyasini kontrol edin.")
raise FileNotFoundError(f"Proxy dosyası bulunamadı veya boş: {proxy_file}")
def load_proxies(self):
"""Dosyadan proxy'leri yükler"""
try:
with open(self.proxy_file, 'r') as file:
for line in file:
line = line.strip()
if line:
try:
parts = line.split(':')
if len(parts) >= 4:
host = parts[0]
port = parts[1]
username = parts[2]
password = ':'.join(parts[3:])
self.proxies.append({
"server": f"http://{host}:{port}",
"username": username,
"password": password,
"original": line
})
else:
logger.warning(f"Gecersiz proxy formati: {line}")
except Exception as e:
logger.error(f"Proxy ayristirma hatasi: {e} - Satir: {line}")
logger.info(f"Toplam {len(self.proxies)} proxy okundu.")
self.working_proxies = self.proxies
except Exception as e:
logger.error(f"Dosya okuma hatasi: {e}")
async def test_proxies(self, max_test=10):
"""
"""
logger.info("Proxy testi atlanıyor, tüm proxy'lerin çalıştığı varsayılıyor...")
# Tüm proxy'lerin çalıştığını varsay
self.working_proxies = self.proxies
def get_random_proxy(self):
"""Rastgele bir proxy döndürür"""
if not self.proxies:
return None
return random.choice(self.proxies)
def get_proxies_for_sessions(self, num_sessions):
if len(self.proxies) < num_sessions:
# Proxy sayısı yetersizse, tekrar kullan
return [self.proxies[i % len(self.proxies)] for i in range(num_sessions)]
else:
# Rastgele num_sessions kadar proxy seç
return random.sample(self.proxies, num_sessions)
def mask_proxy_url(self, proxy_info):
if not proxy_info:
return "Proxy yok"
try:
server = proxy_info["server"]
username = proxy_info["username"]
masked_username = username[:3] + '*' * (len(username) - 3) if len(username) > 3 else username
parsed = urlparse(server)
host_port = f"{parsed.netloc}"
return f"{parsed.scheme}://{masked_username}:******@{host_port}"
except:
return str(proxy_info)
class UserAgentManager:
def __init__(self):
"""User-Agent yönetim sınıfı""
self.user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.101 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:90.0) Gecko/20100101 Firefox/90.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:90.0) Gecko/20100101 Firefox/90.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36 Edg/92.0.902.55'
]
try:
from fake_useragent import UserAgent
self.ua = UserAgent()
logger.info("Fake UserAgent basariyla baslatildi.")
self.has_fake_ua = True
except Exception as e:
logger.warning(f"Fake UserAgent baslatma hatasi: {e}")
logger.warning("Onceden tanimlanmis user agent'lar kullanilacak.")
self.has_fake_ua = False
def get_random_user_agent(self):
"""Rastgele bir User-Agent döndürür"""
if hasattr(self, 'has_fake_ua') and self.has_fake_ua:
try:
return self.ua.random
except:
pass
return random.choice(self.user_agents)
class PlaywrightManager:
def __init__(self, target_urls, proxy_file, num_sessions=2000, visit_duration=60, page_interaction=True, browser_type="all"):
"""
Args:
target_urls (list): Ziyaret edilecek URL'ler listesi
proxy_file (str): Proxy dosyasının yolu
num_sessions (int): Oturum sayısı
visit_duration (int): Her sayfada kalma süresi (saniye)
page_interaction (bool): Sayfa ile etkileşim kurulsun mu?
browser_type (str): Kullanılacak tarayıcı tipi ("all", "chromium", "firefox", "webkit")
"""
self.target_urls = target_urls if isinstance(target_urls, list) else [target_urls]
self.proxy_manager = ProxyManager(proxy_file)
self.ua_manager = UserAgentManager()
self.num_sessions = num_sessions
self.visit_duration = visit_duration
self.page_interaction = page_interaction
self.browser_type = browser_type
self.session_stats = {i: {"success": 0, "failed": 0} for i in range(num_sessions)}
self.total_visits = 0
self.start_time = None
self.browsers = []
self.tasks = []
async def check_playwright_browsers(self):
try:
# Tarayıcıların kurulu olup olmadığını kontrol et
result = subprocess.run(
["playwright", "install", "--dry-run"],
capture_output=True,
text=True
)
if "Browsers are already installed" not in result.stdout:
logger.warning("Playwright tarayıcıları kurulu değil!")
logger.info("Tarayıcılar indiriliyor... Bu işlem birkaç dakika sürebilir.")
install_process = subprocess.run(
["playwright", "install"],
capture_output=True,
text=True
)
if install_process.returncode == 0:
logger.info("Tarayıcılar başarıyla kuruldu!")
else:
logger.error("Tarayıcı kurulumu başarısız oldu!")
logger.error(install_process.stderr)
logger.info("Lütfen manuel olarak 'playwright install' komutunu çalıştırın.")
sys.exit(1)
else:
logger.info("Playwright tarayıcıları zaten kurulu.")
except Exception as e:
logger.error(f"Tarayıcı kontrolü sırasında hata: {e}")
logger.info("Lütfen manuel olarak 'playwright install' komutunu çalıştırın.")
sys.exit(1)
async def setup_browser(self, playwright, proxy_info, user_agent, session_id):
"""
Args:
playwright: Playwright nesnesi
proxy_info (dict): Proxy bilgileri
user_agent (str): User-Agent
session_id (int): Oturum ID'si
Returns:
tuple: (browser, context) tarayıcı ve bağlam nesneleri
"""
try:
# Tarayıcı tipini seç
if self.browser_type == "all":
browser_types = ["chromium", "firefox", "webkit"]
browser_weights = [0.7, 0.2, 0.1] # Chromium daha yaygın
browser_type = random.choices(browser_types, weights=browser_weights, k=1)[0]
else:
browser_type = self.browser_type
l
if browser_type == "chromium":
browser_obj = playwright.chromium
elif browser_type == "firefox":
browser_obj = playwright.firefox
else:
browser_obj = playwright.webkit
browser = await browser_obj.launch(
headless=True, # Görünmez modda çalıştır
slow_mo=50, # İşlemleri yavaşlat (ms)
args=[
'--disable-extensions',
'--disable-gpu',
'--disable-dev-shm-usage',
'--no-sandbox',
'--disable-web-security'
]
)
context = await browser.new_context(
proxy={
"server": proxy_info["server"],
"username": proxy_info["username"],
"password": proxy_info["password"]
},
user_agent=user_agent,
viewport={"width": 1366, "height": 768},
locale="tr-TR", # Türkçe dil ayarı
timezone_id="Europe/Istanbul", # Türkiye zaman dilimi
geolocation={"latitude": 31.995408, "longitude": 36.544443}, # İstanbul konumu
permissions=["geolocation", "notifications"] # İzinler
)
await context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => false,
});
""")
logger.info(f"Oturum #{session_id+1} - {browser_type.capitalize()} tarayıcı başlatıldı")
return browser, context
except Exception as e:
logger.error(f"Oturum #{session_id+1} - Tarayıcı başlatma hatası: {e}")
return None, None
async def interact_with_page(self, page):
"""
Sayfa ile etkileşim kurar (scroll, tıklama vb.)
Args:
page: Playwright page nesnesi
"""
try:
for _ in range(random.randint(2, 5)):
await page.evaluate("window.scrollBy(0, window.innerHeight / 2)")
await asyncio.sleep(random.uniform(1, 2))
await page.evaluate("window.scrollTo(0, 0)")
await asyncio.sleep(random.uniform(0.5, 1))
try:
links = await page.query_selector_all('a[href^="/"], a[href^="http"]')
if links and random.random() < 0.3: # %30 ihtimalle bir linke tıkla
random_link = random.choice(links)
try:
async with page.expect_navigation(timeout=10000, wait_until="domcontentloaded"):
await random_link.click()
await asyncio.sleep(random.uniform(3, 5))
await page.go_back()
await asyncio.sleep(random.uniform(1, 2))
except Exception as e:
logger.debug(f"Link tıklama hatası: {e}")
except Exception as e:
logger.debug(f"Link seçme hatası: {e}")
except Exception as e:
logger.debug(f"Sayfa etkileşim hatası: {e}")
async def custom_page_load_strategy(self, page, url):
"""
Args:
page: Playwright page nesnesi
url (str): Ziyaret edilecek URL
Returns:
bool: Başarılı ise True, değilse False
"""
try:
# Temel navigasyon - minimum bekleme ile
await page.goto(url, wait_until="commit", timeout=20000)
for _ in range(30): # Maksimum 30 saniye bekle
# Sayfa hazır mı kontrol et
is_ready = await page.evaluate("""() => {
return document.readyState === 'complete' || document.readyState === 'interactive';
}""")
if is_ready:
has_content = await page.evaluate("""() => {
return !!document.querySelector('body') &&
document.body.innerHTML.length > 500;
}""")
if has_content:
return True
# 1 saniye bekle ve tekrar dene
await asyncio.sleep(1)
return False
except Exception as e:
logger.error(f"Özel yükleme stratejisi hatası: {e}")
return False
async def visit_url(self, context, url, session_id):
"""
Belirtilen URL'yi ziyaret eder - İyileştirilmiş sürüm
Args:
context: Playwright context nesnesi
url (str): Ziyaret edilecek URL
session_id (int): Oturum ID'si
Returns:
bool: Başarılı ise True, değilse False
"""
try:
# Yeni sayfa oluştur
page = await context.new_page()
start_time = time.time()
i
try:
response = await page.goto(
url,
wait_until="domcontentloaded",
timeout=40000
)
except Exception as e:
logger.warning(f"Oturum #{session_id+1} - domcontentloaded ile yükleme başarısız: {e}")
# Başarısız olursa, commit ile dene
try:
response = await page.goto(
url,
wait_until="commit",
timeout=20000
)
except Exception as e2:
logger.error(f"Oturum #{session_id+1} - Sayfa yükleme tamamen başarısız: {e2}")
self.session_stats[session_id]["failed"] += 1
await page.close()
return False
# Yükleme süresini hesapla
load_time = time.time() - start_time
if not response:
logger.warning(f"Oturum #{session_id+1} - Yanıt alınamadı: {url}")
self.session_stats[session_id]["failed"] += 1
await page.close()
return False
status = response.status
if 200 <= status < 400:
# Sayfa başlığını al
title = await page.title()
logger.info(f"Oturum #{session_id+1} - Başarılı ziyaret - {url} - Durum: {status} - Yükleme: {load_time:.2f}s - Başlık: {title}")
# Sayfanın yüklenmesi için ek bekleme
try:
# Sayfa içeriğinin görünür olmasını bekle
await page.wait_for_selector('body', timeout=10000)
# Sayfanın stabil olması için biraz bekle
await asyncio.sleep(3)
except Exception as e:
logger.warning(f"Oturum #{session_id+1} - Sayfa içeriği bekleme hatası: {e}")
await asyncio.sleep(3)
if self.page_interaction:
await self.interact_with_page(page)
await asyncio.sleep(self.visit_duration)
try:
has_analytics = await page.evaluate("""() => {
return !!(
window.ga ||
window.gtag ||
window.dataLayer ||
window.google_tag_manager ||
document.querySelector('script[src*="google-analytics.com"]') ||
document.querySelector('script[src*="googletagmanager.com"]')
);
}""", timeout=5000)
if has_analytics:
logger.info(f"Oturum #{session_id+1} - Google Analytics tespit edildi: {url}")
except Exception as e:
logger.debug(f"Oturum #{session_id+1} - Analytics kontrolü hatası: {e}")
self.session_stats[session_id]["success"] += 1
await page.close()
return True
else:
logger.warning(f"Oturum #{session_id+1} - Başarısız ziyaret - {url} - Durum: {status}")
self.session_stats[session_id]["failed"] += 1
await page.close()
return False
except Exception as e:
logger.error(f"Oturum #{session_id+1} - Ziyaret hatası - {url}: {e}")
self.session_stats[session_id]["failed"] += 1
return False
async def run_session(self, playwright, session_id):
"""
Bir oturumu çalıştırır
Args:
playwright: Playwright nesnesi
session_id (int): Oturum ID'si
"""
# Proxy ve User-Agent seç
proxy_info = self.proxy_manager.get_random_proxy()
user_agent = self.ua_manager.get_random_user_agent()
if not proxy_info:
logger.error(f"Oturum #{session_id+1} - Proxy bulunamadı!")
return
masked_proxy = self.proxy_manager.mask_proxy_url(proxy_info)
logger.info(f"Oturum #{session_id+1} - Proxy: {masked_proxy}")
browser, context = await self.setup_browser(playwright, proxy_info, user_agent, session_id)
if not browser or not context:
return
self.browsers.append((browser, context))
try:
for url in self.target_urls:
success = await self.visit_url(context, url, session_id)
self.total_visits += 1
if success and len(self.target_urls) > 1:
await asyncio.sleep(random.uniform(3, 8))
except Exception as e:
logger.error(f"Oturum #{session_id+1} - Çalışma hatası: {e}")
finally:
try:
await context.close()
await browser.close()
logger.info(f"Oturum #{session_id+1} - Tarayıcı kapatıldı")
except:
pass
async def run_all_sessions(self):
"""Tüm oturumları çalıştırır"""
self.start_time = time.time()
# Tarayıcıların kurulu olup olmadığını kontrol et
await self.check_playwright_browsers()
# Proxy test etme adımını atla - tüm proxy'lerin çalıştığını varsay
logger.info("Proxy testi atlanıyor, tüm proxy'lerin çalıştığı varsayılıyor...")
async with async_playwright() as playwright:
# Tüm oturumları başlat
self.tasks = []
for i in range(self.num_sessions):
# Oturumlar arasında kısa bir gecikme ekle (aynı anda çok fazla tarayıcı başlatmamak için)
await asyncio.sleep(random.uniform(1, 3))
task = asyncio.create_task(self.run_session(playwright, i))
self.tasks.append(task)
# Tüm görevlerin tamamlanmasını bekle
await asyncio.gather(*self.tasks, return_exceptions=True)
# İstatistikleri göster
self.show_stats()
def show_stats(self):
"""İstatistikleri gösterir"""
end_time = time.time()
elapsed_time = end_time - self.start_time
logger.info("\n--- İstatistikler ---")
logger.info(f"Toplam çalışma süresi: {elapsed_time:.2f} saniye")
logger.info(f"Toplam ziyaret sayısı: {self.total_visits}")
total_success = sum(stats["success"] for stats in self.session_stats.values())
total_failed = sum(stats["failed"] for stats in self.session_stats.values())
if self.total_visits > 0:
success_rate = total_success / self.total_visits * 100
logger.info(f"Başarılı ziyaretler: {total_success} ({success_rate:.2f}%)")
logger.info(f"Başarısız ziyaretler: {total_failed} ({100 - success_rate:.2f}%)")
# Oturum bazında istatistikleri göster
logger.info("\nOturum bazında istatistikler:")
for session_id, stats in self.session_stats.items():
total = stats["success"] + stats["failed"]
success_rate = stats["success"] / total * 100 if total > 0 else 0
logger.info(f"Oturum #{session_id+1} - Başarılı: {stats['success']}, Başarısız: {stats['failed']}, Başarı oranı: {success_rate:.2f}%")
async def save_results_to_csv(self, filename="analytics_results.csv"):
"""Sonuçları CSV dosyasına kaydeder"""
import csv
try:
with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['session_id', 'success', 'failed', 'success_rate']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for session_id, stats in self.session_stats.items():
total = stats["success"] + stats["failed"]
success_rate = stats["success"] / total * 100 if total > 0 else 0
writer.writerow({
'session_id': session_id + 1,
'success': stats["success"],
'failed': stats["failed"],
'success_rate': f"{success_rate:.2f}%"
})
logger.info(f"Sonuçlar {filename} dosyasına kaydedildi.")
except Exception as e:
logger.error(f"CSV kaydetme hatası: {e}")
async def main():
# Argüman ayrıştırıcı
parser = argparse.ArgumentParser(description='Playwright ile web sitesi ziyareti ve Google Analytics\'e veri gönderme')
parser.add_argument('--urls', type=str, help='Ziyaret edilecek URL\'ler (virgülle ayrılmış)')
parser.add_argument('--proxy-file', type=str, default='proxy.txt', help='Proxy dosyası (varsayılan: proxy.txt)')
parser.add_argument('--sessions', type=int, default=20000, help='Oturum sayısı (varsayılan: 5)')
parser.add_argument('--duration', type=int, default=30, help='Her sayfada kalma süresi saniye (varsayılan: 30)')
parser.add_argument('--no-interaction', action='store_true', help='Sayfa ile etkileşim kurma')
parser.add_argument('--browser', type=str, default='chromium', choices=['all', 'chromium', 'firefox', 'webkit'],
help='Kullanılacak tarayıcı (varsayılan: chromium)')
parser.add_argument('--csv', type=str, help='Sonuçları CSV dosyasına kaydet')
args = parser.parse_args()
# URL'leri al
if args.urls:
target_urls = [url.strip() for url in args.urls.split(',')]
else:
target_urls = input("Ziyaret edilecek URL'leri girin (virgülle ayırarak): ").split(',')
target_urls = [url.strip() for url in target_urls]
# URL formatını kontrol et
for i, url in enumerate(target_urls):
if not url.startswith(('http://', 'https://')):
target_urls[i] = 'https://' + url
# Proxy dosyasını kontrol et
proxy_file = args.proxy_file
if not os.path.exists(proxy_file):
logger.error(f"{proxy_file} dosyası bulunamadı!")
# Örnek proxy.txt oluştur
logger.info("Örnek proxy.txt dosyası oluşturuluyor...")
with open(proxy_file, 'w') as f:
f.write("rotating2.we1.town:1000005:5112e6eee4b8e8d28ac3b8__cr.tr:e4f98ea6ff6763a0c\n")
logger.info(f"{proxy_file} dosyası oluşturuldu.")
# Ayarlar
num_sessions = args.sessions
visit_duration = args.duration
page_interaction = not args.no_interaction
browser_type = args.browser
# Uyarı mesajı
logger.warning("\n" + "!" * 80)
logger.warning("UYARI: Bu script web sitelerine gerçek tarayıcı trafiği gönderecektir!")
logger.warning(f"- Hedef URL'ler: {', '.join(target_urls)}")
logger.warning(f"- {num_sessions} oturum aynı anda çalışacak")
logger.warning(f"- Her sayfa ziyareti yaklaşık {visit_duration} saniye sürecek")
logger.warning(f"- Sayfa etkileşimi: {'Açık' if page_interaction else 'Kapalı'}")
logger.warning(f"- Tarayıcı tipi: {browser_type}")
logger.warning("Bu, hedef sitelere yük bindirebilir ve IP adresinizin engellenmesine neden olabilir.")
logger.warning("Sadece izin verilen sitelerde ve test amaçlı kullanın!")
logger.warning("!" * 80 + "\n")
# Kullanıcı onayı
try:
input("Devam etmek için Enter tuşuna basın (İptal etmek için Ctrl+C)...")
except KeyboardInterrupt:
logger.info("İşlem iptal edildi.")
return
# Playwright ile web sitesi ziyareti
manager = PlaywrightManager(
target_urls=target_urls,
proxy_file=proxy_file,
num_sessions=num_sessions,
visit_duration=visit_duration,
page_interaction=page_interaction,
browser_type=browser_type
)
# Tüm oturumları çalıştır
await manager.run_all_sessions()
# Sonuçları CSV'ye kaydet
if args.csv:
await manager.save_results_to_csv(args.csv)
if __name__ == "__main__":
# Ana fonksiyonu çalıştır
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Program kullanıcı tarafından durduruldu.")
except Exception as e:
logger.error(f"Beklenmeyen hata: {e}")