misc-utilities
/
gitea-repo-migrator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Now able to filter source repos by required tags

This commit is contained in:
Mike 2023-02-11 01:07:42 -08:00
parent 9d904924a4
commit 9c0640f2d1
1 changed files with 15 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
domain/Migrator.py
View File

@ -112,7 +112,7 @@ class Migrator:
assert destination_repo_name is not None, "Destination repo name must be specified"
assert destination_topics is not None, "Destination topics must 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
@ -127,6 +127,7 @@ class Migrator:
# Filter
source_repos = self._filter_repos_for_required_topics(repos=source_repos, topics=source_topics)
print()
repos_migrate = []
repos_ignore = []
@ -226,7 +227,13 @@ class Migrator:
return source_repos
def _filter_repos_for_required_topics(self, repos: list[giteapy.Repository], topics: list[str]) -> list[giteapy.Repository]:
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}")
@ -244,21 +251,21 @@ class Migrator:
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
self.__logger.error(f"Repo topics: {repo_topics[repo.id]}")
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(f"Keeping the following repos because they contain all required topics ({topics}):")
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("Rejecting the following repos because they don't contain all required topics:")
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]})")
@ -267,7 +274,8 @@ class Migrator:
return repos_keep
def _check_required_topics(self, topics_present: list[str], topics_required: list[str]) -> bool:
@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: