Instalowanie zależności

Upewnij się, że masz zainstalowany Redis i redis-py biblioteka do pracy z Redis z Pythona. Zainstaluj niezbędne zależności za pomocą pip:

pip install redis

Tworzenie Menedżera Redis

Stwórzmy RedisManager klasa do obsługi interakcji z Redis.

import redis
import json
import random
import logging

class RedisManager:
    def __init__(self, host='localhost', port=6379, db=0):
        self.db = redis.StrictRedis(host=host, port=port, db=db, decode_responses=True)
        self.logger = logging.getLogger('proxy_manager')
        self.logger.setLevel(logging.INFO)
        if not self.logger.hasHandlers():
            ch = logging.StreamHandler()
            ch.setLevel(logging.INFO)
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            ch.setFormatter(formatter)
            self.logger.addHandler(ch)
        self.proxies = []
        self.load_proxy_list()

    def update_proxies(self, filepath):
        proxies = self.read_proxies_from_file(filepath)
        for proxy in proxies:
            if proxy.get('anonymitylevel') == 'elite' and proxy.get('protocol') in ['http', 'https']:
                if not self.proxy_exists(proxy):
                    self.logger.info(f"proxy: {proxy['ipaddress']} ({proxy['protocol']}, {proxy['anonymitylevel']})")
                    self.proxy_save(proxy)
        self.load_proxy_list()

    def get_proxy(self):
        return random.choice(self.proxies)

    def load_proxy_list(self):
        self.proxies = []
        ret = self.db.smembers("proxies")
        for pid in ret:
            proxy = self.load_proxy(pid)
            if proxy:
                self.proxies.append(proxy)

    def load_proxy(self, pid):
        data = self.db.hmget(f"proxy:{pid}", "ipaddress", "port", "protocol", "anonymitylevel", "source", "country")
        if not all(data):
            return None
        return {k: v for k, v in zip(["ipaddress", "port", "protocol", "anonymitylevel", "source", "country"], data)}

    def proxy_save(self, proxy):
        next_id = self.db.incr("proxies_next_id")
        self.db.hmset(f"proxy:{next_id}", proxy)
        self.db.sadd("proxies", next_id)
        self.db.hset("proxies_ids", proxy['ipaddress'], next_id)

    def proxy_exists(self, proxy):
        return self.db.hexists("proxies_ids", proxy['ipaddress'])

    def read_proxies_from_file(self, filepath):
        with open(filepath, 'r') as file:
            values = json.load(file)
        proxies = []
        for data in values:
            proxy = {
                "ipaddress": data.get("ipaddress", "").lower(),
                "port": str(data.get("port", "")),
                "protocol": data.get("protocols", [""])[0].lower(),
                "anonymitylevel": data.get("anonymitylevel", "").lower(),
                "source": data.get("source", "").lower(),
                "country": data.get("country", "").lower()
            }
            proxies.append(proxy)
        return proxies

Skrypt 1: Zbiorcze dodawanie serwerów proxy

Skrypt ten odczytuje listę serwerów proxy z pliku i dodaje je do bazy danych Redis.

import logging

logging.basicConfig(level=logging.INFO)

def main():
    manager = RedisManager()
    filepath = 'path/to/proxy_list.json'
    manager.update_proxies(filepath)
    logging.info("Proxies updated successfully")

if __name__ == "__main__":
    main()

Skrypt 2: Pobieranie losowego serwera proxy

Ten skrypt pobiera losowy serwer proxy z bazy danych Redis.

import logging

logging.basicConfig(level=logging.INFO)

def main():
    manager = RedisManager()
    proxy = manager.get_proxy()
    logging.info(f"Random proxy: {proxy}")

if __name__ == "__main__":
    main()

Skrypt 3: Sprawdzanie istnienia serwera proxy

Ten skrypt sprawdza, czy w bazie danych Redis istnieje określony serwer proxy.

import logging

logging.basicConfig(level=logging.INFO)

def main():
    manager = RedisManager()
    test_proxy = {
        "ipaddress": "192.168.1.1",
        "port": "8080",
        "protocol": "http",
        "anonymitylevel": "elite",
        "source": "test_source",
        "country": "us"
    }
    exists = manager.proxy_exists(test_proxy)
    if exists:
        logging.info("Proxy exists in the database")
    else:
        logging.info("Proxy does not exist in the database")

if __name__ == "__main__":
    main()

Skrypt 4: Usuwanie serwera proxy

Ten skrypt usuwa serwer proxy z bazy danych Redis. Najpierw dodaj a delete_proxy metoda do RedisManager klasa:

def delete_proxy(self, pid):
    self.db.delete(f"proxy:{pid}")
    self.db.srem("proxies", pid)

Teraz skrypt do usuwania proxy:

import logging

logging.basicConfig(level=logging.INFO)

def main():
    manager = RedisManager()
    proxy_id = 1  # Replace with the ID of the proxy you want to delete
    manager.delete_proxy(proxy_id)
    logging.info("Proxy deleted successfully")

if __name__ == "__main__":
    main()

Wnioski

W tym przewodniku stworzyliśmy kilka skryptów do zarządzania serwerami proxy przy użyciu Pythona i Redis. Te skrypty umożliwiają dodawanie, pobieranie, sprawdzanie i usuwanie serwerów proxy z bazy danych Redis, zapewniając niezawodne rozwiązanie do zarządzania serwerami proxy. Bazując na tych przykładach, możesz jeszcze bardziej dostosować i rozszerzyć system zarządzania proxy, aby dopasować go do swoich konkretnych potrzeb.

Komentarze (0)

Nie ma tu jeszcze żadnych komentarzy, możesz być pierwszy!

Dodaj komentarz

Twój adres e-mail nie zostanie opublikowany. Wymagane pola są oznaczone *


Wybierz i kup proxy

Serwery proxy dla centrów danych

Obrotowe proxy

Serwery proxy UDP

Zaufało nam ponad 10000 klientów na całym świecie

Klient proxy
Klient proxy
Klient proxy flowch.ai
Klient proxy
Klient proxy
Klient proxy