Bot Automation Guide 2025 – Complete Bot Development and Usage Tutorial

# Basic Python bot structure
import requests
from bs4 import BeautifulSoup
import time

class WebScraperBot:
    def __init__(self, url, delay=1):
        self.url = url
        self.delay = delay
        self.session = requests.Session()

    def scrape_data(self):
        try:
            response = self.session.get(self.url)
            soup = BeautifulSoup(response.text, 'html.parser')

            # Extract data
            data = self.extract_information(soup)
            return data

        except Exception as e:
            print(f"Error scraping {self.url}: {e}")
            return None

    def extract_information(self, soup):
        # Implement data extraction logic
        return {}

    def run(self, iterations=10):
        for i in range(iterations):
            data = self.scrape_data()
            if data:
                self.process_data(data)
            time.sleep(self.delay)

const puppeteer = require('puppeteer');
const fs = require('fs');

class BrowserBot {
    constructor(options = {}) {
        this.options = {
            headless: options.headless || false,
            userAgent: options.userAgent || 'Mozilla/5.0...',
            ...options
        };
    }

    async init() {
        this.browser = await puppeteer.launch({
            headless: this.options.headless,
            args: ['--no-sandbox', '--disable-setuid-sandbox']
        });
    }

    async scrapeWebsite(url) {
        const page = await this.browser.newPage();
        await page.setUserAgent(this.options.userAgent);

        try {
            await page.goto(url, { waitUntil: 'networkidle2' });

            // Wait for content to load
            await page.waitForSelector('.content-selector');

            // Extract data
            const data = await page.evaluate(() => {
                const elements = document.querySelectorAll('.data-selector');
                return Array.from(elements).map(el => el.textContent);
            });

            return data;

        } catch (error) {
            console.error(`Error scraping ${url}:`, error);
            return null;
        } finally {
            await page.close();
        }
    }

    async close() {
        if (this.browser) {
            await this.browser.close();
        }
    }
}

# Create virtual environment
python -m venv bot_env
source bot_env/bin/activate  # Linux/Mac
# bot_env\Scripts\activate   # Windows

# Install dependencies
pip install requests beautifulsoup4 selenium webdriver-manager

import requests
from bs4 import BeautifulSoup
import json
import time
from urllib.parse import urljoin

