work

domain/Migrator.py

@@ -19,6 +19,8 @@ class Migrator:
 		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
@@ -35,33 +37,139 @@ class Migrator:
 		logger.addHandler(stdout_handler)
 		
 		self.__logger = logger
-		
+	
 	def _get_org_apis(self):
 		
-		conf_source = giteapy.Configuration()
+		api_source = self._get_org_api(
-		conf_source.api_key['access_token'] = self.__source_token
+			hostname=self.__source_host, port=self.__source_port,
-		conf_source.host = self._make_api_base(host=self.__source_host, port=self.__source_port)
+			token=self.__source_token
-		api_source = giteapy.OrganizationApi(giteapy.ApiClient(conf_source))
+		)
-		
+		api_destination = self._get_org_api(
-		conf_destination = giteapy.Configuration()
+			hostname=self.__destination_host, port=self.__destination_port,
-		conf_destination.api_key['access_token'] = self.__destination_token
+			token=self.__destination_token
-		conf_destination.host = self._make_api_base(host=self.__destination_host, port=self.__destination_port)
+		)
-		api_destination = giteapy.OrganizationApi(giteapy.ApiClient(conf_destination))
 		
 		return api_source, api_destination
 	
-	def _make_api_base(self, host, port):
+	def _get_org_api(self, hostname, port, token):
 		
-		base = f"https://{host}"
+		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.")
+		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
+			)
+			
+			self.__logger.info("Migrate body:")
+			self.__logger.info(migrate_body)

main.py

@@ -64,6 +64,14 @@ def main():
 		help="Name of the destination organization"
 	)
 	
+	parser.add_argument(
+		"--no-verify-ssl",
+		dest="verify_ssl",
+		default=True,
+		action="store_false",
+		help="Don't verify SSL certificates",
+	)
+	
 	args = parser.parse_args()
 	mig = Migrator(
 		source_host=args.source_hostname,
@@ -73,9 +81,10 @@ def main():
 		destination_port=args.destination_port,
 		destination_token=args.destination_token
 	)
+	mig.set_verify_ssl(args.verify_ssl)
 	mig.migrate_entire_org(
 		source_org=args.source_org,
-		destination_org=args.destination_org
+		destination_org=args.destination_org,
 	)