Code
import requests
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Semaphore
from tqdm import tqdm
from colorama import Fore, Style, init
# Initialize colorama
init(autoreset=True)
# Configure the maximum number of concurrent threads
MAX_THREADS = 100
semaphore = Semaphore(MAX_THREADS)
def check_plugin(domain, plugin_slug):
"""Check if the specified plugin is present on the domain and return its version if found"""
url = f"/wp-content/plugins/{plugin_slug}/readme.txt"
try:
response = requests.get(url, timeout=10)
# Check if the response is OK
if response.status_code == 200 and 'Stable tag' in response.text:
content = response.text
# Extract version from "Stable tag: "
version_line = next((line for line in content.splitlines() if line.startswith("Stable tag:")), None)
version = version_line.split(": ")[1] if version_line else "Version not found"
return True, version
except Exception as e:
print(f"Error checking {domain}: {str(e)}")
return False, None
def process_domain(domain, plugin_slug, output_file, pbar):
"""Process a single domain"""
with semaphore: # Limit the number of concurrent threads
print(f"{Fore.YELLOW}Checking domain: {domain}{Style.RESET_ALL}")
is_plugin_active, version = check_plugin(domain, plugin_slug)
if is_plugin_active:
print(f"{Fore.GREEN}✅ {plugin_slug.replace('-', ' ').title()} plugin detected on {domain} (Version: {version}){Style.RESET_ALL}")
# Write result to CSV output file immediately
with open(output_file, mode='a', newline='', encoding='utf-8') as outfile:
writer = csv.writer(outfile)
writer.writerow([domain, version])
else:
print(f"{Fore.RED}❌ {plugin_slug.replace('-', ' ').title()} plugin not detected on {domain}{Style.RESET_ALL}")
pbar.update(1)
def main(input_file, output_file, plugin_slug):
try:
# Read domains from TXT input file
with open(input_file, mode='r', encoding='utf-8') as infile:
domains = [line.strip() for line in infile if line.strip()]
print(f"Found {len(domains)} domains to process")
# Create output file
with open(output_file, mode='w', newline='', encoding='utf-8') as outfile:
writer = csv.writer(outfile)
writer.writerow(['domain', 'version']) # Header for output CSV
with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
futures = {}
with tqdm(total=len(domains), desc="Processing Domains") as pbar:
for domain in domains:
future = executor.submit(process_domain, domain, plugin_slug, output_file, pbar)
futures[future] = domain
# Check any remaining futures
for future in as_completed(futures):
futures.pop(future) # Remove completed futures
print(f"{Fore.CYAN}Processing complete.{Style.RESET_ALL}")
except Exception as e:
print(f"General error: {e}")
if __name__ == "__main__":
input_file = 'wp_domains.txt' # Change this to your input file path
output_file = 'output_results.csv' # Change this to your output file path
plugin_slug = 'wp-health' # Change this to your desired plugin slug
main(input_file, output_file, plugin_slug)If you want wordpress list, I can help you! I have 2M+ wp domain list. 🙃