class DataExtractorBot:
    def __init__(self, base_url, config=None):
        self.base_url = base_url
        self.session = requests.Session()
        self.config = config or {}
        self.results = []

        # Configure session
        self.session.headers.update({
            'User-Agent': self.config.get('user_agent', 'Mozilla/5.0...'),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
        })

    def scrape_page(self, url):
        """Scrape a single page"""
        try:
            response = self.session.get(url, timeout=30)
            response.raise_for_status()

            soup = BeautifulSoup(response.content, 'html.parser')
            return soup

        except requests.RequestException as e:
            print(f"Error fetching {url}: {e}")
            return None

    def extract_product_data(self, soup):
        """Extract product information"""
        products = []

        product_elements = soup.find_all('div', class_='product-item')

        for product in product_elements:
            try:
                name = product.find('h3', class_='product-name')
                price = product.find('span', class_='price')
                link = product.find('a', class_='product-link')

                if name and price:
                    product_data = {
                        'name': name.get_text().strip(),
                        'price': price.get_text().strip(),
                        'url': urljoin(self.base_url, link['href']) if link else None,
                        'timestamp': time.time()
                    }
                    products.append(product_data)

            except Exception as e:
                print(f"Error extracting product data: {e}")
                continue

        return products

    def save_results(self, filename='results.json'):
        """Save scraped data to file"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, indent=2, ensure_ascii=False)

    def run(self, pages=5):
        """Run the scraping process"""
        for page in range(1, pages + 1):
            url = f"{self.base_url}/page/{page}"

            print(f"Scraping page {page}: {url}")
            soup = self.scrape_page(url)

            if soup:
                page_products = self.extract_product_data(soup)
                self.results.extend(page_products)
                print(f"Found {len(page_products)} products on page {page}")

            # Respectful delay
            time.sleep(self.config.get('delay', 2))

        self.save_results()
        print(f"Scraping complete. Total products: {len(self.results)}")

import requests
from itertools import cycle

class ProxyBot:
    def __init__(self, proxies_list):
        self.proxies = cycle(proxies_list)
        self.session = requests.Session()

    def get_proxy(self):
        """Get next proxy from rotation"""
        proxy = next(self.proxies)
        return {
            'http': f'http://{proxy}',
            'https': f'https://{proxy}'
        }

    def make_request(self, url, **kwargs):
        """Make request with proxy rotation"""
        proxy = self.get_proxy()
        kwargs['proxies'] = proxy

        try:
            response = self.session.get(url, **kwargs)
            return response
        except requests.RequestException:
            # Try with different proxy
            proxy = self.get_proxy()
            kwargs['proxies'] = proxy
            response = self.session.get(url, **kwargs)
            return response

import random
import time
from selenium.webdriver.chrome.options import Options

class StealthBot:
    def __init__(self):
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36...',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36...',
        ]

    def get_random_user_agent(self):
        return random.choice(self.user_agents)

    def human_like_delay(self, min_delay=1, max_delay=5):
        """Add human-like delays between actions"""
        delay = random.uniform(min_delay, max_delay)
        time.sleep(delay)

    def simulate_human_behavior(self, driver):
        """Simulate human mouse movements and typing"""
        # Random mouse movements
        actions = webdriver.ActionChains(driver)
        actions.move_by_offset(random.randint(-100, 100), random.randint(-100, 100))
        actions.perform()

        # Random scrolling
        driver.execute_script(f"window.scrollTo(0, {random.randint(0, 1000)});")

        self.human_like_delay(0.5, 2)

    def setup_stealth_options(self):
        """Configure browser for stealth operation"""
        options = Options()
        options.add_argument(f'--user-agent={self.get_random_user_agent()}')
        options.add_argument('--disable-blink-features=AutomationControlled')
        options.add_experimental_option("excludeSwitches", ["enable-automation"])
        options.add_experimental_option('useAutomationExtension', False)

        return options

# Clone repository
git clone https://github.com/yourusername/bot-project.git
cd bot-project

# Create virtual environment
python -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Run development server
python main.py --debug

import unittest
from unittest.mock import Mock, patch
from bot import DataExtractorBot

class TestDataExtractorBot(unittest.TestCase):
    def setUp(self):
        self.bot = DataExtractorBot('https://example.com')

    @patch('requests.Session.get')
    def test_scrape_page_success(self, mock_get):
        # Mock successful response
        mock_response = Mock()
        mock_response.text = '<html><body><h1>Test</h1></body></html>'
        mock_response.raise_for_status.return_value = None
        mock_get.return_value = mock_response

        soup = self.bot.scrape_page('https://example.com')
        self.assertIsNotNone(soup)

    @patch('requests.Session.get')
    def test_scrape_page_error(self, mock_get):
        # Mock failed response
        mock_get.side_effect = requests.RequestException("Connection error")

        soup = self.bot.scrape_page('https://example.com')
        self.assertIsNone(soup)

if __name__ == '__main__':
    unittest.main()

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "main.py"]

import logging
import logging.handlers
from logging.config import dictConfig

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        },
        'simple': {
            'format': '%(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'simple',
            'level': 'INFO'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': 'bot.log',
            'maxBytes': 10485760,  # 10MB
            'backupCount': 5,
            'formatter': 'detailed',
            'level': 'DEBUG'
        }
    },
    'root': {
        'handlers': ['console', 'file'],
        'level': 'DEBUG'
    }
}

def setup_logging():
    dictConfig(LOGGING_CONFIG)

# Usage
setup_logging()
logger = logging.getLogger(__name__)
logger.info("Bot started successfully")

import requests

class CaptchaSolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://2captcha.com'

    def solve_recaptcha(self, site_key, url):
        """Solve reCAPTCHA v2"""
        payload = {
            'key': self.api_key,
            'method': 'userrecaptcha',
            'googlekey': site_key,
            'pageurl': url,
            'json': 1
        }

        # Submit CAPTCHA
        response = requests.post(f'{self.base_url}/in.php', data=payload)
        request_id = response.json()['request']

        # Wait for solution
        import time
        while True:
            time.sleep(5)
            result = requests.get(f'{self.base_url}/res.php?key={self.api_key}&action=get&id={request_id}&json=1')
            if result.json()['status'] == 1:
                return result.json()['request']

from sklearn.ensemble import RandomForestClassifier
import pandas as pd

class PredictiveBot:
    def __init__(self):
        self.model = RandomForestClassifier()
        self.is_trained = False

    def train_model(self, training_data):
        """Train ML model on historical data"""
        X = training_data.drop('target', axis=1)
        y = training_data['target']

        self.model.fit(X, y)
        self.is_trained = True

    def predict_action(self, current_state):
        """Predict optimal action based on current state"""
        if not self.is_trained:
            return self.default_action()

        prediction = self.model.predict([current_state])
        return prediction[0]

    def learn_from_outcome(self, state, action, outcome):
        """Update model based on action outcomes"""
        # Implement online learning
        pass

# Basic code obfuscation techniques
import base64
import marshal

def obfuscate_code(code_string):
    """Basic code obfuscation"""
    # Compile to bytecode
    compiled = compile(code_string, '<string>', 'exec')

    # Serialize bytecode
    bytecode = marshal.dumps(compiled)

    # Base64 encode
    encoded = base64.b64encode(bytecode).decode()

    return encoded

def deobfuscate_and_run(encoded_code):
    """Deobfuscate and execute code"""
    # Decode from base64
    bytecode = base64.b64decode(encoded_code)

    # Deserialize bytecode
    compiled = marshal.loads(bytecode)

    # Execute
    exec(compiled)

import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

class SecureBot:
    def __init__(self):
        self.api_key = os.getenv('API_KEY')
        self.database_url = os.getenv('DATABASE_URL')
        self.secret_token = os.getenv('SECRET_TOKEN')

    def validate_credentials(self):
        """Validate that all required credentials are present"""
        required_vars = ['API_KEY', 'DATABASE_URL', 'SECRET_TOKEN']

        missing = [var for var in required_vars if not os.getenv(var)]
        if missing:
            raise ValueError(f"Missing required environment variables: {missing}")

        return True

import asyncio
import aiohttp

class BotOrchestrator:
    def __init__(self, bot_configs):
        self.bots = []
        self.session = None

    async def init_bots(self):
        """Initialize multiple bots asynchronously"""
        self.session = aiohttp.ClientSession()

        for config in self.bot_configs:
            bot = await self.create_bot(config)
            self.bots.append(bot)

    async def run_bots_parallel(self):
        """Run all bots in parallel"""
        tasks = [bot.run() for bot in self.bots]
        await asyncio.gather(*tasks)

    async def monitor_bots(self):
        """Monitor bot performance and health"""
        while True:
            for i, bot in enumerate(self.bots):
                status = await bot.get_status()
                if status['health'] != 'healthy':
                    print(f"Bot {i} unhealthy: {status}")
                    await self.restart_bot(i)

            await asyncio.sleep(60)  # Check every minute

from collections import deque

class LoadBalancer:
    def __init__(self, bots):
        self.bots = deque(bots)
        self.request_count = 0

    def get_next_bot(self):
        """Round-robin bot selection"""
        bot = self.bots[0]
        self.bots.rotate(-1)  # Move to end of queue
        return bot

    def distribute_request(self, request):
        """Distribute request to next available bot"""
        bot = self.get_next_bot()

        try:
            result = bot.process_request(request)
            self.request_count += 1
            return result
        except Exception as e:
            # Try next bot if current fails
            bot = self.get_next_bot()
            result = bot.process_request(request)
            return result