139 lines
5.9 KiB
Python
139 lines
5.9 KiB
Python
import os
|
||
import time
|
||
import logging
|
||
from urllib.parse import urlparse, urljoin
|
||
from bs4 import BeautifulSoup
|
||
from markdownify import markdownify as md
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
||
|
||
class DocsCrawler:
|
||
def __init__(self, start_url: str, max_pages: int = 50, headless: bool = True, allowed_path: str = None):
|
||
self.start_url = start_url
|
||
self.max_pages = max_pages
|
||
self.headless = headless
|
||
self.visited = set()
|
||
self.queue = [start_url]
|
||
|
||
parsed_url = urlparse(start_url)
|
||
self.base_domain = parsed_url.netloc
|
||
self.base_path = allowed_path if allowed_path is not None else parsed_url.path
|
||
|
||
# Sadece bu prefix ile başlayan linkleri gezeceğiz
|
||
self.url_prefix = f"{parsed_url.scheme}://{self.base_domain}{self.base_path}"
|
||
|
||
def is_valid_url(self, url: str) -> bool:
|
||
"""Sadece aynı domain ve prefix altındaki doküman linklerini geçerli sayar."""
|
||
if url in self.visited:
|
||
return False
|
||
if not url.startswith(self.url_prefix):
|
||
return False
|
||
# Gereksiz dosyaları atla
|
||
if any(url.endswith(ext) for ext in [".png", ".jpg", ".pdf", ".zip", ".json"]):
|
||
return False
|
||
return True
|
||
|
||
def save_markdown(self, url: str, content: str):
|
||
"""Çekilen markdown'ı URL yapısına göre klasörleyerek kaydeder."""
|
||
parsed = urlparse(url)
|
||
# Örn: /docs/app/building-your-application -> docs/app/building-your-application
|
||
path = parsed.path.strip("/")
|
||
if not path:
|
||
path = "index"
|
||
|
||
# Dosya yolu oluştur
|
||
file_path = os.path.join("data", "md_docs", self.base_domain, f"{path}.md")
|
||
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
||
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
logging.info(f"Kaydedildi: {file_path}")
|
||
|
||
def extract_main_content(self, html_content: str) -> str:
|
||
"""HTML içerisinden menüleri/sidearları atıp asıl metni alır."""
|
||
soup = BeautifulSoup(html_content, "lxml")
|
||
|
||
# Çoğu modern docs sitesi içeriği article veya main etiketinde tutar.
|
||
main_content = soup.find("article") or soup.find("main")
|
||
|
||
# Eğer bulunamazsa div id="content" gibi ortak yapıları ara
|
||
if not main_content:
|
||
main_content = soup.find("div", id="content") or soup.find("div", class_="content")
|
||
|
||
# Hiçbiri yoksa fallback olarak body kullan
|
||
if not main_content:
|
||
main_content = soup.find("body")
|
||
|
||
# Gereksiz script, style ve nav etiketlerini temizle
|
||
if main_content:
|
||
for unwanted in main_content(["script", "style", "nav", "footer", "header"]):
|
||
unwanted.decompose()
|
||
return str(main_content)
|
||
|
||
return ""
|
||
|
||
def crawl(self):
|
||
"""Kuyruktaki URL'leri sırayla gezip indirir."""
|
||
logging.info(f"Tarama başlatılıyor: {self.start_url}")
|
||
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=self.headless)
|
||
context = browser.new_context(
|
||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||
)
|
||
page = context.new_page()
|
||
|
||
count = 0
|
||
while self.queue and count < self.max_pages:
|
||
current_url = self.queue.pop(0)
|
||
|
||
# Sayfadaki # (anchor) linklerini temizleyerek ziyaret edildi say
|
||
clean_url = current_url.split("#")[0]
|
||
|
||
if clean_url in self.visited:
|
||
continue
|
||
|
||
self.visited.add(clean_url)
|
||
count += 1
|
||
|
||
logging.info(f"[{count}/{self.max_pages}] Çekiliyor: {clean_url}")
|
||
|
||
try:
|
||
# networkidle JS ile yüklenen siteler için önemlidir
|
||
page.goto(clean_url, wait_until="networkidle", timeout=30000)
|
||
html_content = page.content()
|
||
|
||
# 1. Main içeriği bul ve ayıkla
|
||
content_html = self.extract_main_content(html_content)
|
||
if not content_html:
|
||
logging.warning(f"İçerik bulunamadı: {clean_url}")
|
||
continue
|
||
|
||
# 2. HTML'i Markdown'a dönüştür
|
||
# heading_style="ATX" -> ## Başlık (daha iyi markdown formatı)
|
||
md_text = md(content_html, heading_style="ATX", default_title=True)
|
||
|
||
# URL'i başlığa ekleyebiliriz referans için
|
||
md_text = f"# Source: {clean_url}\n\n" + md_text
|
||
|
||
# 3. Kaydet
|
||
self.save_markdown(clean_url, md_text)
|
||
|
||
# 4. Sayfadaki yeni linkleri bul ve sıraya ekle
|
||
soup = BeautifulSoup(html_content, "lxml")
|
||
for a_tag in soup.find_all("a", href=True):
|
||
href = a_tag["href"]
|
||
# Tam URL'yi oluştur
|
||
full_url = urljoin(clean_url, href).split("#")[0]
|
||
if self.is_valid_url(full_url) and full_url not in self.queue:
|
||
self.queue.append(full_url)
|
||
|
||
except Exception as e:
|
||
logging.error(f"Hata oluştu ({clean_url}): {e}")
|
||
|
||
time.sleep(1) # Aşırı yüklenmemek için ufak bekleme
|
||
|
||
browser.close()
|
||
logging.info("Tarama tamamlandı.")
|