ഡിപൻഡൻസികൾ ഇൻസ്റ്റാൾ ചെയ്യുന്നു
നിങ്ങൾ Redis ഇൻസ്റ്റാൾ ചെയ്തിട്ടുണ്ടെന്ന് ഉറപ്പാക്കുക redis-py
പൈത്തണിൽ നിന്നുള്ള റെഡിസിനൊപ്പം പ്രവർത്തിക്കാനുള്ള ലൈബ്രറി. പൈപ്പ് ഉപയോഗിച്ച് ആവശ്യമായ ഡിപൻഡൻസികൾ ഇൻസ്റ്റാൾ ചെയ്യുക:
pip install redis
Redis മാനേജർ സൃഷ്ടിക്കുന്നു
നമുക്ക് ഒരു സൃഷ്ടിക്കാം RedisManager
റെഡിസുമായുള്ള ഇടപെടലുകൾ കൈകാര്യം ചെയ്യുന്നതിനുള്ള ക്ലാസ്.
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()
ഉപസംഹാരം
ഈ ഗൈഡിൽ, പൈത്തണും റെഡിസും ഉപയോഗിച്ച് പ്രോക്സികൾ കൈകാര്യം ചെയ്യുന്നതിനായി ഞങ്ങൾ നിരവധി സ്ക്രിപ്റ്റുകൾ സൃഷ്ടിച്ചിട്ടുണ്ട്. ഒരു Redis ഡാറ്റാബേസിൽ നിന്ന് പ്രോക്സികൾ ചേർക്കാനും വീണ്ടെടുക്കാനും പരിശോധിക്കാനും ഇല്ലാതാക്കാനും ഈ സ്ക്രിപ്റ്റുകൾ നിങ്ങളെ അനുവദിക്കുന്നു, ഇത് പ്രോക്സി മാനേജ്മെൻ്റിന് ശക്തമായ ഒരു പരിഹാരം നൽകുന്നു. ഈ ഉദാഹരണങ്ങൾ അടിസ്ഥാനമാക്കി, നിങ്ങളുടെ നിർദ്ദിഷ്ട ആവശ്യങ്ങൾക്ക് അനുയോജ്യമായ രീതിയിൽ നിങ്ങളുടെ പ്രോക്സി മാനേജ്മെൻ്റ് സിസ്റ്റം കൂടുതൽ ഇഷ്ടാനുസൃതമാക്കാനും വിപുലീകരിക്കാനും നിങ്ങൾക്ക് കഴിയും.
അഭിപ്രായങ്ങൾ (0)
ഇവിടെ ഇതുവരെ അഭിപ്രായങ്ങളൊന്നുമില്ല, നിങ്ങൾക്ക് ആദ്യത്തെയാളാകാം!