gitea-repo-migrator/domain/Migrator.py

import giteapy
import logging
import sys


class Migrator:
	
	__DEFAULT_API_PATH = "/api/v1"
	
	def __init__(
		self,
		source_host, source_port, source_token,
		destination_host, destination_port, destination_token,
	):
		
		# noinspection PyTypeChecker
		self.__logger: logging.Logger = None
		self._init_logger()
		
		self.__verify_ssl = True
		
		self.__source_host = source_host
		self.__source_port = source_port
		self.__source_token = source_token
		
		self.__destination_host = destination_host
		self.__destination_port = destination_port
		self.__destination_token = destination_token
	
	def _init_logger(self):
		
		logger = logging.Logger(name=f"{type(self).__name__}", level=logging.INFO)
		
		stdout_handler = logging.StreamHandler(stream=sys.stdout)
		logger.addHandler(stdout_handler)
		
		self.__logger = logger
	
	def _get_repo_api(self, hostname, port, token):
		
		conf = giteapy.Configuration()
		conf.api_key['access_token'] = token
		conf.host = self._make_api_base(hostname=hostname, port=port)
		conf.verify_ssl = self.__verify_ssl
		api = giteapy.RepositoryApi(giteapy.ApiClient(conf))
		
		return api
	
	def _get_org_apis(self):
		
		api_source = self._get_org_api(
			hostname=self.__source_host, port=self.__source_port,
			token=self.__source_token
		)
		api_destination = self._get_org_api(
			hostname=self.__destination_host, port=self.__destination_port,
			token=self.__destination_token
		)
		
		return api_source, api_destination
	
	def _get_org_api(self, hostname, port, token):
		
		conf = giteapy.Configuration()
		conf.api_key['access_token'] = token
		conf.host = self._make_api_base(hostname=hostname, port=port)
		conf.verify_ssl = self.__verify_ssl
		api = giteapy.OrganizationApi(giteapy.ApiClient(conf))
		
		return api
	
	def _make_api_base(self, hostname, port):
		
		base = f"https://{hostname}"
		if port is not None:
			base += f":{port}"
		base += self.__DEFAULT_API_PATH
		
		return base
	
	def set_verify_ssl(self, b: bool):
		
		self.__verify_ssl = b
	
	def migrate_entire_org(self, source_org, destination_org):
		
		api_source, api_destination = self._get_org_apis()
		
		source_repos = api_source.org_list_repos(source_org)
		self.__logger.info(f"Found {len(source_repos)} repos on source:")
		for repo in source_repos:
			repo: giteapy.Repository
			self.__logger.info(f"- #{repo.id} {repo.full_name}")
		
		print()
		
		repos_migrate = []
		repos_ignore = []
		go_right_now = False
		for repo in source_repos:
			
			repo: giteapy.Repository
			
			while True:
				
				response = input(f"Migrate repo #{repo.id} \"{repo.full_name}\" ? (Y)es, (N)o, (G)o right now, (Q)uit ==> ")
				response = response.lower()
				
				valid_input = True
				if response == "y":
					repos_migrate.append(repo)
				elif response == "n":
					repos_ignore.append(repo)
				elif response == "g":
					self.__logger.info("Okay, done asking questions, migrating existing selections.")
					go_right_now = True
				elif response == "q":
					go_right_now = True
					repos_migrate.clear()
					repos_ignore.clear()
					self.__logger.info("Okay, quitting instead.")
				else:
					valid_input = False
					self.__logger.warning(f"Invalid input: {response}")
				
				if valid_input:
					break
			
			if go_right_now:
				break
		
		#
		self.__logger.info("")
		if len(repos_migrate):
			self.__logger.info("Repos to migrate:")
			for repo in repos_migrate:
				repo: giteapy.Repository
				self.__logger.info(f"#{repo.id} \"{repo.name}\"")
		else:
			self.__logger.info("No repos marked to migrate")
		
		self.__logger.info("")
		if len(repos_ignore):
			self.__logger.info("Repos to ignore:")
			for repo in repos_ignore:
				repo: giteapy.Repository
				self.__logger.info(f"#{repo.id} \"{repo.name}\"")
		else:
			self.__logger.info("No repos marked to ignore")
		
		if len(repos_migrate):
			confirmation = input("Do you confirm the above selections? Enter CONFIRM ==> ")
			if confirmation == "CONFIRM":
				self.__logger.info("Confirmation received; Processing ... ")
				self._migrate_repos(destination_org_name=destination_org, repos=repos_migrate)
			else:
				self.__logger.info("Confirmation not received; Won't do anything.")
	
	def _migrate_repos(self, destination_org_name: str, repos: list):
		
		api_source, api_destination = self._get_org_apis()
		
		destination_org = api_destination.org_get(org=destination_org_name)
		destination_org: giteapy.Organization
		
		self.__logger.info(f"Destination organization: {destination_org.full_name}")
		
		for repo in repos:
			
			repo: giteapy.Repository
			
			migrate_body = giteapy.MigrateRepoForm(
				mirror=False,
				clone_addr=repo.clone_url,
				uid=destination_org.id,
				private=repo.private,
				repo_name=repo.name,
				description=repo.description,
				labels=True, issues=True, pull_requests=True, releases=True, milestones=True, wiki=True
			)
			migrate_body.auth_token = self.__source_token
			migrate_body.swagger_types["auth_token"] = "str"
			migrate_body.attribute_map["auth_token"] = "auth_token"
			
			self.__logger.info("Migrate body:")
			self.__logger.info(migrate_body)
			
			destination_api = self._get_repo_api(
				hostname=self.__destination_host,
				port=self.__destination_port,
				token=self.__destination_token,
			)
			migration_result = destination_api.repo_migrate(body=migrate_body)
			self.__logger.info(f"Migration result: {migration_result}")