catchzeng / bing_images Goto Github PK

try:
    from util import get_file_name, rename, make_image_dir, download_image
except ImportError:  # Python 3
    from .util import get_file_name, rename, make_image_dir, download_image
try:
    from crawler import crawl_image_urls
except ImportError:  # Python 3
    from .crawler import crawl_image_urls
from typing import List
from multiprocessing.pool import ThreadPool
from time import time as timer
import os
import math
import pathlib
from colorama import init, Fore, Style
init(autoreset=True)

print(Fore.RED + r'''
    ██████  ██ ███    ██  ██████        ██ ███    ███  █████   ██████  ███████ ███████ 
    ██   ██ ██ ████   ██ ██             ██ ████  ████ ██   ██ ██       ██      ██      
    ██████  ██ ██ ██  ██ ██   ███ █████ ██ ██ ████ ██ ███████ ██   ███ █████   ███████ 
    ██   ██ ██ ██  ██ ██ ██    ██       ██ ██  ██  ██ ██   ██ ██    ██ ██           ██ 
    ██████  ██ ██   ████  ██████        ██ ██      ██ ██   ██  ██████  ███████ ███████ ''' + Style.DIM + '''
    //////// ''' + Style.NORMAL + '''Automated, Multithreaded Chrome URL Fetcher and Image Downloader''' + Style.DIM +''' ////////''' + Fore.WHITE + '''

    https://github.com/CatchZeng/bing_images
''')

_FINISH = False


def fetch_image_urls(
    query: str,
    limit: int = 20,
    file_type: str = '',
    filters: str = ''
) -> List[str]:
    result = list()
    keywords = query
    if len(file_type) > 0:
        keywords = query + " " + file_type
    urls = crawl_image_urls(keywords, filters, limit)
    for url in urls:
        if isValidURL(url, file_type) and url not in result:
            result.append(url)
            if len(result) >= limit:
                break
    return result


def isValidURL(url, file_type):
    if len(file_type) < 1:
        return True
    return url.endswith(file_type)


def download_images(
    query: str,
    limit: int = 20,
    output_dir='',
    pool_size: int = 20,
    file_type: str = '',
    filters: str = '',
    force_replace=False
):
    start = timer()
    image_dir = make_image_dir(output_dir, force_replace)
    print(f"📁 Save path: {Fore.BLUE}{image_dir}")

    # Fetch more image URLs to avoid some images are invalid.
    max_number = math.ceil(limit*1.5)
    urls = fetch_image_urls(query, max_number, file_type, filters)
    entries = get_image_entries(urls, image_dir)

    print(f"⬇️  Downloading images\n")
    ps = pool_size
    if limit < pool_size:
        ps = limit
    download_image_entries(entries, ps, limit)

    rename_images(image_dir, query)

    print(f"✅ {Fore.GREEN}Done\n")
    elapsed = timer() - start
    print(f"⏱  {Fore.WHITE}Elapsed time: {Fore.RED}%.2fs\n" % elapsed)


def rename_images(dir, prefix):
    files = os.listdir(dir)
    index = 1
    print(f"📝  {Fore.BLUE}Renaming images{Fore.LIGHTBLACK_EX}...")
    for f in files:
        if f.startswith("."):
            print(f"{Fore.YELLOW}Escaping name of {f}{Fore.LIGHTBLACK_EX}...\n")
            continue
        src = os.path.join(dir, f)
        name = rename(f, index, prefix)
        dst = os.path.join(dir, name)
        os.rename(src, dst)
        index = index + 1
    print(f"{Fore.GREEN}    Finished renaming 👍\n")


def download_image_entries(entries, pool_size, limit):
    global _FINISH
    counter = 1
    _FINISH = False
    pool = ThreadPool(pool_size)
    results = pool.imap_unordered(
        download_image_with_thread, entries)
    for (url, result) in results:
        if counter > limit:
            _FINISH = True
            pool.terminate()
            break
        if result:
            urldir = pathlib.PurePath(url)
            urlp = urldir.parents[0]
            print(f"{Fore.YELLOW}   #{str(format(counter, '03'))}{Fore.LIGHTBLACK_EX}: {Fore.LIGHTBLACK_EX}{urlp}/{Fore.WHITE}{urldir.name}\n\t {Fore.GREEN}Downloaded! \n")
            counter = counter + 1


def get_image_entries(urls, dir):
    entries = []
    i = 0
    for url in urls:
        name = get_file_name(url, i, "#tmp#")
        path = os.path.join(dir, name)
        entries.append((url, path))
        i = i + 1
    return entries


def download_image_with_thread(entry):
    if _FINISH:
        return
    url, path = entry
    result = download_image(url, path)
    return (url, result)


if __name__ == '__main__':
    download_images("cat",
                    20,
                    output_dir="/Users/catchzeng/Desktop/cat",
                    pool_size=10,
                    file_type="png",
                    force_replace=True)

import requests
import shutil
import posixpath
import urllib
import os
from colorama import init, Fore, Style
init(autoreset=True)

