""" Module that manage the behaviour of source allocated in a http """ import re import os from copy import deepcopy import urllib.request import urllib.parse import http.client import requests from tools.protocols.masterProtocol import MasterProtocol from tools import const class Http(MasterProtocol): ConnectionError = type('ConnectionError', (Exception,), {}) LoginError = type('LoginError', (Exception,), {}) NotEnoughPrivileges = type('NotEnoughPrivileges', (Exception, ), {}) NoDirectoryOrFile = type('NoDirectoryOrFile', (Exception,), {}) DirectoryOrFileExists = type('DirectoryOrFileExists', (Exception, ), {}) CONNECTION_REQUIRED = {'user': 'user', 'password': 'password', 'ip': 'ip', 'fileType': 'driver_type', 'port': 'port', 'filePath': 'source', 'domain': 'domain', 'layerName': 'layer_name', 'subtype': 'type'} NAME = const.HTTP_KEY VSI_CURL_STREAMING_STRING = '/vsicurl_streaming/' def __init__(self, ip, port, user="", password=""): if ip == 'www.dropbox.com': self.ip = 'www.dl.dropboxusercontent.com' else: self.ip = ip self.port = port self.user = user self.password = password self.conn = http.client.HTTPConnection(self.ip, self.port) self.http_scheme = 'http' def exists_layer(self, layer): if self.exists(layer['source']): return True raise self.NoDirectoryOrFile('{} not exists on {}:{}'.format( layer['source'], self.ip, self.port)) def connect(self): try: self.conn.connect() except ConnectionRefusedError: raise self.ConnectionError('{} on {}:{} off'.format( self.http_scheme, self.ip, self.port)) def exists(self, layer): try: path = layer.get_source() if self.port: port = ':{}'.format(self.port) else: port = '' url = """{}://{}{}{}""".format(self.http_scheme, self.ip, port, path) if '.zip/' in url: file = url.split("/")[-1] url = url.replace("/" + file, "") response = requests.get(url) if response.status_code in [200, 201]: return response.status_code == 200 except Exception: return False def list_files(self, directory, suffix=None): if self.ip == 'www.dl.dropboxusercontent.com': return True else: list_files = [] if not self.user: login_string = '' else: login_string = '{}:{}@'.format(urllib.parse.quote(self.user), urllib.parse.quote(self.password)) src_directory = "{}://{}{}:{}{}".format(self.http_scheme, login_string, self.ip, self.port, directory) page = urllib.request.urlopen(src_directory) try: page_decode = str(page.read().decode()) except UnicodeDecodeError: return False regex = re.compile(r'