Also add ability to copy source topics

Pipfile

@@ -4,7 +4,7 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-giteapy = "*"
+giteapy-soteria = {git = "https://github.com/Yousif-CS/giteapy.git"}
 
 [dev-packages]
 

Pipfile.lock

@@ -1,7 +1,7 @@
 {
     "_meta": {
         "hash": {
-            "sha256": "824c02127733d501ae21f3e5ccc7ee72fc4b2fe92bbca23c77ebd204d294b0fc"
+            "sha256": "1dc9e96fd5a12468ed7d0869b11b9fbca2464e4b806fb8b9c17391a41b6f0eb8"
         },
         "pipfile-spec": 6,
         "requires": {
@@ -24,19 +24,16 @@
             "markers": "python_version >= '3.6'",
             "version": "==2022.12.7"
         },
-        "giteapy": {
-            "hashes": [
-                "sha256:2078c802a4626bf311e911c969c34b7d19fbe9175e2910e1965b24ff69221470"
-            ],
-            "index": "pypi",
-            "version": "==1.0.8"
+        "giteapy-soteria": {
+            "git": "https://github.com/Yousif-CS/giteapy.git",
+            "ref": "e0a089bdfb7ef6130b43727c50e78f176379db20"
         },
         "python-dateutil": {
             "hashes": [
                 "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86",
                 "sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"
             ],
-            "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
+            "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'",
             "version": "==2.8.2"
         },
         "six": {
@@ -44,7 +41,7 @@
                 "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926",
                 "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"
             ],
-            "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
+            "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2'",
             "version": "==1.16.0"
         },
         "urllib3": {

README.md

@@ -1,2 +1,57 @@
-# gitea-repo-migrator
+# Mike's Gitea Repo Migrator
+
+Just a script to help make it a little easier to migrate an entire organization (with bulk selection) from one Gitea instance to another.
+
+Supports changing destination names a bit, and adding topics to each transferred repo, as well as optional bulk-selected deletion from the source.
+
+by Mike Peralta
+
+Current license: You are free to clone and use this program but all other rights reserved, provided you accept 100% of all liability of any outcome of use/download/etc this program. Better license coming soon.
+
+## Requirements
+
+* python 3.10
+* pipenv
+
+## Installation
+
+1. Clone this repo and `cd` to its root directory
+
+2. Execute `pipenv install` (and optionally check with `pipenv check`)
+
+## Execution
+
+Ask `pipenv` to execute *main.py* on your behalf with the following:
+
+```console
+
+$ cd /path/to/gitea-repo-migrator
+$ pipenv run python ./main.py --help
+```
+
+Optionally ask for a shell to run multiple times a bit faster:
+
+```console
+
+$ cd /path/to/gitea-repo-migrator
+$ pipenv shell
+$ python ./main.py --help
+$ python ./main.py --help
+```
+
+### App Tokens
+
+You'll need to generate an *Application Token* for both the source and destination servers, and pass the token along to the command line.
+
+### SSL Verification
+
+Pass the long switch `--no-verify-ssl` if any server sits behind a self-signed or wonky SSL certificate.
+
+### Destination Repo Names
+
+You can tweak the destination repo names a bit by using a string that includes `%N%` somewhere. The string `%N%` will expand to the original name. For example, if we use a repo originally named `my-test-repo` with the string `what-%N%`, the resulting destination repo name will be `what-my-test-repo`. This string will be recomputed for every migrated repo.
+
+### Destination Topics
+
+Topics will be duplicated from all source repos to their corresponding destination repos. You can specify additional topics with the `--destination-topic` switch. For example, to add the topic `migrated` to every repo, pass the switch `--destination-topic migrated`.
 

domain/Migrator.py

@@ -103,13 +103,23 @@ class Migrator:
 	
 	def migrate_entire_org(
 			self,
-		source_org: str,
-		destination_org: str, destination_repo_name: str, destination_topics: list
+			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, api_destination = self._get_org_apis()
+		api_source: giteapy.OrganizationApi
+		api_destination: giteapy.OrganizationApi
 		
-		source_repos = api_source.org_list_repos(source_org)
+		# Grab all org repos
+		source_repos = self._fetch_all_org_repos(org=source_org)
 		self.__logger.info(f"Found {len(source_repos)} repos on source:")
 		for repo in source_repos:
 			repo: giteapy.Repository
@@ -117,6 +127,10 @@ class Migrator:
 		
 		print()
 		
+		# Filter
+		source_repos = self._filter_repos_for_required_topics(repos=source_repos, topics=source_topics)
+		print()
+		
 		repos_migrate = []
 		repos_ignore = []
 		go_right_now = False
@@ -126,8 +140,11 @@ class Migrator:
 			
 			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":
@@ -183,6 +200,7 @@ class Migrator:
 					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.")
@@ -192,11 +210,88 @@ class Migrator:
 			else:
 				self.__logger.info("Confirmation not received; Won't do anything.")
 	
+	def _fetch_all_org_repos(self, org: str):
+		
+		api_source, api_destination = self._get_org_apis()
+		api_source: giteapy.OrganizationApi
+		
+		source_repos = []
+		
+		page = 0
+		while True:
+			
+			page += 1  # Starts at 1 for some reason
+			source_repos_page = api_source.org_list_repos(org, page=page, limit=25)
+			
+			if len(source_repos_page) == 0:
+				break
+			
+			source_repos.extend(source_repos_page)
+		
+		return source_repos
+	
+	def _filter_repos_for_required_topics(
+			
+			self,
+			repos: list[giteapy.Repository],
+			topics: list[str]
+			
+	) -> list[giteapy.Repository]:
+		
+		self.__logger.info(f"Filtering source repos for required topics: {topics}")
+		
+		repos_keep = []
+		repos_reject = []
+		repo_topics = {}
+		
+		api_source_repos = self._get_repo_api(
+			hostname=self.__source_host, port=self.__source_port,
+			token=self.__source_token
+		)
+		
+		for repo in repos:
+			
+			repo_topics[repo.id] = api_source_repos.repo_list_topics(owner=repo.owner.login, repo=repo.name)
+			repo_topics[repo.id] = repo_topics[repo.id].topics
+			
+			if self._check_required_topics(topics_present=repo_topics[repo.id], topics_required=topics):
+				repos_keep.append(repo)
+			else:
+				repos_reject.append(repo)
+		
+		self.__logger.info("")
+		self.__logger.info(f"\nKeeping {len(repos_keep)} repos because they contain all required topics ({topics}):")
+		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} ({repo_topics[repo.id]})")
+		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
 	):
 		
@@ -204,9 +299,13 @@ class Migrator:
 		
 		destination_org = api_destination.org_get(org=destination_org_name)
 		destination_org: giteapy.Organization
-		
 		self.__logger.info(f"Destination organization: {destination_org.full_name}")
 		
+		api_source_repos = self._get_repo_api(
+			hostname=self.__source_host, port=self.__source_port,
+			token=self.__source_token
+		)
+		
 		source_repos_successful = []
 		for source_repo in repos:
 			
@@ -214,6 +313,9 @@ class Migrator:
 			
 			this_destination_repo_name = destination_repo_name.replace("%N%", source_repo.name)
 			
+			source_repo_topics = api_source_repos.repo_list_topics(owner=source_repo.owner.login, repo=source_repo.name)
+			source_repo_topics = source_repo_topics.topics
+			
 			migrate_body = giteapy.MigrateRepoForm(
 				mirror=False,
 				clone_addr=source_repo.clone_url,
@@ -243,6 +345,17 @@ class Migrator:
 			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}")
+					destination_api.repo_add_topc(
+						owner=destination_org.username,
+						repo=repo_new.name,
+						topic=topic,
+					)
+			
+			# Add specified topics
 			for topic in destination_topics:
 				self.__logger.debug(f"Appending topic to new repo: {topic}")
 				destination_api.repo_add_topc(

main.py

@@ -37,6 +37,14 @@ def main():
 		required=True,
 		help="Name of the source organization"
 	)
+	parser.add_argument(
+		"--source-required-topic", "--source-topic",
+		dest="source_topics",
+		default=[],
+		action="append",
+		help="Specify zero or more topics required topics for source repositories."
+		" Any repository that doesn't have all required topics will be skipped"
+	)
 	
 	parser.add_argument(
 		"--destination-hostname", "--dest-hostname", "--destination-host", "--dest-host",
@@ -68,16 +76,46 @@ def main():
 		"--destination-repo-name", "--destination-name", "--dest-repo-name", "--dest-name",
 		dest="destination_repo_name",
 		default="%N%",
-		help="Specify the destination repository name(s). Use wildcard %N% anywhere to denote the original name"
+		help="Specify the destination repository name(s). Use wildcard %%N%% anywhere to denote the original name"
 	)
 	parser.add_argument(
-		"--destination-add-topic", "--destination-topic", "-dest-add-topic", "--dest-topic",
+		"--destination-add-topic", "--destination-topic", "--dest-add-topic", "--dest-topic",
 		dest="destination_topics",
 		default=[],
 		action="append",
 		help="Specify zero or more topics to add to each destination repository"
 	)
 	
+	parser.add_argument(
+		"--destination-copy-topics", "--dest-copy-topics",
+		dest="do_destination_copy_topics",
+		default=True,
+		action="store_true",
+		help="Destination repos should copy topics from their source."
+	)
+	parser.add_argument(
+		"--no-destination-copy-topics", "--no-dest-copy-topics",
+		dest="do_destination_copy_topics",
+		default=True,
+		action="store_false",
+		help="Destination repos should NOT copy topics from their source."
+	)
+	
+	parser.add_argument(
+		"--interactive",
+		dest="interactive",
+		default=True,
+		action="store_true",
+		help="Ask to confirm each migration.",
+	)
+	parser.add_argument(
+		"--no-interactive", "--non-interactive",
+		dest="interactive",
+		default=True,
+		action="store_false",
+		help="Do not ask to confirm each migration; Migrate all repos quickly.",
+	)
+	
 	parser.add_argument(
 		"--no-verify-ssl",
 		dest="verify_ssl",
@@ -97,10 +135,13 @@ def main():
 	)
 	mig.set_verify_ssl(args.verify_ssl)
 	mig.migrate_entire_org(
+		interactive=args.interactive,
 		source_org=args.source_org,
+		source_topics=args.source_topics,
 		destination_org=args.destination_org,
 		destination_repo_name=args.destination_repo_name,
-		destination_topics=args.destination_topics
+		destination_topics=args.destination_topics,
+		do_destination_copy_topics=args.do_destination_copy_topics
 	)