codearchiver/codearchiver/core.py

import abc
import collections
#import codearchiver.modules  # In get_module_class
import codearchiver.version
import dataclasses
import logging
import queue
import requests
import time
import typing


logger = logging.getLogger(__name__)


class InputURL:
	def __init__(self, url):
		if 0 < url.find('+') < url.find('://'):
			# '+' and '://' appear in the URL in this order and there is at least one character each before the + as well as between the two
			self._moduleScheme, self._url = url.split('+', 1)
		else:
			self._moduleScheme = None
			self._url = url
		self._response = None

	@property
	def url(self):
		return self._url

	@property
	def moduleScheme(self):
		return self._moduleScheme

	@property
	def content(self):
		if self._response is None:
			self._response = HttpClient().get(self.url)
		return self._response.text

	def __repr__(self):
		return f'{type(self).__module__}.{type(self).__name__}({self._url!r})'


@dataclasses.dataclass
class Result:
	'''Container for the result of a module'''

	id: str
	'''A unique ID for this result'''

	files: typing.List[str] = dataclasses.field(default_factory = list)
	'''List of filenames produced by the run'''

	submoduleResults: typing.List[typing.Tuple['Module', 'Result']] = dataclasses.field(default_factory = list)
	'''List of related submodules and their results'''


class HttpError(Exception):
	pass


class HttpClient:
	defaultRetries: int = 3
	defaultUserAgent: str = f'codearchiver/{codearchiver.version.__version__}'

	def __init__(self, retries = None, userAgent = None):
		self._session = requests.Session()
		self._retries = retries if retries else self.defaultRetries
		self._userAgent = userAgent if userAgent else self.defaultUserAgent

	def request(self, method, url, params = None, data = None, headers = None, timeout = 10, responseOkCallback = None):
		mergedHeaders = {'User-Agent': self._userAgent}
		if headers:
			mergedHeaders.update(headers)
		headers = mergedHeaders
		for attempt in range(self._retries + 1):
			# The request is newly prepared on each retry because of potential cookie updates.
			req = self._session.prepare_request(requests.Request(method, url, params = params, data = data, headers = headers))
			logger.info(f'Retrieving {req.url}')
			logger.debug(f'... with headers: {headers!r}')
			if data:
				logger.debug(f'... with data: {data!r}')
			try:
				r = self._session.send(req, timeout = timeout)
			except requests.exceptions.RequestException as exc:
				if attempt < self._retries:
					retrying = ', retrying'
					level = logging.WARNING
				else:
					retrying = ''
					level = logging.ERROR
				logger.log(level, f'Error retrieving {req.url}: {exc!r}{retrying}')
			else:
				if responseOkCallback is not None:
					success, msg = responseOkCallback(r)
				else:
					success, msg = (True, None)
				msg = f': {msg}' if msg else ''

				if success:
					logger.debug(f'{req.url} retrieved successfully{msg}')
					return r
				else:
					if attempt < self._retries:
						retrying = ', retrying'
						level = logging.WARNING
					else:
						retrying = ''
						level = logging.ERROR
					logger.log(level, f'Error retrieving {req.url}{msg}{retrying}')
			if attempt < self._retries:
				sleepTime = 1.0 * 2**attempt # exponential backoff: sleep 1 second after first attempt, 2 after second, 4 after third, etc.
				logger.info(f'Waiting {sleepTime:.0f} seconds')
				time.sleep(sleepTime)
		else:
			msg = f'{self._retries + 1} requests to {req.url} failed, giving up.'
			logger.fatal(msg)
			raise HttpError(msg)
		raise RuntimeError('Reached unreachable code')

	def get(self, *args, **kwargs):
		return self.request('GET', *args, **kwargs)

	def post(self, *args, **kwargs):
		return self.request('POST', *args, **kwargs)


class Module:
	'''An abstract base class for a module.'''

	name: typing.Optional[str] = None
	'''The name of the module. Modules without a name are ignored, and names must be unique.'''

	@staticmethod
	def matches(inputUrl: InputURL) -> bool:
		'''Whether or not this module is for handling `inputUrl`.'''
		return False

	def __init__(self, inputUrl, id_ = None):
		self._inputUrl = inputUrl
		self._url = inputUrl.url
		self._id = id_
		self._httpClient = HttpClient()

	@abc.abstractmethod
	def process(self) -> Result:
		'''Perform the relevant retrieval(s)'''

	def __repr__(self):
		return f'{type(self).__module__}.{type(self).__name__}({self._inputUrl!r})'


def get_module_class(inputUrl: InputURL) -> typing.Type[Module]:
	'''Get the Module class most suitable for handling `inputUrl`.'''

	# Ensure that modules are imported
	# This can't be done at the top because the modules need to refer back to the Module class.
	import codearchiver.modules

	# Collect all the Module subclasses and names
	modules = set()
	modulesByName = {}  # name: str -> List[Module]
	q = queue.Queue()
	q.put_nowait(Module)
	while not q.empty():
		class_ = q.get_nowait()
		for c in class_.__subclasses__():
			if c.name is not None:
				logger.debug(f'Found {c.name!r} module {c.__module__}.{c.__name__}')
				modules.add(c)
				if c.name not in modulesByName:
					modulesByName[c.name] = []
				modulesByName[c.name].append(c)
			else:
				logger.debug(f'Found nameless module {c.__module__}.{c.__name__}')
			q.put_nowait(c)

	# Verify that there are no module name collisions
	if any(len(x) > 1 for x in modulesByName.values()):
		raise RuntimeError(f'Found multiple modules with the same name')

	# Check if the URL references one of the modules directly
	if inputUrl.moduleScheme:
		if inputUrl.moduleScheme in modulesByName:
			module = modulesByName[inputUrl.moduleScheme][0]
			logger.info(f'Selecting module {module.__module__}.{module.__name__}')
			return module
		else:
			raise RuntimeError(f'No module with name {inputUrl.moduleScheme!r} exists')

	# Check if exactly one of the modules matches
	matches = [class_ for class_ in modules if class_.matches(inputUrl)]
	if len(matches) >= 2:
		logger.error('Multiple matching modules for input URL')
		logger.debug(f'Matching modules: {matches!r}')
		raise RuntimeError('Multiple matching modules for input URL')
	if matches:
		logger.info(f'Selecting module {matches[0].__module__}.{matches[0].__name__}')
		return matches[0]
	raise RuntimeError('No matching modules for input URL')


def get_module_instance(inputUrl: InputURL, **kwargs) -> Module:
	'''Get an instance of the Module class most suitable for handling `inputUrl`.'''
	return get_module_class(inputUrl)(inputUrl, **kwargs)