gitea-repo-migrator/domain/Migrator.py

from domain.API import API


import gitea
import logging
import sys

import certifi


class Migrator:
	
	__REPO_ORIGINAL_NAME_TOKEN = "%N%"
	
	def __init__(
		self,
		source_host, source_port, source_token,
		destination_host, destination_port, destination_token,
		verify_ssl: bool = True, ca_bundle: str = None
	):
		
		# noinspection PyTypeChecker
		self.__logger: logging.Logger = None
		self._init_logger()
		
		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
		
		self.__verify_ssl = verify_ssl
		self.__ca_bundle = ca_bundle
		
		api = API(
			verify_ssl=self.__verify_ssl,
			ca_bundle=self.__ca_bundle,
		)
		
		self.__source_api = api.get(
			hostname=self.__source_host,
			port=self.__source_port,
			token=self.__source_token,
			
		)
		self.__destination_api = api.get(
			hostname=self.__destination_host,
			port=self.__destination_port,
			token=self.__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_user_api(self, hostname, port, token) -> giteapy.UserApi:
		
		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.UserApi(giteapy.ApiClient(conf))
		
		return api
	"""
	
	"""
	def _get_repo_api(self, hostname, port, token) -> giteapy.RepositoryApi:
		
		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) -> (giteapy.OrganizationApi, giteapy.OrganizationApi):
		
		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) -> giteapy.OrganizationApi:
		
		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_destination_repo_name(self, pattern: str, repo: gitea.Repository):
		
		repo_name = pattern.replace(self.__REPO_ORIGINAL_NAME_TOKEN, repo.name)
		
		return repo_name
	
	"""
	def set_ca_bundle(self, bundle_path: str):
		
		self.__logger.info("Setting certificate bundle path")
		
		# Hacky but oh well
		self.__logger.info(f"Old path: {certifi.where()}")
		certifi.core._CACERT_PATH = bundle_path
		self.__logger.info(f"New path: {certifi.where()}")
		
		# TODO: JUST TESTING
		self.__verify_ssl = bundle_path
	"""
	
	def migrate_entire_org(
			self,
			interactive: bool = True,
			source_org: str = None, source_topics: list[str] = None,
			destination_org: str = None, destination_repo_name: str = None, destination_topics: list = None,
			do_destination_copy_topics: bool = True
	):
		assert source_org is not None, "Source org must be specified"
		assert destination_org is not None, "Destination org must be specified"
		assert destination_repo_name is not None, "Destination repo name must be specified"
		assert destination_topics is not None, "Destination topics must be specified"
		assert do_destination_copy_topics is not None, "Destination directive to copy source topics should be specified"
		
		# api_source, api_destination = self._get_org_apis()
		# api_source: giteapy.OrganizationApi
		# api_destination: giteapy.OrganizationApi
		
		# Tattle on certify
		self.__logger.info(f"Certifi is currently using CA bundle: {certifi.where()}")
		
		# Grab all org repos
		source_repos = self._fetch_all_org_repos(
			org_name=source_org
		)
		self.__logger.info(f"Found {len(source_repos)} repos on source:")
		for repo in source_repos:
			repo: gitea.Repository
			self.__logger.info(f"- {repo.get_full_name()}")
		
		print()
		
		# Filter
		source_repos = self._filter_repos_for_required_topics(
			repos=source_repos,
			topics_required=source_topics
		)
		print()
		self.__logger.info(f"Have {len(source_repos)} remaining repos after topic filtering:")
		for repo in source_repos:
			repo: gitea.Repository
			self.__logger.info(f"- {repo.get_full_name()}")
		
		repos_migrate = []
		repos_ignore = []
		go_right_now = False
		for repo in source_repos:
			
			repo: gitea.Repository
			
			while True:
				
				if interactive:
					response = input(
						f"Migrate repo #{repo.id} \"{repo.full_name}\" ?"
						" (Y)es, (N)o, (G)o right now, (Q)uit ==> "
					)
					response = response.lower()
				else:
					response = "y"
				
				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
		
		# Announce repo destination names
		self.__logger.info("")
		if len(repos_migrate):
			self.__logger.info("Repos to migrate:")
			for repo in repos_migrate:
				repo: gitea.Repository
				destination_name = self._make_destination_repo_name(
					pattern=destination_repo_name, repo=repo
				)
				self.__logger.info(
					f"#{repo.id} \"{repo.name}\"\n> \"{destination_name}\""
				)
		else:
			self.__logger.info("No repos marked to migrate")
		
		# Announce manually ignored repos
		self.__logger.info("")
		if len(repos_ignore):
			self.__logger.info("Repos to ignore:")
			for repo in repos_ignore:
				repo: gitea.Repository
				self.__logger.info(f"#{repo.id} \"{repo.get_full_name()}\"")
		else:
			self.__logger.info("No repos marked to ignore")
		
		# Migrate
		if len(repos_migrate):
			
			confirmation = input("Do you confirm the above selections? Enter MIGRATE ==> ")
			
			if confirmation == "MIGRATE":
				
				self.__logger.info("Confirmation received; Processing ... ")
				source_repos_successful, source_repos_failed = self._migrate_repos(
					destination_org_name=destination_org,
					destination_repo_name=destination_repo_name,
					destination_topics=destination_topics,
					do_destination_copy_topics=do_destination_copy_topics,
					repos=repos_migrate
				)
				self.__logger.info(
					f"{len(source_repos_successful)} of {len(repos_migrate)}"
					" repos successfully migrated."
				)
				if len(source_repos_failed) > 0:
					self.__logger.error(f"Failed to migrate {len(source_repos_failed)} repos:")
					for repo, exception in source_repos_failed:
						self.__logger.error(
							f"> {repo.name}"
						)
					self.__logger.error(f"Captured exception data:")
					for repo, exception in source_repos_failed:
						self.__logger.error(
							f"Failed to migrate repo: {repo.name}\n> {exception}"
						)
				
				self._delete_migrated_repos(
					source_org_name=source_org,
					repos=source_repos_successful
				)
				
			else:
				self.__logger.info("Confirmation not received; Won't do anything.")
	
	def _fetch_all_org_repos(self, org_name: str):
		
		org = gitea.Organization.request(
			gitea=self.__source_api,
			name=org_name
		)
		
		# Grabs all pages automatically
		repos = org.get_repositories()
		
		return repos
	
	def _filter_repos_for_required_topics(
			
			self,
			repos: list[gitea.Repository],
			topics_required: list[str]
			
	) -> list[gitea.Repository]:
		
		self.__logger.info(
			f"Filtering source repos for required topics: {topics_required}"
		)
		
		repos_keep = []
		repos_reject = []
		repo_topics = {}
		
		for repo in repos:
			
			repo: gitea.Repository
			
			repo_key = repo.get_full_name()
			
			topics_present = repo.get_topics()
			repo_topics[repo_key] = topics_present
			
			if self._check_required_topics(
				topics_present=repo_topics[repo_key],
				topics_required=topics_required
			):
				repos_keep.append(repo)
			else:
				repos_reject.append(repo)
		
		self.__logger.info("")
		self.__logger.info(
			f"\nKeeping {len(repos_keep)} repos"
			f" because they contain all required topics ({topics_required}):"
		)
		if len(repos_keep) > 0:
			for repo in repos_keep:
				self.__logger.info(f"> {repo.full_name}")
		else:
			self.__logger.info("> None")
		
		self.__logger.info("")
		self.__logger.info(
			f"Rejecting {len(repos_reject)} repos because they don't contain all required topics:"
		)
		if len(repos_reject) > 0:
			for repo in repos_reject:
				self.__logger.info(f"> {repo.full_name}")
		else:
			self.__logger.info("> None")
		
		return repos_keep
	
	@staticmethod
	def _check_required_topics(topics_present: list[str], topics_required: list[str]) -> bool:
		
		for topic in topics_required:
			if topic not in topics_present:
				return False
		
		return True
	
	def _migrate_repos(
			self,
			destination_org_name: str,
			destination_repo_name: str,
			destination_topics: list,
			do_destination_copy_topics: bool,
			repos: list
	):
		
		# api_source, api_destination = self._get_org_apis()
		# destination_org = api_destination.org_get(org=destination_org_name)
		# destination_org: giteapy.Organization
		
		api_dest_org = gitea.Organization.request(
			gitea=self.__destination_api,
			name=destination_org_name
		)
		
		self.__logger.info(f"Destination organization: {api_dest_org.full_name}")
		
		source_repos_successful = []
		source_repos_failed = []
		for source_repo in repos:
			
			source_repo: gitea.Repository
			
			this_destination_repo_name = destination_repo_name.replace(
				"%N%",
				source_repo.name
			)
			
			self.__logger.info(
				f"Migrating: {source_repo.name} ==> {this_destination_repo_name}"
			)
			
			source_repo_topics = source_repo.get_topics()
			
			try:
				
				repo_new = gitea.Repository.migrate_repo(
					gitea=self.__destination_api,
					service="gitea",  # type of remote service
					clone_addr=source_repo.clone_url,
					repo_name=this_destination_repo_name,
					description=source_repo.description,
					private=source_repo.private,
					auth_token=self.__source_token,
					auth_username=None,
					auth_password=None,
					mirror=False,
					mirror_interval=None,
					# lfs=False,
					# lfs_endpoint="",
					wiki=True,
					labels=True,
					issues=True,
					pull_requests=True,
					releases=True,
					milestones=True,
					repo_owner=destination_org_name,
				)
			
			except Exception as e:
				self.__logger.error(
					f"Failed to execute repo migration request:"
					f"\n{e}"
				)
				source_repos_failed.append(
					(source_repo, e)
				)
				continue
			
			self.__logger.debug(f"Migration result: {repo_new}")
			repo_new: gitea.Repository
			
			assert repo_new.name == this_destination_repo_name, \
				"New repository didn't end up with the correct name. Failure?"
			
			# Copy source topics?
			if do_destination_copy_topics:
				
				for topic in source_repo_topics:
					
					self.__logger.debug(f"Appending source topic to new repo: {topic}")
				
					repo_new.add_topic(topic=topic)
			
			# Add specified topics
			for topic in destination_topics:
				
				self.__logger.debug(f"Appending topic to new repo: {topic}")
				
				repo_new.add_topic(topic=topic)
			
			source_repos_successful.append(source_repo)
		
		return source_repos_successful, source_repos_failed
	
	def _delete_migrated_repos(self, source_org_name: str, repos: list[gitea.Repository]):
		
		if len(repos) == 0:
			self.__logger.warning(f"Cannot delete any migrated repos because none were successful!")
			return
		
		self.__logger.info("")
		self.__logger.info(f"Can now delete repos from source org: {source_org_name}")
		self.__logger.info(f"Will delete {len(repos)} successfully migrated repos:")
		for r in repos:
			
			r: gitea.Repository
			
			self.__logger.info(f"> #{r.id} \"{r.full_name}\" ==> {r.clone_url}")
		
		# Ask the user to confirm deletion
		response = input(
			"Would you like to delete the successfully migrated repos? Type DELETE ==> "
		)
		if response != "DELETE":
			self.__logger.info("Okay, won't delete migrated repos.")
			return
		
		do_quit = False
		do_delete_all = False
		for repo in repos:
			
			repo: gitea.Repository
			
			self.__logger.info(f"Next repo to delete: \"{repo.full_name}\"")
			do_delete = True if do_delete_all else False
			
			if do_delete is False:
				
				response = input("Delete this repo? (Y)es, (N), (A)ll, (Q)uit ==> ")
				response = response.lower()
				
				valid_response = False
				
				while valid_response is False:
					
					valid_response = True
					if response == "y":
						self.__logger.info(f"Okay, deleting {repo.name} from source")
						do_delete = True
					elif response == "n":
						self.__logger.info(f"Okay, won't delete {repo.name}")
					elif response == "a":
						self.__logger.info(f"Okay, deleting ALL remaining repos")
						do_delete_all = True
					elif response == "q":
						do_quit = True
					else:
						valid_response = False
						self.__logger.warning(f"Invalid response: {response}")
			
			if do_quit:
				break
			
			if do_delete or do_delete_all:
				
				self.__logger.info(f"Deleting repo: {repo.full_name}")
				
				repo.delete()