salvacybersec/Youtube2Feed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

yt bot protection

Browse Source
This commit is contained in:
salvacybersec
2025-11-13 05:31:43 +03:00
parent 01f9cfc8b2
commit 86ce4913b8
2 changed files with 128 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
src/transcript_extractor.py
View File

@@ -5,35 +5,126 @@ from youtube_transcript_api import YouTubeTranscriptApi
from typing import List, Dict, Optional
import time
import logging
import random
import os
# Logger oluştur
logger = logging.getLogger(__name__)
# Gerçek tarayıcı User-Agent'ları (rotasyon için)
USER_AGENTS = [
# Chrome (Windows)
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
# Chrome (macOS)
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
# Chrome (Linux)
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
# Firefox (Windows)
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Firefox (macOS)
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0',
# Safari (macOS)
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15',
# Edge (Windows)
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
]
class TranscriptExtractor:
"""YouTube transcript çıkarıcı sınıfı"""
def __init__(self, rate_limit: int = 2, time_window: int = 30):
def __init__(self, rate_limit: int = 1, time_window: int = 60):
"""
Args:
rate_limit: Zaman penceresi başına maksimum istek sayısı (YouTube IP blocking'i önlemek için düşük)
time_window: Zaman penceresi (saniye)
rate_limit: Zaman penceresi başına maksimum istek sayısı (YouTube IP blocking'i önlemek için çok düşük)
time_window: Zaman penceresi (saniye) - daha uzun süre
"""
self.rate_limit = rate_limit
self.rate_limit = rate_limit # 1 istek/60 saniye (çok konservatif)
self.time_window = time_window
self.request_times = []
self.last_blocked_time = 0
self.block_count = 0 # Toplam blocking sayısı
# Gerçek tarayıcı header'larını ayarla
self._setup_browser_headers()
def _setup_browser_headers(self):
"""Gerçek tarayıcı gibi HTTP header'larını ayarla"""
try:
import requests
# requests.Session'ın get/post metodlarını patch et
# Bu, youtube-transcript-api'nin yaptığı tüm isteklere header ekler
original_get = requests.Session.get
original_post = requests.Session.post
def patched_get(self, url, **kwargs):
"""requests.Session.get'i patch et - header'ları ekle"""
headers = kwargs.get('headers', {})
browser_headers = TranscriptExtractor._get_browser_headers()
# Mevcut header'ları koru, browser header'larını ekle
merged_headers = {**browser_headers, **headers}
kwargs['headers'] = merged_headers
return original_get(self, url, **kwargs)
def patched_post(self, url, **kwargs):
"""requests.Session.post'i patch et - header'ları ekle"""
headers = kwargs.get('headers', {})
browser_headers = TranscriptExtractor._get_browser_headers()
# Mevcut header'ları koru, browser header'larını ekle
merged_headers = {**browser_headers, **headers}
kwargs['headers'] = merged_headers
return original_post(self, url, **kwargs)
# Patch'i uygula (sadece bir kez)
if not hasattr(requests.Session, '_browser_headers_patched'):
requests.Session.get = patched_get
requests.Session.post = patched_post
requests.Session._browser_headers_patched = True
logger.info("[HEADERS] ✅ requests.Session patch edildi - Gerçek tarayıcı header'ları eklendi")
except Exception as e:
logger.warning(f"[HEADERS] ⚠️ Header patch edilemedi: {e}")
@staticmethod
def _get_browser_headers() -> Dict[str, str]:
"""Gerçek tarayıcı header'larını döndür"""
user_agent = random.choice(USER_AGENTS)
headers = {
'User-Agent': user_agent,
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9,tr;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'DNT': '1',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-User': '?1',
'Cache-Control': 'max-age=0',
'Referer': 'https://www.youtube.com/',
}
logger.debug(f"[HEADERS] Header'lar oluşturuldu - User-Agent: {user_agent[:50]}...")
return headers
def _check_rate_limit(self):
"""Rate limiting kontrolü (YouTube IP blocking'i önlemek için)"""
now = time.time()
# Eğer son 5 dakikada IP blocking hatası aldıysak, daha uzun bekle
if self.last_blocked_time > 0 and (now - self.last_blocked_time) < 300:
wait_time = 60 # 1 dakika bekle
logger.warning(f"[RATE_LIMIT] IP blocking sonrası bekleme: {wait_time} saniye (Son blocking: {int(now - self.last_blocked_time)} saniye önce)")
# Eğer son 10 dakikada IP blocking hatası aldıysak, çok daha uzun bekle
if self.last_blocked_time > 0 and (now - self.last_blocked_time) < 600:
# Blocking sayısına göre bekleme süresini artır
base_wait = 300 # 5 dakika base
wait_time = base_wait + (self.block_count * 60) # Her blocking'de +1 dakika
wait_time = min(wait_time, 1800) # Maksimum 30 dakika
logger.warning(f"[RATE_LIMIT] IP blocking sonrası bekleme: {wait_time} saniye ({wait_time/60:.1f} dakika) - Blocking sayısı: {self.block_count}")
logger.warning(f"[RATE_LIMIT] Son blocking: {int(now - self.last_blocked_time)} saniye önce")
time.sleep(wait_time)
self.last_blocked_time = 0 # Reset
# Reset etme, 10 dakika boyunca hatırla
# Son time_window saniyesindeki istekleri filtrele
self.request_times = [t for t in self.request_times if now - t < self.time_window]
@@ -49,12 +140,18 @@ class TranscriptExtractor:
self.request_times = [t for t in self.request_times if now - t < self.time_window]
# İstekler arasında minimum bekleme (YouTube blocking'i önlemek için)
# Human-like behavior: Random bekleme (10-20 saniye arası)
if self.request_times:
time_since_last = now - self.request_times[-1]
min_interval = 3 # Minimum 3 saniye
# Blocking varsa daha uzun bekle
if self.block_count > 0:
min_interval = 30 + random.uniform(0, 30) # 30-60 saniye random
else:
min_interval = 10 + random.uniform(0, 10) # 10-20 saniye random
if time_since_last < min_interval:
sleep_time = min_interval - time_since_last
logger.debug(f"[RATE_LIMIT] Minimum interval bekleme: {sleep_time:.2f} saniye (Son istek: {time_since_last:.2f} saniye önce)")
logger.info(f"[RATE_LIMIT] Human-like bekleme: {sleep_time:.1f} saniye (Son istek: {time_since_last:.1f} saniye önce, Blocking sayısı: {self.block_count})")
time.sleep(sleep_time)
# İstek zamanını kaydet
@@ -101,10 +198,20 @@ class TranscriptExtractor:
logger.error(f"[TRANSCRIPT] ❌ Video {video_id} transcript çıkarımı başarısız: {error_type} - {error_msg[:200]}")
# IP blocking hatası tespit edilirse işaretle
if "blocking" in error_msg.lower() or "blocked" in error_msg.lower() or "IP" in error_msg:
if "blocking" in error_msg.lower() or "blocked" in error_msg.lower() or "IP" in error_msg or "IpBlocked" in error_type:
self.last_blocked_time = time.time()
logger.warning(f"[TRANSCRIPT] 🚫 IP blocking tespit edildi! Video: {video_id}, Sonraki isteklerde 60 saniye bekleme yapılacak")
self.block_count += 1
wait_time = 300 + (self.block_count * 60) # 5 dakika + (blocking sayısı * 1 dakika)
wait_time = min(wait_time, 1800) # Maksimum 30 dakika
logger.error(f"[TRANSCRIPT] 🚫 IP blocking tespit edildi! Video: {video_id}")
logger.error(f"[TRANSCRIPT] Toplam blocking sayısı: {self.block_count}, Sonraki isteklerde {wait_time} saniye ({wait_time/60:.1f} dakika) bekleme yapılacak")
logger.warning(f"[TRANSCRIPT] IP blocking detayları: {error_msg[:500]}")
# Çok fazla blocking varsa uyar
if self.block_count >= 3:
logger.error(f"[TRANSCRIPT] ⚠️ UYARI: {self.block_count} kez IP blocking alındı! YouTube IP'nizi geçici olarak engellemiş olabilir.")
logger.error(f"[TRANSCRIPT] Öneriler: 1) Daha uzun bekleme (30+ dakika), 2) Farklı IP kullan, 3) Proxy/VPN kullan")
return None

