🚀 Examples & Tutorials

from backlinkprocessor import BacklinkProcessor
import networkx as nx

def seo_analysis(target_url):
    processor = BacklinkProcessor(delay=1.0)

    # Discover backlinks
    processor.crawl_backlinks([target_url], max_depth=2)

    # Build link graph
    processor.build_link_graph()

    # Calculate SEO metrics
    pagerank_scores = processor.calculate_pagerank()
    processor.calculate_domain_authority()

    # Detect spam links
    spam_links = processor.detect_link_spam()

    # Generate comprehensive report
    report = processor.generate_backlink_report(target_url)

    print("SEO Analysis Results:")
    print(f"  Total backlinks: {report['total_backlinks']}")
    print(f"  Unique domains: {report['unique_referring_domains']}")
    print(f"  Domain authority: {report['domain_authority']:.1f}")
    print(f"  Spam links detected: {len(spam_links)}")

    return report

# Analyze your website
report = seo_analysis('https://yourwebsite.com')

#!/usr/bin/env python3
import schedule
import time
from datetime import datetime
from crawler import EnhancedProductionCrawler

def scheduled_crawl():
    """Perform scheduled comprehensive crawl"""
    print(f"🕷️ Starting scheduled crawl at {datetime.now()}")

    config = {
        'delay': 1.5,
        'max_depth': 3,
        'max_pages': 200,
        'db_path': 'scheduled_crawler.db',
        'analyze_backlinks': True,
        'export_json': True
    }

    crawler = EnhancedProductionCrawler(config)
    seed_urls = [
        'https://example.com',
        'https://python.org',
        'https://github.com'
    ]

    try:
        results = crawler.comprehensive_crawl(seed_urls)
        print(f"✅ Crawl completed: {results['pages_crawled']} pages")
    except Exception as e:
        print(f"❌ Crawl failed: {e}")

def main():
    """Main scheduler function"""
    print("🚀 Starting RatCrawler Scheduler")

    # Schedule daily crawl at 2 AM
    schedule.every().day.at("02:00").do(scheduled_crawl)

    # Schedule weekly comprehensive analysis on Sundays
    schedule.every().sunday.at("03:00").do(lambda: print("📊 Weekly analysis would run here"))

    print("⏰ Scheduler active. Waiting for scheduled tasks...")
    print("Press Ctrl+C to stop")

    try:
        while True:
            schedule.run_pending()
            time.sleep(60)  # Check every minute
    except KeyboardInterrupt:
        print("\n🛑 Scheduler stopped by user")

if __name__ == "__main__":
    main()

import torch
import transformers
from crawler import EnhancedProductionCrawler
from sklearn.feature_extraction.text import TfidfVectorizer

class MLAnalysisCrawler(EnhancedProductionCrawler):
    def __init__(self, config):
        super().__init__(config)
        self.vectorizer = TfidfVectorizer(max_features=1000)
        self.sentiment_analyzer = transformers.pipeline('sentiment-analysis')

    def analyze_content_quality(self, page_data):
        """Analyze content using ML models"""
        if not page_data.get('content_text'):
            return {}

        # TF-IDF analysis
        tfidf_matrix = self.vectorizer.fit_transform([page_data['content_text']])
        content_score = tfidf_matrix.sum() / len(page_data['content_text'].split())

        # Sentiment analysis
        sentiment = self.sentiment_analyzer(page_data['content_text'][:512])[0]

        # Readability metrics
        sentences = len(page_data['content_text'].split('.'))
        words = len(page_data['content_text'].split())
        avg_sentence_length = words / sentences if sentences > 0 else 0

        return {
            'content_score': float(content_score),
            'sentiment': sentiment['label'],
            'sentiment_confidence': sentiment['score'],
            'avg_sentence_length': avg_sentence_length,
            'readability_score': 206.835 - 1.015 * (words/sentences) - 84.6 * (sentences/words)
        }

    def comprehensive_crawl_with_ml(self, seed_urls):
        """Enhanced crawl with ML analysis"""
        results = self.comprehensive_crawl(seed_urls)

        # Add ML analysis to each page
        for page in results.get('sample_pages', []):
            page_data = self.database.get_page_content(page['url'])
            if page_data:
                ml_analysis = self.analyze_content_quality(page_data)
                page.update(ml_analysis)

        return results

# Usage
config = {
    'delay': 1.0,
    'max_pages': 50,
    'db_path': 'ml_crawler.db'
}

ml_crawler = MLAnalysisCrawler(config)
results = ml_crawler.comprehensive_crawl_with_ml(['https://example.com'])

#!/usr/bin/env python3
import multiprocessing as mp
import queue
import time
from crawler import EnhancedProductionCrawler

class DistributedCrawler:
    def __init__(self, num_workers=4):
        self.num_workers = num_workers
        self.url_queue = mp.Queue()
        self.result_queue = mp.Queue()
        self.workers = []

    def worker_process(self, worker_id, config):
        """Individual crawler worker"""
        crawler = EnhancedProductionCrawler(config)

        while True:
            try:
                url = self.url_queue.get(timeout=5)
                print(f"Worker {worker_id}: Processing {url}")

                # Crawl single page
                page_data = crawler.crawl_page_content(url)
                if page_data:
                    self.result_queue.put(page_data)

            except queue.Empty:
                break
            except Exception as e:
                print(f"Worker {worker_id} error: {e}")

    def distribute_crawl(self, urls, config):
        """Distribute crawling across multiple processes"""
        # Start worker processes
        for i in range(self.num_workers):
            worker_config = config.copy()
            worker_config['db_path'] = f'worker_{i}.db'

            p = mp.Process(target=self.worker_process, args=(i, worker_config))
            p.start()
            self.workers.append(p)

        # Add URLs to queue
        for url in urls:
            self.url_queue.put(url)

        # Collect results
        results = []
        for _ in range(len(urls)):
            try:
                result = self.result_queue.get(timeout=30)
                results.append(result)
            except queue.Empty:
                break

        # Stop workers
        for p in self.workers:
            p.terminate()
            p.join()

        return results

# Usage
config = {
    'delay': 1.0,
    'max_depth': 1,  # Single page per worker
    'stay_on_domain': True
}

distributed_crawler = DistributedCrawler(num_workers=4)
urls = [f'https://example.com/page{i}' for i in range(20)]

results = distributed_crawler.distribute_crawl(urls, config)
print(f"Distributed crawl completed: {len(results)} pages")