• 日期: 2025-01-09
  • 标签: 常规

Python Google Dork 搜索工具优化：提高效率并过滤无用 URL

本文将对以下代码进行优化，旨在提升搜索效率并过滤掉无用的URL。

import requests
import time
import os
import urllib3
import sys
import random
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
from colorama import Fore, init

now_time = time.strftime('%Y-%m-%d %H-%M')


# 读取Dorks
def work(dorks):
    with open(dorks, mode='r', encoding='utf-8') as file:
        read_content = file.readlines()
        # 将内容加入列表
        content = [result.strip() for result in read_content]
        # 返回数量丢给任务池
        return len(read_content), content


# Google搜索
def google_serach(query, locks, filename):
    try:
        # 关闭HTTPS报错信息
        urllib3.disable_warnings()
        filename = os.path.join(os.getcwd(), f'{filename}.txt')
        domains = ['fr','it','ca','co.uk','ru','co,jp','co.kr','com.au','co.in','com.br','com.ar','co.za','co.nz','es','se','nl','ch','at','dk','be','pl','fi','ie','pt','gr', 'tw', 'com', 'uk', 'de', 'br', 'ca', 'kr', 'mx', 'au', 'za']
        random_domain = random.choice(domains)
        url = f'https://www.google.{random_domain}/search?q={query}&num=100'
        # 请求头
        headers = {
                   'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
                   'accept-language': 'zh-CN,zh;q=0.9',
                   'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
                   'referer': 'https://www.google.com/',
                   'origin': 'https://www.google.com',
                   'Sec-Fetch-Site': 'same-origin',
                   'Sec-Fetch-Mode': 'navigate',
                   'Sec-Fetch-User': '?1',
                   'Sec-Fetch-Dest': 'document'
        }
        # 代理
        proxies = {'http': 'http://127.0.0.1:7890', 'https': 'http://127.0.0.1:7890'}
        response = requests.get(url=url, headers=headers, proxies=proxies, verify=False, timeout=5)
        soup = BeautifulSoup(response.content, 'html.parser')
        # 查找全部div标签
        find_div = soup.find_all('div', {'class': 'yuRUbf'})
        # 开启线程锁
        locks.acquire()
        # 加入列表
        get_url = [url.findNext('a')['href'] + '
' for url in find_div]
        global url_num, dork_finish_num
        url_num += len(get_url)
        dork_finish_num += 1
        print(Fore.GREEN + f'
{now_time}[INFO]{ '-' * 10}>已获取Url数量：{url_num}  Dorsk数量：{dork_finish_num} / {dork_total_num}', end='' + Fore.RESET)
        # 写入文件
        write_info(filename, get_url)
        # 释放线程锁
        locks.release()

    except TimeoutError:
        pass


# 写入文件函数
def write_info(filename, get_url):

    with open(filename, mode='a+', encoding='utf-8') as file:

        file.writelines(get_url)


if __name__ == '__main__':
    while True:
        try:
            init()  # 初始化颜色模块
            dorks_file = input(Fore.YELLOW + f'
{now_time}[INFO]{ '-' * 10}>input file:' + Fore.RESET)
            print('')
            filename = input(Fore.YELLOW + f'
{now_time}[INFO]{ '-' * 10}>output file:' + Fore.RESET)
            # 接受work函数返回的元组
            dork_total_num, query_list = work(dorks_file)
            # 定义全局变量完成数量/URL数量
            dork_finish_num = url_num = 0

            # 定义线程池数量
            threads = ThreadPoolExecutor(max_workers=5)

            # 定义全局锁
            threads_lock = Lock()

            # 分配进程池任务
            for dokr_list in query_list:

                threads.submit(google_serach, dokr_list, threads_lock, filename)

            threads.shutdown()
            if len(sys.argv) == 1:
                pass
            input(Fore.YELLOW + f'

{now_time}[INFO]{'-' * 10}>程序运行完毕，按回车退出' + Fore.RESET)
            break
        # 文件为空
        except FileNotFoundError:
            print(Fore.RED + f'{now_time}[Error]{'-' * 10}>文件不存在' + Fore.RESET)
        # 中断异常
        except KeyboardInterrupt:
            sys.exit(1)

优化建议如下：

  1. 在写入文件时，可以使用批量写入，而不是每次写入一个 url，可以减少 I/O 操作的次数，提高效率。
  2. 可以考虑使用多进程或异步IO的方式，以进一步提高程序效率。
  3. 可以在搜索时加入?问号过滤，避免获取无用的 url。

代码实现：

  1. 批量写入文件

将写入文件的函数 write_info 修改为批量写入：

# 写入文件函数
def write_info(filename, get_url_list):
    with open(filename, mode='a+', encoding='utf-8') as file:
        file.writelines(get_url_list)

同时，在 google_serach 函数中，每获取一次 url 就加入列表 get_url，当列表长度达到一定数量后，就批量写入文件，并清空列表：

# 加入列表
get_url.append(url.findNext('a')['href'] + '
')
if len(get_url) == 20:
    write_info(filename, get_url)
    get_url = []
  1. 异步IO优化

使用 asyncio 和 aiohttp 库实现异步IO优化，将 google_serach 函数改为异步函数，同时使用 async with 异步请求 url，代码如下：

import asyncio
import aiohttp

# Google搜索
async def google_serach(query, filename):
    try:
        filename = os.path.join(os.getcwd(), f'{filename}.txt')
        domains = ['fr','it','ca','co.uk','ru','co,jp','co.kr','com.au','co.in','com.br','com.ar','co.za','co.nz','es','se','nl','ch','at','dk','be','pl','fi','ie','pt','gr', 'tw', 'com', 'uk', 'de', 'br', 'ca', 'kr', 'mx', 'au', 'za']
        random_domain = random.choice(domains)
        url = f'https://www.google.{random_domain}/search?q={query}&num=100'
        # 请求头
        headers = {
                   'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
                   'accept-language': 'zh-CN,zh;q=0.9',
                   'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
                   'referer': 'https://www.google.com/',
                   'origin': 'https://www.google.com',
                   'Sec-Fetch-Site': 'same-origin',
                   'Sec-Fetch-Mode': 'navigate',
                   'Sec-Fetch-User': '?1',
                   'Sec-Fetch-Dest': 'document'
        }
        async with aiohttp.ClientSession(headers=headers) as session:
            async with session.get(url, verify_ssl=False, timeout=5) as response:
                soup = BeautifulSoup(await response.text(), 'html.parser')
        # 查找全部div标签
        find_div = soup.find_all('div', {'class': 'yuRUbf'})
        # 加入列表
        get_url = [url.findNext('a')['href'] + '
' for url in find_div]
        global url_num, dork_finish_num
        url_num += len(get_url)
        dork_finish_num += 1
        print(Fore.GREEN + f'
{now_time}[INFO]{ '-' * 10}>已获取Url数量：{url_num}  Dorsk数量：{dork_finish_num} / {dork_total_num}', end='' + Fore.RESET)
        # 批量写入文件
        write_info(filename, get_url)
    except TimeoutError:
        pass

使用 asyncio 执行任务，代码如下：

# 分配进程池任务
async def main(query_list, filename):
    tasks = []
    for dokr_list in query_list:
        tasks.append(asyncio.create_task(google_serach(dokr_list, filename)))
    await asyncio.wait(tasks)

if __name__ == '__main__':
    while True:
        try:
            init()  # 初始化颜色模块
            dorks_file = input(Fore.YELLOW + f'
{now_time}[INFO]{ '-' * 10}>input file:' + Fore.RESET)
            print('')
            filename = input(Fore.YELLOW + f'
{now_time}[INFO]{ '-' * 10}>output file:' + Fore.RESET)
            # 接受work函数返回的元组
            dork_total_num, query_list = work(dorks_file)
            # 定义全局变量完成数量/URL数量
            dork_finish_num = url_num = 0

            # 异步IO优化
            asyncio.run(main(query_list, filename))

            if len(sys.argv) == 1:
                pass
            input(Fore.YELLOW + f'

{now_time}[INFO]{'-' * 10}>程序运行完毕，按回车退出' + Fore.RESET)
            break
        # 文件为空
        except FileNotFoundError:
            print(Fore.RED + f'{now_time}[Error]{'-' * 10}>文件不存在' + Fore.RESET)
        # 中断异常
        except KeyboardInterrupt:
            sys.exit(1)
  1. 添加?问号过滤

在 google_serach 函数中，将查询字符串中的空格替换为 +，并添加 ? 过滤，代码如下：

# Google搜索
async def google_serach(query, filename):
    try:
        filename = os.path.join(os.getcwd(), f'{filename}.txt')
        domains = ['fr','it','ca','co.uk','ru','co,jp','co.kr','com.au','co.in','com.br','com.ar','co.za','co.nz','es','se','nl','ch','at','dk','be','pl','fi','ie','pt','gr', 'tw', 'com', 'uk', 'de', 'br', 'ca', 'kr', 'mx', 'au', 'za']
        random_domain = random.choice(domains)
        query = query.replace(' ', '+')  # 将空格替换为 +
        url = f'https://www.google.{random_domain}/search?q={query}+?&num=100'
        # 请求头
        headers = {
                   'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
                   'accept-language': 'zh-CN,zh;q=0.9',
                   'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
                   'referer': 'https://www.google.com/',
                   'origin': 'https://www.google.com',
                   'Sec-Fetch-Site': 'same-origin',
                   'Sec-Fetch-Mode': 'navigate',
                   'Sec-Fetch-User': '?1',
                   'Sec-Fetch-Dest': 'document'
        }
        async with aiohttp.ClientSession(headers=headers) as session:
            async with session.get(url, verify_ssl=False, timeout=5) as response:
                soup = BeautifulSoup(await response.text(), 'html.parser')
        # 查找全部div标签
        find_div = soup.find_all('div', {'class': 'yuRUbf'})
        # 加入列表
        get_url = [url.findNext('a')['href'] + '
' for url in find_div if '?' in url.findNext('a')['href']]
        global url_num, dork_finish_num
        url_num += len(get_url)
        dork_finish_num += 1
        print(Fore.GREEN + f'
{now_time}[INFO]{ '-' * 10}>已获取Url数量：{url_num}  Dorsk数量：{dork_finish_num} / {dork_total_num}', end='' + Fore.RESET)
        # 批量写入文件
        write_info(filename, get_url)
    except TimeoutError:
        pass

通过以上优化，代码的效率和结果准确性都得到了提升，可以更有效地进行 Google Dork 搜索。

注意：

  • 以上代码仅供参考，实际使用中可能需要根据具体情况进行调整。
  • 使用 Google Dork 搜索时，请遵守 Google 的使用条款和服务协议。
  • 避免使用 Google Dork 进行任何违法或不道德的行为。
Python Google Dork 搜索工具优化：提高效率并过滤无用 URL

原文地址: https://www.cveoy.top/t/topic/ouOa 著作权归作者所有。请勿转载和采集!

免费AI点我,无需注册和登录