14
src/web_server.py
View File

@@ -8,6 +8,7 @@ import os
import yaml
import time
import logging
import random
from pathlib import Path
# Logger oluştur
@@ -211,10 +212,10 @@ def process_channel(channel_id: str, max_items: int = 50) -> dict:
else:
logger.debug(f"[PROCESS] Tüm videolar zaten veritabanında")
# Bekleyen videoları işle (max_items kadar, 20'şer batch'ler halinde)
# YouTube IP blocking'i önlemek için her batch'te 20 video işlenir
# Bekleyen videoları işle (max_items kadar, küçük batch'ler halinde)
# YouTube IP blocking'i önlemek için her batch'te sadece 5 video işlenir
# max_items: Her istekte kaç video transcript işleneceği (maksimum 100)
batch_size = 20 # Her batch'te işlenecek video sayısı
batch_size = 5 # Her batch'te işlenecek video sayısı (küçük batch = daha az blocking riski)
processed_count = 0 # İşlenen transcript sayısı
# Tüm bekleyen videoları al (channel_id'ye göre filtrele)
@@ -290,10 +291,11 @@ def process_channel(channel_id: str, max_items: int = 50) -> dict:
# Batch özeti
logger.info(f"[BATCH] Batch {current_batch}/{total_batches} tamamlandı - İşlenen: {batch_processed}, Cache: {batch_cached}, Başarısız: {batch_failed}")
# Batch tamamlandı, kısa bir bekleme (rate limiting için)
# Batch tamamlandı, uzun bekleme (YouTube IP blocking önleme için)
if processed_count < max_items and batch_start + batch_size < len(all_pending_videos):
wait_time = 2
logger.debug(f"[BATCH] Batch'ler arası bekleme: {wait_time} saniye")
# Blocking varsa daha uzun bekle
wait_time = 60 + random.uniform(0, 30) # 60-90 saniye random (human-like)
logger.info(f"[BATCH] Batch'ler arası bekleme: {wait_time:.1f} saniye ({wait_time/60:.1f} dakika) - YouTube IP blocking önleme")
time.sleep(wait_time)
# İşlenmiş videoları getir