DEFAULT_OUTPUT_DIR = "bing-images"


def download_image(url, path) -> bool:
    try:
        r = requests.get(url, stream=True)
        if r.status_code == 200:
            with open(path, 'wb') as f:
                r.raw.decode_content = True
                shutil.copyfileobj(r.raw, f)
            return True
        else:
            print(f"{Fore.RED}      ⚠️  Download image: {Fore.YELLOW}{url}\n{Fore.RED}         Err :: {Fore.WHITE}{r.status_code}\n")
            return False
    except Exception as e:
        print(f"{Fore.RED}      ⚠️  Download image: {Fore.YELLOW}{url}{Fore.RED}\n         Err :: {Fore.WHITE}{e}\n")
        return False


def get_file_name(url, index, prefix='image') -> str:
    try:
        path = urllib.parse.urlsplit(url).path
        filename = posixpath.basename(path).split('?')[0]
        type, _ = file_data(filename)
        result = "{}_{}.{}".format(prefix, index, type)
        return result
    except Exception as e:
        print(f"⚠️  {Fore.RED}Get file name: {Fore.YELLOW}{url}{Fore.RED}\n   Err :: {Fore.WHITE}{e}\n")
        return prefix


def rename(name, index, prefix='image') -> str:
    try:
        type, _ = file_data(name)
        result = "{}_{}.{}".format(prefix, index, type)
        return result
    except Exception as e:
        print(f"{Fore.RED}⚠️  Rename: {Fore.YELLOW}{name}{Fore.RED}\n   Err :: {Fore.WHITE}{e}\n")
        return prefix


def file_data(name):
    try:
        type = name.split(".")[-1]
        name = name.split(".")[0]
        if type.lower() not in ["jpe", "jpeg", "jfif", "exif", "tiff", "gif", "bmp", "png", "webp", "jpg"]:
            type = "jpg"
        return (type, name)
    except Exception as e:
        print(f"{Fore.RED}⚠️  Issue getting: {Fore.YELLOW}{name}{Fore.RED}\n    Err :: {Fore.WHITE}{e}\n")
        return (name, "jpg")


def make_image_dir(output_dir, force_replace=False) -> str:
    image_dir = output_dir
    if len(output_dir) < 1:
        image_dir = os.path.join(os.getcwd(), DEFAULT_OUTPUT_DIR)

    if force_replace:
        if os.path.isdir(image_dir):
            shutil.rmtree(image_dir)
    try:
        if not os.path.isdir(image_dir):
            os.makedirs(image_dir)
    except:
        pass

    return image_dir


if __name__ == '__main__':
    print("util")

from urllib.parse import quote
import shutil
from selenium import webdriver
import time
import json
from colorama import init, Fore, Style
init(autoreset=True)

BASE_URL = "https://www.bing.com/images/search?"


def gen_query_url(keywords, filters):
    keywords_str = "&q=" + quote(keywords)
    query_url = BASE_URL + keywords_str
    if len(filters) > 0:
        query_url += "&qft="+filters
    return query_url


def image_url_from_webpage(driver, max_number=10000):
    image_urls = list()

    time.sleep(10)
    img_count = 0

    while True:
        image_elements = driver.find_elements_by_class_name("iusc")
        if len(image_elements) > max_number:
            break
        if len(image_elements) > img_count:
            img_count = len(image_elements)
            driver.execute_script(
                "window.scrollTo(0, document.body.scrollHeight);")
        else:
            smb = driver.find_elements_by_class_name("btn_seemore")
            if len(smb) > 0 and smb[0].is_displayed():
                smb[0].click()
            else:
                break
        time.sleep(3)
    for image_element in image_elements:
        m_json_str = image_element.get_attribute("m")
        m_json = json.loads(m_json_str)
        image_urls.append(m_json["murl"])
    return image_urls


def crawl_image_urls(keywords, filters, max_number=10000, proxy=None, proxy_type="http"):
    chrome_path = shutil.which("chromedriver")
    chrome_path = "./bin/chromedriver" if chrome_path is None else chrome_path
    chrome_options = webdriver.ChromeOptions()
    if proxy is not None and proxy_type is not None:
        chrome_options.add_argument(
            "--proxy-server={}://{}".format(proxy_type, proxy))
    driver = webdriver.Chrome(chrome_path, chrome_options=chrome_options)

    query_url = gen_query_url(keywords, filters)
    driver.set_window_size(1920, 1080)
    driver.get(query_url)
    image_urls = image_url_from_webpage(driver, max_number)
    driver.close()

    if max_number > len(image_urls):
        output_num = len(image_urls)
    else:
        output_num = max_number

    print(f"{Fore.YELLOW}\n🕷  Crawled {Fore.RED}{len(image_urls)}{Fore.YELLOW} image urls.\n")

    return image_urls[0:output_num]


if __name__ == '__main__':
    images = crawl_image_urls(
        "mbot png", "+filterui:aspect-square", max_number=10)
    for i in images:
        print(f"{Fore.BLUE}{i}+\n")