Code
Prompt: melhore esse codigo, a performance e o resultado, eu quero que exista o email no site que pedir
import os
import re
import concurrent.futures
from functools import partial
class CloudDatabase:
def __init__(self, dir_name):
self.dir_name = dir_name
self.file_paths = [os.path.join(self.dir_name, file_name) for file_name in os.listdir(self.dir_name) if file_name.endswith('.txt')]
self.data = self.read_all_files()
print(f"[INFO] Leitura inicial concluída. {len(self.data)} linhas carregadas.")
def read_all_files(self):
data = []
for file_path in self.file_paths:
with open(file_path, 'r', encoding='utf-8') as f:
data.extend(f.readlines())
return data
def search(self, site, search_type):
results = set()
for line in self.data:
if site in line:
results.update(self.extract_matches(line, search_type))
return results
def extract_matches(self, line, search_type):
results = set()
if search_type == "email:senha":
matches = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}:[^\s]+', line)
elif search_type == "cpf:senha":
matches = re.findall(r'\b\d{11}:[^\s]+', line)
else:
return results
for match in matches:
result = re.sub(r'https?://\S+|www\.\S+', '', match).strip()
if result:
results.add(result)
return results
def save_results(results, search_type, site):
result_dir = 'resultados'
os.makedirs(result_dir, exist_ok=True)
site_base_name = site.split('.')[0]
file_name = f'{site_base_name}_{search_type.replace(":", "_")}_results.txt'
file_path = os.path.join(result_dir, file_name)
with open(file_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(results))
print(f"\n[INFO] Resultados salvos em {file_path}\n")
def print_header():
header = """
____ _ _ ____ _
| _ \\| | | | | _ \\ | |
| |_) | |_ _ ___ | |_ ___ _ __ | |_) |_ _ ___ | |_ ___
| _ <| | | | |/ _ \\ | __| / _ \\| '__|| _ <| | | |/ _ \\| __/ _ \\
| |_) | | |_| | __/ | |_ | __/| | | |_) | |_| | __/| || (_) |
|____/|_|\\__,_|\\___| \\__| \\___||_| |____/ \\__,_|\\___| \\__\\___/
"""
print(header)
def print_options():
print("\nEscolha uma opção:")
print("[1] Buscar por email:senha")
print("[2] Buscar por cpf:senha")
print("[3] Sair")
def process_site(site, search_type, db):
print(f'\n[INFO] Buscando dados para "{site}" com tipo de dado "{search_type}"...')
results = db.search(site, search_type)
if results:
print(f'[RESULTADO] Dados encontrados para "{site}":')
for result in results:
print(result)
save_results(results, search_type, site)
else:
print(f'[INFO] Nenhum resultado encontrado para "{site}".')
def main():
dir_name = 'db'
if not os.path.exists(dir_name):
print("[ERROR] Diretório 'db' não encontrado.")
return
cloud_db = CloudDatabase(dir_name)
print_header()
search_type = None
while search_type not in ["email:senha", "cpf:senha"]:
print_options()
option = input("Digite a opção desejada: ")
if option == "1":
search_type = "email:senha"
elif option == "2":
search_type = "cpf:senha"
elif option == "3":
print("Saindo...")
return
else:
print("[ERROR] Opção inválida. Tente novamente.")
continue
sites_input = input("\nDigite os sites separados por espaço (exemplo: facebook.com gmail.com): ")
sites = sites_input.split()
if not sites:
print("[ERROR] Nenhum site foi fornecido. Saindo...")
return
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(process_site, site, search_type, cloud_db) for site in sites]
concurrent.futures.wait(futures)
if __name__ == '__main__':
main()