Code #5367

{
 "title": "Sistema de Busca de Credenciais Otimizado",
 "description": "Este código aprimora um sistema de busca de credenciais em arquivos de texto, melhorando o desempenho com paralelismo de threads para processamento de arquivos e aprimorando a precisão dos resultados de email ao validar o domínio do email com o site solicitado.",
 "text": [
  "# db_manager.py\n\nimport os\nimport re\nimport concurrent.futures\nfrom functools import partial\n\nclass CloudDatabase:\n    \"\"\"\n    Gerencia a leitura e busca de credenciais em múltiplos arquivos de texto.\n    Otimizado para desempenho e precisão na associação de emails/CPFs a sites.\n    \"\"\"\n    \n    # Padrões Regex pré-compilados para eficiência.\n    # email_pattern: Captura o usuário, o domínio e a senha de um email:senha.\n    # cpf_pattern: Captura o CPF e a senha de um cpf:senha.\n    _EMAIL_PATTERN = re.compile(r'\\b([A-Za-z0-9._%+-]+)@([A-Za-z0-9.-]+\\.[A-Z|a-z]{2,7}):([^\s]+)')\n    _CPF_PATTERN = re.compile(r'\\b(\\d{11}):([^\s]+)')\n\n    def __init__(self, dir_name):\n        \"\"\"\n        Inicializa a base de dados, escaneando o diretório por arquivos .txt.\n        Não carrega todos os dados em memória no início para otimizar o uso de RAM.\n        \"\"\"\n        self.dir_name = dir_name\n        # Lista todos os caminhos dos arquivos .txt no diretório especificado.\n        self.file_paths = []\n        if os.path.exists(dir_name) and os.path.isdir(dir_name):\n            self.file_paths = [os.path.join(self.dir_name, file_name) \n                               for file_name in os.listdir(self.dir_name) \n                               if file_name.endswith('.txt')]\n            print(f\"[INFO] {len(self.file_paths)} arquivos .txt encontrados em '{dir_name}'.\")\n        else:\n            print(f\"[ERRO] Diretório '{dir_name}' não encontrado ou não é um diretório.\")\n\n    def _normalize_domain(self, domain):\n        \"\"\"\n        Normaliza um domínio, removendo 'www.' e convertendo para minúsculas\n        para garantir comparações consistentes (ex: 'www.site.com' -> 'site.com').\n        \"\"\"\n        return domain.lower().replace('www.', '')\n\n    def _extract_matches_from_line(self, line, target_site, search_type):\n        \"\"\"\n        Extrai correspondências (email:senha ou cpf:senha) de uma única linha.\n        Para 'email:senha', verifica se o domínio do email corresponde ao site alvo.\n        \"\"\"\n        results = set()\n        normalized_target_site = self._normalize_domain(target_site)\n\n        if search_type == 'email:senha':\n            # Itera sobre todas as correspondências do padrão de email na linha.\n            for match in self._EMAIL_PATTERN.finditer(line):\n                email_user, email_domain, password = match.groups()\n                normalized_email_domain = self._normalize_domain(email_domain)\n                \n                # Garante que o domínio do email extraído corresponde ao site solicitado.\n                if normalized_email_domain == normalized_target_site:\n                    full_match = f\"{email_user}@{email_domain}:{password}\"\n                    # Remove URLs que possam ter sido acidentalmente capturadas na senha.\n                    clean_match = re.sub(r'https?://\\S+|www\\.\\S+', '', full_match).strip()\n                    if clean_match:\n                        results.add(clean_match)\n        elif search_type == 'cpf:senha':\n            # Para CPF, considera uma correspondência se a linha contém o site\n            # e um par CPF:senha. A verificação 'site in line' é feita antes.\n            for match in self._CPF_PATTERN.finditer(line):\n                cpf, password = match.groups()\n                full_match = f\"{cpf}:{password}\"\n                clean_match = re.sub(r'https?://\\S+|www\\.\\S+', '', full_match).strip()\n                if clean_match:\n                    results.add(clean_match)\n        return results\n\n    def _search_in_file(self, file_path, target_site, search_type):\n        \"\"\"\n        Procura por credenciais em um único arquivo.\n        Esta função é executada por um thread no pool para paralelização.\n        \"\"\"\n        file_results = set()\n        try:\n            # Abre o arquivo com encoding UTF-8 e ignora erros para robustez.\n            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:\n                for line in f:\n                    # Otimização: verifica primeiro se o site alvo está na linha\n                    # antes de tentar processamento de regex mais caro.\n                    if target_site in line:\n                        matches = self._extract_matches_from_line(line, target_site, search_type)\n                        file_results.update(matches)\n        except Exception as e:\n            print(f\"[ERRO] Falha ao ler arquivo '{file_path}': {e}\")\n        return file_results\n\n    def search(self, target_site, search_type):\n        \"\"\"\n        Executa a busca principal por credenciais em todos os arquivos da base de dados.\n        Usa um ThreadPoolExecutor para paralelizar a busca em múltiplos arquivos,\n        acelerando o processo para grandes coleções de dados.\n        \"\"\"\n        all_results = set()\n        # max_workers: ajustado para tarefas I/O-bound (leitura de arquivo).\n        with concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count() * 2) as executor:\n            # 'partial' é usado para fixar os argumentos 'target_site' e 'search_type'\n            # para a função '_search_in_file' que será mapeada sobre 'file_paths'.\n            func = partial(self._search_in_file, target_site=target_site, search_type=search_type)\n            # 'executor.map' aplica a função a cada item em 'self.file_paths' em paralelo\n            # e retorna um iterador dos resultados, que são então agregados.\n            for file_results in executor.map(func, self.file_paths):\n                all_results.update(file_results)\n        return all_results\n\n# ----- Funções de Utilitário e UI (Mantidas/Aprimoradas) -----\n\ndef save_results(results, search_type, site):\n    \"\"\"\n    Salva os resultados da busca em um arquivo de texto organizado por site e tipo de busca.\n    \"\"\"\n    result_dir = 'resultados'\n    os.makedirs(result_dir, exist_ok=True)\n    \n    # Adapta o nome do arquivo para ser mais limpo e descritivo.\n    site_base_name = site.split('.')[0] if '.' in site else site\n    file_name = f'{site_base_name}_{search_type.replace(\":\", \"_\")}_results.txt'\n    file_path = os.path.join(result_dir, file_name)\n\n    with open(file_path, 'w', encoding='utf-8') as f:\n        if results:\n            f.write('\\n'.join(sorted(list(results)))) # Salva resultados ordenados\n        else:\n            f.write(\"Nenhum resultado encontrado.\\n\")\n\n    print(f\"\\n[INFO] Resultados salvos em '{file_path}'\\n\")\n\ndef print_header():\n    \"\"\"\n    Exibe um cabeçalho ASCII art para o programa.\n    \"\"\"\n    header = \"\"\"\n ____  _              _               ____              _       \n|  _ \\| |            | |             |  _ \\            | |      \n| |_) | |_   _  ___   | |_   ___  _ __ | |_) |_   _  ___ | |_ ___  \n|  _ <| | | | |/ _ \\  | __| / _ \\| '__||  _ <| | | |/ _ \\| __/ _ \\ \n| |_) | | |_| |  __/  | |_ |  __/| |   | |_) | |_| |  __/| || (_) |\n|____/|_|\\__,_|\\___|   \\__| \\___||_|   |____/ \\__,_|\\___| \\__\\___/ \n                                                                    \n\"\"\"\n    print(header)\n\ndef print_options():\n    \"\"\"\n    Exibe as opções de busca disponíveis para o usuário.\n    \"\"\"\n    print(\"\\nEscolha uma opção:\")\n    print(\"[1] Buscar por email:senha\")\n    print(\"[2] Buscar por cpf:senha\")\n    print(\"[3] Sair\")\n\ndef process_site(site, search_type, db):\n    \"\"\"\n    Processa a busca por um site específico, exibe os resultados e os salva.\n    Esta função é projetada para ser executada em paralelo para múltiplos sites.\n    \"\"\"\n    print(f'\\n[INFO] Iniciando busca para \"{site}\" com tipo de dado \"{search_type}\"...')\n    try:\n        results = db.search(site, search_type)\n        if results:\n            print(f'[RESULTADO] {len(results)} dados encontrados para \"{site}\":')\n            for result in sorted(list(results)): # Exibe resultados ordenados\n                print(result)\n            save_results(results, search_type, site)\n        else:\n            print(f'[INFO] Nenhum resultado encontrado para \"{site}\".')\n    except Exception as e:\n        print(f'[ERRO] Falha ao processar site \"{site}\": {e}')\n\ndef main():\n    \"\"\"\n    Função principal que coordena o fluxo do programa: inicialização,\n    seleção de opções, entrada de sites e execução das buscas paralelas.\n    \"\"\"\n    dir_name = 'db'\n    \n    # Inicializa a base de dados. Se o diretório não existir, o programa não prossegue.\n    cloud_db = CloudDatabase(dir_name)\n    if not cloud_db.file_paths and not (os.path.exists(dir_name) and os.path.isdir(dir_name)):\n        print(\"[ERRO] Operação cancelada. Por favor, crie o diretório 'db' e adicione arquivos .txt.\")\n        return\n\n    print_header()\n\n    search_type = None\n    while search_type not in [\"email:senha\", \"cpf:senha\"]:\n        print_options()\n        option = input(\"Digite a opção desejada: \")\n\n        if option == \"1\":\n            search_type = \"email:senha\"\n        elif option == \"2\":\n            search_type = \"cpf:senha\"\n        elif option == \"3\":\n            print(\"Saindo...\")\n            return\n        else:\n            print(\"[ERRO] Opção inválida. Tente novamente.\")\n            continue\n\n    sites_input = input(\"\\nDigite os sites separados por espaço (exemplo: facebook.com gmail.com): \")\n    sites = [s.strip() for s in sites_input.split() if s.strip()] # Limpa e filtra sites vazios\n\n    if not sites:\n        print(\"[ERRO] Nenhum site foi fornecido. Saindo...\")\n        return\n\n    print(f\"\\n[INFO] Iniciando busca para {len(sites)} site(s) com {os.cpu_count() * 2} threads de arquivo por busca...\")\n    # Utiliza um ThreadPoolExecutor para processar múltiplos sites em paralelo.\n    with concurrent.futures.ThreadPoolExecutor() as executor:\n        futures = [executor.submit(process_site, site, search_type, cloud_db) for site in sites]\n        concurrent.futures.wait(futures) # Aguarda a conclusão de todas as buscas de sites.\n    \n    print(\"\\n[INFO] Todas as buscas foram concluídas.\")\n\nif __name__ == '__main__':\n    # Cria um diretório 'db' de exemplo e um arquivo para testes se não existirem\n    if not os.path.exists('db'):\n        os.makedirs('db')\n        print(\"[INFO] Diretório 'db' criado para exemplos.\")\n    if not os.path.exists('db/sample1.txt'):\n        with open('db/sample1.txt', 'w', encoding='utf-8') as f:\n            f.write(\"user1@facebook.com:pass123\\n\")\n            f.write(\"Alguma linha com texto e user2@gmail.com:pass456 mas buscando facebook.com\\n\")\n            f.write(\"facebook.com user3@facebook.com:senhaforte\\n\")\n            f.write(\"CPF 12345678901:senhacpf no site do banco.com\\n\")\n            f.write(\"Outra linha com user4@outlook.com:outrasenha e amazon.com\\n\")\n        print(\"[INFO] Arquivo 'db/sample1.txt' criado com dados de exemplo.\")\n    if not os.path.exists('db/sample2.txt'):\n        with open('db/sample2.txt', 'w', encoding='utf-8') as f:\n            f.write(\"admin@banco.com:senha_banco_123\\n\")\n            f.write(\"11223344556:minhasenha no site_exemplo.net\\n\")\n        print(\"[INFO] Arquivo 'db/sample2.txt' criado com dados de exemplo.\")\n            \n    main()"
 ]
}