333 lines
12 KiB
Python
333 lines
12 KiB
Python
import os
|
|
from enum import Enum
|
|
from threading import Semaphore, Thread
|
|
from urllib import request
|
|
from bs4 import BeautifulSoup, Tag, ResultSet
|
|
|
|
|
|
class MessageType(Enum):
|
|
"""
|
|
MessageType
|
|
|
|
Is an enumeration to define the different types of messages sent by the webpic messenger.
|
|
|
|
There are 3 types of messages.
|
|
- log -> log
|
|
- error -> err
|
|
- success -> suc
|
|
|
|
@author Jérémi Nihart / EndMove
|
|
@link https://git.endmove.eu/EndMove/WebPicDownloader
|
|
@version 1.0.0
|
|
@since 2022-09-05
|
|
"""
|
|
LOG = 'log'
|
|
ERROR = 'err'
|
|
SUCCESS = 'suc'
|
|
|
|
|
|
class WebPicDownloader(Thread):
|
|
"""
|
|
WebPicDownloader
|
|
|
|
Webpicdownloader is a scraping tool that allows you to browse a web page,
|
|
find the images and download them. This tool is easily usable and implementable
|
|
in an application. It has been designed to be executed in an integrated thread
|
|
in an asynchronous way. This tool allows to define 3 callback functions, one for
|
|
events, one in case of success and one in case of failure. It also has an
|
|
integrated entry point allowing it to be directly executed in terminal mode.
|
|
|
|
@author EndMove <contact@endmove.eu>
|
|
@link https://git.endmove.eu/EndMove/WebPicDownloader
|
|
@version 1.2.1
|
|
@since 2022-09-05
|
|
"""
|
|
# Variables
|
|
__callbacks: dict = None # Callback dictionary
|
|
__settings: dict = None # Webpic basics settings
|
|
__dl_infos: dict = None # Download informations
|
|
__sem: Semaphore = None # Semaphore for the webpic worker
|
|
|
|
_exit: bool = None # When set to True quit the thread
|
|
|
|
# Constructor
|
|
def __init__(self, path: str = None, headers: dict = None, messenger = None,
|
|
success = None, failure = None) -> None:
|
|
"""
|
|
Constructor
|
|
=> It is important to initialize the WebPicDownloader object properly. The callback
|
|
functions can be initialized after the creation of the object.
|
|
|
|
* :path: -> Folder in which the tool will create the download folders and place the images.
|
|
* :headers: -> Dictionary allowing to define the different parameters present in the header
|
|
of the requests sent by WebPic.
|
|
* :messenger: -> Callback function messenger (see setter).
|
|
* :success: -> Callback function success (see setter).
|
|
* :failure: -> Callback function failure (see setter).
|
|
"""
|
|
super().__init__(daemon=True, name='WebPic download worker')
|
|
self.__callbacks = {
|
|
'messenger': messenger if messenger else lambda msg, type: print(msg),
|
|
'success': success if success else lambda: print("Success!"),
|
|
'failure': failure if failure else lambda: print("failure!")
|
|
}
|
|
self.__settings = {
|
|
'root_path': path if path else os.getcwd(),
|
|
'headers': headers if headers else {
|
|
'User-Agent': "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36"
|
|
}
|
|
}
|
|
self.__dl_infos = {
|
|
'website_url': 'url',
|
|
'download_name': 'name',
|
|
'download_path': 'full_path',
|
|
'running': False
|
|
}
|
|
self.__sem = Semaphore(0)
|
|
self.__exit = False
|
|
|
|
self.start() # start deamon
|
|
|
|
|
|
# Internal functions
|
|
def __get_html(self, url: str) -> str:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Allow to retrieve the HTML content of a website.
|
|
|
|
* :url: -> The url of the site for which we want to get the content of the HTML page.
|
|
* RETURN -> Web page content.
|
|
"""
|
|
req = request.Request(url, headers=self.__settings.get('headers'))
|
|
response = request.urlopen(req)
|
|
return response.read().decode('utf-8')
|
|
|
|
|
|
def __find_all_img(self, html: str) -> ResultSet:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Allow to retrieve all images of an html page.
|
|
|
|
* :html: -> Html code in which to search for image balises.
|
|
* RETURN -> Iterable with all image balises.
|
|
"""
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
return soup.find_all('img')
|
|
|
|
|
|
def __find_img_link(self, img: Tag) -> str:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Allow to retrieve the link of a picture.
|
|
|
|
* :img: -> Image tag {@code bs4.Tag} for which to search the link of an image.
|
|
* RETURN -> Image link.
|
|
"""
|
|
if img.get('src'):
|
|
link = img.get('src')
|
|
elif img.get('data-src'):
|
|
link = img.get('data-src')
|
|
elif img.get('data-srcset'):
|
|
link = img.get('data-srcset')
|
|
elif img.get('data-fallback-src'):
|
|
link = img.get('data-fallback-src')
|
|
else:
|
|
raise ValueError("Unable to find image url")
|
|
if not 'http' in link:
|
|
raise ValueError("Bad image url")
|
|
return link
|
|
|
|
|
|
def __find_image_type(self, img_link: str) -> str:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Allow to retrieve the right image type (png, jpeg...)
|
|
|
|
* :img_link: -> Lien de l'image pourllaquel trouver le bon type.
|
|
* RETURN -> Type of image.
|
|
"""
|
|
type = img_link.split('.')[-1]
|
|
if '?' in type:
|
|
type = type.split('?')[0]
|
|
return type
|
|
|
|
|
|
def __download_img(self, url: str, filename: str) -> None:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Allow to download a picture from internet
|
|
|
|
* :url: -> Image url on the web.
|
|
* :filename: -> Full path with name of the future image.
|
|
"""
|
|
req = request.Request(url, headers=self.__settings.get('headers'))
|
|
raw_img = request.urlopen(req).read()
|
|
with open(filename, 'wb') as img:
|
|
img.write(raw_img)
|
|
|
|
|
|
def __initialize_folder(self, folder_path: str) -> None:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Checks if the folder in which to place the images to be uploaded exists and if
|
|
not chalk it up. An exception is raised if this folder already exists.
|
|
|
|
* :folder_path: -> Full path to the working folder (for the download task).
|
|
"""
|
|
if not os.path.exists(folder_path):
|
|
os.mkdir(folder_path)
|
|
else:
|
|
raise ValueError("The folder already exists, it may already contain images")
|
|
|
|
|
|
def __msg(self, message: str, type:MessageType=MessageType.LOG) -> None:
|
|
"""
|
|
Internal Function #do-not-use#
|
|
=> Use the messenger callback to send a message.
|
|
|
|
* :message: -> the message to send through callback
|
|
* :type: -> message type, can be ['log', 'err', 'suc']
|
|
"""
|
|
self.__callbacks.get('messenger')(message, type)
|
|
|
|
|
|
# Public functions
|
|
def set_success_callback(self, callback) -> None:
|
|
"""
|
|
Setter to define the callback function when the download succeeded.
|
|
|
|
* :callback: -> the callback function to call when the download is a success.
|
|
"""
|
|
self.__callbacks['success'] = callback
|
|
|
|
|
|
def set_failure_callback(self, callback) -> None:
|
|
"""
|
|
Setter to define the callback function called when the download fails.
|
|
|
|
* :callback: -> the callback function to call when the download is a failure.
|
|
"""
|
|
self.__callbacks['failure'] = callback
|
|
|
|
|
|
def set_messenger_callback(self, callback) -> None:
|
|
"""
|
|
Setter to define the callback function called when new messages arrive.
|
|
|
|
* :callback: -> the callback function to call when a message event is emited.
|
|
"""
|
|
self.__callbacks['messenger'] = callback
|
|
|
|
|
|
def start_downloading(self, url: str, name: str) -> None:
|
|
"""
|
|
Start downloading all pictures of a website.
|
|
|
|
* :url: -> The url of the website to annalyse.
|
|
* :folder_name: -> The name of the folder in which to upload the photos.
|
|
"""
|
|
if not self.is_alive:
|
|
self.__msg("Opss, the download thread is not running, please restart webpic.", MessageType.ERROR)
|
|
elif self.__dl_infos.get('running'):
|
|
self.__msg("Opss, the download thread is busy.", MessageType.ERROR)
|
|
else:
|
|
self.__dl_infos['website_url'] = url
|
|
self.__dl_infos['download_name'] = name
|
|
self.__sem.release()
|
|
|
|
|
|
def stop_downloading(self, block=False) -> None:
|
|
"""
|
|
Stops the download after the current item is processed and exit the downloading thread.
|
|
|
|
<!> Attention once called it will not be possible any more to download. <!>
|
|
|
|
* :block: -> If true, the function will block until the worker has finished working, if
|
|
False(default value), the stop message will be thrown and the program will continue.
|
|
"""
|
|
self.__exit = True
|
|
self.__sem.release()
|
|
if block:
|
|
self.join()
|
|
|
|
|
|
def is_download_running(self) -> bool:
|
|
"""
|
|
Indique si un téléchargement est en cours
|
|
|
|
* RETURN -> True if yes, False else.
|
|
"""
|
|
return self.__dl_infos['running'];
|
|
|
|
|
|
# Thread corp function
|
|
def run(self) -> None:
|
|
while True:
|
|
self.__sem.acquire() # waiting the authorization to process
|
|
|
|
if self.__exit: # check if the exiting is requested
|
|
return
|
|
|
|
self.__dl_infos['running'] = True # indicate that the thread is busy
|
|
|
|
try:
|
|
# parse infos from url
|
|
html = self.__get_html(self.__dl_infos.get('website_url')) # website html
|
|
images = self.__find_all_img(html) # find all img balises ing html
|
|
|
|
# setting up download informaations
|
|
tot_count = len(images) # count total image
|
|
dl_count = 0 # set download count to 0
|
|
self.__dl_infos['download_path'] = f"{self.__settings.get('root_path')}/{self.__dl_infos.get('download_name')}/" # format path
|
|
|
|
# init working directory
|
|
self.__initialize_folder(self.__dl_infos.get('download_path')) # Init download folder
|
|
self.__msg(f"WebPicDownloader found {tot_count} images on the website.")
|
|
|
|
# start images processing
|
|
for i, img in enumerate(images):
|
|
try:
|
|
self.__msg(f"Start downloading image {i}.")
|
|
|
|
img_link = self.__find_img_link(img) # find image link
|
|
self.__download_img(img_link, f"{self.__dl_infos.get('download_path')}image-{i}.{self.__find_image_type(img_link)}") # download the image
|
|
|
|
self.__msg(f"Download of image {i}, done!")
|
|
dl_count += 1 # increment download counter
|
|
except Exception as err:
|
|
self.__msg(f"ERROR: Unable to process image {i} -> err[{err}].")
|
|
# end images processing
|
|
|
|
self.__msg(f"WebPicDownloader has processed {dl_count} images out of {tot_count}.")
|
|
self.__callbacks.get('success')() # success, launch callback
|
|
except Exception as err:
|
|
self.__msg(f"ERROR: An error occured -> err[{err}]")
|
|
self.__callbacks.get('failure')() # error, launch callback
|
|
|
|
self.__dl_infos['running'] = False # inficate that the thread is free
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Internal entry point for testing and console use.
|
|
import time
|
|
|
|
wpd = WebPicDownloader() # Instance of webpic
|
|
|
|
# Callback functions
|
|
def success():
|
|
print("\nDownload completed with success.")
|
|
|
|
def failed():
|
|
print("\nDownload completed with errors.")
|
|
|
|
# Set-up callback functions for webpic
|
|
wpd.set_success_callback(success)
|
|
wpd.set_failure_callback(failed)
|
|
|
|
# Ask for download
|
|
print("\nWelcome to WebPicDownloader!")
|
|
url = input("Website URL ? ")
|
|
name = input("Folder name ? ")
|
|
wpd.start_downloading(url, name) # Start downloading
|
|
time.sleep(1) # We wait for the download to start before ask to stop it
|
|
wpd.stop_downloading(block=True) # Stop downloading but block till the download end. |