종속성 설치

Redis가 설치되어 있고 redis-py Python에서 Redis와 함께 작동하는 라이브러리입니다. pip를 사용하여 필요한 종속성을 설치합니다.

pip install redis

Redis 관리자 생성

다음을 만들어 보겠습니다. RedisManager 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

스크립트 1: 대량으로 프록시 추가

이 스크립트는 파일에서 프록시 목록을 읽고 이를 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()

스크립트 2: 무작위 프록시 검색

이 스크립트는 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()

스크립트 3: 프록시 존재 확인

이 스크립트는 Redis 데이터베이스에 특정 프록시 서버가 존재하는지 확인합니다.

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()

스크립트 4: 프록시 삭제

이 스크립트는 Redis 데이터베이스에서 프록시를 삭제합니다. 먼저 delete_proxy 에 대한 방법 RedisManager 수업:

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

이제 프록시 삭제 스크립트는 다음과 같습니다.

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()

결론

이 가이드에서는 Python 및 Redis를 사용하여 프록시를 관리하기 위한 여러 스크립트를 만들었습니다. 이러한 스크립트를 사용하면 Redis 데이터베이스에서 프록시를 추가, 검색, 확인 및 삭제할 수 있어 프록시 관리를 위한 강력한 솔루션을 제공할 수 있습니다. 이러한 예를 기반으로 특정 요구 사항에 맞게 프록시 관리 시스템을 추가로 사용자 정의하고 확장할 수 있습니다.

댓글 (0)

여기에는 아직 댓글이 없습니다. 첫 번째 댓글이 되실 수 있습니다!

답글 남기기

이메일 주소는 공개되지 않습니다. 필수 필드는 *로 표시됩니다


프록시 선택 및 구매

데이터센터 프록시

회전 프록시

UDP 프록시

전 세계 10,000명 이상의 고객이 신뢰함

대리 고객
대리 고객
대리 고객 flowch.ai
대리 고객
대리 고객
대리 고객