import gradio as gr
import requests
from bs4 import BeautifulSoup
import time
import threading
from datetime import datetime
import re
# HTML'den linkleri çıkar - Güncel liste (36 link)
HTML_CONTENT = """
"""
# Base URL - gerçek site URL'i
BASE_URL = "https://malatyayenises.com"
# Global değişkenler
is_running = False
scraping_thread = None
status_log = []
def parse_urls_from_html(html_content):
"""HTML içeriğinden URL'leri çıkar"""
soup = BeautifulSoup(html_content, 'html.parser')
urls = []
for link in soup.find_all('a', href=True):
href = link.get('href')
text = link.get_text(strip=True)
# Time elementini çıkar
time_elem = link.find('time')
if time_elem:
text = text.replace(time_elem.get_text(strip=True), '').strip()
if href and href.startswith('/irfan-basaranoglu/'):
full_url = BASE_URL + href
urls.append({
'url': full_url,
'title': text,
'path': href
})
return urls
def scrape_page(url_info):
"""Bir sayfayı scrape et"""
try:
response = requests.get(url_info['url'], timeout=10, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# Sayfa başlığını al
title = soup.find('title')
title_text = title.get_text(strip=True) if title else url_info['title']
# Ana içeriği al (içeriğe göre ayarlanabilir)
content = soup.find('article') or soup.find('div', class_='content') or soup.find('main')
content_text = content.get_text(strip=True)[:500] if content else "İçerik bulunamadı"
return {
'success': True,
'url': url_info['url'],
'title': title_text,
'preview': content_text,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
except Exception as e:
return {
'success': False,
'url': url_info['url'],
'error': str(e),
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
def continuous_scraping(interval_seconds, url_list, status_callback=None):
"""Sürekli olarak sayfaları ziyaret et"""
global is_running
url_index = 0
while is_running:
if url_list:
url_info = url_list[url_index % len(url_list)]
result = scrape_page(url_info)
status_msg = f"[{result['timestamp']}] Ziyaret: {url_info['title'][:50]}..."
if result['success']:
status_msg += f" ✓ Başarılı"
else:
status_msg += f" ✗ Hata: {result.get('error', 'Bilinmeyen')}"
status_log.append(status_msg)
if len(status_log) > 100: # Son 100 kaydı tut
status_log.pop(0)
if status_callback:
status_callback("\n".join(status_log[-20:])) # Son 20 satırı göster
url_index += 1
time.sleep(interval_seconds)
else:
time.sleep(1)
def start_scraping(base_url, interval):
"""Scraping'i başlat"""
global is_running, scraping_thread
if is_running:
return "Zaten çalışıyor!", "\n".join(status_log[-20:]) if status_log else "Henüz kayıt yok."
# Base URL'yi güncelle
global BASE_URL
BASE_URL = base_url or "https://malatyayenises.com"
# URL listesini oluştur
url_list = parse_urls_from_html(HTML_CONTENT)
if not url_list:
return "URL bulunamadı!", "HTML içeriğinden link çıkarılamadı."
is_running = True
status_log.clear()
def update_status(new_status):
# Bu fonksiyon Gradio'dan çağrılacak
pass
scraping_thread = threading.Thread(
target=continuous_scraping,
args=(interval, url_list, update_status),
daemon=True
)
scraping_thread.start()
return f"Scraping başlatıldı! {len(url_list)} sayfa izleniyor. Interval: {interval} saniye.", "\n".join(status_log[-20:]) if status_log else "Başlatılıyor..."
def stop_scraping():
"""Scraping'i durdur"""
global is_running
if not is_running:
return "Zaten durmuş!", "\n".join(status_log[-20:]) if status_log else "Henüz kayıt yok."
is_running = False
return "Scraping durduruldu.", "\n".join(status_log[-20:]) if status_log else "Durduruldu."
def get_status():
"""Mevcut durumu döndür"""
global is_running, status_log
status_text = "Çalışıyor" if is_running else "Durmuyor"
log_text = "\n".join(status_log[-20:]) if status_log else "Henüz kayıt yok."
return f"Durum: {status_text}", log_text
# Gradio Interface
def create_interface():
with gr.Blocks(title="Babam Yazılar - Sürekli Sayfa Gezici") as demo:
gr.Markdown("# 📚 Babam Yazılar - Sürekli Sayfa Gezici")
gr.Markdown("Bu uygulama, belirtilen sayfaları sürekli olarak ziyaret eder ve durumu gösterir.")
with gr.Row():
with gr.Column():
base_url_input = gr.Textbox(
label="Base URL",
value="https://malatyayenises.com",
placeholder="Örn: https://malatyayenises.com"
)
interval_input = gr.Slider(
label="Ziyaret Aralığı (saniye)",
minimum=1,
maximum=300,
value=10,
step=1
)
with gr.Row():
start_btn = gr.Button("Başlat", variant="primary")
stop_btn = gr.Button("Durdur", variant="stop")
status_btn = gr.Button("Durum", variant="secondary")
status_output = gr.Textbox(
label="Durum Mesajı",
interactive=False,
lines=2
)
with gr.Column():
log_output = gr.Textbox(
label="İşlem Logları",
interactive=False,
lines=20,
max_lines=20
)
# Button actions
def start_action(base_url, interval):
msg, log = start_scraping(base_url, int(interval))
return msg, log
def stop_action():
msg, log = stop_scraping()
return msg, log
def status_action():
msg, log = get_status()
return msg, log
start_btn.click(
fn=start_action,
inputs=[base_url_input, interval_input],
outputs=[status_output, log_output]
)
stop_btn.click(
fn=stop_action,
outputs=[status_output, log_output]
)
status_btn.click(
fn=status_action,
outputs=[status_output, log_output]
)
# Otomatik güncelleme
def update_logs():
"""Logları güncelle"""
global status_log
if status_log:
return "\n".join(status_log[-20:])
return "Henüz kayıt yok."
demo.load(
fn=update_logs,
outputs=[log_output],
every=3 # Her 3 saniyede bir güncelle
)
# Log output'u otomatik güncelle
log_output.change(
fn=update_logs,
outputs=[log_output],
every=3
)
return demo
if __name__ == "__main__":
demo = create_interface()
demo.launch(server_name="0.0.0.0", server_port=7860)